DPL: introduce dangling-outputs-policy (#4381)

Default / "file": old behaviour. Messages matching --keep rule
will be dumped to a file.

"fairmq": a new "output proxy" device is created and all the dangling
outputs are bound to, which will then push them to a fairmq channel.

Author: ktf

Framework/Core/include/Framework/CommonDataProcessors.h

@@ -17,9 +17,7 @@
 
 #include <vector>
 
-namespace o2
-{
-namespace framework
+namespace o2::framework
 {
 using outputTasks = std::vector<std::pair<uint32_t, std::string>>;
 using outputObjects = std::vector<std::pair<uint32_t, std::vector<std::string>>>;
@@ -38,6 +36,12 @@ struct CommonDataProcessors {
   /// not going to be used by the returned DataProcessorSpec.
   static DataProcessorSpec getGlobalFileSink(std::vector<InputSpec> const& danglingInputs,
                                              std::vector<InputSpec>& unmatched);
+  /// Given the list of @a danglingInputs @return a DataProcessor which
+  /// exposes them through a FairMQ channel.
+  /// @fixme: for now we only support shmem and ipc
+  /// @fixme: for now only the dangling inputs are forwarded.
+  static DataProcessorSpec getGlobalFairMQSink(std::vector<InputSpec> const& danglingInputs);
+
   /// writes inputs of kind AOD to file
   static DataProcessorSpec getGlobalAODSink(std::vector<InputSpec> const& OutputInputs,
                                             std::vector<bool> const& isdangling);
@@ -46,7 +50,6 @@ struct CommonDataProcessors {
   static DataProcessorSpec getDummySink(std::vector<InputSpec> const& danglingInputs);
 };
 
-} // namespace framework
-} // namespace o2
+} // namespace o2::framework
 
 #endif // o2_framework_CommonDataProcessors_H_INCLUDED

Framework/Core/include/Framework/runDataProcessing.h

@@ -126,6 +126,8 @@ int main(int argc, char** argv)
     UserCustomizationsHelper::userDefinedCustomization(workflowOptions, 0);
     workflowOptions.push_back(ConfigParamSpec{"readers", VariantType::Int64, 1ll, {"number of parallel readers to use"}});
     workflowOptions.push_back(ConfigParamSpec{"pipeline", VariantType::String, "", {"override default pipeline size"}});
+    workflowOptions.push_back(ConfigParamSpec{"dangling-outputs-policy", VariantType::String, "file", {"what to do with dangling outputs. file: write to file, fairmq: send to output proxy"}});
+
     std::vector<ChannelConfigurationPolicy> channelPolicies;
     UserCustomizationsHelper::userDefinedCustomization(channelPolicies, 0);
     auto defaultChannelPolicies = ChannelConfigurationPolicy::createDefaultPolicies();

Framework/Core/src/CommonDataProcessors.cxx

@@ -29,6 +29,8 @@
 #include "Framework/OutputObjHeader.h"
 #include "Framework/TableTreeHelpers.h"
 #include "Framework/StringHelpers.h"
+#include "Framework/ChannelSpec.h"
+#include "Framework/ExternalFairMQDeviceProxy.h"
 
 #include "TFile.h"
 #include "TTree.h"
@@ -574,6 +576,30 @@ DataProcessorSpec
   return spec;
 }
 
+DataProcessorSpec CommonDataProcessors::getGlobalFairMQSink(std::vector<InputSpec> const& danglingOutputInputs)
+{
+
+  // 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
+  // vectored options
+  // use the OutputChannelSpec as a tool to create the default configuration for the out-of-band channel
+  OutputChannelSpec externalChannelSpec;
+  externalChannelSpec.name = "output_0_0";
+  externalChannelSpec.type = ChannelType::Push;
+  externalChannelSpec.method = ChannelMethod::Bind;
+  externalChannelSpec.hostname = "localhost";
+  externalChannelSpec.port = 0;
+  externalChannelSpec.listeners = 0;
+  // in principle, protocol and transport are two different things but fur simplicity
+  // we use ipc when shared memory is selected and the normal tcp url whith zeromq,
+  // this is for building the default configuration which can be simply changed from the
+  // command line
+  externalChannelSpec.protocol = ChannelProtocol::IPC;
+  std::string defaultChannelConfig = formatExternalChannelConfiguration(externalChannelSpec);
+  // at some point the formatting tool might add the transport as well so we have to check
+  return specifyFairMQDeviceOutputProxy("internal-dpl-output-proxy", danglingOutputInputs, defaultChannelConfig.c_str());
+}
+
 DataProcessorSpec CommonDataProcessors::getDummySink(std::vector<InputSpec> const& danglingOutputInputs)
 {
   return DataProcessorSpec{

Framework/Core/src/WorkflowHelpers.cxx

@@ -400,11 +400,14 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
   }
 
   std::vector<InputSpec> unmatched;
-  if (outputsInputsDangling.size() > 0) {
+  if (outputsInputsDangling.size() > 0 && ctx.options().get<std::string>("dangling-outputs-policy") == "file") {
     auto fileSink = CommonDataProcessors::getGlobalFileSink(outputsInputsDangling, unmatched);
     if (unmatched.size() != outputsInputsDangling.size()) {
       extraSpecs.push_back(fileSink);
     }
+  } else if (outputsInputsDangling.size() > 0 && ctx.options().get<std::string>("dangling-outputs-policy") == "fairmq") {
+    auto fairMQSink = CommonDataProcessors::getGlobalFairMQSink(outputsInputsDangling);
+    extraSpecs.push_back(fairMQSink);
   }
   if (unmatched.size() > 0) {
     extraSpecs.push_back(CommonDataProcessors::getDummySink(unmatched));

Framework/Core/test/test_WorkflowHelpers.cxx

@@ -30,6 +30,7 @@ std::unique_ptr<ConfigContext> makeEmptyConfigContext()
   //        either owns or shares ownership of the registry.
   std::vector<std::unique_ptr<ParamRetriever>> retrievers;
   static std::vector<ConfigParamSpec> specs;
+  specs.push_back(ConfigParamSpec{"dangling-outputs-policy", VariantType::String, "file", {"what to do with dangling outputs. file: write to file, fairmq: send to output proxy"}});
   auto store = std::make_unique<ConfigParamStore>(specs, std::move(retrievers));
   store->preload();
   store->activate();