From 1f709bbf99a9ccd98be6a83d057355281e7b3573 Mon Sep 17 00:00:00 2001
From: Richard Fuchs <rfuchs@sipwise.com>
Date: Mon, 7 Jul 2025 13:54:48 -0400
Subject: [PATCH] MT#63171 AmThread: introduce lock & cond

Add a central mutex and condition variable to AmThread, which can be
used by subclasses in their ::run() to determine whether they should
sleep, run, or shut down. This will replace identical mechanism in
subclasses that have this implemented on their own.

Subclasses which don't need this are not affected.

Add `_state` to the list of members protected by this mutex, obsoleting
the need for atomic ops.

Refactor `_joined` as an enum state, also protected by run_mutex,
replacing the separate mutex. Use run_cond as signal to wake up thread
waiting on the join to the complete.

Notify this cond on shutdown, after setting the state to "stopping."

Add ::stop_requested_unlocked() for threads which use `run_mut`.

Change-Id: I0b640b54c9f601e86cbb25413cacb67e31fbe913
---
 core/AmThread.cpp | 45 ++++++++++++++++++++++++++++++++++-----------
 core/AmThread.h   | 32 ++++++++++++++++++++++++++------
 2 files changed, 60 insertions(+), 17 deletions(-)

diff --git a/core/AmThread.cpp b/core/AmThread.cpp
index 54dff29d..31d837b7 100644
--- a/core/AmThread.cpp
+++ b/core/AmThread.cpp
@@ -55,16 +55,19 @@ void AmThread::_start()
 
   DBG("Thread %lu is ending.\n", _pid);
 
+  std::lock_guard<std::mutex> _l(run_mut);
   _state = state::stopped;
 }
 
 void AmThread::start()
 {
-  state expected = state::idle;
-  if(!_state.compare_exchange_strong(expected, state::running)) {
+  std::lock_guard<std::mutex> _l(run_mut);
+
+  if (_state != state::idle) {
     DBG("Thread %lu already running.\n", _pid);
     return;
   }
+  _state = state::running;
 
   _pid = 0;
 
@@ -87,11 +90,17 @@ void AmThread::ready()
 
 void AmThread::stop()
 {
-  state expected = state::running;
-  if(!_state.compare_exchange_strong(expected, state::stopping)){
+  std::unique_lock<std::mutex> _l(run_mut);
+
+  if (_state != state::running) {
     DBG("Thread %lu already stopped\n", _pid);
     return;
   }
+  _state = state::stopping;
+
+  _l.unlock();
+
+  run_cond.notify_all();
 
   // gives the thread a chance to clean up
   DBG("Thread %lu calling on_stop\n", _pid);
@@ -101,18 +110,32 @@ void AmThread::stop()
 
 void AmThread::join()
 {
+  std::unique_lock<std::mutex> _l(run_mut);
+
   // don't attempt to join thread that doesn't exist
   if (_state == state::idle)
     return;
 
-  // make sure only one other thread joins this one. all others
-  // are made to wait through the mutex
-  std::lock_guard<std::mutex> _l(_join_mt);
-  if (!_joined) {
-    if (_td.joinable())
-      _td.join();
-    _joined = true;
+  // nothing to do if already done
+  if (_joined == join_state::joined)
+    return;
+
+  // is somebody else doing the joining? wait until done
+  while (_joined == join_state::joining)
+    run_cond.wait(_l);
+
+  if (_joined == join_state::joined)
+    return;
+
+  // we have to do the joining (state == unjoined)
+  if (_td.joinable()) {
+    _joined = join_state::joining;
+    _l.unlock();
+    _td.join();
+    _l.lock();
   }
+  _joined = join_state::joined;
+  run_cond.notify_all();
 }
 
 
diff --git a/core/AmThread.h b/core/AmThread.h
index 2b572535..e1d13723 100644
--- a/core/AmThread.h
+++ b/core/AmThread.h
@@ -123,10 +123,14 @@ class AmThread
     stopped, // after stop
   };
 
-  std::atomic<state> _state;
+  enum join_state {
+    unjoined, // nobody joining/joined yet
+    joining, // one thread is joining
+    joined, // has been joined
+  };
 
-  std::mutex _join_mt;
-  bool _joined;
+  state _state;
+  join_state _joined;
 
   void _start();
 
@@ -134,11 +138,24 @@ class AmThread
   bool _triggers_ready;
 
 protected:
+  // protects `_state`, `_joined`, and anything else the subclass wants to use it for,
+  // primarily meant to determine whether a thread should run or not
+  std::mutex run_mut;
+
+  // can be used as a waker for threads running in a loop
+  std::condition_variable run_cond;
+
   virtual void run()=0;
   virtual void on_stop() {};
 
+  /** @return true if this thread ought to stop, without obtaining run_mut. */
+  bool stop_requested_unlocked() const { return _state == stopping; }
+
   /** @return true if this thread ought to stop. */
-  bool stop_requested() { return _state == stopping; }
+  bool stop_requested() {
+    std::lock_guard<std::mutex> _l(run_mut);
+    return stop_requested_unlocked();
+  }
 
   std::optional<std::reference_wrapper<SdNotifier>> _sd_notifier;
 
@@ -150,7 +167,7 @@ public:
 
   AmThread(bool triggers_ready = false)
     : _state(state::idle),
-      _joined(false),
+      _joined(join_state::unjoined),
       _triggers_ready(triggers_ready),
       _pid(0)
   {}
@@ -168,7 +185,10 @@ public:
   void join();
 
   /** @return true if this thread has finished. */
-  bool is_stopped() { return _state == stopped; }
+  bool is_stopped() {
+    std::lock_guard<std::mutex> _l(run_mut);
+    return _state == stopped;
+  }
 };
 
 /**