DPL: allow custom orbit when using timer / enumeration

The firstTForbit field is now given by the formula:

firstTForbit = orbit-offset-enumeration + orbit-multiplier-enumeration + dh.tfCounter

for the enumeration and timer cases. One can use, e.g.:

--orbit-offset-enumeration and --orbit-multiplier-enumeration

on the command line to affect the two parameters.

Author: ktf

Framework/Core/include/Framework/ConfigParamsHelper.h

@@ -36,6 +36,10 @@ struct ConfigParamsHelper {
                                options_description& options,
                                boost::program_options::options_description vetos = options_description());
 
+  /// Add the ConfigParamSpec @a spec to @a specs if there is no parameter with
+  /// the same name already.
+  static void addOptionIfMissing(std::vector<ConfigParamSpec>& specs, ConfigParamSpec spec);
+
   /// populate boost program options for a complete workflow
   template <typename ContainerType>
   static boost::program_options::options_description

Framework/Core/include/Framework/DataDescriptorMatcher.h

@@ -41,6 +41,13 @@ struct ContextRef {
   inline bool operator==(ContextRef const& other) const;
 };
 
+/// Special positions for variables in context.
+enum ContextPos {
+  STARTTIME_POS = 0,    /// The DataProcessingHeader::startTime associated to the timeslice
+  TFCOUNTER_POS = 14,   /// The DataHeader::tfCounter associated to the timeslice
+  FIRSTTFORBIT_POS = 15 /// The DataHeader::firstTFOrbit associated to the timeslice
+};
+
 /// An element of the matching context. Context itself is really a vector of
 /// those. It's up to the matcher builder to build the vector in a suitable way.
 /// We do not have any float in the value, because AFAICT there is no need for

Framework/Core/include/Framework/ExpirationHandler.h

@@ -8,17 +8,16 @@
 // granted to it by virtue of its status as an Intergovernmental Organization
 // or submit itself to any jurisdiction.
 
-#ifndef FRAMEWORK_EXPIRATIONHANDLER_H
-#define FRAMEWORK_EXPIRATIONHANDLER_H
+#ifndef O2_FRAMEWORK_EXPIRATIONHANDLER_H_
+#define O2_FRAMEWORK_EXPIRATIONHANDLER_H_
 
 #include "Framework/Lifetime.h"
 #include "Framework/RoutingIndices.h"
+#include "Framework/DataDescriptorMatcher.h"
 #include <cstdint>
 #include <functional>
 
-namespace o2
-{
-namespace framework
+namespace o2::framework
 {
 
 struct PartRef;
@@ -29,7 +28,7 @@ struct TimesliceSlot;
 struct ExpirationHandler {
   using Creator = std::function<TimesliceSlot(TimesliceIndex&)>;
   using Checker = std::function<bool(uint64_t timestamp)>;
-  using Handler = std::function<void(ServiceRegistry&, PartRef& expiredInput, uint64_t timestamp)>;
+  using Handler = std::function<void(ServiceRegistry&, PartRef& expiredInput, uint64_t timestamp, data_matcher::VariableContext& variables)>;
 
   RouteIndex routeIndex;
   Lifetime lifetime;
@@ -38,7 +37,6 @@ struct ExpirationHandler {
   Handler handler;
 };
 
-} // namespace framework
-} // namespace o2
+} // namespace o2::framework
 
-#endif // FRAMEWORK_EXPIRATIONHANDLER_H
+#endif // O2_FRAMEWORK_EXPIRATIONHANDLER_H_

Framework/Core/include/Framework/LifetimeHelpers.h

@@ -76,7 +76,8 @@ struct LifetimeHelpers {
   /// dataOrigin, dataDescrition and dataSpecification of the given @a route.
   /// The payload of each message will contain an incremental number for each
   /// message being created.
-  static ExpirationHandler::Handler enumerate(ConcreteDataMatcher const& spec, std::string const& sourceChannel);
+  static ExpirationHandler::Handler enumerate(ConcreteDataMatcher const& spec, std::string const& sourceChannel,
+                                              int64_t orbitOffset, int64_t orbitMultiplier);
 
   /// Create a dummy (empty) message every time a record expires, suing @a spec
   /// as content of the payload.

Framework/Core/include/Framework/TimesliceIndex.inc

@@ -122,7 +122,7 @@ inline uint32_t TimesliceIndex::getFirstTFOrbitForSlot(TimesliceSlot slot) const
 {
   assert(mVariables.size() > slot.index);
   // firstTForbit is always at register 15
-  auto pval = std::get_if<uint32_t>(&mVariables[slot.index].get(15));
+  auto pval = std::get_if<uint32_t>(&mVariables[slot.index].get(data_matcher::FIRSTTFORBIT_POS));
   if (pval == nullptr) {
     return -1;
   }
@@ -132,7 +132,7 @@ inline uint32_t TimesliceIndex::getFirstTFOrbitForSlot(TimesliceSlot slot) const
 inline uint32_t TimesliceIndex::getFirstTFCounterForSlot(TimesliceSlot slot) const
 {
   assert(mVariables.size() > slot.index);
-  // firstTForbit is always at register 15
+  // tfCounter is always at register 14
   auto pval = std::get_if<uint32_t>(&mVariables[slot.index].get(14));
   if (pval == nullptr) {
     return -1;

Framework/Core/src/ConfigParamsHelper.cxx

@@ -89,6 +89,16 @@ void ConfigParamsHelper::populateBoostProgramOptions(
   }
 }
 
+void ConfigParamsHelper::addOptionIfMissing(std::vector<ConfigParamSpec>& specs, ConfigParamSpec spec)
+{
+  for (auto& old : specs) {
+    if (old.name == spec.name) {
+      return;
+    }
+  }
+  specs.push_back(spec);
+}
+
 /// populate boost program options making all options of type string
 /// this is used for filtering the command line argument
 bool ConfigParamsHelper::dpl2BoostOptions(const std::vector<ConfigParamSpec>& spec,

Framework/Core/src/DataDescriptorMatcher.cxx

@@ -106,9 +106,9 @@ bool StartTimeValueMatcher::match(header::DataHeader const& dh, DataProcessingHe
     }
     context.put({ref->index, dph.startTime / mScale});
     // We always put in 14 the tfCounter
-    context.put({14, dh.tfCounter});
+    context.put({TFCOUNTER_POS, dh.tfCounter});
     // We always put in 15 the firstTForbit
-    context.put({15, dh.firstTForbit});
+    context.put({FIRSTTFORBIT_POS, dh.firstTForbit});
     return true;
   } else if (auto v = std::get_if<uint64_t>(&mValue)) {
     return (dph.startTime / mScale) == *v;

Framework/Core/src/DataRelayer.cxx

@@ -108,6 +108,7 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
     }
     assert(mDistinctRoutesIndex.empty() == false);
     auto timestamp = mTimesliceIndex.getTimesliceForSlot(slot);
+    auto& variables = mTimesliceIndex.getVariablesForSlot(slot);
     // We iterate on all the hanlders checking if they need to be expired.
     for (size_t ei = 0; ei < expirationHandlers.size(); ++ei) {
       auto& expirator = expirationHandlers[ei];
@@ -137,7 +138,7 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
       if (part.size() == 0) {
         part.parts.resize(1);
       }
-      expirator.handler(services, part[0], timestamp.value);
+      expirator.handler(services, part[0], timestamp.value, variables);
       activity.expiredSlots++;
 
       mTimesliceIndex.markAsDirty(slot, true);

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -177,7 +177,10 @@ struct ExpirationHandlerHelpers {
       throw runtime_error("InputSpec for Timers must be fully qualified");
     }
     // We copy the matcher to avoid lifetime issues.
-    return [matcher = *m, sourceChannel](DeviceState&, ConfigParamRegistry const&) { return LifetimeHelpers::enumerate(matcher, sourceChannel); };
+    return [matcher = *m, sourceChannel](DeviceState&, ConfigParamRegistry const& config) {
+      // Timers do not have any orbit associated to them
+      return LifetimeHelpers::enumerate(matcher, sourceChannel, 0, 0);
+    };
   }
 
   static RouteConfigurator::ExpirationConfigurator expiringEnumerationConfigurator(InputSpec const& spec, std::string const& sourceChannel)
@@ -187,8 +190,10 @@ struct ExpirationHandlerHelpers {
       throw runtime_error("InputSpec for Enumeration must be fully qualified");
     }
     // We copy the matcher to avoid lifetime issues.
-    return [matcher = *m, sourceChannel](DeviceState&, ConfigParamRegistry const&) {
-      return LifetimeHelpers::enumerate(matcher, sourceChannel);
+    return [matcher = *m, sourceChannel](DeviceState&, ConfigParamRegistry const& config) {
+      size_t orbitOffset = config.get<int64_t>("orbit-offset-enumeration");
+      size_t orbitMultiplier = config.get<int64_t>("orbit-multiplier-enumeration");
+      return LifetimeHelpers::enumerate(matcher, sourceChannel, orbitOffset, orbitMultiplier);
     };
   }
 
@@ -222,17 +227,17 @@ struct ExpirationHandlerHelpers {
   {
     try {
       ConcreteDataMatcher concrete = DataSpecUtils::asConcreteDataMatcher(spec);
-      return [concrete, sourceChannel](DeviceState&, ConfigParamRegistry const&) {
+      return [concrete, sourceChannel](DeviceState&, ConfigParamRegistry const& config) {
         return LifetimeHelpers::dummy(concrete, sourceChannel);
       };
     } catch (...) {
       ConcreteDataTypeMatcher dataType = DataSpecUtils::asConcreteDataTypeMatcher(spec);
       ConcreteDataMatcher concrete{dataType.origin, dataType.description, 0xdeadbeef};
-      return [concrete, sourceChannel](DeviceState&, ConfigParamRegistry const&) {
+      return [concrete, sourceChannel](DeviceState&, ConfigParamRegistry const& config) {
         return LifetimeHelpers::dummy(concrete, sourceChannel);
       };
+      // We copy the matcher to avoid lifetime issues.
     }
-    // We copy the matcher to avoid lifetime issues.
   }
 };
 

Framework/Core/src/LifetimeHelpers.cxx

@@ -11,6 +11,7 @@
 #include "Framework/DataProcessingHeader.h"
 #include "Framework/InputSpec.h"
 #include "Framework/LifetimeHelpers.h"
+#include "Framework/DataDescriptorMatcher.h"
 #include "Framework/Logger.h"
 #include "Framework/RawDeviceService.h"
 #include "Framework/ServiceRegistry.h"
@@ -143,7 +144,7 @@ ExpirationHandler::Checker LifetimeHelpers::expireTimed(std::chrono::microsecond
 /// expires via this mechanism).
 ExpirationHandler::Handler LifetimeHelpers::doNothing()
 {
-  return [](ServiceRegistry&, PartRef& ref, uint64_t) -> void { return; };
+  return [](ServiceRegistry&, PartRef& ref, uint64_t, data_matcher::VariableContext&) -> void { return; };
 }
 
 // We simply put everything in a stringstream and read it afterwards.
@@ -188,7 +189,7 @@ ExpirationHandler::Handler
   if (matcher == nullptr) {
     throw runtime_error("InputSpec for Conditions must be fully qualified");
   }
-  return [spec, matcher, sourceChannel, serverUrl = prefix, overrideTimestampMilliseconds](ServiceRegistry& services, PartRef& ref, uint64_t timestamp) -> void {
+  return [spec, matcher, sourceChannel, serverUrl = prefix, overrideTimestampMilliseconds](ServiceRegistry& services, PartRef& ref, uint64_t timestamp, data_matcher::VariableContext&) -> void {
     // We should invoke the handler only once.
     assert(!ref.header);
     assert(!ref.payload);
@@ -265,7 +266,7 @@ ExpirationHandler::Handler
 /// FIXME: provide a way to customise the histogram from the configuration.
 ExpirationHandler::Handler LifetimeHelpers::fetchFromQARegistry()
 {
-  return [](ServiceRegistry&, PartRef& ref, uint64_t) -> void {
+  return [](ServiceRegistry&, PartRef& ref, uint64_t, data_matcher::VariableContext&) -> void {
     throw runtime_error("fetchFromQARegistry: Not yet implemented");
     return;
   };
@@ -276,18 +277,19 @@ ExpirationHandler::Handler LifetimeHelpers::fetchFromQARegistry()
 /// FIXME: provide a way to customise the histogram from the configuration.
 ExpirationHandler::Handler LifetimeHelpers::fetchFromObjectRegistry()
 {
-  return [](ServiceRegistry&, PartRef& ref, uint64_t) -> void {
+  return [](ServiceRegistry&, PartRef& ref, uint64_t, data_matcher::VariableContext&) -> void {
     throw runtime_error("fetchFromObjectRegistry: Not yet implemented");
     return;
   };
 }
 
 /// Enumerate entries on every invokation.
-ExpirationHandler::Handler LifetimeHelpers::enumerate(ConcreteDataMatcher const& matcher, std::string const& sourceChannel)
+ExpirationHandler::Handler LifetimeHelpers::enumerate(ConcreteDataMatcher const& matcher, std::string const& sourceChannel,
+                                                      int64_t orbitOffset, int64_t orbitMultiplier)
 {
   using counter_t = int64_t;
   auto counter = std::make_shared<counter_t>(0);
-  auto f = [matcher, counter, sourceChannel](ServiceRegistry& services, PartRef& ref, uint64_t timestamp) -> void {
+  return [matcher, counter, sourceChannel, orbitOffset, orbitMultiplier](ServiceRegistry& services, PartRef& ref, uint64_t timestamp, data_matcher::VariableContext& variables) -> void {
     // We should invoke the handler only once.
     assert(!ref.header);
     assert(!ref.payload);
@@ -299,6 +301,10 @@ ExpirationHandler::Handler LifetimeHelpers::enumerate(ConcreteDataMatcher const&
     dh.subSpecification = matcher.subSpec;
     dh.payloadSize = sizeof(counter_t);
     dh.payloadSerializationMethod = gSerializationMethodNone;
+    dh.tfCounter = timestamp;
+    dh.firstTForbit = timestamp * orbitMultiplier + orbitOffset;
+    variables.put({data_matcher::FIRSTTFORBIT_POS, dh.firstTForbit});
+    variables.put({data_matcher::TFCOUNTER_POS, dh.tfCounter});
 
     DataProcessingHeader dph{timestamp, 1};
 
@@ -312,15 +318,14 @@ ExpirationHandler::Handler LifetimeHelpers::enumerate(ConcreteDataMatcher const&
     ref.payload = std::move(payload);
     (*counter)++;
   };
-  return f;
 }
 
 /// Create a dummy message with the provided ConcreteDataMatcher
 ExpirationHandler::Handler LifetimeHelpers::dummy(ConcreteDataMatcher const& matcher, std::string const& sourceChannel)
 {
   using counter_t = int64_t;
   auto counter = std::make_shared<counter_t>(0);
-  auto f = [matcher, counter, sourceChannel](ServiceRegistry& services, PartRef& ref, uint64_t timestamp) -> void {
+  auto f = [matcher, counter, sourceChannel](ServiceRegistry& services, PartRef& ref, uint64_t timestamp, data_matcher::VariableContext& variables) -> void {
     // We should invoke the handler only once.
     assert(!ref.header);
     assert(!ref.payload);
@@ -332,7 +337,23 @@ ExpirationHandler::Handler LifetimeHelpers::dummy(ConcreteDataMatcher const& mat
     dh.subSpecification = matcher.subSpec;
     dh.payloadSize = 0;
     dh.payloadSerializationMethod = gSerializationMethodNone;
-    dh.tfCounter = timestamp;
+
+    {
+      auto pval = std::get_if<uint32_t>(&variables.get(data_matcher::FIRSTTFORBIT_POS));
+      if (pval == nullptr) {
+        dh.firstTForbit = -1;
+      } else {
+        dh.firstTForbit = *pval;
+      }
+    }
+    {
+      auto pval = std::get_if<uint32_t>(&variables.get(data_matcher::TFCOUNTER_POS));
+      if (pval == nullptr) {
+        dh.tfCounter = timestamp;
+      } else {
+        dh.tfCounter = *pval;
+      }
+    }
 
     DataProcessingHeader dph{timestamp, 1};