DPL: add support for dangling outputs

Sometimes it's handy to leave dangling outputs visible to a data
processor, so that one can reuse a given algorithm in different
scenarious where they might or not be requested.

This changes the previous behavior that was to wipe the output
completely, so that invoking make() on it would have thrown an
exception.

The new behavior will allow the user to invoke make and pass the output
to an internal device that acts as a sink.

Future updates of this feature will allow the user to query if
a given output is dangling or not and provide a callback based API for
make which will invoke the callback only if the output is not dangling.

We could also exploit this feature to have preconfigured actions (e.g. write
to a ROOT file) for dangling outputs, so that the user does not need to
instanciate a writer himself, but can rely on the system to create one
for all the dangling outputs.

Author: ktf

Framework/Core/CMakeLists.txt

@@ -208,6 +208,7 @@ set(TEST_SRCS
       test/test_CustomGUI.cxx
       test/test_CompletionPolicy.cxx
       test/test_DanglingInputs.cxx
+      test/test_DanglingOutputs.cxx
       test/test_DataAllocator.cxx
       test/test_DataProcessorSpec.cxx
       test/test_DataRefUtils.cxx

Framework/Core/include/Framework/OutputSpec.h

@@ -68,7 +68,7 @@ struct OutputSpec {
   {
   }
 
-  bool operator==(const OutputSpec& that)
+  bool operator==(OutputSpec const& that) const
   {
     return origin == that.origin && description == that.description && subSpec == that.subSpec &&
            lifetime == that.lifetime;

Framework/Core/src/WorkflowHelpers.cxx

@@ -157,6 +157,13 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow)
     }
   }
 
+  auto danglingOutputsInputs = computeDanglingOutputs(workflow);
+  DataProcessorSpec dplInternalSync{
+    "internal-dpl-sink",
+    danglingOutputsInputs,
+    {},
+  };
+
   if (ccdbBackend.outputs.empty() == false) {
     workflow.push_back(ccdbBackend);
   }
@@ -169,6 +176,11 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow)
   if (timer.outputs.empty() == false) {
     workflow.push_back(timer);
   }
+  /// This will inject a fake sink so that any dangling
+  /// output is actually requested by it.
+  if (danglingOutputsInputs.empty() == false) {
+    workflow.push_back(dplInternalSync);
+  }
 }
 
 void
@@ -477,5 +489,62 @@ WorkflowHelpers::verifyWorkflow(const o2::framework::WorkflowSpec &workflow) {
   }
 }
 
+struct UnifiedDataSpecType {
+  header::DataOrigin origin;
+  header::DataDescription description;
+  uint64_t subSpec;
+  int isOutput;
+};
+
+std::vector<InputSpec> WorkflowHelpers::computeDanglingOutputs(WorkflowSpec const& workflow)
+{
+  std::vector<UnifiedDataSpecType> tmp;
+  std::vector<InputSpec> results;
+
+  for (auto& spec : workflow) {
+    for (auto& input : spec.inputs) {
+      tmp.push_back({ input.origin, input.description, input.subSpec, 0 });
+    }
+    for (auto& output : spec.outputs) {
+      tmp.push_back({ output.origin, output.description, output.subSpec, 1 });
+    }
+  }
+
+  auto cmp = [](UnifiedDataSpecType const& lhs, UnifiedDataSpecType const& rhs) -> bool {
+    return std::tie(lhs.origin, lhs.description, lhs.subSpec, lhs.isOutput) <
+           std::tie(rhs.origin, rhs.description, rhs.subSpec, rhs.isOutput);
+  };
+  std::sort(tmp.begin(), tmp.end(), cmp);
+
+  // Once sorted, all the transitions which begin with an output
+  // mean there was no input.
+  bool isFirst = true;
+  UnifiedDataSpecType last;
+  int i = 0;
+  for (auto& unified : tmp) {
+    if (last.origin != unified.origin ||
+        last.description != unified.description ||
+        last.subSpec != unified.subSpec) {
+      isFirst = true;
+      last = unified;
+    }
+    if (isFirst && unified.isOutput == 0) {
+      isFirst = false;
+      continue;
+    }
+    if (isFirst && unified.isOutput == 1) {
+      isFirst = false;
+      char buf[64];
+      results.push_back(InputSpec{ (snprintf(buf, 64, "dangling%d", i), buf), unified.origin,
+                                   unified.description, unified.subSpec });
+      ++i;
+      continue;
+    }
+    assert(isFirst == false);
+  }
+
+  return results;
+}
+
 } // namespace framwork
 } // namespace o2

Framework/Core/src/WorkflowHelpers.h

@@ -165,6 +165,11 @@ struct WorkflowHelpers {
   static std::vector<EdgeAction> computeInEdgeActions(
                                  const std::vector<DeviceConnectionEdge> &edges,
                                  const std::vector<size_t> &index);
+
+  /// Given @a workflow it finds the OutputSpec in every module which do not have
+  /// a corresponding InputSpec. I.e. they are dangling.
+  /// @return a vector of InputSpec which would have matched said dangling outputs.
+  static std::vector<InputSpec> computeDanglingOutputs(WorkflowSpec const& workflow);
 };
 
 }

Framework/Core/test/test_DanglingOutputs.cxx

@@ -0,0 +1,51 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+#include "Framework/ConfigParamSpec.h"
+#include "Framework/DeviceSpec.h"
+#include <InfoLogger/InfoLogger.hxx>
+#include <vector>
+#include "Framework/runDataProcessing.h"
+#include "Framework/ControlService.h"
+
+using namespace o2::framework;
+
+AlgorithmSpec simplePipe(std::string const& what, int minDelay)
+{
+  return AlgorithmSpec{ [what, minDelay](InitContext& ic) {
+    srand(getpid());
+    return [what, minDelay](ProcessingContext& ctx) {
+      auto bData = ctx.outputs().make<int>(OutputRef{ what }, 1);
+    };
+  } };
+}
+
+// a1 is not actually used by anything, however it might.
+WorkflowSpec defineDataProcessing(ConfigContext const& specs)
+{
+  return WorkflowSpec{
+    { "A",
+      Inputs{},
+      { OutputSpec{ { "a1" }, "TST", "A1" },
+        OutputSpec{ { "a2" }, "TST", "A2" } },
+      AlgorithmSpec{
+        [](ProcessingContext& ctx) {
+          sleep(rand() % 2 + 1);
+          auto aData1 = ctx.outputs().make<int>(OutputRef{ "a1" }, 1);
+          auto aData2 = ctx.outputs().make<int>(Output{ "TST", "A2" }, 1);
+        } } },
+    { "B",
+      { InputSpec{ { "a1" }, "TST", "A1" } },
+      {},
+      AlgorithmSpec{
+        [](ProcessingContext& ctx) {
+          ctx.services().get<ControlService>().readyToQuit(true);
+        } } }
+  };
+}

Framework/Core/test/test_WorkflowHelpers.cxx

@@ -16,6 +16,7 @@
 #include "../src/WorkflowHelpers.h"
 #include <boost/test/unit_test.hpp>
 #include <boost/test/tools/detail/per_element_manip.hpp>
+#include <algorithm>
 
 using namespace o2::framework;
 
@@ -418,10 +419,59 @@ BOOST_AUTO_TEST_CASE(TestExternalInput)
 
   BOOST_CHECK_EQUAL(workflow.size(), 1);
   WorkflowHelpers::injectServiceDevices(workflow);
-  // The added device is the one which should connect to
-  // the condition DB.
-  BOOST_CHECK_EQUAL(workflow.size(), 2);
+  // The added devices are the one which should connect to
+  // the condition DB and the sink for the dangling outputs.
+  BOOST_CHECK_EQUAL(workflow.size(), 3);
   WorkflowHelpers::constructGraph(workflow, logicalEdges,
                                   outputs,
                                   availableForwardsInfo);
 }
+
+BOOST_AUTO_TEST_CASE(DetermineDanglingOutputs)
+{
+  WorkflowSpec workflow0{
+    { "A", Inputs{}, { OutputSpec{ "TST", "A" } } },
+    { "B", { InputSpec{ "a", "TST", "A" } }, Outputs{} }
+  };
+
+  WorkflowSpec workflow1{
+    { "A",
+      Inputs{},
+      Outputs{ OutputSpec{ "TST", "A" } } }
+  };
+
+  WorkflowSpec workflow2{
+    { "A", Inputs{}, { OutputSpec{ "TST", "A" } } },
+    { "B", { InputSpec{ "a", "TST", "B" } }, Outputs{} }
+  };
+
+  WorkflowSpec workflow3{
+    { "A", Inputs{}, { OutputSpec{ "TST", "A" }, OutputSpec{ "TST", "B" } } },
+    { "B", { InputSpec{ "a", "TST", "A" } }, Outputs{} }
+  };
+
+  WorkflowSpec workflow4{
+    { "A", Inputs{}, { OutputSpec{ "TST", "A" }, OutputSpec{ "TST", "B" }, OutputSpec{ "TST", "C" } } },
+    { "B", { InputSpec{ "a", "TST", "A" } }, Outputs{} }
+  };
+
+  auto dangling0 = WorkflowHelpers::computeDanglingOutputs(workflow0);
+  std::vector<InputSpec> expected0{};
+  BOOST_TEST(dangling0 == expected0, boost::test_tools::per_element());
+
+  auto dangling1 = WorkflowHelpers::computeDanglingOutputs(workflow1);
+  std::vector<InputSpec> expected1{ InputSpec{ "dangling0", "TST", "A" } };
+  BOOST_TEST(dangling1 == expected1, boost::test_tools::per_element());
+
+  auto dangling2 = WorkflowHelpers::computeDanglingOutputs(workflow2);
+  std::vector<InputSpec> expected2{ InputSpec{ "dangling0", "TST", "A" } };
+  BOOST_TEST(dangling2 == expected2, boost::test_tools::per_element());
+
+  auto dangling3 = WorkflowHelpers::computeDanglingOutputs(workflow3);
+  std::vector<InputSpec> expected3{ InputSpec{ "dangling0", "TST", "B" } };
+  BOOST_TEST(dangling3 == expected3, boost::test_tools::per_element());
+
+  auto dangling4 = WorkflowHelpers::computeDanglingOutputs(workflow4);
+  std::vector<InputSpec> expected4{ InputSpec{ "dangling0", "TST", "B" }, InputSpec{ "dangling1", "TST", "C" } };
+  BOOST_TEST(dangling4 == expected4, boost::test_tools::per_element());
+}