DPL: add ability to disable inputs programmatically

Disabled inputs will not result in an actual route for the data, however it
will be stored in the configuration, so that analysis workflows will not need
to have the configuration available at every step.

Framework/Core/include/Framework/InputSpec.h

@@ -74,6 +74,9 @@ struct InputSpec {
 
   /// A set of configurables which can be used to customise the InputSpec.
   std::vector<ConfigParamSpec> metadata;
+  /// Wether or not the input is to be considered enabled.
+  /// Useful to programmatically disable inputs e.g. for the ProcessorOptions.
+  bool enabled = true;
 
   friend std::ostream& operator<<(std::ostream& stream, InputSpec const& arg);
   bool operator==(InputSpec const& that) const;

Framework/Core/include/Framework/OutputSpec.h

@@ -86,6 +86,10 @@ struct OutputSpec {
   /// A set of configurables which can be used to customise the InputSpec.
   std::vector<ConfigParamSpec> metadata;
 
+  /// Wether or not this output is enabled. This is useful to decide programmatically
+  /// wether or not to produce a given output.
+  bool enabled = true;
+
   friend std::ostream& operator<<(std::ostream& stream, OutputSpec const& arg);
 };
 

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -679,19 +679,27 @@ void DeviceSpecHelpers::processOutEdgeActions(ConfigContext const& configContext
     assert(policyPtr != nullptr);
 
     if (edge.isForward == false) {
+      auto& matcher = outputsMatchers[edge.outputGlobalIndex];
+      if (matcher.enabled == false) {
+        throw runtime_error_f("Output %s is disabled but it was still used in topology", DataSpecUtils::describe(matcher).data());
+      }
       OutputRoute route{
         edge.timeIndex,
         consumer.maxInputTimeslices,
-        outputsMatchers[edge.outputGlobalIndex],
+        matcher,
         channel.name,
         policyPtr,
       };
       device.outputs.emplace_back(route);
     } else {
+      auto& matcher = workflow[edge.consumer].inputs[edge.consumerInputIndex];
+      if (matcher.enabled == false) {
+        throw runtime_error_f("Output %s is disabled but it was still used in topology", DataSpecUtils::describe(matcher).data());
+      }
       ForwardRoute route{
         edge.timeIndex,
         consumer.maxInputTimeslices,
-        workflow[edge.consumer].inputs[edge.consumerInputIndex],
+        matcher,
         channel.name};
       device.forwards.emplace_back(route);
     }
@@ -915,6 +923,9 @@ void DeviceSpecHelpers::processInEdgeActions(std::vector<DeviceSpec>& devices,
 
     auto const& inputSpec = consumer.inputs[edge.consumerInputIndex];
     auto const& sourceChannel = consumerDevice.inputChannels[ci].name;
+    if (inputSpec.enabled == false) {
+      throw runtime_error_f("Input %s is disabled but it was still used in topology", DataSpecUtils::describe(inputSpec).data());
+    }
 
     InputRoute route{
       inputSpec,

Framework/Core/src/WorkflowHelpers.cxx

@@ -841,11 +841,13 @@ void WorkflowHelpers::constructGraph(const WorkflowSpec& workflow,
     for (size_t wi = 0; wi < workflow.size(); ++wi) {
       auto& producer = workflow[wi];
 
-      for (size_t oi = 0; oi < producer.outputs.size(); ++oi) {
-        auto& out = producer.outputs[oi];
+      for (auto& output : producer.outputs) {
+        if (output.enabled == false) {
+          continue;
+        }
         auto uniqueOutputId = outputs.size();
         availableOutputsInfo.emplace_back(LogicalOutputInfo{wi, uniqueOutputId, false});
-        outputs.push_back(out);
+        outputs.push_back(output);
       }
     }
   };
@@ -879,6 +881,10 @@ void WorkflowHelpers::constructGraph(const WorkflowSpec& workflow,
   std::vector<bool> matches(constOutputs.size());
   for (size_t consumer = 0; consumer < workflow.size(); ++consumer) {
     for (size_t input = 0; input < workflow[consumer].inputs.size(); ++input) {
+      // Skip disabled inputs.
+      if (workflow[consumer].inputs[input].enabled == false) {
+        continue;
+      }
       forwards.clear();
       for (size_t i = 0; i < constOutputs.size(); i++) {
         matches[i] = DataSpecUtils::match(workflow[consumer].inputs[input], constOutputs[i]);

Framework/Core/src/WorkflowSerializationHelpers.cxx

@@ -61,6 +61,7 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
     IN_INPUT_ORIGIN,
     IN_INPUT_DESCRIPTION,
     IN_INPUT_SUBSPEC,
+    IN_INPUT_ENABLED,
     IN_INPUT_ORIGIN_REF,
     IN_INPUT_DESCRIPTION_REF,
     IN_INPUT_SUBSPEC_REF,
@@ -75,6 +76,7 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
     IN_OUTPUT_BINDING,
     IN_OUTPUT_ORIGIN,
     IN_OUTPUT_DESCRIPTION,
+    IN_OUTPUT_ENABLED,
     IN_OUTPUT_SUBSPEC,
     IN_OUTPUT_LIFETIME,
     IN_OUTPUT_OPTIONS,
@@ -167,6 +169,9 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       case State::IN_INPUT_SUBSPEC:
         s << "IN_INPUT_SUBSPEC";
         break;
+      case State::IN_INPUT_ENABLED:
+        s << "IN_INPUT_ENABLED";
+        break;
       case State::IN_INPUT_ORIGIN_REF:
         s << "IN_INPUT_ORIGIN_REF";
         break;
@@ -218,6 +223,9 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       case State::IN_OUTPUT_OPTIONS:
         s << "IN_OUTPUT_OPTIONS";
         break;
+      case WorkflowImporter::State::IN_OUTPUT_ENABLED:
+        s << "IN_OUTPUT_ENABLED";
+        break;
       case State::IN_OPTION:
         s << "IN_OPTION";
         break;
@@ -302,6 +310,7 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       dataProcessors.push_back(DataProcessorSpec{});
     } else if (in(State::IN_INPUTS)) {
       push(State::IN_INPUT);
+      enabled = true;
       inputMatcherNodes.clear();
     } else if (in(State::IN_INPUT_MATCHER)) {
       // start a new embedded matcher
@@ -313,6 +322,7 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       // will be merged into the parent matcher
     } else if (in(State::IN_OUTPUTS)) {
       push(State::IN_OUTPUT);
+      enabled = true;
       outputHasSubSpec = false;
     } else if (in(State::IN_OPTIONS)) {
       push(State::IN_OPTION);
@@ -370,9 +380,9 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       } else {
         dataProcessors.back().inputs.push_back(InputSpec({binding}, std::move(*matcher), lifetime, inputOptions));
       }
+      dataProcessors.back().inputs.back().enabled = enabled;
       inputMatcherNodes.clear();
       inputOptions.clear();
-
     } else if (in(State::IN_INPUT_MATCHER) && inputMatcherNodes.size() > 1) {
       data_matcher::Node child = std::move(inputMatcherNodes.back());
       inputMatcherNodes.pop_back();
@@ -433,6 +443,7 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       } else {
         dataProcessors.back().outputs.push_back(OutputSpec({binding}, {origin, description}, lifetime));
       }
+      dataProcessors.back().outputs.back().enabled = enabled;
       outputHasSubSpec = false;
     } else if (in(State::IN_OPTION)) {
       std::unique_ptr<ConfigParamSpec> opt{nullptr};
@@ -545,12 +556,14 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       push(State::IN_DATAPROCESSORS);
     } else if (in(State::IN_INPUTS)) {
       push(State::IN_INPUT);
+      enabled = true;
     } else if (in(State::IN_INPUT_OPTIONS)) {
       push(State::IN_OPTION);
     } else if (in(State::IN_OUTPUT_OPTIONS)) {
       push(State::IN_OPTION);
     } else if (in(State::IN_OUTPUTS)) {
       push(State::IN_OUTPUT);
+      enabled = true;
       outputHasSubSpec = false;
     } else if (in(State::IN_OPTIONS)) {
       push(State::IN_OPTION);
@@ -602,6 +615,8 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       push(State::IN_INPUT_DESCRIPTION_REF);
     } else if (in(State::IN_INPUT) && strncmp(str, "subspecRef", length) == 0) {
       push(State::IN_INPUT_SUBSPEC_REF);
+    } else if (in(State::IN_INPUT) && strncmp(str, "enabled", length) == 0) {
+      push(State::IN_INPUT_ENABLED);
     } else if (in(State::IN_INPUT) && strncmp(str, "matcher", length) == 0) {
       // the outermost matcher is starting here
       // we create a placeholder which is being updated later
@@ -664,6 +679,8 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       push(State::IN_OUTPUT_LIFETIME);
     } else if (in(State::IN_OUTPUT) && strncmp(str, "metadata", length) == 0) {
       push(State::IN_OUTPUT_OPTIONS);
+    } else if (in(State::IN_OUTPUT) && strncmp(str, "enabled", length) == 0) {
+      push(State::IN_OUTPUT_ENABLED);
     } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "name", length) == 0) {
       push(State::IN_DATAPROCESSOR_NAME);
     } else if (in(State::IN_DATAPROCESSOR) && strncmp(str, "ranks", length) == 0) {
@@ -734,6 +751,8 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       metadata.back().executable = s;
     } else if (in(State::IN_INPUT_BINDING)) {
       binding = s;
+    } else if (in(State::IN_INPUT_ENABLED)) {
+      enabled = (s == "true");
     } else if (in(State::IN_INPUT_ORIGIN)) {
       origin.runtimeInit(s.c_str(), std::min(s.size(), 4UL));
       std::string v(s.c_str(), std::min(s.size(), 4UL));
@@ -841,6 +860,10 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
       dataProcessors.back().inputTimeSliceId = i;
     } else if (in(State::IN_DATAPROCESSOR_MAX_TIMESLICES)) {
       dataProcessors.back().maxInputTimeslices = i;
+    } else if (in(State::IN_INPUT_ENABLED)) {
+      enabled = (i == 1);
+    } else if (in(State::IN_OUTPUT_ENABLED)) {
+      enabled = (i == 1);
     }
     pop();
     return true;
@@ -915,6 +938,9 @@ struct WorkflowImporter : public rapidjson::BaseReaderHandler<rapidjson::UTF8<>,
   header::DataDescription description;
   size_t subspec;
   size_t ref;
+  // Keep track of the enabled state of the input/output
+  // Unless specified, inputs are enabled by default.
+  bool enabled = true;
   Lifetime lifetime;
   std::string metadatumKey;
   std::string metadatumValue;
@@ -1103,6 +1129,8 @@ void WorkflowSerializationHelpers::dump(std::ostream& out,
       }
       w.Key("lifetime");
       w.Uint((int)input.lifetime);
+      w.Key("enabled");
+      w.Uint((int)input.enabled);
       if (input.metadata.empty() == false) {
         w.Key("metadata");
         w.StartArray();
@@ -1152,6 +1180,8 @@ void WorkflowSerializationHelpers::dump(std::ostream& out,
       }
       w.Key("lifetime");
       w.Uint((int)output.lifetime);
+      w.Key("enabled");
+      w.Uint((int)output.enabled);
       if (output.metadata.empty() == false) {
         w.Key("metadata");
         w.StartArray();

Framework/Core/test/test_FrameworkDataFlowToDDS.cxx

@@ -169,14 +169,16 @@ TEST_CASE("TestDDS")
                     "origin": "TST",
                     "description": "A1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 },
                 {
                     "binding": "TST/A2/0",
                     "origin": "TST",
                     "description": "A2",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "options": [],
@@ -195,7 +197,8 @@ TEST_CASE("TestDDS")
                     "origin": "TST",
                     "description": "A1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "outputs": [
@@ -204,7 +207,8 @@ TEST_CASE("TestDDS")
                     "origin": "TST",
                     "description": "B1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "options": [],
@@ -223,7 +227,8 @@ TEST_CASE("TestDDS")
                     "origin": "TST",
                     "description": "A2",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "outputs": [
@@ -232,7 +237,8 @@ TEST_CASE("TestDDS")
                     "origin": "TST",
                     "description": "C1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "options": [],
@@ -251,14 +257,16 @@ TEST_CASE("TestDDS")
                     "origin": "TST",
                     "description": "B1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 },
                 {
                     "binding": "y",
                     "origin": "TST",
                     "description": "C1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "outputs": [],
@@ -433,14 +441,16 @@ TEST_CASE("TestDDSExpendable")
                     "origin": "TST",
                     "description": "A1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 },
                 {
                     "binding": "TST/A2/0",
                     "origin": "TST",
                     "description": "A2",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "options": [],
@@ -459,7 +469,8 @@ TEST_CASE("TestDDSExpendable")
                     "origin": "TST",
                     "description": "A1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "outputs": [
@@ -468,7 +479,8 @@ TEST_CASE("TestDDSExpendable")
                     "origin": "TST",
                     "description": "B1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "options": [],
@@ -487,7 +499,8 @@ TEST_CASE("TestDDSExpendable")
                     "origin": "TST",
                     "description": "A2",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "outputs": [
@@ -496,7 +509,8 @@ TEST_CASE("TestDDSExpendable")
                     "origin": "TST",
                     "description": "C1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "options": [],
@@ -515,14 +529,16 @@ TEST_CASE("TestDDSExpendable")
                     "origin": "TST",
                     "description": "B1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 },
                 {
                     "binding": "y",
                     "origin": "TST",
                     "description": "C1",
                     "subspec": 0,
-                    "lifetime": 0
+                    "lifetime": 0,
+                    "enabled": 1
                 }
             ],
             "outputs": [],