DPL: enable WebSocket based DriverClient

Author: ktf

Framework/Core/include/Framework/DeviceState.h

@@ -41,7 +41,7 @@ struct DeviceState {
   StreamingState streaming = StreamingState::Streaming;
   bool quitRequested = false;
   // The libuv event loop which serves this device.
-  uv_loop_t* loop;
+  uv_loop_t* loop = nullptr;
   // The list of active timers which notify this device.
   std::vector<uv_timer_t*> activeTimers;
   // The list of pollers for active input channels

Framework/Core/include/Framework/DriverClient.h

@@ -11,6 +11,7 @@
 #define O2_FRAMEWORK_DRIVERCLIENT_H_
 
 #include "Framework/ServiceHandle.h"
+#include <string>
 #include <functional>
 
 namespace o2::framework
@@ -23,7 +24,15 @@ class DriverClient
   constexpr static ServiceKind service_kind = ServiceKind::Global;
 
   /// Report some message to the Driver
-  virtual void tell(const char* msg) = 0;
+  /// @a msg the message to be sent.
+  /// @a size size of the message to be sent.
+  /// @a flush whether the message should be flushed immediately,
+  /// if possible.
+  virtual void tell(char const* msg, size_t s, bool flush = true) = 0;
+  void tell(std::string_view const& msg, bool flush = true)
+  {
+    tell(msg.data(), msg.size(), flush);
+  };
 
   /// Act on some @a event notified by the driver
   virtual void observe(const char* event, std::function<void(char const*)> callback) = 0;

Framework/Core/include/Framework/DriverInfo.h

@@ -137,6 +137,8 @@ struct DriverInfo {
   std::string uniqueWorkflowId = "";
   /// Metrics gathering interval
   unsigned short resourcesMonitoringInterval;
+  /// Port used by the websocket control. 0 means not initialised.
+  unsigned short port = 0;
   /// Last port used for tracy
   short tracyPort = 8086;
   /// Aggregate metrics calculated in the driver itself

Framework/Core/src/CommonServices.cxx

@@ -25,7 +25,10 @@
 #include "Framework/Tracing.h"
 #include "Framework/Monitoring.h"
 #include "TextDriverClient.h"
+#include "WSDriverClient.h"
+#include "HTTPParser.h"
 #include "../src/DataProcessingStatus.h"
+#include "DPLMonitoringBackend.h"
 
 #include <Configuration/ConfigurationInterface.h>
 #include <Configuration/ConfigurationFactory.h>
@@ -35,6 +38,7 @@
 #include <options/FairMQProgOptions.h>
 
 #include <cstdlib>
+#include <cstring>
 
 using AliceO2::InfoLogger::InfoLogger;
 using AliceO2::InfoLogger::InfoLoggerContext;
@@ -58,8 +62,19 @@ struct ServiceKindExtractor<InfoLoggerContext> {
 o2::framework::ServiceSpec CommonServices::monitoringSpec()
 {
   return ServiceSpec{"monitoring",
-                     [](ServiceRegistry&, DeviceState&, fair::mq::ProgOptions& options) -> ServiceHandle {
-                       void* service = MonitoringFactory::Get(options.GetPropertyAsString("monitoring-backend")).release();
+                     [](ServiceRegistry& registry, DeviceState&, fair::mq::ProgOptions& options) -> ServiceHandle {
+                       void* service = nullptr;
+                       bool isWebsocket = strncmp(options.GetPropertyAsString("driver-client-backend").c_str(), "ws://", 4) == 0;
+                       bool isDefault = options.GetPropertyAsString("monitoring-backend") == "default";
+                       bool useDPL = (isWebsocket && isDefault) || options.GetPropertyAsString("monitoring-backend") == "dpl://";
+                       if (useDPL) {
+                         auto monitoring = new Monitoring();
+                         monitoring->addBackend(std::make_unique<DPLMonitoringBackend>(registry));
+                         service = monitoring;
+                       } else {
+                         auto backend = isDefault ? "infologger://" : options.GetPropertyAsString("monitoring-backend");
+                         service = MonitoringFactory::Get(backend).release();
+                       }
                        return ServiceHandle{TypeIdHelpers::uniqueId<Monitoring>(), service};
                      },
                      noConfiguration(),
@@ -223,8 +238,14 @@ o2::framework::ServiceSpec CommonServices::driverClientSpec()
   return ServiceSpec{
     "driverClient",
     [](ServiceRegistry& services, DeviceState& state, fair::mq::ProgOptions& options) -> ServiceHandle {
+      auto backend = options.GetPropertyAsString("driver-client-backend");
+      if (backend == "stdout://") {
+        return ServiceHandle{TypeIdHelpers::uniqueId<DriverClient>(),
+                             new TextDriverClient(services, state)};
+      }
+      auto [ip, port] = o2::framework::parse_websocket_url(backend.c_str());
       return ServiceHandle{TypeIdHelpers::uniqueId<DriverClient>(),
-                           new TextDriverClient(services, state)};
+                           new WSDriverClient(services, state, ip.c_str(), port)};
     },
     noConfiguration(),
     nullptr,

Framework/Core/src/DPLMonitoringBackend.cxx

@@ -76,7 +76,7 @@ void DPLMonitoringBackend::send(o2::monitoring::Metric const& metric)
     mStream << o2::monitoring::tags::TAG_KEY[key] << "=" << o2::monitoring::tags::GetValue(value);
   }
   mStream << '\n';
-  mRegistry.get<framework::DriverClient>().tell(mStream.str().c_str());
+  mRegistry.get<framework::DriverClient>().tell(mStream.str());
 }
 
 } // namespace o2::framework

Framework/Core/src/DPLWebSocket.cxx

@@ -64,6 +64,7 @@ void websocket_server_callback(uv_stream_t* stream, ssize_t nread, const uv_buf_
   }
   try {
     parse_http_request(buf->base, nread, server);
+    free(buf->base);
   } catch (RuntimeErrorRef& ref) {
     auto& err = o2::framework::error_from_ref(ref);
     LOG(ERROR) << "Error while parsing request: " << err.what;
@@ -270,6 +271,22 @@ void ws_client_write_callback(uv_write_t* h, int status)
   }
 }
 
+void ws_client_bulk_write_callback(uv_write_t* h, int status)
+{
+  if (status) {
+    LOG(ERROR) << "uv_write error: " << uv_err_name(status);
+    free(h);
+    return;
+  }
+  std::vector<uv_buf_t>* buffers = (std::vector<uv_buf_t>*)h->data;
+  if (buffers) {
+    for (auto& b : *buffers) {
+      free(b.base);
+    }
+  }
+  delete buffers;
+}
+
 /// Helper to return an error
 void WSDPLClient::write(char const* message, size_t s)
 {
@@ -281,11 +298,14 @@ void WSDPLClient::write(char const* message, size_t s)
 
 void WSDPLClient::write(std::vector<uv_buf_t>& outputs)
 {
-  for (auto& msg : outputs) {
-    uv_write_t* write_req = (uv_write_t*)malloc(sizeof(uv_write_t));
-    write_req->data = msg.base;
-    uv_write(write_req, (uv_stream_t*)mStream, &msg, 1, ws_client_write_callback);
+  if (outputs.empty()) {
+    return;
   }
+  uv_write_t* write_req = (uv_write_t*)malloc(sizeof(uv_write_t));
+  std::vector<uv_buf_t>* buffers = new std::vector<uv_buf_t>;
+  buffers->swap(outputs);
+  write_req->data = buffers;
+  uv_write(write_req, (uv_stream_t*)mStream, &buffers->at(0), buffers->size(), ws_client_bulk_write_callback);
 }
 
 } // namespace o2::framework

Framework/Core/src/DataProcessingDevice.cxx

@@ -33,6 +33,7 @@
 #include "Framework/SourceInfoHeader.h"
 #include "Framework/Logger.h"
 #include "Framework/Monitoring.h"
+#include "Framework/DriverClient.h"
 #include "PropertyTreeHelpers.h"
 #include "DataProcessingStatus.h"
 #include "DataProcessingHelpers.h"
@@ -237,7 +238,8 @@ void DataProcessingDevice::Init()
     } else {
       str = entry.second.get_value<std::string>();
     }
-    LOG(INFO) << "[CONFIG] " << entry.first << "=" << str << " 1 " << configStore->provenance(entry.first.c_str());
+    std::string configString = fmt::format("[CONFIG] {}={} 1 {}", entry.first, str, configStore->provenance(entry.first.c_str())).c_str();
+    mServiceRegistry.get<DriverClient>().tell(configString.c_str());
   }
 
   mConfigRegistry = std::make_unique<ConfigParamRegistry>(std::move(configStore));

Framework/Core/src/DeviceSpecHelpers.cxx

@@ -1216,6 +1216,7 @@ boost::program_options::options_description DeviceSpecHelpers::getForwardedDevic
     ("post-fork-command", bpo::value<std::string>(), "post fork command to execute (e.g. numactl {pid}")                                      //
     ("session", bpo::value<std::string>(), "unique label for the shared memory session")                                                      //
     ("configuration,cfg", bpo::value<std::string>(), "configuration connection string")                                                       //
+    ("driver-client-backend", bpo::value<std::string>(), "driver connection string")                                                          //
     ("monitoring-backend", bpo::value<std::string>(), "monitoring connection string")                                                         //
     ("infologger-mode", bpo::value<std::string>(), "INFOLOGGER_MODE override")                                                                //
     ("infologger-severity", bpo::value<std::string>(), "minimun FairLogger severity which goes to info logger")                               //

Framework/Core/src/TextDriverClient.cxx

@@ -17,9 +17,9 @@ TextDriverClient::TextDriverClient(ServiceRegistry& registry, DeviceState& devic
 {
 }
 
-void TextDriverClient::tell(const char* msg)
+void TextDriverClient::tell(const char* msg, size_t s, bool flush)
 {
-  LOG(INFO) << msg;
+  LOG(INFO) << std::string_view{msg, s};
 }
 
 void TextDriverClient::flushPending()

Framework/Core/src/TextDriverClient.h

@@ -28,10 +28,10 @@ class TextDriverClient : public DriverClient
 
   /// The text based client simply sends a message on stdout which is
   /// (potentially) captured by the driver.
-  void tell(const char* msg) override;
+  void tell(char const* msg, size_t s, bool flush = true) final;
   /// Half duplex communication
-  void observe(const char* event, std::function<void(char const*)> callback) override{};
-  void flushPending() override;
+  void observe(const char* event, std::function<void(char const*)> callback) final{};
+  void flushPending() final;
 };
 
 } // namespace o2::framework