- Proper process termination:

if interrupted with CTRL+C blocking socket calls will return with -1. Each device should call FairMQDevice::Shutdown() before ending the running state to close open sockets, otherwise the interrupt call itself will block.

- FIX: Update number of received messages for FairMQFileSink.
- Add ability to poll on outputs for FairMQPoller.

fairmq/zeromq/FairMQContextZMQ.cxx

@@ -51,8 +51,11 @@ void FairMQContextZMQ::Close()
     int rc = zmq_ctx_destroy(fContext);
     if (rc != 0)
     {
-        LOG(ERROR) << "failed closing context, reason: " << zmq_strerror(errno);
+        if (errno == EINTR) {
+            LOG(ERROR) << " failed closing context, reason: " << zmq_strerror(errno);
+        } else {
+            fContext = NULL;
+            return;
+        }
     }
-
-    fContext = NULL;
 }

fairmq/zeromq/FairMQPollerZMQ.cxx

@@ -15,6 +15,7 @@
 #include <zmq.h>
 
 #include "FairMQPollerZMQ.h"
+#include "FairMQLogger.h"
 
 FairMQPollerZMQ::FairMQPollerZMQ(const vector<FairMQSocket*>& inputs)
 {
@@ -32,7 +33,11 @@ FairMQPollerZMQ::FairMQPollerZMQ(const vector<FairMQSocket*>& inputs)
 
 void FairMQPollerZMQ::Poll(int timeout)
 {
-    zmq_poll(items, fNumItems, timeout);
+    int rc = zmq_poll(items, fNumItems, timeout);
+    if (rc < 0)
+    {
+        LOG(ERROR) << "polling failed, reason: " << zmq_strerror(errno);
+    }
 }
 
 bool FairMQPollerZMQ::CheckInput(int index)
@@ -43,6 +48,14 @@ bool FairMQPollerZMQ::CheckInput(int index)
     return false;
 }
 
+bool FairMQPollerZMQ::CheckOutput(int index)
+{
+    if (items[index].revents & ZMQ_POLLOUT)
+        return true;
+
+    return false;
+}
+
 FairMQPollerZMQ::~FairMQPollerZMQ()
 {
     if (items != NULL)

fairmq/zeromq/FairMQPollerZMQ.h

@@ -29,6 +29,7 @@ class FairMQPollerZMQ : public FairMQPoller
 
     virtual void Poll(int timeout);
     virtual bool CheckInput(int index);
+    virtual bool CheckOutput(int index);
 
     virtual ~FairMQPollerZMQ();
 

fairmq/zeromq/FairMQSocketZMQ.cxx

@@ -40,7 +40,16 @@ FairMQSocketZMQ::FairMQSocketZMQ(const string& type, int num, int numIoThreads)
     rc = zmq_setsockopt(fSocket, ZMQ_IDENTITY, &fId, fId.length());
     if (rc != 0)
     {
-        LOG(ERROR) << "failed setting socket option, reason: " << zmq_strerror(errno);
+        LOG(ERROR) << "failed setting ZMQ_IDENTITY socket option, reason: " << zmq_strerror(errno);
+    }
+
+    // Tell socket to try and send/receive outstanding messages for <linger> milliseconds before terminating.
+    // Default value for ZeroMQ is -1, which is to wait forever.
+    int linger = 500;
+    rc = zmq_setsockopt(fSocket, ZMQ_LINGER, &linger, sizeof(linger));
+    if (rc != 0)
+    {
+        LOG(ERROR) << "failed setting ZMQ_LINGER socket option, reason: " << zmq_strerror(errno);
     }
 
     if (type == "sub")
@@ -48,7 +57,7 @@ FairMQSocketZMQ::FairMQSocketZMQ(const string& type, int num, int numIoThreads)
         rc = zmq_setsockopt(fSocket, ZMQ_SUBSCRIBE, NULL, 0);
         if (rc != 0)
         {
-            LOG(ERROR) << "failed setting socket option, reason: " << zmq_strerror(errno);
+            LOG(ERROR) << "failed setting ZMQ_SUBSCRIBE socket option, reason: " << zmq_strerror(errno);
         }
     }
 
@@ -82,7 +91,7 @@ void FairMQSocketZMQ::Connect(const string& address)
     }
 }
 
-size_t FairMQSocketZMQ::Send(FairMQMessage* msg, const string& flag)
+int FairMQSocketZMQ::Send(FairMQMessage* msg, const string& flag)
 {
     int nbytes = zmq_msg_send(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, GetConstant(flag));
     if (nbytes >= 0)
@@ -93,13 +102,18 @@ size_t FairMQSocketZMQ::Send(FairMQMessage* msg, const string& flag)
     }
     if (zmq_errno() == EAGAIN)
     {
-        return false;
+        return 0;
+    }
+    if (zmq_errno() == ETERM)
+    {
+        LOG(INFO) << "terminating socket #" << fId;
+        return -1;
     }
     LOG(ERROR) << "failed sending on socket #" << fId << ", reason: " << zmq_strerror(errno);
     return nbytes;
 }
 
-size_t FairMQSocketZMQ::Receive(FairMQMessage* msg, const string& flag)
+int FairMQSocketZMQ::Receive(FairMQMessage* msg, const string& flag)
 {
     int nbytes = zmq_msg_recv(static_cast<zmq_msg_t*>(msg->GetMessage()), fSocket, GetConstant(flag));
     if (nbytes >= 0)
@@ -110,7 +124,12 @@ size_t FairMQSocketZMQ::Receive(FairMQMessage* msg, const string& flag)
     }
     if (zmq_errno() == EAGAIN)
     {
-        return false;
+        return 0;
+    }
+    if (zmq_errno() == ETERM)
+    {
+        LOG(INFO) << "terminating socket #" << fId;
+        return -1;
     }
     LOG(ERROR) << "failed receiving on socket #" << fId << ", reason: " << zmq_strerror(errno);
     return nbytes;
@@ -132,6 +151,15 @@ void FairMQSocketZMQ::Close()
     fSocket = NULL;
 }
 
+void FairMQSocketZMQ::Terminate()
+{
+    int rc = zmq_ctx_destroy(fContext->GetContext());
+    if (rc != 0)
+    {
+        LOG(ERROR) << "failed terminating context, reason: " << zmq_strerror(errno);
+    }
+}
+
 void* FairMQSocketZMQ::GetSocket()
 {
     return fSocket;
@@ -204,6 +232,8 @@ int FairMQSocketZMQ::GetConstant(const string& constant)
         return ZMQ_SNDMORE;
     if (constant == "rcv-more")
         return ZMQ_RCVMORE;
+    if (constant == "linger")
+        return ZMQ_LINGER;
 
     return -1;
 }

fairmq/zeromq/FairMQSocketZMQ.h

@@ -32,12 +32,13 @@ class FairMQSocketZMQ : public FairMQSocket
     virtual void Bind(const string& address);
     virtual void Connect(const string& address);
 
-    virtual size_t Send(FairMQMessage* msg, const string& flag="");
-    virtual size_t Receive(FairMQMessage* msg, const string& flag="");
+    virtual int Send(FairMQMessage* msg, const string& flag="");
+    virtual int Receive(FairMQMessage* msg, const string& flag="");
 
     virtual void* GetSocket();
     virtual int GetSocket(int nothing);
     virtual void Close();
+    virtual void Terminate();
 
     virtual void SetOption(const string& option, const void* value, size_t valueSize);
     virtual void GetOption(const string& option, void* value, size_t* valueSize);