DPL: add SendingPolicy for the case destination is expendable

AliceO2Group · Feb 9, 2024 · 7897c37 · 7897c37
1 parent eb9aadc
commit 7897c37
Showing 1 changed file with 18 additions and 1 deletion.
diff --git a/Framework/Core/src/SendingPolicy.cxx b/Framework/Core/src/SendingPolicy.cxx
@@ -29,7 +29,7 @@ std::vector<SendingPolicy> SendingPolicy::createDefaultPolicies()
 {
   return {SendingPolicy{
             .name = "dispatcher",
-            .matcher = [](DataProcessorSpec const& source, DataProcessorSpec const& dest, ConfigContext const&) { 
+            .matcher = [](DataProcessorSpec const& source, DataProcessorSpec const& dest, ConfigContext const&) {
                 if (source.name == "Dispatcher") {
                   return true;
                 }
@@ -103,6 +103,23 @@ std::vector<SendingPolicy> SendingPolicy::createDefaultPolicies()
               } else if (res == (size_t) fair::mq::TransferCode::error) {
                 LOGP(fatal, "Error while sending on channel {}", channel->GetName());
               } }},
+          SendingPolicy{
+            .name = "expendable",
+            .matcher = [](DataProcessorSpec const& source, DataProcessorSpec const& dest, ConfigContext const&) {
+              auto has_label = [](DataProcessorLabel const& label) {
+                return label.value == "expendable";
+              };
+              return std::find_if(dest.labels.begin(), dest.labels.end(), has_label) != dest.labels.end(); },
+            .send = [](fair::mq::Parts& parts, ChannelIndex channelIndex, ServiceRegistryRef registry) {
+              auto &proxy = registry.get<FairMQDeviceProxy>();
+              auto *channel = proxy.getOutputChannel(channelIndex);
+              auto timeout = 1000;
+              auto res = channel->Send(parts, timeout);
+              if (res == (size_t)fair::mq::TransferCode::timeout) {
+                LOGP(warning, "Timed out sending after {}s. Downstream backpressure detected on expendable channel {}.", timeout/1000, channel->GetName());
+              } else if (res == (size_t) fair::mq::TransferCode::error) {
+                LOGP(info, "Error while sending on channel {}", channel->GetName());
+              } }},
           SendingPolicy{
             .name = "default",
             .matcher = [](DataProcessorSpec const&, DataProcessorSpec const&, ConfigContext const&) { return true; },