From f7bb7111748dbc69d72bee643ea5b5775bb357b7 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 1/7] 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 | 7 + SerialPrograms/cmake/SourceFiles.cmake | 2 + 4 files changed, 536 insertions(+) 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 d1b236702e..264d80e6fa 100644 --- a/Common/Cpp/Sockets/Sockets_Tests.cpp +++ b/Common/Cpp/Sockets/Sockets_Tests.cpp @@ -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 #ifdef _WIN32 #include "Common/Cpp/Sockets/ClientSocket_WinSocket.h" #include "Common/Cpp/Sockets/ServerSocket_WinSocket.h" @@ -881,6 +885,9 @@ void add_tests_Sockets(UnitTestDatabase& database, ThreadPool& thread_pool){ "Sockets (Qt) - Completion lifetime under event-thread stall" ); +#ifndef _WIN32 + add_socket_tests(database, thread_pool, "POSIX"); +#endif #ifdef _WIN32 add_socket_tests(database, thread_pool, "WinSocket"); #endif diff --git a/SerialPrograms/cmake/SourceFiles.cmake b/SerialPrograms/cmake/SourceFiles.cmake index 881ae16061..7e167477e9 100644 --- a/SerialPrograms/cmake/SourceFiles.cmake +++ b/SerialPrograms/cmake/SourceFiles.cmake @@ -408,8 +408,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_POSIX.cpp ../Common/Cpp/Sockets/ServerSocket_WinSocket.cpp ../Common/Cpp/Sockets/ServerSocket_Qt.cpp + ../Common/Cpp/Sockets/ServerSocket_POSIX.h ../Common/Cpp/Sockets/ServerSocket_WinSocket.h ../Common/Cpp/Sockets/ServerSocket_Qt.h ../Common/Cpp/Sockets/Sockets_Tests.cpp From 2e127b6b9be4dc6c6e896b1990120b4abc2d04d9 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 2/7] 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 c615e103354f0b1d7d003f6f8327b17997f05fd0 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 3/7] 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 ecd70435b69c20a3f5d5331f89689ebe4579715d 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 4/7] 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 4c1f53812a87a88292059014a26c705aa25f97a2 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 10:31:34 +0000 Subject: [PATCH 5/7] ServerSocket_POSIX.h: remove the excess blank lines The header/.cpp split left 2-3 blank lines between most members; normalize to the single-blank style used by the client headers. --- Common/Cpp/Sockets/ServerSocket_POSIX.h | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.h b/Common/Cpp/Sockets/ServerSocket_POSIX.h index 7626cb64de..008f08d3bb 100644 --- a/Common/Cpp/Sockets/ServerSocket_POSIX.h +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.h @@ -14,7 +14,6 @@ #include "Common/Cpp/Concurrency/Mutex.h" #include "AbstractServerSocket.h" - namespace PokemonAutomation{ class ThreadPool; @@ -25,32 +24,23 @@ class ServerSocket_POSIX final : public AbstractServerSocket{ 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: @@ -77,4 +67,4 @@ class ServerSocket_POSIX final : public AbstractServerSocket{ } -#endif +#endif \ No newline at end of file From ede3f435ee3ab5df17378ac097de67b983ac1538 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sun, 1 Nov 2026 11:53:23 +0000 Subject: [PATCH 6/7] ServerSocket_POSIX: don't call recv() while holding m_lock Same issue as the WinSocket backend (#1820): the connection thread called ::recv() inside the m_lock scope. The socket is non-blocking so it returns immediately in practice, but a socket call must not stretch the lock across a kernel transition: if the socket were ever made blocking, every other caller of m_lock (close(), close_connection(), send(), the accept loop) would hang while the client stays silent. The loop now takes m_lock only for the state check, calls recv() unlocked, and re-checks under the lock that the connection has not been replaced before dispatching on_receive_data. --- Common/Cpp/Sockets/ServerSocket_POSIX.cpp | 59 +++++++++++++---------- 1 file changed, 34 insertions(+), 25 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.cpp b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp index cf375f1d49..1a5a6d9666 100644 --- a/Common/Cpp/Sockets/ServerSocket_POSIX.cpp +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.cpp @@ -377,45 +377,54 @@ void ServerSocket_POSIX::connection_thread_loop(){ 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 - } } + // The socket is non-blocking, so this returns immediately. It is + // deliberately called without holding m_lock: the state check + // above and the result handling below are the only parts that + // need the lock, and no socket call should stretch it. + ssize_t bytes = ::recv(socket, buffer, BUFFER_SIZE, 0); + if (bytes > 0){ + // A new connection may have taken over while we were + // reading: only dispatch the data if we still are the + // active connection. + bool still_active; + { + std::lock_guard lg(m_lock); + still_active = (m_client == socket); + } + if (!still_active){ + break; + } 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: + if (bytes == -1){ + int error_no = errno; + if (error_no == EAGAIN){ + // No data yet. + std::this_thread::sleep_for(std::chrono::milliseconds(1)); continue; - default: - error = "Receive error (errno " + std::to_string(error_no) + "): " + std::string(strerror(error_no)); - break; } +#if EAGAIN != EWOULDBLOCK + if (error_no == EWOULDBLOCK){ + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + continue; + } +#endif + if (error_no == EINTR){ + continue; + } + error = "Receive error (errno " + std::to_string(error_no) + "): " + std::string(strerror(error_no)); + }else if (bytes == 0){ + error = "Connection closed by client."; } break; } From a27eda418fafe5cae0523cea5bbe5deb158f2e47 Mon Sep 17 00:00:00 2001 From: Alexander Yee Date: Sun, 1 Nov 2026 12:36:37 -0800 Subject: [PATCH 7/7] formatting --- Common/Cpp/Sockets/ServerSocket_POSIX.h | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/Common/Cpp/Sockets/ServerSocket_POSIX.h b/Common/Cpp/Sockets/ServerSocket_POSIX.h index 008f08d3bb..9d7965dd57 100644 --- a/Common/Cpp/Sockets/ServerSocket_POSIX.h +++ b/Common/Cpp/Sockets/ServerSocket_POSIX.h @@ -29,9 +29,7 @@ class ServerSocket_POSIX final : public AbstractServerSocket{ } 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; @@ -40,9 +38,9 @@ class ServerSocket_POSIX final : public AbstractServerSocket{ private: void accept_thread_loop(std::string address, uint16_t port); - void connection_thread_loop(); + private: ThreadPool& m_thread_pool; Mutex m_lock; @@ -67,4 +65,4 @@ class ServerSocket_POSIX final : public AbstractServerSocket{ } -#endif \ No newline at end of file +#endif