diff --git a/src/gateway.cpp b/src/gateway.cpp index 3017ce6..5e985d9 100644 --- a/src/gateway.cpp +++ b/src/gateway.cpp @@ -513,6 +513,8 @@ class GatewayImpl : public IGateway, private IClientTransport Server server; std::thread watchdogThread; std::atomic watchdogRunning; + std::condition_variable watchdogCv; + std::mutex watchdogMtx; bool legacyRPCv1; std::map 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 lock(connectionLog_mtx); + needsDisconnection = connectionStarted; + } + if (needsDisconnection) + { + disconnect(); } } @@ -618,6 +623,11 @@ class GatewayImpl : public IGateway, private IClientTransport FIREBOLT_LOG_ERROR("Gateway", "[connect] transport connect failed status=%d", static_cast(status)); return status; } + else + { + std::lock_guard lock(connectionLog_mtx); + connectionStarted = true; + } if (!watchdogRunning.exchange(true)) { @@ -625,9 +635,15 @@ class GatewayImpl : public IGateway, private IClientTransport watchdogThread = std::thread( [this]() { + std::unique_lock 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(std::chrono::duration_cast( - 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(std::chrono::duration_cast( - 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 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(std::chrono::duration_cast( + 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(std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0_nw) + .count())); + { + std::lock_guard lock(connectionLog_mtx); + connectionStarted = false; + } + } }; IGateway& GetGatewayInstance() diff --git a/test/unit/gatewayTest.cpp b/test/unit/gatewayTest.cpp index 67b519e..e638420 100644 --- a/test/unit/gatewayTest.cpp +++ b/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(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); +}