DPL: improve rate limiting

ktf · ktf · commit 195b9767b029 · 2022-05-03T23:25:40.000+02:00
* Now uses consistently the oldest possible timeframe to as
  a rate limiting metric.
* There is now a callback whenever we receive a DomainInfoHeader.
  It can be subscribed by the user and it is used e.g. to implement
  the rate limiting.
diff --git a/Framework/Core/include/Framework/CallbackService.h b/Framework/Core/include/Framework/CallbackService.h
@@ -73,7 +73,11 @@ class CallbackService
     PreProcessing,
     /// Invoked after the processing callback,
     PostProcessing,
-    CCDBDeserialised
+    /// Invoked whenever an object from CCDB is deserialised via ROOT.
+    /// Use this to finalise the initialisation of the object.
+    CCDBDeserialised,
+    /// Invoked when new domain info is available
+    DomainInfoUpdated
   };
 
   using StartCallback = std::function<void()>;
@@ -88,21 +92,23 @@ class CallbackService
   using PreProcessingCallback = std::function<void(ServiceRegistry&, int)>;
   using PostProcessingCallback = std::function<void(ServiceRegistry&, int)>;
   using CCDBDeserializedCallback = std::function<void(ConcreteDataMatcher&, void*)>;
+  using DomainInfoUpdatedCallback = std::function<void(ServiceRegistry&, size_t timeslice)>;
 
-  using Callbacks = CallbackRegistry<Id,                                                              //
-                                     RegistryPair<Id, Id::Start, StartCallback>,                      //
-                                     RegistryPair<Id, Id::Stop, StopCallback>,                        //
-                                     RegistryPair<Id, Id::Reset, ResetCallback>,                      //
-                                     RegistryPair<Id, Id::Idle, IdleCallback>,                        //
-                                     RegistryPair<Id, Id::ClockTick, ClockTickCallback>,              //
-                                     RegistryPair<Id, Id::DataConsumed, DataConsumedCallback>,        //
-                                     RegistryPair<Id, Id::EndOfStream, EndOfStreamCallback>,          //
-                                     RegistryPair<Id, Id::RegionInfoCallback, RegionInfoCallback>,    //
-                                     RegistryPair<Id, Id::NewTimeslice, NewTimesliceCallback>,        //
-                                     RegistryPair<Id, Id::PreProcessing, PreProcessingCallback>,      //
-                                     RegistryPair<Id, Id::PostProcessing, PostProcessingCallback>,    //
-                                     RegistryPair<Id, Id::CCDBDeserialised, CCDBDeserializedCallback> //
-                                     >;                                                               //
+  using Callbacks = CallbackRegistry<Id,                                                                //
+                                     RegistryPair<Id, Id::Start, StartCallback>,                        //
+                                     RegistryPair<Id, Id::Stop, StopCallback>,                          //
+                                     RegistryPair<Id, Id::Reset, ResetCallback>,                        //
+                                     RegistryPair<Id, Id::Idle, IdleCallback>,                          //
+                                     RegistryPair<Id, Id::ClockTick, ClockTickCallback>,                //
+                                     RegistryPair<Id, Id::DataConsumed, DataConsumedCallback>,          //
+                                     RegistryPair<Id, Id::EndOfStream, EndOfStreamCallback>,            //
+                                     RegistryPair<Id, Id::RegionInfoCallback, RegionInfoCallback>,      //
+                                     RegistryPair<Id, Id::NewTimeslice, NewTimesliceCallback>,          //
+                                     RegistryPair<Id, Id::PreProcessing, PreProcessingCallback>,        //
+                                     RegistryPair<Id, Id::PostProcessing, PostProcessingCallback>,      //
+                                     RegistryPair<Id, Id::CCDBDeserialised, CCDBDeserializedCallback>,  //
+                                     RegistryPair<Id, Id::DomainInfoUpdated, DomainInfoUpdatedCallback> //
+                                     >;                                                                 //
 
   // set callback for specified processing step
   template <typename U>
diff --git a/Framework/Core/include/Framework/ExpirationHandler.h b/Framework/Core/include/Framework/ExpirationHandler.h
@@ -29,7 +29,7 @@ struct TimesliceSlot;
 struct InputRecord;
 
 struct ExpirationHandler {
-  using Creator = std::function<TimesliceSlot(TimesliceIndex&)>;
+  using Creator = std::function<TimesliceSlot(ChannelIndex, TimesliceIndex&)>;
   /// Callback type to check if the record must be expired
   using Checker = std::function<bool(ServiceRegistry&, uint64_t timestamp, InputSpan const& record)>;
   /// Callback type to actually materialise a given record
diff --git a/Framework/Core/include/Framework/InputRecord.h b/Framework/Core/include/Framework/InputRecord.h
@@ -682,6 +682,11 @@ class InputRecord
     return {this, size()};
   }
 
+  InputSpan& span()
+  {
+    return mSpan;
+  }
+
  private:
   ServiceRegistry& mRegistry;
   std::vector<InputRoute> const& mInputsSchema;
diff --git a/Framework/Core/include/Framework/MessageContext.h b/Framework/Core/include/Framework/MessageContext.h
@@ -40,7 +40,7 @@
 namespace o2::framework
 {
 
-class Output;
+struct Output;
 
 class MessageContext
 {
diff --git a/Framework/Core/include/Framework/ProcessingContext.h b/Framework/Core/include/Framework/ProcessingContext.h
@@ -30,8 +30,11 @@ class ProcessingContext
   {
   }
 
+  /// The inputs associated with this processing context.
   InputRecord& inputs() { return mInputs; }
+  /// The services registry associated with this processing context.
   ServiceRegistry& services() { return mServices; }
+  /// The data allocator is used to allocate memory for the output data.
   DataAllocator& outputs() { return mAllocator; }
 
   InputRecord& mInputs;
diff --git a/Framework/Core/src/CallbacksPolicy.cxx b/Framework/Core/src/CallbacksPolicy.cxx
@@ -33,13 +33,17 @@ CallbacksPolicy epnProcessReporting()
     .policy = [](CallbackService& callbacks, InitContext& context) -> void {
       callbacks.set(CallbackService::Id::PreProcessing, [](ServiceRegistry& registry, int op) {
         auto& info = registry.get<TimingInfo>();
-        LOGP(info, "Processing timeslice:{}, tfCounter:{}, firstTFOrbit:{}, action:{}",
-             info.timeslice, info.tfCounter, info.firstTFOrbit, op);
+        if ((int)info.firstTFOrbit != -1) {
+          LOGP(info, "Processing timeslice:{}, tfCounter:{}, firstTFOrbit:{}, action:{}",
+               info.timeslice, info.tfCounter, info.firstTFOrbit, op);
+        }
       });
       callbacks.set(CallbackService::Id::PostProcessing, [](ServiceRegistry& registry, int op) {
         auto& info = registry.get<TimingInfo>();
-        LOGP(info, "Done processing timeslice:{}, tfCounter:{}, firstTFOrbit:{}, action:{}",
-             info.timeslice, info.tfCounter, info.firstTFOrbit, op);
+        if ((int)info.firstTFOrbit != -1) {
+          LOGP(info, "Done processing timeslice:{}, tfCounter:{}, firstTFOrbit:{}, action:{}",
+               info.timeslice, info.tfCounter, info.firstTFOrbit, op);
+        }
       });
     }};
 }
diff --git a/Framework/Core/src/CommonDataProcessors.cxx b/Framework/Core/src/CommonDataProcessors.cxx
@@ -28,6 +28,7 @@
 #include "Framework/Logger.h"
 #include "Framework/OutputSpec.h"
 #include "Framework/RawDeviceService.h"
+#include "Framework/TimesliceIndex.h"
 #include "Framework/Variant.h"
 #include "../../../Algorithm/include/Algorithm/HeaderStack.h"
 #include "Framework/OutputObjHeader.h"
@@ -508,19 +509,19 @@ DataProcessorSpec CommonDataProcessors::getDummySink(std::vector<InputSpec> cons
     .name = "internal-dpl-injected-dummy-sink",
     .inputs = danglingOutputInputs,
     .algorithm = AlgorithmSpec{adaptStateful([](CallbackService& callbacks) {
-      auto dataConsumed = [](ServiceRegistry& services) {
-        services.get<DataProcessingStats>().consumedTimeframes++;
+      auto domainInfoUpdated = [](ServiceRegistry& services, size_t timeslice) {
+        auto& timesliceIndex = services.get<TimesliceIndex>();
         auto device = services.get<RawDeviceService>().device();
         auto channel = device->fChannels.find("metric-feedback");
         if (channel != device->fChannels.end()) {
           FairMQMessagePtr payload(device->NewMessage());
-          int64_t* consumed = (int64_t*)malloc(sizeof(int64_t));
-          *consumed = services.get<DataProcessingStats>().consumedTimeframes;
+          size_t* consumed = (size_t*)malloc(sizeof(size_t));
+          *consumed = timesliceIndex.getOldestPossibleOutput().timeslice.value;
           payload->Rebuild(consumed, sizeof(int64_t), nullptr, nullptr);
           channel->second[0].Send(payload);
         }
       };
-      callbacks.set(CallbackService::Id::DataConsumed, dataConsumed);
+      callbacks.set(CallbackService::Id::DomainInfoUpdated, domainInfoUpdated);
 
       return adaptStateless([]() {
       });
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
@@ -1326,7 +1326,14 @@ void DataProcessingDevice::handleData(DataProcessorContext& context, InputChanne
     return;
   }
   if (oldestPossibleTimeslice != (size_t)-1) {
-    context.relayer->setOldestPossibleInput({oldestPossibleTimeslice}, info.id);
+    TimesliceIndex& timesliceIndex = context.registry->get<TimesliceIndex>();
+    auto r = timesliceIndex.setOldestPossibleInput({oldestPossibleTimeslice}, info.id);
+    timesliceIndex.updateOldestPossibleOutput();
+    auto& proxy = context.registry->get<FairMQDeviceProxy>();
+    auto oldestPossibleOutput = context.relayer->getOldestPossibleOutput();
+    LOGP(detail, "Broadcasting possible output {}", oldestPossibleOutput.timeslice.value);
+    context.registry->get<CallbackService>()(CallbackService::Id::DomainInfoUpdated, *(context.registry), (size_t)oldestPossibleOutput.timeslice.value);
+    DataProcessingHelpers::broadcastOldestPossibleTimeslice(proxy, oldestPossibleOutput.timeslice.value);
   }
   handleValidMessages(*inputTypes);
   return;
diff --git a/Framework/Core/src/DataRelayer.cxx b/Framework/Core/src/DataRelayer.cxx
@@ -25,6 +25,7 @@
 #include "Framework/Signpost.h"
 #include "Framework/RoutingIndices.h"
 #include "Framework/VariableContextHelpers.h"
+#include "Framework/FairMQDeviceProxy.h"
 #include "DataProcessingStatus.h"
 #include "DataRelayerHelpers.h"
 #include "InputRouteHelpers.h"
@@ -59,8 +60,8 @@ DataRelayer::DataRelayer(const CompletionPolicy& policy,
                          std::vector<InputRoute> const& routes,
                          monitoring::Monitoring& metrics,
                          TimesliceIndex& index)
-  : mTimesliceIndex{index},
-    mMetrics{metrics},
+  : mMetrics{metrics},
+    mTimesliceIndex{index},
     mCompletionPolicy{policy},
     mDistinctRoutesIndex{DataRelayerHelpers::createDistinctRouteIndex(routes)},
     mInputMatchers{DataRelayerHelpers::createInputMatchers(routes)},
@@ -102,6 +103,7 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
 {
   LOGP(debug, "DataRelayer::processDanglingInputs");
   std::scoped_lock<LockableBase(std::recursive_mutex)> lock(mMutex);
+  auto& deviceProxy = services.get<FairMQDeviceProxy>();
 
   ActivityStats activity;
   /// Nothing to do if nothing can expire.
@@ -115,7 +117,8 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
     LOGP(debug, "Creating new slot");
     for (auto& handler : expirationHandlers) {
       LOGP(debug, "handler.creator for {}", handler.name);
-      slotsCreatedByHandlers.push_back(handler.creator(mTimesliceIndex));
+      auto channelIndex = deviceProxy.getInputChannelIndex(handler.routeIndex);
+      slotsCreatedByHandlers.push_back(handler.creator(channelIndex, mTimesliceIndex));
     }
   }
   if (slotsCreatedByHandlers.empty() == false) {
diff --git a/Framework/Core/src/DataSender.cxx b/Framework/Core/src/DataSender.cxx
@@ -86,6 +86,7 @@ void DataSender::send(FairMQParts& parts, ChannelIndex channelIndex)
   /// FIXME: throttling this information?
   /// FIXME: do it only if it changes?
   TimesliceIndex& index = mRegistry.get<TimesliceIndex>();
+  index.updateOldestPossibleOutput();
 
   auto oldest = index.getOldestPossibleOutput();
   if (oldest.timeslice.value == -1) {
diff --git a/Framework/Core/src/LifetimeHelpers.cxx b/Framework/Core/src/LifetimeHelpers.cxx
@@ -56,7 +56,7 @@ size_t getCurrentTime()
 
 ExpirationHandler::Creator LifetimeHelpers::dataDrivenCreation()
 {
-  return [](TimesliceIndex&) -> TimesliceSlot {
+  return [](ChannelIndex, TimesliceIndex&) -> TimesliceSlot {
     return {TimesliceSlot::ANY};
   };
 }
@@ -66,7 +66,7 @@ ExpirationHandler::Creator LifetimeHelpers::enumDrivenCreation(size_t start, siz
   auto last = std::make_shared<size_t>(start + inputTimeslice * step);
   auto repetition = std::make_shared<size_t>(0);
 
-  return [end, step, last, maxInputTimeslices, maxRepetitions, repetition](TimesliceIndex& index) -> TimesliceSlot {
+  return [end, step, last, maxInputTimeslices, maxRepetitions, repetition](ChannelIndex channelIndex, TimesliceIndex& index) -> TimesliceSlot {
     for (size_t si = 0; si < index.size(); si++) {
       if (*last > end) {
         LOGP(debug, "Last greater than end");
@@ -81,6 +81,12 @@ ExpirationHandler::Creator LifetimeHelpers::enumDrivenCreation(size_t start, siz
         }
         LOGP(debug, "Associating timestamp {} to slot {}", timestamp.value, slot.index);
         index.associate(timestamp, slot);
+        // We know that next association will bring in last
+        // so we can state this will be the latest possible input for the channel
+        // associated with this.
+        LOG(debug) << "Oldest possible input is " << *last;
+        auto newOldest = index.setOldestPossibleInput({*last}, channelIndex);
+        index.updateOldestPossibleOutput();
         return slot;
       }
     }
@@ -95,7 +101,7 @@ ExpirationHandler::Creator LifetimeHelpers::timeDrivenCreation(std::chrono::micr
   auto start = getCurrentTime();
   auto last = std::make_shared<decltype(start)>(start);
   // FIXME: should create timeslices when period expires....
-  return [last, period](TimesliceIndex& index) -> TimesliceSlot {
+  return [last, period](ChannelIndex, TimesliceIndex& index) -> TimesliceSlot {
     // Nothing to do if the time has not expired yet.
     auto current = getCurrentTime();
     auto delta = current - *last;
@@ -192,7 +198,7 @@ ExpirationHandler::Checker LifetimeHelpers::expireIfPresent(std::vector<InputRou
 
 ExpirationHandler::Creator LifetimeHelpers::uvDrivenCreation(int requestedLoopReason, DeviceState& state)
 {
-  return [requestedLoopReason, &state](TimesliceIndex& index) -> TimesliceSlot {
+  return [requestedLoopReason, &state](ChannelIndex, TimesliceIndex& index) -> TimesliceSlot {
     /// Not the expected loop reason, return an invalid slot.
     if ((state.loopReason & requestedLoopReason) == 0) {
       LOGP(debug, "No expiration due to a loop event. Requested: {:b}, reported: {:b}, matching: {:b}",
@@ -354,7 +360,7 @@ ExpirationHandler::Handler
       timestamp = ceilf((VariableContextHelpers::getFirstTFOrbit(variables) * o2::constants::lhc::LHCOrbitNS / 1000 + dataTakingContext.orbitResetTime) / 1000);
     } else {
       // The timestamp used by DPL is in nanoseconds
-      timestamp = ceilf(VariableContextHelpers::getTimeslice(variables).value / 1000);
+      timestamp = ceilf(VariableContextHelpers::getTimeslice(variables).value / 1000.);
     }
 
     std::string path = "";
@@ -500,9 +506,6 @@ ExpirationHandler::Handler LifetimeHelpers::enumerate(ConcreteDataMatcher const&
     ref.payload = std::move(payload);
     (*counter)++;
 
-    auto& timesliceIndex = services.get<TimesliceIndex>();
-    auto newOldest = timesliceIndex.setOldestPossibleInput({dph.startTime}, channelIndex);
-    timesliceIndex.updateOldestPossibleOutput();
   };
 }
 
@@ -552,10 +555,6 @@ ExpirationHandler::Handler LifetimeHelpers::dummy(ConcreteDataMatcher const& mat
     ref.header = std::move(header);
     auto payload = transport->CreateMessage(0);
     ref.payload = std::move(payload);
-
-    auto& timesliceIndex = services.get<TimesliceIndex>();
-    auto newOldest = timesliceIndex.setOldestPossibleInput({dph.startTime}, channelIndex);
-    timesliceIndex.updateOldestPossibleOutput();
   };
   return f;
 }
diff --git a/Framework/TestWorkflows/src/o2DiamondWorkflow.cxx b/Framework/TestWorkflows/src/o2DiamondWorkflow.cxx
@@ -17,6 +17,7 @@
 #include "Framework/Configurable.h"
 #include "Framework/RunningWorkflowInfo.h"
 #include "Framework/CallbackService.h"
+#include "Framework/RateLimiter.h"
 #include <fairmq/Device.h>
 
 #include <iostream>
@@ -43,16 +44,16 @@ void customize(std::vector<CallbacksPolicy>& policies)
     }});
 }
 
-void customize(std::vector<SendingPolicy>& policies)
-{
-  policies.push_back(SendingPolicy{
-    .matcher = DeviceMatchers::matchByName("A"),
-    .send = [](FairMQDeviceProxy& proxy, FairMQParts& parts, ChannelIndex channelIndex) {
-      LOG(info) << "A custom policy for sending invoked!";
-      auto* channel = proxy.getOutputChannel(channelIndex);
-      channel->Send(parts, 0);
-    }});
-}
+// void customize(std::vector<SendingPolicy>& policies)
+//{
+//   policies.push_back(SendingPolicy{
+//     .matcher = DeviceMatchers::matchByName("A"),
+//     .send = [](FairMQDeviceProxy& proxy, FairMQParts& parts, ChannelIndex channelIndex) {
+//       LOG(info) << "A custom policy for sending invoked!";
+//       auto* channel = proxy.getOutputChannel(channelIndex);
+//       channel->Send(parts, 0);
+//     }});
+// }
 
 #include "Framework/runDataProcessing.h"
 
@@ -77,12 +78,15 @@ WorkflowSpec defineDataProcessing(ConfigContext const& specs)
     .outputs = {OutputSpec{{"a1"}, "TST", "A1"},
                 OutputSpec{{"a2"}, "TST", "A2"}},
     .algorithm = AlgorithmSpec{adaptStateless(
-      [](DataAllocator& outputs, RawDeviceService& device, DataTakingContext& context) {
-        device.device()->WaitFor(std::chrono::seconds(rand() % 2));
+      [](DataAllocator& outputs, RawDeviceService& device, DataTakingContext& context, ProcessingContext& pcx) {
+        // static RateLimiter limiter;
+        // limiter.check(pcx, std::stoi(device.device()->fConfig->GetValue<std::string>("timeframes-rate-limit")), 2000);
+        LOG(error) << "Foo";
         auto& aData = outputs.make<int>(OutputRef{"a1"}, 1);
         auto& bData = outputs.make<int>(OutputRef{"a2"}, 1);
       })},
-    .options = {ConfigParamSpec{"some-device-param", VariantType::Int, 1, {"Some device parameter"}}}};
+    .options = {ConfigParamSpec{"some-device-param", VariantType::Int, 1, {"Some device parameter"}},
+                ConfigParamSpec{"channel-config", VariantType::String, "name=metric-feedback", {"Timeframes per second limit"}}}};
   DataProcessorSpec b{
     .name = "B",
     .inputs = {InputSpec{"x", "TST", "A1", Lifetime::Timeframe, {ConfigParamSpec{"somestring", VariantType::String, "", {"Some input param"}}}}},

Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@`
`40`	`40`	`namespace o2::framework`
`41`	`41`	`{`
`42`	`42`
`43`		`-class Output;`
	`43`	`+struct Output;`
`44`	`44`
`45`	`45`	`class MessageContext`
`46`	`46`	`{`
Original file line number	Diff line number	Diff line change
`@@ -30,8 +30,11 @@ class ProcessingContext`
`30`	`30`	`{`
`31`	`31`	`}`
`32`	`32`
	`33`	`+ /// The inputs associated with this processing context.`
`33`	`34`	`InputRecord& inputs() { return mInputs; }`
	`35`	`+ /// The services registry associated with this processing context.`
`34`	`36`	`ServiceRegistry& services() { return mServices; }`
	`37`	`+ /// The data allocator is used to allocate memory for the output data.`
`35`	`38`	`DataAllocator& outputs() { return mAllocator; }`
`36`	`39`
`37`	`40`	`InputRecord& mInputs;`