diff --git a/tools/server/server-models.cpp b/tools/server/server-models.cpp index 59532af65c..ced5eea132 100644 --- a/tools/server/server-models.cpp +++ b/tools/server/server-models.cpp @@ -424,6 +424,7 @@ server_models::server_models( LOG_WRN("using original argv[0] as fallback: %s\n", argv[0]); } load_models(); + debug_fake_timing = !common_get_env("LLAMA_DEBUG_FAKE_TIMING").empty(); } server_models::~server_models() = default; @@ -924,6 +925,11 @@ void server_models::load(const std::string & name) { } void server_models::load(const std::string & name, const load_options & opts) { + if (debug_fake_timing) { + // do not hold the mutex here, other requests must keep making progress + std::this_thread::sleep_for(std::chrono::seconds(2)); + } + if (!opts.custom_meta.has_value()) { if (!has_model(name)) { throw std::runtime_error("model name=" + name + " is not found"); @@ -1426,7 +1432,7 @@ bool server_models::ensure_model_ready(const std::string & name, const std::func continue; } - cv.wait_for(lk, std::chrono::milliseconds(500)); + cv.wait_for(lk, std::chrono::milliseconds(200)); } } catch (...) { leave_queue(); @@ -1452,6 +1458,10 @@ server_http_res_ptr server_models::proxy_request(const server_http_req & req, co } mapping[name].req_count++; } + if (debug_fake_timing) { + // sleep after req_count++, so the model counts as busy while we wait here + std::this_thread::sleep_for(std::chrono::seconds(2)); + } SRV_INF("proxying request to model %s on port %d\n", name.c_str(), meta->port); std::string proxy_path = req.path; if (!req.query_string.empty()) { diff --git a/tools/server/server-models.h b/tools/server/server-models.h index b1acef954b..bd45f165ca 100644 --- a/tools/server/server-models.h +++ b/tools/server/server-models.h @@ -200,6 +200,9 @@ private: // queue of requests waiting for a models_max slot std::unique_ptr sched; + // if true, add some delay to simulate works (useful for testing) + bool debug_fake_timing = false; + void update_meta(const std::string & name, const server_model_meta & meta); // unload least recently used models if the limit is reached diff --git a/tools/server/tests/unit/test_router.py b/tools/server/tests/unit/test_router.py index 94165e520e..34b4ddb71c 100644 --- a/tools/server/tests/unit/test_router.py +++ b/tools/server/tests/unit/test_router.py @@ -145,6 +145,156 @@ def test_router_models_max_evicts_lru(): assert _get_model_status(first) == "unloaded" +# lru_sched tests (relying on LLAMA_DEBUG_FAKE_TIMING) + +MODEL_A = "ggml-org/tinygemma3-GGUF:Q8_0" +MODEL_B = "ggml-org/test-model-stories260K:F32" +MODEL_C = "ggml-org/test-model-stories260K-infill:F32" + + +def _tokenize(model_id: str, timeout: float | None = DEFAULT_REQUEST_TIMEOUT) -> ServerResponse: + return server.make_request( + "POST", "/tokenize", data={"model": model_id, "content": "hello world"}, timeout=timeout + ) + + +class _Bg: + """runs one request in a thread, keeps its result, error and finish time""" + + def __init__(self, fn): + self.result = None + self.error: Exception | None = None + self.done_at: float = 0.0 + self._thread = threading.Thread(target=self._run, args=(fn,), daemon=True) + + def _run(self, fn): + try: + self.result = fn() + except Exception as e: + self.error = e + self.done_at = time.time() + + def start(self): + self._thread.start() + return self + + def join(self, timeout: int = 180): + self._thread.join(timeout) + assert not self._thread.is_alive(), "background request did not finish in time" + return self + + def assert_ok(self, what: str): + assert self.error is None, f"{what} raised {self.error!r}" + assert self.result is not None and self.result.status_code == 200, \ + f"{what} failed: {self.result.status_code if self.result else None} {self.result.body if self.result else None}" + + +def test_router_queue_does_not_evict_busy_model(): + """a request that finds no free slot waits, and the model serving a request survives it""" + global server + server.models_max = 1 + server.start() + + _load_model_and_wait(MODEL_A, timeout=120) + + busy = _Bg(lambda: _tokenize(MODEL_A)).start() + time.sleep(0.5) # let the request reach the child and take the only slot + + # no slot free and MODEL_A is busy, so this queues instead of evicting mid-request + queued = _Bg(lambda: _tokenize(MODEL_B)).start() + + busy.join() + queued.join() + + # had MODEL_A been evicted while serving, its own request would have died + busy.assert_ok("request against the busy model") + queued.assert_ok("queued request") + + _wait_for_model_status(MODEL_B, {"loaded"}, timeout=120) + assert _get_model_status(MODEL_A) == "unloaded" + + +def test_router_queue_coalesces_requests_for_same_model(): + """many requests for one missing model share a slot, so only one model is given up""" + global server + server.models_max = 2 + server.start() + + _load_model_and_wait(MODEL_A, timeout=120) + _load_model_and_wait(MODEL_B, timeout=120) + + # keep MODEL_A busy so MODEL_B is the only model that can be given up + busy = _Bg(lambda: _tokenize(MODEL_A)).start() + time.sleep(0.5) + + waiters = [_Bg(lambda: _tokenize(MODEL_C)).start() for _ in range(3)] + + busy.join() + for w in waiters: + w.join() + + busy.assert_ok("request against the busy model") + for i, w in enumerate(waiters): + w.assert_ok(f"queued request {i}") + + _wait_for_model_status(MODEL_C, {"loaded"}, timeout=120) + # one entry for 3 requests means one eviction: MODEL_B goes, MODEL_A is left alone. + # without coalescing the leftover entries still ask for a slot, + # and MODEL_A is taken too as soon as it goes idle + assert _get_model_status(MODEL_A) == "loaded" + assert _get_model_status(MODEL_B) == "unloaded" + + +def test_router_queue_client_disconnect_keeps_model(): + """a client that leaves while queued must not cost a running model its slot""" + global server + server.models_max = 1 + server.start() + + _load_model_and_wait(MODEL_A, timeout=120) + + busy = _Bg(lambda: _tokenize(MODEL_A)).start() + time.sleep(0.5) + + # queues behind MODEL_A, then gives up long before MODEL_A goes idle + with pytest.raises(requests.exceptions.RequestException): + _tokenize(MODEL_B, timeout=1) + + busy.join() + busy.assert_ok("request against the busy model") + + # nobody is waiting anymore, so MODEL_A keeps its slot + time.sleep(3) + assert _get_model_status(MODEL_A) == "loaded" + assert _get_model_status(MODEL_B) == "unloaded" + + +def test_router_queue_is_fifo(): + """the queue is served in arrival order""" + global server + server.models_max = 1 + server.start() + + _load_model_and_wait(MODEL_A, timeout=120) + + busy = _Bg(lambda: _tokenize(MODEL_A)).start() + time.sleep(0.5) + + first = _Bg(lambda: _tokenize(MODEL_B)).start() + time.sleep(1) # keep the arrival order unambiguous + second = _Bg(lambda: _tokenize(MODEL_C)).start() + + busy.join() + first.join() + second.join() + + busy.assert_ok("request against the busy model") + first.assert_ok("first queued request") + second.assert_ok("second queued request") + + assert first.done_at < second.done_at, "queue was not served in arrival order" + + def test_router_no_models_autoload(): global server server.no_models_autoload = True diff --git a/tools/server/tests/utils.py b/tools/server/tests/utils.py index ae56bc70a1..9d9e300a7d 100644 --- a/tools/server/tests/utils.py +++ b/tools/server/tests/utils.py @@ -132,7 +132,10 @@ class ServerProcess: self.external_server = "DEBUG_EXTERNAL" in os.environ def start(self, timeout_seconds: int = DEFAULT_HTTP_TIMEOUT) -> None: - env = {**os.environ} + env = { + **os.environ, + "LLAMA_DEBUG_FAKE_TIMING": "1", + } if "LLAMA_CACHE" not in os.environ: env["LLAMA_CACHE"] = "tmp" if self.external_server: