From c614694a259c7eab6e4c051d79b77709188a647f Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sat, 31 Oct 2026 08:16:15 +0000 Subject: [PATCH 1/2] Sockets: fixes to ClientSocket_POSIX Ports the same state-machine fixes as the WinSocket PR to the POSIX implementation: - Detect connection loss: recv()==0 (graceful close) and fatal recv() errors (ECONNRESET, ENOTCONN, ...) now fire on_disconnect() and return the socket to NOT_RUNNING instead of exiting the thread silently with the state stuck at CONNECTED. - Close connections properly: close() now joins the receive thread, closes the socket, and returns to NOT_RUNNING so connect() can be called again. It is safe to call multiple times. - New sockets per connect(): a socket whose previous connection ended cannot be reconnected, so connect() always starts from a fresh socket. m_socket is now std::atomic since the receive thread and send() share it. - Non-blocking connect done correctly: only one connect() call; while in progress, wait with poll() for the socket to become writable and check SO_ERROR for the result. (Calling connect() again while in progress returns EISCONN on Linux even when the attempt is still pending or will fail, which the old retry loop treated as success - connecting to a dead port marked the socket CONNECTED with an empty error.) - Connect failures (refused, unreachable, ...) now report the error through on_connect_finished() and return to NOT_RUNNING. --- Common/Cpp/Sockets/ClientSocket_POSIX.h | 240 +++++++++++++++++------- 1 file changed, 172 insertions(+), 68 deletions(-) diff --git a/Common/Cpp/Sockets/ClientSocket_POSIX.h b/Common/Cpp/Sockets/ClientSocket_POSIX.h index 324090c69c..eadf4d8f01 100644 --- a/Common/Cpp/Sockets/ClientSocket_POSIX.h +++ b/Common/Cpp/Sockets/ClientSocket_POSIX.h @@ -2,13 +2,13 @@ * * From: https://github.com/PokemonAutomation/ * - * This file is mostly untested! - * */ #ifndef PokemonAutomation_ClientSocket_POSIX_H #define PokemonAutomation_ClientSocket_POSIX_H +#include +#include #include #include #include @@ -32,58 +32,110 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ public: ClientSocket_POSIX(ThreadPool& thread_pool) : m_thread_pool(thread_pool) - , m_socket(socket(AF_INET, SOCK_STREAM, 0)){ + { // Ignore SIGPIPE. Handle errors via errno instead signal(SIGPIPE, SIG_IGN); - - if (m_socket != -1){ - fcntl(m_socket, F_SETFL, O_NONBLOCK); - } } virtual ~ClientSocket_POSIX(){ close(); - m_thread.wait_and_ignore_exceptions(); - close_socket(); } + // Stops the receive thread, closes the socket, and returns the socket to + // State::NOT_RUNNING so "connect()" can be called again. This is safe to + // call multiple times. + // + // This blocks until the receive thread has exited, so it must not be + // called from within a listener callback (which runs on that thread). virtual void close() noexcept override{ + AsyncTask thread; { - std::lock_guard lg1(m_lock); - m_state.store(State::DESTRUCTING, std::memory_order_relaxed); + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) == State::NOT_RUNNING && !m_thread){ + // Never connected, or already fully closed. + return; + } + m_state.store(State::DESTRUCTING, std::memory_order_release); + m_cv.notify_all(); + thread = std::move(m_thread); + } + + thread.wait_and_ignore_exceptions(); + + { + std::lock_guard lg(m_lock); + close_socket_locked(); + m_state.store(State::NOT_RUNNING, std::memory_order_release); m_cv.notify_all(); } } virtual void connect(const std::string& address, uint16_t port) override{ - std::lock_guard lg1(m_lock); - if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING){ + std::string error; + { + std::lock_guard lg(m_lock); + if (m_state.load(std::memory_order_relaxed) != State::NOT_RUNNING){ + return; + } + + // A socket whose previous connection has ended cannot be + // reconnected, so always start from a fresh socket. + close_socket_locked(); + + int socket = ::socket(AF_INET, SOCK_STREAM, 0); + if (socket == -1){ + error = "socket() failed: " + std::string(strerror(errno)); + }else{ + int flags = fcntl(socket, F_GETFL, 0); + if (flags == -1 || fcntl(socket, F_SETFL, flags | O_NONBLOCK) == -1){ + error = "fcntl() failed: " + std::string(strerror(errno)); + ::close(socket); + }else{ + m_socket.store(socket, std::memory_order_release); + } + } + if (!error.empty()){ + m_error = error; + return; + } + + m_state.store(State::CONNECTING, std::memory_order_relaxed); + } + if (!error.empty()){ + m_listeners.run_method(&Listener::on_connect_finished, error); return; } + try{ - m_state.store(State::CONNECTING, std::memory_order_relaxed); m_thread = m_thread_pool.dispatch_now_blocking([=, this]{ thread_loop(address, port); }); }catch (...){ - m_state.store(State::NOT_RUNNING, std::memory_order_relaxed); + { + std::lock_guard lg(m_lock); + m_error = "Failed to start the receive thread."; + close_socket_locked(); + m_state.store(State::NOT_RUNNING, std::memory_order_relaxed); + } + m_listeners.run_method(&Listener::on_connect_finished, m_error); throw; } } virtual size_t send(const void* data, size_t bytes) override{ - if (m_socket == -1){ - return 0; - } - - constexpr int BLOCK_SIZE = (int)1 << 30; const char* ptr = (const char*)data; size_t sent = 0; while (bytes > 0 && state() == State::CONNECTED){ + int socket = m_socket.load(std::memory_order_acquire); + if (socket == -1){ + return sent; + } + + constexpr int BLOCK_SIZE = (int)1 << 30; size_t current = std::min(bytes, BLOCK_SIZE); - ssize_t current_sent = ::send(m_socket, ptr, (int)current, MSG_DONTWAIT | MSG_NOSIGNAL); + ssize_t current_sent = ::send(socket, ptr, (int)current, MSG_DONTWAIT | MSG_NOSIGNAL); if (current_sent != -1){ sent += current_sent; if ((size_t)current_sent < current){ @@ -94,13 +146,14 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ continue; } + int error = errno; +// cout << "error = " << error << endl; + std::unique_lock lg(m_lock); if (state() == State::DESTRUCTING){ break; } - int error = errno; -// cout << "error = " << error << endl; switch (error){ case EAGAIN: #if EAGAIN != EWOULDBLOCK @@ -124,18 +177,41 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ private: - void close_socket(){ - if (m_socket == -1){ + // Closes the underlying socket, if any. Requires "m_lock" to be held. + void close_socket_locked(){ + int socket = m_socket.exchange(-1, std::memory_order_acquire); + if (socket == -1){ return; } - int ret = ::close(m_socket); - if (ret != 0){ + if (::close(socket) != 0){ try{ std::cout << "Failed close(): errno = " << errno << " (" << strerror(errno) << ")" << std::endl; }catch (...){} } } + // A connection ended on its own (the connect attempt failed, the server + // closed the connection, or a socket error occurred). This returns the + // socket to State::NOT_RUNNING so "connect()" can be called again, and + // fires the appropriate callback outside of "m_lock". + // Requires "lg" to be locked. It unlocks it. + void connection_lost_locked( + std::unique_lock& lg, + const std::string& error_message, + bool was_connected + ){ + m_error = error_message; + m_state.store(State::NOT_RUNNING, std::memory_order_release); + close_socket_locked(); + lg.unlock(); + + if (was_connected){ + m_listeners.run_method(&Listener::on_disconnect, error_message); + }else{ + m_listeners.run_method(&Listener::on_connect_finished, error_message); + } + } + void thread_loop(std::string address, uint16_t port){ try{ thread_loop_internal(std::move(address), port); @@ -159,51 +235,74 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ server.sin_addr.s_addr = inet_addr(address.c_str()); // Connect + int socket = m_socket.load(std::memory_order_acquire); + if (socket == -1){ + return; + } + + bool in_progress = false; while (true){ State state = m_state.load(std::memory_order_relaxed); if (state == State::DESTRUCTING){ return; } - if (::connect(m_socket, (struct sockaddr*)&server, sizeof(server)) != -1){ - break; + if (!in_progress){ + int ret = ::connect(socket, (struct sockaddr*)&server, sizeof(server)); + if (ret == 0){ + break; + } + + int error = errno; +// cout << "error = " << error << endl; + if (error != EAGAIN && error != EALREADY && error != EINPROGRESS){ + if (error == ETIMEDOUT){ + connection_lost_locked(lg, "Connection timed out.", false); + }else{ + connection_lost_locked(lg, "Connect error (errno " + std::to_string(error) + "): " + std::string(strerror(error)), false); + } + return; + } + in_progress = true; } - int error = errno; -// cout << "error = " << error << endl; + // While the connect is in progress, calling connect() again + // returns EISCONN, so instead wait for the socket to become + // writable, then check SO_ERROR for the result. + pollfd pf; + pf.fd = socket; + pf.events = POLLOUT; + pf.revents = 0; + int pret = poll(&pf, 1, 10); + if (pret == -1){ + if (errno != EINTR){ + connection_lost_locked(lg, "poll() error: " + std::string(strerror(errno)), false); + return; + } + continue; + } + if (pret == 0){ + continue; + } - switch (error){ - case EISCONN: - goto Connected; - case EAGAIN: - case EALREADY: - case EINPROGRESS: + int so_error = 0; + socklen_t slen = sizeof(so_error); + if (getsockopt(socket, SOL_SOCKET, SO_ERROR, &so_error, &slen) != 0){ + so_error = errno; + } + if (so_error == 0){ break; - case ETIMEDOUT: - m_state.store(State::NOT_RUNNING, std::memory_order_relaxed); - m_error = "Connection timed out."; - lg.unlock(); - m_listeners.run_method(&Listener::on_connect_finished, m_error); - return; - case ECONNREFUSED: - m_state.store(State::NOT_RUNNING, std::memory_order_relaxed); - m_error = "Connection refused."; - lg.unlock(); - m_listeners.run_method(&Listener::on_connect_finished, m_error); - return; - default: - m_state.store(State::NOT_RUNNING, std::memory_order_relaxed); - m_error = "Connect error (errno " + std::to_string(error) + "): " + std::string(strerror(error)); - lg.unlock(); - m_listeners.run_method(&Listener::on_connect_finished, m_error); - return; } - - m_cv.wait_for(lg, std::chrono::milliseconds(10)); - + if (so_error == ETIMEDOUT){ + connection_lost_locked(lg, "Connection timed out.", false); + }else if (so_error == ECONNREFUSED){ + connection_lost_locked(lg, "Connection refused.", false); + }else{ + connection_lost_locked(lg, "Connect error (errno " + std::to_string(so_error) + "): " + std::string(strerror(so_error)), false); + } + return; } -Connected: m_state.store(State::CONNECTED, std::memory_order_relaxed); } @@ -222,7 +321,13 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ if (state == State::DESTRUCTING){ return; } - bytes = ::recv(m_socket, buffer, BUFFER_SIZE, 0); + + int socket = m_socket.load(std::memory_order_acquire); + if (socket == -1){ + return; + } + + bytes = ::recv(socket, buffer, BUFFER_SIZE, 0); // cout << "error = " << error << endl; if (bytes == -1){ error = errno; @@ -239,7 +344,11 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ return; } -// cout << "error = " << error << endl; + if (bytes == 0){ + connection_lost_locked(lg, "Connection closed by server.", true); + return; + } + if (bytes == -1){ switch (error){ case EAGAIN: @@ -247,13 +356,8 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ case EWOULDBLOCK: #endif break; - case ECONNRESET: - case ENOTCONN: - case EPIPE: - m_error = "Receive error (errno " + std::to_string(error) + "): " + std::string(strerror(error)); - return; default: - m_error = "Receive error (errno " + std::to_string(error) + "): " + std::string(strerror(error)); + connection_lost_locked(lg, "Receive error (errno " + std::to_string(error) + "): " + std::string(strerror(error)), true); return; } } @@ -266,7 +370,7 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ private: ThreadPool& m_thread_pool; - const int m_socket; + std::atomic m_socket{-1}; std::string m_error; From 51484f355b4037473c86a4bcdba1b994b1e6ca50 Mon Sep 17 00:00:00 2001 From: PA Dev Bot <344001113+pa-devbot@users.noreply.github.com> Date: Sat, 31 Oct 2026 11:53:54 +0000 Subject: [PATCH 2/2] Sockets: start the receive thread with dispatch() Now that dispatch() spawns threads as needed, the receive thread no longer needs dispatch_now_blocking(). The blocking variant can also deadlock if connect() is called from a thread that is itself running on a capped pool, since it blocks until a pool thread is free. --- Common/Cpp/Sockets/ClientSocket_POSIX.h | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/Common/Cpp/Sockets/ClientSocket_POSIX.h b/Common/Cpp/Sockets/ClientSocket_POSIX.h index eadf4d8f01..36155343b6 100644 --- a/Common/Cpp/Sockets/ClientSocket_POSIX.h +++ b/Common/Cpp/Sockets/ClientSocket_POSIX.h @@ -106,8 +106,13 @@ class ClientSocket_POSIX final : public AbstractClientSocket{ return; } + // Use the plain dispatch() rather than dispatch_now_blocking(): + // connect() 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 receive thread. try{ - m_thread = m_thread_pool.dispatch_now_blocking([=, this]{ + m_thread = m_thread_pool.dispatch([=, this]{ thread_loop(address, port); }); }catch (...){