DPL: introduce Lifetime::OutOfBand

This lifetime can be used to trigger processing on messages
incoming to an out of band fairmq channel. The idea is that
we can use this e.g. for the feedback look required by rate
limiting the readout-proxy.

Author: ktf

Framework/Core/include/Framework/DeviceState.h

@@ -55,7 +55,8 @@ struct DeviceState {
     WS_CLOSING = 256,       // Events related to WS shutting down
     WS_READING = 512,       // Events related to WS shutting down
     WS_WRITING = 1024,      // Events related to WS shutting down
-    ASYNC_NOTIFICATION = 2048
+    ASYNC_NOTIFICATION = 2048,
+    OOB_ACTIVITY = 4096 // Out of band activity
   };
 
   std::vector<InputChannelInfo> inputChannelInfos;
@@ -79,6 +80,8 @@ struct DeviceState {
   std::vector<uv_poll_t*> activeOutputPollers;
   /// The list of active signal handlers
   std::vector<uv_signal_t*> activeSignals;
+  /// The list for active out-of-bound pollers
+  std::vector<uv_poll_t*> activeOutOfBandPollers;
 
   uv_async_t* awakeMainThread = nullptr;
 

Framework/Core/include/Framework/InputRoute.h

@@ -27,7 +27,7 @@ struct DeviceState;
 class ConfigParamRegistry;
 
 struct RouteConfigurator {
-  using CreationConfigurator = std::function<ExpirationHandler::Creator(DeviceState&, ConfigParamRegistry const&)>;
+  using CreationConfigurator = std::function<ExpirationHandler::Creator(DeviceState&, ServiceRegistry&, ConfigParamRegistry const&)>;
   using DanglingConfigurator = std::function<ExpirationHandler::Checker(DeviceState&, ConfigParamRegistry const&)>;
   using ExpirationConfigurator = std::function<ExpirationHandler::Handler(DeviceState&, ConfigParamRegistry const&)>;
 

Framework/Core/include/Framework/Lifetime.h

@@ -36,7 +36,10 @@ enum struct Lifetime {
   /// This comes handy e.g. to handle Raw Data, since DataDistribution will provide
   /// everything in one go so whatever is expected but not there, for whatever reason
   /// will be substituted with a dummy entry.
-  Optional
+  Optional,
+  /// An input which is materialised with the contents of some out of band
+  /// FairMQ channel.
+  OutOfBand
 };
 
 } // namespace o2::framework

Framework/Core/include/Framework/LifetimeHelpers.h

@@ -66,6 +66,12 @@ struct LifetimeHelpers {
                                                        std::string const& overrideTimestamp,
                                                        std::string const& sourceChannel);
 
+  /// Build a fetcher for an object from an out of band FairMQ channel whenever the record is expired.
+  /// @a spec is the associated InputSpec
+  /// @a channelName the channel we should Receive data from
+  static ExpirationHandler::Handler fetchFromFairMQ(InputSpec const& spec,
+                                                    std::string const& channelName);
+
   /// Create an entry in the registry for histograms on the first
   /// FIXME: actually implement this
   /// FIXME: provide a way to customise the histogram from the configuration.

Framework/Core/src/DataProcessingDevice.cxx

@@ -289,22 +289,6 @@ void DataProcessingDevice::Init()
 
   mExpirationHandlers.clear();
 
-  auto distinct = DataRelayerHelpers::createDistinctRouteIndex(mSpec.inputs);
-  int i = 0;
-  for (auto& di : distinct) {
-    auto& route = mSpec.inputs[di];
-    if (route.configurator.has_value() == false) {
-      i++;
-      continue;
-    }
-    ExpirationHandler handler{
-      RouteIndex{i++},
-      route.matcher.lifetime,
-      route.configurator->creatorConfigurator(mState, *mConfigRegistry),
-      route.configurator->danglingConfigurator(mState, *mConfigRegistry),
-      route.configurator->expirationConfigurator(mState, *mConfigRegistry)};
-    mExpirationHandlers.emplace_back(std::move(handler));
-  }
 
   if (mInit) {
     InitContext initContext{*mConfigRegistry, mServiceRegistry};
@@ -380,6 +364,23 @@ void on_awake_main_thread(uv_async_t* handle)
 } // namespace
 void DataProcessingDevice::InitTask()
 {
+  auto distinct = DataRelayerHelpers::createDistinctRouteIndex(mSpec.inputs);
+  int i = 0;
+  for (auto& di : distinct) {
+    auto& route = mSpec.inputs[di];
+    if (route.configurator.has_value() == false) {
+      i++;
+      continue;
+    }
+    ExpirationHandler handler{
+      RouteIndex{i++},
+      route.matcher.lifetime,
+      route.configurator->creatorConfigurator(mState, mServiceRegistry, *mConfigRegistry),
+      route.configurator->danglingConfigurator(mState, *mConfigRegistry),
+      route.configurator->expirationConfigurator(mState, *mConfigRegistry)};
+    mExpirationHandlers.emplace_back(std::move(handler));
+  }
+
   if (mState.awakeMainThread == nullptr) {
     mState.awakeMainThread = (uv_async_t*)malloc(sizeof(uv_async_t));
     mState.awakeMainThread->data = &mState;

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -33,6 +33,7 @@
 #include "Framework/ComputingResource.h"
 #include "Framework/Logger.h"
 #include "Framework/RuntimeError.h"
+#include "Framework/RawDeviceService.h"
 #include "ProcessingPoliciesHelpers.h"
 
 #include "WorkflowHelpers.h"
@@ -74,12 +75,12 @@ void signal_callback(uv_signal_t* handle, int)
 struct ExpirationHandlerHelpers {
   static RouteConfigurator::CreationConfigurator dataDrivenConfigurator()
   {
-    return [](DeviceState&, ConfigParamRegistry const&) { return LifetimeHelpers::dataDrivenCreation(); };
+    return [](DeviceState&, ServiceRegistry&, ConfigParamRegistry const&) { return LifetimeHelpers::dataDrivenCreation(); };
   }
 
   static RouteConfigurator::CreationConfigurator timeDrivenConfigurator(InputSpec const& matcher)
   {
-    return [matcher](DeviceState& state, ConfigParamRegistry const& options) {
+    return [matcher](DeviceState& state, ServiceRegistry&, ConfigParamRegistry const& options) {
       std::string rateName = std::string{"period-"} + matcher.binding;
       auto period = options.get<int>(rateName.c_str());
       // We create a timer to wake us up. Notice the actual
@@ -97,7 +98,7 @@ struct ExpirationHandlerHelpers {
 
   static RouteConfigurator::CreationConfigurator signalDrivenConfigurator(InputSpec const& matcher, size_t inputTimeslice, size_t maxInputTimeslices)
   {
-    return [matcher, inputTimeslice, maxInputTimeslices](DeviceState& state, ConfigParamRegistry const& options) {
+    return [matcher, inputTimeslice, maxInputTimeslices](DeviceState& state, ServiceRegistry&, ConfigParamRegistry const& options) {
       std::string startName = std::string{"start-value-"} + matcher.binding;
       std::string endName = std::string{"end-value-"} + matcher.binding;
       std::string stepName = std::string{"step-value-"} + matcher.binding;
@@ -119,7 +120,7 @@ struct ExpirationHandlerHelpers {
 
   static RouteConfigurator::CreationConfigurator enumDrivenConfigurator(InputSpec const& matcher, size_t inputTimeslice, size_t maxInputTimeslices)
   {
-    return [matcher, inputTimeslice, maxInputTimeslices](DeviceState&, ConfigParamRegistry const& options) {
+    return [matcher, inputTimeslice, maxInputTimeslices](DeviceState&, ServiceRegistry&, ConfigParamRegistry const& options) {
       std::string startName = std::string{"start-value-"} + matcher.binding;
       std::string endName = std::string{"end-value-"} + matcher.binding;
       std::string stepName = std::string{"step-value-"} + matcher.binding;
@@ -164,6 +165,52 @@ struct ExpirationHandlerHelpers {
     };
   }
 
+  static RouteConfigurator::CreationConfigurator fairmqDrivenConfiguration(InputSpec const& spec, int inputTimeslice, int maxInputTimeslices)
+  {
+    return [spec, inputTimeslice, maxInputTimeslices](DeviceState& state, ServiceRegistry& services, ConfigParamRegistry const& options) {
+      std::string channelNameOption = std::string{"out-of-band-channel-name-"} + spec.binding;
+      auto channelName = options.get<std::string>(channelNameOption.c_str());
+      auto device = services.get<RawDeviceService>().device();
+      auto& channel = device->fChannels[channelName];
+
+      // We assume there is always a ZeroMQ socket behind.
+      int zmq_fd = 0;
+      size_t zmq_fd_len = sizeof(zmq_fd);
+      uv_poll_t* poller = (uv_poll_t*)malloc(sizeof(uv_poll_t));
+      channel[0].GetSocket().GetOption("fd", &zmq_fd, &zmq_fd_len);
+      if (zmq_fd == 0) {
+        throw runtime_error_f("Cannot get file descriptor for channel %s", channelName.c_str());
+      }
+      LOG(debug) << "Polling socket for " << channel[0].GetName();
+
+      state.activeOutOfBandPollers.push_back(poller);
+
+      // We always create entries whenever we get invoked.
+      // Notice this works only if we are the only input.
+      // Otherwise we should check the channel for new data,
+      // before we create an entry.
+      return LifetimeHelpers::enumDrivenCreation(0, -1, 1, inputTimeslice, maxInputTimeslices, 1);
+    };
+  }
+
+  static RouteConfigurator::DanglingConfigurator danglingOutOfBandConfigurator()
+  {
+    return [](DeviceState&, ConfigParamRegistry const& options) {
+      // If the entry is there it means that something awoke
+      // the loop, so we can materialise it immediately.
+      return LifetimeHelpers::expireAlways();
+    };
+  }
+
+  static RouteConfigurator::ExpirationConfigurator expiringOutOfBandConfigurator(InputSpec const& spec)
+  {
+    return [spec](DeviceState&, ConfigParamRegistry const& options) {
+      std::string channelNameOption = std::string{"out-of-band-channel-name-"} + spec.binding;
+      auto channelName = options.get<std::string>(channelNameOption.c_str());
+      return LifetimeHelpers::fetchFromFairMQ(spec, channelName);
+    };
+  }
+
   static RouteConfigurator::DanglingConfigurator danglingQAConfigurator()
   {
     // FIXME: this should really be expireAlways. However, since we do not have
@@ -234,7 +281,7 @@ struct ExpirationHandlerHelpers {
   /// This behaves as data. I.e. we never create it unless data arrives.
   static RouteConfigurator::CreationConfigurator createOptionalConfigurator()
   {
-    return [](DeviceState&, ConfigParamRegistry const&) { return LifetimeHelpers::dataDrivenCreation(); };
+    return [](DeviceState&, ServiceRegistry&, ConfigParamRegistry const&) { return LifetimeHelpers::dataDrivenCreation(); };
   }
 
   /// This will always exipire an optional record when no data is received.
@@ -663,6 +710,12 @@ void DeviceSpecHelpers::processInEdgeActions(std::vector<DeviceSpec>& devices,
       std::nullopt};
 
     switch (consumer.inputs[edge.consumerInputIndex].lifetime) {
+      case Lifetime::OutOfBand:
+        route.configurator = {
+          ExpirationHandlerHelpers::fairmqDrivenConfiguration(inputSpec, consumerDevice.inputTimesliceId, consumerDevice.maxInputTimeslices),
+          ExpirationHandlerHelpers::danglingOutOfBandConfigurator(),
+          ExpirationHandlerHelpers::expiringOutOfBandConfigurator(inputSpec)};
+        break;
       case Lifetime::Condition:
         route.configurator = {
           ExpirationHandlerHelpers::dataDrivenConfigurator(),

Framework/Core/src/DeviceSpecHelpers.h

@@ -27,6 +27,8 @@
 #include "Framework/ProcessingPolicies.h"
 #include "ResourceManager.h"
 #include "WorkflowHelpers.h"
+
+#include <FairMQDevice.h>
 #include <boost/program_options.hpp>
 
 #include <vector>

Framework/Core/src/LifetimeHelpers.cxx

@@ -374,6 +374,24 @@ ExpirationHandler::Handler
   };
 }
 
+ExpirationHandler::Handler
+  LifetimeHelpers::fetchFromFairMQ(InputSpec const& spec,
+                                   std::string const& channelName)
+{
+  return [spec, channelName](ServiceRegistry& services, PartRef& ref, data_matcher::VariableContext& variables) -> void {
+    auto& rawDeviceService = services.get<RawDeviceService>();
+    auto device = rawDeviceService.device();
+
+    // Receive parts and put them in the PartRef
+    // we know this is not blocking because we were polled
+    // on the channel.
+    FairMQParts parts;
+    device->Receive(parts, channelName, 0);
+    ref.header = std::move(parts.At(0));
+    ref.payload = std::move(parts.At(1));
+  };
+}
+
 /// Create an entry in the registry for histograms on the first
 /// FIXME: actually implement this
 /// FIXME: provide a way to customise the histogram from the configuration.
@@ -498,6 +516,7 @@ std::ostream& operator<<(std::ostream& oss, Lifetime const& val)
     STREAM_ENUM(Lifetime::Enumeration)
     STREAM_ENUM(Lifetime::Signal)
     STREAM_ENUM(Lifetime::Optional)
+    STREAM_ENUM(Lifetime::OutOfBand)
   };
   return oss;
 }

Framework/Core/src/WorkflowHelpers.cxx

@@ -336,6 +336,13 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
           }
           requestedCCDBs.emplace_back(input);
         } break;
+        case Lifetime::OutOfBand: {
+          auto concrete = DataSpecUtils::asConcreteDataMatcher(input);
+          auto hasOption = std::any_of(processor.options.begin(), processor.options.end(), [&input](auto const& option) { return (option.name == "out-of-band-channel-name-" + input.binding); });
+          if (hasOption == false) {
+            processor.options.push_back(ConfigParamSpec{"out-of-band-channel-name-" + input.binding, VariantType::String, "out-of-band", {"channel to listen for out of band data"}});
+          }
+        } break;
         case Lifetime::QA:
         case Lifetime::Transient:
         case Lifetime::Timeframe: