DPL: Add DPL_OUTPUT_PROXY_WHENANY env variable setting for output proxy

davidrohr · davidrohr · commit 7c80e4ffdf3d · 2022-06-10T12:04:04.000+02:00
diff --git a/Framework/Core/include/Framework/CompletionPolicyHelpers.h b/Framework/Core/include/Framework/CompletionPolicyHelpers.h
@@ -49,6 +49,8 @@ struct CompletionPolicyHelpers {
   {
     return consumeWhenAny("consume-any", matcher);
   }
+  static CompletionPolicy consumeWhenAny(std::string matchName);
+
   /// When any of the parts of the record have been received, process the existing and free the associated payloads.
   /// This allows freeing things as early as possible, while still being able to wait
   /// all the parts before disposing the timeslice completely
diff --git a/Framework/Core/src/CompletionPolicyHelpers.cxx b/Framework/Core/src/CompletionPolicyHelpers.cxx
@@ -204,6 +204,14 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(const char* name, Compl
   return CompletionPolicy{name, matcher, callback};
 }
 
+CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(std::string matchName)
+{
+  auto matcher = [matchName](DeviceSpec const& device) -> bool {
+    return std::regex_match(device.name.begin(), device.name.end(), std::regex(matchName));
+  };
+  return consumeWhenAny(matcher);
+}
+
 CompletionPolicy CompletionPolicyHelpers::processWhenAny(const char* name, CompletionPolicy::Matcher matcher)
 {
   auto callback = [](InputSpan const& inputs) -> CompletionPolicy::CompletionOp {
diff --git a/Framework/Utils/src/dpl-output-proxy.cxx b/Framework/Utils/src/dpl-output-proxy.cxx
@@ -59,8 +59,11 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)
 void customize(std::vector<o2::framework::CompletionPolicy>& policies)
 {
   static bool doOrdered = getenv("DPL_OUTPUT_PROXY_ORDERED") && atoi(getenv("DPL_OUTPUT_PROXY_ORDERED"));
+  static bool doAny = getenv("DPL_OUTPUT_PROXY_WHENANY") && atoi(getenv("DPL_OUTPUT_PROXY_WHENANY"));
   if (doOrdered) {
     policies.push_back(CompletionPolicyHelpers::consumeWhenAllOrdered(".*"));
+  } else if (doAny) {
+    policies.push_back(CompletionPolicyHelpers::consumeWhenAny(".*"));
   }
 }
 

Original file line number	Diff line number	Diff line change
`@@ -49,6 +49,8 @@ struct CompletionPolicyHelpers {`
`49`	`49`	`{`
`50`	`50`	`return consumeWhenAny("consume-any", matcher);`
`51`	`51`	`}`
	`52`	`+ static CompletionPolicy consumeWhenAny(std::string matchName);`
	`53`	`+`
`52`	`54`	`/// When any of the parts of the record have been received, process the existing and free the associated payloads.`
`53`	`55`	`/// This allows freeing things as early as possible, while still being able to wait`
`54`	`56`	`/// all the parts before disposing the timeslice completely`
Original file line number	Diff line number	Diff line change
`@@ -204,6 +204,14 @@ CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(const char* name, Compl`
`204`	`204`	`return CompletionPolicy{name, matcher, callback};`
`205`	`205`	`}`
`206`	`206`
	`207`	`+CompletionPolicy CompletionPolicyHelpers::consumeWhenAny(std::string matchName)`
	`208`	`+{`
	`209`	`+ auto matcher = [matchName](DeviceSpec const& device) -> bool {`
	`210`	`+ return std::regex_match(device.name.begin(), device.name.end(), std::regex(matchName));`
	`211`	`+ };`
	`212`	`+ return consumeWhenAny(matcher);`
	`213`	`+}`
	`214`	`+`
`207`	`215`	`CompletionPolicy CompletionPolicyHelpers::processWhenAny(const char* name, CompletionPolicy::Matcher matcher)`
`208`	`216`	`{`
`209`	`217`	`auto callback = [](InputSpan const& inputs) -> CompletionPolicy::CompletionOp {`
Original file line number	Diff line number	Diff line change
`@@ -59,8 +59,11 @@ void customize(std::vector<ConfigParamSpec>& workflowOptions)`
`59`	`59`	`void customize(std::vector<o2::framework::CompletionPolicy>& policies)`
`60`	`60`	`{`
`61`	`61`	`static bool doOrdered = getenv("DPL_OUTPUT_PROXY_ORDERED") && atoi(getenv("DPL_OUTPUT_PROXY_ORDERED"));`
	`62`	`+ static bool doAny = getenv("DPL_OUTPUT_PROXY_WHENANY") && atoi(getenv("DPL_OUTPUT_PROXY_WHENANY"));`
`62`	`63`	`if (doOrdered) {`
`63`	`64`	`policies.push_back(CompletionPolicyHelpers::consumeWhenAllOrdered(".*"));`
	`65`	`+ } else if (doAny) {`
	`66`	`+ policies.push_back(CompletionPolicyHelpers::consumeWhenAny(".*"));`
`64`	`67`	`}`
`65`	`68`	`}`
`66`	`69`