diff --git a/ggml/src/ggml-moe-gguf-source.h b/ggml/src/ggml-moe-gguf-source.h index d3cb46c40d50..068d934c9593 100644 --- a/ggml/src/ggml-moe-gguf-source.h +++ b/ggml/src/ggml-moe-gguf-source.h @@ -70,7 +70,7 @@ class GgufMoeSource { const int total = std::atoi(first.substr(of + pat.size(), 5).c_str()); const size_t idx0 = first.rfind('-', of - 1); const std::string prefix = first.substr(0, idx0 + 1), suffix = first.substr(of); - for (int i = 1; i <= total; i++) { char n[8]; std::snprintf(n, sizeof(n), "%05d", i); files.push_back(prefix + n + suffix); } + for (int i = 1; i <= total; i++) { char n[16]; std::snprintf(n, sizeof(n), "%05d", i); files.push_back(prefix + n + suffix); } } for (const auto & p : files) { Shard s; gguf_init_params gp{}; gp.no_alloc = true; gp.ctx = nullptr; diff --git a/tools/moe-fit/fit-device.cpp b/tools/moe-fit/fit-device.cpp index ead67de7b186..9202697db1f3 100644 --- a/tools/moe-fit/fit-device.cpp +++ b/tools/moe-fit/fit-device.cpp @@ -39,7 +39,7 @@ static std::vector shard_paths(const std::string & first) { const int total = std::atoi(first.substr(of + pat.size(), 5).c_str()); const size_t idx0 = first.rfind('-', of - 1); const std::string prefix = first.substr(0, idx0 + 1), suffix = first.substr(of); - for (int i = 1; i <= total; i++) { char b[8]; std::snprintf(b, sizeof(b), "%05d", i); files.push_back(prefix + b + suffix); } + for (int i = 1; i <= total; i++) { char b[16]; std::snprintf(b, sizeof(b), "%05d", i); files.push_back(prefix + b + suffix); } return files; } diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 36a38608de93..32e2a55f1d50 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -2473,6 +2473,42 @@ struct server_context_impl { send_error(task, "Unable to save slot", ERROR_TYPE_SERVER); break; } + // Continuum fork: the CONTEXT CHECKPOINTS are part of the register file. + // On a hybrid/recurrent (or SWA) cache the next request can only reuse a + // prefix by rolling back to a checkpoint at or before the divergence — + // and even an identical prompt rolls back one token [TAG_PROMPT_LOGITS]. + // A restored slot with no checkpoints therefore always hits do_reset: + // "restore ok", then a full re-prefill (measured 2026-09-04, n_restored + // 11298 → cache_n 0). The state file carries KV + tokens only, so the + // checkpoints ride in a sidecar beside it; a missing or unreadable + // sidecar degrades to the pre-fork cold restore, never to an error. + { + const std::string ckpt_path = filepath + ".ckpt"; + std::ofstream out(ckpt_path, std::ios::binary | std::ios::trunc); + auto put = [&](const void * ptr, size_t n) { out.write(reinterpret_cast(ptr), n); }; + const uint32_t magic = 0x434B5054u; // 'CKPT' + const uint32_t version = 1; + const uint32_t n_ckpt = (uint32_t) slot->prompt.checkpoints.size(); + put(&magic, sizeof(magic)); put(&version, sizeof(version)); put(&n_ckpt, sizeof(n_ckpt)); + for (const auto & c : slot->prompt.checkpoints) { + const int64_t n_tokens = c.n_tokens; + const int32_t pos_min = c.pos_min; + const int32_t pos_max = c.pos_max; + const uint64_t sz_tgt = c.data_tgt.size(); + const uint64_t sz_dft = c.data_dft.size(); + const uint64_t sz_spec = c.data_spec.size(); + put(&n_tokens, sizeof(n_tokens)); put(&pos_min, sizeof(pos_min)); put(&pos_max, sizeof(pos_max)); + put(&sz_tgt, sizeof(sz_tgt)); put(&sz_dft, sizeof(sz_dft)); put(&sz_spec, sizeof(sz_spec)); + put(c.data_tgt.data(), c.data_tgt.size()); + put(c.data_dft.data(), c.data_dft.size()); + put(c.data_spec.data(), c.data_spec.size()); + } + if (!out) { + SLT_WRN(*slot, "checkpoint sidecar could not be written: %s (the restore will be cold)\n", ckpt_path.c_str()); + } else { + SLT_INF(*slot, "saved %u context checkpoints beside the slot state: %s\n", n_ckpt, ckpt_path.c_str()); + } + } const int64_t t_end = ggml_time_us(); const double t_save_ms = (t_end - t_start) / 1000.0; @@ -2533,6 +2569,43 @@ struct server_context_impl { slot->prompt.clear(); slot->prompt.tokens = std::move(restored); + // Continuum fork: bring the context checkpoints back with the state (see + // SLOT_SAVE). Any defect in the sidecar leaves the slot exactly as a + // pre-fork restore left it — valid tokens + state, no checkpoints. + { + const std::string ckpt_path = filepath + ".ckpt"; + std::ifstream in(ckpt_path, std::ios::binary); + std::list loaded; + bool ok = (bool) in; + auto get = [&](void * ptr, size_t n) { in.read(reinterpret_cast(ptr), n); ok = ok && (bool) in; }; + if (ok) { + uint32_t magic = 0, version = 0, n_ckpt = 0; + get(&magic, sizeof(magic)); get(&version, sizeof(version)); get(&n_ckpt, sizeof(n_ckpt)); + ok = ok && magic == 0x434B5054u && version == 1 && n_ckpt <= 4096; + for (uint32_t i = 0; ok && i < n_ckpt; ++i) { + int64_t n_tokens = 0; int32_t pos_min = 0, pos_max = 0; + uint64_t sz_tgt = 0, sz_dft = 0, sz_spec = 0; + get(&n_tokens, sizeof(n_tokens)); get(&pos_min, sizeof(pos_min)); get(&pos_max, sizeof(pos_max)); + get(&sz_tgt, sizeof(sz_tgt)); get(&sz_dft, sizeof(sz_dft)); get(&sz_spec, sizeof(sz_spec)); + // a checkpoint is at most one sequence state; refuse absurd sizes + ok = ok && sz_tgt < (1ull << 36) && sz_dft < (1ull << 36) && sz_spec < (1ull << 30); + if (!ok) break; + auto & c = loaded.emplace_back(); + c.update_pos(n_tokens, pos_min, pos_max); + c.id_task = -1; + c.data_tgt.resize(sz_tgt); get(c.data_tgt.data(), sz_tgt); + c.data_dft.resize(sz_dft); get(c.data_dft.data(), sz_dft); + c.data_spec.resize(sz_spec); get(c.data_spec.data(), sz_spec); + } + } + if (ok) { + slot->prompt.checkpoints = std::move(loaded); + SLT_INF(*slot, "restored %zu context checkpoints from the sidecar: %s\n", slot->prompt.checkpoints.size(), ckpt_path.c_str()); + } else { + slot->prompt.checkpoints.clear(); + SLT_WRN(*slot, "no usable checkpoint sidecar (%s): the restore is cold — the next divergence re-prefills from the head\n", ckpt_path.c_str()); + } + } } catch (const std::exception & err) { slot->prompt_clear(); send_error(task, std::string("Unable to restore slot: ") + err.what(), ERROR_TYPE_INVALID_REQUEST); diff --git a/tools/server/tests/unit/test_slot_save.py b/tools/server/tests/unit/test_slot_save.py index 2d905055d7a5..f0d5dd87239c 100644 --- a/tools/server/tests/unit/test_slot_save.py +++ b/tools/server/tests/unit/test_slot_save.py @@ -168,7 +168,11 @@ def test_deferred_restore_is_bound_to_its_slot(): global server server.server_slots = True # /slots GET to observe processing server.n_slots = 2 - server.n_predict = 512 + # The long slot must OUTLAST the whole restore round-trip on a fast CI runner (a 512-token + # gen of the tiny model ends in ~2 s on hosted ubuntu/windows runners — measured 2026-09-13: + # the binding assertion then failed for timing, not for binding). 4096 keeps slot 1 busy. + server.n_predict = 4096 + server.n_ctx = 8192 # 4096 per slot: room for the long slot to actually run 4096 tokens server.start() # A reusable page on disk: process a prompt on slot 0, save it, let slot 0 idle. @@ -181,20 +185,23 @@ def test_deferred_restore_is_bound_to_its_slot(): done0 = threading.Event() done1 = threading.Event() + t_done = {} def gen(id_slot, n_predict, done): try: for _ in server.make_stream_request("POST", "/completion", data={ "prompt": "Tell me a very long and detailed story about a cat.", "id_slot": id_slot, "n_predict": n_predict, "cache_prompt": True, "stream": True, + "ignore_eos": True, # the tiny model emits EOS within a few tokens: the long slot must really run long }): pass finally: + t_done[id_slot] = time.time() done.set() - # slot 0 finishes QUICKLY (short), slot 1 runs LONG. The returner wants slot 0. - t0 = threading.Thread(target=gen, args=(0, 24, done0), daemon=True) - t1 = threading.Thread(target=gen, args=(1, 512, done1), daemon=True) + # slot 0 finishes FIRST (256 tokens: long enough that a fast runner can observe it busy), slot 1 runs LONG. The returner wants slot 0. + t0 = threading.Thread(target=gen, args=(0, 128, done0), daemon=True) + t1 = threading.Thread(target=gen, args=(1, 4096, done1), daemon=True) t1.start() # ensure slot 1 is processing before slot 0 starts, so the slot-1 op queues first deadline = time.time() + 20 @@ -233,15 +240,21 @@ def restore_slot1(): assert res.status_code == 200, f"slot-0 restore failed: {res.status_code} {res.body}" assert res.body["n_restored"] > 0 - # The binding proof: the slot-0 restore returned while slot 1 was STILL generating. - # Before the fix it starved until slot 1 finished (done1 set); with the bind it is - # serviced the moment its own slot (0) releases. - assert not done1.is_set(), \ - "slot-0 restore only returned after the long slot-1 gen ended — it was not bound to its slot" - assert elapsed < 20, f"slot-0 restore took {elapsed:.1f}s — starved behind the long slot" - + t_ret = time.time() t1.join(timeout=DEFAULT_HTTP_TIMEOUT) tr1.join(timeout=DEFAULT_HTTP_TIMEOUT) + # The binding proof, in TIMESTAMPS (2026-09-13: "returned while slot 1 was still + # generating" was a race against the runner's speed — a tiny model ends 4096 + # tokens in ~4 s on Metal and ~2 s on hosted runners). With the bind, the slot-0 + # restore is serviced the moment ITS slot releases: it returns within a beat of + # slot 0's own generation ending. Unbound, it waited for the LONG slot: it returns + # only after slot 1 ended. So: the restore's return sits next to done0, not done1. + after_own = t_ret - t_done[0] + gap = t_done[1] - t_done[0] + if gap < 1.0: + pytest.skip(f"runner too fast to separate the slots (slot 1 ended {gap:.2f}s after slot 0)") + assert after_own < gap / 2, \ + f"slot-0 restore returned {after_own:.2f}s after its own slot freed (slot 1 ended {gap:.2f}s later) — not bound to its slot" def test_slot_restore_legacy_token_list():