From a03734a34ce2c2039f04a21d5024213e7f770ddd Mon Sep 17 00:00:00 2001
From: Kostas Alexopoulos <kostorr.alex@gmail.com>
Date: Thu, 11 Nov 2021 16:24:25 +0100
Subject: [PATCH] [lla] Make thread safe

---
 include/Lla/Session.h |  2 ++
 src/Session.cxx       | 28 +++++++++++++++++++++++-----
 src/SocketLock.cxx    |  2 ++
 3 files changed, 27 insertions(+), 5 deletions(-)

diff --git a/include/Lla/Session.h b/include/Lla/Session.h
index 336cc34..bcbb8ea 100644
--- a/include/Lla/Session.h
+++ b/include/Lla/Session.h
@@ -23,6 +23,7 @@
 #include "Lla/LockParameters.h"
 
 #include <memory>
+#include <mutex>
 
 namespace o2
 {
@@ -75,6 +76,7 @@ class Session
   LockParameters mLockParams;
   std::unique_ptr<InterprocessLockInterface> mLock;
   bool mIsStarted = false;
+  std::mutex mMutex;
 };
 
 } // namespace lla
diff --git a/src/Session.cxx b/src/Session.cxx
index 95e8632..f633f3d 100644
--- a/src/Session.cxx
+++ b/src/Session.cxx
@@ -115,6 +115,10 @@ void Session::checkAndSetParameters()
 
 bool Session::start()
 {
+  // In case of start when mutex is kept, immediately return
+  std::unique_lock<std::mutex> ul(mMutex, std::try_to_lock);
+  if (!ul.owns_lock()) { return false; }
+
   if (!isStarted()) {
     if (mLock->tryLock()) {
       mIsStarted = true;
@@ -128,19 +132,33 @@ bool Session::start()
 
 bool Session::timedStart(int timeOut)
 {
-  if (!isStarted()) {
-    if (mLock->timedLock(timeOut)) {
-      mIsStarted = true;
+  // In case of timed start keep trying to take the mutex and start
+  const auto start = std::chrono::steady_clock::now();
+  auto timeExceeded = [&]() { return ((std::chrono::steady_clock::now() - start) > std::chrono::milliseconds(timeOut)); };
+
+  while (!timeExceeded()) {
+
+    std::unique_lock<std::mutex> ul(mMutex, std::try_to_lock);
+    if (!ul.owns_lock()) { continue; }
+
+    if (!isStarted()) {
+      if (mLock->tryLock()) {
+        mIsStarted = true;
+        return true;
+      }
+    } else {
       return true;
     }
-    return false;
   }
 
-  return true;
+  return false;
 }
 
 void Session::stop()
 {
+  // In case of stop, block until mutex acquired
+  std::unique_lock<std::mutex> ul(mMutex);
+  
   if (isStarted()) {
     mLock->unlock();
     mIsStarted = false;
diff --git a/src/SocketLock.cxx b/src/SocketLock.cxx
index 87f68fe..2eefccc 100644
--- a/src/SocketLock.cxx
+++ b/src/SocketLock.cxx
@@ -71,6 +71,8 @@ bool SocketLock::tryLock()
     lock();
     return true;
   } catch (const LlaException& e) {
+    // DEBUG
+    //std::cerr << e.what() << std::endl;
   }
   return false;
 }