From 690b95e050a220a1d4fbab0ca3aecf9b044d9f95 Mon Sep 17 00:00:00 2001 From: Jon Saad-Falcon <41205309+jonsaadfalcon@users.noreply.github.com> Date: Mon, 1 Jun 2026 13:17:38 -0700 Subject: [PATCH] perf(serve): parallelize engine discovery + async version check (#470) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `jarvis serve` startup was slow on two counts addressed here: 1. discover_engines() probed each registered engine's health() serially, and every probe is a blocking network check with its own ~2s timeout — so N dead/slow localhost ports cost N*2s. Run the probes concurrently in a ThreadPoolExecutor; the existing healthy.sort() normalizes order, so the result is identical to the serial version. health() impls are read-only on per-instance HTTP clients with no shared mutable state. 2. The PyPI update check ran a blocking urlopen (up to 3s on a cache miss) inline before dispatch, delaying every command. Move it to a daemon thread — it's best-effort and never raises. Together these remove ~10-30s from cold startup. Adds a regression test asserting discovery probes overlap (concurrency), not just that output is unchanged (covered by existing tests). Note: the larger ~30-40s win — the duplicate SystemBuilder.build() in serve.py — is NOT addressed here; it's not redundant (the second build wires a JarvisSystem the inline path never constructs) and needs a design pass. Deferred. Co-authored-by: Claude Opus 4.8 (1M context) --- src/openjarvis/cli/__init__.py | 18 ++++++++--- src/openjarvis/engine/_discovery.py | 24 +++++++++++--- tests/engine/test_discovery.py | 49 +++++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 8 deletions(-) diff --git a/src/openjarvis/cli/__init__.py b/src/openjarvis/cli/__init__.py index b1553595..40c72491 100644 --- a/src/openjarvis/cli/__init__.py +++ b/src/openjarvis/cli/__init__.py @@ -68,9 +68,21 @@ def cli(ctx: click.Context, verbose: bool, quiet: bool) -> None: research_mode_active = "--research" in sys.argv if not quiet and ctx.invoked_subcommand and not research_mode_active: + import threading + from openjarvis.cli._version_check import check_for_updates - check_for_updates(ctx.invoked_subcommand) + # Run the PyPI version poll off the hot path: on a cache miss it does + # a blocking urlopen (up to 3s) that otherwise delays every command, + # notably `jarvis serve` startup (#263). It's best-effort and never + # raises, and the nudge prints to stderr, so a daemon thread is safe — + # for long-lived commands (serve) it finishes; for short commands that + # exit first, the check is simply skipped this run (same as a miss). + threading.Thread( + target=check_for_updates, + args=(ctx.invoked_subcommand,), + daemon=True, + ).start() # First-run guard — routes bare `jarvis` to chat or init. if ctx.invoked_subcommand is None: @@ -129,9 +141,7 @@ try: except Exception as _dr_exc: import logging as _logging - _logging.getLogger(__name__).debug( - "deep-research command unavailable: %s", _dr_exc - ) + _logging.getLogger(__name__).debug("deep-research command unavailable: %s", _dr_exc) cli.add_command(self_update, "self-update") cli.add_command(bootstrap_cmd, "_bootstrap") diff --git a/src/openjarvis/engine/_discovery.py b/src/openjarvis/engine/_discovery.py index 8cec7c25..6de6c1be 100644 --- a/src/openjarvis/engine/_discovery.py +++ b/src/openjarvis/engine/_discovery.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from concurrent.futures import ThreadPoolExecutor from typing import Any, Dict, List, Tuple from openjarvis.core.config import JarvisConfig @@ -106,15 +107,30 @@ def discover_engines(config: JarvisConfig) -> List[Tuple[str, InferenceEngine]]: Results are sorted with the config default engine first. """ _maybe_register_mining_sidecar_engine() - healthy: List[Tuple[str, InferenceEngine]] = [] - for key in EngineRegistry.keys(): + + # Probe engines concurrently: each health() does a blocking network + # check with its own timeout, so a serial loop costs the SUM of all + # probe timeouts (dead localhost ports especially). Running them in + # threads collapses that to roughly the slowest single probe. The + # healthy.sort() below normalizes order, so completion order is + # irrelevant and the result is identical to the serial version (#263). + keys = list(EngineRegistry.keys()) + + def _probe(key: str) -> Tuple[str, InferenceEngine] | None: try: engine = _make_engine(key, config) if engine.health(): - healthy.append((key, engine)) + return (key, engine) except Exception as exc: logger.debug("Engine %r failed during discovery: %s", key, exc) - continue + return None + + healthy: List[Tuple[str, InferenceEngine]] = [] + if keys: + with ThreadPoolExecutor(max_workers=len(keys)) as pool: + for result in pool.map(_probe, keys): + if result is not None: + healthy.append(result) default_key = config.engine.default diff --git a/tests/engine/test_discovery.py b/tests/engine/test_discovery.py index 8d398f4d..e8642c41 100644 --- a/tests/engine/test_discovery.py +++ b/tests/engine/test_discovery.py @@ -73,6 +73,55 @@ class TestDiscoverEngines: result = discover_engines(cfg) assert result[0][0] == "b" + def test_health_checks_run_concurrently(self) -> None: + """Regression for #263 — discovery must probe engines in parallel. + + Each engine's health() does a blocking network probe with its own + timeout, so serial discovery cost = sum of probe times. With N + engines each sleeping S, parallel discovery wall-time must stay far + below N*S (closer to S). We don't measure real time precisely (CI is + noisy); instead we record concurrency: the max number of health() + calls in flight simultaneously must exceed 1. + """ + import threading + import time + + n_engines = 6 + sleep_s = 0.15 + for i in range(n_engines): + _reg(f"slow{i}", f"slow{i}") + + lock = threading.Lock() + in_flight = 0 + max_in_flight = 0 + + class _SlowEngine(_FakeEngine): + def health(self) -> bool: + nonlocal in_flight, max_in_flight + with lock: + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + time.sleep(sleep_s) + with lock: + in_flight -= 1 + return True + + cfg = JarvisConfig() + with mock.patch( + "openjarvis.engine._discovery._make_engine", + side_effect=lambda k, c: _SlowEngine(healthy=True), + ): + start = time.monotonic() + result = discover_engines(cfg) + elapsed = time.monotonic() - start + + # All slow engines were discovered (plus any real registered ones). + assert len([r for r in result if r[0].startswith("slow")]) == n_engines + # Concurrency actually happened — more than one probe overlapped. + assert max_in_flight > 1, f"probes ran serially (max_in_flight={max_in_flight})" + # Wall-time is well under the serial sum (n*sleep), allowing slack. + assert elapsed < n_engines * sleep_s * 0.7 + class TestDiscoverModels: def test_aggregate_models(self) -> None: