Sockets: make the Qt client's close() cancel in-flight sends

Common/Cpp/Sockets/ClientSocket_Qt.cpp

@@ -5,6 +5,7 @@
  */
 
 #include <chrono>
+#include <algorithm>
 #include <QSharedPointer>
 #include <QTimer>
 #include <QTcpSocket>
@@ -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<Mutex> 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<Mutex> 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<Mutex> 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<Mutex> 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<QSharedPointer<SendData>> cancelled;
+    {
+        std::lock_guard<Mutex> lg(m_lock);
+        cancelled = std::move(m_inflight_sends);
+    }
+    for (const auto& send_data : cancelled){
+        {
+            std::lock_guard<Mutex> 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<QTcpSocket> ClientSocket_Qt::make_socket(){
         this, &ClientSocket_Qt::internal_send,
         socket.get(), [this, sock = socket.get()](QSharedPointer<SendData> send_data){
             {
-                std::lock_guard<Mutex> lg(m_lock);
-                m_sends_in_flight.fetch_add(1, std::memory_order_acq_rel);
-                m_cv.notify_all();
+                std::lock_guard<Mutex> 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<QTcpSocket> ClientSocket_Qt::make_socket(){
                     && sock->state() == QAbstractSocket::ConnectedState
                     && std::chrono::steady_clock::now() < send_data->deadline
                 ){
+                    {
+                        std::lock_guard<Mutex> 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<QTcpSocket> 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<Mutex> lg(send_data->lock);
+                        send_data->bytes_sent = send_data->packet.size() - remaining;
+                    }
                 }
                 bytes_sent = send_data->packet.size() - remaining;
             }
 
             {
                 std::lock_guard<Mutex> 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()
+                );
             }
 
             {

Common/Cpp/Sockets/ClientSocket_Qt.h

@@ -9,7 +9,7 @@
 
 #include <chrono>
 #include <string>
-#include <atomic>
+#include <vector>
 #include <QObject>
 #include <QSharedPointer>
 #include "Common/Cpp/Concurrency/Mutex.h"
@@ -34,7 +34,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 +63,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 +97,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<QTcpSocket> make_socket();
 
 
 private:
     Mutex m_lock;
-    ConditionVariable m_cv;
 
     QTcpSocket* m_socket = nullptr;
     bool m_success_pending = false;
-    std::atomic<int> 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<QSharedPointer<SendData>> m_inflight_sends;
 };