RDKEMW-22043: Upgrade the watchdog logic to use wait_for

src/gateway.cpp

@@ -513,6 +513,8 @@ class GatewayImpl : public IGateway, private IClientTransport
     Server server;
     std::thread watchdogThread;
     std::atomic<bool> watchdogRunning;
+    std::condition_variable watchdogCv;
+    std::mutex watchdogMtx;
     bool legacyRPCv1;
 
     std::map<MessageID, std::string> rpcv1_eventMap;
@@ -521,9 +523,11 @@ class GatewayImpl : public IGateway, private IClientTransport
     std::mutex connectionLog_mtx;
     bool hasLastConnectionLog{false};
     bool lastConnectionState{false};
+    bool connectionStarted{false};
     Firebolt::Error lastConnectionError{Firebolt::Error::None};
     std::chrono::steady_clock::time_point lastConnectionLogTs{};
     size_t suppressedConnectionNoticeCount{0};
+    std::mutex cleanup_mtx;
 
 public:
     GatewayImpl()
@@ -536,13 +540,14 @@ class GatewayImpl : public IGateway, private IClientTransport
 
     ~GatewayImpl()
     {
-        if (watchdogRunning)
+        bool needsDisconnection = false;
         {
-            watchdogRunning = false;
-            if (watchdogThread.joinable())
-            {
-                watchdogThread.join();
-            }
+            std::lock_guard<std::mutex> lock(connectionLog_mtx);
+            needsDisconnection = connectionStarted;
+        }
+        if (needsDisconnection)
+        {
+            disconnect();
         }
     }
 
@@ -618,16 +623,27 @@ class GatewayImpl : public IGateway, private IClientTransport
             FIREBOLT_LOG_ERROR("Gateway", "[connect] transport connect failed status=%d", static_cast<int>(status));
             return status;
         }
+        else
+        {
+            std::lock_guard<std::mutex> lock(connectionLog_mtx);
+            connectionStarted = true;
+        }
 
         if (!watchdogRunning.exchange(true))
         {
             FIREBOLT_LOG_DEBUG("Gateway", "[watchdog] starting thread (interval=%u ms)", watchdog_interval_ms);
             watchdogThread = std::thread(
                 [this]()
                 {
+                    std::unique_lock<std::mutex> lock(watchdogMtx);
                     while (watchdogRunning)
                     {
-                        std::this_thread::sleep_for(std::chrono::milliseconds(watchdog_interval_ms));
+                        if (watchdogCv.wait_for(lock, std::chrono::milliseconds(watchdog_interval_ms),
+                                                [this] { return !watchdogRunning; }))
+                        {
+                            break;
+                        }
+                        lock.unlock();
                         try
                         {
                             client.checkPromises();
@@ -640,6 +656,7 @@ class GatewayImpl : public IGateway, private IClientTransport
                         {
                             FIREBOLT_LOG_ERROR("Gateway", "[watchdog] checkPromises() threw unknown exception");
                         }
+                        lock.lock();
                     }
                 });
             FIREBOLT_LOG_DEBUG("Gateway", "[watchdog] thread started");
@@ -662,27 +679,7 @@ class GatewayImpl : public IGateway, private IClientTransport
         {
             return status;
         }
-        if (watchdogRunning.exchange(false))
-        {
-            FIREBOLT_LOG_DEBUG("Gateway", "[disconnect] waiting for watchdog thread join...");
-            auto t0_wdog = std::chrono::steady_clock::now();
-            if (watchdogThread.joinable())
-            {
-                watchdogThread.join();
-            }
-            FIREBOLT_LOG_DEBUG("Gateway", "[disconnect] watchdog joined in %lld ms",
-                               static_cast<long long>(std::chrono::duration_cast<std::chrono::milliseconds>(
-                                                          std::chrono::steady_clock::now() - t0_wdog)
-                                                          .count()));
-        }
-        client.cancelAll();
-        FIREBOLT_LOG_DEBUG("Gateway", "[disconnect] stopping notification worker...");
-        auto t0_nw = std::chrono::steady_clock::now();
-        server.stopNotificationWorker();
-        FIREBOLT_LOG_DEBUG("Gateway", "[disconnect] notification worker stopped in %lld ms",
-                           static_cast<long long>(std::chrono::duration_cast<std::chrono::milliseconds>(
-                                                      std::chrono::steady_clock::now() - t0_nw)
-                                                      .count()));
+        cleanupInternalState();
         return Error::None;
     }
 
@@ -945,6 +942,12 @@ class GatewayImpl : public IGateway, private IClientTransport
             }
         }
 
+        // Disconnection can also happen from the server, it's necessary to cleanup if this ever happens.
+        if (!connected)
+        {
+            cleanupInternalState();
+        }
+
         if (emitNotice)
         {
             FIREBOLT_LOG_NOTICE("Gateway", "[connection] state=%s error=%d suppressed_repeats=%zu",
@@ -978,6 +981,37 @@ class GatewayImpl : public IGateway, private IClientTransport
     {
         return transport.getResponseHeader(headerName);
     }
+
+    void cleanupInternalState()
+    {
+        std::lock_guard<std::mutex> lock(cleanup_mtx);
+        if (watchdogRunning.exchange(false))
+        {
+            watchdogCv.notify_all();
+            FIREBOLT_LOG_DEBUG("Gateway", "[disconnect] waiting for watchdog thread join...");
+            auto t0_wdog = std::chrono::steady_clock::now();
+            if (watchdogThread.joinable())
+            {
+                watchdogThread.join();
+            }
+            FIREBOLT_LOG_DEBUG("Gateway", "[disconnect] watchdog joined in %lld ms",
+                               static_cast<long long>(std::chrono::duration_cast<std::chrono::milliseconds>(
+                                                          std::chrono::steady_clock::now() - t0_wdog)
+                                                          .count()));
+        }
+        client.cancelAll();
+        FIREBOLT_LOG_DEBUG("Gateway", "[disconnect] stopping notification worker...");
+        auto t0_nw = std::chrono::steady_clock::now();
+        server.stopNotificationWorker();
+        FIREBOLT_LOG_DEBUG("Gateway", "[disconnect] notification worker stopped in %lld ms",
+                           static_cast<long long>(std::chrono::duration_cast<std::chrono::milliseconds>(
+                                                      std::chrono::steady_clock::now() - t0_nw)
+                                                      .count()));
+        {
+            std::lock_guard<std::mutex> lock(connectionLog_mtx);
+            connectionStarted = false;
+        }
+    }
 };
 
 IGateway& GetGatewayInstance()

test/unit/gatewayTest.cpp

@@ -1555,3 +1555,41 @@ TEST_F(GatewayUTest, DisconnectCancelsPendingRequests)
     EXPECT_FALSE(result.has_value());
     EXPECT_EQ(result.error(), Firebolt::Error::NotConnected);
 }
+
+// ---------------------------------------------------------------------------
+// Test name: GatewayUTest.DisconnectIsNotTimebound
+// Covers: disconnect() completes immediately without waiting for watchdog interval
+// Scenario type: performance
+// ---------------------------------------------------------------------------
+TEST_F(GatewayUTest, DisconnectIsNotTimebound)
+{
+    m_messageHandler = [](connection_hdl, server::message_ptr) {};
+
+    IGateway& gateway = connectAndWait();
+
+    // Fire a request that the server will never answer.
+    auto responseFuture = gateway.request("test.neverResponds", nlohmann::json{});
+
+    // The request is now in-flight and the future is pending. Disconnect and
+    // measure the time it takes. With the condition_variable::wait_for refactoring,
+    // disconnect() should complete immediately (< 100ms) rather than waiting
+    // for the full watchdog interval (500ms).
+    auto t0 = std::chrono::steady_clock::now();
+    Firebolt::Error disconnectErr = gateway.disconnect();
+    auto t1 = std::chrono::steady_clock::now();
+    auto disconnectDuration = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count();
+
+    EXPECT_EQ(disconnectErr, Firebolt::Error::None);
+
+    // Disconnect should complete well within the watchdog interval (500ms).
+    // Allow some overhead but it should be significantly faster than 500ms.
+    EXPECT_LT(disconnectDuration, 200) << "disconnect() took " << disconnectDuration
+                                       << "ms, expected < 200ms (watchdog interval is 500ms)";
+
+    // After disconnect() returns, cancelAll() must have resolved the promise.
+    ASSERT_EQ(responseFuture.wait_for(std::chrono::milliseconds(0)), std::future_status::ready);
+
+    auto result = responseFuture.get();
+    EXPECT_FALSE(result.has_value());
+    EXPECT_EQ(result.error(), Firebolt::Error::NotConnected);
+}