FairMQ: Fix missing region acks if receiver doesn't call OnData & clean includes.

fairmq/shmem/FairMQMessageSHM.cxx

@@ -5,15 +5,16 @@
  *              GNU Lesser General Public Licence (LGPL) version 3,             *
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
-#include <string>
+#include <fairmq/shmem/Common.h>
-#include <cstdlib>
+#include <fairmq/shmem/Region.h>
-
-#include <boost/date_time/posix_time/posix_time.hpp>
 
 #include "FairMQMessageSHM.h"
 #include "FairMQUnmanagedRegionSHM.h"
 #include "FairMQLogger.h"
-#include "Common.h"
+
+#include <boost/date_time/posix_time/posix_time.hpp>
+
+#include <cstdlib>
 
 using namespace std;
 using namespace fair::mq::shmem;
@@ -30,6 +31,7 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager)
     , fQueued(false)
     , fMetaCreated(false)
     , fRegionId(0)
+    , fRegionPtr(nullptr)
     , fHandle(-1)
     , fSize(0)
     , fLocalPtr(nullptr)
@@ -47,6 +49,7 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager, const size_t size)
     , fQueued(false)
     , fMetaCreated(false)
     , fRegionId(0)
+    , fRegionPtr(nullptr)
     , fHandle(-1)
     , fSize(0)
     , fLocalPtr(nullptr)
@@ -60,6 +63,7 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager, void* data, const size_t si
     , fQueued(false)
     , fMetaCreated(false)
     , fRegionId(0)
+    , fRegionPtr(nullptr)
     , fHandle(-1)
     , fSize(0)
     , fLocalPtr(nullptr)
@@ -84,6 +88,7 @@ FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQUnmanagedRegionPtr& r
     , fQueued(false)
     , fMetaCreated(false)
     , fRegionId(static_cast<FairMQUnmanagedRegionSHM*>(region.get())->fRegionId)
+    , fRegionPtr(nullptr)
     , fHandle(-1)
     , fSize(size)
     , fLocalPtr(data)
@@ -209,10 +214,10 @@ void* FairMQMessageSHM::GetData()
         }
         else
         {
-            boost::interprocess::mapped_region* region = fManager.GetRemoteRegion(fRegionId);
+            fRegionPtr = fManager.GetRemoteRegion(fRegionId);
-            if (region)
+            if (fRegionPtr)
             {
-                fLocalPtr = reinterpret_cast<char*>(region->get_address()) + fHandle;
+                fLocalPtr = reinterpret_cast<char*>(fRegionPtr->fRegion.get_address()) + fHandle;
             }
             else
             {
@@ -290,10 +295,13 @@ void FairMQMessageSHM::CloseMessage()
             do
             {
                 auto sndTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(200);
-                bipc::message_queue* q = fManager.GetRegionQueue(fRegionId);
+                if (!fRegionPtr)
-                if (q)
                 {
-                    if (q->timed_send(&block, sizeof(RegionBlock), 0, sndTill))
+                    fRegionPtr = fManager.GetRemoteRegion(fRegionId);
+                }
+                if (fRegionPtr)
+                {
+                    if (fRegionPtr->fQueue->timed_send(&block, sizeof(RegionBlock), 0, sndTill))
                     {
                         success = true;
                     }

fairmq/shmem/FairMQMessageSHM.h

@@ -8,17 +8,17 @@
 #ifndef FAIRMQMESSAGESHM_H_
 #define FAIRMQMESSAGESHM_H_
 
-#include <cstddef> // size_t
+#include <fairmq/shmem/Manager.h>
-#include <string>
+
-#include <atomic>
+#include "FairMQMessage.h"
+#include "FairMQUnmanagedRegion.h"
 
 #include <zmq.h>
 
 #include <boost/interprocess/mapped_region.hpp>
 
-#include "FairMQMessage.h"
+#include <cstddef> // size_t
-#include "FairMQUnmanagedRegion.h"
+#include <atomic>
-#include "Manager.h"
 
 class FairMQMessageSHM : public FairMQMessage
 {
@@ -61,6 +61,7 @@ class FairMQMessageSHM : public FairMQMessage
     static std::atomic<bool> fInterrupted;
     static FairMQ::Transport fTransportType;
     uint64_t fRegionId;
+    fair::mq::shmem::Region* fRegionPtr;
     boost::interprocess::managed_shared_memory::handle_t fHandle;
     size_t fSize;
     void* fLocalPtr;

fairmq/shmem/FairMQPollerSHM.cxx

@@ -12,11 +12,11 @@
  * @author A. Rybalchenko
  */
 
-#include <zmq.h>
-
 #include "FairMQPollerSHM.h"
 #include "FairMQLogger.h"
 
+#include <zmq.h>
+
 using namespace std;
 
 FairMQPollerSHM::FairMQPollerSHM(const vector<FairMQChannel>& channels)

fairmq/shmem/FairMQPollerSHM.h

@@ -10,7 +10,6 @@
 
 #include <vector>
 #include <unordered_map>
-#include <initializer_list>
 
 #include <zmq.h>
 

fairmq/shmem/FairMQSocketSHM.cxx

@@ -5,15 +5,14 @@
  *              GNU Lesser General Public Licence (LGPL) version 3,             *  
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
-#include <sstream>
+#include <fairmq/shmem/Common.h>
-
-#include <zmq.h>
 
 #include "FairMQSocketSHM.h"
 #include "FairMQMessageSHM.h"
 #include "FairMQUnmanagedRegionSHM.h"
 #include "FairMQLogger.h"
-#include "Common.h"
+
+#include <zmq.h>
 
 using namespace std;
 using namespace fair::mq::shmem;

fairmq/shmem/FairMQSocketSHM.h

@@ -8,13 +8,13 @@
 #ifndef FAIRMQSOCKETSHM_H_
 #define FAIRMQSOCKETSHM_H_
 
-#include <atomic>
-
-#include <memory> // unique_ptr
-
 #include "FairMQSocket.h"
 #include "FairMQMessage.h"
-#include "Manager.h"
+
+#include <fairmq/shmem/Manager.h>
+
+#include <atomic>
+#include <memory> // unique_ptr
 
 class FairMQSocketSHM : public FairMQSocket
 {

fairmq/shmem/FairMQTransportFactorySHM.h

@@ -9,12 +9,13 @@
 #ifndef FAIRMQTRANSPORTFACTORYSHM_H_
 #define FAIRMQTRANSPORTFACTORYSHM_H_
 
+#include <fairmq/shmem/Manager.h>
+#include <fairmq/shmem/Common.h>
+
 #include "FairMQTransportFactory.h"
-#include "Manager.h"
 #include "FairMQMessageSHM.h"
 #include "FairMQSocketSHM.h"
 #include "FairMQPollerSHM.h"
-#include "Common.h"
 #include "FairMQUnmanagedRegionSHM.h"
 #include <options/FairMQProgOptions.h>
 

fairmq/shmem/FairMQUnmanagedRegionSHM.cxx

@@ -6,8 +6,9 @@
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
 
+#include <fairmq/shmem/Common.h>
+
 #include "FairMQUnmanagedRegionSHM.h"
-#include "Common.h"
 
 using namespace std;
 using namespace fair::mq::shmem;

fairmq/shmem/FairMQUnmanagedRegionSHM.h

@@ -9,18 +9,15 @@
 #ifndef FAIRMQUNMANAGEDREGIONSHM_H_
 #define FAIRMQUNMANAGEDREGIONSHM_H_
 
+#include <fairmq/shmem/Manager.h>
+
 #include "FairMQUnmanagedRegion.h"
 #include "FairMQLogger.h"
-#include "Manager.h"
 
 #include <boost/interprocess/shared_memory_object.hpp>
 #include <boost/interprocess/mapped_region.hpp>
 
 #include <cstddef> // size_t
-#include <atomic>
-#include <memory>
-#include <string>
-#include <unordered_map>
 
 class FairMQUnmanagedRegionSHM : public FairMQUnmanagedRegion
 {

fairmq/shmem/Manager.cxx

@@ -6,8 +6,8 @@
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
 
- #include "Manager.h"
+#include <fairmq/shmem/Manager.h>
- #include "Common.h"
+#include <fairmq/shmem/Common.h>
 
 namespace fair
 {
@@ -71,20 +71,20 @@ bipc::mapped_region* Manager::CreateRegion(const size_t size, const uint64_t id,
     }
 }
 
-bipc::mapped_region* Manager::GetRemoteRegion(const uint64_t id)
+Region* Manager::GetRemoteRegion(const uint64_t id)
 {
     // remote region could actually be a local one if a message originates from this device (has been sent out and returned)
     auto it = fRegions.find(id);
     if (it != fRegions.end())
     {
-        return &(it->second.fRegion);
+        return &(it->second);
     }
     else
     {
         try
         {
             auto r = fRegions.emplace(id, Region{*this, id, 0, true, nullptr});
-            return &(r.first->second.fRegion);
+            return &(r.first->second);
         }
         catch (bipc::interprocess_exception& e)
         {
@@ -100,18 +100,6 @@ void Manager::RemoveRegion(const uint64_t id)
     fRegions.erase(id);
 }
 
-bipc::message_queue* Manager::GetRegionQueue(const uint64_t id)
-{
-    try
-    {
-        return fRegions.at(id).fQueue.get();
-    }
-    catch (out_of_range& oor)
-    {
-        return nullptr;
-    }
-}
-
 void Manager::RemoveSegment()
 {
     if (bipc::shared_memory_object::remove(fSegmentName.c_str()))

fairmq/shmem/Manager.h

@@ -15,23 +15,18 @@
 #ifndef FAIR_MQ_SHMEM_MANAGER_H_
 #define FAIR_MQ_SHMEM_MANAGER_H_
 
+#include <fairmq/Tools.h>
+#include <fairmq/shmem/Region.h>
+#include <fairmq/shmem/Common.h>
+
 #include "FairMQLogger.h"
 #include "FairMQMessage.h"
-#include "fairmq/Tools.h"
-#include "Region.h"
-#include "Common.h"
 
 #include <boost/interprocess/managed_shared_memory.hpp>
 #include <boost/interprocess/ipc/message_queue.hpp>
 
-#include <thread>
-#include <queue>
 #include <string>
-#include <vector>
-#include <functional>
 #include <unordered_map>
-#include <mutex>
-#include <condition_variable>
 
 namespace fair
 {
@@ -58,11 +53,9 @@ class Manager
     void Resume();
 
     boost::interprocess::mapped_region* CreateRegion(const size_t size, const uint64_t id, FairMQRegionCallback callback);
-    boost::interprocess::mapped_region* GetRemoteRegion(const uint64_t id);
+    Region* GetRemoteRegion(const uint64_t id);
     void RemoveRegion(const uint64_t id);
 
-    boost::interprocess::message_queue* GetRegionQueue(const uint64_t id);
-
     void RemoveSegment();
 
     boost::interprocess::managed_shared_memory& ManagementSegment();

fairmq/shmem/Monitor.cxx

@@ -6,8 +6,8 @@
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
 
-#include "Monitor.h"
+#include <fairmq/shmem/Monitor.h>
-#include "Common.h"
+#include <fairmq/shmem/Common.h>
 
 #include <boost/interprocess/managed_shared_memory.hpp>
 #include <boost/interprocess/containers/vector.hpp>

fairmq/shmem/Region.cxx

@@ -6,14 +6,12 @@
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
 
-#include "Region.h"
+#include <fairmq/shmem/Region.h>
-#include "Common.h"
+#include <fairmq/shmem/Common.h>
-#include "Manager.h"
+#include <fairmq/shmem/Manager.h>
 
 #include <boost/date_time/posix_time/posix_time.hpp>
 
-#include <chrono>
-
 namespace fair
 {
 namespace mq

fairmq/shmem/Region.h

@@ -17,13 +17,13 @@
 
 #include "FairMQLogger.h"
 #include "FairMQUnmanagedRegion.h"
-#include "fairmq/Tools.h"
+
+#include <fairmq/Tools.h>
 
 #include <boost/interprocess/managed_shared_memory.hpp>
 #include <boost/interprocess/ipc/message_queue.hpp>
 
 #include <thread>
-#include <queue>
 #include <unordered_map>
 
 namespace fair

fairmq/shmem/runMonitor.cxx

@@ -5,7 +5,7 @@
  *              GNU Lesser General Public Licence (LGPL) version 3,             *
  *                  copied verbatim in the file "LICENSE"                       *
  ********************************************************************************/
-#include "Monitor.h"
+#include <fairmq/shmem/Monitor.h>
 
 #include <boost/program_options.hpp>