Cancel debug socket waits portably without timing out valid fragmented commands

runtime/CMakeLists.txt

@@ -23,6 +23,16 @@ psxrecomp_add_runtime_target(psx-runtime
 # (docs/ENHANCEMENTS.md G1.10). Titles get their <exe>_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)

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;

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 <winsock2.h>
+typedef SOCKET PsxDebugSocket;
+#define PSX_DEBUG_SOCKET_INVALID INVALID_SOCKET
+#define psx_debug_socket_close closesocket
+#else
+#include <sys/socket.h>
+#include <sys/select.h>
+#include <netinet/in.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <errno.h>
+typedef int PsxDebugSocket;
+#define PSX_DEBUG_SOCKET_INVALID (-1)
+#define psx_debug_socket_close close
+#endif
+#include <string.h>
+
+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

runtime/tests/test_debug_socket_shutdown.cpp

@@ -0,0 +1,108 @@
+#include "debug_socket_io.h"
+#ifdef NDEBUG
+#undef NDEBUG
+#endif
+#include <atomic>
+#include <cassert>
+#include <chrono>
+#include <future>
+#include <thread>
+#include <cstdio>
+#include <string>
+
+using namespace std::chrono_literals;
+static int running(void *ctx) { return static_cast<std::atomic<int> *>(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<sockaddr *>(&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<sockaddr *>(&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<sockaddr *>(&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<int> 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<int> 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<int> 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");
+}