Use condition_variable for interruptible thread sleep in updateLoop

CHANGELOG.md

@@ -1,3 +1,6 @@
+## XX.XX.XX
+- Added `enableImmediateRequestOnStop` configuration option. When enabled, the update loop uses a condition variable instead of polling, allowing `stop()` and `setUpdateInterval()` to take effect immediately rather than waiting for the current sleep interval to expire.
+
 ## 23.2.4
 - Mitigated an issue where cached events were not queued when a user property was recorded.
 

CMakeLists.txt

@@ -113,7 +113,8 @@ if(COUNTLY_BUILD_TESTS)
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/event.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/crash.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/request.cpp
-    ${CMAKE_CURRENT_SOURCE_DIR}/tests/config.cpp)
+    ${CMAKE_CURRENT_SOURCE_DIR}/tests/config.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/tests/immediate_stop.cpp)
 
     target_compile_options(countly-tests PRIVATE -g)
     target_compile_definitions(countly-tests PRIVATE COUNTLY_BUILD_TESTS)

include/countly.hpp

@@ -5,6 +5,7 @@
 #include "countly/countly_configuration.hpp"
 
 #include <chrono>
+#include <condition_variable>
 #include <functional>
 #include <iterator>
 #include <map>
@@ -60,6 +61,8 @@ class Countly : public cly::CountlyDelegates {
 
   void disableAutoEventsOnUserProperties();
 
+  void enableImmediateRequestOnStop();
+
   void setHTTPClient(HTTPClientFunction fun);
 
   void setMetrics(const std::string &os, const std::string &os_version, const std::string &device, const std::string &resolution, const std::string &carrier, const std::string &app_version);
@@ -351,6 +354,7 @@ class Countly : public cly::CountlyDelegates {
   bool enable_automatic_session = false;
   bool stop_thread = false;
   bool running = false;
+  std::condition_variable stop_cv; // Wakes updateLoop immediately on stop
   size_t wait_milliseconds = COUNTLY_KEEPALIVE_INTERVAL;
 
   size_t max_events = COUNTLY_MAX_EVENTS_DEFAULT;

include/countly/countly_configuration.hpp

@@ -72,6 +72,13 @@ struct CountlyConfiguration {
 
   bool autoEventsOnUserProperties = true;
 
+  /**
+   * Enable immediate stop notification using a condition variable.
+   * When enabled, the update loop wakes immediately on stop instead of
+   * waiting for the current sleep interval to expire.
+   */
+  bool immediateRequestOnStop = false;
+
   HTTPClientFunction http_client_function = nullptr;
 
   nlohmann::json metrics;

src/countly.cpp

@@ -156,6 +156,16 @@ void Countly::disableAutoEventsOnUserProperties() {
   mutex->unlock();
 }
 
+void Countly::enableImmediateRequestOnStop() {
+  if (is_sdk_initialized) {
+    log(LogLevel::WARNING, "[Countly][enableImmediateRequestOnStop] You can not enable immediate request on stop after SDK initialization.");
+    return;
+  }
+
+  std::lock_guard<std::mutex> lk(*mutex);
+  configuration->immediateRequestOnStop = true;
+}
+
 void Countly::setMetrics(const std::string &os, const std::string &os_version, const std::string &device, const std::string &resolution, const std::string &carrier, const std::string &app_version) {
   if (is_sdk_initialized) {
     log(LogLevel::WARNING, "[Countly][setMetrics] You can not set metrics after SDK initialization.");
@@ -501,9 +511,13 @@ void Countly::stop() {
 }
 
 void Countly::_deleteThread() {
-  mutex->lock();
-  stop_thread = true;
-  mutex->unlock();
+  {
+    std::lock_guard<std::mutex> lk(*mutex);
+    stop_thread = true;
+  }
+  if (configuration->immediateRequestOnStop) {
+    stop_cv.notify_one();
+  }
   if (thread && thread->joinable()) {
     try {
       thread->join();
@@ -515,9 +529,13 @@ void Countly::_deleteThread() {
 }
 
 void Countly::setUpdateInterval(size_t milliseconds) {
-  mutex->lock();
-  wait_milliseconds = milliseconds;
-  mutex->unlock();
+  {
+    std::lock_guard<std::mutex> lk(*mutex);
+    wait_milliseconds = milliseconds;
+  }
+  if (configuration->immediateRequestOnStop) {
+    stop_cv.notify_one();
+  }
 }
 
 void Countly::addEvent(const cly::Event &event) {
@@ -1213,29 +1231,58 @@ std::chrono::system_clock::duration Countly::getSessionDuration() { return Count
 
 void Countly::updateLoop() {
   log(LogLevel::DEBUG, "[Countly][updateLoop]");
-  mutex->lock();
-  running = true;
-  mutex->unlock();
-  while (true) {
-    mutex->lock();
-    if (stop_thread) {
-      stop_thread = false;
+  {
+    std::lock_guard<std::mutex> lk(*mutex);
+    running = true;
+  }
+  if (configuration->immediateRequestOnStop) {
+    try {
+      while (true) {
+        {
+          std::unique_lock<std::mutex> lk(*mutex);
+          stop_cv.wait_for(lk, std::chrono::milliseconds(wait_milliseconds), [this] {
+            return stop_thread;
+          });
+          if (stop_thread) {
+            stop_thread = false;
+            running = false;
+            return;
+          }
+        }
+        if (enable_automatic_session == true && configuration->manualSessionControl == false) {
+          updateSession();
+        } else if (configuration->manualSessionControl == true) {
+          packEvents();
+        }
+        requestModule->processQueue(mutex);
+      }
+    } catch (...) {
+      std::lock_guard<std::mutex> lk(*mutex);
+      running = false;
+      log(LogLevel::ERROR, "[Countly][updateLoop] unexpected exception, stopping update loop");
+    }
+  } else {
+    while (true) {
+      mutex->lock();
+      if (stop_thread) {
+        stop_thread = false;
+        mutex->unlock();
+        break;
+      }
+      size_t last_wait_milliseconds = wait_milliseconds;
       mutex->unlock();
-      break;
+      std::this_thread::sleep_for(std::chrono::milliseconds(last_wait_milliseconds));
+      if (enable_automatic_session == true && configuration->manualSessionControl == false) {
+        updateSession();
+      } else if (configuration->manualSessionControl == true) {
+        packEvents();
+      }
+      requestModule->processQueue(mutex);
     }
-    size_t last_wait_milliseconds = wait_milliseconds;
+    mutex->lock();
+    running = false;
     mutex->unlock();
-    std::this_thread::sleep_for(std::chrono::milliseconds(last_wait_milliseconds));
-    if (enable_automatic_session == true && configuration->manualSessionControl == false) {
-      updateSession();
-    } else if (configuration->manualSessionControl == true) {
-      packEvents();
-    }
-    requestModule->processQueue(mutex);
   }
-  mutex->lock();
-  running = false;
-  mutex->unlock();
 }
 
 void Countly::enableRemoteConfig() {

src/storage_module_db.cpp

@@ -172,7 +172,7 @@ void StorageModuleDB::RQRemoveFront(std::shared_ptr<DataEntry> request) {
     }
 
     // Log the request ID being removed
-    _logger->log(LogLevel::DEBUG, "[Countly][StorageModuleDB] RQRemoveFront RequestID = " + request->getId());
+    _logger->log(LogLevel::DEBUG, "[Countly][StorageModuleDB] RQRemoveFront RequestID = " + std::to_string(request->getId()));
 
 #ifdef COUNTLY_USE_SQLITE
     sqlite3 *database;

tests/config.cpp

@@ -52,6 +52,7 @@ TEST_CASE("Validate setting configuration values") {
     CHECK(config.forcePost == false);
     CHECK(config.port == 443);
     CHECK(config.manualSessionControl == false);
+    CHECK(config.immediateRequestOnStop == false);
     CHECK(config.sha256_function == nullptr);
     CHECK(config.http_client_function == nullptr);
     CHECK(config.metrics.empty());
@@ -78,6 +79,7 @@ TEST_CASE("Validate setting configuration values") {
     ct.SetPath(TEST_DATABASE_NAME);
     ct.setMaxRQProcessingBatchSize(10);
     ct.enableManualSessionControl();
+    ct.enableImmediateRequestOnStop();
     ct.start("YOUR_APP_KEY", "https://try.count.ly", -1, false);
 
     // Get configuration values using Countly getters
@@ -97,6 +99,7 @@ TEST_CASE("Validate setting configuration values") {
     CHECK(config.forcePost == true);
     CHECK(config.port == 443);
     CHECK(config.manualSessionControl == true);
+    CHECK(config.immediateRequestOnStop == true);
     CHECK(config.sha256_function("custom SHA256") == customSha_1_returnValue);
 
     HTTPResponse response = config.http_client_function(true, "", "");
@@ -182,6 +185,7 @@ TEST_CASE("Validate setting configuration values") {
     ct.setSalt("new-salt");
     ct.setMaxRequestQueueSize(100);
     ct.SetPath("new_database.db");
+    ct.enableImmediateRequestOnStop();
 
     // get SDK configuration again and make sure that they haven't changed
     config = ct.getConfiguration();
@@ -199,6 +203,7 @@ TEST_CASE("Validate setting configuration values") {
     CHECK(config.breadcrumbsThreshold == 100);
     CHECK(config.forcePost == true);
     CHECK(config.port == 443);
+    CHECK(config.immediateRequestOnStop == false); // was never enabled before init, should stay false
     CHECK(config.sha256_function("custom SHA256") == customSha_1_returnValue);
 
     response = config.http_client_function(true, "", "");