DPL: allow parallel WS backed drivers

Author: ktf

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -961,7 +961,7 @@ void split(const std::string& str, Container& cont)
 }
 } // namespace
 
-void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
+void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped, unsigned short driverPort,
                                          std::vector<DataProcessorInfo> const& processorInfos,
                                          std::vector<DeviceSpec> const& deviceSpecs,
                                          std::vector<DeviceExecution>& deviceExecutions,
@@ -1037,6 +1037,7 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
     // has option --session been specified on the command line?
     bool haveSessionArg = false;
     using FilterFunctionT = std::function<void(decltype(argc), decltype(argv), decltype(od))>;
+    bool useDefaultWS = false;
 
     // the filter function will forward command line arguments based on the option
     // definition passed to it. All options of the program option definition will be forwarded
@@ -1109,6 +1110,7 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
       }
 
       haveSessionArg = haveSessionArg || varmap.count("session") != 0;
+      useDefaultWS = useDefaultWS || (varmap.count("driver-client-backend") != 0) && varmap["driver-client-backend"].as<std::string>() == "ws://";
 
       for (const auto varit : varmap) {
         // find the option belonging to key, add if the option has been parsed
@@ -1160,6 +1162,16 @@ void DeviceSpecHelpers::prepareArguments(bool defaultQuiet, bool defaultStopped,
       tmpArgs.emplace_back(std::string("--session"));
       tmpArgs.emplace_back("dpl_" + uniqueWorkflowId);
     }
+    // In case we use only ws://, we need to expand the address
+    // with the correct port.
+    if (useDefaultWS) {
+      auto it = std::find(tmpArgs.begin(), tmpArgs.end(), "--driver-client-backend");
+      if ((it != tmpArgs.end()) && (it + 1 != tmpArgs.end())) {
+        tmpArgs.erase(it, it + 2);
+      }
+      tmpArgs.emplace_back(std::string("--driver-client-backend"));
+      tmpArgs.emplace_back("ws://0.0.0.0:" + std::to_string(driverPort));
+    }
 
     if (spec.resourceMonitoringInterval > 0) {
       tmpArgs.emplace_back(std::string("--resources-monitoring"));

Framework/Core/src/DeviceSpecHelpers.h

@@ -101,6 +101,7 @@ struct DeviceSpecHelpers {
   static void prepareArguments(
     bool defaultQuiet,
     bool defaultStopped,
+    unsigned short driverPort,
     std::vector<DataProcessorInfo> const& processorInfos,
     std::vector<DeviceSpec> const& deviceSpecs,
     std::vector<DeviceExecution>& deviceExecutions,

Framework/Core/src/runDataProcessing.cxx

@@ -7,6 +7,7 @@
 // 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 <stdexcept>
 #include "Framework/BoostOptionsRetriever.h"
 #include "Framework/ChannelConfigurationPolicy.h"
 #include "Framework/ChannelMatching.h"
@@ -1159,6 +1160,9 @@ int runStateMachine(DataProcessorSpecs const& workflow,
       if (result != 0) {
         driverInfo.port++;
       }
+      if (driverInfo.port > 64000) {
+        throw runtime_error_f("Unable to find a free port for the driver. Last attempt returned %d", result);
+      }
     } while (result != 0);
   }
 
@@ -1398,6 +1402,7 @@ int runStateMachine(DataProcessorSpecs const& workflow,
           DeviceSpecHelpers::reworkShmSegmentSize(dataProcessorInfos);
           DeviceSpecHelpers::prepareArguments(driverControl.defaultQuiet,
                                               driverControl.defaultStopped,
+                                              driverInfo.port,
                                               dataProcessorInfos,
                                               deviceSpecs,
                                               deviceExecutions, controls,

Framework/Core/test/test_DeviceSpecHelpers.cxx

@@ -75,14 +75,13 @@ void check(const std::vector<std::string>& arguments,
       workflowOptions,
     });
   }
-  DeviceSpecHelpers::prepareArguments(true, true,
+  DeviceSpecHelpers::prepareArguments(true, true, 8080,
                                       dataProcessorInfos,
                                       deviceSpecs,
                                       deviceExecutions,
                                       deviceControls,
                                       "workflow-id");
 
-
   for (size_t index = 0; index < deviceSpecs.size(); index++) {
     const auto& deviceSpec = deviceSpecs[index];
     const auto& deviceExecution = deviceExecutions[index];

Framework/Core/test/test_FrameworkDataFlowToDDS.cxx

@@ -92,7 +92,7 @@ BOOST_AUTO_TEST_CASE(TestDDS)
       {"C", "foo", {}, workflowOptions},
       {"D", "foo", {}, workflowOptions},
     }};
-  DeviceSpecHelpers::prepareArguments(false, false,
+  DeviceSpecHelpers::prepareArguments(false, false, 8080,
                                       dataProcessorInfos,
                                       devices, executions, controls,
                                       "workflow-id");