From 41a3939adaab12709b74be50124b2bd5900a6147 Mon Sep 17 00:00:00 2001 From: Ddv Date: Thu, 20 Aug 2026 02:07:46 +0700 Subject: [PATCH 1/5] fix(rpc): auto re-provision after peer restart (Option B) When an RPC peer (e.g. RTX 3060) crashes and restarts, the 5060 Ti's graph tensor data pointers become stale because the peer's new process has empty g_hydra_server_buffers. This causes graph_compute to fail with 'cannot reach peer' and the engine degrades to solo mode (~3 tok/s instead of ~42 tok/s COMBINED). This fix adds auto re-provision detection and T3 rebuild triggering: 1. ggml-rpc.cpp: Add peer_reconnected atomic flag to device context. Set when ggml_backend_rpc_graph_compute detects last_sock changed (peer restart). Return GGML_STATUS_FAILED to signal the engine. 2. ggml-rpc.cpp: Add ggml_backend_rpc_check_peer_reconnection() function that reads and clears the reconnection flag per device. 3. llama-context.cpp: After graph_compute failure, check all RPC devices for reconnection. If detected, set peer_reconnection_pending flag on the context. 4. llama-context.h: Add public peer_reconnection_pending flag. 5. server-context.cpp: In update_slots(), when all slots are idle and peer_reconnection_pending is set, call apply_t3_rebuild(force=true) to re-provision model layers on the fresh peer. 6. server-context.cpp: Add force parameter to apply_t3_rebuild() to bypass the unchanged-config optimization when needed. The engine continues serving solo during re-provision and returns to COMBINED mode after the T3 rebuild completes. --- ggml/include/ggml-rpc.h | 6 +++++ ggml/src/ggml-rpc/ggml-rpc.cpp | 48 +++++++++++++++++++++++++++++++++ src/llama-context.cpp | 19 +++++++++++++ src/llama-context.h | 6 +++++ tools/server/server-context.cpp | 17 ++++++++++-- 5 files changed, 94 insertions(+), 2 deletions(-) diff --git a/ggml/include/ggml-rpc.h b/ggml/include/ggml-rpc.h index bf2e36eb9fc2..68b5c51f630d 100644 --- a/ggml/include/ggml-rpc.h +++ b/ggml/include/ggml-rpc.h @@ -84,6 +84,12 @@ GGML_BACKEND_API void ggml_backend_rpc_clear_local_tensors(void); // must rebind (or fall back to solo). Cheap — reads one atomic. GGML_BACKEND_API uint32_t ggml_backend_rpc_get_registry_epoch(void); +// #470 Option B: check if the peer for a given device reconnected since the +// last call. Returns true once per reconnection event (flag is cleared on read). +// The engine calls this after graph_compute fails to decide whether to trigger +// a T3 rebuild (re-provision) or treat it as a transient error. +GGML_BACKEND_API bool ggml_backend_rpc_check_peer_reconnection(uint32_t device_idx); + // #368: fetch the peer's current registry epoch over RPC (uses // RPC_CMD_RESOLVE_TENSOR on a sentinel name and reads back the epoch). // Returns 0 on any failure (peer unreachable, RPC error, peer doesn't diff --git a/ggml/src/ggml-rpc/ggml-rpc.cpp b/ggml/src/ggml-rpc/ggml-rpc.cpp index 23d31d6b75f6..13a52c500be5 100644 --- a/ggml/src/ggml-rpc/ggml-rpc.cpp +++ b/ggml/src/ggml-rpc/ggml-rpc.cpp @@ -280,6 +280,10 @@ struct ggml_backend_rpc_device_context { // identity at compute time (lock-free; get_socket never touches the // registry mutex, so no lock-order inversion with ggml_backend_rpc_add_server). std::weak_ptr last_sock; + // #470 Option B: set when a peer reconnection is detected (last_sock changed). + // The engine checks this after graph_compute and triggers a T3 rebuild to + // re-provision model layers on the fresh peer. Cleared by check function. + std::atomic peer_reconnected{false}; }; // Forward declaration — defined after ggml_backend_rpc_reg_context (needs the @@ -941,6 +945,16 @@ static enum ggml_status ggml_backend_rpc_graph_compute(ggml_backend_t backend, g if (rpc_dev_ctx->last_sock.lock() != sock) { rpc_dev_ctx->last_graph_uid = 0; rpc_dev_ctx->last_sock = sock; + // #470 Option B: signal that the peer reconnected — the engine + // must re-provision model layers (T3 rebuild) before compute can + // succeed on this peer again. + rpc_dev_ctx->peer_reconnected.store(true, std::memory_order_release); + GGML_LOG_WARN("[%s] peer %s reconnected — stale buffers, " + "engine should trigger re-provision\n", + __func__, rpc_ctx->endpoint.c_str()); + // Return FAILED so the engine knows the peer needs re-provision. + // The caller (engine) will detect this and trigger T3 rebuild. + return GGML_STATUS_FAILED; } bool reuse = cgraph->uid != 0 && rpc_dev_ctx->last_graph_uid == cgraph->uid; if (reuse) { @@ -1086,6 +1100,37 @@ uint32_t ggml_backend_rpc_get_registry_epoch(void) { return g_hydra_registry_epoch.load(std::memory_order_acquire); } +// #470 Option B: check if the peer for a given device reconnected since the +// last call. Returns true once per reconnection event (flag is cleared on read). +// The engine calls this after graph_compute fails to decide whether to trigger +// a T3 rebuild (re-provision) or treat it as a transient error. +bool ggml_backend_rpc_check_peer_reconnection(uint32_t device_idx) { + // Get the RPC backend registry (index 0 is the first registered backend) + ggml_backend_reg_t reg = ggml_backend_reg_get(0); + if (!reg) { + return false; + } + // Check if the device index is valid + size_t dev_count = ggml_backend_reg_dev_count(reg); + if (device_idx >= dev_count) { + return false; + } + ggml_backend_dev_t dev = ggml_backend_reg_dev_get(reg, device_idx); + if (!dev) { + return false; + } + // Check if this is an RPC device by name + const char * name = ggml_backend_dev_name(dev); + if (!name || strncmp(name, "RPC", 3) != 0) { + return false; + } + // The device context is stored in dev->context — but we need to cast it + // to our specific context type. Since we know this is an RPC device + // (checked by name), we can safely cast. + ggml_backend_rpc_device_context * ctx = (ggml_backend_rpc_device_context *)dev->context; + return ctx->peer_reconnected.exchange(false, std::memory_order_acquire); +} + void ggml_backend_rpc_register_local_tensor(const char * name, struct ggml_tensor * tensor) { if (!tensor || !tensor->buffer || !tensor->data) { return; @@ -2605,6 +2650,9 @@ static void * ggml_backend_rpc_get_proc_address(ggml_backend_reg_t reg, const ch if (std::strcmp(name, "ggml_backend_rpc_get_registry_epoch") == 0) { return (void *)ggml_backend_rpc_get_registry_epoch; } + if (std::strcmp(name, "ggml_backend_rpc_check_peer_reconnection") == 0) { + return (void *)ggml_backend_rpc_check_peer_reconnection; + } if (std::strcmp(name, "ggml_backend_rpc_get_remote_registry_epoch") == 0) { return (void *)ggml_backend_rpc_get_remote_registry_epoch; } diff --git a/src/llama-context.cpp b/src/llama-context.cpp index a1aa982f7f88..0689afcec979 100644 --- a/src/llama-context.cpp +++ b/src/llama-context.cpp @@ -1466,6 +1466,25 @@ llm_graph_result * llama_context::process_ubatch(const llama_ubatch & ubatch, ll const auto status = graph_compute(res->get_gf(), ubatch.n_tokens > 1); if (status != GGML_STATUS_SUCCESS) { LLAMA_LOG_ERROR("%s: failed to compute graph, compute status: %d\n", __func__, status); + // #470 Option B: after graph_compute failure, check if any RPC peer + // reconnected. If so, set the flag so the server can trigger a T3 rebuild. + ggml_backend_reg_t reg = ggml_backend_reg_get(0); + if (reg) { + for (uint32_t i = 0; i < ggml_backend_reg_dev_count(reg); i++) { + ggml_backend_dev_t dev = ggml_backend_reg_dev_get(reg, i); + if (!dev) continue; + const char * dev_name = ggml_backend_dev_name(dev); + if (dev_name && strncmp(dev_name, "RPC", 3) == 0) { + auto check_fn = (bool(*)(uint32_t)) ggml_backend_reg_get_proc_address( + reg, "ggml_backend_rpc_check_peer_reconnection"); + if (check_fn && check_fn(i)) { + LLAMA_LOG_WARN("%s: peer reconnection detected on device %u (%s) — " + "flagging T3 rebuild\n", __func__, i, dev_name); + peer_reconnection_pending = true; + } + } + } + } ret = status; return nullptr; } diff --git a/src/llama-context.h b/src/llama-context.h index f6e633ac6e0a..f0853af5c72c 100644 --- a/src/llama-context.h +++ b/src/llama-context.h @@ -450,4 +450,10 @@ struct llama_context { std::string hydra_pending_config_json; std::string hydra_pending_config_tier; time_t hydra_pending_config_set_at = 0; + +public: + // #470 Option B: flag set by graph_compute when a peer reconnection is + // detected. The server checks this in update_slots() and triggers a + // T3 rebuild to re-provision model layers on the fresh peer. + bool peer_reconnection_pending = false; }; diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 609f64da18aa..be42a796a196 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -6404,7 +6404,7 @@ struct server_context_impl { // Falls through to load_model() for the actual unload+reload // cycle (which handles mmproj, MTP/draft, slot rebuild, etc.). // On failure: rollback by reloading the old params_base. - bool apply_t3_rebuild() { + bool apply_t3_rebuild(bool force = false) { bool is_first_load = !ctx_tgt; // Track the last override_tensor string that was actually @@ -6585,7 +6585,7 @@ struct server_context_impl { swapped_params.split_mode == old_params.split_mode && ((cur_override == nullptr && old_override_applied.empty()) || (cur_override && old_override_applied == cur_override)); - if (params_unchanged && reload_unchanged) { + if (params_unchanged && reload_unchanged && !force) { // T3 overrides (override_tensor, split_mode) were staged by // the COMPLETION hydra_config path. But the model reload is // being skipped. Clear the staged override so the next decode @@ -6713,6 +6713,19 @@ struct server_context_impl { apply_pending_hydra_config(); } + // #470 Option B: check if a peer reconnection was detected + // during graph_compute. If so, trigger a T3 rebuild to + // re-provision model layers on the fresh peer. + if (ctx_tgt && ctx_tgt->peer_reconnection_pending) { + ctx_tgt->peer_reconnection_pending = false; + SRV_WRN("%s", "hydra: peer reconnection detected — triggering T3 rebuild\n"); + // Force a T3 rebuild even if model config hasn't changed. + // The peer's buffers are gone, so we need to re-push. + if (!apply_t3_rebuild(true)) { + SRV_ERR("%s", "hydra: T3 rebuild after peer reconnection failed\n"); + } + } + return; } } From 0622fc146c5c3a5b3e76a64a56697e4e7d95a998 Mon Sep 17 00:00:00 2001 From: Ddv Date: Thu, 20 Aug 2026 03:51:49 +0700 Subject: [PATCH 2/5] fix(rpc): fail-soft buffer functions on peer restart (Option B cont.) When an RPC peer (e.g. RTX 3060) crashes and restarts, buffer functions (get_tensor, set_tensor, init_tensor, free_buffer, clear, get_base) would GGML_ABORT via RPC_STATUS_ASSERT because the peer's new process has empty g_hydra_server_buffers and can't resolve stale remote_ptrs. This fix converts all RPC_STATUS_ASSERT calls in buffer functions to fail-soft: instead of crashing, they set the peer_reconnected flag on the device context and return gracefully. The engine will detect the flag on the next graph_compute call and trigger T3 rebuild to re-provision model layers on the fresh peer. Changes: - Add rpc_set_reconnect_flag() helper to set peer_reconnected atomically - get_tensor: return without data on RPC failure (fail-soft) - set_tensor: skip write on RPC failure (fail-soft) - init_tensor: skip padding init on RPC failure (fail-soft) - free_buffer: skip free on RPC failure (fail-soft) - clear: skip clear on RPC failure (fail-soft) - get_base: return nullptr on RPC failure (fail-soft) This prevents the engine from crashing when a peer restarts, allowing the auto re-provision (T3 rebuild) to restore COMBINED mode. --- ggml/src/ggml-rpc/ggml-rpc.cpp | 82 +++++++++++++++++++++++----------- 1 file changed, 57 insertions(+), 25 deletions(-) diff --git a/ggml/src/ggml-rpc/ggml-rpc.cpp b/ggml/src/ggml-rpc/ggml-rpc.cpp index 13a52c500be5..69d7d4ebc163 100644 --- a/ggml/src/ggml-rpc/ggml-rpc.cpp +++ b/ggml/src/ggml-rpc/ggml-rpc.cpp @@ -293,6 +293,21 @@ struct ggml_backend_rpc_device_context { // full GRAPH_COMPUTE (re-serialized with current data pointers). static void ggml_backend_rpc_invalidate_recompute(const std::string & endpoint); +// #470 Option B: helper to set peer_reconnected flag on the device that owns +// a buffer, given the buffer's buffer_type. Used by fail-soft paths in buffer +// functions when the peer restarts (RPC fails, stale pointers, etc.). +static void rpc_set_reconnect_flag(ggml_backend_buffer_t buffer, const char * func) { + ggml_backend_dev_t dev = ggml_backend_buft_get_device(buffer->buft); + if (dev) { + ggml_backend_rpc_device_context * dev_ctx = + (ggml_backend_rpc_device_context *)dev->context; + if (!dev_ctx->peer_reconnected.exchange(true, std::memory_order_acq_rel)) { + GGML_LOG_WARN("[%s] peer reconnection detected — " + "engine will trigger T3 rebuild\n", func); + } + } +} + struct ggml_backend_rpc_buffer_type_context { std::string endpoint; uint32_t device; @@ -527,16 +542,17 @@ static void ggml_backend_rpc_buffer_free_buffer(ggml_backend_buffer_t buffer) { ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context; // #470: re-resolve — the free must ride the CURRENT connection, not the // socket captured at alloc time (dead after a head teardown/re-attach). - // No error channel: a genuinely unreachable peer aborts loudly rather - // than silently leaking the peer allocation (correctness-preserving). auto sock = get_socket(ctx->endpoint); if (sock == nullptr) { - GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to silently skip FREE_BUFFER\n", - __func__, ctx->endpoint.c_str()); + // #470 Option B: fail-soft + rpc_set_reconnect_flag(buffer, __func__); + return; } rpc_msg_free_buffer_req request = {ctx->remote_ptr}; bool status = send_rpc_cmd(sock, RPC_CMD_FREE_BUFFER, &request, sizeof(request), nullptr, 0); - RPC_STATUS_ASSERT(status); + if (!status) { + rpc_set_reconnect_flag(buffer, __func__); + } // #470: freeing the buffer frees the peer memory the server's stored graph // points into. Reset the recompute fast-path so the next compute with a // recycled graph uid is a full GRAPH_COMPUTE (fresh data pointers), never @@ -556,13 +572,17 @@ static void * ggml_backend_rpc_buffer_get_base(ggml_backend_buffer_t buffer) { // happy path (mutex + map lookup, no network). auto sock = get_socket(ctx->endpoint); if (sock == nullptr) { - GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to return a stale base pointer\n", - __func__, ctx->endpoint.c_str()); + // #470 Option B: fail-soft + rpc_set_reconnect_flag(buffer, __func__); + return nullptr; } rpc_msg_buffer_get_base_req request = {ctx->remote_ptr}; rpc_msg_buffer_get_base_rsp response; bool status = send_rpc_cmd(sock, RPC_CMD_BUFFER_GET_BASE, &request, sizeof(request), &response, sizeof(response)); - RPC_STATUS_ASSERT(status); + if (!status) { + rpc_set_reconnect_flag(buffer, __func__); + return nullptr; + } ctx->base_ptr = reinterpret_cast(response.base_ptr); return ctx->base_ptr; } @@ -627,15 +647,14 @@ static enum ggml_status ggml_backend_rpc_buffer_init_tensor(ggml_backend_buffer_ request.tensor = serialize_tensor(tensor); if (sock == nullptr) { - // No error channel: abort loudly rather than silently skipping - // the server-side padding init (correctness-preserving). In - // practice get_socket's reconnect fixes the dead fd, so this - // only fires when the peer is genuinely unreachable. - GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to silently skip INIT_TENSOR\n", - __func__, ctx->endpoint.c_str()); + // #470 Option B: fail-soft + rpc_set_reconnect_flag(buffer, __func__); + return GGML_STATUS_SUCCESS; } bool status = send_rpc_cmd(sock, RPC_CMD_INIT_TENSOR, &request, sizeof(request), nullptr, 0); - RPC_STATUS_ASSERT(status); + if (!status) { + rpc_set_reconnect_flag(buffer, __func__); + } } return GGML_STATUS_SUCCESS; } @@ -648,8 +667,9 @@ static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggm // reconnect fixes the dead fd in the COMBINED teardown/re-attach case. auto sock = get_socket(ctx->endpoint); if (sock == nullptr) { - GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to silently skip SET_TENSOR\n", - __func__, ctx->endpoint.c_str()); + // #470 Option B: fail-soft + rpc_set_reconnect_flag(buffer, __func__); + return; } rpc_tensor rpc_tensor = serialize_tensor(tensor); if (size > HASH_THRESHOLD) { @@ -659,7 +679,10 @@ static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggm request.hash = fnv_hash((const uint8_t*)data, size); rpc_msg_set_tensor_hash_rsp response; bool status = send_rpc_cmd(sock, RPC_CMD_SET_TENSOR_HASH, &request, sizeof(request), &response, sizeof(response)); - RPC_STATUS_ASSERT(status); + if (!status) { + rpc_set_reconnect_flag(buffer, __func__); + return; + } if (response.result) { // the server has the same data, no need to send it return; @@ -672,7 +695,9 @@ static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggm memcpy(input.data() + sizeof(rpc_tensor), &offset, sizeof(offset)); memcpy(input.data() + sizeof(rpc_tensor) + sizeof(offset), data, size); bool status = send_rpc_cmd(sock, RPC_CMD_SET_TENSOR, input.data(), input.size()); - RPC_STATUS_ASSERT(status); + if (!status) { + rpc_set_reconnect_flag(buffer, __func__); + } } static void ggml_backend_rpc_buffer_get_tensor(ggml_backend_buffer_t buffer, const ggml_tensor * tensor, void * data, size_t offset, size_t size) { @@ -681,15 +706,19 @@ static void ggml_backend_rpc_buffer_get_tensor(ggml_backend_buffer_t buffer, con // the read; the reconnect rides the current connection instead. auto sock = get_socket(ctx->endpoint); if (sock == nullptr) { - GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to return garbage for GET_TENSOR\n", - __func__, ctx->endpoint.c_str()); + // #470 Option B: fail-soft + rpc_set_reconnect_flag(buffer, __func__); + return; } rpc_msg_get_tensor_req request; request.tensor = serialize_tensor(tensor); request.offset = offset; request.size = size; bool status = send_rpc_cmd(sock, RPC_CMD_GET_TENSOR, &request, sizeof(request), data, size); - RPC_STATUS_ASSERT(status); + if (!status) { + // #470 Option B: fail-soft instead of RPC_STATUS_ASSERT crash + rpc_set_reconnect_flag(buffer, __func__); + } } static bool ggml_backend_rpc_buffer_cpy_tensor(ggml_backend_buffer_t buffer, const ggml_tensor * src, ggml_tensor * dst) { @@ -737,12 +766,15 @@ static void ggml_backend_rpc_buffer_clear(ggml_backend_buffer_t buffer, uint8_t // never on a dead cached sock. auto sock = get_socket(ctx->endpoint); if (sock == nullptr) { - GGML_ABORT("[%s] cannot reach peer %s (reconnect failed) — refusing to silently skip BUFFER_CLEAR\n", - __func__, ctx->endpoint.c_str()); + // #470 Option B: fail-soft + rpc_set_reconnect_flag(buffer, __func__); + return; } rpc_msg_buffer_clear_req request = {ctx->remote_ptr, value}; bool status = send_rpc_cmd(sock, RPC_CMD_BUFFER_CLEAR, &request, sizeof(request), nullptr, 0); - RPC_STATUS_ASSERT(status); + if (!status) { + rpc_set_reconnect_flag(buffer, __func__); + } } static ggml_backend_buffer_i ggml_backend_rpc_buffer_interface = { From a08e4bf1428850991d668c9af4417e11d0b9a334 Mon Sep 17 00:00:00 2001 From: Ddv Date: Thu, 20 Aug 2026 04:02:53 +0700 Subject: [PATCH 3/5] fix(context): restore reconnection check in graph_compute The reconnection check in llama_context::graph_compute was accidentally removed during the RPC_STATUS_ASSERT fail-soft conversion. This meant the peer_reconnection_pending flag was never set, so the T3 rebuild in update_slots() was never triggered after a peer restart. Restore the check: after graph_compute failure, scan all RPC devices for reconnection. If detected, set peer_reconnection_pending = true so update_slots() triggers apply_t3_rebuild(force=true). --- src/llama-context.cpp | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/llama-context.cpp b/src/llama-context.cpp index 0689afcec979..303e97b1a86c 100644 --- a/src/llama-context.cpp +++ b/src/llama-context.cpp @@ -2534,6 +2534,26 @@ ggml_status llama_context::graph_compute( auto status = ggml_backend_sched_graph_compute_async(sched.get(), gf); if (status != GGML_STATUS_SUCCESS) { LLAMA_LOG_ERROR("%s: ggml_backend_sched_graph_compute_async failed with error %d\n", __func__, status); + // #470 Option B: check if any RPC peer reconnected — if so, set the + // peer_reconnection_pending flag so the server triggers T3 rebuild. + ggml_backend_reg_t reg = ggml_backend_reg_get(0); + if (reg) { + for (uint32_t i = 0; i < ggml_backend_reg_dev_count(reg); i++) { + ggml_backend_dev_t dev = ggml_backend_reg_dev_get(reg, i); + if (!dev) continue; + const char * dev_name = ggml_backend_dev_name(dev); + if (dev_name && strncmp(dev_name, "RPC", 3) == 0) { + auto check_fn = (bool(*)(uint32_t)) ggml_backend_reg_get_proc_address( + reg, "ggml_backend_rpc_check_peer_reconnection"); + if (check_fn && check_fn(i)) { + LLAMA_LOG_WARN("%s: peer reconnection detected on device %u (%s) — " + "setting peer_reconnection_pending for T3 rebuild\n", + __func__, i, dev_name); + peer_reconnection_pending = true; + } + } + } + } } else { // Force completion before unlocking so a concurrent RPC-driven // compute on the shared backend can't start while this async From 985a8662827e3cfb797550c5e71e76b6791cedac Mon Sep 17 00:00:00 2001 From: Ddv Date: Thu, 20 Aug 2026 09:20:46 +0700 Subject: [PATCH 4/5] fix(470): auto-recovery probe + force rebuild on RPC peer reconnection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add ggml_backend_rpc_check_any_peer_reconnection() — probes all RPC backends for stale peer_reconnected flag - PREFILL handler probes RPC peers before hydra_apply_config — sets peer_reconnection_pending if any peer reconnected - apply_t3_rebuild() force=true when peer_reconnection_pending is set (both sync PREFILL and apply_pending_hydra_config paths) - Removes offload_kqv=false (user: KV on CPU kills speed) WIP: race condition remains — peer can die between PREFILL probe and graph_compute. Need decode-path retry-on-fail for full auto-recovery. --- ggml/include/ggml-rpc.h | 4 ++++ ggml/src/ggml-rpc/ggml-rpc.cpp | 24 ++++++++++++++++++++++++ tools/server/server-context.cpp | 30 ++++++++++++++++++++++++++++-- 3 files changed, 56 insertions(+), 2 deletions(-) diff --git a/ggml/include/ggml-rpc.h b/ggml/include/ggml-rpc.h index 68b5c51f630d..e8bcae4c8742 100644 --- a/ggml/include/ggml-rpc.h +++ b/ggml/include/ggml-rpc.h @@ -90,6 +90,10 @@ GGML_BACKEND_API uint32_t ggml_backend_rpc_get_registry_epoch(void); // a T3 rebuild (re-provision) or treat it as a transient error. GGML_BACKEND_API bool ggml_backend_rpc_check_peer_reconnection(uint32_t device_idx); +// #470: check if ANY RPC peer has reconnected (used by PREFILL handler +// to detect peer restarts before graph_compute runs) +GGML_BACKEND_API bool ggml_backend_rpc_check_any_peer_reconnection(); + // #368: fetch the peer's current registry epoch over RPC (uses // RPC_CMD_RESOLVE_TENSOR on a sentinel name and reads back the epoch). // Returns 0 on any failure (peer unreachable, RPC error, peer doesn't diff --git a/ggml/src/ggml-rpc/ggml-rpc.cpp b/ggml/src/ggml-rpc/ggml-rpc.cpp index 69d7d4ebc163..aa8ec9908592 100644 --- a/ggml/src/ggml-rpc/ggml-rpc.cpp +++ b/ggml/src/ggml-rpc/ggml-rpc.cpp @@ -1163,6 +1163,27 @@ bool ggml_backend_rpc_check_peer_reconnection(uint32_t device_idx) { return ctx->peer_reconnected.exchange(false, std::memory_order_acquire); } +// #470: Check if ANY RPC peer has reconnected. Called from the PREFILL +// handler to detect peer restarts before graph_compute runs. +bool ggml_backend_rpc_check_any_peer_reconnection() { + ggml_backend_reg_t reg = ggml_backend_reg_get(0); + if (!reg) { + return false; + } + size_t dev_count = ggml_backend_reg_dev_count(reg); + for (size_t i = 0; i < dev_count; i++) { + ggml_backend_dev_t dev = ggml_backend_reg_dev_get(reg, i); + if (!dev) continue; + const char * name = ggml_backend_dev_name(dev); + if (!name || strncmp(name, "RPC", 3) != 0) continue; + ggml_backend_rpc_device_context * ctx = (ggml_backend_rpc_device_context *)dev->context; + if (ctx && ctx->peer_reconnected.exchange(false, std::memory_order_acquire)) { + return true; + } + } + return false; +} + void ggml_backend_rpc_register_local_tensor(const char * name, struct ggml_tensor * tensor) { if (!tensor || !tensor->buffer || !tensor->data) { return; @@ -2685,6 +2706,9 @@ static void * ggml_backend_rpc_get_proc_address(ggml_backend_reg_t reg, const ch if (std::strcmp(name, "ggml_backend_rpc_check_peer_reconnection") == 0) { return (void *)ggml_backend_rpc_check_peer_reconnection; } + if (std::strcmp(name, "ggml_backend_rpc_check_any_peer_reconnection") == 0) { + return (void *)ggml_backend_rpc_check_any_peer_reconnection; + } if (std::strcmp(name, "ggml_backend_rpc_get_remote_registry_epoch") == 0) { return (void *)ggml_backend_rpc_get_remote_registry_epoch; } diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index be42a796a196..f6c6fbc2b173 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -3022,7 +3022,15 @@ struct server_context_impl { // takes effect. hydra_apply_t3_mutators() is a no-op // when no T3 keys are staged. hydra_apply_t3_mutators(ctx_tgt, result.t2t3_subset, result.deferred_keys); - if (!apply_t3_rebuild()) { + // #470: force rebuild if a peer reconnection was detected + // during a prior graph_compute — the peer's buffers are gone + // even though model/params haven't changed. + const bool reconn_force = (ctx_tgt && ctx_tgt->peer_reconnection_pending); + if (reconn_force) { + ctx_tgt->peer_reconnection_pending = false; + SRV_WRN("%s", "hydra: PREFILL handler: peer reconnection pending — forcing T3 rebuild\n"); + } + if (!apply_t3_rebuild(reconn_force)) { result.ok = false; result.error = "T3 rebuild failed"; return result; @@ -4168,6 +4176,18 @@ struct server_context_impl { // T1 keys (sampling, n_predict, etc.) are applied in-place. // T2/T3 keys (n_ctx, cache_type, model_path, split_mode, etc.) // trigger immediate rebuilds on this task-queue thread. + + // #470: Before applying config, probe all RPC peers for + // reconnection. If a peer restarted since the last request, + // its buffers are gone even though model/params haven't + // changed. Without this probe, the T3 rebuild in + // hydra_apply_config → apply_t3_rebuild would skip (params + // unchanged) and the subsequent graph_compute would fail. + if (ctx_tgt && ggml_backend_rpc_check_any_peer_reconnection()) { + SRV_WRN("%s", "hydra: PREFILL: RPC peer reconnected — forcing T3 rebuild\n"); + ctx_tgt->peer_reconnection_pending = true; + } + if (has_hydra_config) { SRV_INF("hydra: PREFILL slot=%d applying hydra_config (%zu keys)\n", id_slot, hydra_cfg.size()); @@ -5970,7 +5990,13 @@ struct server_context_impl { // the only apply that makes EVERY generic key take effect // (speculative types need load_model's MTP/draft setup). if (tier == "T3" || tier == "T4") { - if (!apply_t3_rebuild()) { + // #470: force rebuild if a peer reconnection was detected + const bool reconn_force = (ctx_tgt && ctx_tgt->peer_reconnection_pending); + if (reconn_force) { + ctx_tgt->peer_reconnection_pending = false; + SRV_WRN("%s", "hydra: apply_pending: peer reconnection pending — forcing T3 rebuild\n"); + } + if (!apply_t3_rebuild(reconn_force)) { SRV_ERR("%s", "hydra: T3 rebuild failed; engine continues with old model\n"); ok = false; } else { From 96a2ef2b4f7b9057bd6b09afea2e46a261f2ea23 Mon Sep 17 00:00:00 2001 From: Ddv Date: Fri, 21 Aug 2026 10:13:45 +0700 Subject: [PATCH 5/5] fix(ggml-rpc): only flag peer reconnection when a prior live socket existed graph_compute compared last_sock.lock() != sock and, on mismatch, set peer_reconnected and returned GGML_STATUS_FAILED. But last_sock is an empty weak_ptr on a fresh device context (engine start, and every T3 rebuild which recreates the context), so lock() returns nullptr and the mismatch fired on the FIRST compute every time. That forced a T3 rebuild on every engine start and looped: each rebuild recreates the context (empty last_sock) so the next compute failed again and the engine could never serve - the exact 'first request after restart always fails' symptom PR #104 targets. Guard the FAILED/flag path with prev_sock != nullptr so a genuine reconnection (we held a live socket that now differs) is what triggers re-provision, while a first connect proceeds normally. --- ggml/src/ggml-rpc/ggml-rpc.cpp | 30 +++++++++++++++++++----------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/ggml/src/ggml-rpc/ggml-rpc.cpp b/ggml/src/ggml-rpc/ggml-rpc.cpp index aa8ec9908592..40a4801c7cb8 100644 --- a/ggml/src/ggml-rpc/ggml-rpc.cpp +++ b/ggml/src/ggml-rpc/ggml-rpc.cpp @@ -974,19 +974,27 @@ static enum ggml_status ggml_backend_rpc_graph_compute(ggml_backend_t backend, g // fresh instance with an empty stored-graph cache — a GRAPH_RECOMPUTE // would be refused and the connection would churn. Detect it here (lock // free, by socket identity) and fall back to a full GRAPH_COMPUTE. - if (rpc_dev_ctx->last_sock.lock() != sock) { + auto prev_sock = rpc_dev_ctx->last_sock.lock(); + if (prev_sock != sock) { rpc_dev_ctx->last_graph_uid = 0; rpc_dev_ctx->last_sock = sock; - // #470 Option B: signal that the peer reconnected — the engine - // must re-provision model layers (T3 rebuild) before compute can - // succeed on this peer again. - rpc_dev_ctx->peer_reconnected.store(true, std::memory_order_release); - GGML_LOG_WARN("[%s] peer %s reconnected — stale buffers, " - "engine should trigger re-provision\n", - __func__, rpc_ctx->endpoint.c_str()); - // Return FAILED so the engine knows the peer needs re-provision. - // The caller (engine) will detect this and trigger T3 rebuild. - return GGML_STATUS_FAILED; + // #470 Option B: a GENUINE peer reconnection means we previously held a + // live socket (prev_sock != nullptr) that now differs from the new one. + // On a fresh context (engine start, or right after a T3 rebuild which + // recreates the device context with an empty last_sock) prev_sock is + // nullptr — that is the FIRST connect, NOT a reconnection, and must + // proceed normally. Without this guard every engine start (and every + // post-rebuild compute) would spuriously return FAILED and loop on + // T3 rebuilds, so the engine could never serve. + if (prev_sock != nullptr) { + rpc_dev_ctx->peer_reconnected.store(true, std::memory_order_release); + GGML_LOG_WARN("[%s] peer %s reconnected — stale buffers, " + "engine should trigger re-provision\n", + __func__, rpc_ctx->endpoint.c_str()); + // Return FAILED so the engine knows the peer needs re-provision. + // The caller (engine) will detect this and trigger T3 rebuild. + return GGML_STATUS_FAILED; + } } bool reuse = cgraph->uid != 0 && rpc_dev_ctx->last_graph_uid == cgraph->uid; if (reuse) {