DPL: event driven FairMQ data loop with libuv

ktf · ktf · commit 809b0c7dbe2b · 2020-06-22T15:21:21.000+02:00
diff --git a/Framework/Core/include/Framework/AlgorithmSpec.h b/Framework/Core/include/Framework/AlgorithmSpec.h
@@ -45,6 +45,12 @@ struct AlgorithmSpec {
   using ProcessCallback = std::function<void(ProcessingContext&)>;
   using InitCallback = std::function<ProcessCallback(InitContext&)>;
   using ErrorCallback = std::function<void(ErrorContext&)>;
+
+  static AlgorithmSpec dummyAlgorithm()
+  {
+    return AlgorithmSpec{ProcessCallback{nullptr}};
+  }
+
   static ErrorCallback& emptyErrorCallback()
   {
     static ErrorCallback callback = nullptr;
diff --git a/Framework/Core/include/Framework/ChannelConfigurationPolicyHelpers.h b/Framework/Core/include/Framework/ChannelConfigurationPolicyHelpers.h
@@ -52,6 +52,10 @@ struct ChannelConfigurationPolicyHelpers {
   static InputChannelModifier reqInput;
   /// Makes the passed output channel bind and reply
   static OutputChannelModifier replyOutput;
+  /// Makes the passed input channel connect and pair
+  static InputChannelModifier pairInput;
+  /// Makes the passed output channel bind and pair
+  static OutputChannelModifier pairOutput;
 };
 
 } // namespace framework
diff --git a/Framework/Core/include/Framework/ChannelSpec.h b/Framework/Core/include/Framework/ChannelSpec.h
@@ -30,6 +30,7 @@ enum struct ChannelType {
   Sub,
   Push,
   Pull,
+  Pair
 };
 
 /// The kind of backend to use for the channels
diff --git a/Framework/Core/include/Framework/DataProcessingDevice.h b/Framework/Core/include/Framework/DataProcessingDevice.h
@@ -55,6 +55,7 @@ class DataProcessingDevice : public FairMQDevice
   void SetErrorPolicy(enum TerminationPolicy policy) { mErrorPolicy = policy; }
 
  protected:
+  bool doRun();
   bool handleData(FairMQParts&, InputChannelInfo&);
   bool tryDispatchComputation(std::vector<DataRelayer::RecordAction>& completed);
   void error(const char* msg);
@@ -91,6 +92,7 @@ class DataProcessingDevice : public FairMQDevice
   int mCurrentBackoff = 0;                           /// The current exponential backoff value.
   std::vector<FairMQRegionInfo> mPendingRegionInfos; /// A list of the region infos not yet notified.
   enum TerminationPolicy mErrorPolicy = TerminationPolicy::WAIT; /// What to do when an error arises
+  bool mWasActive = false;                                       /// Whether or not the device was active at last iteration.
 };
 
 } // namespace o2::framework
diff --git a/Framework/Core/include/Framework/DeviceState.h b/Framework/Core/include/Framework/DeviceState.h
@@ -16,6 +16,8 @@
 #include <map>
 #include <utility>
 
+typedef struct uv_loop_s uv_loop_t;
+
 namespace o2::framework
 {
 
@@ -35,6 +37,7 @@ struct DeviceState {
   std::vector<InputChannelInfo> inputChannelInfos;
   StreamingState streaming = StreamingState::Streaming;
   bool quitRequested = false;
+  uv_loop_t* loop;
 };
 
 } // namespace o2::framework
diff --git a/Framework/Core/src/ChannelConfigurationPolicyHelpers.cxx b/Framework/Core/src/ChannelConfigurationPolicyHelpers.cxx
@@ -61,5 +61,17 @@ ChannelConfigurationPolicyHelpers::OutputChannelModifier ChannelConfigurationPol
     channel.type = ChannelType::Push;
   };
 
+ChannelConfigurationPolicyHelpers::InputChannelModifier ChannelConfigurationPolicyHelpers::pairInput =
+  [](InputChannelSpec& channel) {
+    channel.method = ChannelMethod::Connect;
+    channel.type = ChannelType::Pair;
+  };
+
+ChannelConfigurationPolicyHelpers::OutputChannelModifier ChannelConfigurationPolicyHelpers::pairOutput =
+  [](OutputChannelSpec& channel) {
+    channel.method = ChannelMethod::Bind;
+    channel.type = ChannelType::Pair;
+  };
+
 } // namespace framework
 } // namespace o2
diff --git a/Framework/Core/src/ChannelSpecHelpers.cxx b/Framework/Core/src/ChannelSpecHelpers.cxx
@@ -27,6 +27,8 @@ char const* ChannelSpecHelpers::typeAsString(enum ChannelType type)
       return "push";
     case ChannelType::Pull:
       return "pull";
+    case ChannelType::Pair:
+      return "pair";
   }
   throw std::runtime_error("Unknown ChannelType");
 }
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
diff --git a/Framework/Core/src/WorkflowHelpers.cxx b/Framework/Core/src/WorkflowHelpers.cxx
diff --git a/Framework/Core/src/runDataProcessing.cxx b/Framework/Core/src/runDataProcessing.cxx

-Original file line number
+Diff line change
   Sub,
   Push,
   Pull,
 +  Pair
 };
 /// The kind of backend to use for the channels
Original file line number	Diff line number	Diff line change
`@@ -27,6 +27,8 @@ char const* ChannelSpecHelpers::typeAsString(enum ChannelType type)`
`27`	`27`	`return "push";`
`28`	`28`	`case ChannelType::Pull:`
`29`	`29`	`return "pull";`
	`30`	`+ case ChannelType::Pair:`
	`31`	`+ return "pair";`
`30`	`32`	`}`
`31`	`33`	`throw std::runtime_error("Unknown ChannelType");`
`32`	`34`	`}`