From f1e6a975de08d4155d18acc599c9bd0ddaf4c9c6 Mon Sep 17 00:00:00 2001
From: Andre Puschmann <andre@softwareradiosystems.com>
Date: Tue, 18 May 2021 06:32:23 +0200
Subject: [PATCH] thread_pool: fix concurrent access of state variable

the thread workers need access to their current state to exit properly
when they are set to state STOP. However, since the state is kept in
a std::vector for all workers, it seems more appropiate to add a per-thread
running variable rather then mutexing the entire vector.
---
 lib/include/srsran/common/thread_pool.h |  3 +++
 lib/src/common/thread_pool.cc           | 10 ++++++++--
 2 files changed, 11 insertions(+), 2 deletions(-)

diff --git a/lib/include/srsran/common/thread_pool.h b/lib/include/srsran/common/thread_pool.h
index a2ad3f1c6..0c3eb1674 100644
--- a/lib/include/srsran/common/thread_pool.h
+++ b/lib/include/srsran/common/thread_pool.h
@@ -23,6 +23,7 @@
 #include "srsran/adt/circular_buffer.h"
 #include "srsran/adt/move_callback.h"
 #include "srsran/srslog/srslog.h"
+#include <atomic>
 #include <condition_variable>
 #include <functional>
 #include <memory>
@@ -45,6 +46,7 @@ public:
     worker();
     ~worker() = default;
     void     setup(uint32_t id, thread_pool* parent, uint32_t prio = 0, uint32_t mask = 255);
+    void     stop();
     uint32_t get_id();
     void     release();
 
@@ -54,6 +56,7 @@ public:
   private:
     uint32_t     my_id     = 0;
     thread_pool* my_parent = nullptr;
+    std::atomic<bool> running   = {true};
 
     void run_thread();
     void wait_to_start();
diff --git a/lib/src/common/thread_pool.cc b/lib/src/common/thread_pool.cc
index ea40aa6c8..e74111d6b 100644
--- a/lib/src/common/thread_pool.cc
+++ b/lib/src/common/thread_pool.cc
@@ -42,15 +42,20 @@ void thread_pool::worker::setup(uint32_t id, thread_pool* parent, uint32_t prio,
 void thread_pool::worker::run_thread()
 {
   set_name(std::string("WORKER") + std::to_string(my_id));
-  while (my_parent->status[my_id] != STOP) {
+  while (running.load(std::memory_order_relaxed)) {
     wait_to_start();
-    if (my_parent->status[my_id] != STOP) {
+    if (running.load(std::memory_order_relaxed)) {
       work_imp();
       finished();
     }
   }
 }
 
+void thread_pool::worker::stop()
+{
+  running = false;
+}
+
 uint32_t thread_pool::worker::get_id()
 {
   return my_id;
@@ -92,6 +97,7 @@ void thread_pool::stop()
     for (uint32_t i = 0; i < nof_workers; i++) {
       if (workers[i]) {
         debug_thread("stop(): stopping %d\n", i);
+        workers[i]->stop();
         status[i] = STOP;
         cvar_worker[i].notify_all();
         cvar_queue.notify_all();