Fix race condition in the control plugin

Fixes #32

CMakeLists.txt

@@ -48,7 +48,11 @@ endif()
 
 if(BUILD_NANOMSG_TRANSPORT)
   find_package2(PRIVATE nanomsg VERSION 1.0.0 REQUIRED)
+  find_package2(PRIVATE msgpack VERSION 3.0.0)
+  set(PROJECT_msgpack_VERSION 2.1.5)
+  if(NOT msgpack_FOUND)
     find_package2(PRIVATE msgpack VERSION 2.1.5 REQUIRED)
+  endif()
   set(msgpack_ROOT ${PACKAGE_PREFIX_DIR})
 endif()
 

CONTRIBUTORS

@@ -2,6 +2,7 @@ Aphecetche, Laurent
 Binet, Sebastien
 Eulisse, Giulio
 Karabowicz, Radoslaw
+Kretz, Matthias <kretz@kde.org>
 Krzewicki, Mikolaj 
 Neskovic, Gvozden
 Richter, Matthias

cmake/FairMQLib.cmake

@@ -280,6 +280,7 @@ macro(find_package2 qualifier pkgname)
   set(CMAKE_PREFIX_PATH ${old_CPP})
   unset(old_CPP)
 
+  if(${pkgname}_FOUND)
     if(${qualifier} STREQUAL PRIVATE)
       set(PROJECT_${pkgname}_VERSION ${ARGS_VERSION})
       set(PROJECT_${pkgname}_COMPONENTS ${ARGS_COMPONENTS})
@@ -296,4 +297,5 @@ macro(find_package2 qualifier pkgname)
       set(PROJECT_INTERFACE_${pkgname}_COMPONENTS ${ARGS_COMPONENTS})
       set(PROJECT_INTERFACE_PACKAGE_DEPENDENCIES ${PROJECT_INTERFACE_PACKAGE_DEPENDENCIES} ${pkgname})
     endif()
+  endif()
 endmacro()

fairmq/FairMQChannel.cxx

@@ -499,7 +499,7 @@ bool FairMQChannel::ValidateChannel()
         else
         {
             vector<string> endpoints;
-            boost::algorithm::split(endpoints, fAddress, boost::algorithm::is_any_of(","));
+            boost::algorithm::split(endpoints, fAddress, boost::algorithm::is_any_of(";"));
             for (const auto endpoint : endpoints)
             {
                 string address;

fairmq/FairMQDevice.cxx

@@ -280,7 +280,7 @@ bool FairMQDevice::AttachChannel(FairMQChannel& ch)
             }
             catch (fair::mq::SocketError& se)
             {
-                LOG(ERROR) << se.what();
+                LOG(error) << se.what();
                 return false;
             }
         }
@@ -823,7 +823,7 @@ void FairMQDevice::CreateOwnConfig()
     try {
         fDefaultTransportType = fair::mq::TransportTypes.at(fConfig->GetValue<string>("transport"));
     } catch(const exception& e) {
-        LOG(ERROR) << "invalid transport type provided: " << fConfig->GetValue<string>("transport");
+        LOG(error) << "invalid transport type provided: " << fConfig->GetValue<string>("transport");
     }
 }
 
@@ -867,7 +867,7 @@ void FairMQDevice::SetConfig(FairMQProgOptions& config)
     try {
         fDefaultTransportType = fair::mq::TransportTypes.at(fConfig->GetValue<string>("transport"));
     } catch(const exception& e) {
-        LOG(ERROR) << "invalid transport type provided: " << fConfig->GetValue<string>("transport");
+        LOG(error) << "invalid transport type provided: " << fConfig->GetValue<string>("transport");
     }
     SetTransport(fConfig->GetValue<string>("transport"));
 }

fairmq/options/FairMQProgOptions.cxx

@@ -413,7 +413,7 @@ int FairMQProgOptions::PrintOptions()
 
     for (const auto& o : fUnregisteredOptions)
     {
-        LOG(WARN) << "detected unregistered option: " << o;
+        LOG(debug) << "detected unregistered option: " << o;
     }
 
     stringstream ss;

fairmq/plugins/Control.cxx

@@ -227,7 +227,7 @@ auto Control::WaitForNextState() -> DeviceState
     unique_lock<mutex> lock{fEventsMutex};
     while (fEvents.empty())
     {
-        fNewEvent.wait(lock);
+        fNewEvent.wait_for(lock, chrono::milliseconds(50));
     }
 
     auto result = fEvents.front();

fairmq/plugins/DDS/DDS.cxx

@@ -110,12 +110,12 @@ auto DDS::HandleControl() -> void
         // and propagate addresses of bound channels to DDS.
         FillChannelContainers();
 
-        LOG(DEBUG) << "$DDS_TASK_PATH: " << getenv("DDS_TASK_PATH");
+        LOG(debug) << "$DDS_TASK_PATH: " << getenv("DDS_TASK_PATH");
-        LOG(DEBUG) << "$DDS_GROUP_NAME: " << getenv("DDS_GROUP_NAME");
+        LOG(debug) << "$DDS_GROUP_NAME: " << getenv("DDS_GROUP_NAME");
-        LOG(DEBUG) << "$DDS_COLLECTION_NAME: " << getenv("DDS_COLLECTION_NAME");
+        LOG(debug) << "$DDS_COLLECTION_NAME: " << getenv("DDS_COLLECTION_NAME");
-        LOG(DEBUG) << "$DDS_TASK_NAME: " << getenv("DDS_TASK_NAME");
+        LOG(debug) << "$DDS_TASK_NAME: " << getenv("DDS_TASK_NAME");
-        LOG(DEBUG) << "$DDS_TASK_INDEX: " << getenv("DDS_TASK_INDEX");
+        LOG(debug) << "$DDS_TASK_INDEX: " << getenv("DDS_TASK_INDEX");
-        LOG(DEBUG) << "$DDS_COLLECTION_INDEX: " << getenv("DDS_COLLECTION_INDEX");
+        LOG(debug) << "$DDS_COLLECTION_INDEX: " << getenv("DDS_COLLECTION_INDEX");
 
         // start DDS service - subscriptions will only start firing after this step
         fService.start();

fairmq/shmem/FairMQSocketSHM.cxx

@@ -196,7 +196,7 @@ int FairMQSocketSHM::ReceiveImpl(FairMQMessagePtr& msg, const int flags, const i
             const auto numMsgs = nbytes / sizeof(MetaHeader);
             if (numMsgs > 1)
             {
-                LOG(ERROR) << "Receiving SHM multipart with a single message receive call";
+                LOG(error) << "Receiving SHM multipart with a single message receive call";
             }
 
             assert (numMsgs == 1);
@@ -311,13 +311,13 @@ int64_t FairMQSocketSHM::SendImpl(vector<FairMQMessagePtr>& msgVec, const int fl
         else if (zmq_errno() == ETERM)
         {
             zmq_msg_close (&lZmqMsg);
-            LOG(INFO) << "terminating socket " << fId;
+            LOG(info) << "terminating socket " << fId;
             return -1;
         }
         else
         {
             zmq_msg_close (&lZmqMsg);
-            LOG(ERROR) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
+            LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
             return nbytes;
         }
     }

fairmq/shmem/FairMQTransportFactorySHM.cxx

@@ -69,7 +69,7 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const Fai
     }
     else
     {
-        LOG(warn) << "FairMQProgOptions not available! Using defaults.";
+        LOG(debug) << "FairMQProgOptions not available! Using defaults.";
     }
 
     fSessionName.resize(8, '_'); // shorten the session name, to accommodate for name size limit on some systems (MacOS)

fairmq/zeromq/FairMQTransportFactoryZMQ.cxx

@@ -39,7 +39,7 @@ FairMQTransportFactoryZMQ::FairMQTransportFactoryZMQ(const string& id, const Fai
     }
     else
     {
-        LOG(warn) << "FairMQProgOptions not available! Using defaults.";
+        LOG(debug) << "FairMQProgOptions not available! Using defaults.";
     }
 
     if (zmq_ctx_set(fContext, ZMQ_IO_THREADS, numIoThreads) != 0)