Polymorphic allocators in DPL

DataFormats/Headers/include/Headers/Stack.h

@@ -33,25 +33,19 @@ namespace header
 ///   - returns a Stack ready to be shipped.
 struct Stack {
 
- private:
-  static void freefn(void* data, void* hint)
-  {
-    boost::container::pmr::memory_resource* resource = static_cast<boost::container::pmr::memory_resource*>(hint);
-    resource->deallocate(data, 0, 0);
-  }
+  using memory_resource = o2::pmr::memory_resource;
 
+ private:
   struct freeobj {
-    freeobj() {}
-    freeobj(boost::container::pmr::memory_resource* mr) : resource(mr) {}
-
-    boost::container::pmr::memory_resource* resource{ nullptr };
-    void operator()(o2::byte* ptr) { Stack::freefn(ptr, resource); }
+    freeobj(memory_resource* mr) : resource(mr) {}
+    memory_resource* resource{ nullptr };
+    void operator()(o2::byte* ptr) { resource->deallocate(ptr, 0, 0); }
   };
 
  public:
   using allocator_type = boost::container::pmr::polymorphic_allocator<o2::byte>;
   using value_type = o2::byte;
-  using BufferType = std::unique_ptr<value_type[], freeobj>;
+  using BufferType = std::unique_ptr<value_type[], freeobj>; //this gives us proper default move semantics for free
 
   Stack() = default;
   Stack(Stack&&) = default;
@@ -64,8 +58,6 @@ struct Stack {
   allocator_type get_allocator() const { return allocator; }
 
   //
-  boost::container::pmr::memory_resource* getFreefnHint() const noexcept { return allocator.resource(); }
-  static auto getFreefn() noexcept { return &freefn; }
 
   /// The magic constructors: take arbitrary number of headers and serialize them
   /// into the buffer buffer allocated by the specified polymorphic allocator. By default
@@ -86,15 +78,15 @@ struct Stack {
     : allocator{ allocatorArg },
       bufferSize{ calculateSize(std::forward<Headers>(headers)...) },
       buffer{ static_cast<o2::byte*>(allocator.resource()->allocate(bufferSize, alignof(std::max_align_t))),
-              freeobj(getFreefnHint()) }
+              freeobj(allocator.resource()) }
   {
     inject(buffer.get(), std::forward<Headers>(headers)...);
   }
 
  private:
   allocator_type allocator{ boost::container::pmr::new_delete_resource() };
   size_t bufferSize{ 0 };
-  BufferType buffer{ nullptr, freeobj{ getFreefnHint() } };
+  BufferType buffer{ nullptr, freeobj{ allocator.resource() } };
 
   template <typename T, typename... Args>
   static size_t calculateSize(T&& h, Args&&... args) noexcept

DataFormats/MemoryResources/include/MemoryResources/MemoryResources.h

@@ -41,17 +41,17 @@
 #include <FairMQTransportFactory.h>
 #include <fairmq/MemoryResources.h>
 #include <fairmq/MemoryResourceTools.h>
+#include "Types.h"
 
 namespace o2
 {
 
-using byte = unsigned char;
-
-namespace memory_resource
+namespace pmr
 {
 
 using FairMQMemoryResource = fair::mq::FairMQMemoryResource;
 using ChannelResource = fair::mq::ChannelResource;
+using namespace fair::mq::pmr;
 
 template <typename ContainerT>
 FairMQMessagePtr getMessage(ContainerT&& container, FairMQMemoryResource* targetResource = nullptr)
@@ -209,6 +209,8 @@ class OwningMessageSpectatorAllocator
 
 using ByteSpectatorAllocator = SpectatorAllocator<o2::byte>;
 using BytePmrAllocator = boost::container::pmr::polymorphic_allocator<o2::byte>;
+template <class T>
+using vector = std::vector<T, o2::pmr::polymorphic_allocator<T>>;
 
 //__________________________________________________________________________________________________
 /// Return a std::vector spanned over the contents of the message, takes ownership of the message
@@ -227,7 +229,11 @@ inline static ChannelResource* getTransportAllocator(FairMQTransportFactory* fac
   return factory->GetMemoryResource();
 }
 
-}; //namespace memory_resource
+}; //namespace pmr
+
+template <class T>
+using vector = std::vector<T, o2::pmr::polymorphic_allocator<T>>;
+
 }; //namespace o2
 
 #endif

DataFormats/MemoryResources/test/testMemoryResources.cxx

@@ -19,7 +19,7 @@
 
 namespace o2
 {
-namespace memory_resource
+namespace pmr
 {
 auto factoryZMQ = FairMQTransportFactory::CreateTransportFactory("zeromq");
 auto factorySHM = FairMQTransportFactory::CreateTransportFactory("shmem");
@@ -103,7 +103,7 @@ BOOST_AUTO_TEST_CASE(getMessage_test)
     v.emplace_back(2);
     v.emplace_back(3);
     void* vectorBeginPtr = &v[0];
-    message = o2::memory_resource::getMessage(std::move(v));
+    message = o2::pmr::getMessage(std::move(v));
     BOOST_CHECK(message != nullptr);
     BOOST_CHECK(message->GetData() == vectorBeginPtr);
   }
@@ -118,7 +118,7 @@ BOOST_AUTO_TEST_CASE(getMessage_test)
     v.emplace_back(5);
     v.emplace_back(6);
     void* vectorBeginPtr = &v[0];
-    message = o2::memory_resource::getMessage(std::move(v), allocSHM);
+    message = o2::pmr::getMessage(std::move(v), allocSHM);
     BOOST_CHECK(message != nullptr);
     BOOST_CHECK(message->GetData() != vectorBeginPtr);
   }
@@ -146,7 +146,7 @@ BOOST_AUTO_TEST_CASE(adoptVector_test)
   BOOST_CHECK(adoptedOwner[1].i == 2);
   BOOST_CHECK(adoptedOwner[2].i == 1);
 
-  auto reclaimedMessage = o2::memory_resource::getMessage(std::move(adoptedOwner));
+  auto reclaimedMessage = o2::pmr::getMessage(std::move(adoptedOwner));
   BOOST_CHECK(reclaimedMessage.get() == messageAddr);
   BOOST_CHECK(adoptedOwner.size() == 0);
 

Framework/Core/CMakeLists.txt

@@ -157,6 +157,7 @@ set(HEADERS
       include/Framework/FairMQResizableBuffer.h
       include/Framework/Metric2DViewIndex.h
       include/Framework/RawBufferContext.h
+      include/Framework/observer_ptr.h
       src/ComputingResource.h
       src/DDSConfigHelpers.h
       src/O2ControlHelpers.h

Framework/Core/include/Framework/DataAllocator.h

@@ -41,7 +41,7 @@
 #include <cstddef>
 
 // Do not change this for a full inclusion of FairMQDevice.
-class FairMQDevice;
+#include <FairMQDevice.h>
 class FairMQMessage;
 
 namespace o2
@@ -465,6 +465,42 @@ class DataAllocator
     return adopt(getOutputByBind(std::move(ref)), obj);
   }
 
+  //make a stl (pmr) vector
+  template <typename T, typename... Args>
+  o2::vector<T> makeVector(const Output& spec, Args&&... args)
+  {
+    std::string channel = matchDataHeader(spec, mTimingInfo->timeslice);
+    auto context = mContextRegistry->get<MessageContext>();
+    o2::pmr::FairMQMemoryResource* targetResource = context->proxy().getDevice()->GetChannel(channel).Transport()->GetMemoryResource();
+    return o2::vector<T>{ targetResource, std::forward<Args>(args)... };
+  }
+
+  //adopt container (if PMR is used with the appropriate memory resource iin container it is ZERO-copy)
+  template <typename ContainerT>
+  void adoptContainer(const Output& spec, ContainerT& container) = delete; //only bind to moved-from containers
+  template <typename ContainerT>
+  void adoptContainer(const Output& spec, ContainerT&& container)
+  {
+    // Find a matching channel, create a new message for it and put it in the
+    // queue to be sent at the end of the processing
+    std::string channel = matchDataHeader(spec, mTimingInfo->timeslice);
+
+    FairMQParts parts;
+
+    auto context = mContextRegistry->get<MessageContext>();
+    o2::pmr::FairMQMemoryResource* targetResource = context->proxy().getDevice()->GetChannel(channel).Transport()->GetMemoryResource();
+    FairMQMessagePtr payloadMessage = o2::pmr::getMessage(std::forward<ContainerT>(container), targetResource);
+
+    FairMQMessagePtr headerMessage = headerMessageFromOutput(spec, channel,                        //
+                                                             o2::header::gSerializationMethodNone, //
+                                                             payloadMessage->GetSize()             //
+    );
+
+    parts.AddPart(std::move(headerMessage));
+    parts.AddPart(std::move(payloadMessage));
+    context->addPart(std::move(parts), channel);
+  }
+
   /// snapshot object and route to output specified by OutputRef
   /// Framework makes a (serialized) copy of object content.
   ///

Framework/Core/include/Framework/observer_ptr.h

@@ -0,0 +1,74 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#ifndef FRAMEWORK_OBSERVER_PTR_H
+#define FRAMEWORK_OBSERVER_PTR_H
+
+#include <cstddef>
+#include <type_traits>
+
+namespace o2
+{
+
+template <typename W>
+class observer_ptr
+{
+ public:
+  using element_type = W;
+
+  constexpr observer_ptr() noexcept = default;
+  constexpr observer_ptr(std::nullptr_t) noexcept {}
+  explicit observer_ptr(element_type* ptr) noexcept : mptr{ ptr } {}
+  template <typename W2, typename std::enable_if<std::is_convertible<element_type*, W2*>::value, int>::type = 1>
+  observer_ptr(observer_ptr<W2> other) noexcept : mptr{ other.get() }
+  {
+  }
+  observer_ptr(const observer_ptr& other) = default;
+  observer_ptr(observer_ptr&& other) = default;
+
+  constexpr element_type* release() noexcept
+  {
+    auto tmp = *this;
+    this->reset();
+    return tmp.get();
+  }
+  constexpr void reset(element_type* p = nullptr) noexcept { *this = p; }
+  constexpr void swap(observer_ptr& other) noexcept
+  {
+    observer_ptr<element_type> tmp(*this);
+    *this = other;
+    other = tmp;
+  };
+  constexpr void swap(std::nullptr_t) noexcept
+  {
+    *this = nullptr;
+  };
+  constexpr element_type* get() const noexcept { return mptr; }
+  constexpr std::add_lvalue_reference_t<element_type> operator*() const { return *get(); }
+  constexpr element_type* operator->() const noexcept { return get(); }
+  constexpr std::add_lvalue_reference_t<observer_ptr<element_type>> operator=(const std::add_lvalue_reference_t<observer_ptr<element_type>> other) { mptr = other.mptr; }
+  constexpr std::add_lvalue_reference_t<observer_ptr<element_type>> operator=(element_type* const other) { mptr = other; }
+
+  constexpr explicit operator element_type*() const noexcept { return get(); }
+  constexpr explicit operator bool() const noexcept { return get() != nullptr; }
+
+ private:
+  element_type* mptr{ nullptr };
+};
+
+template <typename W>
+observer_ptr<W> make_observer(W* p) noexcept
+{
+  return observer_ptr(p);
+}
+
+} //namespace o2
+
+#endif

Framework/Core/src/DataAllocator.cxx

@@ -117,8 +117,8 @@ FairMQMessagePtr DataAllocator::headerMessageFromOutput(Output const& spec,
   DataProcessingHeader dph{mTimingInfo->timeslice, 1};
   auto context = mContextRegistry->get<MessageContext>();
 
-  auto channelAlloc = o2::memory_resource::getTransportAllocator(context->proxy().getTransport(channel, 0));
-  return o2::memory_resource::getMessage(o2::header::Stack{ channelAlloc, dh, dph, spec.metaHeader });
+  auto channelAlloc = o2::pmr::getTransportAllocator(context->proxy().getTransport(channel, 0));
+  return o2::pmr::getMessage(o2::header::Stack{ channelAlloc, dh, dph, spec.metaHeader });
 }
 
 void DataAllocator::addPartToContext(FairMQMessagePtr&& payloadMessage, const Output& spec,

Framework/Core/src/Dispatcher.cxx

@@ -104,8 +104,8 @@ void Dispatcher::sendFairMQ(FairMQDevice* device, const DataRef& inputData, cons
   DataProcessingHeader dphout{ dph->startTime, dph->duration };
   o2::header::Stack headerStack{ dhout, dphout };
 
-  auto channelAlloc = o2::memory_resource::getTransportAllocator(device->Transport());
-  FairMQMessagePtr msgHeaderStack = o2::memory_resource::getMessage(std::move(headerStack), channelAlloc);
+  auto channelAlloc = o2::pmr::getTransportAllocator(device->Transport());
+  FairMQMessagePtr msgHeaderStack = o2::pmr::getMessage(std::move(headerStack), channelAlloc);
 
   char* payloadCopy = new char[dh->payloadSize];
   memcpy(payloadCopy, inputData.payload, dh->payloadSize);

Framework/Core/src/ExternalFairMQDeviceProxy.cxx

@@ -40,8 +40,8 @@ void broadcastMessage(FairMQDevice &device, o2::header::Stack &&headerStack, Fai
 
     // FIXME: this assumes there is only one output from here... This should
     //        really do the matchmaking between inputs and output channels.
-    auto channelAlloc = o2::memory_resource::getTransportAllocator(channelInfo.second[index].Transport());
-    FairMQMessagePtr headerMessage = o2::memory_resource::getMessage(std::move(headerStack), channelAlloc);
+    auto channelAlloc = o2::pmr::getTransportAllocator(channelInfo.second[index].Transport());
+    FairMQMessagePtr headerMessage = o2::pmr::getMessage(std::move(headerStack), channelAlloc);
 
     FairMQParts out;
     out.AddPart(std::move(headerMessage));

Framework/Core/src/LifetimeHelpers.cxx

@@ -126,8 +126,8 @@ ExpirationHandler::Handler LifetimeHelpers::enumerate(ConcreteDataMatcher const&
     DataProcessingHeader dph{ timestamp, 1 };
 
     auto&& transport = rawDeviceService.device()->GetChannel(sourceChannel, 0).Transport();
-    auto channelAlloc = o2::memory_resource::getTransportAllocator(transport);
-    auto header = o2::memory_resource::getMessage(o2::header::Stack{ channelAlloc, dh, dph });
+    auto channelAlloc = o2::pmr::getTransportAllocator(transport);
+    auto header = o2::pmr::getMessage(o2::header::Stack{ channelAlloc, dh, dph });
     ref.header = std::move(header);
 
     auto payload = rawDeviceService.device()->NewMessage(*counter);

Framework/Utils/include/Utils/Utils.h

@@ -18,6 +18,7 @@
 
 #include "Framework/DataProcessorSpec.h"
 #include <functional>
+#include "MemoryResources/MemoryResources.h"
 
 namespace o2f = o2::framework;
 
@@ -36,7 +37,7 @@ o2f::DataProcessorSpec defineBroadcaster(std::string devName, o2f::InputSpec usr
                                          size_t fixMsgSize);
 o2f::DataProcessorSpec defineBroadcaster(std::string devName, o2f::InputSpec usrInput, o2f::Outputs usrOutputs);
 
-using OutputBuffer = std::vector<char>;
+using OutputBuffer = o2::vector<char>;
 // Merger implementations
 o2f::DataProcessorSpec defineMerger(std::string devName, o2f::Inputs usrInputs, o2f::OutputSpec usrOutput,
                                     std::function<void(OutputBuffer, const o2f::DataRef)> const mergerFunc);

Framework/Utils/src/DPLMerger.cxx

@@ -40,14 +40,13 @@ o2f::DataProcessorSpec defineMerger(std::string devName, o2f::Inputs usrInputs,
 
              // Defining the ProcessCallback as returned object of InitCallback
              return [outputPtr, mergerFuncPtr](o2f::ProcessingContext& ctx) {
-               OutputBuffer outputBuffer;
+               OutputBuffer outputBuffer = ctx.outputs().makeVector<char>(*outputPtr);
                // Iterating over the InputSpecs to aggregate msgs from the connected devices
                for (const auto& itInputs : ctx.inputs()) {
                  (*mergerFuncPtr)(outputBuffer, itInputs);
                }
                // Adopting the buffer as new chunk
-               ctx.outputs().adoptChunk((*outputPtr), &outputBuffer[0], outputBuffer.size(), header::Stack::getFreefn(),
-                                        nullptr);
+               ctx.outputs().adoptContainer((*outputPtr), std::move(outputBuffer));
              };
            } } };
 }

Utilities/O2Device/README.md

@@ -23,7 +23,7 @@ addDataBlock(O2Message& message, o2::header::Stack&& headerStack, T data);
 - data:
   - ```FairMQMessagePtr``` (```std::unique_ptr<FairMQMessage>```)
   - STL-like container that uses a ```pmr::polymorphic_allocator``` as allocator. Note: this is only efficient if
-    the memory resource used to construct the allocator is ```o2::memory_resource::ChannelResource```, otherwise an implicit copy occurs.
+    the memory resource used to construct the allocator is ```o2::pmr::ChannelResource```, otherwise an implicit copy occurs.
 
 ### forEach()
 Executes a function on each data block within the message.
@@ -39,11 +39,11 @@ the function is a callable object (lambda, functor, std::function) and needs to
 #### basic example, inside a FairMQDevice:
 ```C++
 // make sure we have the allocator associated with the transport appropriate for the selected channel:
-auto outputChannelAllocator = o2::memory_resource::getTransportAllocator(GetChannel("dataOut").Transport());
+auto outputChannelAllocator = o2::pmr::getTransportAllocator(GetChannel("dataOut").Transport());
 
 //the data
 using namespace boost::container::pmr;
-std::vector<int, polymorphic_allocator<int>> dataVector(polymorphic_allocator<int>{outputChannelAllocator});
+o2::pmr::vector<int> dataVector(outputChannelAllocator);
 dataVector.reserve(10);
 dataVector.push_back(1);
 dataVector.push_back(2);

Utilities/O2Device/include/O2Device/Utilities.h

@@ -44,7 +44,7 @@ using O2Message = FairMQParts;
 //__________________________________________________________________________________________________
 // addDataBlock for generic (compatible) containers, that is contiguous containers using the pmr allocator
 template <typename ContainerT, typename std::enable_if<!std::is_same<ContainerT, FairMQMessagePtr>::value, int>::type = 0>
-bool addDataBlock(O2Message& parts, o2::header::Stack&& inputStack, ContainerT&& inputData, o2::memory_resource::FairMQMemoryResource* targetResource = nullptr)
+bool addDataBlock(O2Message& parts, o2::header::Stack&& inputStack, ContainerT&& inputData, o2::pmr::FairMQMemoryResource* targetResource = nullptr)
 {
   using std::move;
   using std::forward;
@@ -62,7 +62,7 @@ bool addDataBlock(O2Message& parts, o2::header::Stack&& inputStack, ContainerT&&
 // addDataBlock for data already wrapped in FairMQMessagePtr
 // note: since we cannot partially specialize function templates, use SFINAE here instead
 template <typename ContainerT, typename std::enable_if<std::is_same<ContainerT, FairMQMessagePtr>::value, int>::type = 0>
-bool addDataBlock(O2Message& parts, o2::header::Stack&& inputStack, ContainerT&& dataMessage, o2::memory_resource::FairMQMemoryResource* targetResource = nullptr)
+bool addDataBlock(O2Message& parts, o2::header::Stack&& inputStack, ContainerT&& dataMessage, o2::pmr::FairMQMemoryResource* targetResource = nullptr)
 {
   using std::move;