DPL: prevent InputSpec default constructor

As we move closer to have InputSpec as a veritable "matcher" object
rather than a set of variables to be equal compared, I need to tight
the access to the member variables. The first adiabatic step in such a
direction is to make sure an InputSpec is built all at once, and not
incrementally, as that ability will be lost once we move to use
a DataDescriptorMatcher to describe the InputSpec internal State.

Author: ktf

Framework/Core/include/Framework/InputSpec.h

@@ -25,11 +25,22 @@ namespace framework
 /// input or in output. This can be used, for example to match
 /// specific payloads in a timeframe.
 struct InputSpec {
+  /// This is the legacy way to construct things. For the moment we still allow
+  /// accessing directly the members, but this will change as well at some point.
+  InputSpec(std::string binding_, header::DataOrigin origin_, header::DataDescription description_, header::DataHeader::SubSpecificationType subSpec_ = 0, enum Lifetime lifetime_ = Lifetime::Timeframe)
+    : binding{binding_},
+      origin{origin_},
+      description{description_},
+      subSpec{subSpec_},
+      lifetime{lifetime_}
+  {
+  }
+
   std::string binding;
   header::DataOrigin origin;
   header::DataDescription description;
-  header::DataHeader::SubSpecificationType subSpec = 0;
-  enum Lifetime lifetime = Lifetime::Timeframe;
+  header::DataHeader::SubSpecificationType subSpec;
+  enum Lifetime lifetime;
 
   bool operator==(InputSpec const& that) const
   {

Framework/Core/include/Framework/LifetimeHelpers.h

@@ -13,15 +13,16 @@
 #include "Framework/ExpirationHandler.h"
 #include "Framework/PartRef.h"
 
-#include <functional>
 #include <chrono>
+#include <functional>
+#include <string>
 
 namespace o2
 {
 namespace framework
 {
 
-struct InputRoute;
+struct InputSpec;
 
 /// Lifetime handlers are used to manage the cases in which data is not coming
 /// from the dataflow, but from some other source or trigger, e.g.,
@@ -58,13 +59,13 @@ struct LifetimeHelpers {
   /// FIXME: provide a way to customise the histogram from the configuration.
   static ExpirationHandler::Handler fetchFromObjectRegistry();
 
-  /// Enumerate entries on every invokation.  @a route is the route which the
+  /// Enumerate entries on every invokation.  @a matcher is the InputSpec which the
   /// given enumeration refers to. In particular messages created by the
   /// returned ExpirationHandler will have an header which matches the
   /// 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(InputRoute const& route);
+  static ExpirationHandler::Handler enumerate(InputSpec const& spec, std::string const& sourceChannel);
 };
 
 } // namespace framework

Framework/Core/src/DataSampling.cxx

@@ -185,21 +185,24 @@ QcTaskConfigurations DataSampling::readQcTasksConfiguration(const std::string& c
 
     for (auto&& input : taskInputsSplit) {
 
-      InputSpec desiredData;
-      try {
-        desiredData.binding = configFile->get<std::string>(prefixConfigTasks + input + ".inputName");
-
-        std::string origin = configFile->get<std::string>(prefixConfigTasks + input + ".dataOrigin");
-        origin.copy(desiredData.origin.str, (size_t)desiredData.origin.size);
+      std::string binding;
+      header::DataOrigin origin;
+      header::DataDescription description;
 
-        std::string description = configFile->get<std::string>(prefixConfigTasks + input + ".dataDescription");
-        description.copy(desiredData.description.str, (size_t)desiredData.description.size);
+      try {
+        binding = configFile->get<std::string>(prefixConfigTasks + input + ".inputName");
+        std::string originStr = configFile->get<std::string>(prefixConfigTasks + input + ".dataOrigin");
+        std::string descriptionStr = configFile->get<std::string>(prefixConfigTasks + input + ".dataDescription");
 
+        originStr.copy(origin.str, (size_t)origin.size);
+        descriptionStr.copy(description.str, (size_t)description.size);
       } catch (...) {
         LOG(ERROR) << "QC Task configuration error. In file " << configurationSource << " input " << input
                    << " has missing values";
         continue;
       }
+
+      InputSpec desiredData{binding, origin, description, 0};
       task.desiredDataSpecs.push_back(desiredData);
 
       // for temporary feature

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -58,10 +58,10 @@ struct ExpirationHandlerHelpers {
     return [](ConfigParamRegistry const&) { return LifetimeHelpers::expireNever(); };
   }
 
-  static InputRoute::ExpirationConfigurator expiringConditionConfigurator(InputRoute& route)
+  static InputRoute::ExpirationConfigurator expiringConditionConfigurator(InputSpec const& matcher)
   {
-    return [route](ConfigParamRegistry const&) {
-      std::string prefix = std::string{ "/" } + route.matcher.origin.str + "/" + route.matcher.description.str;
+    return [matcher](ConfigParamRegistry const&) {
+      std::string prefix = std::string{ "/" } + matcher.origin.str + "/" + matcher.description.str;
       return LifetimeHelpers::fetchFromCCDBCache(prefix);
     };
   }
@@ -79,18 +79,18 @@ struct ExpirationHandlerHelpers {
     return [](ConfigParamRegistry const&) { return LifetimeHelpers::fetchFromQARegistry(); };
   }
 
-  static InputRoute::DanglingConfigurator danglingTimerConfigurator(InputRoute& route)
+  static InputRoute::DanglingConfigurator danglingTimerConfigurator(InputSpec const& matcher)
   {
-    return [route](ConfigParamRegistry const& options) {
-      std::string rateName = std::string{ "period-" } + route.matcher.origin.str + "-" + route.matcher.description.str;
+    return [matcher](ConfigParamRegistry const& options) {
+      std::string rateName = std::string{ "period-" } + matcher.binding;
       auto period = options.get<int>(rateName.c_str());
       return LifetimeHelpers::expireTimed(std::chrono::milliseconds(period));
     };
   }
 
-  static InputRoute::ExpirationConfigurator expiringTimerConfigurator(InputRoute& route)
+  static InputRoute::ExpirationConfigurator expiringTimerConfigurator(InputSpec const& matcher, std::string const& sourceChannel)
   {
-    return [route](ConfigParamRegistry const&) { return LifetimeHelpers::enumerate(route); };
+    return [matcher, sourceChannel](ConfigParamRegistry const&) { return LifetimeHelpers::enumerate(matcher, sourceChannel); };
   }
 
   static InputRoute::DanglingConfigurator danglingTransientConfigurator()
@@ -101,7 +101,7 @@ struct ExpirationHandlerHelpers {
     return [](ConfigParamRegistry const&) { return LifetimeHelpers::expireNever(); };
   }
 
-  static InputRoute::ExpirationConfigurator expiringTransientConfigurator(InputRoute& route)
+  static InputRoute::ExpirationConfigurator expiringTransientConfigurator(InputSpec const& matcher)
   {
     return [](ConfigParamRegistry const&) { return LifetimeHelpers::fetchFromObjectRegistry(); };
   }
@@ -252,11 +252,12 @@ void DeviceSpecHelpers::processOutEdgeActions(std::vector<DeviceSpec>& devices,
       };
       device.outputs.emplace_back(route);
     } else {
-      ForwardRoute route;
-      route.matcher = workflow[edge.consumer].inputs[edge.consumerInputIndex];
-      route.channel = channel.name;
-      route.timeslice = edge.timeIndex;
-      route.maxTimeslices = consumer.maxInputTimeslices;
+      ForwardRoute route{
+        edge.timeIndex,
+        consumer.maxInputTimeslices,
+        workflow[edge.consumer].inputs[edge.consumerInputIndex],
+        channel.name
+      };
       device.forwards.emplace_back(route);
     }
   };
@@ -423,32 +424,43 @@ void DeviceSpecHelpers::processInEdgeActions(std::vector<DeviceSpec>& devices,
     auto const& edge = logicalEdges[ei];
     auto const& consumer = workflow[edge.consumer];
     auto& consumerDevice = devices[di];
-    InputRoute route;
-    route.matcher = consumer.inputs[edge.consumerInputIndex];
-    route.sourceChannel = consumerDevice.inputChannels[ci].name;
-    route.timeslice = edge.producerTimeIndex;
+
+    InputRoute::DanglingConfigurator danglingConfigurator;
+    InputRoute::ExpirationConfigurator expirationConfigurator;
+    auto const& inputSpec = consumer.inputs[edge.consumerInputIndex];
+    auto const& sourceChannel = consumerDevice.inputChannels[ci].name;
+
     switch (consumer.inputs[edge.consumerInputIndex].lifetime) {
       case Lifetime::Timeframe:
-        route.danglingConfigurator = ExpirationHandlerHelpers::danglingTimeframeConfigurator();
-        route.expirationConfigurator = ExpirationHandlerHelpers::expiringTimeframeConfigurator();
+        danglingConfigurator = ExpirationHandlerHelpers::danglingTimeframeConfigurator();
+        expirationConfigurator = ExpirationHandlerHelpers::expiringTimeframeConfigurator();
         break;
       case Lifetime::Condition:
-        route.danglingConfigurator = ExpirationHandlerHelpers::danglingConditionConfigurator();
-        route.expirationConfigurator = ExpirationHandlerHelpers::expiringConditionConfigurator(route);
+        danglingConfigurator = ExpirationHandlerHelpers::danglingConditionConfigurator();
+        expirationConfigurator = ExpirationHandlerHelpers::expiringConditionConfigurator(inputSpec);
         break;
       case Lifetime::QA:
-        route.danglingConfigurator = ExpirationHandlerHelpers::danglingQAConfigurator();
-        route.expirationConfigurator = ExpirationHandlerHelpers::expiringQAConfigurator();
+        danglingConfigurator = ExpirationHandlerHelpers::danglingQAConfigurator();
+        expirationConfigurator = ExpirationHandlerHelpers::expiringQAConfigurator();
         break;
       case Lifetime::Timer:
-        route.danglingConfigurator = ExpirationHandlerHelpers::danglingTimerConfigurator(route);
-        route.expirationConfigurator = ExpirationHandlerHelpers::expiringTimerConfigurator(route);
+        danglingConfigurator = ExpirationHandlerHelpers::danglingTimerConfigurator(inputSpec);
+        expirationConfigurator = ExpirationHandlerHelpers::expiringTimerConfigurator(inputSpec, sourceChannel);
         break;
       case Lifetime::Transient:
-        route.danglingConfigurator = ExpirationHandlerHelpers::danglingTransientConfigurator();
-        route.expirationConfigurator = ExpirationHandlerHelpers::expiringTransientConfigurator(route);
+        danglingConfigurator = ExpirationHandlerHelpers::danglingTransientConfigurator();
+        expirationConfigurator = ExpirationHandlerHelpers::expiringTransientConfigurator(inputSpec);
         break;
     }
+
+    InputRoute route{
+      inputSpec,
+      sourceChannel,
+      edge.producerTimeIndex,
+      danglingConfigurator,
+      expirationConfigurator
+    };
+
     consumerDevice.inputs.push_back(route);
   };
 

Framework/Core/src/LifetimeHelpers.cxx

@@ -11,7 +11,7 @@
 #include "Framework/LifetimeHelpers.h"
 #include "Framework/ServiceRegistry.h"
 #include "Framework/RawDeviceService.h"
-#include "Framework/InputRoute.h"
+#include "Framework/InputSpec.h"
 #include "Headers/DataHeader.h"
 #include "Headers/Stack.h"
 #include "MemoryResources/MemoryResources.h"
@@ -107,25 +107,25 @@ ExpirationHandler::Handler LifetimeHelpers::fetchFromObjectRegistry()
 }
 
 /// Enumerate entries on every invokation.
-ExpirationHandler::Handler LifetimeHelpers::enumerate(InputRoute const& route)
+ExpirationHandler::Handler LifetimeHelpers::enumerate(InputSpec const& matcher, std::string const& sourceChannel)
 {
   auto counter = std::make_shared<int64_t>(0);
-  auto f = [route, counter](ServiceRegistry& services, PartRef& ref, uint64_t timestamp) -> void {
+  auto f = [matcher, counter, sourceChannel](ServiceRegistry& services, PartRef& ref, uint64_t timestamp) -> void {
     // We should invoke the handler only once.
     assert(!ref.header);
     assert(!ref.payload);
     auto& rawDeviceService = services.get<RawDeviceService>();
 
     DataHeader dh;
-    dh.dataOrigin = route.matcher.origin;
-    dh.dataDescription = route.matcher.description;
-    dh.subSpecification = route.matcher.subSpec;
+    dh.dataOrigin = matcher.origin;
+    dh.dataDescription = matcher.description;
+    dh.subSpecification = matcher.subSpec;
     dh.payloadSize = 8;
     dh.payloadSerializationMethod = gSerializationMethodNone;
 
     DataProcessingHeader dph{ timestamp, 1 };
 
-    auto&& transport = rawDeviceService.device()->GetChannel(route.sourceChannel, 0).Transport();
+    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 });
     ref.header = std::move(header);

Framework/Core/test/test_DataRelayer.cxx

@@ -32,21 +32,12 @@ using Stack = o2::header::Stack;
 // and the subsequent InputRecord is immediately requested.
 BOOST_AUTO_TEST_CASE(TestNoWait) {
   Monitoring metrics;
-  InputSpec spec;
-  spec.binding = "clusters";
-  spec.description = "CLUSTERS";
-  spec.origin = "TPC";
-  spec.subSpec = 0;
-  spec.lifetime = Lifetime::Timeframe;
-
-  InputRoute route;
-  route.sourceChannel = "Fake";
-  route.matcher = spec;
-  route.timeslice = 0;
+  InputSpec spec{ "clusters", "TPC", "CLUSTERS" };
 
   std::vector<InputRoute> inputs = {
-    route
+    InputRoute{ spec, "Fake", 0 }
   };
+
   std::vector<ForwardRoute> forwards;
   TimesliceIndex index;
 
@@ -81,31 +72,22 @@ BOOST_AUTO_TEST_CASE(TestNoWait) {
 // correctly relayed before being processed.
 BOOST_AUTO_TEST_CASE(TestRelay) {
   Monitoring metrics;
-  InputSpec spec1;
-  spec1.binding = "clusters";
-  spec1.description = "CLUSTERS";
-  spec1.origin = "TPC";
-  spec1.subSpec = 0;
-  spec1.lifetime = Lifetime::Timeframe;
-
-  InputSpec spec2;
-  spec2.binding = "clusters_its";
-  spec2.description = "CLUSTERS";
-  spec2.origin = "ITS";
-  spec2.subSpec = 0;
-  spec2.lifetime = Lifetime::Timeframe;
-
-  InputRoute route1;
-  route1.sourceChannel = "Fake";
-  route1.matcher = spec1;
-  route1.timeslice = 0;
-
-  InputRoute route2;
-  route2.sourceChannel = "Fake";
-  route2.matcher = spec2;
-  route2.timeslice = 0;
-
-  std::vector<InputRoute> inputs = { route1, route2 };
+  InputSpec spec1{
+    "clusters",
+    "TPC",
+    "CLUSTERS",
+  };
+  InputSpec spec2{
+    "clusters_its",
+    "ITS",
+    "CLUSTERS",
+  };
+
+  std::vector<InputRoute> inputs = {
+    InputRoute{ spec1, "Fake1", 0 },
+    InputRoute{ spec2, "Fake2", 0 }
+  };
+
   std::vector<ForwardRoute> forwards;
 
   TimesliceIndex index;
@@ -157,20 +139,10 @@ BOOST_AUTO_TEST_CASE(TestRelay) {
 // the cache.
 BOOST_AUTO_TEST_CASE(TestCache) {
   Monitoring metrics;
-  InputSpec spec;
-  spec.binding = "clusters";
-  spec.description = "CLUSTERS";
-  spec.origin = "TPC";
-  spec.subSpec = 0;
-  spec.lifetime = Lifetime::Timeframe;
-
-  InputRoute route;
-  route.sourceChannel = "Fake";
-  route.matcher = spec;
-  route.timeslice = 0;
+  InputSpec spec{ "clusters", "TPC", "CLUSTERS" };
 
   std::vector<InputRoute> inputs = {
-    route
+    InputRoute{ spec, "Fake", 0 }
   };
   std::vector<ForwardRoute> forwards;
 
@@ -231,34 +203,14 @@ BOOST_AUTO_TEST_CASE(TestCache) {
 // it will run immediately.
 BOOST_AUTO_TEST_CASE(TestPolicies) {
   Monitoring metrics;
-  InputSpec spec1;
-  spec1.binding = "clusters";
-  spec1.description = "CLUSTERS";
-  spec1.origin = "TPC";
-  spec1.subSpec = 0;
-  spec1.lifetime = Lifetime::Timeframe;
-
-  InputSpec spec2;
-  spec2.binding = "tracks";
-  spec2.description = "TRACKS";
-  spec2.origin = "TPC";
-  spec2.subSpec = 0;
-  spec2.lifetime = Lifetime::Timeframe;
-
-  InputRoute route1;
-  route1.sourceChannel = "Fake";
-  route1.matcher = spec1;
-  route1.timeslice = 0;
-
-  InputRoute route2;
-  route2.sourceChannel = "Fake2";
-  route2.matcher = spec2;
-  route2.timeslice = 0;
+  InputSpec spec1{ "clusters", "TPC", "CLUSTERS" };
+  InputSpec spec2{ "tracks", "TPC", "TRACKS" };
 
   std::vector<InputRoute> inputs = {
-    route1,
-    route2
+    InputRoute{ spec1, "Fake1", 0 },
+    InputRoute{ spec2, "Fake2", 0 },
   };
+
   std::vector<ForwardRoute> forwards;
   TimesliceIndex index;