DPL Proxies: add --sporadic-inputs / --sporadic-outputs option to proxies

Author: ktf

Framework/Utils/src/dpl-output-proxy.cxx

@@ -39,6 +39,10 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
     ConfigParamSpec{
       "output-proxy-method", VariantType::String, "bind", {"proxy socket method: bind, connect"}});
 
+  workflowOptions.push_back(
+    ConfigParamSpec{
+      "sporadic-inputs", VariantType::Bool, false, {"consider all the inputs as sporadic"}});
+
   workflowOptions.push_back(
     ConfigParamSpec{
       "output-proxy-address", VariantType::String, "0.0.0.0", {"address to connect / bind to"}});
@@ -74,6 +78,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
   auto processorName = config.options().get<std::string>("proxy-name");
   auto inputConfig = config.options().get<std::string>("dataspec");
   int defaultPort = config.options().get<int>("default-port");
+  bool sporadicInputs = config.options().get<bool>("sporadic-inputs");
   auto defaultTransportConfig = config.options().get<std::string>("default-transport");
   if (defaultTransportConfig == "zeromq") {
     // nothing to do for the moment
@@ -87,6 +92,12 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
   if (inputs.size() == 0) {
     throw std::runtime_error("invalid dataspec '" + inputConfig + "'");
   }
+  // we need to set the lifetime of the inputs to sporadic if requested
+  if (sporadicInputs) {
+    for (auto& input : inputs) {
+      input.lifetime = Lifetime::Sporadic;
+    }
+  }
 
   // we build the default channel configuration from the binding of the first input
   // in order to have more than one we would need to possibility to have support for

Framework/Utils/src/raw-proxy.cxx

@@ -35,6 +35,10 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
     ConfigParamSpec{
       "inject-missing-data", VariantType::Bool, false, {"inject missing data according to dataspec if not found in the input"}});
 
+  workflowOptions.push_back(
+    ConfigParamSpec{
+      "sporadic-outputs", VariantType::Bool, false, {"consider all the outputs as sporadic"}});
+
   workflowOptions.push_back(
     ConfigParamSpec{
       "print-input-sizes", VariantType::Int, 0, {"print statistics about sizes per input spec every n TFs"}});
@@ -52,16 +56,18 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
 
 WorkflowSpec defineDataProcessing(ConfigContext const& config)
 {
-  std::string processorName = config.options().get<std::string>("proxy-name");
-  std::string outputconfig = config.options().get<std::string>("dataspec");
+  auto processorName = config.options().get<std::string>("proxy-name");
+  auto outputconfig = config.options().get<std::string>("dataspec");
   bool injectMissingData = config.options().get<bool>("inject-missing-data");
-  unsigned int printSizes = config.options().get<unsigned int>("print-input-sizes");
+  bool sporadicOutputs = config.options().get<bool>("sporadic-outputs");
+  auto printSizes = config.options().get<unsigned int>("print-input-sizes");
   bool throwOnUnmatched = config.options().get<bool>("throwOnUnmatched");
   uint64_t minSHM = std::stoul(config.options().get<std::string>("timeframes-shm-limit"));
   std::vector<InputSpec> matchers = select(outputconfig.c_str());
   Outputs readoutProxyOutput;
   for (auto const& matcher : matchers) {
     readoutProxyOutput.emplace_back(DataSpecUtils::asOutputSpec(matcher));
+    readoutProxyOutput.back().lifetime = sporadicOutputs ? Lifetime::Sporadic : Lifetime::Timeframe;
   }
 
   // we use the same specs as filters in the dpl adaptor