Merge pull request #149 from Countly/instant_stop

Use condition_variable for interruptible thread sleep in updateLoop

Author: arifBurakDemiray

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## 26.1.0
 - ! Minor breaking change ! SDK Behavior Settings is now enabled by default. Changes made on SDK Manager > SDK Behavior Settings on your server will affect SDK behavior directly.
 
+- 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.
 - Added init config method "disableSDKBehaviorSettingsUpdates" to disable periodic SBS updates from the server.
 - Added init config method "setSDKBehaviorSettings" to provide server configuration in JSON format during initialization.
 

CMakeLists.txt

@@ -118,6 +118,7 @@ if(COUNTLY_BUILD_TESTS)
     ${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/immediate_stop.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tests/sbs.cpp)
 
     target_compile_options(countly-tests PRIVATE -g)
@@ -159,4 +160,4 @@ endif()
     CXX_EXTENSIONS NO)
 endif()
 
-install(TARGETS countly ARCHIVE DESTINATION lib PUBLIC_HEADER DESTINATION include/countly)
+install(TARGETS countly ARCHIVE DESTINATION lib PUBLIC_HEADER DESTINATION include/countly)

include/countly.hpp

@@ -4,7 +4,9 @@
 #include "countly/constants.hpp"
 #include "countly/countly_configuration.hpp"
 
+#include <atomic>
 #include <chrono>
+#include <condition_variable>
 #include <functional>
 #include <iterator>
 #include <map>
@@ -62,6 +64,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);
@@ -363,8 +367,8 @@ class Countly : public cly::CountlyDelegates {
   void updateLoop();
   void packEvents();
   bool began_session = false;
-  bool is_being_disposed = false;
-  bool is_sdk_initialized = false;
+  std::atomic<bool> is_being_disposed{false};
+  std::atomic<bool> is_sdk_initialized{false};
 
   std::chrono::system_clock::time_point last_sent_session_request;
   nlohmann::json session_params;
@@ -382,9 +386,10 @@ class Countly : public cly::CountlyDelegates {
   std::shared_ptr<std::mutex> mutex = std::make_shared<std::mutex>();
 
   bool is_queue_being_processed = false;
-  bool enable_automatic_session = false;
-  bool stop_thread = false;
-  bool running = false;
+  std::atomic<bool> enable_automatic_session{false};
+  std::atomic<bool> stop_thread{false};
+  std::atomic<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

@@ -45,7 +45,11 @@ Countly &Countly::getInstance() {
 }
 
 #ifdef COUNTLY_BUILD_TESTS
-void Countly::halt() { _sharedInstance.reset(new Countly()); }
+void Countly::halt() { 
+    if (_sharedInstance) {
+        _sharedInstance->stop();  // joins threads, releases mutex normally
+    }
+    _sharedInstance.reset(new Countly()); }
 #endif
 
 /**
@@ -157,6 +161,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, This method can't be called after SDK initialization. Returning.");
@@ -566,9 +580,11 @@ void Countly::stop() {
 }
 
 void Countly::_deleteThread() {
-  mutex->lock();
-  stop_thread = true;
-  mutex->unlock();
+  {
+    std::lock_guard<std::mutex> lk(*mutex);
+    stop_thread = true;
+  }
+  stop_cv.notify_one();
   if (thread && thread->joinable()) {
     try {
       thread->join();
@@ -580,9 +596,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) {
@@ -1380,30 +1400,69 @@ std::chrono::system_clock::duration Countly::getSessionDuration(std::chrono::sys
 std::chrono::system_clock::duration Countly::getSessionDuration() { return Countly::getSessionDuration(Countly::getTimestamp()); }
 
 void Countly::updateLoop() {
-  log(LogLevel::DEBUG, "[Countly] updateLoop, Start");
-  mutex->lock();
-  running = true;
-  mutex->unlock();
-  while (true) {
-    mutex->lock();
-    if (stop_thread) {
-      stop_thread = false;
+  log(LogLevel::DEBUG, "[Countly][updateLoop]");
+  {
+    std::lock_guard<std::mutex> lk(*mutex);
+    running = true;
+  }
+  try {
+    if (configuration->immediateRequestOnStop) {
+      while (true) {
+        {
+          std::unique_lock<std::mutex> lk(*mutex);
+          stop_cv.wait_for(lk, std::chrono::milliseconds(wait_milliseconds), [this] {
+            return stop_thread.load();
+          });
+          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);
+      }
+    } else {
+      while (true) {
+        size_t last_wait_milliseconds;
+        {
+          std::lock_guard<std::mutex> lk(*mutex);
+          if (stop_thread) {
+            stop_thread = false;
+            break;
+          }
+          last_wait_milliseconds = wait_milliseconds;
+        }
+        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);
+      }
+      std::lock_guard<std::mutex> lk(*mutex);
+      running = false;
+    }
+  } catch (const std::exception &e) {
+    bool acquired = mutex->try_lock();
+    running = false;
+    log(LogLevel::ERROR, std::string("[Countly][updateLoop] exception in update loop: ") + e.what());
+    if (acquired) {
       mutex->unlock();
-      break;
     }
-    size_t last_wait_milliseconds = wait_milliseconds;
-    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();
+  } catch (...) {
+    bool acquired = mutex->try_lock();
+    running = false;
+    log(LogLevel::FATAL, "[Countly][updateLoop] unknown non-std::exception caught, stopping update loop");
+    if (acquired) {
+      mutex->unlock();
     }
-    requestModule->processQueue(mutex);
   }
-  mutex->lock();
-  running = false;
-  mutex->unlock();
 }
 
 void Countly::enableRemoteConfig() {

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, "", "");