From 00e26792b213c0b7a2a9ad2be80e5b3a113b765e Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 02:44:49 +0000 Subject: [PATCH 01/15] Sockets: add the Qt ServerSocket and the loopback test harness Implements ServerSocket_Qt (a QTcpServer on the Qt event thread pool, mirroring ClientSocket_Qt; the implementation lives in ServerSocket_Qt.cpp, following the split of the client backends) and adds loopback tests that combine a ClientSocket and a ServerSocket against each other on 127.0.0.1. The test logic is written against the abstract AbstractClientSocket/AbstractServerSocket classes and is instantiated per backend; this PR instantiates the Qt backend on all platforms. The POSIX and WinSocket backends (and their test instantiations) land in separate PRs on top of this one, so that every platform runs 2 of the 3 backends and CI as a whole covers all 3. Scenarios: echo round trip, zero-length sends, a 1 MiB transfer with integrity check, rapid reconnect cycles, a send/close race, rejection of a second client, and a port-in-use start failure. Also documents that ServerSocket::close_connection() is safe to call from within a listener callback (close() is not: it joins the worker threads). --- Common/Cpp/Sockets/AbstractServerSocket.h | 5 +- Common/Cpp/Sockets/ServerSocket_Qt.cpp | 387 +++++++++ Common/Cpp/Sockets/ServerSocket_Qt.h | 124 +++ Common/Cpp/Sockets/Sockets_Tests.cpp | 763 ++++++++++++++++++ Common/Cpp/Sockets/Sockets_Tests.h | 28 + .../ComputerPrograms_UnitTestRunner.cpp | 2 + SerialPrograms/cmake/SourceFiles.cmake | 5 + 7 files changed, 1313 insertions(+), 1 deletion(-) create mode 100644 Common/Cpp/Sockets/ServerSocket_Qt.cpp create mode 100644 Common/Cpp/Sockets/ServerSocket_Qt.h create mode 100644 Common/Cpp/Sockets/Sockets_Tests.cpp create mode 100644 Common/Cpp/Sockets/Sockets_Tests.h diff --git a/Common/Cpp/Sockets/AbstractServerSocket.h b/Common/Cpp/Sockets/AbstractServerSocket.h index 8f9e13aacb..a3b53c921b 100644 --- a/Common/Cpp/Sockets/AbstractServerSocket.h +++ b/Common/Cpp/Sockets/AbstractServerSocket.h @@ -78,7 +78,10 @@ class AbstractServerSocket{ // State::LISTENING on success or State::NOT_RUNNING on failure. virtual void start(const std::string& address, uint16_t port) = 0; - // Closes the active connection, if any, and keeps listening. + // Closes the active connection, if any, and keeps listening. Unlike + // "close()", this does not wait for the worker threads, so it is safe + // to call from within a listener callback. "Listener::on_disconnect" + // is not called for the connection closed this way. virtual void close_connection() = 0; // Stops the server: it stops accepting new connections, closes the diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.cpp b/Common/Cpp/Sockets/ServerSocket_Qt.cpp new file mode 100644 index 0000000000..679c3014fc --- /dev/null +++ b/Common/Cpp/Sockets/ServerSocket_Qt.cpp @@ -0,0 +1,387 @@ +/* Server Socket (Qt) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#include +#include +#include +#include +#include "Common/Qt/QtThreadPool.h" +#include "Common/Qt/GlobalThreadPoolsQt.h" +#include "ServerSocket_Qt.h" + +namespace PokemonAutomation{ + + + +void ServerSocket_Qt::start(const std::string& address, uint16_t port){ + bool starting; + { + std::lock_guard lg(m_lock); + starting = + (m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING) + && m_server == nullptr + && !m_starting; + m_starting = true; + } + if (!starting){ + return; + } + + m_listen_error = ""; + QTcpServer* server = nullptr; + try{ + server = static_cast( + GlobalThreadPools::qt_event_threadpool().add_object( + [this, address, port]{ + return make_server(address, port); + } + ) + ); + } + catch (...){ + { + std::lock_guard lg(m_lock); + m_starting = false; + m_cv.notify_all(); + } + m_listeners.run_method(&Listener::on_listen_finished, "Failed to start the listener."); + throw; + } + bool listening; + { + std::lock_guard lg(m_lock); + m_starting = false; + m_server = server; + if (m_state.load(std::memory_order_relaxed) == State::DESTRUCTING){ + // "close()" is already in progress: it will destroy the + // server we just created. Leave the state for it. + }else{ + listening = m_listen_error.empty(); + m_state.store( + listening ? State::LISTENING : State::NOT_RUNNING, + std::memory_order_release + ); + } + m_cv.notify_all(); + } + + m_listeners.run_method(&Listener::on_listen_finished, m_listen_error); + } + +void ServerSocket_Qt::close_connection(){ + QThread* event_thread; + { + std::lock_guard lg(m_lock); + event_thread = m_event_thread; + if (m_server == nullptr){ + // Not running. + return; + } + } + + CloseCompletion completion; + emit internal_close_connection(&completion); + + // If called from a thread other than the server's event thread, the + // call is queued: wait until it has run, so the state is up to date + // when this returns (same as the POSIX backend). If called from the + // event thread (e.g. a listener callback), the call ran directly and + // this does not block. + if (QThread::currentThread() != event_thread){ + std::unique_lock lg(completion.lock); + completion.cv.wait_for( + lg, + std::chrono::seconds(5), + [&completion]{ return completion.done; } + ); + } + } +void ServerSocket_Qt::close() noexcept{ + QTcpServer* server; + { + std::lock_guard lg(m_lock); + if (m_server == nullptr && !m_starting){ + // Never started, or already closed. + return; + } + m_close_requested = true; + m_state.store(State::DESTRUCTING, std::memory_order_release); + } + + // Wait for an in-flight "start()" to finish so we destroy the + // server it creates, and for in-flight "send()" calls to finish so + // we don't destroy the server out from under them. + { + std::unique_lock lg(m_lock); + m_cv.wait(lg, [this]{ + return !m_starting + && m_sends_in_flight.load(std::memory_order_acquire) == 0; + }); + server = m_server; + m_server = nullptr; + m_socket = nullptr; + } + if (server == nullptr){ + // "start()" failed or was never reached: nothing to destroy. + std::lock_guard lg(m_lock); + m_state.store(State::NOT_RUNNING, std::memory_order_release); + m_close_requested = false; + return; + } + + // Destroy the server on its event thread. This blocks until the + // destruction has happened. + GlobalThreadPools::qt_event_threadpool().remove_object(server); + + std::lock_guard lg(m_lock); + m_state.store(State::NOT_RUNNING, std::memory_order_release); + m_close_requested = false; + } + +size_t ServerSocket_Qt::send(const void* data, size_t bytes){ + QTcpSocket* socket; + { + std::lock_guard lg(m_lock); + socket = + (m_state.load(std::memory_order_relaxed) == State::CONNECTED) ? m_socket : nullptr; + } + if (socket == nullptr){ + return 0; + } + + SendData send_data; + send_data.packet.assign((const char*)data, bytes); + + 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. + { + std::unique_lock lg(send_data.lock); + send_data.cv.wait_for( + lg, + std::chrono::seconds(5), + [&send_data]{ return send_data.done; } + ); + } + + return send_data.bytes_sent; + } + +void ServerSocket_Qt::on_new_connection(){ + QTcpServer* server; + { + std::lock_guard lg(m_lock); + server = m_server; + } + if (server == nullptr){ + return; + } + + while (true){ + QTcpSocket* sock = server->nextPendingConnection(); + if (sock == nullptr){ + break; + } + // The connection socket lives with the server, so "close()" + // destroys it on the event thread. + sock->setParent(server); + + bool accepted; + bool rejected; + { + std::lock_guard lg(m_lock); + State state = m_state.load(std::memory_order_relaxed); + if (state == State::CONNECTED){ + // A client is already connected: reject this one. + accepted = false; + rejected = true; + }else if (state != State::LISTENING){ + // The server is shutting down. + accepted = false; + rejected = false; + }else{ + m_socket = sock; + m_close_requested = false; + m_state.store(State::CONNECTED, std::memory_order_release); + accepted = true; + rejected = false; + } + } + + if (!accepted){ + if (rejected){ + m_listeners.run_method(&Listener::on_connect_rejected); + } + sock->abort(); + continue; + } + + QObject::connect( + sock, &QAbstractSocket::readyRead, + sock, [this, sock]{ + // readyRead only fires once per delivery, and more than + // one buffer's worth of data may be waiting, so read + // until the socket has no more data available. + while (true){ + qint64 available = sock->bytesAvailable(); + if (available <= 0){ + break; + } + + constexpr size_t BUFFER_SIZE = 4096; + char buffer[BUFFER_SIZE]; + qint64 bytes = sock->read(buffer, std::min(BUFFER_SIZE, available)); + if (bytes <= 0){ + break; + } + + m_listeners.run_method(&Listener::on_receive_data, buffer, bytes); + } + } + ); + QObject::connect( + sock, &QAbstractSocket::disconnected, + sock, [this, sock]{ + connection_lost(sock, "Connection closed by client."); + } + ); + QObject::connect( + sock, &QAbstractSocket::errorOccurred, + sock, [this, sock](QAbstractSocket::SocketError){ + connection_lost(sock, sock->errorString().toStdString()); + } + ); + + m_listeners.run_method(&Listener::on_connect); + } + } + +void ServerSocket_Qt::connection_lost(QTcpSocket* sock, const std::string& error_message){ + bool report; + { + std::lock_guard lg(m_lock); + if (m_socket != sock){ + // Already replaced or closed; the first callback wins. + return; + } + report = + !m_close_requested + && m_state.load(std::memory_order_relaxed) == State::CONNECTED; + m_socket = nullptr; + if (m_state.load(std::memory_order_relaxed) == State::CONNECTED){ + m_state.store(State::LISTENING, std::memory_order_release); + } + } + + if (report){ + m_listeners.run_method(&Listener::on_disconnect, error_message); + } + } + +void ServerSocket_Qt::on_close_connection(CloseCompletion* completion){ + QTcpSocket* sock; + { + std::lock_guard lg(m_lock); + if (m_socket != nullptr){ + m_close_requested = true; + m_state.store(State::LISTENING, std::memory_order_release); + sock = m_socket; + m_socket = nullptr; + }else{ + sock = nullptr; + } + } + if (sock != nullptr){ + sock->abort(); + } + + { + std::lock_guard lg(completion->lock); + completion->done = true; + } + completion->cv.notify_all(); + } + +void ServerSocket_Qt::on_send(SendData* send_data){ + QTcpSocket* sock; + { + std::lock_guard lg(m_lock); + m_sends_in_flight.fetch_add(1, std::memory_order_acq_rel); + m_cv.notify_all(); + sock = m_socket; + } + + size_t bytes_sent = 0; + if (sock != nullptr){ + const char* ptr = send_data->packet.data(); + size_t remaining = send_data->packet.size(); + while (remaining > 0 && sock->state() == QAbstractSocket::ConnectedState){ + qint64 current_sent = sock->write(ptr, remaining); + if (current_sent <= 0){ + // The kernel's send buffer is full. Wait for some of it + // to drain, then try again. + sock->waitForBytesWritten(10); + continue; + } + ptr += current_sent; + remaining -= (size_t)current_sent; + } + 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(); + } + + { + std::lock_guard lg(send_data->lock); + send_data->bytes_sent = bytes_sent; + send_data->done = true; + } + send_data->cv.notify_all(); + } + +std::unique_ptr ServerSocket_Qt::make_server(const std::string& address, uint16_t port){ + auto server = std::make_unique(); + if (!server->listen(QHostAddress(QString::fromStdString(address)), port)){ + std::lock_guard lg(m_lock); + m_listen_error = server->errorString().toStdString(); + } + + { + std::lock_guard lg(m_lock); + m_event_thread = QThread::currentThread(); + } + + QObject::connect( + this, &ServerSocket_Qt::internal_send, + server.get(), [this](SendData* send_data){ + on_send(send_data); + } + ); + QObject::connect( + this, &ServerSocket_Qt::internal_close_connection, + server.get(), [this](CloseCompletion* completion){ + on_close_connection(completion); + } + ); + QObject::connect( + server.get(), &QTcpServer::newConnection, + server.get(), [this]{ + on_new_connection(); + } + ); + + return server; + } + + +} diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.h b/Common/Cpp/Sockets/ServerSocket_Qt.h new file mode 100644 index 0000000000..0fd0e8a3b7 --- /dev/null +++ b/Common/Cpp/Sockets/ServerSocket_Qt.h @@ -0,0 +1,124 @@ +/* Server Socket (Qt) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#ifndef PokemonAutomation_ServerSocket_Qt_H +#define PokemonAutomation_ServerSocket_Qt_H + +#include +#include +#include +#include +#include "Common/Cpp/Concurrency/Mutex.h" +#include "Common/Cpp/Concurrency/ConditionVariable.h" +#include "AbstractServerSocket.h" + +class QThread; +class QTcpServer; +class QTcpSocket; + +namespace PokemonAutomation{ + + +class ThreadPool; + + +class ServerSocket_Qt final : public QObject, public AbstractServerSocket{ + Q_OBJECT + + struct SendData{ + std::string packet; + size_t bytes_sent = 0; + bool done = false; + + Mutex lock; + std::condition_variable cv; + }; + + // Synchronizes "close_connection()" with the QTcpServer's event thread + // when called from another thread. + struct CloseCompletion{ + Mutex lock; + std::condition_variable cv; + bool done = false; + }; + + +public: + ServerSocket_Qt(ThreadPool&){} + + virtual ~ServerSocket_Qt(){ + close(); + } + + // Binds to "address" and listens on "port" on the Qt event thread. This + // blocks until the QTcpServer has been created on that thread. + virtual void start(const std::string& address, uint16_t port) override; + + + virtual void close_connection() override; + + + // Stops the server, destroys the QTcpServer (and its connection + // socket), and returns the socket to State::NOT_RUNNING so "start()" + // can be called again. This blocks until the QTcpServer has been + // destroyed, so any in-flight listener callback is finished by the + // time it returns. It is safe to call multiple times. + // + // This must not be called from within a listener callback, since the + // callbacks run on the QTcpServer's event thread and this destroys the + // server that is running them. + virtual void close() noexcept override; + + + virtual size_t send(const void* data, size_t bytes) override; + + +signals: + void internal_send(SendData* send_data); + void internal_close_connection(CloseCompletion* completion); + + +private: + // Runs on the QTcpServer's event thread. + void on_new_connection(); + + // The active connection ended, or was aborted by "close_connection()". + // Only the first of the (errorOccurred, disconnected) signals for a + // given connection fires a callback, and "on_disconnect" is not fired + // for a connection closed by "close_connection()" or "close()". + // Runs on the QTcpServer's event thread. + void connection_lost(QTcpSocket* sock, const std::string& error_message); + + // Runs on the QTcpServer's event thread. + void on_close_connection(CloseCompletion* completion); + + // Runs on the QTcpServer's event thread. The context object is the + // server, so the call still runs if the connection socket has already + // been aborted, and it just reports 0 bytes in that case. + void on_send(SendData* send_data); + + std::unique_ptr make_server(const std::string& address, uint16_t port); + +private: + Mutex m_lock; + ConditionVariable m_cv; + + QTcpServer* m_server = nullptr; + QTcpSocket* m_socket = nullptr; + QThread* m_event_thread = nullptr; + bool m_close_requested = false; + bool m_starting = false; + std::atomic m_sends_in_flight{0}; + + // Written on the event thread by "make_server()", read after + // "add_object()" has returned. + std::string m_listen_error; +}; + + + +} +#endif diff --git a/Common/Cpp/Sockets/Sockets_Tests.cpp b/Common/Cpp/Sockets/Sockets_Tests.cpp new file mode 100644 index 0000000000..09cba432b1 --- /dev/null +++ b/Common/Cpp/Sockets/Sockets_Tests.cpp @@ -0,0 +1,763 @@ +/* Sockets Tests + * + * From: https://github.com/PokemonAutomation/ + * + * Loopback tests that combine a ClientSocket and a ServerSocket against + * each other on 127.0.0.1. The test logic is written against the abstract + * classes, and is instantiated for every backend that compiles on the + * current platform. Qt is instantiated here; the POSIX and WinSocket + * backends add their own instantiations in separate PRs, so that every + * platform runs 2 of the 3 backends and CI as a whole covers all 3. + * + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "Common/Cpp/Concurrency/AsyncTask.h" +#include "Common/Cpp/Concurrency/ThreadPool.h" +#include "Common/Cpp/Sockets/ClientSocket_Qt.h" +#include "Common/Cpp/Sockets/ServerSocket_Qt.h" +#include "Common/Cpp/TestRunners/UnitTestDatabase.h" +#include "Sockets_Tests.h" + +namespace PokemonAutomation{ + +namespace{ + + +struct Waiter{ + std::mutex lock; + std::condition_variable cv; + + bool wait_for(std::chrono::milliseconds timeout, const std::function& pred){ + std::unique_lock lg(lock); + return cv.wait_for(lg, timeout, pred); + } + + void notify(){ + std::lock_guard lg(lock); + cv.notify_all(); + } +}; + + +struct ClientListener : AbstractClientSocket::Listener{ + Waiter waiter; + std::atomic connect_finished{0}; + std::string last_connect_error; + std::atomic disconnected{0}; + std::string last_disconnect_error; + std::vector received; + + virtual void on_connect_finished(const std::string& error_message) override{ + connect_finished++; + last_connect_error = error_message; + waiter.notify(); + } + + virtual void on_receive_data(const void* data, size_t bytes) override{ + const uint8_t* ptr = (const uint8_t*)data; + received.insert(received.end(), ptr, ptr + bytes); + waiter.notify(); + } + + virtual void on_disconnect(const std::string& error_message) override{ + disconnected++; + last_disconnect_error = error_message; + waiter.notify(); + } +}; + + +struct ServerListener : AbstractServerSocket::Listener{ + Waiter waiter; + std::atomic listen_finished{0}; + std::string last_listen_error; + std::atomic connected{0}; + std::atomic rejected{0}; + std::atomic disconnected{0}; + std::string last_disconnect_error; + std::atomic bytes_received{0}; + + // When set, received data is echoed back to the client. + AbstractServerSocket* server = nullptr; + + virtual void on_listen_finished(const std::string& error_message) override{ + listen_finished++; + last_listen_error = error_message; + waiter.notify(); + } + + virtual void on_connect() override{ + connected++; + waiter.notify(); + } + + virtual void on_connect_rejected() override{ + rejected++; + waiter.notify(); + } + + virtual void on_receive_data(const void* data, size_t bytes) override{ + bytes_received += bytes; + if (server != nullptr){ + server->send(data, bytes); + } + waiter.notify(); + } + + virtual void on_disconnect(const std::string& error_message) override{ + disconnected++; + last_disconnect_error = error_message; + waiter.notify(); + } +}; + + +template +class TestSockets : public UnitTest{ +public: + TestSockets(ThreadPool& thread_pool, std::string name) + : UnitTest(std::move(name)) + , m_thread_pool(thread_pool) + { + m_threads = 2; + } + + +protected: + bool start_server(ServerSocketType& server, ServerListener& listener, uint16_t& port) const{ + for (int attempt = 0; attempt < 8; attempt++){ + uint16_t candidate; + { + std::lock_guard lg(m_rng_lock); + candidate = (uint16_t)(40000 + m_rng() % 20000); + } + server.start("127.0.0.1", candidate); + if (!listener.waiter.wait_for( + std::chrono::seconds(5), + [&listener]{ return listener.listen_finished.load() > 0; } + )){ + continue; + } + if (listener.last_listen_error.empty()){ + port = candidate; + return true; + } + } + return false; + } + + bool connect_client( + ClientSocketType& client, + ClientListener& listener, + uint16_t port, + int expected_connect + ) const{ + client.connect("127.0.0.1", port); + return listener.waiter.wait_for( + std::chrono::seconds(5), + [&listener, expected_connect]{ + return listener.connect_finished.load() >= expected_connect; + } + ); + } + +protected: + ThreadPool& m_thread_pool; + +private: + mutable std::mutex m_rng_lock; + mutable std::mt19937 m_rng{std::random_device{}()}; +}; + + +// Connect, send a packet, verify the echo, disconnect, and check that the +// server observed both the connect and the disconnect. +template +class Test_Sockets_Echo : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + error = "The client did not finish connecting"; + }else if (!client_listener.last_connect_error.empty()){ + error = "The client failed to connect: " + client_listener.last_connect_error; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else if (client.send("hello world", 11) != 11){ + error = "The client did not send all 11 bytes"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.bytes_received.load() >= 11; } + )){ + error = "The server did not receive the data"; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.received.size() >= 11; } + )){ + error = "The client did not receive the echo"; + }else if (std::memcmp(client_listener.received.data(), "hello world", 11) != 0){ + error = "The echo data was corrupted"; + }else{ + ok = true; + } + + client.close(); + } + + if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.disconnected.load() >= 1; } + )){ + error = "The server did not call on_disconnect"; + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Zero-length sends must be a valid no-op that returns 0 without +// touching the connection. +template +class Test_Sockets_ZeroLengthSend : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + error = "The client did not finish connecting"; + }else if (client.send("", 0) != 0){ + error = "A zero-length client send did not return 0"; + }else if (server.send("", 0) != 0){ + error = "A zero-length server send did not return 0"; + }else{ + // The connection must still be usable afterwards. + if (client.send("ping", 4) != 4){ + error = "The client did not send all 4 bytes after a zero-length send"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.bytes_received.load() >= 4; } + )){ + error = "The server did not receive the data"; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.received.size() >= 4; } + )){ + error = "The client did not receive the echo"; + }else if (std::memcmp(client_listener.received.data(), "ping", 4) != 0){ + error = "The echo data was corrupted"; + }else{ + ok = true; + } + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Transfer a megabyte each way (the server echoes it back) and verify the +// data integrity. +template +class Test_Sockets_LargeTransfer : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + constexpr size_t SIZE = 1 << 20; + std::vector payload(SIZE); + for (size_t i = 0; i < SIZE; i++){ + payload[i] = (uint8_t)(i % 251); + } + + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + error = "The client did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else if (client.send(payload.data(), payload.size()) != payload.size()){ + error = "The client did not send all " + std::to_string(payload.size()) + " bytes"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(30), + [&server_listener]{ return server_listener.bytes_received.load() >= SIZE; } + )){ + error = "The server did not receive all the data"; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(30), + [&client_listener]{ return client_listener.received.size() >= SIZE; } + )){ + error = "The client did not receive all the echo data"; + }else if (std::memcmp(client_listener.received.data(), payload.data(), SIZE) != 0){ + error = "The echo data was corrupted"; + }else{ + ok = true; + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Rapidly connect and disconnect many times. Each cycle transfers a +// packet to make sure the new connection actually works. +template +class Test_Sockets_RapidReconnect : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + constexpr int CYCLES = 50; + + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + size_t expected_echo = 0; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + for (int cycle = 0; cycle < CYCLES; cycle++){ + int cycle_no = cycle + 1; + + if (!this->connect_client(client, client_listener, port, cycle_no)){ + error = "The client did not finish connecting (cycle " + std::to_string(cycle_no) + ")"; + break; + } + if (!client_listener.last_connect_error.empty()){ + error = "The client failed to connect (cycle " + std::to_string(cycle_no) + "): " + client_listener.last_connect_error; + break; + } + if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener, cycle_no]{ return server_listener.connected.load() >= cycle_no; } + )){ + error = "The server did not call on_connect (cycle " + std::to_string(cycle_no) + ")"; + break; + } + + std::string packet = "cycle " + std::to_string(cycle_no); + expected_echo += packet.size(); + if (client.send(packet.data(), packet.size()) != packet.size()){ + error = "The client did not send the packet (cycle " + std::to_string(cycle_no) + ")"; + break; + } + if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener, expected_echo]{ return client_listener.received.size() >= expected_echo; } + )){ + error = "The client did not receive the echo (cycle " + std::to_string(cycle_no) + ")"; + break; + } + + client.close(); + if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener, cycle_no]{ return server_listener.disconnected.load() >= cycle_no; } + )){ + error = "The server did not call on_disconnect (cycle " + std::to_string(cycle_no) + ")"; + break; + } + } + + if (error.empty()){ + ok = true; + } + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Send a large packet while the server closes the connection underneath +// it: the send must finish (returning however much was written), the +// server must go back to listening, and the client must observe the +// disconnect. Then the server must accept a new connection. +template +class Test_Sockets_SendCloseRace : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + constexpr size_t SIZE = 1 << 20; + std::vector payload(SIZE, 0xAB); + + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + error = "The client did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else{ + std::atomic bytes_sent{0}; + AsyncTask send_task = this->m_thread_pool.dispatch([&client, &payload, &bytes_sent]{ + bytes_sent = client.send(payload.data(), payload.size()); + }); + + // Give the send a head start, then cut the connection from + // the server side. + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + server.close_connection(); + + send_task.wait_and_ignore_exceptions(); + + if (server.state() != AbstractServerSocket::State::LISTENING){ + error = "The server is not listening after close_connection()"; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.disconnected.load() >= 1; } + )){ + error = "The client did not observe the disconnect"; + }else if (bytes_sent.load() > SIZE){ + error = "The client reported sending more bytes than the packet size"; + }else{ + // The server must accept a new connection. + if (!this->connect_client(client, client_listener, port, 2)){ + error = "The client did not reconnect"; + }else if (!client_listener.last_connect_error.empty()){ + error = "The client failed to reconnect: " + client_listener.last_connect_error; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 2; } + )){ + error = "The server did not accept the new connection"; + }else if (client.send("x", 1) != 1){ + error = "The new connection does not work"; + }else{ + ok = true; + } + } + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// While a client is connected, a second client must be rejected, and the +// first client must keep working. +template +class Test_Sockets_RejectSecondClient : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocketType client_a(this->m_thread_pool); + ClientListener client_a_listener; + client_a.add_listener(client_a_listener); + + ClientSocketType client_b(this->m_thread_pool); + ClientListener client_b_listener; + client_b.add_listener(client_b_listener); + + if (!this->connect_client(client_a, client_a_listener, port, 1)){ + error = "Client A did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else{ + client_b.connect("127.0.0.1", port); + if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.rejected.load() >= 1; } + )){ + error = "The server did not reject the second client"; + }else{ + // The first client must keep working. + if (client_a.send("still here?", 11) != 11){ + error = "Client A did not send all 11 bytes"; + }else if (!client_a_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_a_listener]{ return client_a_listener.received.size() >= 11; } + )){ + error = "Client A did not receive the echo"; + }else if (std::memcmp(client_a_listener.received.data(), "still here?", 11) != 0){ + error = "The echo data was corrupted"; + }else{ + // Client B must have got a final answer to its + // connect attempt, whatever it was. + if (!client_b_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_b_listener]{ return client_b_listener.connect_finished.load() >= 1; } + )){ + error = "Client B did not finish its connect attempt"; + }else{ + ok = true; + } + } + } + } + + client_a.close(); + client_b.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } +}; + + +// Starting a second server on an already-bound port must fail through +// on_listen_finished and leave the server not running. +template +class Test_Sockets_PortInUse : public TestSockets{ +public: + using TestSockets::TestSockets; + + virtual UnitTestResult run(Logger& logger, CancellableScope& scope) const override{ + ServerSocketType server(this->m_thread_pool); + ServerListener server_listener; + server_listener.server = &server; + server.add_listener(server_listener); + + uint16_t port; + if (!this->start_server(server, server_listener, port)){ + return "Failed to start the server"; + } + + ServerSocketType second_server(this->m_thread_pool); + ServerListener second_listener; + second_server.add_listener(second_listener); + second_server.start("127.0.0.1", port); + if (!second_listener.waiter.wait_for( + std::chrono::seconds(5), + [&second_listener]{ return second_listener.listen_finished.load() > 0; } + )){ + server.close(); + return "The second server did not report its start"; + } + if (second_listener.last_listen_error.empty()){ + server.close(); + second_server.close(); + return "The second server succeeded on a port that is already in use"; + } + if (second_server.state() != AbstractServerSocket::State::NOT_RUNNING){ + server.close(); + second_server.close(); + return "The second server is not NOT_RUNNING after a failed start"; + } + + // The original server must still work. + bool ok = false; + { + ClientSocketType client(this->m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + + if (!this->connect_client(client, client_listener, port, 1)){ + ok = false; + }else if (client.send("ok", 2) != 2){ + ok = false; + }else if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.received.size() >= 2; } + )){ + ok = false; + }else{ + ok = true; + } + + client.close(); + } + + server.close(); + second_server.close(); + + if (!ok){ + return "The original server stopped working"; + } + return true; + } +}; + + +template +void add_socket_tests(UnitTestDatabase& database, ThreadPool& thread_pool, const char* backend_name){ + const std::string prefix = "Sockets (" + std::string(backend_name) + ") - "; + + database.add>( + thread_pool, prefix + "Client/server echo round trip" + ); + database.add>( + thread_pool, prefix + "Zero-length send" + ); + database.add>( + thread_pool, prefix + "Large transfer" + ); + database.add>( + thread_pool, prefix + "Rapid reconnect" + ); + database.add>( + thread_pool, prefix + "Send/close race" + ); + database.add>( + thread_pool, prefix + "Reject a second client" + ); + database.add>( + thread_pool, prefix + "Port in use" + ); +} + + + +} // anonymous namespace + + +void add_tests_Sockets(UnitTestDatabase& database, ThreadPool& thread_pool){ + add_socket_tests(database, thread_pool, "Qt"); +} + + + +} diff --git a/Common/Cpp/Sockets/Sockets_Tests.h b/Common/Cpp/Sockets/Sockets_Tests.h new file mode 100644 index 0000000000..0b651959a4 --- /dev/null +++ b/Common/Cpp/Sockets/Sockets_Tests.h @@ -0,0 +1,28 @@ +/* Sockets Tests + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#ifndef PokemonAutomation_Sockets_Tests_H +#define PokemonAutomation_Sockets_Tests_H + +namespace PokemonAutomation{ + +class ThreadPool; +class UnitTestDatabase; + + +// Loopback tests that combine a ClientSocket and a ServerSocket against +// each other on 127.0.0.1: echo round trip, zero-length sends, a large +// transfer, rapid reconnects, a send/close race, connection rejection, +// and a port-in-use start failure. Each scenario runs for every backend +// that compiles on the current platform (Qt: always, POSIX: non-Windows, +// WinSocket: Windows). `thread_pool` runs the sockets and the test +// workers; it must be a long-lived pool (e.g. +// `GlobalThreadPools::unlimited_normal()`). +void add_tests_Sockets(UnitTestDatabase& database, ThreadPool& thread_pool); + + +} +#endif diff --git a/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp b/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp index 1c19573466..419f0f36a3 100644 --- a/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp +++ b/SerialPrograms/Source/ComputerPrograms/ComputerPrograms_UnitTestRunner.cpp @@ -7,6 +7,7 @@ #include "Common/Cpp/ListenerSet.h" #include "Common/Cpp/ScopeExit.h" #include "Common/Cpp/PrettyPrint.h" +#include "Common/Cpp/Sockets/Sockets_Tests.h" #include "Common/Cpp/TestRunners/TestImageDistortionManager.h" #include "Common/Cpp/TestRunners/UnitTestDatabase.h" #include "Common/Qt/QtThreadPool_Tests.h" @@ -49,6 +50,7 @@ UnitTestDatabase make_UNIT_TESTS_ALL(){ UnitTestDatabase ret; add_tests_QtEventThreadPool(ret, GlobalThreadPools::computation_normal()); + add_tests_Sockets(ret, GlobalThreadPools::unlimited_normal()); add_tests_BlackBorderDetector(ret); add_tests_ListenerSet(ret); add_tests_AudioInfo(ret); diff --git a/SerialPrograms/cmake/SourceFiles.cmake b/SerialPrograms/cmake/SourceFiles.cmake index b0f27c7490..fb89038897 100644 --- a/SerialPrograms/cmake/SourceFiles.cmake +++ b/SerialPrograms/cmake/SourceFiles.cmake @@ -396,6 +396,7 @@ file(GLOB LIBRARY_SOURCES ../Common/Cpp/Rectangle.tpp ../Common/Cpp/SIMDDebuggers.h ../Common/Cpp/Sockets/AbstractClientSocket.h + ../Common/Cpp/Sockets/AbstractServerSocket.h ../Common/Cpp/Sockets/ClientSocket.cpp ../Common/Cpp/Sockets/ClientSocket.h ../Common/Cpp/Sockets/ClientSocket_POSIX.cpp @@ -404,6 +405,10 @@ file(GLOB LIBRARY_SOURCES ../Common/Cpp/Sockets/ClientSocket_Qt.h ../Common/Cpp/Sockets/ClientSocket_WinSocket.cpp ../Common/Cpp/Sockets/ClientSocket_WinSocket.h + ../Common/Cpp/Sockets/ServerSocket_Qt.cpp + ../Common/Cpp/Sockets/ServerSocket_Qt.h + ../Common/Cpp/Sockets/Sockets_Tests.cpp + ../Common/Cpp/Sockets/Sockets_Tests.h ../Common/Cpp/StreamConnections/MockDevice.cpp ../Common/Cpp/StreamConnections/MockDevice.h ../Common/Cpp/TestRunners/ParallelUnitTestRunner.cpp From 20140b76393e59ef4e3005dc7f54eb8650421d3b Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 04:53:20 +0000 Subject: [PATCH 02/15] ServerSocket_Qt.cpp: use CRLF line endings throughout --- Common/Cpp/Sockets/ServerSocket_Qt.cpp | 38 +++++++++++++------------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.cpp b/Common/Cpp/Sockets/ServerSocket_Qt.cpp index 679c3014fc..7546831b48 100644 --- a/Common/Cpp/Sockets/ServerSocket_Qt.cpp +++ b/Common/Cpp/Sockets/ServerSocket_Qt.cpp @@ -1,22 +1,22 @@ -/* Server Socket (Qt) - * - * From: https://github.com/PokemonAutomation/ - * - */ - -#include -#include -#include -#include -#include "Common/Qt/QtThreadPool.h" -#include "Common/Qt/GlobalThreadPoolsQt.h" -#include "ServerSocket_Qt.h" - -namespace PokemonAutomation{ - - - -void ServerSocket_Qt::start(const std::string& address, uint16_t port){ +/* Server Socket (Qt) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#include +#include +#include +#include +#include "Common/Qt/QtThreadPool.h" +#include "Common/Qt/GlobalThreadPoolsQt.h" +#include "ServerSocket_Qt.h" + +namespace PokemonAutomation{ + + + +void ServerSocket_Qt::start(const std::string& address, uint16_t port){ bool starting; { std::lock_guard lg(m_lock); From 5b8910edb26d0915a2c2aa09cca41016e32676ed Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 06:59:23 +0000 Subject: [PATCH 03/15] Sockets: use the Mutex/ConditionVariable wrappers per AGENTS.md AGENTS.md forbids using std::mutex/std::condition_variable directly; use the PokemonAutomation::Mutex/ConditionVariable wrappers from Common/Cpp/Concurrency instead. Migrate the Qt server's completion structs and the test harness's Waiter/rng lock. --- Common/Cpp/Sockets/ServerSocket_Qt.h | 5 ++--- Common/Cpp/Sockets/Sockets_Tests.cpp | 15 ++++++++------- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.h b/Common/Cpp/Sockets/ServerSocket_Qt.h index 0fd0e8a3b7..ce5010a242 100644 --- a/Common/Cpp/Sockets/ServerSocket_Qt.h +++ b/Common/Cpp/Sockets/ServerSocket_Qt.h @@ -9,7 +9,6 @@ #include #include -#include #include #include "Common/Cpp/Concurrency/Mutex.h" #include "Common/Cpp/Concurrency/ConditionVariable.h" @@ -34,14 +33,14 @@ class ServerSocket_Qt final : public QObject, public AbstractServerSocket{ bool done = false; Mutex lock; - std::condition_variable cv; + ConditionVariable cv; }; // Synchronizes "close_connection()" with the QTcpServer's event thread // when called from another thread. struct CloseCompletion{ Mutex lock; - std::condition_variable cv; + ConditionVariable cv; bool done = false; }; diff --git a/Common/Cpp/Sockets/Sockets_Tests.cpp b/Common/Cpp/Sockets/Sockets_Tests.cpp index 09cba432b1..773d8b0200 100644 --- a/Common/Cpp/Sockets/Sockets_Tests.cpp +++ b/Common/Cpp/Sockets/Sockets_Tests.cpp @@ -13,7 +13,6 @@ #include #include -#include #include #include #include @@ -23,6 +22,8 @@ #include #include #include "Common/Cpp/Concurrency/AsyncTask.h" +#include "Common/Cpp/Concurrency/ConditionVariable.h" +#include "Common/Cpp/Concurrency/Mutex.h" #include "Common/Cpp/Concurrency/ThreadPool.h" #include "Common/Cpp/Sockets/ClientSocket_Qt.h" #include "Common/Cpp/Sockets/ServerSocket_Qt.h" @@ -35,16 +36,16 @@ namespace{ struct Waiter{ - std::mutex lock; - std::condition_variable cv; + Mutex lock; + ConditionVariable cv; bool wait_for(std::chrono::milliseconds timeout, const std::function& pred){ - std::unique_lock lg(lock); + std::unique_lock lg(lock); return cv.wait_for(lg, timeout, pred); } void notify(){ - std::lock_guard lg(lock); + std::lock_guard lg(lock); cv.notify_all(); } }; @@ -139,7 +140,7 @@ class TestSockets : public UnitTest{ for (int attempt = 0; attempt < 8; attempt++){ uint16_t candidate; { - std::lock_guard lg(m_rng_lock); + std::lock_guard lg(m_rng_lock); candidate = (uint16_t)(40000 + m_rng() % 20000); } server.start("127.0.0.1", candidate); @@ -176,7 +177,7 @@ class TestSockets : public UnitTest{ ThreadPool& m_thread_pool; private: - mutable std::mutex m_rng_lock; + mutable Mutex m_rng_lock; mutable std::mt19937 m_rng{std::random_device{}()}; }; From eaa4d588769b2d95f7053579fedb7e88accce42f Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 06:59:23 +0000 Subject: [PATCH 04/15] ClientSocket_Qt: use the ConditionVariable wrapper (AGENTS.md) --- Common/Cpp/Sockets/ClientSocket_Qt.h | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Common/Cpp/Sockets/ClientSocket_Qt.h b/Common/Cpp/Sockets/ClientSocket_Qt.h index c8d235faef..c944ec2b6f 100644 --- a/Common/Cpp/Sockets/ClientSocket_Qt.h +++ b/Common/Cpp/Sockets/ClientSocket_Qt.h @@ -9,7 +9,6 @@ #include #include -#include #include #include "Common/Cpp/Concurrency/Mutex.h" #include "Common/Cpp/Concurrency/ConditionVariable.h" @@ -32,7 +31,7 @@ class ClientSocket_Qt final : public QObject, public AbstractClientSocket{ bool done = false; Mutex lock; - std::condition_variable cv; + ConditionVariable cv; }; From 2625ac9162e05910fedb37be2a8a4b753ff54130 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 07:04:27 +0000 Subject: [PATCH 05/15] ServerSocket_Qt.cpp: fix the indentation from the header/.cpp split The method bodies kept their in-class 8-space indentation from the monolithic header; un-dedent them to file scope (and drop the stray blank lines the split left behind). --- Common/Cpp/Sockets/ServerSocket_Qt.cpp | 611 ++++++++++++------------- 1 file changed, 304 insertions(+), 307 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.cpp b/Common/Cpp/Sockets/ServerSocket_Qt.cpp index 7546831b48..57a078958b 100644 --- a/Common/Cpp/Sockets/ServerSocket_Qt.cpp +++ b/Common/Cpp/Sockets/ServerSocket_Qt.cpp @@ -14,374 +14,371 @@ namespace PokemonAutomation{ - - void ServerSocket_Qt::start(const std::string& address, uint16_t port){ - bool starting; - { - std::lock_guard lg(m_lock); - starting = - (m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING) - && m_server == nullptr - && !m_starting; - m_starting = true; - } - if (!starting){ - return; - } + bool starting; + { + std::lock_guard lg(m_lock); + starting = + (m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING) + && m_server == nullptr + && !m_starting; + m_starting = true; + } + if (!starting){ + return; + } - m_listen_error = ""; - QTcpServer* server = nullptr; - try{ - server = static_cast( - GlobalThreadPools::qt_event_threadpool().add_object( - [this, address, port]{ - return make_server(address, port); - } - ) - ); - } - catch (...){ - { - std::lock_guard lg(m_lock); - m_starting = false; - m_cv.notify_all(); - } - m_listeners.run_method(&Listener::on_listen_finished, "Failed to start the listener."); - throw; - } - bool listening; + m_listen_error = ""; + QTcpServer* server = nullptr; + try{ + server = static_cast( + GlobalThreadPools::qt_event_threadpool().add_object( + [this, address, port]{ + return make_server(address, port); + } + ) + ); + } + catch (...){ { std::lock_guard lg(m_lock); m_starting = false; - m_server = server; - if (m_state.load(std::memory_order_relaxed) == State::DESTRUCTING){ - // "close()" is already in progress: it will destroy the - // server we just created. Leave the state for it. - }else{ - listening = m_listen_error.empty(); - m_state.store( - listening ? State::LISTENING : State::NOT_RUNNING, - std::memory_order_release - ); - } m_cv.notify_all(); } - - m_listeners.run_method(&Listener::on_listen_finished, m_listen_error); + m_listeners.run_method(&Listener::on_listen_finished, "Failed to start the listener."); + throw; + } + bool listening; + { + std::lock_guard lg(m_lock); + m_starting = false; + m_server = server; + if (m_state.load(std::memory_order_relaxed) == State::DESTRUCTING){ + // "close()" is already in progress: it will destroy the + // server we just created. Leave the state for it. + }else{ + listening = m_listen_error.empty(); + m_state.store( + listening ? State::LISTENING : State::NOT_RUNNING, + std::memory_order_release + ); + } + m_cv.notify_all(); } + m_listeners.run_method(&Listener::on_listen_finished, m_listen_error); +} + void ServerSocket_Qt::close_connection(){ - QThread* event_thread; - { - std::lock_guard lg(m_lock); - event_thread = m_event_thread; - if (m_server == nullptr){ - // Not running. - return; - } + QThread* event_thread; + { + std::lock_guard lg(m_lock); + event_thread = m_event_thread; + if (m_server == nullptr){ + // Not running. + return; } + } - CloseCompletion completion; - emit internal_close_connection(&completion); - - // If called from a thread other than the server's event thread, the - // call is queued: wait until it has run, so the state is up to date - // when this returns (same as the POSIX backend). If called from the - // event thread (e.g. a listener callback), the call ran directly and - // this does not block. - if (QThread::currentThread() != event_thread){ - std::unique_lock lg(completion.lock); - completion.cv.wait_for( - lg, - std::chrono::seconds(5), - [&completion]{ return completion.done; } - ); - } + CloseCompletion completion; + emit internal_close_connection(&completion); + + // If called from a thread other than the server's event thread, the + // call is queued: wait until it has run, so the state is up to date + // when this returns (same as the POSIX backend). If called from the + // event thread (e.g. a listener callback), the call ran directly and + // this does not block. + if (QThread::currentThread() != event_thread){ + std::unique_lock lg(completion.lock); + completion.cv.wait_for( + lg, + std::chrono::seconds(5), + [&completion]{ return completion.done; } + ); } +} void ServerSocket_Qt::close() noexcept{ - QTcpServer* server; - { - std::lock_guard lg(m_lock); - if (m_server == nullptr && !m_starting){ - // Never started, or already closed. - return; - } - m_close_requested = true; - m_state.store(State::DESTRUCTING, std::memory_order_release); - } - - // Wait for an in-flight "start()" to finish so we destroy the - // server it creates, and for in-flight "send()" calls to finish so - // we don't destroy the server out from under them. - { - std::unique_lock lg(m_lock); - m_cv.wait(lg, [this]{ - return !m_starting - && m_sends_in_flight.load(std::memory_order_acquire) == 0; - }); - server = m_server; - m_server = nullptr; - m_socket = nullptr; - } - if (server == nullptr){ - // "start()" failed or was never reached: nothing to destroy. - std::lock_guard lg(m_lock); - m_state.store(State::NOT_RUNNING, std::memory_order_release); - m_close_requested = false; + QTcpServer* server; + { + std::lock_guard lg(m_lock); + if (m_server == nullptr && !m_starting){ + // Never started, or already closed. return; } + m_close_requested = true; + m_state.store(State::DESTRUCTING, std::memory_order_release); + } - // Destroy the server on its event thread. This blocks until the - // destruction has happened. - GlobalThreadPools::qt_event_threadpool().remove_object(server); - + // Wait for an in-flight "start()" to finish so we destroy the + // server it creates, and for in-flight "send()" calls to finish so + // we don't destroy the server out from under them. + { + std::unique_lock lg(m_lock); + m_cv.wait(lg, [this]{ + return !m_starting + && m_sends_in_flight.load(std::memory_order_acquire) == 0; + }); + server = m_server; + m_server = nullptr; + m_socket = nullptr; + } + if (server == nullptr){ + // "start()" failed or was never reached: nothing to destroy. std::lock_guard lg(m_lock); m_state.store(State::NOT_RUNNING, std::memory_order_release); m_close_requested = false; + return; } -size_t ServerSocket_Qt::send(const void* data, size_t bytes){ - QTcpSocket* socket; - { - std::lock_guard lg(m_lock); - socket = - (m_state.load(std::memory_order_relaxed) == State::CONNECTED) ? m_socket : nullptr; - } - if (socket == nullptr){ - return 0; - } - - SendData send_data; - send_data.packet.assign((const char*)data, bytes); + // Destroy the server on its event thread. This blocks until the + // destruction has happened. + GlobalThreadPools::qt_event_threadpool().remove_object(server); - emit internal_send(&send_data); + std::lock_guard lg(m_lock); + m_state.store(State::NOT_RUNNING, std::memory_order_release); + m_close_requested = false; +} - // 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. - { - std::unique_lock lg(send_data.lock); - send_data.cv.wait_for( - lg, - std::chrono::seconds(5), - [&send_data]{ return send_data.done; } - ); - } +size_t ServerSocket_Qt::send(const void* data, size_t bytes){ + QTcpSocket* socket; + { + std::lock_guard lg(m_lock); + socket = + (m_state.load(std::memory_order_relaxed) == State::CONNECTED) ? m_socket : nullptr; + } + if (socket == nullptr){ + return 0; + } - return send_data.bytes_sent; + SendData send_data; + send_data.packet.assign((const char*)data, bytes); + + 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. + { + std::unique_lock lg(send_data.lock); + send_data.cv.wait_for( + lg, + std::chrono::seconds(5), + [&send_data]{ return send_data.done; } + ); } + return send_data.bytes_sent; +} + void ServerSocket_Qt::on_new_connection(){ - QTcpServer* server; + QTcpServer* server; + { + std::lock_guard lg(m_lock); + server = m_server; + } + if (server == nullptr){ + return; + } + + while (true){ + QTcpSocket* sock = server->nextPendingConnection(); + if (sock == nullptr){ + break; + } + // The connection socket lives with the server, so "close()" + // destroys it on the event thread. + sock->setParent(server); + + bool accepted; + bool rejected; { std::lock_guard lg(m_lock); - server = m_server; - } - if (server == nullptr){ - return; + State state = m_state.load(std::memory_order_relaxed); + if (state == State::CONNECTED){ + // A client is already connected: reject this one. + accepted = false; + rejected = true; + }else if (state != State::LISTENING){ + // The server is shutting down. + accepted = false; + rejected = false; + }else{ + m_socket = sock; + m_close_requested = false; + m_state.store(State::CONNECTED, std::memory_order_release); + accepted = true; + rejected = false; + } } - while (true){ - QTcpSocket* sock = server->nextPendingConnection(); - if (sock == nullptr){ - break; - } - // The connection socket lives with the server, so "close()" - // destroys it on the event thread. - sock->setParent(server); - - bool accepted; - bool rejected; - { - std::lock_guard lg(m_lock); - State state = m_state.load(std::memory_order_relaxed); - if (state == State::CONNECTED){ - // A client is already connected: reject this one. - accepted = false; - rejected = true; - }else if (state != State::LISTENING){ - // The server is shutting down. - accepted = false; - rejected = false; - }else{ - m_socket = sock; - m_close_requested = false; - m_state.store(State::CONNECTED, std::memory_order_release); - accepted = true; - rejected = false; - } + if (!accepted){ + if (rejected){ + m_listeners.run_method(&Listener::on_connect_rejected); } + sock->abort(); + continue; + } - if (!accepted){ - if (rejected){ - m_listeners.run_method(&Listener::on_connect_rejected); - } - sock->abort(); - continue; - } + QObject::connect( + sock, &QAbstractSocket::readyRead, + sock, [this, sock]{ + // readyRead only fires once per delivery, and more than + // one buffer's worth of data may be waiting, so read + // until the socket has no more data available. + while (true){ + qint64 available = sock->bytesAvailable(); + if (available <= 0){ + break; + } - QObject::connect( - sock, &QAbstractSocket::readyRead, - sock, [this, sock]{ - // readyRead only fires once per delivery, and more than - // one buffer's worth of data may be waiting, so read - // until the socket has no more data available. - while (true){ - qint64 available = sock->bytesAvailable(); - if (available <= 0){ - break; - } - - constexpr size_t BUFFER_SIZE = 4096; - char buffer[BUFFER_SIZE]; - qint64 bytes = sock->read(buffer, std::min(BUFFER_SIZE, available)); - if (bytes <= 0){ - break; - } - - m_listeners.run_method(&Listener::on_receive_data, buffer, bytes); + constexpr size_t BUFFER_SIZE = 4096; + char buffer[BUFFER_SIZE]; + qint64 bytes = sock->read(buffer, std::min(BUFFER_SIZE, available)); + if (bytes <= 0){ + break; } + + m_listeners.run_method(&Listener::on_receive_data, buffer, bytes); } - ); - QObject::connect( - sock, &QAbstractSocket::disconnected, - sock, [this, sock]{ - connection_lost(sock, "Connection closed by client."); - } - ); - QObject::connect( - sock, &QAbstractSocket::errorOccurred, - sock, [this, sock](QAbstractSocket::SocketError){ - connection_lost(sock, sock->errorString().toStdString()); - } - ); + } + ); + QObject::connect( + sock, &QAbstractSocket::disconnected, + sock, [this, sock]{ + connection_lost(sock, "Connection closed by client."); + } + ); + QObject::connect( + sock, &QAbstractSocket::errorOccurred, + sock, [this, sock](QAbstractSocket::SocketError){ + connection_lost(sock, sock->errorString().toStdString()); + } + ); - m_listeners.run_method(&Listener::on_connect); - } + m_listeners.run_method(&Listener::on_connect); } +} void ServerSocket_Qt::connection_lost(QTcpSocket* sock, const std::string& error_message){ - bool report; - { - std::lock_guard lg(m_lock); - if (m_socket != sock){ - // Already replaced or closed; the first callback wins. - return; - } - report = - !m_close_requested - && m_state.load(std::memory_order_relaxed) == State::CONNECTED; - m_socket = nullptr; - if (m_state.load(std::memory_order_relaxed) == State::CONNECTED){ - m_state.store(State::LISTENING, std::memory_order_release); - } + bool report; + { + std::lock_guard lg(m_lock); + if (m_socket != sock){ + // Already replaced or closed; the first callback wins. + return; } - - if (report){ - m_listeners.run_method(&Listener::on_disconnect, error_message); + report = + !m_close_requested + && m_state.load(std::memory_order_relaxed) == State::CONNECTED; + m_socket = nullptr; + if (m_state.load(std::memory_order_relaxed) == State::CONNECTED){ + m_state.store(State::LISTENING, std::memory_order_release); } } + if (report){ + m_listeners.run_method(&Listener::on_disconnect, error_message); + } +} + void ServerSocket_Qt::on_close_connection(CloseCompletion* completion){ - QTcpSocket* sock; - { - std::lock_guard lg(m_lock); - if (m_socket != nullptr){ - m_close_requested = true; - m_state.store(State::LISTENING, std::memory_order_release); - sock = m_socket; - m_socket = nullptr; - }else{ - sock = nullptr; - } - } - if (sock != nullptr){ - sock->abort(); + QTcpSocket* sock; + { + std::lock_guard lg(m_lock); + if (m_socket != nullptr){ + m_close_requested = true; + m_state.store(State::LISTENING, std::memory_order_release); + sock = m_socket; + m_socket = nullptr; + }else{ + sock = nullptr; } + } + if (sock != nullptr){ + sock->abort(); + } - { - std::lock_guard lg(completion->lock); - completion->done = true; - } - completion->cv.notify_all(); + { + std::lock_guard lg(completion->lock); + completion->done = true; } + completion->cv.notify_all(); +} void ServerSocket_Qt::on_send(SendData* send_data){ - QTcpSocket* sock; - { - std::lock_guard lg(m_lock); - m_sends_in_flight.fetch_add(1, std::memory_order_acq_rel); - m_cv.notify_all(); - sock = m_socket; - } + QTcpSocket* sock; + { + std::lock_guard lg(m_lock); + m_sends_in_flight.fetch_add(1, std::memory_order_acq_rel); + m_cv.notify_all(); + sock = m_socket; + } - size_t bytes_sent = 0; - if (sock != nullptr){ - const char* ptr = send_data->packet.data(); - size_t remaining = send_data->packet.size(); - while (remaining > 0 && sock->state() == QAbstractSocket::ConnectedState){ - qint64 current_sent = sock->write(ptr, remaining); - if (current_sent <= 0){ - // The kernel's send buffer is full. Wait for some of it - // to drain, then try again. - sock->waitForBytesWritten(10); - continue; - } - ptr += current_sent; - remaining -= (size_t)current_sent; + size_t bytes_sent = 0; + if (sock != nullptr){ + const char* ptr = send_data->packet.data(); + size_t remaining = send_data->packet.size(); + while (remaining > 0 && sock->state() == QAbstractSocket::ConnectedState){ + qint64 current_sent = sock->write(ptr, remaining); + if (current_sent <= 0){ + // The kernel's send buffer is full. Wait for some of it + // to drain, then try again. + sock->waitForBytesWritten(10); + continue; } - bytes_sent = send_data->packet.size() - remaining; + ptr += current_sent; + remaining -= (size_t)current_sent; } + 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(); - } + { + std::lock_guard lg(m_lock); + m_sends_in_flight.fetch_sub(1, std::memory_order_acq_rel); + m_cv.notify_all(); + } - { - std::lock_guard lg(send_data->lock); - send_data->bytes_sent = bytes_sent; - send_data->done = true; - } - send_data->cv.notify_all(); + { + std::lock_guard lg(send_data->lock); + send_data->bytes_sent = bytes_sent; + send_data->done = true; } + send_data->cv.notify_all(); +} std::unique_ptr ServerSocket_Qt::make_server(const std::string& address, uint16_t port){ - auto server = std::make_unique(); - if (!server->listen(QHostAddress(QString::fromStdString(address)), port)){ - std::lock_guard lg(m_lock); - m_listen_error = server->errorString().toStdString(); - } - - { - std::lock_guard lg(m_lock); - m_event_thread = QThread::currentThread(); - } - - QObject::connect( - this, &ServerSocket_Qt::internal_send, - server.get(), [this](SendData* send_data){ - on_send(send_data); - } - ); - QObject::connect( - this, &ServerSocket_Qt::internal_close_connection, - server.get(), [this](CloseCompletion* completion){ - on_close_connection(completion); - } - ); - QObject::connect( - server.get(), &QTcpServer::newConnection, - server.get(), [this]{ - on_new_connection(); - } - ); + auto server = std::make_unique(); + if (!server->listen(QHostAddress(QString::fromStdString(address)), port)){ + std::lock_guard lg(m_lock); + m_listen_error = server->errorString().toStdString(); + } - return server; + { + std::lock_guard lg(m_lock); + m_event_thread = QThread::currentThread(); } + QObject::connect( + this, &ServerSocket_Qt::internal_send, + server.get(), [this](SendData* send_data){ + on_send(send_data); + } + ); + QObject::connect( + this, &ServerSocket_Qt::internal_close_connection, + server.get(), [this](CloseCompletion* completion){ + on_close_connection(completion); + } + ); + QObject::connect( + server.get(), &QTcpServer::newConnection, + server.get(), [this]{ + on_new_connection(); + } + ); + + return server; +} } From 6f1eadf6d1b01146ce58bb003ca22ef4beef31d7 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 08:59:47 +0000 Subject: [PATCH 06/15] ServerSocket_Qt: give signal completions shared ownership The 5 s completion waits in send()/close_connection() were only safe while the queued call ran within the window: the completion objects were stack-local, so if the wait expired and the event thread delivered the call later, on_send/on_close_connection wrote through a dead stack pointer (stack-use-after-return; reproduced under ASAN by stalling the event thread in on_receive_data for 6 s). The signals now pass QSharedPointer/QSharedPointer by value: the queued call's own copy keeps the payload alive for as long as the call is pending or running (no UAF after a timed-out caller), and if close() destroys the QTcpServer first, Qt discards the queued call and the caller's copy is the last reference (no leak). The 5 s bounds remain as plain safety valves. Also add the regression test 'Sockets (Qt) - Completion lifetime under event-thread stall', which stalls the event thread inside on_receive_data, verifies close_connection() times out at ~5 s instead of hanging, and that the late handler still restores LISTENING. Takes ~12 s to run. --- Common/Cpp/Sockets/ServerSocket_Qt.cpp | 38 ++++---- Common/Cpp/Sockets/ServerSocket_Qt.h | 18 ++-- Common/Cpp/Sockets/Sockets_Tests.cpp | 119 +++++++++++++++++++++++++ 3 files changed, 152 insertions(+), 23 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.cpp b/Common/Cpp/Sockets/ServerSocket_Qt.cpp index 57a078958b..9bae4f48da 100644 --- a/Common/Cpp/Sockets/ServerSocket_Qt.cpp +++ b/Common/Cpp/Sockets/ServerSocket_Qt.cpp @@ -80,20 +80,23 @@ void ServerSocket_Qt::close_connection(){ } } - CloseCompletion completion; - emit internal_close_connection(&completion); + QSharedPointer completion = QSharedPointer::create(); + emit internal_close_connection(completion); // If called from a thread other than the server's event thread, the // call is queued: wait until it has run, so the state is up to date // when this returns (same as the POSIX backend). If called from the // event thread (e.g. a listener callback), the call ran directly and // this does not block. + // + // The shared ownership makes the timeout safe: if it expires, the + // queued call's own copy keeps the completion alive until it runs. if (QThread::currentThread() != event_thread){ - std::unique_lock lg(completion.lock); - completion.cv.wait_for( + std::unique_lock lg(completion->lock); + completion->cv.wait_for( lg, std::chrono::seconds(5), - [&completion]{ return completion.done; } + [&completion]{ return completion->done; } ); } } @@ -150,25 +153,26 @@ size_t ServerSocket_Qt::send(const void* data, size_t bytes){ return 0; } - SendData send_data; - send_data.packet.assign((const char*)data, bytes); + QSharedPointer send_data = QSharedPointer::create(); + send_data->packet.assign((const char*)data, bytes); - emit internal_send(&send_data); + 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. + // complete on its own; the shared ownership keeps the packet alive + // until the queued call runs (or is discarded with the server). { - std::unique_lock lg(send_data.lock); - send_data.cv.wait_for( + std::unique_lock lg(send_data->lock); + send_data->cv.wait_for( lg, std::chrono::seconds(5), - [&send_data]{ return send_data.done; } + [&send_data]{ return send_data->done; } ); } - return send_data.bytes_sent; + return send_data->bytes_sent; } void ServerSocket_Qt::on_new_connection(){ @@ -282,7 +286,7 @@ void ServerSocket_Qt::connection_lost(QTcpSocket* sock, const std::string& error } } -void ServerSocket_Qt::on_close_connection(CloseCompletion* completion){ +void ServerSocket_Qt::on_close_connection(QSharedPointer completion){ QTcpSocket* sock; { std::lock_guard lg(m_lock); @@ -306,7 +310,7 @@ void ServerSocket_Qt::on_close_connection(CloseCompletion* completion){ completion->cv.notify_all(); } -void ServerSocket_Qt::on_send(SendData* send_data){ +void ServerSocket_Qt::on_send(QSharedPointer send_data){ QTcpSocket* sock; { std::lock_guard lg(m_lock); @@ -361,13 +365,13 @@ std::unique_ptr ServerSocket_Qt::make_server(const std::string& addr QObject::connect( this, &ServerSocket_Qt::internal_send, - server.get(), [this](SendData* send_data){ + server.get(), [this](QSharedPointer send_data){ on_send(send_data); } ); QObject::connect( this, &ServerSocket_Qt::internal_close_connection, - server.get(), [this](CloseCompletion* completion){ + server.get(), [this](QSharedPointer completion){ on_close_connection(completion); } ); diff --git a/Common/Cpp/Sockets/ServerSocket_Qt.h b/Common/Cpp/Sockets/ServerSocket_Qt.h index ce5010a242..34fb1660dc 100644 --- a/Common/Cpp/Sockets/ServerSocket_Qt.h +++ b/Common/Cpp/Sockets/ServerSocket_Qt.h @@ -10,6 +10,7 @@ #include #include #include +#include #include "Common/Cpp/Concurrency/Mutex.h" #include "Common/Cpp/Concurrency/ConditionVariable.h" #include "AbstractServerSocket.h" @@ -27,6 +28,11 @@ class ThreadPool; class ServerSocket_Qt final : public QObject, public AbstractServerSocket{ Q_OBJECT + // Carries shared ownership (the signals pass it by value) so that a call still + // queued when the caller times out, or when "close()" destroys the server and + // Qt discards the pending call, keeps the payload alive exactly as long as the + // event itself: the handler's copy outlives a timed-out caller, and a + // discarded event leaves the caller's copy as the last reference. struct SendData{ std::string packet; size_t bytes_sent = 0; @@ -36,8 +42,8 @@ class ServerSocket_Qt final : public QObject, public AbstractServerSocket{ ConditionVariable cv; }; - // Synchronizes "close_connection()" with the QTcpServer's event thread - // when called from another thread. + // Synchronizes "close_connection()" with the QTcpServer's event thread when + // called from another thread. Lifetime as SendData. struct CloseCompletion{ Mutex lock; ConditionVariable cv; @@ -76,8 +82,8 @@ class ServerSocket_Qt final : public QObject, public AbstractServerSocket{ signals: - void internal_send(SendData* send_data); - void internal_close_connection(CloseCompletion* completion); + void internal_send(QSharedPointer send_data); + void internal_close_connection(QSharedPointer completion); private: @@ -92,12 +98,12 @@ class ServerSocket_Qt final : public QObject, public AbstractServerSocket{ void connection_lost(QTcpSocket* sock, const std::string& error_message); // Runs on the QTcpServer's event thread. - void on_close_connection(CloseCompletion* completion); + void on_close_connection(QSharedPointer completion); // Runs on the QTcpServer's event thread. The context object is the // server, so the call still runs if the connection socket has already // been aborted, and it just reports 0 bytes in that case. - void on_send(SendData* send_data); + void on_send(QSharedPointer send_data); std::unique_ptr make_server(const std::string& address, uint16_t port); diff --git a/Common/Cpp/Sockets/Sockets_Tests.cpp b/Common/Cpp/Sockets/Sockets_Tests.cpp index 773d8b0200..5a4e37764e 100644 --- a/Common/Cpp/Sockets/Sockets_Tests.cpp +++ b/Common/Cpp/Sockets/Sockets_Tests.cpp @@ -723,6 +723,118 @@ class Test_Sockets_PortInUse : public TestSockets 0; } + ) && server_listener.last_listen_error.empty()){ + port = candidate; + break; + } + } + } + if (port == 0){ + return "Failed to start the server"; + } + + bool ok = false; + std::string error; + { + ClientSocket_Qt client(m_thread_pool); + ClientListener client_listener; + client.add_listener(client_listener); + client.connect("127.0.0.1", port); + if (!client_listener.waiter.wait_for( + std::chrono::seconds(5), + [&client_listener]{ return client_listener.connect_finished.load() >= 1; } + )){ + error = "The client did not finish connecting"; + }else if (!server_listener.waiter.wait_for( + std::chrono::seconds(5), + [&server_listener]{ return server_listener.connected.load() >= 1; } + )){ + error = "The server did not call on_connect"; + }else{ + // One byte in: the event thread blocks 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{ + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + + auto start = std::chrono::steady_clock::now(); + server.close_connection(); + auto waited = std::chrono::steady_clock::now() - start; + if (waited < std::chrono::seconds(4) || waited >= std::chrono::seconds(8)){ + error = "close_connection() did not time out as expected (" + + std::to_string( + (long long)std::chrono::duration_cast(waited).count() + ) + " ms)"; + }else{ + // Wait for the stall to finish so the late handler actually + // runs, then check that it restored the state. + std::this_thread::sleep_for(std::chrono::seconds(6)); + if (server.state() != AbstractServerSocket::State::LISTENING){ + error = "The late handler did not restore LISTENING"; + }else{ + ok = true; + } + } + } + } + + client.close(); + } + + server.close(); + + if (!ok && !error.empty()){ + return error; + } + return true; + } + +private: + ThreadPool& m_thread_pool; +}; + + template void add_socket_tests(UnitTestDatabase& database, ThreadPool& thread_pool, const char* backend_name){ const std::string prefix = "Sockets (" + std::string(backend_name) + ") - "; @@ -757,6 +869,13 @@ void add_socket_tests(UnitTestDatabase& database, ThreadPool& thread_pool, const void add_tests_Sockets(UnitTestDatabase& database, ThreadPool& thread_pool){ add_socket_tests(database, thread_pool, "Qt"); + + // Qt-specific: the completion-lifetime guarantee is only at risk in the + // Qt backend (a queued call can outlive its caller there). + database.add( + thread_pool, + "Sockets (Qt) - Completion lifetime under event-thread stall" + ); } From c4cdc792f14611b1327a920bd6c9fd0c358ee0f9 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 03:04:27 +0000 Subject: [PATCH 07/15] Sockets: add the POSIX ServerSocket Implements ServerSocket_POSIX (an accept-loop thread plus one connection thread at a time, both from the given thread pool via plain dispatch(), with non-blocking sockets; the implementation lives in ServerSocket_POSIX.cpp, following the split of the client backends) and enables the POSIX instantiation of the loopback tests on non-Windows. close()/close_connection() only flip the state: the connection thread notices within ~1 ms and closes the socket itself, so close_connection() is safe to call from within a listener callback. The accept loop keeps running while a connection is active and rejects new clients via on_connect_rejected. The connection task is started and published without holding the main lock: assigning to the task slot destroys the previous connection task's AsyncTask, whose destructor waits for that task to finish - and the task may be blocked on the main lock. --- Common/Cpp/Sockets/ServerSocket_POSIX.cpp | 447 ++++++++++++++++++++++ Common/Cpp/Sockets/ServerSocket_POSIX.h | 80 ++++ Common/Cpp/Sockets/Sockets_Tests.cpp | 14 +- SerialPrograms/cmake/SourceFiles.cmake | 2 + 4 files changed, 540 insertions(+), 3 deletions(-) create mode 100644 Common/Cpp/Sockets/ServerSocket_POSIX.cpp create mode 100644 Common/Cpp/Sockets/ServerSocket_POSIX.h diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.cpp b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp new file mode 100644 index 0000000000..e66877f2cf --- /dev/null +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp @@ -0,0 +1,447 @@ +/* Server Socket (POSIX) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#if !_WIN32 + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "Common/Cpp/Concurrency/ThreadPool.h" +#include "ServerSocket_POSIX.h" + +namespace PokemonAutomation{ + + + +ServerSocket_POSIX::ServerSocket_POSIX(ThreadPool& thread_pool) + : m_thread_pool(thread_pool) + { + // Ignore SIGPIPE. Handle errors via errno instead + signal(SIGPIPE, SIG_IGN); + } + +void ServerSocket_POSIX::start(const std::string& address, uint16_t port){ + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING + || m_starting){ + return; + } + m_starting = true; + } + + // Use the plain dispatch() rather than dispatch_now_blocking(): + // start() may be called from a thread that is itself running on + // this pool, and blocking here until a pool thread is free could + // deadlock a capped pool. The returned AsyncTask still lets + // close() join the listener thread. + try{ + AsyncTask task = m_thread_pool.dispatch([=, this]{ + accept_thread_loop(address, port); + }); + { + // m_task_lock, not m_lock: close() only reads the slot + // after it has waited for m_starting to clear, so this + // assignment is the only writer besides close(). + std::lock_guard lg(m_task_lock); + m_accept_thread = std::move(task); + } + }catch (...){ + { + std::lock_guard lg(m_lock); + m_starting = false; + m_cv.notify_all(); + } + m_listeners.run_method(&Listener::on_listen_finished, "Failed to start the listener thread."); + throw; + } + + { + std::lock_guard lg(m_lock); + m_starting = false; + m_cv.notify_all(); + } + } + +void ServerSocket_POSIX::close_connection(){ + int socket; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + return; + } + m_close_requested = true; + m_state.store(State::LISTENING, std::memory_order_release); + socket = m_client; + } + if (socket != -1){ + // Tells the connection thread to stop; it closes the socket + // itself. This is safe to call from within a listener callback. + ::shutdown(socket, SHUT_RDWR); + } + } + +void ServerSocket_POSIX::close() noexcept{ + { + std::unique_lock lg(m_lock); + // Wait for an in-progress start() to publish the listener task + // before we take the task slots. + m_cv.wait(lg, [&]{ return !m_starting; }); + if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING){ + m_close_requested = true; + m_state.store(State::DESTRUCTING, std::memory_order_release); + m_closing++; + } + } + + bool drain = false; + { + std::lock_guard lg(m_lock); + drain = (m_closing == 1); + } + if (!drain){ + // Either nothing was running (m_closing == 0), or another + // close() is already draining the threads (m_closing > 1). + // In the second case, wait for it to finish. + std::unique_lock lg(m_lock); + m_cv.wait(lg, [&]{ + return m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING; + }); + return; + } + + AsyncTask accept_thread; + AsyncTask client_thread; + { + // m_task_lock, not m_lock: moving the tasks out destroys the + // old AsyncTask objects, and their destructors wait for the + // tasks to finish - which they can only do if they can acquire + // m_lock. + std::lock_guard lg(m_task_lock); + accept_thread = std::move(m_accept_thread); + m_accept_thread = AsyncTask(); + client_thread = std::move(m_client_thread); + m_client_thread = AsyncTask(); + } + + // The threads notice the new state within a few milliseconds and + // close their own sockets. + accept_thread.wait_and_ignore_exceptions(); + client_thread.wait_and_ignore_exceptions(); + + { + std::lock_guard lg(m_lock); + m_state.store(State::NOT_RUNNING, std::memory_order_release); + m_close_requested = false; + m_closing = 0; + m_cv.notify_all(); + } + } + +size_t ServerSocket_POSIX::send(const void* data, size_t bytes){ + int socket; + { + std::lock_guard lg(m_lock); + socket = + (m_state.load(std::memory_order_relaxed) == State::CONNECTED) ? m_client : -1; + } + if (socket == -1){ + return 0; + } + + const char* ptr = (const char*)data; + size_t sent = 0; + + while (bytes > 0){ + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + return sent; + } + } + + constexpr size_t BLOCK_SIZE = 1 << 30; + size_t current = std::min(BLOCK_SIZE, bytes); + ssize_t current_sent = ::send(socket, ptr, current, MSG_DONTWAIT | MSG_NOSIGNAL); + if (current_sent != -1){ + sent += (size_t)current_sent; + ptr += current_sent; + bytes -= (size_t)current_sent; + continue; + } + + int error = errno; + switch (error){ + case EAGAIN: +#if EAGAIN != EWOULDBLOCK + case EWOULDBLOCK: +#endif + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + continue; + default: + // EPIPE, ECONNRESET, or any other error: the connection is + // gone. + return sent; + } + } + + return sent; + } + +void ServerSocket_POSIX::accept_thread_loop(std::string address, uint16_t port){ + int listen_socket = ::socket(AF_INET, SOCK_STREAM, 0); + if (listen_socket == -1){ + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to create the socket: " + std::string(strerror(errno)) + ); + return; + } + + int yes = 1; + setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)); + + sockaddr_in server; + std::memset(&server, 0, sizeof(server)); + server.sin_family = AF_INET; + server.sin_port = htons(port); + server.sin_addr.s_addr = inet_addr(address.c_str()); + + if (bind(listen_socket, (struct sockaddr*)&server, sizeof(server)) == -1){ + int error = errno; + ::close(listen_socket); + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to bind the socket: " + std::string(strerror(error)) + ); + return; + } + + if (::listen(listen_socket, 16) == -1){ + int error = errno; + ::close(listen_socket); + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to listen on the socket: " + std::string(strerror(error)) + ); + return; + } + + std::string start_error; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) == State::DESTRUCTING){ + // close() won the race. + start_error = "The server was closed before it could start listening."; + }else{ + m_listen = listen_socket; + listen_socket = -1; + m_state.store(State::LISTENING, std::memory_order_release); + } + } + if (listen_socket != -1){ + ::close(listen_socket); + } + + m_listeners.run_method(&Listener::on_listen_finished, start_error); + if (!start_error.empty()){ + return; + } + + // Accept loop. Runs for the whole lifetime of the server: it keeps + // polling while a connection is active, and rejects new clients in + // the meantime. + while (true){ + int listen; + { + std::lock_guard lg(m_lock); + State state = m_state.load(std::memory_order_relaxed); + if (state == State::DESTRUCTING || state == State::NOT_RUNNING){ + break; + } + listen = m_listen; + } + + pollfd pf; + pf.fd = listen; + pf.events = POLLIN; + pf.revents = 0; + int ret = ::poll(&pf, 1, 10); + if (ret == -1){ + if (errno == EINTR){ + continue; + } + break; + } + if (ret == 0){ + continue; + } + + int client_socket = ::accept(listen, nullptr, nullptr); + if (client_socket == -1){ + continue; + } + + bool rejected = false; + bool accepted = false; + { + std::lock_guard lg(m_lock); + State state = m_state.load(std::memory_order_relaxed); + if (state == State::DESTRUCTING || state == State::NOT_RUNNING){ + ::close(client_socket); + break; + } + if (state == State::CONNECTED){ + // A client is already connected: reject this one. + ::close(client_socket); + rejected = true; + }else{ + fcntl(client_socket, F_SETFL, O_NONBLOCK); + m_client = client_socket; + m_close_requested = false; + m_state.store(State::CONNECTED, std::memory_order_release); + accepted = true; + } + } + + if (accepted){ + // Start the connection thread without holding m_lock. + // The assignment to m_client_thread destroys the previous + // connection task's AsyncTask, whose destructor waits for + // that task to finish - and that task may be blocked on + // m_lock. + try{ + AsyncTask task = m_thread_pool.dispatch([this]{ + connection_thread_loop(); + }); + std::lock_guard lg(m_task_lock); + m_client_thread = std::move(task); + }catch (...){ + // The connection thread could not be started. Drop + // this connection and keep listening. + std::lock_guard lg(m_lock); + ::close(client_socket); + m_client = -1; + m_state.store(State::LISTENING, std::memory_order_release); + } + } + + if (rejected){ + m_listeners.run_method(&Listener::on_connect_rejected); + } + } + + int listen; + { + std::lock_guard lg(m_lock); + listen = m_listen; + m_listen = -1; + } + if (listen != -1){ + ::close(listen); + } + } + +void ServerSocket_POSIX::connection_thread_loop(){ + m_listeners.run_method(&Listener::on_thread_start); + + int socket; + bool active; + { + std::lock_guard lg(m_lock); + socket = m_client; + active = (m_state.load(std::memory_order_relaxed) == State::CONNECTED); + if (!active && m_client == socket){ + m_client = -1; + } + } + if (!active){ + // The connection was closed before this thread started. + ::close(socket); + return; + } + + m_listeners.run_method(&Listener::on_connect); + + constexpr size_t BUFFER_SIZE = 4096; + char buffer[BUFFER_SIZE]; + std::string error; + + while (true){ + ssize_t bytes; + int error_no = 0; + bool no_data; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + break; + } + bytes = ::recv(socket, buffer, BUFFER_SIZE, 0); + no_data = (bytes == -1); + if (no_data){ + error_no = errno; + no_data = (error_no == EAGAIN); +#if EAGAIN != EWOULDBLOCK + no_data = no_data || (error_no == EWOULDBLOCK); +#endif + } + } + + if (bytes > 0){ + m_listeners.run_method(&Listener::on_receive_data, buffer, bytes); + continue; + } + + if (no_data){ + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + continue; + } + + if (bytes == 0){ + error = "Connection closed by client."; + }else{ + switch (error_no){ + case EINTR: + continue; + default: + error = "Receive error (errno " + std::to_string(error_no) + "): " + std::string(strerror(error_no)); + break; + } + } + break; + } + + bool report_disconnect; + { + std::lock_guard lg(m_lock); + report_disconnect = false; + if (m_client == socket){ + // Still the active connection (or it was closed before a + // new one took over): we own the socket and the state. + m_client = -1; + if (m_state.load(std::memory_order_relaxed) == State::CONNECTED){ + m_state.store(State::LISTENING, std::memory_order_release); + report_disconnect = !m_close_requested; + } + } + } + ::close(socket); + if (report_disconnect){ + m_listeners.run_method(&Listener::on_disconnect, error); + } + } + +} + +#endif diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.h b/Common/Cpp/Sockets/ServerSocket_POSIX.h new file mode 100644 index 0000000000..7626cb64de --- /dev/null +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.h @@ -0,0 +1,80 @@ +/* Server Socket (POSIX) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#ifndef PokemonAutomation_ServerSocket_POSIX_H +#define PokemonAutomation_ServerSocket_POSIX_H + +#include +#include +#include "Common/Cpp/Concurrency/AsyncTask.h" +#include "Common/Cpp/Concurrency/ConditionVariable.h" +#include "Common/Cpp/Concurrency/Mutex.h" +#include "AbstractServerSocket.h" + + +namespace PokemonAutomation{ + +class ThreadPool; + + +class ServerSocket_POSIX final : public AbstractServerSocket{ +public: + + ServerSocket_POSIX(ThreadPool& thread_pool); + + + virtual ~ServerSocket_POSIX(){ + close(); + } + + + + virtual void start(const std::string& address, uint16_t port) override; + + + virtual void close_connection() override; + + + + virtual void close() noexcept override; + + + virtual size_t send(const void* data, size_t bytes) override; + + +private: + + + void accept_thread_loop(std::string address, uint16_t port); + + + void connection_thread_loop(); + +private: + ThreadPool& m_thread_pool; + Mutex m_lock; + ConditionVariable m_cv; + bool m_starting = false; + int m_closing = 0; + + // Protects m_accept_thread and m_client_thread. Kept separate from + // m_lock: assigning to a task slot destroys the previous task's + // AsyncTask, and its destructor waits for that task to finish - which + // requires m_lock, so the two locks must never be held at once. + Mutex m_task_lock; + + int m_listen = -1; + int m_client = -1; + bool m_close_requested = false; + + AsyncTask m_accept_thread; + AsyncTask m_client_thread; +}; + + + +} +#endif diff --git a/Common/Cpp/Sockets/Sockets_Tests.cpp b/Common/Cpp/Sockets/Sockets_Tests.cpp index 5a4e37764e..57a8398628 100644 --- a/Common/Cpp/Sockets/Sockets_Tests.cpp +++ b/Common/Cpp/Sockets/Sockets_Tests.cpp @@ -5,9 +5,9 @@ * Loopback tests that combine a ClientSocket and a ServerSocket against * each other on 127.0.0.1. The test logic is written against the abstract * classes, and is instantiated for every backend that compiles on the - * current platform. Qt is instantiated here; the POSIX and WinSocket - * backends add their own instantiations in separate PRs, so that every - * platform runs 2 of the 3 backends and CI as a whole covers all 3. + * current platform (Qt: always, POSIX: on non-Windows, WinSocket: on + * Windows; each backend lands in its own PR), so that every platform + * runs 2 of the 3 backends and CI as a whole covers all 3. * */ @@ -27,6 +27,10 @@ #include "Common/Cpp/Concurrency/ThreadPool.h" #include "Common/Cpp/Sockets/ClientSocket_Qt.h" #include "Common/Cpp/Sockets/ServerSocket_Qt.h" +#ifndef _WIN32 +#include "Common/Cpp/Sockets/ClientSocket_POSIX.h" +#include "Common/Cpp/Sockets/ServerSocket_POSIX.h" +#endif #include "Common/Cpp/TestRunners/UnitTestDatabase.h" #include "Sockets_Tests.h" @@ -876,6 +880,10 @@ void add_tests_Sockets(UnitTestDatabase& database, ThreadPool& thread_pool){ thread_pool, "Sockets (Qt) - Completion lifetime under event-thread stall" ); + +#ifndef _WIN32 + add_socket_tests(database, thread_pool, "POSIX"); +#endif } diff --git a/SerialPrograms/cmake/SourceFiles.cmake b/SerialPrograms/cmake/SourceFiles.cmake index fb89038897..b4725b5616 100644 --- a/SerialPrograms/cmake/SourceFiles.cmake +++ b/SerialPrograms/cmake/SourceFiles.cmake @@ -405,7 +405,9 @@ file(GLOB LIBRARY_SOURCES ../Common/Cpp/Sockets/ClientSocket_Qt.h ../Common/Cpp/Sockets/ClientSocket_WinSocket.cpp ../Common/Cpp/Sockets/ClientSocket_WinSocket.h + ../Common/Cpp/Sockets/ServerSocket_POSIX.cpp ../Common/Cpp/Sockets/ServerSocket_Qt.cpp + ../Common/Cpp/Sockets/ServerSocket_POSIX.h ../Common/Cpp/Sockets/ServerSocket_Qt.h ../Common/Cpp/Sockets/Sockets_Tests.cpp ../Common/Cpp/Sockets/Sockets_Tests.h From 8669c05da61e3fcd5c414fb20712b6a385b48624 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 04:59:27 +0000 Subject: [PATCH 08/15] ServerSocket_POSIX.cpp: use CRLF line endings throughout --- Common/Cpp/Sockets/ServerSocket_POSIX.cpp | 52 +++++++++++------------ 1 file changed, 26 insertions(+), 26 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.cpp b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp index e66877f2cf..f3c53c2495 100644 --- a/Common/Cpp/Sockets/ServerSocket_POSIX.cpp +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp @@ -1,29 +1,29 @@ -/* Server Socket (POSIX) - * - * From: https://github.com/PokemonAutomation/ - * - */ - -#if !_WIN32 - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include "Common/Cpp/Concurrency/ThreadPool.h" -#include "ServerSocket_POSIX.h" - -namespace PokemonAutomation{ - - - +/* Server Socket (POSIX) + * + * From: https://github.com/PokemonAutomation/ + * + */ + +#if !_WIN32 + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "Common/Cpp/Concurrency/ThreadPool.h" +#include "ServerSocket_POSIX.h" + +namespace PokemonAutomation{ + + + ServerSocket_POSIX::ServerSocket_POSIX(ThreadPool& thread_pool) : m_thread_pool(thread_pool) { From 4eafe7cfeeb00a90403f7f466434561891236877 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 07:16:44 +0000 Subject: [PATCH 09/15] POSIX client: finish partial sends instead of returning early send() with MSG_DONTWAIT can return after writing only part of the buffer when the kernel's send buffer is full. macOS does this much more readily than Linux, so the 1 MiB loopback transfer test failed on the Mac CI: client.send() returned early with fewer bytes. Loop until the whole buffer has been written, matching the Qt backend's send() behavior. --- Common/Cpp/Sockets/ClientSocket_POSIX.cpp | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/Common/Cpp/Sockets/ClientSocket_POSIX.cpp b/Common/Cpp/Sockets/ClientSocket_POSIX.cpp index 8e46b363a4..caf9d17773 100644 --- a/Common/Cpp/Sockets/ClientSocket_POSIX.cpp +++ b/Common/Cpp/Sockets/ClientSocket_POSIX.cpp @@ -133,10 +133,11 @@ size_t ClientSocket_POSIX::send(const void* data, size_t bytes){ size_t current = std::min(bytes, BLOCK_SIZE); ssize_t current_sent = ::send(socket, ptr, (int)current, MSG_DONTWAIT | MSG_NOSIGNAL); if (current_sent != -1){ + // send() may return after writing only part of the + // buffer (macOS does this when the kernel's send + // buffer is full); loop until everything has been + // written, the same way the Qt backend does. sent += current_sent; - if ((size_t)current_sent < current){ - return sent; - } ptr += current_sent; bytes -= current_sent; continue; From ca15f0a20b6b057465d8f1c466324c6c7f13b136 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 07:16:44 +0000 Subject: [PATCH 10/15] ServerSocket_POSIX.cpp: fix the indentation from the header/.cpp split The method bodies kept their in-class 8-space indentation from the monolithic header; un-dedent them to file scope. --- Common/Cpp/Sockets/ServerSocket_POSIX.cpp | 680 +++++++++++----------- 1 file changed, 339 insertions(+), 341 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.cpp b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp index f3c53c2495..cf375f1d49 100644 --- a/Common/Cpp/Sockets/ServerSocket_POSIX.cpp +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp @@ -22,425 +22,423 @@ namespace PokemonAutomation{ - - ServerSocket_POSIX::ServerSocket_POSIX(ThreadPool& thread_pool) - : m_thread_pool(thread_pool) - { - // Ignore SIGPIPE. Handle errors via errno instead - signal(SIGPIPE, SIG_IGN); - } + : m_thread_pool(thread_pool) +{ + // Ignore SIGPIPE. Handle errors via errno instead + signal(SIGPIPE, SIG_IGN); +} void ServerSocket_POSIX::start(const std::string& address, uint16_t port){ - { - std::lock_guard lg(m_lock); - if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING - || m_starting){ - return; - } - m_starting = true; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING + || m_starting){ + return; } + m_starting = true; + } - // Use the plain dispatch() rather than dispatch_now_blocking(): - // start() may be called from a thread that is itself running on - // this pool, and blocking here until a pool thread is free could - // deadlock a capped pool. The returned AsyncTask still lets - // close() join the listener thread. - try{ - AsyncTask task = m_thread_pool.dispatch([=, this]{ - accept_thread_loop(address, port); - }); - { - // m_task_lock, not m_lock: close() only reads the slot - // after it has waited for m_starting to clear, so this - // assignment is the only writer besides close(). - std::lock_guard lg(m_task_lock); - m_accept_thread = std::move(task); - } - }catch (...){ - { - std::lock_guard lg(m_lock); - m_starting = false; - m_cv.notify_all(); - } - m_listeners.run_method(&Listener::on_listen_finished, "Failed to start the listener thread."); - throw; + // Use the plain dispatch() rather than dispatch_now_blocking(): + // start() may be called from a thread that is itself running on + // this pool, and blocking here until a pool thread is free could + // deadlock a capped pool. The returned AsyncTask still lets + // close() join the listener thread. + try{ + AsyncTask task = m_thread_pool.dispatch([=, this]{ + accept_thread_loop(address, port); + }); + { + // m_task_lock, not m_lock: close() only reads the slot + // after it has waited for m_starting to clear, so this + // assignment is the only writer besides close(). + std::lock_guard lg(m_task_lock); + m_accept_thread = std::move(task); } - + }catch (...){ { std::lock_guard lg(m_lock); m_starting = false; m_cv.notify_all(); } + m_listeners.run_method(&Listener::on_listen_finished, "Failed to start the listener thread."); + throw; } + { + std::lock_guard lg(m_lock); + m_starting = false; + m_cv.notify_all(); + } +} + void ServerSocket_POSIX::close_connection(){ - int socket; - { - std::lock_guard lg(m_lock); - if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ - return; - } - m_close_requested = true; - m_state.store(State::LISTENING, std::memory_order_release); - socket = m_client; - } - if (socket != -1){ - // Tells the connection thread to stop; it closes the socket - // itself. This is safe to call from within a listener callback. - ::shutdown(socket, SHUT_RDWR); + int socket; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + return; } + m_close_requested = true; + m_state.store(State::LISTENING, std::memory_order_release); + socket = m_client; } + if (socket != -1){ + // Tells the connection thread to stop; it closes the socket + // itself. This is safe to call from within a listener callback. + ::shutdown(socket, SHUT_RDWR); + } +} void ServerSocket_POSIX::close() noexcept{ - { - std::unique_lock lg(m_lock); - // Wait for an in-progress start() to publish the listener task - // before we take the task slots. - m_cv.wait(lg, [&]{ return !m_starting; }); - if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING){ - m_close_requested = true; - m_state.store(State::DESTRUCTING, std::memory_order_release); - m_closing++; - } + { + std::unique_lock lg(m_lock); + // Wait for an in-progress start() to publish the listener task + // before we take the task slots. + m_cv.wait(lg, [&]{ return !m_starting; }); + if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING){ + m_close_requested = true; + m_state.store(State::DESTRUCTING, std::memory_order_release); + m_closing++; } + } - bool drain = false; - { - std::lock_guard lg(m_lock); - drain = (m_closing == 1); - } - if (!drain){ - // Either nothing was running (m_closing == 0), or another - // close() is already draining the threads (m_closing > 1). - // In the second case, wait for it to finish. - std::unique_lock lg(m_lock); - m_cv.wait(lg, [&]{ - return m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING; - }); - return; - } + bool drain = false; + { + std::lock_guard lg(m_lock); + drain = (m_closing == 1); + } + if (!drain){ + // Either nothing was running (m_closing == 0), or another + // close() is already draining the threads (m_closing > 1). + // In the second case, wait for it to finish. + std::unique_lock lg(m_lock); + m_cv.wait(lg, [&]{ + return m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING; + }); + return; + } - AsyncTask accept_thread; - AsyncTask client_thread; - { - // m_task_lock, not m_lock: moving the tasks out destroys the - // old AsyncTask objects, and their destructors wait for the - // tasks to finish - which they can only do if they can acquire - // m_lock. - std::lock_guard lg(m_task_lock); - accept_thread = std::move(m_accept_thread); - m_accept_thread = AsyncTask(); - client_thread = std::move(m_client_thread); - m_client_thread = AsyncTask(); - } + AsyncTask accept_thread; + AsyncTask client_thread; + { + // m_task_lock, not m_lock: moving the tasks out destroys the + // old AsyncTask objects, and their destructors wait for the + // tasks to finish - which they can only do if they can acquire + // m_lock. + std::lock_guard lg(m_task_lock); + accept_thread = std::move(m_accept_thread); + m_accept_thread = AsyncTask(); + client_thread = std::move(m_client_thread); + m_client_thread = AsyncTask(); + } - // The threads notice the new state within a few milliseconds and - // close their own sockets. - accept_thread.wait_and_ignore_exceptions(); - client_thread.wait_and_ignore_exceptions(); + // The threads notice the new state within a few milliseconds and + // close their own sockets. + accept_thread.wait_and_ignore_exceptions(); + client_thread.wait_and_ignore_exceptions(); - { - std::lock_guard lg(m_lock); - m_state.store(State::NOT_RUNNING, std::memory_order_release); - m_close_requested = false; - m_closing = 0; - m_cv.notify_all(); - } + { + std::lock_guard lg(m_lock); + m_state.store(State::NOT_RUNNING, std::memory_order_release); + m_close_requested = false; + m_closing = 0; + m_cv.notify_all(); } +} size_t ServerSocket_POSIX::send(const void* data, size_t bytes){ - int socket; - { - std::lock_guard lg(m_lock); - socket = - (m_state.load(std::memory_order_relaxed) == State::CONNECTED) ? m_client : -1; - } - if (socket == -1){ - return 0; - } + int socket; + { + std::lock_guard lg(m_lock); + socket = + (m_state.load(std::memory_order_relaxed) == State::CONNECTED) ? m_client : -1; + } + if (socket == -1){ + return 0; + } - const char* ptr = (const char*)data; - size_t sent = 0; + const char* ptr = (const char*)data; + size_t sent = 0; - while (bytes > 0){ - { - std::lock_guard lg(m_lock); - if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ - return sent; - } + while (bytes > 0){ + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + return sent; } + } - constexpr size_t BLOCK_SIZE = 1 << 30; - size_t current = std::min(BLOCK_SIZE, bytes); - ssize_t current_sent = ::send(socket, ptr, current, MSG_DONTWAIT | MSG_NOSIGNAL); - if (current_sent != -1){ - sent += (size_t)current_sent; - ptr += current_sent; - bytes -= (size_t)current_sent; - continue; - } + constexpr size_t BLOCK_SIZE = 1 << 30; + size_t current = std::min(BLOCK_SIZE, bytes); + ssize_t current_sent = ::send(socket, ptr, current, MSG_DONTWAIT | MSG_NOSIGNAL); + if (current_sent != -1){ + sent += (size_t)current_sent; + ptr += current_sent; + bytes -= (size_t)current_sent; + continue; + } - int error = errno; - switch (error){ - case EAGAIN: + int error = errno; + switch (error){ + case EAGAIN: #if EAGAIN != EWOULDBLOCK - case EWOULDBLOCK: + case EWOULDBLOCK: #endif - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - continue; - default: - // EPIPE, ECONNRESET, or any other error: the connection is - // gone. - return sent; - } + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + continue; + default: + // EPIPE, ECONNRESET, or any other error: the connection is + // gone. + return sent; } - - return sent; } + return sent; +} + void ServerSocket_POSIX::accept_thread_loop(std::string address, uint16_t port){ - int listen_socket = ::socket(AF_INET, SOCK_STREAM, 0); - if (listen_socket == -1){ - m_listeners.run_method( - &Listener::on_listen_finished, - "Failed to create the socket: " + std::string(strerror(errno)) - ); - return; - } + int listen_socket = ::socket(AF_INET, SOCK_STREAM, 0); + if (listen_socket == -1){ + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to create the socket: " + std::string(strerror(errno)) + ); + return; + } - int yes = 1; - setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)); - - sockaddr_in server; - std::memset(&server, 0, sizeof(server)); - server.sin_family = AF_INET; - server.sin_port = htons(port); - server.sin_addr.s_addr = inet_addr(address.c_str()); - - if (bind(listen_socket, (struct sockaddr*)&server, sizeof(server)) == -1){ - int error = errno; - ::close(listen_socket); - m_listeners.run_method( - &Listener::on_listen_finished, - "Failed to bind the socket: " + std::string(strerror(error)) - ); - return; - } + int yes = 1; + setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)); + + sockaddr_in server; + std::memset(&server, 0, sizeof(server)); + server.sin_family = AF_INET; + server.sin_port = htons(port); + server.sin_addr.s_addr = inet_addr(address.c_str()); + + if (bind(listen_socket, (struct sockaddr*)&server, sizeof(server)) == -1){ + int error = errno; + ::close(listen_socket); + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to bind the socket: " + std::string(strerror(error)) + ); + return; + } - if (::listen(listen_socket, 16) == -1){ - int error = errno; - ::close(listen_socket); - m_listeners.run_method( - &Listener::on_listen_finished, - "Failed to listen on the socket: " + std::string(strerror(error)) - ); - return; + if (::listen(listen_socket, 16) == -1){ + int error = errno; + ::close(listen_socket); + m_listeners.run_method( + &Listener::on_listen_finished, + "Failed to listen on the socket: " + std::string(strerror(error)) + ); + return; + } + + std::string start_error; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) == State::DESTRUCTING){ + // close() won the race. + start_error = "The server was closed before it could start listening."; + }else{ + m_listen = listen_socket; + listen_socket = -1; + m_state.store(State::LISTENING, std::memory_order_release); } + } + if (listen_socket != -1){ + ::close(listen_socket); + } - std::string start_error; + m_listeners.run_method(&Listener::on_listen_finished, start_error); + if (!start_error.empty()){ + return; + } + + // Accept loop. Runs for the whole lifetime of the server: it keeps + // polling while a connection is active, and rejects new clients in + // the meantime. + while (true){ + int listen; { std::lock_guard lg(m_lock); - if (m_state.load(std::memory_order_relaxed) == State::DESTRUCTING){ - // close() won the race. - start_error = "The server was closed before it could start listening."; - }else{ - m_listen = listen_socket; - listen_socket = -1; - m_state.store(State::LISTENING, std::memory_order_release); + State state = m_state.load(std::memory_order_relaxed); + if (state == State::DESTRUCTING || state == State::NOT_RUNNING){ + break; } - } - if (listen_socket != -1){ - ::close(listen_socket); + listen = m_listen; } - m_listeners.run_method(&Listener::on_listen_finished, start_error); - if (!start_error.empty()){ - return; + pollfd pf; + pf.fd = listen; + pf.events = POLLIN; + pf.revents = 0; + int ret = ::poll(&pf, 1, 10); + if (ret == -1){ + if (errno == EINTR){ + continue; + } + break; + } + if (ret == 0){ + continue; } - // Accept loop. Runs for the whole lifetime of the server: it keeps - // polling while a connection is active, and rejects new clients in - // the meantime. - while (true){ - int listen; - { - std::lock_guard lg(m_lock); - State state = m_state.load(std::memory_order_relaxed); - if (state == State::DESTRUCTING || state == State::NOT_RUNNING){ - break; - } - listen = m_listen; - } + int client_socket = ::accept(listen, nullptr, nullptr); + if (client_socket == -1){ + continue; + } - pollfd pf; - pf.fd = listen; - pf.events = POLLIN; - pf.revents = 0; - int ret = ::poll(&pf, 1, 10); - if (ret == -1){ - if (errno == EINTR){ - continue; - } + bool rejected = false; + bool accepted = false; + { + std::lock_guard lg(m_lock); + State state = m_state.load(std::memory_order_relaxed); + if (state == State::DESTRUCTING || state == State::NOT_RUNNING){ + ::close(client_socket); break; } - if (ret == 0){ - continue; - } - - int client_socket = ::accept(listen, nullptr, nullptr); - if (client_socket == -1){ - continue; + if (state == State::CONNECTED){ + // A client is already connected: reject this one. + ::close(client_socket); + rejected = true; + }else{ + fcntl(client_socket, F_SETFL, O_NONBLOCK); + m_client = client_socket; + m_close_requested = false; + m_state.store(State::CONNECTED, std::memory_order_release); + accepted = true; } + } - bool rejected = false; - bool accepted = false; - { + if (accepted){ + // Start the connection thread without holding m_lock. + // The assignment to m_client_thread destroys the previous + // connection task's AsyncTask, whose destructor waits for + // that task to finish - and that task may be blocked on + // m_lock. + try{ + AsyncTask task = m_thread_pool.dispatch([this]{ + connection_thread_loop(); + }); + std::lock_guard lg(m_task_lock); + m_client_thread = std::move(task); + }catch (...){ + // The connection thread could not be started. Drop + // this connection and keep listening. std::lock_guard lg(m_lock); - State state = m_state.load(std::memory_order_relaxed); - if (state == State::DESTRUCTING || state == State::NOT_RUNNING){ - ::close(client_socket); - break; - } - if (state == State::CONNECTED){ - // A client is already connected: reject this one. - ::close(client_socket); - rejected = true; - }else{ - fcntl(client_socket, F_SETFL, O_NONBLOCK); - m_client = client_socket; - m_close_requested = false; - m_state.store(State::CONNECTED, std::memory_order_release); - accepted = true; - } - } - - if (accepted){ - // Start the connection thread without holding m_lock. - // The assignment to m_client_thread destroys the previous - // connection task's AsyncTask, whose destructor waits for - // that task to finish - and that task may be blocked on - // m_lock. - try{ - AsyncTask task = m_thread_pool.dispatch([this]{ - connection_thread_loop(); - }); - std::lock_guard lg(m_task_lock); - m_client_thread = std::move(task); - }catch (...){ - // The connection thread could not be started. Drop - // this connection and keep listening. - std::lock_guard lg(m_lock); - ::close(client_socket); - m_client = -1; - m_state.store(State::LISTENING, std::memory_order_release); - } - } - - if (rejected){ - m_listeners.run_method(&Listener::on_connect_rejected); + ::close(client_socket); + m_client = -1; + m_state.store(State::LISTENING, std::memory_order_release); } } - int listen; - { - std::lock_guard lg(m_lock); - listen = m_listen; - m_listen = -1; - } - if (listen != -1){ - ::close(listen); + if (rejected){ + m_listeners.run_method(&Listener::on_connect_rejected); } } + int listen; + { + std::lock_guard lg(m_lock); + listen = m_listen; + m_listen = -1; + } + if (listen != -1){ + ::close(listen); + } +} + void ServerSocket_POSIX::connection_thread_loop(){ - m_listeners.run_method(&Listener::on_thread_start); + m_listeners.run_method(&Listener::on_thread_start); - int socket; - bool active; - { - std::lock_guard lg(m_lock); - socket = m_client; - active = (m_state.load(std::memory_order_relaxed) == State::CONNECTED); - if (!active && m_client == socket){ - m_client = -1; - } - } - if (!active){ - // The connection was closed before this thread started. - ::close(socket); - return; + int socket; + bool active; + { + std::lock_guard lg(m_lock); + socket = m_client; + active = (m_state.load(std::memory_order_relaxed) == State::CONNECTED); + if (!active && m_client == socket){ + m_client = -1; } + } + if (!active){ + // The connection was closed before this thread started. + ::close(socket); + return; + } - m_listeners.run_method(&Listener::on_connect); + m_listeners.run_method(&Listener::on_connect); - constexpr size_t BUFFER_SIZE = 4096; - char buffer[BUFFER_SIZE]; - std::string error; + constexpr size_t BUFFER_SIZE = 4096; + char buffer[BUFFER_SIZE]; + std::string error; - while (true){ - ssize_t bytes; - int error_no = 0; - bool no_data; - { - std::lock_guard lg(m_lock); - if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ - break; - } - bytes = ::recv(socket, buffer, BUFFER_SIZE, 0); - no_data = (bytes == -1); - if (no_data){ - error_no = errno; - no_data = (error_no == EAGAIN); + while (true){ + ssize_t bytes; + int error_no = 0; + bool no_data; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::CONNECTED){ + break; + } + bytes = ::recv(socket, buffer, BUFFER_SIZE, 0); + no_data = (bytes == -1); + if (no_data){ + error_no = errno; + no_data = (error_no == EAGAIN); #if EAGAIN != EWOULDBLOCK - no_data = no_data || (error_no == EWOULDBLOCK); + no_data = no_data || (error_no == EWOULDBLOCK); #endif - } } + } - if (bytes > 0){ - m_listeners.run_method(&Listener::on_receive_data, buffer, bytes); - continue; - } + if (bytes > 0){ + m_listeners.run_method(&Listener::on_receive_data, buffer, bytes); + continue; + } - if (no_data){ - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - continue; - } + if (no_data){ + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + continue; + } - if (bytes == 0){ - error = "Connection closed by client."; - }else{ - switch (error_no){ - case EINTR: - continue; - default: - error = "Receive error (errno " + std::to_string(error_no) + "): " + std::string(strerror(error_no)); - break; - } + if (bytes == 0){ + error = "Connection closed by client."; + }else{ + switch (error_no){ + case EINTR: + continue; + default: + error = "Receive error (errno " + std::to_string(error_no) + "): " + std::string(strerror(error_no)); + break; } - break; } + break; + } - bool report_disconnect; - { - std::lock_guard lg(m_lock); - report_disconnect = false; - if (m_client == socket){ - // Still the active connection (or it was closed before a - // new one took over): we own the socket and the state. - m_client = -1; - if (m_state.load(std::memory_order_relaxed) == State::CONNECTED){ - m_state.store(State::LISTENING, std::memory_order_release); - report_disconnect = !m_close_requested; - } + bool report_disconnect; + { + std::lock_guard lg(m_lock); + report_disconnect = false; + if (m_client == socket){ + // Still the active connection (or it was closed before a + // new one took over): we own the socket and the state. + m_client = -1; + if (m_state.load(std::memory_order_relaxed) == State::CONNECTED){ + m_state.store(State::LISTENING, std::memory_order_release); + report_disconnect = !m_close_requested; } } - ::close(socket); - if (report_disconnect){ - m_listeners.run_method(&Listener::on_disconnect, error); - } } + ::close(socket); + if (report_disconnect){ + m_listeners.run_method(&Listener::on_disconnect, error); + } +} } From d3ea387f72ec23f7a68e80b9a00408b8b13847bb Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 06:47:46 +0000 Subject: [PATCH 11/15] ML: temporary [ORT-DIAG] stderr markers and ORT INFO logging While investigating the intermittent Mac Intel PaddleOCR segfault: - Print the ONNX Runtime version, hardware_concurrency() and the chosen intra/inter-op thread counts to stderr (the file logger is not in the CI log). - Run the global Ort::Env at ORT_LOGGING_LEVEL_INFO so ONNX Runtime's own session-creation progress (Session Options, thread pool params, graph transforms, BFCArena, ...) lands in the CI log. - Add [ORT-DIAG] stderr markers around create_session() and the first PaddleOCRPipeline::recognize_line() Run() calls, to pin down which stage the crash is in. All of this is temporary: revert when the crash is understood. --- .../Source/ML/Inference/ML_PaddleOCRPipeline.cpp | 5 +++++ .../Source/ML/Models/ML_ONNXRuntimeHelpers.cpp | 4 ++++ SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp | 10 +++++++++- 3 files changed, 18 insertions(+), 1 deletion(-) diff --git a/SerialPrograms/Source/ML/Inference/ML_PaddleOCRPipeline.cpp b/SerialPrograms/Source/ML/Inference/ML_PaddleOCRPipeline.cpp index ab9ab35fb7..066d79e626 100644 --- a/SerialPrograms/Source/ML/Inference/ML_PaddleOCRPipeline.cpp +++ b/SerialPrograms/Source/ML/Inference/ML_PaddleOCRPipeline.cpp @@ -87,6 +87,7 @@ PaddleOCRPipeline::PaddleOCRPipeline( , m_det_path(std::move(det_path)) { load_dictionary(Filesystem::Path(dict_path)); + std::cerr << "[ORT-DIAG] PaddleOCRPipeline ctor finished: " << rec_path << std::endl; } PaddleOCRDetector& PaddleOCRPipeline::detector(){ @@ -381,6 +382,10 @@ std::string PaddleOCRPipeline::recognize_line(const cv::Mat& cropped_image){ const char* input_names[] = {m_input_name.c_str()}; const char* output_names[] = {m_output_name.c_str()}; + static int diag_run_count = 0; + if (++diag_run_count <= 20){ + std::cerr << "[ORT-DIAG] recognize_line: about to call session->Run() (call " << diag_run_count << ")" << std::endl; + } try{ if (debugging){ m_logger.log("[OCR-DEBUG] Calling m_rec_session.Run() now..."); diff --git a/SerialPrograms/Source/ML/Models/ML_ONNXRuntimeHelpers.cpp b/SerialPrograms/Source/ML/Models/ML_ONNXRuntimeHelpers.cpp index d0dac9d2a8..620173c414 100644 --- a/SerialPrograms/Source/ML/Models/ML_ONNXRuntimeHelpers.cpp +++ b/SerialPrograms/Source/ML/Models/ML_ONNXRuntimeHelpers.cpp @@ -167,9 +167,11 @@ Ort::Session create_session( const std::string& model_cache_path, bool try_gpu ){ + std::cerr << "[ORT-DIAG] create_session(" << model_path << ", try_gpu=" << (try_gpu ? 1 : 0) << ")" << std::endl; bool write_flag_file = true; std::string file_hash; std::tie(write_flag_file, file_hash) = clean_up_old_model_cache(model_cache_path, model_path); + std::cerr << "[ORT-DIAG] clean_up_old_model_cache done, write_flag_file=" << write_flag_file << std::endl; auto onnx_path = str_to_onnx_str(model_path); @@ -205,7 +207,9 @@ Ort::Session create_session( Ort::SessionOptions cpu_options = create_session_options(model_cache_path, false); cpu_options.DisablePerSessionThreads(); + std::cerr << "[ORT-DIAG] calling Ort::Session constructor (CPU, per-session threads disabled)..." << std::endl; Ort::Session session{global_ort_env(), onnx_path.c_str(), cpu_options}; + std::cerr << "[ORT-DIAG] Ort::Session constructor returned: " << model_path << std::endl; logger.log("Ort::Session created"); return session; } diff --git a/SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp b/SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp index c69ace4a39..e16195442e 100644 --- a/SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp +++ b/SerialPrograms/Source/ML/Models/ML_OrtEnv.cpp @@ -4,6 +4,8 @@ * */ +#include +#include #include "Common/Cpp/Exceptions.h" #include "Common/Cpp/PrettyPrint.h" #include "CommonFramework/Logging/Logger.h" @@ -19,11 +21,15 @@ Ort::ThreadingOptions make_ort_threading_options(){ { int threads = (int)PerformanceOptions::instance().ONNX_OPTIONS.MAX_INTRA_OP_THREADS; global_logger_tagged().log("Setting ONNX Intra-Op Threads: " + tostr_u_commas(threads)); + std::cerr << "[ORT-DIAG] ONNX Runtime " << Ort::GetVersionString() + << ", hardware_concurrency() = " << std::thread::hardware_concurrency() + << ", SetGlobalIntraOpNumThreads(" << threads << ")" << std::endl; ret.SetGlobalIntraOpNumThreads(threads); } { int threads = (int)PerformanceOptions::instance().ONNX_OPTIONS.MAX_INTER_OP_THREADS; global_logger_tagged().log("Setting ONNX Inter-Op Threads: " + tostr_u_commas(threads)); + std::cerr << "[ORT-DIAG] SetGlobalInterOpNumThreads(" << threads << ")" << std::endl; ret.SetGlobalInterOpNumThreads(threads); } ret.SetGlobalSpinControl(0); @@ -31,6 +37,7 @@ Ort::ThreadingOptions make_ort_threading_options(){ // them, and they are very slow on Intel CPUs. ONNX Runtime only supports this on x86; // see Common/Cpp/Hardware/FloatingPointMode.h. ret.SetGlobalDenormalAsZero(); + std::cerr << "[ORT-DIAG] SetGlobalSpinControl(0), SetGlobalDenormalAsZero()" << std::endl; return ret; } @@ -42,9 +49,10 @@ Ort::Env& global_ort_env(){ #endif static Ort::Env env( make_ort_threading_options(), - ORT_LOGGING_LEVEL_WARNING, + ORT_LOGGING_LEVEL_INFO, // [ORT-DIAG] was WARNING "Global_ThreadPool_Env" ); + std::cerr << "[ORT-DIAG] Ort::Env created" << std::endl; return env; } From eee9c0d1c1c71d55bb95329e10a6ec14f57b6dbe Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 07:18:57 +0000 Subject: [PATCH 12/15] Tests: [ORT-DIAG] dump a backtrace to stderr on a fatal signal in test mode The CI log lost the output right up to the segfault because stdout is fully buffered when piped: in the #1812 Mac Intel run the crash happened ~2.5 s after 'Running parallel unit tests...' and not a single line after that was captured. Two temporary diagnostics: - A SIGSEGV/SIGBUS/SIGABRT/SIGILL/SIGFPE handler (command-line test mode only) that writes a native backtrace of the crashing thread straight to fd 2 with backtrace_symbols_fd(), then restores the default disposition and re-raises so the OS crash report and exit code are unchanged. - setvbuf(stdout, _IOLBF) in test mode, so the test 'Starting:'/'Passed:' lines reach the log as they happen and the log shows the last thing the crashing process was doing. --- .../Source/Tests/CommandLineTests.cpp | 121 +++++++++++++----- 1 file changed, 88 insertions(+), 33 deletions(-) diff --git a/SerialPrograms/Source/Tests/CommandLineTests.cpp b/SerialPrograms/Source/Tests/CommandLineTests.cpp index a723fef213..898d5cf8db 100644 --- a/SerialPrograms/Source/Tests/CommandLineTests.cpp +++ b/SerialPrograms/Source/Tests/CommandLineTests.cpp @@ -1,33 +1,88 @@ -/* Command Line Tests - * - * From: https://github.com/PokemonAutomation/ - * - */ - - -#include -#include "CommonFramework/Logging/Logger.h" -#include "ComputerPrograms/ComputerPrograms_UnitTestRunner.h" - -#include "CommandLineTests.h" - -using std::cout; -using std::endl; - -namespace PokemonAutomation{ - - - -int run_command_line_tests(bool augment_color_distortions){ - cout << "Running parallel unit tests..." << endl; - ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); - if (runner.run(augment_color_distortions)){ - return 1; - } - cout << "Running parallel unit tests... Done!" << endl; - return 0; -} - - - -} +/* Command Line Tests + * + * From: https://github.com/PokemonAutomation/ + * + */ + + +#include +#include "CommonFramework/Logging/Logger.h" +#include "ComputerPrograms/ComputerPrograms_UnitTestRunner.h" + +#include "CommandLineTests.h" + +#if defined(__APPLE__) || defined(__linux__) +#include +#include +#include +#endif + +using std::cout; +using std::endl; + +namespace PokemonAutomation{ + + +#if defined(__APPLE__) || defined(__linux__) +// On a fatal signal, dump a native backtrace of the crashing thread straight to stderr +// (fd 2, unbuffered) so it survives in the CI log even though stdout's buffer is lost when +// the process segfaults. Then restore the default disposition and re-raise, so the OS still +// writes its crash report (.ips on macOS) and the exit code still reflects the crash. +static void crash_backtrace_handler(int sig){ + // Re-entrancy guard: if we crash while handling a signal, just die. + static volatile sig_atomic_t in_handler = 0; + if (in_handler){ + _exit(128 + sig); + } + in_handler = 1; + + const char msg[] = "\n[PA-CRASH] Fatal signal, backtrace of the crashing thread:\n"; + if (write(2, msg, sizeof(msg) - 1) < 0){ + // Nothing more we can do. + } + + void* frames[128]; + int n = backtrace(frames, 128); + // Writes one symbolized frame per line to fd 2. glibc declares it void; Apple's returns int. +#if defined(__linux__) + backtrace_symbols_fd(frames, n, 2); +#else + if (backtrace_symbols_fd(frames, n, 2) < 0){ + // Nothing more we can do. + } +#endif + + // Let the OS do its normal crash handling (crash report + exit code). + signal(sig, SIG_DFL); + raise(sig); +} + +static void install_crash_backtrace_handlers(){ + static const int sigs[] = { SIGSEGV, SIGBUS, SIGABRT, SIGILL, SIGFPE }; + for (int sig : sigs){ + signal(sig, crash_backtrace_handler); + } +} +#else +static void install_crash_backtrace_handlers(){} +#endif + + +int run_command_line_tests(bool augment_color_distortions){ + install_crash_backtrace_handlers(); + // [ORT-DIAG] Keep stdout line-buffered in CI: when piped, stdio defaults to full + // buffering, so the last few KB of test output (and where the crash happened) is lost + // in the buffer when the process dies. + setvbuf(stdout, nullptr, _IOLBF, 0); + cout << "Running parallel unit tests..." << endl; + ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); + if (runner.run(augment_color_distortions)){ + return 1; + } + cout << "Running parallel unit tests... Done!" << endl; + return 0; +} + + + +} From 736430548584d12c05c6cb8fc38fdd9c1ca29463 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 08:08:24 +0000 Subject: [PATCH 13/15] Tests: [ORT-DIAG] fix backtrace_symbols_fd() on Apple, restore CRLF Apple's execinfo.h declares backtrace_symbols_fd() as void (glibc's returns int), so ignore its result on both. Also restore the CRLF line endings the previous commit accidentally rewrote to LF. --- .../Source/Tests/CommandLineTests.cpp | 169 +++++++++--------- 1 file changed, 81 insertions(+), 88 deletions(-) diff --git a/SerialPrograms/Source/Tests/CommandLineTests.cpp b/SerialPrograms/Source/Tests/CommandLineTests.cpp index 898d5cf8db..c673a074a8 100644 --- a/SerialPrograms/Source/Tests/CommandLineTests.cpp +++ b/SerialPrograms/Source/Tests/CommandLineTests.cpp @@ -1,88 +1,81 @@ -/* Command Line Tests - * - * From: https://github.com/PokemonAutomation/ - * - */ - - -#include -#include "CommonFramework/Logging/Logger.h" -#include "ComputerPrograms/ComputerPrograms_UnitTestRunner.h" - -#include "CommandLineTests.h" - -#if defined(__APPLE__) || defined(__linux__) -#include -#include -#include -#endif - -using std::cout; -using std::endl; - -namespace PokemonAutomation{ - - -#if defined(__APPLE__) || defined(__linux__) -// On a fatal signal, dump a native backtrace of the crashing thread straight to stderr -// (fd 2, unbuffered) so it survives in the CI log even though stdout's buffer is lost when -// the process segfaults. Then restore the default disposition and re-raise, so the OS still -// writes its crash report (.ips on macOS) and the exit code still reflects the crash. -static void crash_backtrace_handler(int sig){ - // Re-entrancy guard: if we crash while handling a signal, just die. - static volatile sig_atomic_t in_handler = 0; - if (in_handler){ - _exit(128 + sig); - } - in_handler = 1; - - const char msg[] = "\n[PA-CRASH] Fatal signal, backtrace of the crashing thread:\n"; - if (write(2, msg, sizeof(msg) - 1) < 0){ - // Nothing more we can do. - } - - void* frames[128]; - int n = backtrace(frames, 128); - // Writes one symbolized frame per line to fd 2. glibc declares it void; Apple's returns int. -#if defined(__linux__) - backtrace_symbols_fd(frames, n, 2); -#else - if (backtrace_symbols_fd(frames, n, 2) < 0){ - // Nothing more we can do. - } -#endif - - // Let the OS do its normal crash handling (crash report + exit code). - signal(sig, SIG_DFL); - raise(sig); -} - -static void install_crash_backtrace_handlers(){ - static const int sigs[] = { SIGSEGV, SIGBUS, SIGABRT, SIGILL, SIGFPE }; - for (int sig : sigs){ - signal(sig, crash_backtrace_handler); - } -} -#else -static void install_crash_backtrace_handlers(){} -#endif - - -int run_command_line_tests(bool augment_color_distortions){ - install_crash_backtrace_handlers(); - // [ORT-DIAG] Keep stdout line-buffered in CI: when piped, stdio defaults to full - // buffering, so the last few KB of test output (and where the crash happened) is lost - // in the buffer when the process dies. - setvbuf(stdout, nullptr, _IOLBF, 0); - cout << "Running parallel unit tests..." << endl; - ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); - if (runner.run(augment_color_distortions)){ - return 1; - } - cout << "Running parallel unit tests... Done!" << endl; - return 0; -} - - - -} +/* Command Line Tests + * + * From: https://github.com/PokemonAutomation/ + * + */ + + +#include +#include "CommonFramework/Logging/Logger.h" +#include "ComputerPrograms/ComputerPrograms_UnitTestRunner.h" + +#include "CommandLineTests.h" + +#if defined(__APPLE__) || defined(__linux__) +#include +#include +#include +#endif + +using std::cout; +using std::endl; + +namespace PokemonAutomation{ + + +#if defined(__APPLE__) || defined(__linux__) +// On a fatal signal, dump a native backtrace of the crashing thread straight to stderr +// (fd 2, unbuffered) so it survives in the CI log even though stdout's buffer is lost when +// the process segfaults. Then restore the default disposition and re-raise, so the OS still +// writes its crash report (.ips on macOS) and the exit code still reflects the crash. +static void crash_backtrace_handler(int sig){ + // Re-entrancy guard: if we crash while handling a signal, just die. + static volatile sig_atomic_t in_handler = 0; + if (in_handler){ + _exit(128 + sig); + } + in_handler = 1; + + const char msg[] = "\n[PA-CRASH] Fatal signal, backtrace of the crashing thread:\n"; + if (write(2, msg, sizeof(msg) - 1) < 0){ + // Nothing more we can do. + } + + void* frames[128]; + int n = backtrace(frames, 128); + backtrace_symbols_fd(frames, n, 2); // writes one symbolized frame per line to fd 2 + + // Let the OS do its normal crash handling (crash report + exit code). + signal(sig, SIG_DFL); + raise(sig); +} + +static void install_crash_backtrace_handlers(){ + static const int sigs[] = { SIGSEGV, SIGBUS, SIGABRT, SIGILL, SIGFPE }; + for (int sig : sigs){ + signal(sig, crash_backtrace_handler); + } +} +#else +static void install_crash_backtrace_handlers(){} +#endif + + +int run_command_line_tests(bool augment_color_distortions){ + install_crash_backtrace_handlers(); + // [ORT-DIAG] Keep stdout line-buffered in CI: when piped, stdio defaults to full + // buffering, so the last few KB of test output (and where the crash happened) is lost + // in the buffer when the process dies. + setvbuf(stdout, nullptr, _IOLBF, 0); + cout << "Running parallel unit tests..." << endl; + ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); + if (runner.run(augment_color_distortions)){ + return 1; + } + cout << "Running parallel unit tests... Done!" << endl; + return 0; +} + + + +} From 0f4c65ae9f0ecf555400e7809f82cb3e8b1be028 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 09:14:40 +0000 Subject: [PATCH 14/15] Tests: [ORT-DIAG] give setvbuf() an explicit 8 KB buffer setvbuf(stdout, nullptr, _IOLBF, 0) crashed the Windows CI runner with 0xC0000409 (stack-buffer-overrun) ~0.7 s into the test run, on the first stdout write after the call: with size 0, MSVC's ucrt appears to install a zero-size stdio buffer for stdout. 8192 is the conventional default. --- SerialPrograms/Source/Tests/CommandLineTests.cpp | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/SerialPrograms/Source/Tests/CommandLineTests.cpp b/SerialPrograms/Source/Tests/CommandLineTests.cpp index c673a074a8..62cc4323ec 100644 --- a/SerialPrograms/Source/Tests/CommandLineTests.cpp +++ b/SerialPrograms/Source/Tests/CommandLineTests.cpp @@ -66,7 +66,10 @@ int run_command_line_tests(bool augment_color_distortions){ // [ORT-DIAG] Keep stdout line-buffered in CI: when piped, stdio defaults to full // buffering, so the last few KB of test output (and where the crash happened) is lost // in the buffer when the process dies. - setvbuf(stdout, nullptr, _IOLBF, 0); + // 8192 bytes: 0 means "implementation-defined" and MSVC's ucrt can end up with a + // zero-size buffer, which corrupts the next stdout write (observed 0xC0000409 on the + // Windows CI runner). + setvbuf(stdout, nullptr, _IOLBF, 8192); cout << "Running parallel unit tests..." << endl; ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); if (runner.run(augment_color_distortions)){ From 48dc961009079371303714b87fdd4b05ebe1e8e6 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 09:40:47 +0000 Subject: [PATCH 15/15] Tests: [ORT-DIAG] run the suite 3x, re-creating the PaddleOCR sessions The Mac Intel crash is probabilistic (~60% of runs) and lands in the PaddleOCR ONNX session creation, which happens only once per process because the sessions are cached. Run the full suite three times and clear the PaddleOCR cache between runs so the sessions are re-created each time, raising the per-CI-run reproduction odds to ~93%. Stop and fail on the first reported test failure. --- .../Source/Tests/CommandLineTests.cpp | 23 ++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/SerialPrograms/Source/Tests/CommandLineTests.cpp b/SerialPrograms/Source/Tests/CommandLineTests.cpp index 62cc4323ec..ab6fe321f0 100644 --- a/SerialPrograms/Source/Tests/CommandLineTests.cpp +++ b/SerialPrograms/Source/Tests/CommandLineTests.cpp @@ -7,6 +7,7 @@ #include #include "CommonFramework/Logging/Logger.h" +#include "CommonTools/OCR/OCR_PaddleOCR.h" #include "ComputerPrograms/ComputerPrograms_UnitTestRunner.h" #include "CommandLineTests.h" @@ -71,10 +72,26 @@ int run_command_line_tests(bool augment_color_distortions){ // Windows CI runner). setvbuf(stdout, nullptr, _IOLBF, 8192); cout << "Running parallel unit tests..." << endl; - ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); - if (runner.run(augment_color_distortions)){ - return 1; + + // [ORT-DIAG] The Mac Intel crash is probabilistic (~60% of CI runs) and happens at + // PaddleOCR's ONNX session creation, which normally happens only once per process + // (the sessions are cached). To raise the odds of reproducing it in CI, run the full + // suite several times, clearing the PaddleOCR cache between runs so the sessions are + // re-created each time. + constexpr int k_repeat_runs = 3; + for (int run_index = 1; run_index <= k_repeat_runs; ++run_index){ + std::cerr << "[ORT-DIAG] Test suite run " << run_index << " of " << k_repeat_runs << std::endl; + if (run_index > 1){ + OCR::clear_paddle_ocr_cache(); // Force PaddleOCR to re-create its ONNX sessions. + } + ComputerPrograms::CommandLineUnitTestRunner runner(global_logger_command_line()); + if (runner.run(augment_color_distortions)){ + std::cerr << "[ORT-DIAG] Test suite run " << run_index << " reported failures; stopping." + << std::endl; + return 1; + } } + cout << "Running parallel unit tests... Done!" << endl; return 0; }