From 62f6670da96a2cfc9f63607d743adfacb01eba54 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Mon, 2 Nov 2026 10:45:52 +0000 Subject: [PATCH 1/2] Sockets: make the Qt client's close() cancel in-flight sends close() now cancels the in-flight send() calls instead of waiting for them to finish: it marks them done so their callers return promptly with the number of bytes actually written so far (best effort), then destroys the QTcpSocket as before. How it works: - In-flight SendData objects are tracked in a list under m_lock (replacing the old send-in-flight counter). send() adds its entry before emitting the queued write. - close() and connect() (which destroys the previous socket) claim the list and mark every entry done + cancelled, waking the send() callers. The queued write loops see the flag and finish without writing (or update the running byte count as they go, so a close mid-write reports the bytes written so far). - The infinite default no longer carries the 5 s safety bound: "done" is now guaranteed to be set by either the event-thread write loop (deadline / socket error / Qt's userspace acceptance) or the cancel, so the wait can be unbounded. Finite timeouts keep their "timeout + 1 s" queueing margin. - The write loop reports its running byte count, so cancellation reports a best-effort count instead of always 0. Also adds a test: with the client's event thread blocked for 6 s in on_receive_data, a send started on the test thread queues behind the stall, and a close() from a worker thread must wake the send within a few milliseconds of starting (without the cancel, the send would only wake when the stall ends and the queued write finally runs). --- Common/Cpp/Sockets/ClientSocket_Qt.cpp | 126 ++++++++++----- Common/Cpp/Sockets/ClientSocket_Qt.h | 28 +++- Common/Cpp/Sockets/Sockets_Tests.cpp | 214 +++++++++++++++++++++++++ 3 files changed, 329 insertions(+), 39 deletions(-) diff --git a/Common/Cpp/Sockets/ClientSocket_Qt.cpp b/Common/Cpp/Sockets/ClientSocket_Qt.cpp index 55ee74f958..a866b33d00 100644 --- a/Common/Cpp/Sockets/ClientSocket_Qt.cpp +++ b/Common/Cpp/Sockets/ClientSocket_Qt.cpp @@ -5,6 +5,7 @@ */ #include +#include #include #include #include @@ -30,22 +31,19 @@ void ClientSocket_Qt::close() noexcept{ } m_state.store(State::DESTRUCTING, std::memory_order_release); m_success_pending = false; - m_cv.notify_all(); socket = m_socket; m_socket = nullptr; } - // Wait for in-flight "send()" calls to finish so we don't destroy - // the socket out from under them. - { - std::unique_lock lg(m_lock); - m_cv.wait(lg, [this]{ - return m_sends_in_flight.load(std::memory_order_acquire) == 0; - }); - } + // Cancel the in-flight "send()" calls so their callers return + // promptly with the best-effort byte count, instead of waiting for + // the event thread to finish the writes. + cancel_inflight_sends(); // Destroy the socket on its event thread. This blocks until the - // destruction has happened. + // destruction has happened. A queued "send()" call still runs before + // the destruction (the event loop is FIFO), but it is already + // cancelled and finishes without writing. GlobalThreadPools::qt_event_threadpool().remove_object(socket); std::lock_guard lg(m_lock); @@ -62,6 +60,12 @@ void ClientSocket_Qt::connect(const std::string& address, uint16_t port){ m_socket = nullptr; } if (old_socket != nullptr){ + // Cancel the in-flight sends before destroying the socket: a + // call still queued is discarded with the socket, and its + // caller must not be left waiting for a "done" that never + // comes. + cancel_inflight_sends(); + // Destroy the socket from the previous connection. This blocks // until the destruction has happened. GlobalThreadPools::qt_event_threadpool().remove_object(old_socket); @@ -103,40 +107,61 @@ size_t ClientSocket_Qt::send(const void* data, size_t bytes, std::chrono::millis ? std::chrono::steady_clock::time_point::max() : std::chrono::steady_clock::now() + timeout; + { + std::lock_guard lg(m_lock); + m_inflight_sends.push_back(send_data); + } + emit internal_send(send_data); - // Wait for the socket's event thread to finish writing the packet. - // A finite "timeout" waits for the deadline plus 1 s of queueing - // margin: the write loop stops at the deadline, so anything beyond - // that is just the queued call waiting to start. + // Wait for the socket's event thread to finish writing the packet, + // or for "close()"/"connect()" to cancel the send. A finite + // "timeout" waits for the deadline plus 1 s of queueing margin: the + // write loop stops at the deadline, so anything beyond that is just + // the queued call waiting to start. // - // The infinite default cannot wait indefinitely - a limitation of - // this wrapper, not of Qt: if "close()" runs from another thread - // and the event loop is destroyed while this call is still queued - // (posted after the destruction event), Qt silently discards the - // pending event, the handler never runs, and "done" is never set; - // an unbounded wait would then hang the caller forever. A send - // queued before close() is safe: the event loop is FIFO, so it - // runs first. The 5 s bound turns that post-close race into a - // "0 bytes written" result instead of a hang. The shared + // The infinite default waits without a time bound: "done" is set by + // either the event-thread write loop (which stops at the deadline, + // on a socket error, or when Qt's userspace buffer accepts the + // packet) or by "close()"/"connect()" cancelling the send before + // the socket is destroyed, so the wait is always ended. The shared // ownership keeps the packet alive until the queued call runs // (or is discarded with the socket). - const std::chrono::milliseconds wait_time = - (timeout == std::chrono::milliseconds::max()) - ? std::chrono::milliseconds(5000) - : timeout + std::chrono::milliseconds(1000); { std::unique_lock lg(send_data->lock); - send_data->cv.wait_for( - lg, - wait_time, - [&send_data]{ return send_data->done; } - ); + if (timeout == std::chrono::milliseconds::max()){ + send_data->cv.wait( + lg, + [&send_data]{ return send_data->done; } + ); + }else{ + send_data->cv.wait_for( + lg, + timeout + std::chrono::milliseconds(1000), + [&send_data]{ return send_data->done; } + ); + } } return send_data->bytes_sent; } +void ClientSocket_Qt::cancel_inflight_sends(){ + std::vector> cancelled; + { + std::lock_guard lg(m_lock); + cancelled = std::move(m_inflight_sends); + } + for (const auto& send_data : cancelled){ + { + std::lock_guard lg(send_data->lock); + send_data->cancelled = true; + send_data->done = true; + } + send_data->cv.notify_all(); + } +} + void ClientSocket_Qt::connection_failed(const std::string& error_message){ bool as_connect_failure; { @@ -263,9 +288,12 @@ std::unique_ptr ClientSocket_Qt::make_socket(){ this, &ClientSocket_Qt::internal_send, socket.get(), [this, sock = socket.get()](QSharedPointer send_data){ { - std::lock_guard lg(m_lock); - m_sends_in_flight.fetch_add(1, std::memory_order_acq_rel); - m_cv.notify_all(); + std::lock_guard lg(send_data->lock); + if (send_data->cancelled){ + // "close()"/"connect()" already cancelled this send + // and woken its caller; finish without writing. + return; + } } size_t bytes_sent = 0; @@ -280,6 +308,15 @@ std::unique_ptr ClientSocket_Qt::make_socket(){ && sock->state() == QAbstractSocket::ConnectedState && std::chrono::steady_clock::now() < send_data->deadline ){ + { + std::lock_guard lg(send_data->lock); + if (send_data->cancelled){ + // "close()" cancelled the send mid-write; its + // caller has already been woken with the + // count written so far. + break; + } + } qint64 current_sent = sock->write(ptr, remaining); if (current_sent <= 0){ // The kernel's send buffer is full. Wait for @@ -289,14 +326,29 @@ std::unique_ptr ClientSocket_Qt::make_socket(){ } ptr += current_sent; remaining -= (size_t)current_sent; + // Track the running count, so a "close()" that + // cancels this send mid-write reports the bytes + // written so far. + { + std::lock_guard lg(send_data->lock); + send_data->bytes_sent = send_data->packet.size() - remaining; + } } bytes_sent = send_data->packet.size() - remaining; } { std::lock_guard lg(m_lock); - m_sends_in_flight.fetch_sub(1, std::memory_order_acq_rel); - m_cv.notify_all(); + // "close()"/"connect()" may already have claimed this + // send; only remove it if it still owns its slot. + m_inflight_sends.erase( + std::remove( + m_inflight_sends.begin(), + m_inflight_sends.end(), + send_data + ), + m_inflight_sends.end() + ); } { diff --git a/Common/Cpp/Sockets/ClientSocket_Qt.h b/Common/Cpp/Sockets/ClientSocket_Qt.h index fd09116730..559b1a3c17 100644 --- a/Common/Cpp/Sockets/ClientSocket_Qt.h +++ b/Common/Cpp/Sockets/ClientSocket_Qt.h @@ -10,6 +10,7 @@ #include #include #include +#include #include #include #include "Common/Cpp/Concurrency/Mutex.h" @@ -34,7 +35,15 @@ class ClientSocket_Qt final : public QObject, public AbstractClientSocket{ // the socket's event thread forever and block "close()". "max()" // means no deadline (the infinite default). std::chrono::steady_clock::time_point deadline = std::chrono::steady_clock::time_point::max(); + // The running byte count: the write loop updates it as it goes, + // so "close()" cancelling the send mid-write can report the + // bytes written so far (best effort). size_t bytes_sent = 0; + // Set by "close()"/"connect()" when they cancel the send; the + // queued write loop then finishes without writing. + bool cancelled = false; + // Set when the event-thread write loop finishes or when + // "close()"/"connect()" cancel the send. bool done = false; Mutex lock; @@ -55,6 +64,10 @@ class ClientSocket_Qt final : public QObject, public AbstractClientSocket{ // callback is finished by the time it returns. It is safe to call // multiple times. // + // It cancels the in-flight "send()" calls: their callers return + // promptly with the number of bytes actually written so far (best + // effort), instead of waiting for the event thread to finish them. + // // This must not be called from within a listener callback, since the // callbacks run on the QTcpSocket's event thread and this destroys the // socket that is running them. @@ -85,16 +98,27 @@ class ClientSocket_Qt final : public QObject, public AbstractClientSocket{ // Runs on the QTcpSocket's event thread. void connection_failed(const std::string& error_message); + // Cancels the in-flight "send()" calls: marks them done so their + // callers return promptly with the bytes written so far (best + // effort). The queued write loops finish without writing. The + // shared ownership keeps each packet alive until its queued call + // runs (or is discarded with the socket). + void cancel_inflight_sends(); + std::unique_ptr make_socket(); private: Mutex m_lock; - ConditionVariable m_cv; QTcpSocket* m_socket = nullptr; bool m_success_pending = false; - std::atomic m_sends_in_flight{0}; + + // The "send()" calls that are queued on the event thread but not + // finished yet, so that "close()"/"connect()" can cancel them (and + // wake their callers) before the socket is destroyed. The write + // loops remove their entry when they finish. Protected by m_lock. + std::vector> m_inflight_sends; }; diff --git a/Common/Cpp/Sockets/Sockets_Tests.cpp b/Common/Cpp/Sockets/Sockets_Tests.cpp index 2d1d2b855d..9d3ef7b7e9 100644 --- a/Common/Cpp/Sockets/Sockets_Tests.cpp +++ b/Common/Cpp/Sockets/Sockets_Tests.cpp @@ -1002,6 +1002,215 @@ class Test_Sockets_Qt_CompletionLifetime : public UnitTest{ }; +// Verifies the Qt client's close()-cancels-sends guarantee under an +// event-thread stall: with the event thread blocked for 6 s in +// on_receive_data, a "send()" started on the calling thread is queued +// behind the stall, so "close()" is the only thing that can end it. +// "close()" must cancel it so the send's caller is woken promptly with +// a best-effort byte count, instead of the send's own bound expiring. +class Test_Sockets_Qt_CloseCancelsSend : public UnitTest{ +public: + Test_Sockets_Qt_CloseCancelsSend(ThreadPool& thread_pool, std::string name) + : UnitTest(std::move(name)) + , m_thread_pool(thread_pool) + { + m_threads = 2; + } + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + if (timing_unreliable()){ + return UnitTestResult( + UnitTestResult::SKIPPED, + "Machine too starved for a timing-sensitive socket test (timing_unreliable())." + ); + } + + // Blocks the client's event thread for 6 s on the first data, + // and flags it when the block starts. + struct StallingClientListener : AbstractClientSocket::Listener{ + Waiter waiter; + std::atomic stall_started{false}; + + virtual void on_receive_data(const void*, size_t) override{ + stall_started.store(true, std::memory_order_release); + waiter.notify(); + std::this_thread::sleep_for(std::chrono::seconds(6)); + } + }; + + ServerSocket_Qt server(m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port = 0; + { + std::mt19937 rng{std::random_device{}()}; + for (int attempt = 0; attempt < 8; attempt++){ + uint16_t candidate = (uint16_t)(40000 + rng() % 20000); + server.start("127.0.0.1", candidate); + if (server_listener.waiter.wait_for( + SOCKET_TEST_WAIT, + [&server_listener]{ return server_listener.listen_finished.load() > 0; } + ) && server_listener.last_listen_error.empty()){ + port = candidate; + break; + } + } + } + if (port == 0){ + if (timing_unreliable()){ + return UnitTestResult( + UnitTestResult::SKIPPED, + "The server did not start listening on a starved machine." + ); + } + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocket_Qt client(m_thread_pool); + ClientListener client_listener; + StallingClientListener stall_listener; + client.add_listener(client_listener); + client.add_listener(stall_listener); + client.connect("127.0.0.1", port); + if (!client_listener.waiter.wait_for( + SOCKET_TEST_WAIT, + [&client_listener]{ return client_listener.connect_finished.load() >= 1; } + )){ + if (timing_unreliable()){ + return UnitTestResult( + UnitTestResult::SKIPPED, + "The client did not finish connecting on a starved machine." + ); + } + error = "The client did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + SOCKET_TEST_WAIT, + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + if (timing_unreliable()){ + return UnitTestResult( + UnitTestResult::SKIPPED, + "The server did not call on_connect on a starved machine." + ); + } + error = "The server did not call on_connect"; + }else{ + // One byte out: the server echoes it back, and the echo + // blocks the client's event thread in on_receive_data + // for 6 s. + char byte = 'x'; + if (client.send(&byte, 1) != 1){ + error = "The client did not send the byte"; + }else if (!stall_listener.waiter.wait_for( + SOCKET_TEST_WAIT, + [&stall_listener]{ return stall_listener.stall_started.load(); } + )){ + error = "The client's event thread did not reach the stall"; + }else{ + // The event thread is now blocked for 6 s; anything + // queued to it waits behind the stall. This send + // (on the test thread, so it is queued well before + // the close below) will queue behind the stall, and + // the close() from the worker thread is the only + // thing that can end it. + constexpr size_t SIZE = 1 << 20; + std::vector payload(SIZE, 0xAB); + + auto close_start = std::chrono::steady_clock::time_point::max(); + long long close_ms = 0; + AsyncTask close_task = m_thread_pool.dispatch( + [&client, &close_start, &close_ms]{ + // Give the send a head start so it is in + // flight when close() runs. + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + auto t0 = std::chrono::steady_clock::now(); + close_start = t0; + client.close(); + close_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0 + ).count(); + } + ); + + size_t sent = client.send(payload.data(), payload.size()); + const auto send_end = std::chrono::steady_clock::now(); + + close_task.wait_and_ignore_exceptions(); + + // The send must have been woken by the cancel: it + // returns within a few milliseconds of close() + // starting, no matter when close() itself got to + // run. (Without cancellation it would only wake + // when the 6 s stall ends and the queued write + // finally runs.) + const long long cancel_lag = std::chrono::duration_cast( + send_end - close_start + ).count(); + + if (sent == payload.size()){ + // The event thread caught up before close() and + // the send completed on its own: the + // cancellation was not exercised, so there is + // nothing to assert on the send's timing. + if (close_ms > 60000){ + if (timing_unreliable()){ + return UnitTestResult( + UnitTestResult::SKIPPED, + "close() did not finish within 60 s on a starved machine." + ); + } + error = "close() did not finish within 60 s (" + std::to_string(close_ms) + " ms)"; + }else{ + ok = true; + } + }else if (cancel_lag < 0){ + // The send returned before close() even started: + // it was never in flight, so the test's premise + // did not hold. + error = "The send returned (" + std::to_string(sent) + " bytes) before close() even started"; + }else if (cancel_lag > 2000){ + if (timing_unreliable()){ + return UnitTestResult( + UnitTestResult::SKIPPED, + "send() was not woken promptly by close() on a starved machine." + ); + } + error = "send() did not return promptly after close() cancelled it (lag " + std::to_string(cancel_lag) + " ms)"; + }else if (close_ms > 60000){ + if (timing_unreliable()){ + return UnitTestResult( + UnitTestResult::SKIPPED, + "close() did not finish within 60 s on a starved machine." + ); + } + error = "close() did not finish within 60 s (" + std::to_string(close_ms) + " ms)"; + }else{ + ok = true; + } + } + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } + +private: + ThreadPool& m_thread_pool; +}; + + // The peer accepts the connection but stalls reading: a send() with a // deadline must stop at the deadline with a partial count, and // close() must still be able to finish afterwards (on the Qt backend @@ -1525,6 +1734,11 @@ void add_tests_Sockets(UnitTestDatabase& database, ThreadPool& thread_pool){ "Sockets (Qt) - Completion lifetime under event-thread stall" ); + database.add( + thread_pool, + "Sockets (Qt) - close() cancels a send queued behind an event-thread stall" + ); + #ifndef _WIN32 add_socket_tests(database, thread_pool, "POSIX"); #endif From 7a753ae4cd518f4b3a328012b9d43102d1e40047 Mon Sep 17 00:00:00 2001 From: Alexander Yee Date: Mon, 2 Nov 2026 23:52:44 -0800 Subject: [PATCH 2/2] Cleanup unused include. --- Common/Cpp/Sockets/ClientSocket_Qt.h | 1 - 1 file changed, 1 deletion(-) diff --git a/Common/Cpp/Sockets/ClientSocket_Qt.h b/Common/Cpp/Sockets/ClientSocket_Qt.h index 559b1a3c17..2e7c0f6918 100644 --- a/Common/Cpp/Sockets/ClientSocket_Qt.h +++ b/Common/Cpp/Sockets/ClientSocket_Qt.h @@ -9,7 +9,6 @@ #include #include -#include #include #include #include