diff --git a/ggml/include/ggml-rpc.h b/ggml/include/ggml-rpc.h index bf2e36eb9fc2..e8bcae4c8742 100644 --- a/ggml/include/ggml-rpc.h +++ b/ggml/include/ggml-rpc.h @@ -84,6 +84,16 @@ 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); + +// #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 23d31d6b75f6..40a4801c7cb8 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 @@ -289,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; @@ -523,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 @@ -552,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; } @@ -623,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; } @@ -644,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) { @@ -655,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; @@ -668,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) { @@ -677,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) { @@ -733,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 = { @@ -938,9 +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: 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) { @@ -1086,6 +1140,58 @@ 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); +} + +// #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; @@ -2605,6 +2711,12 @@ 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_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/src/llama-context.cpp b/src/llama-context.cpp index a1aa982f7f88..303e97b1a86c 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; } @@ -2515,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 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..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 { @@ -6404,7 +6430,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 +6611,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 +6739,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; } }