From f3c4576d02b26afb5fef9ae62ecc33bdbecbd173 Mon Sep 17 00:00:00 2001 From: "ru.nazarov" Date: Tue, 28 Jul 2026 15:50:57 +0300 Subject: [PATCH 1/3] fix some race --- statshouse.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/statshouse.hpp b/statshouse.hpp index 1dc399f..779112a 100644 --- a/statshouse.hpp +++ b/statshouse.hpp @@ -987,9 +987,9 @@ class TransportTCP : public TransportUDPBase { return flush_packet_result::ok; } if (enqueue(*secondary, payload)) { - primary->force_reconnect.store(true, std::memory_order_relaxed); { std::lock_guard lock{primary->mu}; + primary->force_reconnect.store(true, std::memory_order_relaxed); primary->cv.notify_one(); } std::lock_guard lock{route_mu_}; From 5b4f7d911135a811f03e688c35d46e4379beacb0 Mon Sep 17 00:00:00 2001 From: "ru.nazarov" Date: Tue, 28 Jul 2026 15:56:10 +0300 Subject: [PATCH 2/3] fix some race --- statshouse.hpp | 3 --- 1 file changed, 3 deletions(-) diff --git a/statshouse.hpp b/statshouse.hpp index 779112a..a065b3d 100644 --- a/statshouse.hpp +++ b/statshouse.hpp @@ -1051,9 +1051,6 @@ class TransportTCP : public TransportUDPBase { void send_payload_with_reconnect(worker &w, const std::vector &payload) { for (;;) { - if (w.force_reconnect.exchange(false, std::memory_order_relaxed)) { - maybe_stuck_reconnect(w); - } if (w.sock < 0 && !reconnect(w)) { if (stop_.load()) { return; From d2d9e257dcc0d98542013bab05f3b61bc2d04c9a Mon Sep 17 00:00:00 2001 From: "ru.nazarov" Date: Tue, 28 Jul 2026 16:32:17 +0300 Subject: [PATCH 3/3] fix some race --- statshouse.hpp | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/statshouse.hpp b/statshouse.hpp index a065b3d..c1834a3 100644 --- a/statshouse.hpp +++ b/statshouse.hpp @@ -940,7 +940,7 @@ class TransportTCP : public TransportUDPBase { int sock = -1; std::thread th; std::atomic would_block_bytes{0}; - std::atomic_bool force_reconnect{false}; + bool force_reconnect = false; std::chrono::steady_clock::time_point last_stuck_recon{std::chrono::steady_clock::now()}; }; struct send_result { @@ -989,7 +989,7 @@ class TransportTCP : public TransportUDPBase { if (enqueue(*secondary, payload)) { { std::lock_guard lock{primary->mu}; - primary->force_reconnect.store(true, std::memory_order_relaxed); + primary->force_reconnect = true; primary->cv.notify_one(); } std::lock_guard lock{route_mu_}; @@ -1018,9 +1018,10 @@ class TransportTCP : public TransportUDPBase { { std::unique_lock lock{w->mu}; w->cv.wait(lock, [this, w]() { - return stop_.load() || !w->queue.empty() || w->force_reconnect.load(std::memory_order_relaxed); + return stop_.load() || !w->queue.empty() || w->force_reconnect; }); - if (w->force_reconnect.exchange(false, std::memory_order_relaxed)) { + if (w->force_reconnect) { + w->force_reconnect = false; need_recon = true; } if (!w->queue.empty()) {