From cbb5a3a3bf90acf4867babd56b00e8f1560b6f5b Mon Sep 17 00:00:00 2001
From: Richard Fuchs <rfuchs@sipwise.com>
Date: Mon, 7 Jul 2025 14:19:20 -0400
Subject: [PATCH] MT#63171 AmSessionProcessorThread: replace runcond

Use AmMutex facilities to protect members dictating whether to run or
not. Replaces `runcond`, `process_sessions_mut`, and ::on_stop().

Remove redundant wake-up.

Change-Id: I75c5c4ff2be2b66b36c7354c1e127bc4f5d9993c
---
 core/AmSessionProcessor.cpp | 27 +++++++++++----------------
 core/AmSessionProcessor.h   |  3 ---
 2 files changed, 11 insertions(+), 19 deletions(-)

diff --git a/core/AmSessionProcessor.cpp b/core/AmSessionProcessor.cpp
index aa93741c..9ce365b8 100644
--- a/core/AmSessionProcessor.cpp
+++ b/core/AmSessionProcessor.cpp
@@ -85,7 +85,7 @@ void AmSessionProcessor::stopThreads()
 
 
 AmSessionProcessorThread::AmSessionProcessorThread() 
-  : events(this), runcond(false)
+  : events(this)
 {
 }
 
@@ -93,27 +93,27 @@ AmSessionProcessorThread::~AmSessionProcessorThread() {
 }
 
 void AmSessionProcessorThread::notify(AmEventQueue* sender) {
-  process_sessions_mut.lock();
-  runcond.set(true);
+  run_mut.lock();
   process_sessions.insert(sender);
-  process_sessions_mut.unlock();
+  run_mut.unlock();
+  run_cond.notify_all();
 }
 
 void AmSessionProcessorThread::run() {
+  std::unique_lock<std::mutex> _l(run_mut);
 
-  while (!stop_requested()) {
+  while (!stop_requested_unlocked()) {
 
-    runcond.wait_for();
+    if (process_sessions.empty())
+      run_cond.wait(_l);
 
     DBG("running processing loop\n");
 
-    process_sessions_mut.lock();
-    runcond.set(false);
     // get the list of session s that need processing
     std::set<AmEventQueue*> pending_process_sessions 
       = process_sessions;
     process_sessions.clear();
-    process_sessions_mut.unlock();
+    _l.unlock();
 
     // process control events (AmSessionProcessorThreadAddEvent)
     events.processEvents();
@@ -166,11 +166,9 @@ void AmSessionProcessorThread::run() {
 	(*it)->finalize();
       }
     }
-  }
-}
 
-void AmSessionProcessorThread::on_stop() {
-  runcond.set(true);
+    _l.lock();
+  }
 }
 
 // AmEventHandler interface
@@ -195,9 +193,6 @@ void AmSessionProcessorThread::startSession(AmSession* s) {
 
   // trigger processing of events already in queue at startup
   notify(s);
-
-  // wakeup the thread
-  runcond.set(true);
 }
 
 #endif
diff --git a/core/AmSessionProcessor.h b/core/AmSessionProcessor.h
index 0a6c652a..b2377cf1 100644
--- a/core/AmSessionProcessor.h
+++ b/core/AmSessionProcessor.h
@@ -70,9 +70,7 @@ class AmSessionProcessorThread
   std::list<AmSession*> sessions;
   std::vector<AmSession*> startup_sessions;
 
-  AmCondition runcond;
   std::set<AmEventQueue*> process_sessions;
-  AmMutex process_sessions_mut;
 
   // AmEventHandler interface
   void process(AmEvent* e);
@@ -83,7 +81,6 @@ class AmSessionProcessorThread
 
   // AmThread interface
   void run();
-  void on_stop();
 
   // AmEventNotificationSink interface
   void notify(AmEventQueue* sender);