AliceO2Group · ktf · Feb 24, 2021 · Feb 23, 2021
@@ -13,6 +13,7 @@
 #include "Framework/DeviceSpec.h"
 #include "HTTPParser.h"
 #include <algorithm>
+#include <atomic>
 #include <uv.h>
 #include <sys/types.h>
 #include <unistd.h>

@@ -16,6 +16,7 @@
 #include <string>
 #include <map>
 #include <functional>
+#include <atomic>

 class uv_stream_s;

@@ -70,11 +71,11 @@ struct WSDPLClient : public HTTPParser {
  /// Dump headers
  void dumpHeaders();
  void sendHandshake();
-  bool isConnected() { return mHandshaken; }
+  bool isHandshaken() { return mHandshaken; }

  std::string mNonce;
  DeviceSpec const& mSpec;
-  bool mHandshaken = false;
+  std::atomic<bool> mHandshaken = false;
  std::function<void()> mHandshake;
  uv_stream_t* mStream = nullptr;
  std::map<std::string, std::string> mHeaders;

@@ -28,6 +28,7 @@ void on_connect(uv_connect_t* connection, int status)
  auto onHandshake = [client]() {
    client->flushPending();
  };
+  std::lock_guard<std::mutex> lock(client->mutex());
  client->setDPLClient(std::make_unique<WSDPLClient>(connection->handle, client->spec(), onHandshake));
  client->sendHandshake();
 }
@@ -62,6 +63,7 @@ void sendMessageToDriver(std::unique_ptr<o2::framework::WSDPLClient>& client, ch
 void WSDriverClient::setDPLClient(std::unique_ptr<WSDPLClient> client)
 {
  mClient = std::move(client);
+  mConnected = true;
 }

 void WSDriverClient::sendHandshake()
@@ -76,23 +78,23 @@ void WSDriverClient::observe(const char*, std::function<void(char const*)>)

 void WSDriverClient::tell(const char* msg, size_t s, bool flush)
 {
-  static std::mutex tellMutex;
-  std::lock_guard<std::mutex> guard(tellMutex);
-
  static bool printed1 = false;
  static bool printed2 = false;
-  if (mClient && mClient->isConnected() && flush) {
+  if (mConnected && mClient->isHandshaken() && flush) {
    flushPending();
+    std::lock_guard<std::mutex> lock(mClientMutex);
    std::vector<uv_buf_t> outputs;
    encode_websocket_frames(outputs, msg, s, WebSocketOpCode::Binary, 0);
    mClient->write(outputs);
  } else {
+    std::lock_guard<std::mutex> lock(mClientMutex);
    encode_websocket_frames(mBacklog, msg, s, WebSocketOpCode::Binary, 0);
  }
 }

 void WSDriverClient::flushPending()
 {
+  std::lock_guard<std::mutex> lock(mClientMutex);
  static bool printed1 = false;
  static bool printed2 = false;
  if (!mClient) {
@@ -104,7 +106,7 @@ void WSDriverClient::flushPending()
    }
    return;
  }
-  if (!(mClient->isConnected())) {
+  if (!(mClient->isHandshaken())) {
    if (mBacklog.size() > 1000) {
      if (!printed2) {
        LOG(WARNING) << "Unable to communicate with driver because client is not connected";

@@ -16,6 +16,8 @@
 #include <memory>
 #include <string>
 #include <vector>
+#include <mutex>
+#include <atomic>

 typedef struct uv_connect_s uv_connect_t;

@@ -42,10 +44,13 @@ class WSDriverClient : public DriverClient
  DeviceSpec const& spec() { return mSpec; }
  // Initiate a websocket session
  void sendHandshake();
+  std::mutex& mutex() { return mClientMutex; }

 private:
+  // Whether or not we managed to connect.
+  std::atomic<bool> mConnected = false;
+  std::mutex mClientMutex;
  DeviceSpec const& mSpec;
-  bool mConnected = false;
  std::vector<uv_buf_t> mBacklog;
  uv_connect_t* mConnection = nullptr;
  std::unique_ptr<WSDPLClient> mClient;