From cae4f945f261b5a5a62ca13ab82802f61584cc91 Mon Sep 17 00:00:00 2001 From: Giulio Eulisse Date: Tue, 5 May 2020 23:50:21 +0200 Subject: [PATCH] DPL: initial support for FairMQRegionInfo callback This should allow users to register a callback and be notified whenever FairMQ creates a new shared memory region. Guaranteed to be called synchronously wrt the processing. --- .../Core/include/Framework/CallbackService.h | 38 ++++++++++++++----- .../include/Framework/DataProcessingDevice.h | 2 + Framework/Core/src/DataProcessingDevice.cxx | 21 ++++++++++ 3 files changed, 52 insertions(+), 9 deletions(-) diff --git a/Framework/Core/include/Framework/CallbackService.h b/Framework/Core/include/Framework/CallbackService.h index 1e5279701ce01..004a660d830aa 100644 --- a/Framework/Core/include/Framework/CallbackService.h +++ b/Framework/Core/include/Framework/CallbackService.h @@ -13,6 +13,8 @@ #include "CallbackRegistry.h" #include +class FairMQRegionInfo; + namespace o2 { namespace framework @@ -35,7 +37,23 @@ class CallbackService /// Invoked when we are notified that no further data will arrive. /// Notice that one could have more "EndOfData" notifications. Because /// we could be signaled by control that the data flow restarted. - EndOfStream + EndOfStream, + + /// Invoked whenever FairMQ notifies us of a new region + /// + /// return AlgorithmSpec::InitCallback{[=](InitContext& ic) { + /// auto& callbacks = ic.services().get(); + /// callbacks.set(CallbackService::Id::RegionInfoCallback, [](FairMQRegionInfo const& info) { + /// ... do GPU init ... + /// }); + /// } + /// ... + /// return [task](ProcessingContext& pc) { + /// // your processing loop. Guaranteed to be called synchronously + /// // with the callback + /// }; + /// }}; + RegionInfoCallback }; template @@ -51,15 +69,17 @@ class CallbackService using IdleCallback = std::function; using ClockTickCallback = std::function; using EndOfStreamCallback = std::function; + using RegionInfoCallback = std::function; - using Callbacks = CallbackRegistry, // - RegistryPair, // - RegistryPair, // - RegistryPair, // - RegistryPair, // - RegistryPair // - >; // + using Callbacks = CallbackRegistry, // + RegistryPair, // + RegistryPair, // + RegistryPair, // + RegistryPair, // + RegistryPair, // + RegistryPair // + >; // // set callback for specified processing step template diff --git a/Framework/Core/include/Framework/DataProcessingDevice.h b/Framework/Core/include/Framework/DataProcessingDevice.h index 40e3906e1ad4c..478a985b194f3 100644 --- a/Framework/Core/include/Framework/DataProcessingDevice.h +++ b/Framework/Core/include/Framework/DataProcessingDevice.h @@ -45,6 +45,7 @@ class DataProcessingDevice : public FairMQDevice public: DataProcessingDevice(DeviceSpec const& spec, ServiceRegistry&, DeviceState& state); void Init() final; + void InitTask() final; void PreRun() final; void PostRun() final; void Reset() final; @@ -84,6 +85,7 @@ class DataProcessingDevice : public FairMQDevice uint64_t mBeginIterationTimestamp = 0; /// The timestamp of when the current ConditionalRun was started DataProcessingStats mStats; /// Stats about the actual data processing. int mCurrentBackoff = 0; /// The current exponential backoff value. + std::vector mPendingRegionInfos; /// A list of the region infos not yet notified. }; } // namespace o2::framework diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx index 6df86108091a4..4002264bd53bc 100644 --- a/Framework/Core/src/DataProcessingDevice.cxx +++ b/Framework/Core/src/DataProcessingDevice.cxx @@ -165,6 +165,20 @@ void DataProcessingDevice::Init() } } +void DataProcessingDevice::InitTask() +{ + for (auto& channel : fChannels) { + channel.second.at(0).Transport()->SubscribeToRegionEvents([& pendingRegionInfos = mPendingRegionInfos](FairMQRegionInfo info) { + LOG(debug) << ">>> Region info event" << info.event; + LOG(debug) << "id: " << info.id; + LOG(debug) << "ptr: " << info.ptr; + LOG(debug) << "size: " << info.size; + LOG(debug) << "flags: " << info.flags; + pendingRegionInfos.push_back(info); + }); + } +} + void DataProcessingDevice::PreRun() { mServiceRegistry.get()(CallbackService::Id::Start); } void DataProcessingDevice::PostRun() { mServiceRegistry.get()(CallbackService::Id::Stop); } @@ -250,6 +264,13 @@ bool DataProcessingDevice::ConditionalRun() auto now = std::chrono::high_resolution_clock::now(); mBeginIterationTimestamp = (uint64_t)std::chrono::duration(now.time_since_epoch()).count(); + if (mPendingRegionInfos.empty() == false) { + std::vector toBeNotified; + toBeNotified.swap(mPendingRegionInfos); // avoid any MT issue. + for (auto& info : toBeNotified) { + mServiceRegistry.get()(CallbackService::Id::RegionInfoCallback, info); + } + } mServiceRegistry.get()(CallbackService::Id::ClockTick); // Whether or not we had something to do. bool active = false;