From 8a5559cc7c992c1bd3872ebf797f30c1cc6d4077 Mon Sep 17 00:00:00 2001 From: Matthew Stanley <1379tech@gmail.com> Date: Wed, 14 Oct 2026 11:29:13 -0700 Subject: [PATCH] Cancel debug socket waits portably without timing out valid fragmented commands --- runtime/CMakeLists.txt | 10 ++ runtime/src/debug_server.c | 74 ++++++------- runtime/src/debug_socket_io.h | 105 ++++++++++++++++++ runtime/tests/test_debug_socket_shutdown.cpp | 108 +++++++++++++++++++ 4 files changed, 254 insertions(+), 43 deletions(-) create mode 100644 runtime/src/debug_socket_io.h create mode 100644 runtime/tests/test_debug_socket_shutdown.cpp diff --git a/runtime/CMakeLists.txt b/runtime/CMakeLists.txt index 28494162..0a58eb55 100644 --- a/runtime/CMakeLists.txt +++ b/runtime/CMakeLists.txt @@ -23,6 +23,16 @@ psxrecomp_add_runtime_target(psx-runtime # (docs/ENHANCEMENTS.md G1.10). Titles get their _pgxp the same way. if(BUILD_TESTING) + add_executable(debug_socket_shutdown_test tests/test_debug_socket_shutdown.cpp) + target_include_directories(debug_socket_shutdown_test PRIVATE src) + find_package(Threads REQUIRED) + target_link_libraries(debug_socket_shutdown_test PRIVATE Threads::Threads) + if(WIN32) + target_link_libraries(debug_socket_shutdown_test PRIVATE ws2_32) + endif() + add_test(NAME debug_socket_shutdown_test COMMAND debug_socket_shutdown_test) + set_tests_properties(debug_socket_shutdown_test PROPERTIES TIMEOUT 15) + add_executable(video_texture_filter_test tests/test_video_texture_filter.c src/video_texture_filter.c) target_include_directories(video_texture_filter_test PRIVATE include) add_test(NAME video_texture_filter_test COMMAND video_texture_filter_test) diff --git a/runtime/src/debug_server.c b/runtime/src/debug_server.c index 7d230cdf..b8028672 100644 --- a/runtime/src/debug_server.c +++ b/runtime/src/debug_server.c @@ -113,6 +113,7 @@ #endif #include "psx_sdl.h" +#include "debug_socket_io.h" /* ---- Externs from runtime ---- */ extern uint32_t i_stat; @@ -148,7 +149,15 @@ static SDL_mutex *s_io_mutex = NULL; static SDL_cond *s_io_req_cv = NULL; /* emu waits on this for a request */ static SDL_cond *s_io_resp_cv = NULL; /* I/O waits on this for a response */ static volatile int s_io_state = IO_IDLE; -static volatile int s_io_running = 0; +#if defined(PSX_SDL3) +static SDL_AtomicInt s_io_running; +static int io_is_running(void *ctx) { (void)ctx; return SDL_GetAtomicInt(&s_io_running); } +static void io_set_running(int on) { SDL_SetAtomicInt(&s_io_running, on); } +#else +static SDL_atomic_t s_io_running; +static int io_is_running(void *ctx) { (void)ctx; return SDL_AtomicGet(&s_io_running); } +static void io_set_running(int on) { SDL_AtomicSet(&s_io_running, on); } +#endif static char s_io_req[RECV_BUF_SIZE]; /* request line (I/O -> emu) */ static char *s_resp_buf = NULL; /* response bytes (emu -> I/O) */ static size_t s_resp_len = 0, s_resp_cap = 0; @@ -16479,28 +16488,19 @@ void debug_server_init(int port) * 16 leaves room for a few probes to queue while we investigate. This * is observability infrastructure, not a freeze fix. */ listen(s_listen, 16); - /* Bounded accept(): close() of a socket blocked in accept() does NOT - * reliably wake the sleeper on Linux (the fd number can even be recycled - * under us), so the close-to-break-accept + SDL_WaitThread dance in - * debug_server_shutdown() could hang window-close forever. A receive - * timeout makes accept() return EWOULDBLOCK/EINTR periodically; the loop - * re-checks s_io_running, so shutdown joins within ~200 ms. */ -#ifdef _WIN32 - { DWORD accept_to = 200; - setsockopt(s_listen, SOL_SOCKET, SO_RCVTIMEO, (const char *)&accept_to, - sizeof(accept_to)); } -#else - { struct timeval accept_to = { 0, 200 * 1000 }; - setsockopt(s_listen, SOL_SOCKET, SO_RCVTIMEO, &accept_to, - sizeof(accept_to)); } -#endif + if (!psx_debug_socket_nonblocking(s_listen, 1)) { + s_listen_err = sock_error(); + sock_close(s_listen); + s_listen = SOCK_INVALID; + return; + } fprintf(stdout, "psxrecomp: debug server LISTENING on 127.0.0.1:%d\n", s_port); /* Always-on wall-time phase sampler (phase_profile). Own thread: the * server pumps on the main thread, so sampling must not live there. */ phase_sampler_start(); - /* Blocking accept — the dedicated I/O thread waits on it (no busy-poll). */ + /* The I/O worker waits for readiness with bounded cancellation. */ /* Start the TCP I/O thread + its handoff primitives. */ s_io_mutex = SDL_CreateMutex(); @@ -16509,9 +16509,9 @@ void debug_server_init(int port) s_resp_cap = 65536; s_resp_buf = (char *)malloc(s_resp_cap); if (s_io_mutex && s_io_req_cv && s_io_resp_cv && s_resp_buf) { - s_io_running = 1; + io_set_running(1); s_io_thread = SDL_CreateThread(io_thread_main, "psx-dbg-io", NULL); - if (!s_io_thread) s_io_running = 0; + if (!s_io_thread) io_set_running(0); } if (!s_frame_history) { @@ -16851,17 +16851,7 @@ void debug_server_init(int port) * line length (NUL-terminated, newline stripped), or -1 on close/error. */ static int recv_line(sock_t c, char *buf, int cap) { - int len = 0; - while (len < cap - 1) { - int n = recv(c, buf + len, cap - 1 - len, 0); - if (n <= 0) return -1; - len += n; - buf[len] = '\0'; - char *nl = strchr(buf, '\n'); - if (nl) { *nl = '\0'; return (int)(nl - buf); } - } - buf[cap - 1] = '\0'; - return cap - 1; /* over-long line: take what we have */ + return psx_debug_socket_recv_line(c, buf, cap, io_is_running, NULL); } /* The TCP I/O thread: owns accept/recv/send. Hands each request to the emu @@ -16871,11 +16861,9 @@ static int recv_line(sock_t c, char *buf, int cap) static int io_thread_main(void *arg) { (void)arg; - while (s_io_running) { - struct sockaddr_in caddr; - int clen = sizeof(caddr); - sock_t c = accept(s_listen, (struct sockaddr *)&caddr, &clen); - if (c == SOCK_INVALID) { if (!s_io_running) break; SDL_Delay(5); continue; } + while (io_is_running(NULL)) { + sock_t c = psx_debug_socket_accept(s_listen, io_is_running, NULL); + if (c == SOCK_INVALID) { if (!io_is_running(NULL)) break; SDL_Delay(5); continue; } char req[RECV_BUF_SIZE]; int rl = recv_line(c, req, sizeof(req)); @@ -16897,7 +16885,7 @@ static int io_thread_main(void *arg) s_io_state = IO_REQ; SDL_CondSignal(s_io_req_cv); int waited = 0; - while (s_io_state != IO_RESP && s_io_running) { + while (s_io_state != IO_RESP && io_is_running(NULL)) { if (SDL_CondWaitTimeout(s_io_resp_cv, s_io_mutex, 1000) == SDL_MUTEX_TIMEDOUT) { waited += 1000; if (waited >= 30000) break; /* emu wedged: give up on this one */ @@ -17078,18 +17066,18 @@ void debug_server_check_watchpoints(void) void debug_server_shutdown(void) { - /* Stop the I/O thread: clear the flag, close the listen socket to break the - * blocking accept(), then join. */ - s_io_running = 0; - if (s_listen != SOCK_INVALID) { - sock_close(s_listen); - s_listen = SOCK_INVALID; - } + /* Join before closing: the worker keeps sole ownership of its sockets + * while its cancellable waits are active. */ + io_set_running(0); if (s_io_thread) { if (s_io_mutex) { SDL_LockMutex(s_io_mutex); SDL_CondSignal(s_io_resp_cv); SDL_UnlockMutex(s_io_mutex); } SDL_WaitThread(s_io_thread, NULL); s_io_thread = NULL; } + if (s_listen != SOCK_INVALID) { + sock_close(s_listen); + s_listen = SOCK_INVALID; + } if (s_client != SOCK_INVALID) { sock_close(s_client); s_client = SOCK_INVALID; diff --git a/runtime/src/debug_socket_io.h b/runtime/src/debug_socket_io.h new file mode 100644 index 00000000..d2204f9e --- /dev/null +++ b/runtime/src/debug_socket_io.h @@ -0,0 +1,105 @@ +#ifndef PSX_DEBUG_SOCKET_IO_H +#define PSX_DEBUG_SOCKET_IO_H + +/* The worker owns its sockets until it exits. Readiness waits are cancellable; + * they do not impose a deadline on a client sending a fragmented request. */ +#ifdef _WIN32 +#include +typedef SOCKET PsxDebugSocket; +#define PSX_DEBUG_SOCKET_INVALID INVALID_SOCKET +#define psx_debug_socket_close closesocket +#else +#include +#include +#include +#include +#include +#include +typedef int PsxDebugSocket; +#define PSX_DEBUG_SOCKET_INVALID (-1) +#define psx_debug_socket_close close +#endif +#include + +typedef int (*PsxDebugSocketRunning)(void *ctx); + +static inline int psx_debug_socket_nonblocking(PsxDebugSocket s, int on) { +#ifdef _WIN32 + u_long mode = on ? 1 : 0; + return ioctlsocket(s, FIONBIO, &mode) == 0; +#else + int flags = fcntl(s, F_GETFL, 0); + return flags >= 0 && fcntl(s, F_SETFL, + on ? (flags | O_NONBLOCK) : (flags & ~O_NONBLOCK)) == 0; +#endif +} + +static inline int psx_debug_socket_retryable(void) { +#ifdef _WIN32 + int e = WSAGetLastError(); + return e == WSAEWOULDBLOCK || e == WSAEINTR; +#else + return errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR; +#endif +} + +static inline int psx_debug_socket_readable(PsxDebugSocket s) { + fd_set readfds; + struct timeval wait = {0, 200000}; +#ifndef _WIN32 + if (s < 0 || s >= FD_SETSIZE) return -1; +#endif + FD_ZERO(&readfds); + FD_SET(s, &readfds); +#ifdef _WIN32 + int ready = select(0, &readfds, NULL, NULL, &wait); +#else + int ready = select(s + 1, &readfds, NULL, NULL, &wait); +#endif + return ready < 0 && psx_debug_socket_retryable() ? 0 : ready; +} + +static inline PsxDebugSocket psx_debug_socket_accept( + PsxDebugSocket listener, PsxDebugSocketRunning running, void *ctx) { + while (running(ctx)) { + int ready = psx_debug_socket_readable(listener); + if (ready < 0) return PSX_DEBUG_SOCKET_INVALID; + if (!ready || !running(ctx)) continue; + /* The listener is nonblocking: an aborted connection between select + * and accept must not strand the worker in a blocking accept. */ + PsxDebugSocket client = accept(listener, NULL, NULL); + if (client == PSX_DEBUG_SOCKET_INVALID) { + if (psx_debug_socket_retryable()) continue; + return PSX_DEBUG_SOCKET_INVALID; + } + /* Winsock inherits FIONBIO. Request reads retain the existing blocking + * connection semantics, guarded by the cancellable readiness wait. */ + if (!psx_debug_socket_nonblocking(client, 0)) { + psx_debug_socket_close(client); + return PSX_DEBUG_SOCKET_INVALID; + } + return client; + } + return PSX_DEBUG_SOCKET_INVALID; +} + +static inline int psx_debug_socket_recv_line(PsxDebugSocket client, char *buf, + int cap, PsxDebugSocketRunning running, void *ctx) { + int len = 0; + while (len < cap - 1 && running(ctx)) { + int ready = psx_debug_socket_readable(client); + if (ready < 0) return -1; + if (!ready || !running(ctx)) continue; + int n = recv(client, buf + len, cap - 1 - len, 0); + if (n < 0 && psx_debug_socket_retryable()) continue; + if (n <= 0) return -1; + len += n; + buf[len] = '\0'; + char *nl = strchr(buf, '\n'); + if (nl) { *nl = '\0'; return (int)(nl - buf); } + } + if (!running(ctx)) return -1; + buf[cap - 1] = '\0'; + return cap - 1; +} +#endif diff --git a/runtime/tests/test_debug_socket_shutdown.cpp b/runtime/tests/test_debug_socket_shutdown.cpp new file mode 100644 index 00000000..2f40e079 --- /dev/null +++ b/runtime/tests/test_debug_socket_shutdown.cpp @@ -0,0 +1,108 @@ +#include "debug_socket_io.h" +#ifdef NDEBUG +#undef NDEBUG +#endif +#include +#include +#include +#include +#include +#include +#include + +using namespace std::chrono_literals; +static int running(void *ctx) { return static_cast *>(ctx)->load(); } +static PsxDebugSocket listener(unsigned short *port) { + auto s = socket(AF_INET, SOCK_STREAM, 0); + assert(s != PSX_DEBUG_SOCKET_INVALID); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + assert(bind(s, reinterpret_cast(&addr), sizeof addr) == 0); + assert(listen(s, 16) == 0); + assert(psx_debug_socket_nonblocking(s, 1)); +#ifdef _WIN32 + int size = sizeof addr; +#else + socklen_t size = sizeof addr; +#endif + assert(getsockname(s, reinterpret_cast(&addr), &size) == 0); + *port = ntohs(addr.sin_port); + return s; +} +static PsxDebugSocket connect_to(unsigned short port) { + auto s = socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = htons(port); + assert(connect(s, reinterpret_cast(&addr), sizeof addr) == 0); + return s; +} +int main() { +#ifdef _WIN32 + WSADATA wsa; + assert(WSAStartup(MAKEWORD(2, 2), &wsa) == 0); +#endif + for (int cycle = 0; cycle < 3; ++cycle) { + unsigned short port; + auto ls = listener(&port); + std::atomic live{1}; + auto pending = std::async(std::launch::async, [&] { + return psx_debug_socket_accept(ls, running, &live); + }); + assert(pending.wait_for(350ms) == std::future_status::timeout); + live = 0; + assert(pending.wait_for(1500ms) == std::future_status::ready); + assert(pending.get() == PSX_DEBUG_SOCKET_INVALID); + psx_debug_socket_close(ls); + } + for (int partial = 0; partial < 2; ++partial) { + unsigned short port; + auto ls = listener(&port); + std::atomic live{1}; + auto peer = connect_to(port); + auto worker = std::async(std::launch::async, [&] { + auto client = psx_debug_socket_accept(ls, running, &live); + assert(client != PSX_DEBUG_SOCKET_INVALID); + char buf[128]; + int n = psx_debug_socket_recv_line(client, buf, sizeof buf, running, &live); + psx_debug_socket_close(client); + return n; + }); + if (partial) assert(send(peer, "{\"cmd\":", 7, 0) == 7); + assert(worker.wait_for(350ms) == std::future_status::timeout); + live = 0; + assert(worker.wait_for(1500ms) == std::future_status::ready); + assert(worker.get() == -1); + psx_debug_socket_close(peer); + psx_debug_socket_close(ls); + } + unsigned short port; + auto ls = listener(&port); + std::atomic live{1}; + auto peer = connect_to(port); + auto worker = std::async(std::launch::async, [&] { + auto client = psx_debug_socket_accept(ls, running, &live); + char buf[128]; + int n = psx_debug_socket_recv_line(client, buf, sizeof buf, running, &live); + assert(n == 14 && std::string(buf) == "{\"cmd\":\"ping\"}"); + assert(send(client, "pong\n", 5, 0) == 5); + psx_debug_socket_close(client); + }); + std::this_thread::sleep_for(350ms); + assert(send(peer, "{\"cmd\":", 7, 0) == 7); + std::this_thread::sleep_for(350ms); + assert(send(peer, "\"ping\"}\n", 8, 0) == 8); + assert(worker.wait_for(1500ms) == std::future_status::ready); + worker.get(); + char pong[8]{}; + assert(recv(peer, pong, sizeof pong, 0) == 5); + assert(std::string(pong) == "pong\n"); + psx_debug_socket_close(peer); + psx_debug_socket_close(ls); +#ifdef _WIN32 + WSACleanup(); +#endif + puts("debug socket: idle/partial shutdown, delayed fragments, response and repeated initialization PASS"); +}