Implement bulk callbacks for unmanaged regions

examples/readout/Readout.h

@@ -37,10 +37,10 @@ class Readout : public FairMQDevice
         fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("rb",
                                                                 0,
                                                                 10000000,
-                                                                [this](void* /*data*/, size_t /*size*/, void* /*hint*/) { // callback to be called when message buffers no longer needed by transport
+                                                                [this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport
-                                                                    --fNumUnackedMsgs;
+                                                                    fNumUnackedMsgs -= blocks.size();
                                                                     if (fMaxIterations > 0) {
-                                                                        LOG(debug) << "Received ack";
+                                                                        LOG(debug) << "Received " << blocks.size() << " acks";
                                                                     }
                                                                 }
                                                                 ));

examples/region/CMakeLists.txt

@@ -32,10 +32,10 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fairmq-start-ex-region.sh.in ${CMAKE_
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/test-ex-region.sh.in ${CMAKE_CURRENT_BINARY_DIR}/test-ex-region.sh)
 
 add_test(NAME Example.Region.zeromq COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-region.sh zeromq)
-set_tests_properties(Example.Region.zeromq PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received ack")
+set_tests_properties(Example.Region.zeromq PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received [0-9*] acks")
 
 add_test(NAME Example.Region.shmem COMMAND ${CMAKE_CURRENT_BINARY_DIR}/test-ex-region.sh shmem)
-set_tests_properties(Example.Region.shmem PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received ack")
+set_tests_properties(Example.Region.shmem PROPERTIES TIMEOUT "30" RUN_SERIAL true PASS_REGULAR_EXPRESSION "Received [0-9*] acks")
 
 # install
 

examples/region/Sampler.cxx

@@ -46,10 +46,10 @@ void Sampler::InitTask()
     fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("data",
                                                              0,
                                                              10000000,
-                                                             [this](void* /*data*/, size_t /*size*/, void* /*hint*/) { // callback to be called when message buffers no longer needed by transport
+                                                             [this](const std::vector<fair::mq::RegionBlock>& blocks) { // callback to be called when message buffers no longer needed by transport
-                                                                 --fNumUnackedMsgs;
+                                                                 fNumUnackedMsgs -= blocks.size();
                                                                  if (fMaxIterations > 0) {
-                                                                     LOG(debug) << "Received ack";
+                                                                    LOG(debug) << "Received " << blocks.size() << " acks";
                                                                  }
                                                              }
                                                              ));

fairmq/FairMQChannel.h

@@ -335,14 +335,10 @@ class FairMQChannel
         return Transport()->NewStaticMessage(data);
     }
 
-    FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr, const std::string& path = "", int flags = 0)
+    template<typename... Args>
+    FairMQUnmanagedRegionPtr NewUnmanagedRegion(Args&&... args)
     {
-        return Transport()->CreateUnmanagedRegion(size, callback, path, flags);
+        return Transport()->CreateUnmanagedRegion(std::forward<Args>(args)...);
-    }
-
-    FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size, const int64_t userFlags, FairMQRegionCallback callback = nullptr, const std::string& path = "", int flags = 0)
-    {
-        return Transport()->CreateUnmanagedRegion(size, userFlags, callback, path, flags);
     }
 
     static constexpr fair::mq::Transport DefaultTransportType = fair::mq::Transport::DEFAULT;

fairmq/FairMQDevice.h

@@ -217,45 +217,17 @@ class FairMQDevice
     }
 
     // creates unamanaged region with the default device transport
-    FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size,
+    template<typename... Args>
-                                                FairMQRegionCallback callback = nullptr,
+    FairMQUnmanagedRegionPtr NewUnmanagedRegion(Args&&... args)
-                                                const std::string& path = "",
-                                                int flags = 0)
     {
-        return Transport()->CreateUnmanagedRegion(size, callback, path, flags);
+        return Transport()->CreateUnmanagedRegion(std::forward<Args>(args)...);
-    }
-
-    // creates unamanaged region with the default device transport
-    FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size,
-                                                const int64_t userFlags,
-                                                FairMQRegionCallback callback = nullptr,
-                                                const std::string& path = "",
-                                                int flags = 0)
-    {
-        return Transport()->CreateUnmanagedRegion(size, userFlags, callback, path, flags);
     }
 
     // creates unmanaged region with the transport of the specified channel
-    FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel,
+    template<typename... Args>
-                                                   int index,
+    FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, Args&&... args)
-                                                   const size_t size,
-                                                   FairMQRegionCallback callback = nullptr,
-                                                   const std::string& path = "",
-                                                   int flags = 0)
     {
-        return GetChannel(channel, index).NewUnmanagedRegion(size, callback, path, flags);
+        return GetChannel(channel, index).NewUnmanagedRegion(std::forward<Args>(args)...);
-    }
-
-    // creates unmanaged region with the transport of the specified channel
-    FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel,
-                                                   int index,
-                                                   const size_t size,
-                                                   const int64_t userFlags,
-                                                   FairMQRegionCallback callback = nullptr,
-                                                   const std::string& path = "",
-                                                   int flags = 0)
-    {
-        return GetChannel(channel, index).NewUnmanagedRegion(size, userFlags, callback, path, flags);
     }
 
     template<typename ...Ts>

fairmq/FairMQTransportFactory.h

@@ -85,6 +85,7 @@ class FairMQTransportFactory
     /// @param flags optional parameter to pass to the underlying transport
     /// @return pointer to UnmanagedRegion
     virtual FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr, const std::string& path = "", int flags = 0) = 0;
+    virtual FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionBulkCallback callback = nullptr, const std::string& path = "", int flags = 0) = 0;
     /// @brief Create new UnmanagedRegion
     /// @param size size of the region
     /// @param userFlags flags to be stored with the region, have no effect on the transport, but can be retrieved from the region by the user
@@ -93,6 +94,7 @@ class FairMQTransportFactory
     /// @param flags optional parameter to pass to the underlying transport
     /// @return pointer to UnmanagedRegion
     virtual FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, const int64_t userFlags, FairMQRegionCallback callback = nullptr, const std::string& path = "", int flags = 0) = 0;
+    virtual FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, const int64_t userFlags, FairMQRegionBulkCallback callback = nullptr, const std::string& path = "", int flags = 0) = 0;
 
     /// @brief Subscribe to region events (creation, destruction, ...)
     /// @param callback the callback that is called when a region event occurs

fairmq/FairMQUnmanagedRegion.h

@@ -13,6 +13,7 @@
 #include <memory> // std::unique_ptr
 #include <functional> // std::function
 #include <ostream> // std::ostream
+#include <vector>
 
 class FairMQTransportFactory;
 
@@ -48,14 +49,25 @@ struct FairMQRegionInfo
     FairMQRegionEvent event;
 };
 
+struct FairMQRegionBlock {
+    void* ptr;
+    size_t size;
+    void* hint;
+
+    FairMQRegionBlock(void* p, size_t s, void* h)
+        : ptr(p), size(s), hint(h)
+    {}
+};
+
 using FairMQRegionCallback = std::function<void(void*, size_t, void*)>;
+using FairMQRegionBulkCallback = std::function<void(const std::vector<FairMQRegionBlock>&)>;
 using FairMQRegionEventCallback = std::function<void(FairMQRegionInfo)>;
 
 class FairMQUnmanagedRegion
 {
   public:
     FairMQUnmanagedRegion() {}
-    FairMQUnmanagedRegion(FairMQTransportFactory* factory): fTransport(factory) {}
+    FairMQUnmanagedRegion(FairMQTransportFactory* factory) : fTransport(factory) {}
 
     virtual void* GetData() const = 0;
     virtual size_t GetSize() const = 0;
@@ -92,9 +104,11 @@ namespace mq
 {
 
 using RegionCallback = FairMQRegionCallback;
+using RegionBulkCallback = FairMQRegionBulkCallback;
 using RegionEventCallback = FairMQRegionEventCallback;
 using RegionEvent = FairMQRegionEvent;
 using RegionInfo = FairMQRegionInfo;
+using RegionBlock = FairMQRegionBlock;
 using UnmanagedRegion = FairMQUnmanagedRegion;
 using UnmanagedRegionPtr = FairMQUnmanagedRegionPtr;
 

fairmq/ofi/TransportFactory.cxx

@@ -90,11 +90,21 @@ auto TransportFactory::CreateUnmanagedRegion(const size_t /*size*/, FairMQRegion
     throw runtime_error{"Not yet implemented UMR."};
 }
 
+auto TransportFactory::CreateUnmanagedRegion(const size_t /*size*/, FairMQRegionBulkCallback /*callback*/, const std::string& /* path = "" */, int /* flags = 0 */) -> UnmanagedRegionPtr
+{
+    throw runtime_error{"Not yet implemented UMR."};
+}
+
 auto TransportFactory::CreateUnmanagedRegion(const size_t /*size*/, const int64_t /*userFlags*/, FairMQRegionCallback /*callback*/, const std::string& /* path = "" */, int /* flags = 0 */) -> UnmanagedRegionPtr
 {
     throw runtime_error{"Not yet implemented UMR."};
 }
 
+auto TransportFactory::CreateUnmanagedRegion(const size_t /*size*/, const int64_t /*userFlags*/, FairMQRegionBulkCallback /*callback*/, const std::string& /* path = "" */, int /* flags = 0 */) -> UnmanagedRegionPtr
+{
+    throw runtime_error{"Not yet implemented UMR."};
+}
+
 auto TransportFactory::GetType() const -> Transport
 {
     return Transport::OFI;

fairmq/ofi/TransportFactory.h

@@ -47,7 +47,9 @@ class TransportFactory final : public FairMQTransportFactory
     auto CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const -> PollerPtr override;
 
     auto CreateUnmanagedRegion(const size_t size, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0) -> UnmanagedRegionPtr override;
+    auto CreateUnmanagedRegion(const size_t size, RegionBulkCallback callback = nullptr, const std::string& path = "", int flags = 0) -> UnmanagedRegionPtr override;
     auto CreateUnmanagedRegion(const size_t size, int64_t userFlags, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0) -> UnmanagedRegionPtr override;
+    auto CreateUnmanagedRegion(const size_t size, int64_t userFlags, RegionBulkCallback callback = nullptr, const std::string& path = "", int flags = 0) -> UnmanagedRegionPtr override;
 
     void SubscribeToRegionEvents(RegionEventCallback /* callback */) override { LOG(error) << "SubscribeToRegionEvents not yet implemented for OFI"; }
     bool SubscribedToRegionEvents() override { LOG(error) << "Region event subscriptions not yet implemented for OFI"; return false; }

fairmq/shmem/Manager.cxx

@@ -101,7 +101,12 @@ void Manager::StartMonitor(const string& id)
     }
 }
 
-pair<bipc::mapped_region*, uint64_t> Manager::CreateRegion(const size_t size, const int64_t userFlags, RegionCallback callback, const string& path /* = "" */, int flags /* = 0 */)
+pair<bipc::mapped_region*, uint64_t> Manager::CreateRegion(const size_t size,
+                                                           const int64_t userFlags,
+                                                           RegionCallback callback,
+                                                           RegionBulkCallback bulkCallback,
+                                                           const string& path /* = "" */,
+                                                           int flags /* = 0 */)
 {
     try {
 
@@ -134,7 +139,7 @@ pair<bipc::mapped_region*, uint64_t> Manager::CreateRegion(const size_t size, co
             // create region info
             fRegionInfos->emplace(id, RegionInfo(path.c_str(), flags, userFlags, fShmVoidAlloc));
 
-            auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, size, false, callback, path, flags));
+            auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, size, false, callback, bulkCallback, path, flags));
             // LOG(debug) << "Created region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
 
             r.first->second->StartReceivingAcks();
@@ -182,7 +187,7 @@ Region* Manager::GetRegionUnsafe(const uint64_t id)
             int flags = regionInfo.fFlags;
             // LOG(debug) << "Located remote region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
 
-            auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, 0, true, nullptr, path, flags));
+            auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, 0, true, nullptr, nullptr, path, flags));
             return r.first->second.get();
         } catch (bie& e) {
             LOG(warn) << "Could not get remote region for id: " << id;

fairmq/shmem/Manager.h

@@ -70,7 +70,12 @@ class Manager
     int IncrementDeviceCounter();
     int DecrementDeviceCounter();
 
-    std::pair<boost::interprocess::mapped_region*, uint64_t> CreateRegion(const size_t size, const int64_t userFlags, RegionCallback callback, const std::string& path = "", int flags = 0);
+    std::pair<boost::interprocess::mapped_region*, uint64_t> CreateRegion(const size_t size,
+                                                                          const int64_t userFlags,
+                                                                          RegionCallback callback,
+                                                                          RegionBulkCallback bulkCallback,
+                                                                          const std::string& path = "",
+                                                                          int flags = 0);
     Region* GetRegion(const uint64_t id);
     Region* GetRegionUnsafe(const uint64_t id);
     void RemoveRegion(const uint64_t id);

fairmq/shmem/Region.cxx

@@ -33,7 +33,7 @@ namespace mq
 namespace shmem
 {
 
-Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, RegionCallback callback, const string& path /* = "" */, int flags /* = 0 */)
+Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, RegionCallback callback, RegionBulkCallback bulkCallback, const string& path, int flags)
     : fManager(manager)
     , fRemote(remote)
     , fStop(false)
@@ -46,6 +46,7 @@ Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, Region
     , fReceiveAcksWorker()
     , fSendAcksWorker()
     , fCallback(callback)
+    , fBulkCallback(bulkCallback)
 {
     if (path != "") {
         fName = string(path + fName);
@@ -110,14 +111,22 @@ void Region::ReceiveAcks()
     unsigned int priority;
     bipc::message_queue::size_type recvdSize;
     unique_ptr<RegionBlock[]> blocks = tools::make_unique<RegionBlock[]>(fAckBunchSize);
+    std::vector<fair::mq::RegionBlock> result;
+    result.reserve(fAckBunchSize);
 
     while (!fStop) { // end thread condition (should exist until region is destroyed)
         auto rcvTill = bpt::microsec_clock::universal_time() + bpt::milliseconds(500);
 
         while (fQueue->timed_receive(blocks.get(), fAckBunchSize * sizeof(RegionBlock), recvdSize, priority, rcvTill)) {
             // LOG(debug) << "received: " << block.fHandle << " " << block.fSize << " " << block.fMessageId;
-            if (fCallback) {
             const auto numBlocks = recvdSize / sizeof(RegionBlock);
+            if (fBulkCallback) {
+                result.clear();
+                for (size_t i = 0; i < numBlocks; i++) {
+                    result.emplace_back(reinterpret_cast<char*>(fRegion.get_address()) + blocks[i].fHandle, blocks[i].fSize, reinterpret_cast<void*>(blocks[i].fHint));
+                }
+                fBulkCallback(result);
+            } else if (fCallback) {
                 for (size_t i = 0; i < numBlocks; i++) {
                     fCallback(reinterpret_cast<char*>(fRegion.get_address()) + blocks[i].fHandle, blocks[i].fSize, reinterpret_cast<void*>(blocks[i].fHint));
                 }
@@ -125,7 +134,7 @@ void Region::ReceiveAcks()
         }
     } // while !fStop
 
-    LOG(debug) << "receive ack worker for " << fName << " leaving.";
+    LOG(debug) << "ReceiveAcks() worker for " << fName << " leaving.";
 }
 
 void Region::ReleaseBlock(const RegionBlock &block)

fairmq/shmem/Region.h

@@ -40,7 +40,7 @@ class Manager;
 
 struct Region
 {
-    Region(Manager& manager, uint64_t id, uint64_t size, bool remote, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0);
+    Region(Manager& manager, uint64_t id, uint64_t size, bool remote, RegionCallback callback, RegionBulkCallback bulkCallback, const std::string& path, int flags);
 
     Region() = delete;
 
@@ -76,6 +76,7 @@ struct Region
     std::thread fReceiveAcksWorker;
     std::thread fSendAcksWorker;
     RegionCallback fCallback;
+    RegionBulkCallback fBulkCallback;
 };
 
 } // namespace shmem

fairmq/shmem/TransportFactory.cxx

@@ -161,12 +161,22 @@ PollerPtr TransportFactory::CreatePoller(const unordered_map<string, vector<Fair
 
 UnmanagedRegionPtr TransportFactory::CreateUnmanagedRegion(const size_t size, RegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */)
 {
-    return tools::make_unique<UnmanagedRegion>(*fManager, size, callback, path, flags, this);
+    return tools::make_unique<UnmanagedRegion>(*fManager, size, 0, callback, nullptr, path, flags, this);
+}
+
+UnmanagedRegionPtr TransportFactory::CreateUnmanagedRegion(const size_t size, RegionBulkCallback bulkCallback, const std::string& path /* = "" */, int flags /* = 0 */)
+{
+    return tools::make_unique<UnmanagedRegion>(*fManager, size, 0, nullptr, bulkCallback, path, flags, this);
 }
 
 UnmanagedRegionPtr TransportFactory::CreateUnmanagedRegion(const size_t size, const int64_t userFlags, RegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */)
 {
-    return tools::make_unique<UnmanagedRegion>(*fManager, size, userFlags, callback, path, flags, this);
+    return tools::make_unique<UnmanagedRegion>(*fManager, size, userFlags, callback, nullptr, path, flags, this);
+}
+
+UnmanagedRegionPtr TransportFactory::CreateUnmanagedRegion(const size_t size, const int64_t userFlags, RegionBulkCallback bulkCallback, const std::string& path /* = "" */, int flags /* = 0 */)
+{
+    return tools::make_unique<UnmanagedRegion>(*fManager, size, userFlags, nullptr, bulkCallback, path, flags, this);
 }
 
 void TransportFactory::SubscribeToRegionEvents(RegionEventCallback callback)

fairmq/shmem/TransportFactory.h

@@ -50,7 +50,9 @@ class TransportFactory final : public fair::mq::TransportFactory
     PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
 
     UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0) override;
+    UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, RegionBulkCallback callback = nullptr, const std::string& path = "", int flags = 0) override;
     UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, int64_t userFlags, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0) override;
+    UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, int64_t userFlags, RegionBulkCallback callback = nullptr, const std::string& path = "", int flags = 0) override;
 
     void SubscribeToRegionEvents(RegionEventCallback callback) override;
     bool SubscribedToRegionEvents() override;

fairmq/shmem/UnmanagedRegion.h

@@ -36,17 +36,20 @@ class UnmanagedRegion final : public fair::mq::UnmanagedRegion
     friend class Socket;
 
   public:
-    UnmanagedRegion(Manager& manager, const size_t size, RegionCallback callback, const std::string& path = "", int flags = 0, FairMQTransportFactory* factory = nullptr)
+    UnmanagedRegion(Manager& manager,
-        : UnmanagedRegion(manager, size, 0, callback, path, flags, factory)
+                    const size_t size,
-    {}
+                    const int64_t userFlags,
-
+                    RegionCallback callback,
-    UnmanagedRegion(Manager& manager, const size_t size, const int64_t userFlags, RegionCallback callback, const std::string& path = "", int flags = 0, FairMQTransportFactory* factory = nullptr)
+                    RegionBulkCallback bulkCallback,
+                    const std::string& path = "",
+                    int flags = 0,
+                    FairMQTransportFactory* factory = nullptr)
         : FairMQUnmanagedRegion(factory)
         , fManager(manager)
         , fRegion(nullptr)
         , fRegionId(0)
     {
-        auto result = fManager.CreateRegion(size, userFlags, callback, path, flags);
+        auto result = fManager.CreateRegion(size, userFlags, callback, bulkCallback, path, flags);
         fRegion = result.first;
         fRegionId = result.second;
     }

fairmq/zeromq/FairMQMessageZMQ.cxx

@@ -81,7 +81,12 @@ FairMQMessageZMQ::FairMQMessageZMQ(FairMQUnmanagedRegionPtr& region, void* data,
 
     memcpy(zmq_msg_data(fMsg.get()), data, size);
     // call region callback
-    static_cast<FairMQUnmanagedRegionZMQ*>(region.get())->fCallback(data, size, hint);
+    auto ptr = static_cast<FairMQUnmanagedRegionZMQ*>(region.get());
+    if (ptr->fBulkCallback) {
+        ptr->fBulkCallback({{data, size, hint}});
+    } else if (ptr->fCallback) {
+        ptr->fCallback(data, size, hint);
+    }
 
     // if (zmq_msg_init_data(fMsg.get(), data, size, [](void*, void*){}, nullptr) != 0)
     // {

fairmq/zeromq/FairMQSocketZMQ.cxx

@@ -84,7 +84,7 @@ bool FairMQSocketZMQ::Bind(const string& address)
             // do not print error in this case, this is handled by FairMQDevice in case no connection could be established after trying a number of random ports from a range.
             return false;
         }
-        LOG(error) << "Failed binding socket " << fId << ", reason: " << zmq_strerror(errno);
+        LOG(error) << "Failed binding socket " << fId << ", address: " << address << ", reason: " << zmq_strerror(errno);
         return false;
     }
 
@@ -97,7 +97,7 @@ bool FairMQSocketZMQ::Connect(const string& address)
 
     if (zmq_connect(fSocket, address.c_str()) != 0)
     {
-        LOG(error) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
+        LOG(error) << "Failed connecting socket " << fId << ", address: " << address << ", reason: " << zmq_strerror(errno);
         return false;
     }
 

fairmq/zeromq/FairMQTransportFactoryZMQ.cxx

@@ -94,19 +94,55 @@ FairMQPollerPtr FairMQTransportFactoryZMQ::CreatePoller(const unordered_map<stri
     return unique_ptr<FairMQPoller>(new FairMQPollerZMQ(channelsMap, channelList));
 }
 
-FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback, const string& path /* = "" */, int flags /* = 0 */)
+FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(
+    const size_t size,
+    FairMQRegionCallback callback,
+    const string& path /* = "" */,
+    int flags /* = 0 */)
 {
-    return CreateUnmanagedRegion(size, 0, callback, path, flags);
+    return CreateUnmanagedRegion(size, 0, callback, nullptr, path, flags);
+}
+FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(
+    const size_t size,
+    FairMQRegionBulkCallback bulkCallback,
+    const string& path /* = "" */,
+    int flags /* = 0 */)
+{
+    return CreateUnmanagedRegion(size, 0, nullptr, bulkCallback, path, flags);
+}
+FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(
+    const size_t size,
+    const int64_t userFlags,
+    FairMQRegionCallback callback,
+    const string& path /* = "" */,
+    int flags /* = 0 */)
+{
+    return CreateUnmanagedRegion(size, userFlags, callback, nullptr, path, flags);
+}
+FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(
+    const size_t size,
+    const int64_t userFlags,
+    FairMQRegionBulkCallback bulkCallback,
+    const string& path /* = "" */,
+    int flags /* = 0 */)
+{
+    return CreateUnmanagedRegion(size, userFlags, nullptr, bulkCallback, path, flags);
 }
 
-FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(const size_t size, const int64_t userFlags, FairMQRegionCallback callback, const string& path /* = "" */, int flags /* = 0 */)
+FairMQUnmanagedRegionPtr FairMQTransportFactoryZMQ::CreateUnmanagedRegion(
+    const size_t size,
+    const int64_t userFlags,
+    FairMQRegionCallback callback,
+    FairMQRegionBulkCallback bulkCallback,
+    const string& path /* = "" */,
+    int flags /* = 0 */)
 {
     unique_ptr<FairMQUnmanagedRegion> ptr = nullptr;
     {
         lock_guard<mutex> lock(fMtx);
 
         ++fRegionCounter;
-        ptr = unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionZMQ(fRegionCounter, size, userFlags, callback, path, flags, this));
+        ptr = unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionZMQ(fRegionCounter, size, userFlags, callback, bulkCallback, path, flags, this));
         auto zPtr = static_cast<FairMQUnmanagedRegionZMQ*>(ptr.get());
         fRegionInfos.emplace_back(zPtr->GetId(), zPtr->GetData(), zPtr->GetSize(), zPtr->GetUserFlags(), fair::mq::RegionEvent::created);
         fRegionEvents.emplace(zPtr->GetId(), zPtr->GetData(), zPtr->GetSize(), zPtr->GetUserFlags(), fair::mq::RegionEvent::created);

fairmq/zeromq/FairMQTransportFactoryZMQ.h

@@ -49,7 +49,10 @@ class FairMQTransportFactoryZMQ final : public FairMQTransportFactory
     FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
 
     FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback, const std::string& path = "", int flags = 0) override;
+    FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionBulkCallback bulkCallback, const std::string& path = "", int flags = 0) override;
     FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, const int64_t userFlags, FairMQRegionCallback callback, const std::string& path = "", int flags = 0) override;
+    FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, const int64_t userFlags, FairMQRegionBulkCallback bulkCallback, const std::string& path = "", int flags = 0) override;
+    FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, const int64_t userFlags, FairMQRegionCallback callback, FairMQRegionBulkCallback bulkCallback, const std::string& path = "", int flags = 0);
 
     void SubscribeToRegionEvents(FairMQRegionEventCallback callback) override;
     bool SubscribedToRegionEvents() override;

fairmq/zeromq/FairMQUnmanagedRegionZMQ.cxx

@@ -10,13 +10,21 @@
 #include "FairMQTransportFactoryZMQ.h"
 #include "FairMQLogger.h"
 
-FairMQUnmanagedRegionZMQ::FairMQUnmanagedRegionZMQ(uint64_t id, const size_t size, int64_t userFlags, FairMQRegionCallback callback, const std::string& /* path = "" */, int /* flags = 0 */, FairMQTransportFactory* factory /* = nullptr */)
+FairMQUnmanagedRegionZMQ::FairMQUnmanagedRegionZMQ(uint64_t id,
+                                                   const size_t size,
+                                                   int64_t userFlags,
+                                                   FairMQRegionCallback callback,
+                                                   FairMQRegionBulkCallback bulkCallback,
+                                                   const std::string& /* path = "" */,
+                                                   int /* flags = 0 */,
+                                                   FairMQTransportFactory* factory /* = nullptr */)
     : FairMQUnmanagedRegion(factory)
     , fId(id)
     , fBuffer(malloc(size))
     , fSize(size)
     , fUserFlags(userFlags)
     , fCallback(callback)
+    , fBulkCallback(bulkCallback)
 {}
 
 void* FairMQUnmanagedRegionZMQ::GetData() const

fairmq/zeromq/FairMQUnmanagedRegionZMQ.h

@@ -21,7 +21,7 @@ class FairMQUnmanagedRegionZMQ final : public FairMQUnmanagedRegion
     friend class FairMQMessageZMQ;
 
   public:
-    FairMQUnmanagedRegionZMQ(uint64_t id, const size_t size, int64_t userFlags, FairMQRegionCallback callback, const std::string& /* path = "" */, int /* flags = 0 */, FairMQTransportFactory* factory = nullptr);
+    FairMQUnmanagedRegionZMQ(uint64_t id, const size_t size, int64_t userFlags, FairMQRegionCallback callback, FairMQRegionBulkCallback bulkCallback, const std::string& /* path = "" */, int /* flags = 0 */, FairMQTransportFactory* factory = nullptr);
 
     FairMQUnmanagedRegionZMQ(const FairMQUnmanagedRegionZMQ&) = delete;
     FairMQUnmanagedRegionZMQ operator=(const FairMQUnmanagedRegionZMQ&) = delete;
@@ -39,6 +39,7 @@ class FairMQUnmanagedRegionZMQ final : public FairMQUnmanagedRegion
     size_t fSize;
     int64_t fUserFlags;
     FairMQRegionCallback fCallback;
+    FairMQRegionBulkCallback fBulkCallback;
 };
 
 #endif /* FAIRMQUNMANAGEDREGIONZMQ_H_ */

test/region/_region.cxx

@@ -32,7 +32,7 @@ void RegionEventSubscriptions(const string& transport)
     constexpr int size1 = 1000000;
     constexpr int size2 = 5000000;
     constexpr int64_t userFlags = 12345;
-    fair::mq::tools::SharedSemaphore blocker;
+    fair::mq::tools::Semaphore blocker;
 
     {
         auto region1 = factory->CreateUnmanagedRegion(size1, [](void*, size_t, void*) {});
@@ -90,6 +90,72 @@ void RegionEventSubscriptions(const string& transport)
     ASSERT_EQ(factory->SubscribedToRegionEvents(), false);
 }
 
+void RegionCallbacks(const string& transport, const string& _address)
+{
+    size_t session(fair::mq::tools::UuidHash());
+    std::string address(fair::mq::tools::ToString(_address, "_", transport));
+
+    fair::mq::ProgOptions config;
+    config.SetProperty<string>("session", to_string(session));
+
+    auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config);
+
+    unique_ptr<int> intPtr1 = fair::mq::tools::make_unique<int>(42);
+    unique_ptr<int> intPtr2 = fair::mq::tools::make_unique<int>(43);
+    fair::mq::tools::Semaphore blocker;
+
+    FairMQChannel push("Push", "push", factory);
+    push.Bind(address);
+
+    FairMQChannel pull("Pull", "pull", factory);
+    pull.Connect(address);
+
+    void* ptr1 = nullptr;
+    size_t size1 = 100;
+    void* ptr2 = nullptr;
+    size_t size2 = 200;
+
+    auto region1 = factory->CreateUnmanagedRegion(2000000, [&](void* ptr, size_t size, void* hint) {
+        ASSERT_EQ(ptr, ptr1);
+        ASSERT_EQ(size, size1);
+        ASSERT_EQ(hint, intPtr1.get());
+        ASSERT_EQ(*static_cast<int*>(hint), 42);
+        blocker.Signal();
+    });
+    ptr1 = region1->GetData();
+
+    auto region2 = factory->CreateUnmanagedRegion(3000000, [&](const std::vector<fair::mq::RegionBlock>& blocks) {
+        ASSERT_EQ(blocks.size(), 1);
+        ASSERT_EQ(blocks.at(0).ptr, ptr2);
+        ASSERT_EQ(blocks.at(0).size, size2);
+        ASSERT_EQ(blocks.at(0).hint, intPtr2.get());
+        ASSERT_EQ(*static_cast<int*>(blocks.at(0).hint), 43);
+        blocker.Signal();
+    });
+    ptr2 = region2->GetData();
+
+
+    {
+        FairMQMessagePtr msg1out(push.NewMessage(region1, ptr1, size1, intPtr1.get()));
+        FairMQMessagePtr msg2out(push.NewMessage(region2, ptr2, size2, intPtr2.get()));
+        ASSERT_EQ(push.Send(msg1out), size1);
+        ASSERT_EQ(push.Send(msg2out), size2);
+    }
+
+    {
+        FairMQMessagePtr msg1in(pull.NewMessage());
+        FairMQMessagePtr msg2in(pull.NewMessage());
+        ASSERT_EQ(pull.Receive(msg1in), size1);
+        ASSERT_EQ(pull.Receive(msg2in), size2);
+    }
+
+    LOG(info) << "waiting for blockers...";
+    blocker.Wait();
+    LOG(info) << "1 done.";
+    blocker.Wait();
+    LOG(info) << "2 done.";
+}
+
 TEST(EventSubscriptions, zeromq)
 {
     RegionEventSubscriptions("zeromq");
@@ -100,4 +166,14 @@ TEST(EventSubscriptions, shmem)
     RegionEventSubscriptions("shmem");
 }
 
+TEST(Callbacks, zeromq)
+{
+    RegionCallbacks("zeromq", "ipc://test_region_callbacks");
+}
+
+TEST(Callbacks, shmem)
+{
+    RegionCallbacks("shmem", "ipc://test_region_callbacks");
+}
+
 } // namespace

test/transport/_options.cxx

@@ -25,7 +25,7 @@ namespace
 
 using namespace std;
 
-void CheckOldOptionInterface(FairMQChannel& channel, const string& option, const string& transport)
+void CheckOldOptionInterface(FairMQChannel& channel, const string& option)
 {
     int value = 500;
     channel.GetSocket().SetOption(option, &value, sizeof(value));
@@ -44,11 +44,11 @@ void RunOptionsTest(const string& transport)
     auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config);
     FairMQChannel channel("Push", "push", factory);
 
-    CheckOldOptionInterface(channel, "linger", transport);
+    CheckOldOptionInterface(channel, "linger");
-    CheckOldOptionInterface(channel, "snd-hwm", transport);
+    CheckOldOptionInterface(channel, "snd-hwm");
-    CheckOldOptionInterface(channel, "rcv-hwm", transport);
+    CheckOldOptionInterface(channel, "rcv-hwm");
-    CheckOldOptionInterface(channel, "snd-size", transport);
+    CheckOldOptionInterface(channel, "snd-size");
-    CheckOldOptionInterface(channel, "rcv-size", transport);
+    CheckOldOptionInterface(channel, "rcv-size");
 
     channel.GetSocket().SetLinger(300);
     ASSERT_EQ(channel.GetSocket().GetLinger(), 300);