From 9583c5b3785599979658cdf226f8d341e773eb38 Mon Sep 17 00:00:00 2001 From: Hydra Engineering Date: Sat, 29 Aug 2026 15:14:06 +0700 Subject: [PATCH 1/3] fix(rpc): retry EAGAIN/EWOULDBLOCK in state-stream recv paths (#713) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously, recv() returning EAGAIN on a non-blocking socket was treated identically to EOF or a hard error. During an M2 state-stream restore, this caused a short buffer to be passed to state_seq_set_data, which threw 'unexpectedly reached end of buffer' and silently left a corrupt KV pool. Extracted a shared inline helper hydra_recv_with_retry() in common/hydra-socket-retry.h that both call sites (llama_context:: refill and hydra_recv_all) now use: - Polls with short slices (1 s) up to a 30 s per-recv-call budget (not per-transfer). - EAGAIN/EWOULDBLOCK triggers poll+retry instead of failing. - EINTR is retried on both poll() and recv() syscalls. - On POLLHUP/POLLERR, performs a final recv() to drain buffered data before declaring EOF — prevents discarding up to 800 MB of a legitimate STATE_PUT transfer when the peer writes final bytes then closes. - True EOF (recv returns 0) and hard errors remain terminal. Rewrote test-hydra-recv-eagainst to include and test the real header function directly (no copy-paste mirror). Added deterministic Case E (64 KB payload + immediate close, verifying all bytes received before EOF) and Case F (POLLHUP+POLLIN set together on small payload). --- common/hydra-socket-retry.h | 138 +++++++++++++++++++ src/llama-context.cpp | 10 +- tests/CMakeLists.txt | 8 ++ tests/test-hydra-recv-eagain.cpp | 226 +++++++++++++++++++++++++++++++ tools/server/server-context.cpp | 21 ++- 5 files changed, 395 insertions(+), 8 deletions(-) create mode 100644 common/hydra-socket-retry.h create mode 100644 tests/test-hydra-recv-eagain.cpp diff --git a/common/hydra-socket-retry.h b/common/hydra-socket-retry.h new file mode 100644 index 00000000000..fd5ffe91fd9 --- /dev/null +++ b/common/hydra-socket-retry.h @@ -0,0 +1,138 @@ +// hydra#713: shared recv-with-EAGAIN-retry helper for non-blocking sockets. +// +// Both the M2 state-stream path (llama_io_read_socket::refill in +// llama-context.cpp) and the RPC framing path (hydra_recv_all in +// server-context.cpp) need identical EAGAIN/EWOULDBLOCK handling: poll + retry +// with a bounded deadline, drain on POLLHUP/POLLERR before declaring EOF, and +// EINTR retry on every syscall. A single copy here avoids silent drift +// between the two call sites. +// +// Header-only, POSIX only (`#if !defined(_WIN32)`). +// +// timeout_ms is a per-recv-call budget (not a per-transfer budget): +// each call to hydra_recv_with_retry waits at most timeout_ms for the +// requested n bytes. The caller's outer loop (hydra_recv_all, refill) +// may invoke this repeatedly for large transfers. + +#ifndef LLAMA_HYDRA_SOCKET_RETRY_H +#define LLAMA_HYDRA_SOCKET_RETRY_H + +#if !defined(_WIN32) + +#include +#include +#include +#include +#include + +#include +#include +#include + +// Attempt a non-blocking recv with bounded poll-retry on EAGAIN/EWOULDBLOCK. +// +// Returns: +// >0 — bytes read (always == n on success, caller loop handles short reads) +// 0 — clean EOF (peer closed after draining any buffered data) +// -1 — hard error or timeout; errno is set: +// ETIMEDOUT — timeout_ms elapsed with no data +// ECONNRESET / EPIPE / etc. — peer-level failure +// EBADF — bad fd +// +// On POLLHUP or POLLERR the helper performs one final recv to drain any +// data the peer wrote before closing. Only if that recv also returns 0 +// (EOF) or a hard error does the function return. This prevents +// discarding buffered data when the peer writes its final bytes then +// closes (a common pattern for large STATE_PUT transfers up to 800 MB). +inline ssize_t hydra_recv_with_retry(int fd, void * buf, size_t n, int timeout_ms) { + char * p = reinterpret_cast(buf); + + // Fast path: attempt recv immediately — no poll overhead for the common case. + ssize_t r = ::recv(fd, p, n, 0); + if (r > 0) { + return r; + } + if (r == 0) { + return 0; // clean EOF + } + // r < 0 — check errno before entering the retry loop. + if (errno != EAGAIN && errno != EWOULDBLOCK) { + return -1; // hard error (ECONNRESET, EBADF, …) + } + + // EAGAIN: poll + retry loop with a bounded wall-clock deadline. + // The deadline is relative to the FIRST EAGAIN, not to the original call, + // so the caller's per-call budget is respected. + const int64_t deadline_ms = + static_cast(timeout_ms) > 0 ? timeout_ms : 30000; + // We track elapsed time via poll slices rather than a clock to avoid + // clock-resolution issues on all platforms; the loop simply counts + // down `remaining_ms`. + int remaining_ms = deadline_ms; + + for (;;) { + if (remaining_ms <= 0) { + errno = ETIMEDOUT; + return -1; + } + const int wait_ms = std::min(remaining_ms, 1000); + struct pollfd pfd = { fd, POLLIN, 0 }; + int pr = ::poll(&pfd, 1, wait_ms); + if (pr < 0) { + if (errno == EINTR) { + // EINTR on poll: subtract the slice we waited and retry. + remaining_ms -= wait_ms; + continue; + } + return -1; // real poll error (EBADF, EINVAL, …) + } + if (pr == 0) { + // Timeout slice expired — subtract and loop. + remaining_ms -= wait_ms; + continue; + } + + // poll returned > 0: at least one event is ready. + + if (pfd.revents & (POLLHUP | POLLERR)) { + // On Linux, POLLIN is often set together with POLLHUP when the + // peer wrote final bytes then closed. We must drain buffered + // data before declaring EOF — otherwise up to 800 MB of a + // legitimate STATE_PUT transfer is silently discarded. + r = ::recv(fd, p, n, 0); + if (r > 0) { + return r; + } + if (r == 0) { + return 0; // true EOF after drain + } + // r < 0: recv error after HUP/ERR — propagate. + return -1; + } + + // POLLIN ready — attempt the actual recv. + r = ::recv(fd, p, n, 0); + if (r > 0) { + return r; + } + if (r == 0) { + return 0; // clean EOF + } + // r < 0 + if (errno == EAGAIN || errno == EWOULDBLOCK) { + // Still EAGAIN — subtract the poll slice and loop. + remaining_ms -= wait_ms; + continue; + } + if (errno == EINTR) { + // EINTR on recv: subtract the poll slice and retry (don't + // count this as a "no data" cycle against the deadline). + remaining_ms -= wait_ms; + continue; + } + return -1; // hard error (ECONNRESET, etc.) + } +} + +#endif // !_WIN32 +#endif // LLAMA_HYDRA_SOCKET_RETRY_H diff --git a/src/llama-context.cpp b/src/llama-context.cpp index 42cd3bf151f..185cdf40fff 100644 --- a/src/llama-context.cpp +++ b/src/llama-context.cpp @@ -3011,6 +3011,8 @@ size_t llama_context::state_get_size() { // hydra: zero-copy socket streaming (class stays here; C wrapper in llama-hydra.cpp) #if !defined(_WIN32) #include +// hydra#713: shared EAGAIN/EWOULDBLOCK retry helper (poll + drain-on-HUP). +#include "../common/hydra-socket-retry.h" // xxh3 for M2 decode-side wire-hash verification (see state_seq_set_data_from_fd) #include "../vendor/xxhash/xxhash.h" @@ -3143,9 +3145,13 @@ class llama_io_read_socket : public llama_io_read_i { if (staging_pos < staging_len) { return; } - ssize_t r = ::recv(fd, staging.data(), staging.size(), 0); + ssize_t r = hydra_recv_with_retry(fd, staging.data(), staging.size(), 30000); if (r <= 0) { - throw std::runtime_error("hydra: socket recv failed during state restore"); + if (r == 0) { + throw std::runtime_error("hydra: socket recv EOF during state restore"); + } + throw std::runtime_error(std::string("hydra: socket recv failed during state restore: ") + + std::strerror(errno)); } staging_pos = 0; staging_len = (size_t)r; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 9b955a1c77f..4ee845700d9 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -182,6 +182,14 @@ target_link_libraries(test-hydra-rpc-stale-sock PRIVATE ggml) # #470: test uses ggml_backend_buffer_copy_tensor (declared in ggml-backend-impl.h). target_include_directories(test-hydra-rpc-stale-sock PRIVATE ${PROJECT_SOURCE_DIR}/ggml/src) +# hydra#713: EAGAIN/EWOULDBLOCK retry in hydra_recv_all and +# llama_io_read_socket::refill(). Pure socket test — no model/GPU needed. +# Verifies that recv returning EAGAIN is retried with poll (not treated as +# EOF) and that true EOF / timeout still fail cleanly. +if (NOT WIN32) + llama_build_and_test(test-hydra-recv-eagain.cpp) +endif() + if (NOT WIN32 OR NOT BUILD_SHARED_LIBS) # these tests are disabled on Windows because they use internal functions not exported with LLAMA_API (when building with shared libraries) llama_build_and_test(test-sampling.cpp) diff --git a/tests/test-hydra-recv-eagain.cpp b/tests/test-hydra-recv-eagain.cpp new file mode 100644 index 00000000000..70bbc57d983 --- /dev/null +++ b/tests/test-hydra-recv-eagain.cpp @@ -0,0 +1,226 @@ +// hydra#713: EAGAIN/EWOULDBLOCK retry in hydra_recv_with_retry(). +// +// Bug: when recv() returns -1/EAGAIN mid-state-stream (non-blocking socket +// under backpressure), the old code treated it identically to EOF -> short +// buffer passed to state_seq_set_data -> "unexpectedly reached end of buffer" +// -> silent corrupt KV pool. +// +// Fix: hydra_recv_with_retry() (common/hydra-socket-retry.h) polls + retries +// on EAGAIN/EWOULDBLOCK with a bounded deadline, drains buffered data on +// POLLHUP before declaring EOF, and retries recv() on EINTR. +// +// This test exercises the REAL function directly (no copy-paste mirror). +// No model or GPU needed -- pure socket test. +#include "hydra-socket-retry.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +static int g_failures = 0; + +static void expect(const char * what, bool ok) { + if (!ok) { + fprintf(stderr, "FAIL: %s\n", what); + g_failures++; + } +} + +// Helper: read exactly n bytes using hydra_recv_with_retry in a loop. +// Returns true on success, false on error/timeout/EOF. +static bool recv_all(int fd, void * buf, size_t n) { + char * p = reinterpret_cast(buf); + while (n > 0) { + ssize_t r = hydra_recv_with_retry(fd, p, n, 30000); + if (r > 0) { + p += r; n -= (size_t)r; + } else { + return false; + } + } + return true; +} + +int main() { + fprintf(stderr, "test-hydra-recv-eagain: running\n"); + + // ── Case A: delayed write with EAGAIN retry. Writer sleeps 200 ms then + // sends; reader's first recv() returns EAGAIN, poll() blocks until + // POLLIN, then recv() succeeds. + { + int sv[2]; + expect("A: socketpair", socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); + int flags = fcntl(sv[0], F_GETFL, 0); + fcntl(sv[0], F_SETFL, flags | O_NONBLOCK); + + const char payload[] = "hello hydra#713"; + std::thread writer([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + ssize_t w = ::send(sv[1], payload, sizeof(payload), 0); + expect("A: send", w == (ssize_t)sizeof(payload)); + }); + + char buf[sizeof(payload)] = {}; + bool ok = recv_all(sv[0], buf, sizeof(payload)); + expect("A: recv retries past EAGAIN", ok); + expect("A: payload matches", ok && memcmp(buf, payload, sizeof(payload)) == 0); + + writer.join(); + ::close(sv[0]); + ::close(sv[1]); + } + + // ── Case B: deterministic fast-path (no EAGAIN). Data is written to the + // socket BEFORE the first recv() call, so the reader sees data + // immediately -- the common case for small payloads. This validates + // that the fast path (no poll overhead) works correctly. + { + int sv[2]; + expect("B: socketpair", socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); + + const char payload[] = "immediate-send"; + // Write BEFORE setting non-blocking -- data is in kernel buffer. + ssize_t w = ::send(sv[1], payload, sizeof(payload), 0); + expect("B: send", w == (ssize_t)sizeof(payload)); + + // Now set non-blocking and read -- should succeed on first recv(). + int flags = fcntl(sv[0], F_GETFL, 0); + fcntl(sv[0], F_SETFL, flags | O_NONBLOCK); + + char buf[sizeof(payload)] = {}; + bool ok = recv_all(sv[0], buf, sizeof(payload)); + expect("B: immediate data (fast path, no EAGAIN)", ok); + expect("B: payload matches", ok && memcmp(buf, payload, sizeof(payload)) == 0); + + ::close(sv[0]); + ::close(sv[1]); + } + + // ── Case C: EOF (writer closes without sending). recv must return false, + // not hang. + { + int sv[2]; + expect("C: socketpair", socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); + int flags = fcntl(sv[0], F_GETFL, 0); + fcntl(sv[0], F_SETFL, flags | O_NONBLOCK); + + ::close(sv[1]); // EOF + + char buf[8] = {}; + bool ok = recv_all(sv[0], buf, sizeof(buf)); + expect("C: EOF returns false (no hang)", !ok); + + ::close(sv[0]); + } + + // ── Case D: timeout on slow peer. Custom 100 ms timeout; writer delays + // 500 ms -- must time out cleanly. + { + int sv[2]; + expect("D: socketpair", socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); + int flags = fcntl(sv[0], F_GETFL, 0); + fcntl(sv[0], F_SETFL, flags | O_NONBLOCK); + + std::thread writer([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + ::send(sv[1], "late", 4, 0); + }); + + char buf[8] = {}; + ssize_t r = hydra_recv_with_retry(sv[0], buf, sizeof(buf), 100); + expect("D: slow peer -> timeout returns -1", r == -1); + expect("D: errno is ETIMEDOUT", errno == ETIMEDOUT); + + writer.join(); + ::close(sv[0]); + ::close(sv[1]); + } + + // ── Case E: POLLHUP with buffered data. Writer writes a large payload + // (64 KB -- larger than typical socket buffer to ensure some kernel + // buffering) then immediately close()s. On Linux, poll() returns + // POLLHUP|POLLIN when the peer closes with data still in the buffer. + // The reader MUST receive all bytes before seeing EOF -- not fail + // with a spurious error. This is the root-cause bug from the t2 + // review (discarding buffered data on legitimate STATE_PUT transfers + // up to 800 MB). + { + int sv[2]; + expect("E: socketpair", socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); + int flags = fcntl(sv[0], F_GETFL, 0); + fcntl(sv[0], F_SETFL, flags | O_NONBLOCK); + + // Build a 64 KB payload. + const size_t payload_sz = 64 * 1024; + std::vector payload(payload_sz); + for (size_t i = 0; i < payload_sz; i++) { + payload[i] = (char)(i & 0xFF); + } + + // Writer: send large payload then immediately close. + std::thread writer([&] { + ssize_t w = ::send(sv[1], payload.data(), payload_sz, 0); + // May be partial if kernel buffer is small; loop to be safe. + size_t sent = (w > 0) ? (size_t)w : 0; + while (sent < payload_sz) { + w = ::send(sv[1], payload.data() + sent, payload_sz - sent, 0); + if (w <= 0) break; + sent += (size_t)w; + } + ::close(sv[1]); + }); + + // Read all bytes then expect EOF. + std::vector buf(payload_sz); + bool ok = recv_all(sv[0], buf.data(), payload_sz); + expect("E: all bytes received before EOF", ok); + expect("E: payload matches", ok && memcmp(buf.data(), payload.data(), payload_sz) == 0); + + // After draining, the next recv should return 0 (EOF). + char junk[8]; + ssize_t r = hydra_recv_with_retry(sv[0], junk, sizeof(junk), 1000); + expect("E: EOF after full drain", r == 0); + + writer.join(); + ::close(sv[0]); + } + + // ── Case F: POLLHUP set together with POLLIN (the common Linux pattern). + // Write a small payload, close the writer, then read. poll() may + // return POLLIN|POLLHUP in a single call. The reader must get the + // data, not discard it. + { + int sv[2]; + expect("F: socketpair", socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); + int flags = fcntl(sv[0], F_GETFL, 0); + fcntl(sv[0], F_SETFL, flags | O_NONBLOCK); + + const char payload[] = "pollhup-with-data"; + // Write then close in same thread -- fast, no delay. + ::send(sv[1], payload, sizeof(payload), 0); + ::close(sv[1]); + + char buf[sizeof(payload)] = {}; + bool ok = recv_all(sv[0], buf, sizeof(payload)); + expect("F: data received despite POLLHUP", ok); + expect("F: payload matches", ok && memcmp(buf, payload, sizeof(payload)) == 0); + + ::close(sv[0]); + } + + if (g_failures == 0) { + fprintf(stderr, "test-hydra-recv-eagain: all checks passed\n"); + } else { + fprintf(stderr, "test-hydra-recv-eagain: %d check(s) FAILED\n", g_failures); + } + return g_failures; +} diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index f6c6fbc2b17..2db5e1b522b 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -48,8 +48,11 @@ # include # include # include +# include # include # include +// hydra#713: shared EAGAIN/EWOULDBLOCK retry helper (poll + drain-on-HUP). +# include "../../common/hydra-socket-retry.h" #endif // fix problem with std::min and std::max @@ -10333,21 +10336,27 @@ struct hydra_rpc_ctx { // `false` return as "give up" but none logged *why*, so a wedged RPC // response looked identical to a client that vanished. Log once, centrally, // instead of touching the ~30 call sites. +// +// hydra#713: EAGAIN/EWOULDBLOCK is retried via hydra_recv_with_retry() +// (common/hydra-socket-retry.h) with a 30 s per-call budget. True EOF +// and hard errors remain terminal. static bool hydra_recv_all(int fd, void * buf, size_t n) { char * p = reinterpret_cast(buf); const size_t total = n; while (n > 0) { - ssize_t r = ::recv(fd, p, n, 0); - if (r < 0) { - SRV_WRN("hydra rpc: recv failed on fd=%d (%zu/%zu bytes): %s\n", - fd, total - n, total, std::strerror(errno)); - return false; + ssize_t r = hydra_recv_with_retry(fd, p, n, 30000); + if (r > 0) { + p += r; n -= r; + continue; } if (r == 0) { SRV_DBG("hydra rpc: recv EOF on fd=%d (%zu/%zu bytes)\n", fd, total - n, total); return false; } - p += r; n -= r; + // r < 0: hard error or timeout. errno is set by hydra_recv_with_retry. + SRV_WRN("hydra rpc: recv failed on fd=%d (%zu/%zu bytes): %s\n", + fd, total - n, total, std::strerror(errno)); + return false; } return true; } From 75a1ceea31247f64a3a64e2a41a802b9e5943101 Mon Sep 17 00:00:00 2001 From: Hydra Engineering Date: Sat, 29 Aug 2026 17:32:48 +0700 Subject: [PATCH 2/3] fix(713): quarantine slot via prompt_clear on STATE_PUT zero-read Full slot cleanup (KV cells + tokens + logits) instead of manual partial clear, so PREFILL cannot run on a dirty KV pool after a failed restore. Loud SRV_ERR with slot id + state_len. Mirrors DECODE_APPLY cleanup path. Co-Authored-By: hydra-dev --- tools/server/server-context.cpp | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 2db5e1b522b..25b74b1ab48 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -3803,15 +3803,18 @@ struct server_context_impl { const size_t state_len = has_hdr ? buf.size() - hdr_offset : buf.size(); const size_t n_read = llama_state_seq_set_data(ctx_tgt, state_ptr, state_len, slot->id); if (n_read == 0) { + // #713 quarantine: restore failed (state_seq_set_data error + // / short stream). Clear the slot fully so PREFILL cannot + // run on a dirty KV pool — mirror the DECODE_APPLY cleanup + // (~line 5260). The RPC handler sends HYDRA_STATUS_ERROR + // back to the coordinator, which maps it to a retry-after-clean. + SRV_ERR("hydra: STATE_PUT slot=%d quarantine: restore failed " + "(state_len=%zu n_read=0) — clearing slot\n", + id_slot, state_len); res->rpc_status = HYDRA_STATUS_ERROR; - res->error = "llama_state_set_data returned 0"; - // Tokens were registered before set_data — clear them so the slot - // is not left poisoned (n_past > 0 with no KV cells → pos_min == -1 - // abort on the next decode that touches this slot). - slot->prompt.tokens.clear(); - slot->prompt.checkpoints.clear(); + res->error = "KV restore failed (llama_state_seq_set_data returned 0)"; + slot->prompt_clear(false); // clears KV cells + tokens + logits slot->n_prompt_tokens_cache = 0; - llama_memory_seq_rm(llama_get_memory(ctx_tgt), slot->id, -1, -1); } else { // D4: Inject trailing logits into per-slot buffer instead of the // shared context-wide llama_get_logits(). This avoids the race where From 5f28a97342259a9f1a13ee2785dfa9492e9fe420 Mon Sep 17 00:00:00 2001 From: Ddv Date: Wed, 2 Sep 2026 22:27:32 +0700 Subject: [PATCH 3/3] =?UTF-8?q?fix(713):=20review=20findings=20M1/M2=20+?= =?UTF-8?q?=20minors=20=E2=80=94=20pipelined=20EAGAIN,=20quarantine=20comp?= =?UTF-8?q?leteness,=20clock,=20errno,=20SHUT=5FRD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M1: src/llama-context.cpp pipelined read_tensor (dominant CUDA path) now uses hydra_recv_with_retry instead of raw ::recv, distinguishing EAGAIN/EWOULDBLOCK (retry) from EOF (0) and hard error (-1). M2: tools/server/server-context.cpp — extract hydra_quarantine_slot() helper to converge STATE_PUT zero-read and DECODE_APPLY status==0 cleanup. prompt_clear(false) alone misses checkpoints/just_restored/ n_prompt_tokens_cache/n_decoded, so helper restores completeness to prevent pos_min==-1/#641 class on next decode. Minors: - common/hydra-socket-retry.h: steady_clock deadline accounting (not slice-subtraction), so EINTR/spurious POLLIN don't prematurely burn the 30s budget; fixes type mismatch std::min. - tests/test-hydra-recv-eagain.cpp: errno capture immediately after hydra_recv_with_retry (NIT-10); Case G blocking+SO_RCVTIMEO; Case H EINTR storm with steady_clock verification. - tools/server/server-context.cpp: hydra_handle_state_put SHUT_RD drain on short-read, mirroring DECODE_APPLY; quarantine observability (n_checkpoints/just_restored) on STATE_META and hydra_state result. - src/llama-context.cpp: XXH_INLINE_ALL for vendor xxhash header-only inline (fixes libllama undefined XXH3_64bits_update). Co-Authored-By: hydra-dev --- common/hydra-socket-retry.h | 46 +++++++------ src/llama-context.cpp | 19 +++++- tests/test-hydra-recv-eagain.cpp | 113 ++++++++++++++++++++++++++++++- tools/server/server-context.cpp | 70 ++++++++++++++++--- tools/server/server-task.cpp | 3 + tools/server/server-task.h | 7 ++ 6 files changed, 222 insertions(+), 36 deletions(-) diff --git a/common/hydra-socket-retry.h b/common/hydra-socket-retry.h index fd5ffe91fd9..f1c69b9fb27 100644 --- a/common/hydra-socket-retry.h +++ b/common/hydra-socket-retry.h @@ -24,6 +24,7 @@ #include #include #include +#include #include #include @@ -32,7 +33,8 @@ // Attempt a non-blocking recv with bounded poll-retry on EAGAIN/EWOULDBLOCK. // // Returns: -// >0 — bytes read (always == n on success, caller loop handles short reads) +// >0 — bytes read (1..n). recv may return a short read (r < n), so the +// caller's loop must keep reading until n bytes are in or EOF/error. // 0 — clean EOF (peer closed after draining any buffered data) // -1 — hard error or timeout; errno is set: // ETIMEDOUT — timeout_ms elapsed with no data @@ -63,32 +65,38 @@ inline ssize_t hydra_recv_with_retry(int fd, void * buf, size_t n, int timeout_m // EAGAIN: poll + retry loop with a bounded wall-clock deadline. // The deadline is relative to the FIRST EAGAIN, not to the original call, // so the caller's per-call budget is respected. - const int64_t deadline_ms = - static_cast(timeout_ms) > 0 ? timeout_ms : 30000; - // We track elapsed time via poll slices rather than a clock to avoid - // clock-resolution issues on all platforms; the loop simply counts - // down `remaining_ms`. - int remaining_ms = deadline_ms; + // + // Elapsed time is measured with a monotonic clock (steady_clock), not by + // subtracting the poll slice each iteration. Slice-subtraction burns the + // whole budget on events that consumed no wall time (EINTR, a spurious + // POLLIN that still yields EAGAIN), which would time out a healthy + // transfer prematurely. Charging real elapsed time also makes the EINTR + // handling actually correct rather than merely bounded. + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(static_cast(timeout_ms > 0 ? timeout_ms : 30000)); for (;;) { - if (remaining_ms <= 0) { + const auto now = std::chrono::steady_clock::now(); + if (now >= deadline) { errno = ETIMEDOUT; return -1; } - const int wait_ms = std::min(remaining_ms, 1000); + const auto rem_ms = std::chrono::duration_cast(deadline - now).count(); + const int wait_ms = static_cast(std::min(rem_ms, 1000LL)); + struct pollfd pfd = { fd, POLLIN, 0 }; int pr = ::poll(&pfd, 1, wait_ms); if (pr < 0) { if (errno == EINTR) { - // EINTR on poll: subtract the slice we waited and retry. - remaining_ms -= wait_ms; + // Interrupted before any data: real elapsed time is already + // charged against the deadline at the top of the loop. continue; } return -1; // real poll error (EBADF, EINVAL, …) } if (pr == 0) { - // Timeout slice expired — subtract and loop. - remaining_ms -= wait_ms; + // Poll slice expired with no data: loop back — the deadline check + // accounts for the time that actually passed. continue; } @@ -119,15 +127,9 @@ inline ssize_t hydra_recv_with_retry(int fd, void * buf, size_t n, int timeout_m return 0; // clean EOF } // r < 0 - if (errno == EAGAIN || errno == EWOULDBLOCK) { - // Still EAGAIN — subtract the poll slice and loop. - remaining_ms -= wait_ms; - continue; - } - if (errno == EINTR) { - // EINTR on recv: subtract the poll slice and retry (don't - // count this as a "no data" cycle against the deadline). - remaining_ms -= wait_ms; + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { + // Still no data (spurious readiness / interrupted): retry within + // the deadline. Real elapsed time is charged at the top of the loop. continue; } return -1; // hard error (ECONNRESET, etc.) diff --git a/src/llama-context.cpp b/src/llama-context.cpp index 185cdf40fff..78b2f0bf6b6 100644 --- a/src/llama-context.cpp +++ b/src/llama-context.cpp @@ -3014,7 +3014,12 @@ size_t llama_context::state_get_size() { // hydra#713: shared EAGAIN/EWOULDBLOCK retry helper (poll + drain-on-HUP). #include "../common/hydra-socket-retry.h" // xxh3 for M2 decode-side wire-hash verification (see state_seq_set_data_from_fd) +// Use header-only inline variant so libllama does not need an external +// xxhash.c object — server-context.cpp already compiles the non-inline +// implementation for the server binary, but tests link only libllama. +#define XXH_INLINE_ALL #include "../vendor/xxhash/xxhash.h" +#undef XXH_INLINE_ALL class llama_io_write_socket : public llama_io_write_i { // hydra#334: chunk size is caller-supplied (see llama_cparams::hydra_state_chunk_size, @@ -3215,10 +3220,18 @@ class llama_io_read_socket : public llama_io_read_i { staging_pos += take; done += take; if (staging_pos >= staging_len && done < size) { - // recv the next chunk WHILE the H2D copy of the current one runs - ssize_t r = ::recv(fd, other.data(), other.size(), 0); + // recv the next chunk WHILE the H2D copy of the current one runs. + // hydra#713 review (finding 1): this pipelined branch is the + // dominant CUDA path (tensor_backend != nullptr) and carries the + // whole ~800 MB stream — it must retry EAGAIN/EWOULDBLOCK like + // refill() does, not treat the first -1 as stream-end. + ssize_t r = hydra_recv_with_retry(fd, other.data(), other.size(), 30000); if (r <= 0) { - throw std::runtime_error("hydra: socket recv failed during state restore"); + if (r == 0) { + throw std::runtime_error("hydra: socket recv EOF during state restore"); + } + throw std::runtime_error(std::string("hydra: socket recv failed during state restore: ") + + std::strerror(errno)); } other_len = (size_t)r; if (hash_state != nullptr) { diff --git a/tests/test-hydra-recv-eagain.cpp b/tests/test-hydra-recv-eagain.cpp index 70bbc57d983..ee6060971aa 100644 --- a/tests/test-hydra-recv-eagain.cpp +++ b/tests/test-hydra-recv-eagain.cpp @@ -23,11 +23,20 @@ #include #include +#include #include +#include #include +#include + static int g_failures = 0; +// Case H bookkeeping: tick counter for the periodic SIGALRM that interrupts +// the reader's poll() calls. Handler is async-signal-safe (counter bump only). +static volatile sig_atomic_t g_sigalrm_ticks = 0; +static void sigalrm_handler(int) { g_sigalrm_ticks++; } + static void expect(const char * what, bool ok) { if (!ok) { fprintf(stderr, "FAIL: %s\n", what); @@ -137,8 +146,11 @@ int main() { char buf[8] = {}; ssize_t r = hydra_recv_with_retry(sv[0], buf, sizeof(buf), 100); + // Capture errno IMMEDIATELY — expect()'s fprintf may clobber it + // (review NIT-10). + const int r_errno = errno; expect("D: slow peer -> timeout returns -1", r == -1); - expect("D: errno is ETIMEDOUT", errno == ETIMEDOUT); + expect("D: errno is ETIMEDOUT", r_errno == ETIMEDOUT); writer.join(); ::close(sv[0]); @@ -217,6 +229,105 @@ int main() { ::close(sv[0]); } + // ── Case G: production socket mode — BLOCKING socket + SO_RCVTIMEO + // (no O_NONBLOCK). The engine's RPC sockets (hydra_handle_connection) + // are exactly this: blocking with a 120 s SO_RCVTIMEO, where "no data + // yet" surfaces as -1/EAGAIN only after the kernel receive timeout + // elapses. All cases above use O_NONBLOCK; this one pins the + // blocking-mode contract: that EAGAIN must be retried within the + // budget, not treated as a hard error or stream-end. + { + int sv[2]; + expect("G: socketpair", socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); + // sv[0] stays BLOCKING; give it a short kernel receive timeout so the + // first recv() fails fast with EAGAIN instead of parking for 120 s. + struct timeval tv = { 0, 100 * 1000 }; // 100 ms + expect("G: setsockopt SO_RCVTIMEO", + setsockopt(sv[0], SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) == 0); + + const char payload[] = "blocking-mode payload"; + std::thread writer([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + ssize_t w = ::send(sv[1], payload, sizeof(payload), 0); + expect("G: send", w == (ssize_t)sizeof(payload)); + }); + + char buf[sizeof(payload)] = {}; + // Timeline: fast-path recv() blocks ~100 ms -> -1/EAGAIN; poll() then + // wakes when the writer's bytes land at ~250 ms; the next recv() returns + // the data. All within the 30 s per-call budget (1 s here via recv_all + // would also be fine — the point is the EAGAIN retry, not the timing). + bool ok = recv_all(sv[0], buf, sizeof(payload)); + expect("G: blocking+SO_RCVTIMEO EAGAIN retried, data received", ok); + expect("G: payload matches", ok && memcmp(buf, payload, sizeof(payload)) == 0); + + writer.join(); + ::close(sv[0]); + ::close(sv[1]); + } + + // ── Case H: EINTR storm — clock-based deadline accounting (finding 5). + // A periodic SIGALRM interrupts the reader's poll() every ~50 ms. + // The writer's data lands at ~800 ms, inside the budget. With the old + // slice-subtraction accounting each EINTR burned a full poll slice + // (up to 1 s) of budget, so a couple of interrupts timed the call out + // long before the data arrived. With steady_clock accounting the real + // elapsed time is charged and the read succeeds. + { + struct sigaction sa{}; + sa.sa_handler = sigalrm_handler; + sigemptyset(&sa.sa_mask); + expect("H: sigaction", sigaction(SIGALRM, &sa, nullptr) == 0); + + // Block SIGALRM in the writer thread so every tick lands on the + // reader (main) — the only thread in poll() — and deterministically + // exercises the EINTR path. + sigset_t alarm_set; + sigemptyset(&alarm_set); + sigaddset(&alarm_set, SIGALRM); + pthread_sigmask(SIG_UNBLOCK, &alarm_set, nullptr); // main: ensure unblocked + + struct itimerval itv{}; + itv.it_interval.tv_usec = 50 * 1000; // tick every 50 ms + itv.it_value.tv_usec = 20 * 1000; // first tick at 20 ms + expect("H: setitimer", setitimer(ITIMER_REAL, &itv, nullptr) == 0); + + int sv[2]; + expect("H: socketpair", socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); + int flags = fcntl(sv[0], F_GETFL, 0); + fcntl(sv[0], F_SETFL, flags | O_NONBLOCK); + + const char payload[] = "eintr storm survivor"; + g_sigalrm_ticks = 0; + std::thread writer([&] { + pthread_sigmask(SIG_BLOCK, &alarm_set, nullptr); + std::this_thread::sleep_for(std::chrono::milliseconds(800)); + ssize_t w = ::send(sv[1], payload, sizeof(payload), 0); + expect("H: send", w == (ssize_t)sizeof(payload)); + }); + + char buf[sizeof(payload)] = {}; + bool ok = recv_all(sv[0], buf, sizeof(payload)); + + // Disarm before asserting so a late tick cannot race the cleanup. + itv.it_interval.tv_usec = 0; + itv.it_value.tv_usec = 0; + setitimer(ITIMER_REAL, &itv, nullptr); + + expect("H: data received despite repeated poll EINTR", ok); + expect("H: payload matches", ok && memcmp(buf, payload, sizeof(payload)) == 0); + expect("H: EINTRs actually occurred", g_sigalrm_ticks > 0); + + writer.join(); + ::close(sv[0]); + ::close(sv[1]); + + struct sigaction sa_def{}; + sa_def.sa_handler = SIG_DFL; + sigemptyset(&sa_def.sa_mask); + sigaction(SIGALRM, &sa_def, nullptr); + } + if (g_failures == 0) { fprintf(stderr, "test-hydra-recv-eagain: all checks passed\n"); } else { diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 25b74b1ab48..f556e997171 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -673,6 +673,39 @@ struct server_slot { } }; +// hydra#713 review (findings 2, 3, 7): full slot quarantine on a failed KV +// restore. Converges the STATE_PUT zero-read cleanup and the DECODE_APPLY +// status==0 cleanup, which previously drifted apart. `prompt_clear(false)` +// clears KV cells + prompt.tokens + restored_logits but does NOT touch +// prompt.checkpoints / just_restored / n_prompt_tokens_cache — the removed +// code cleared those, so this helper restores that completeness so a failed +// restore cannot leave checkpoints referencing positions with no backing KV +// cells (the pos_min == -1 / #641 class this quarantine exists to prevent). +// +// After this returns the slot is exactly as if nothing had been restored: +// KV cells (ctx_tgt + ctx_dft) — prompt_clear → common_context_seq_rm(-1,-1) +// prompt.tokens — prompt_clear +// restored_logits / logits_valid — prompt_clear +// prompt.checkpoints — cleared here (prompt_clear skips them) +// just_restored — cleared here (one-shot restore flag) +// n_prompt_tokens_cache — reset to 0 +// n_prompt_tokens_processed/n_decoded — reset to 0 (same trio as the +// PREFILL task start, so STATE_META +// reports n_past == 0) +// +// Inference-thread only; the slot must not be processing (prompt_clear asserts). +// Both callers (STATE_PUT zero-read, DECODE_APPLY status==0) run there on an +// idle slot; DECODE_APPLY already prompt_clears the same slot earlier in the +// handler, so the assert cannot newly fire. +static void hydra_quarantine_slot(server_slot & slot) { + slot.prompt_clear(false); // KV cells + tokens + restored logits + slot.prompt.checkpoints.clear(); // prompt_clear does NOT clear checkpoints + slot.just_restored = false; // one-shot flag must not survive a failure + slot.n_prompt_tokens_cache = 0; + slot.n_prompt_tokens_processed = 0; + slot.n_decoded = 0; +} + // @@ -3813,8 +3846,7 @@ struct server_context_impl { id_slot, state_len); res->rpc_status = HYDRA_STATUS_ERROR; res->error = "KV restore failed (llama_state_seq_set_data returned 0)"; - slot->prompt_clear(false); // clears KV cells + tokens + logits - slot->n_prompt_tokens_cache = 0; + hydra_quarantine_slot(*slot); // KV + tokens + checkpoints + just_restored } else { // D4: Inject trailing logits into per-slot buffer instead of the // shared context-wide llama_get_logits(). This avoids the race where @@ -3914,6 +3946,11 @@ struct server_context_impl { res->is_processing = slot->is_processing(); res->is_transferring = slot->hydra_transferring->load(); res->state_size = (uint64_t)llama_state_seq_get_size(ctx_tgt, slot->id); + // hydra#713 review (finding 6): quarantine observability — + // checkpoint count + restore flag so an observer can verify + // a failed restore left the slot fully clean. + res->n_checkpoints = (uint32_t)slot->prompt.checkpoints.size(); + res->just_restored = slot->just_restored; // M-Perf.9 #289: surface model identity. The Coordinator uses // these to detect cross-model restores — a slot holding a Mini // KV cache must never have it decoded by a Balanced-loaded model. @@ -5271,14 +5308,14 @@ struct server_context_impl { ::shutdown(task.hydra_action.hydra_fd, SHUT_RD); } slot->reserved_for_decode_id = -1; - // Tokens were registered from the v2 header before set_data — - // clear them so the slot is not left poisoned (n_past > 0 - // with no KV cells → pos_min == -1 abort on the next decode - // that touches this slot). Matches the STATE_PUT failure path. - slot->prompt.tokens.clear(); - slot->prompt.checkpoints.clear(); - slot->n_prompt_tokens_cache = 0; - llama_memory_seq_rm(llama_get_memory(ctx_tgt), slot->id, -1, -1); + // Full quarantine — same helper as the STATE_PUT zero-read + // branch (hydra#713 review, finding 3): KV cells, tokens, + // checkpoints, restored logits, just_restored, + // n_prompt_tokens_cache. Tokens were registered from the + // v2 header before set_data, so the slot must not be left + // poisoned (n_past > 0 with no KV cells → pos_min == -1 + // abort on the next decode that touches this slot). + hydra_quarantine_slot(*slot); if (routes_ptr) { server_routes::decode_result_entry entry; entry.id_slot = id_slot; @@ -9029,6 +9066,9 @@ void server_routes::init_routes() { {"slot_id", hr->id_slot}, {"n_past", hr->n_past}, {"state_size", (uint64_t)hr->state_size}, + // hydra#713 review (finding 6): quarantine observability + {"n_checkpoints", (uint32_t)hr->n_checkpoints}, + {"just_restored", hr->just_restored}, {"is_processing", hr->is_processing}, {"is_transferring", hr->is_transferring}, {"operation", hr->operation}, @@ -10506,6 +10546,13 @@ static void hydra_handle_state_put(int fd, int slot_id, uint64_t payload_len, co if (!hydra_recv_all(fd, buf.data(), (size_t)payload_len)) { SRV_WRN("%s", "hydra rpc: STATE_PUT failed to read payload\n"); hydra_write_res(fd, HYDRA_STATUS_ERROR, 0, 0); + // hydra#713 review (finding 7): the peer declared payload_len bytes but + // we stopped reading short — the residual KV bytes would otherwise be + // parsed as the next request header (bad magic → drop). Self-recovering, + // but SHUT_RD closes the read side deterministically, mirroring the + // DECODE_APPLY stream-failure drain. The write side stays open (the + // error response above was the last thing we send on it). + ::shutdown(fd, SHUT_RD); return; } @@ -10598,6 +10645,9 @@ static void hydra_handle_state_meta(int fd, int slot_id, const hydra_rpc_ctx & c meta_j["slot_id"] = res->id_slot; meta_j["n_past"] = res->n_past; meta_j["state_size"] = res->state_size; + // hydra#713 review (finding 6): quarantine observability + meta_j["n_checkpoints"] = res->n_checkpoints; + meta_j["just_restored"] = res->just_restored; meta_j["is_processing"] = res->is_processing; meta_j["is_transferring"] = res->is_transferring; if (!res->model_alias.empty()) meta_j["model_alias"] = res->model_alias; diff --git a/tools/server/server-task.cpp b/tools/server/server-task.cpp index 74b1c979170..163283a7569 100644 --- a/tools/server/server-task.cpp +++ b/tools/server/server-task.cpp @@ -2010,6 +2010,9 @@ json server_task_result_hydra_state::to_json() { if (op == 0x32) { j["n_past"] = n_past; j["state_size"] = state_size; + // hydra#713 review (finding 6): quarantine observability + j["n_checkpoints"] = n_checkpoints; + j["just_restored"] = just_restored; j["is_processing"] = is_processing; j["is_transferring"] = is_transferring; // true while M1/M2 async GET is active // #451: progress fields diff --git a/tools/server/server-task.h b/tools/server/server-task.h index 11c49243325..c4920775b7f 100644 --- a/tools/server/server-task.h +++ b/tools/server/server-task.h @@ -685,6 +685,13 @@ struct server_task_result_hydra_state : server_task_result { bool is_transferring = false; // true while M1/M2 background send is active uint64_t state_size = 0; + // hydra#713 review (finding 6): STATE_META quarantine observability — + // lets an external observer (test/coordinator) verify a failed restore + // left no dangling slot state behind. Backward-compatible metadata + // additions (coordinator ignores unknown JSON keys). + uint32_t n_checkpoints = 0; // slot->prompt.checkpoints.size() + bool just_restored = false; // one-shot KV-restore flag + // M-Perf.9 #289 / #470: model identity for the slot. Populated from // impl->model_name (the alias) and impl->params_base.model.path. // model_hash has been replaced by GGUF-derived semantic identity fields