C++ Trial: Client/Server: close() and dtor waits for pending running_threads using jau::latch

- Helps avoiding runaway threads at unit test closing and object destruction (if at all)
- Shows running_thread count

2 files changed, 35 insertions, 0 deletions

diff --git a/trial/direct_bt/dbt_client01.hpp b/trial/direct_bt/dbt_client01.hpp
index 9d6dcdce..8ef6e6ce 100644
--- a/trial/direct_bt/dbt_client01.hpp
+++ b/trial/direct_bt/dbt_client01.hpp
@@ -29,6 +29,8 @@
 
 #include "dbt_client_test.hpp"
 
+#include <jau/latch.hpp>
+
 class DBTClient01;
 typedef std::shared_ptr<DBTClient01> DBTClient01Ref;
 
@@ -51,6 +53,7 @@ class DBTClient01 : public DBTClientTest {
 
         jau::sc_atomic_int deviceReadyCount = 0;
 
+        jau::latch running_threads = jau::latch(0);
         jau::sc_atomic_int disconnectCount = 0;
         jau::sc_atomic_int notificationsReceived = 0;
         jau::sc_atomic_int indicationsReceived = 0;
@@ -104,6 +107,7 @@ class DBTClient01 : public DBTClientTest {
                         const uint64_t td = jau::getCurrentMilliseconds() - parent.timestamp_t0; // adapter-init -> now
                         fprintf_td(stderr, "PERF: adapter-init -> FOUND__-0  %" PRIu64 " ms\n", td);
                     }
+                    parent.running_threads.count_up();
                     std::thread dc(&DBTClient01::connectDiscoveredDevice, &parent, device); // @suppress("Invalid arguments")
                     dc.detach();
                     return true;
@@ -200,6 +204,7 @@ class DBTClient01 : public DBTClientTest {
                 } else {
                     fprintf_td(stderr, "****** Client i470 disconnectDevice(delayed %d ms): client null\n", sleep_dur);
                 }
+                parent.running_threads.count_down();
             }
 
             void deviceReady(BTDeviceRef device, const uint64_t timestamp) override {
@@ -209,11 +214,13 @@ class DBTClient01 : public DBTClientTest {
                     fprintf_td(stderr, "****** Client READY-0: Processing[%d] %s\n", parent.deviceReadyCount.load(), device->toString(true).c_str());
 
                     // Be nice to Test* case, allowing to reach its own listener.deviceReady() added later
+                    parent.running_threads.count_up();
                     std::thread dc(&DBTClient01::processReadyDevice, &parent, device);
                     dc.detach();
                     // processReadyDevice(device); // AdapterStatusListener::deviceReady() explicitly allows prolonged and complex code execution!
 
                     if( parent.do_disconnect ) {
+                        parent.running_threads.count_up();
                         std::thread disconnectThread(&MyAdapterStatusListener::disconnectDevice, this, device);
                         disconnectThread.detach();
                     }
@@ -228,6 +235,7 @@ class DBTClient01 : public DBTClientTest {
 
                 parent.disconnectCount++;
 
+                parent.running_threads.count_up();
                 std::thread dc(&DBTClient01::removeDevice, &parent, device); // @suppress("Invalid arguments")
                 dc.detach();
             }
@@ -288,6 +296,11 @@ class DBTClient01 : public DBTClientTest {
             this->do_disconnect = do_disconnect_;
         }
 
+        ~DBTClient01() {
+            fprintf_td(stderr, "****** Client dtor: running_threads %zu\n", running_threads.value());
+            running_threads.wait_for( 10_s );
+        }
+
         std::string getName() override { return adapterName; }
 
         void setAdapter(BTAdapterRef clientAdapter_) override {
@@ -374,6 +387,7 @@ class DBTClient01 : public DBTClientTest {
             const uint16_t supervision_timeout = (uint16_t) getHCIConnSupervisorTimeout(conn_latency, (int) ( conn_interval_max * 1.25 ) /* ms */);
             res = device->connectLE(le_scan_interval, le_scan_window, conn_interval_min, conn_interval_max, conn_latency, supervision_timeout);
             fprintf_td(stderr, "****** Client Connecting Device: End result %s of %s\n", to_string(res).c_str(), device->toString().c_str());
+            running_threads.count_down();
         }
 
         void processReadyDevice(BTDeviceRef device) {
@@ -606,6 +620,7 @@ class DBTClient01 : public DBTClientTest {
                             success, completedMeasurementsSuccess.load(),
                             measurementsLeft.load(), notificationsReceived.load(), indicationsReceived.load(),
                             completedGATTCommands.load(), device->getAddressAndType().toString().c_str());
+            running_threads.count_down();
         }
 
         void removeDevice(BTDeviceRef device) {
@@ -614,6 +629,7 @@ class DBTClient01 : public DBTClientTest {
             if( REMOVE_DEVICE ) {
                 device->remove();
             }
+            running_threads.count_down();
         }
 
         static const bool le_scan_active = true; // default value
@@ -640,6 +656,8 @@ class DBTClient01 : public DBTClientTest {
             fprintf_td(stderr, "****** Client Close: %s\n", msg.c_str());
             clientAdapter->stopDiscovery();
             clientAdapter->removeStatusListener( myAdapterStatusListener );
+            fprintf_td(stderr, "****** Client close: running_threads %zu\n", running_threads.value());
+            running_threads.wait_for( 10_s );
         }
 
         bool initAdapter(BTAdapterRef adapter) override {
diff --git a/trial/direct_bt/dbt_server01.hpp b/trial/direct_bt/dbt_server01.hpp
index 4db62506..cdc47127 100644
--- a/trial/direct_bt/dbt_server01.hpp
+++ b/trial/direct_bt/dbt_server01.hpp
@@ -31,6 +31,8 @@
 
 #include "dbt_server_test.hpp"
 
+#include <jau/latch.hpp>
+
 class DBTServer01;
 typedef std::shared_ptr<DBTServer01> DBTServer01Ref;
 
@@ -72,6 +74,7 @@ class DBTServer01 : public DBTServerTest {
         bool use_SC = true;
         BTSecurityLevel adapterSecurityLevel = BTSecurityLevel::UNSET;
 
+        jau::latch running_threads = jau::latch(0);
         jau::sc_atomic_int disconnectCount = 0;
         jau::sc_atomic_int servedProtocolSessionsTotal = 0;
         jau::sc_atomic_int servedProtocolSessionsSuccess = 0;
@@ -296,6 +299,7 @@ class DBTServer01 : public DBTServerTest {
                 if( match ) {
                     parent.setDevice(nullptr);
                 }
+                parent.running_threads.count_up();
                 std::thread sd(&DBTServer01::processDisconnectedDevice, &parent, device); // @suppress("Invalid arguments")
                 sd.detach();
                 (void)timestamp;
@@ -377,6 +381,7 @@ class DBTServer01 : public DBTServerTest {
                             }
                         }
                     }
+                    parent.running_threads.count_down();
                 }
 
                 void disconnectDevice() {
@@ -395,6 +400,7 @@ class DBTServer01 : public DBTServerTest {
                     } else {
                         fprintf_td(stderr, "****** Server i470 disconnectDevice(delayed %d ms): client null\n", sleep_dur);
                     }
+                    parent.running_threads.count_down();
                 }
 
             public:
@@ -459,6 +465,7 @@ class DBTServer01 : public DBTServerTest {
                             match, parent.servedProtocolSessionsTotal.load(), parent.servingProtocolSessionsLeft.load(),
                             match ? (int)usedMTU_old : 0, (int)mtu, device->toString().c_str());
                     if( parent.do_disconnect ) {
+                        parent.running_threads.count_up();
                         std::thread disconnectThread(&MyGATTServerListener::disconnectDevice, this);
                         disconnectThread.detach();
                     }
@@ -518,6 +525,7 @@ class DBTServer01 : public DBTServerTest {
                                 parent.servingProtocolSessionsLeft--;
                             }
                         }
+                        parent.running_threads.count_up();
                         jau::POctets value2(value);
                         std::thread senderThread(&MyGATTServerListener::sendResponse, this, value2);
                         senderThread.detach();
@@ -590,6 +598,11 @@ class DBTServer01 : public DBTServerTest {
             dbGattServer->addListener( gattServerListener );
         }
 
+        ~DBTServer01() {
+            fprintf_td(stderr, "****** Server dtor: running_threads %zu\n", running_threads.value());
+            running_threads.wait_for( 10_s );
+        }
+
         std::string getName() override { return adapterName; }
 
         BTSecurityLevel getSecurityLevel() override { return adapterSecurityLevel; }
@@ -621,6 +634,9 @@ class DBTServer01 : public DBTServerTest {
                 }
             }
             gattServerListener->close();
+            fprintf_td(stderr, "****** Server close: running_threads %zu\n", running_threads.value());
+            running_threads.wait_for( 10_s );
+
             // dbGattServer = nullptr; // keep alive
             stopAdvertising(msg); // try once more in case of already started AdapterStatusListener
             fprintf_td(stderr, "****** Server Close.X: %s\n", msg.c_str());
@@ -704,6 +720,7 @@ class DBTServer01 : public DBTServerTest {
             }
 
             fprintf_td(stderr, "****** Server Disonnected Device: End %s\n", device->toString().c_str());
+            running_threads.count_down();
         }
 
     public: