diff options
Diffstat (limited to 'app/backends/servers.py')
| -rw-r--r-- | app/backends/servers.py | 134 |
1 files changed, 130 insertions, 4 deletions
diff --git a/app/backends/servers.py b/app/backends/servers.py index d352717..fd28866 100644 --- a/app/backends/servers.py +++ b/app/backends/servers.py @@ -23,6 +23,7 @@ boot screen, which renders the same events. Pid/log files live under """ import os +import re import signal import subprocess import sys @@ -62,6 +63,12 @@ _BOOT_HINTS = ( "re-install the model via Configure Backends (checking the model " "installs its companion packages), or pip-install it into the " "backend's venv manually"), + # A launcher that could not take its configured port — uvicorn's bind + # failure dies outright, sglang-omni's silently moves to a random one + # (the live detection for that is _PORT_FALLBACK_RE below). + ("already in use", + "the configured port is held by another process — stop that process " + "or move this server to a free port, then start it again"), ) # Progress callback: called with an event dict. KIND is one of: @@ -70,6 +77,10 @@ _BOOT_HINTS = ( # "ready" {name, url} server is up and answering # "exited" {name, returncode, log_tail, hint} process died while booting # "timeout" {name, seconds, log_tail, hint} readiness deadline elapsed +# "port_taken" {name, taken, moved, message, log_tail} +# launcher moved the boot +# to a random port (the +# configured one was taken) # "running" {name, url} already up (no spawn) # "cancelled" {name} boot aborted via cancel # "error" {message} could not spawn the executable @@ -104,6 +115,9 @@ def _console_progress(event: dict) -> None: _print_tail(event.get("log_tail")) if event.get("hint"): print(f"[WARNING] hint: {event['hint']}") + elif kind == "port_taken": + print(f"[ERROR] {event['message']}") + _print_tail(event.get("log_tail")) elif kind == "error": print(f"[ERROR] {event['message']}") @@ -154,6 +168,62 @@ def _print_tail(tail: List[str]) -> None: print("---") +# sglang-omni's launcher, finding the requested port taken, silently binds +# a random one instead — the boot then stays healthy but unreachable at +# the configured URL (every client keeps polling the taken port until the +# start timeout). Scanned from the boot log so the boot fails in seconds +# with both port numbers named instead. +_PORT_FALLBACK_RE = re.compile( + r"Port (?P<taken>\d+) is already in use.*?" + r"Using port (?P<moved>\d+) instead", re.DOTALL) + +# How much of the previous log read is rescanned with the next chunk, so a +# fallback message split across two reads still matches (the launcher +# prints its two lines back to back; half a kilobyte is generous). +_LOG_CARRY_BYTES = 512 + + +def _read_new_log(path: Path, offset: int) -> tuple: + """Read the bytes appended to the server log since OFFSET (best effort). + + Returns ``(new_offset, text)`` — where to resume and what was read + (undecodable bytes replaced). An unreadable or unchanged file yields + the offset unchanged and ""; a shrunken file (truncated or rotated) + restarts from zero so nothing new is skipped. + """ + try: + size = path.stat().st_size + except OSError: + return offset, "" + if size < offset: + offset = 0 + if size == offset: + return offset, "" + try: + with path.open("rb") as fh: + fh.seek(offset) + raw = fh.read() + except OSError: + return offset, "" + return offset + len(raw), raw.decode("utf-8", errors="replace") + + +def _port_taken_message(spec) -> str: + """The refusal message for spawning while SPEC's URL already answers.""" + return (f"another process is listening at {spec.url} but it is not a " + f"usable {spec.name} server — stop that process (or move this " + f"server to a free port) and start again") + + +def _port_fallback_message(spec, taken: str, moved: str) -> str: + """The boot-failure message when the launcher moved ports on us.""" + return (f"the {spec.name} launcher moved the server from port " + f"{taken} (already in use by another process) to port {moved}; " + f"clients poll {taken}, so this boot cannot become ready — " + f"stop whatever holds port {taken} (or move this server to a " + f"free port) and start again") + + def _pid_alive(pid: int) -> bool: """True when a process with PID is still running (POSIX signal-0 probe).""" if sys.platform == "win32": @@ -286,15 +356,19 @@ def _kill_pid(pid: int) -> bool: return True -def _server_ready(spec) -> bool: +def _server_ready(spec, listening: Optional[bool] = None) -> bool: """True when the server described by SPEC is usable, not just listening. Without an IDENTITY this is the plain TCP-connect check. With one, the server must also answer HTTP as that backend (``probe.identify_server``); for the faster backend (whose model loads after the port opens) the - ``/health`` model_loaded flag must additionally be true. + ``/health`` model_loaded flag must additionally be true. LISTENING, when + given, is a caller's fresh ``common.server_running`` result — reused so + one start pass probes the port only once. """ - if not common.server_running(spec.url): + if listening is None: + listening = common.server_running(spec.url) + if not listening: return False identity = getattr(spec, "identity", None) if identity is None: @@ -320,6 +394,13 @@ def start(spec, progress: ProgressCallback = None, exit reports the log tail and returns False. A no-op (True) when the server is already running. + Two port-conflict guards fail fast instead of letting a doomed boot + run out the clock: a foreign process already listening on the spec's + URL (but not answering as the backend) refuses the spawn outright, and + a launcher that logs a port fallback mid-boot ("Port N is already in + use ... Using port M instead" — sglang-omni's) aborts the boot with a + "port_taken" event and kills the misdirected server. + PROGRESS, when given, receives each boot event (see ProgressCallback); the default ``_console_progress`` prints them, preserving the old console output. CANCEL (a threading.Event) aborts the boot: the spawned process @@ -333,7 +414,11 @@ def start(spec, progress: ProgressCallback = None, "message": f"server executable not found: {exe}. Run 'Set " "up a backend' first."}) return False - if _server_ready(spec): + # One TCP probe feeds both checks: an already-usable server takes the + # "running" path, and a listener that is NOT usable (identity probe + # failed) is exactly the foreign-holder conflict refused below. + listening = common.server_running(spec.url) + if _server_ready(spec, listening): report({"kind": "running", "name": spec.name, "url": spec.url}) return True @@ -350,6 +435,15 @@ def start(spec, progress: ProgressCallback = None, f"{pid_for(spec.name)}) is already starting or " "running; stop it first"}) return False + # Refuse to spawn onto a port a foreign process already holds: TCP-up + # but probe-down means the listener is not a usable instance of this + # server. A fresh spawn would then either die on the bind or (launchers + # that fall back silently, like sglang-omni) move to a random port and + # leave every client polling the taken one — the boot watchdog below + # catches that late, so name the conflict here. + if listening: + report({"kind": "error", "message": _port_taken_message(spec)}) + return False pid_file = _pid_path(spec.name) if pid_file.exists(): try: @@ -419,6 +513,13 @@ def start(spec, progress: ProgressCallback = None, started = time.time() next_heartbeat = started + 15 deadline = started + start_timeout + # Boot-log watchdog state: scan only what this boot appends (the log + # file accumulates across boots, so a previous boot's fallback lines + # must not re-fire here), carrying a tail between reads so a fallback + # message split across two polls still matches. + log_path = _log_path(spec.name) + log_offset = log_path.stat().st_size if log_path.exists() else 0 + log_carry = "" while time.time() < deadline: if cancel is not None and cancel.is_set(): # User cancelled while booting: kill what we spawned (the @@ -441,6 +542,31 @@ def start(spec, progress: ProgressCallback = None, except OSError: pass return False + # Watchdog: a launcher that silently moved to another port (the + # configured one was taken) keeps booting healthily where no + # client will ever call it — abort now instead of polling the + # taken port until the timeout. + log_offset, new_text = _read_new_log(log_path, log_offset) + if new_text: + scan = log_carry + new_text + match = _PORT_FALLBACK_RE.search(scan) + if match is not None: + # Kill the misdirected server: it would serve on a port + # no client will call while holding GPU memory. + _kill_pid(proc.pid) + try: + pid_file.unlink() + except OSError: + pass + report({"kind": "port_taken", "name": spec.name, + "taken": int(match.group("taken")), + "moved": int(match.group("moved")), + "message": _port_fallback_message( + spec, match.group("taken"), + match.group("moved")), + "log_tail": _read_log_tail(spec.name)}) + return False + log_carry = scan[-_LOG_CARRY_BYTES:] if _server_ready(spec): report({"kind": "ready", "name": spec.name, "url": spec.url}) return True |