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

@@ -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();
 }
 
 

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;
+  }
 };
 
 /**