MT#62181 AmThread: streamline

Use an atomic enum to track state: idle, running, stopping, or stopped.

Use atomic compare and exchange to ensure consistency without requiring
a mutex.

Never detach threads. Always require a join. This makes for an easy and
reliable way to wait for a thread to finish.

Provide a default on_stop so that it can be omitted in child classes.

Change-Id: Id59942638b7f769090ac91c2049b7d934e876ffc

core/AmThread.cpp

@@ -36,11 +36,6 @@ using std::string;
 #include <system_error>
 #include <exception>
 
-AmThread::AmThread()
-  : _stopped(true)
-{
-}
-
 void AmThread::_start()
 {
   _pid = static_cast<unsigned long>(_td.native_handle());
@@ -48,60 +43,41 @@ void AmThread::_start()
   run();
 
   DBG("Thread %lu is ending.\n", _pid);
-  _stopped = true;
-    
+
+  _state = state::stopped;
 }
 
 void AmThread::start()
 {
-  _pid = 0;
-
-  // unless placed here, a call seq like run(); join(); will not wait to join
-  // b/c creating the thread can take too long
-  bool expected = true;
-  if (!this->_stopped.compare_exchange_strong(expected, false)) {
-    ERROR("thread already running\n");
+  state expected = state::idle;
+  if(!_state.compare_exchange_strong(expected, state::running)) {
+    DBG("Thread %lu already running.\n", _pid);
     return;
   }
 
+  _pid = 0;
+
   _td = std::thread(&AmThread::_start, this);
-  //DBG("Thread %lu is just created.\n", (unsigned long int) _pid);
 }
 
 void AmThread::stop()
 {
-  lock_guard<AmMutex> _l(_m_td);
-
-  if(is_stopped()){
+  state expected = state::running;
+  if(!_state.compare_exchange_strong(expected, state::stopping)){
+    DBG("Thread %lu already stopped\n", _pid);
     return;
   }
 
   // gives the thread a chance to clean up
-  DBG("Thread %lu calling on_stop, give it a chance to clean up.\n", _pid);
-
-  try { on_stop(); } catch(...) {}
-
-  try {
-    _td.detach();
-  }
-  catch (std::system_error &e) {
-    WARN("Failed to detach thread: code %d (%s)\n", e.code().value(), e.what());
-  }
-  catch (std::exception &e) {
-    WARN("Failed to detach thread: %s\n", e.what());
-  }
-  catch (...) {
-    WARN("Failed to detach thread: unknown error\n");
-  }
-
-  DBG("Thread %lu finished detach.\n", _pid);
+  DBG("Thread %lu calling on_stop\n", _pid);
 
-  //pthread_cancel(_td);
+  on_stop();
 }
 
 void AmThread::join()
 {
-  if(!is_stopped())
+  // only when neither stopped nor joined
+  if (_td.joinable())
     _td.join();
 }
 
@@ -164,6 +140,7 @@ void AmThreadWatcher::run()
 	DBG("thread %lu is to be processed in thread watcher.\n", cur_thread->_pid);
 	if(cur_thread->is_stopped()){
 	  DBG("thread %lu has been destroyed.\n", cur_thread->_pid);
+	  cur_thread->join();
 	  delete cur_thread;
 	}
 	else {

core/AmThread.h

@@ -112,29 +112,42 @@ public:
 class AmThread
 {
   std::thread _td;
-  AmMutex   _m_td;
 
-  atomic_bool _stopped;
+  enum state {
+    idle, // not started yet
+    running,
+    stopping, // waiting to stop
+    stopped, // after stop
+  };
+
+  std::atomic<state> _state;
 
   void _start();
 
 protected:
   virtual void run()=0;
-  virtual void on_stop()=0;
+  virtual void on_stop() {};
 
 public:
   unsigned long _pid;
-  AmThread();
+
+  AmThread()
+    : _state(state::idle)
+  {}
+
   virtual ~AmThread() {}
 
   /** Start it ! */
   void start();
+
   /** Stop it ! */
   void stop();
-  /** @return true if this thread doesn't run. */
-  bool is_stopped() { return _stopped; }
+
   /** Wait for this thread to finish */
   void join();
+
+  /** @return true if this thread has finished. */
+  bool is_stopped() { return _state == stopped; }
 };
 
 /**