Sockets: add an optional deadline to ServerSocket::send()

Common/Cpp/Sockets/AbstractServerSocket.h

@@ -12,6 +12,7 @@
 #ifndef PokemonAutomation_Sockets_AbstractServerSocket_H
 #define PokemonAutomation_Sockets_AbstractServerSocket_H
 
+#include <chrono>
 #include <string>
 #include "Common/Cpp/ListenerSet.h"
 
@@ -93,9 +94,17 @@ class AbstractServerSocket{
     //  threads).
     virtual void close() noexcept = 0;
 
-    //  Sends data to the active connection.
+    //  Sends data to the active connection. This blocks until every byte
+    //  has been written, the connection is lost, or "timeout" elapses.
+    //  "std::chrono::milliseconds::max()" (the default) waits indefinitely.
+    //  Returns the number of bytes actually written, which may be less
+    //  than "bytes" if the deadline elapsed or the connection was lost.
     //  If there is no active connection, returns 0.
-    virtual size_t send(const void* data, size_t bytes) = 0;
+    virtual size_t send(
+        const void* data,
+        size_t bytes,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds::max()
+    ) = 0;
 
 
 protected:

Common/Cpp/Sockets/ClientSocket_Qt.cpp

@@ -106,12 +106,21 @@ size_t ClientSocket_Qt::send(const void* data, size_t bytes, std::chrono::millis
     emit internal_send(send_data);
 
     //  Wait for the socket's event thread to finish writing the packet.
-    //  The write loop itself stops at "deadline", so this bound only
-    //  covers the queued call's start-up delay. The infinite default
-    //  keeps a bounded 5 s safety: if "close()" destroys the socket
-    //  before the queued call ever runs, it will never complete on its
-    //  own. The shared ownership makes the timeout safe: if it expires,
-    //  the queued call's own copy keeps the payload alive until it runs.
+    //  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
+    //  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)

Common/Cpp/Sockets/ServerSocket_POSIX.cpp

@@ -147,7 +147,7 @@ void ServerSocket_POSIX::close() noexcept{
     }
 }
 
-size_t ServerSocket_POSIX::send(const void* data, size_t bytes){
+size_t ServerSocket_POSIX::send(const void* data, size_t bytes, std::chrono::milliseconds timeout){
     int socket;
     {
         std::lock_guard<Mutex> lg(m_lock);
@@ -161,13 +161,25 @@ size_t ServerSocket_POSIX::send(const void* data, size_t bytes){
     const char* ptr = (const char*)data;
     size_t sent = 0;
 
+    //  "time_point::max()" (the infinite default) is never reached; using
+    //  it also avoids overflowing "now() + timeout".
+    const std::chrono::steady_clock::time_point deadline =
+        (timeout == std::chrono::milliseconds::max())
+            ? std::chrono::steady_clock::time_point::max()
+            : std::chrono::steady_clock::now() + timeout;
+
     while (bytes > 0){
         {
             std::lock_guard<Mutex> lg(m_lock);
             if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){
                 return sent;
             }
         }
+        if (std::chrono::steady_clock::now() >= deadline){
+            //  The deadline elapsed with bytes still unwritten; the
+            //  connection is still fine, so just report the partial count.
+            return sent;
+        }
 
         constexpr size_t BLOCK_SIZE = 1 << 30;
         size_t current = std::min(BLOCK_SIZE, bytes);

Common/Cpp/Sockets/ServerSocket_POSIX.h

@@ -9,6 +9,7 @@
 
 #include <string>
 #include <atomic>
+#include <chrono>
 #include "Common/Cpp/Concurrency/AsyncTask.h"
 #include "Common/Cpp/Concurrency/ConditionVariable.h"
 #include "Common/Cpp/Concurrency/Mutex.h"
@@ -32,7 +33,11 @@ class ServerSocket_POSIX final : public AbstractServerSocket{
     virtual void close_connection() override;
     virtual void close() noexcept override;
 
-    virtual size_t send(const void* data, size_t bytes) override;
+    virtual size_t send(
+        const void* data,
+        size_t bytes,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds::max()
+    ) override;
 
 
 private:

Common/Cpp/Sockets/ServerSocket_Qt.cpp

@@ -141,7 +141,7 @@ void ServerSocket_Qt::close() noexcept{
     m_close_requested = false;
 }
 
-size_t ServerSocket_Qt::send(const void* data, size_t bytes){
+size_t ServerSocket_Qt::send(const void* data, size_t bytes, std::chrono::milliseconds timeout){
     QTcpSocket* socket;
     {
         std::lock_guard<Mutex> lg(m_lock);
@@ -155,19 +155,40 @@ size_t ServerSocket_Qt::send(const void* data, size_t bytes){
 
     QSharedPointer<SendData> send_data = QSharedPointer<SendData>::create();
     send_data->packet.assign((const char*)data, bytes);
+    //  "time_point::max()" (the infinite default) is never reached; using
+    //  it also avoids overflowing "now() + timeout".
+    send_data->deadline =
+        (timeout == std::chrono::milliseconds::max())
+            ? std::chrono::steady_clock::time_point::max()
+            : std::chrono::steady_clock::now() + timeout;
 
     emit internal_send(send_data);
 
     //  Wait for the server's event thread to finish writing the packet.
-    //  The timeout covers the case where "close()" destroys the server
-    //  before the queued call ever runs, in which case it will never
-    //  complete on its own; the shared ownership keeps the packet alive
-    //  until the queued call runs (or is discarded with the server).
+    //  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()"/"close_connection()" 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 the 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 ownership keeps the packet alive until the
+    //  queued call runs (or is discarded with the server).
+    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,
-            std::chrono::seconds(5),
+            wait_time,
             [&send_data]{ return send_data->done; }
         );
     }
@@ -323,7 +344,13 @@ void ServerSocket_Qt::on_send(QSharedPointer<SendData> send_data){
     if (sock != nullptr){
         const char* ptr = send_data->packet.data();
         size_t remaining = send_data->packet.size();
-        while (remaining > 0 && sock->state() == QAbstractSocket::ConnectedState){
+        //  The deadline keeps a client that never reads from pinning this
+        //  event thread forever (which would also block "close()").
+        while (
+            remaining > 0
+            && sock->state() == QAbstractSocket::ConnectedState
+            && std::chrono::steady_clock::now() < send_data->deadline
+        ){
             qint64 current_sent = sock->write(ptr, remaining);
             if (current_sent <= 0){
                 //  The kernel's send buffer is full. Wait for some of it

Common/Cpp/Sockets/ServerSocket_Qt.h

@@ -7,6 +7,7 @@
 #ifndef PokemonAutomation_ServerSocket_Qt_H
 #define PokemonAutomation_ServerSocket_Qt_H
 
+#include <chrono>
 #include <string>
 #include <atomic>
 #include <QObject>
@@ -35,6 +36,11 @@ class ServerSocket_Qt final : public QObject, public AbstractServerSocket{
     //  discarded event leaves the caller's copy as the last reference.
     struct SendData{
         std::string packet;
+        //  The event-thread write loop stops at this deadline, so a send
+        //  that can never complete (a client that never reads) cannot pin
+        //  the server'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();
         size_t bytes_sent = 0;
         bool done = false;
 
@@ -74,7 +80,11 @@ class ServerSocket_Qt final : public QObject, public AbstractServerSocket{
     //  server that is running them.
     virtual void close() noexcept override;
 
-    virtual size_t send(const void* data, size_t bytes) override;
+    virtual size_t send(
+        const void* data,
+        size_t bytes,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds::max()
+    ) override;
 
 
 signals:

Common/Cpp/Sockets/ServerSocket_WinSocket.cpp

@@ -157,7 +157,7 @@ void ServerSocket_WinSocket::close() noexcept{
     }
 }
 
-size_t ServerSocket_WinSocket::send(const void* data, size_t bytes){
+size_t ServerSocket_WinSocket::send(const void* data, size_t bytes, std::chrono::milliseconds timeout){
     SOCKET socket;
     {
         std::lock_guard<Mutex> lg(m_lock);
@@ -171,13 +171,25 @@ size_t ServerSocket_WinSocket::send(const void* data, size_t bytes){
     const char* ptr = (const char*)data;
     size_t sent = 0;
 
+    //  "time_point::max()" (the infinite default) is never reached; using
+    //  it also avoids overflowing "now() + timeout".
+    const std::chrono::steady_clock::time_point deadline =
+        (timeout == std::chrono::milliseconds::max())
+            ? std::chrono::steady_clock::time_point::max()
+            : std::chrono::steady_clock::now() + timeout;
+
     while (bytes > 0){
         {
             std::lock_guard<Mutex> lg(m_lock);
             if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){
                 return sent;
             }
         }
+        if (std::chrono::steady_clock::now() >= deadline){
+            //  The deadline elapsed with bytes still unwritten; the
+            //  connection is still fine, so just report the partial count.
+            return sent;
+        }
 
         int current_sent = ::send(socket, ptr, (int)bytes, 0);
         if (current_sent != SOCKET_ERROR){

Common/Cpp/Sockets/ServerSocket_WinSocket.h

@@ -9,6 +9,7 @@
 
 #include <string>
 #include <atomic>
+#include <chrono>
 #include <winsock2.h>
 #include "Common/Cpp/Concurrency/AsyncTask.h"
 #include "Common/Cpp/Concurrency/ConditionVariable.h"
@@ -35,7 +36,11 @@ class ServerSocket_WinSocket final : public AbstractServerSocket{
     virtual void close_connection() override;
     virtual void close() noexcept override;
 
-    virtual size_t send(const void* data, size_t bytes) override;
+    virtual size_t send(
+        const void* data,
+        size_t bytes,
+        std::chrono::milliseconds timeout = std::chrono::milliseconds::max()
+    ) override;
 
 
 private: