diff options
Diffstat (limited to 'app/backends/servers.py')
| -rw-r--r-- | app/backends/servers.py | 189 |
1 files changed, 116 insertions, 73 deletions
diff --git a/app/backends/servers.py b/app/backends/servers.py index fd28866..781ecb5 100644 --- a/app/backends/servers.py +++ b/app/backends/servers.py @@ -22,6 +22,7 @@ boot screen, which renders the same events. Pid/log files live under ``app/logs/`` which is already gitignored. """ +import contextlib import os import re import signal @@ -142,6 +143,43 @@ def _pid_path(name: str) -> Path: return LOG_DIR / f"{name}-server.pid" +@contextlib.contextmanager +def _start_lock(name: str): + """Serialize a ``start``'s stale-pid cleanup, spawn and pid + publication for NAME. + + Two starters racing through ``start`` could otherwise unlink each + other's just-created, still-empty pid-file reservation (created + before the spawned pid is written) and both end up spawning a + server. The lock (an advisory fcntl/msvcrt lock, blocking so the + loser waits only as long as the winner's spawn takes) is held from + the liveness check until the pid file carries the spawned pid; the + boot wait happens outside it. The lock dies with its holder, so a + crashed starter never wedges later starts; platforms without file + locking degrade to the old unlocked behavior. + """ + LOG_DIR.mkdir(parents=True, exist_ok=True) + handle = open(LOG_DIR / f"{name}-server.start.lock", "w") + try: + try: + import fcntl + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + except OSError: + pass # degrade rather than fail the start + except ImportError: + try: + import msvcrt + handle.write("0") + handle.flush() + msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1) + except (ImportError, OSError): + pass + yield + finally: + handle.close() # releases the lock + + def _read_log_tail(name: str, lines: int = 20) -> List[str]: """Return the last LINES of the server's log (best-effort).""" try: @@ -338,10 +376,11 @@ def _kill_pid(pid: int) -> bool: except PermissionError: return False for _ in range(int(STOP_GRACE_SECONDS * 10)): - # Reap first so an exited (zombie) child ends the wait immediately - # instead of keeping the killpg(0) probe "alive" until SIGKILL. - if _reap_exited(pid): - return True + # Reap each round so an exited (zombie) child stops keeping the + # killpg(0) probe "alive" until SIGKILL. Reaping the launcher is + # NOT proof the group is gone — workers can outlive it — so the + # group probe below, not the reap, decides when the wait ends. + _reap_exited(pid) try: os.killpg(pgid, 0) except ProcessLookupError: @@ -429,82 +468,86 @@ def start(spec, progress: ProgressCallback = None, # Refuse to double-start: a live pid file means a previous start is # still booting (or its process is wedged). Spawning a second server # on the same port would orphan the first with no pid record left. - if alive(spec.name): - report({"kind": "error", - "message": f"a {spec.name} server (pid " - 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(): + # The start lock keeps the liveness check, stale cleanup, spawn and + # pid publication serialized against a concurrent starter (see + # _start_lock); the boot wait below runs outside it. + with _start_lock(spec.name): + if alive(spec.name): + report({"kind": "error", + "message": f"a {spec.name} server (pid " + 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: + pid_file.unlink() + except OSError: + pass + # Reserve the slot atomically (O_EXCL): two concurrent starters can + # both pass the liveness check above, but only one wins the create — + # the loser refuses instead of spawning a duplicate server on the port. try: - pid_file.unlink() + pid_handle = pid_file.open("x") + except FileExistsError: + report({"kind": "error", + "message": f"a {spec.name} server is already starting " + "(its pid file appeared while this start was " + "running); stop it first"}) + return False except OSError: - pass - # Reserve the slot atomically (O_EXCL): two concurrent starters can - # both pass the liveness check above, but only one wins the create — - # the loser refuses instead of spawning a duplicate server on the port. - try: - pid_handle = pid_file.open("x") - except FileExistsError: - report({"kind": "error", - "message": f"a {spec.name} server is already starting " - "(its pid file appeared while this start was " - "running); stop it first"}) - return False - except OSError: - pid_handle = None - - cwd = getattr(spec, "cwd", None) - # Append so an earlier boot's output survives (crash-loop debugging); - # the child inherits the handle and the parent's copy is closed right - # after the spawn, so nothing leaks here. - log_handle = _log_path(spec.name).open("a", encoding="utf-8") - popen_kwargs = {"stdout": log_handle, "stderr": subprocess.STDOUT} - if cwd is not None: - popen_kwargs["cwd"] = str(cwd) - if sys.platform == "win32": - popen_kwargs["creationflags"] = \ - subprocess.CREATE_NEW_PROCESS_GROUP # type: ignore[attr-defined] - else: - popen_kwargs["start_new_session"] = True - try: - proc = subprocess.Popen(argv, **popen_kwargs) - except OSError as exc: - report({"kind": "error", - "message": f"could not start server: {exc}"}) + pid_handle = None + + cwd = getattr(spec, "cwd", None) + # Append so an earlier boot's output survives (crash-loop debugging); + # the child inherits the handle and the parent's copy is closed right + # after the spawn, so nothing leaks here. + log_handle = _log_path(spec.name).open("a", encoding="utf-8") + popen_kwargs = {"stdout": log_handle, "stderr": subprocess.STDOUT} + if cwd is not None: + popen_kwargs["cwd"] = str(cwd) + if sys.platform == "win32": + popen_kwargs["creationflags"] = \ + subprocess.CREATE_NEW_PROCESS_GROUP # type: ignore[attr-defined] + else: + popen_kwargs["start_new_session"] = True + try: + proc = subprocess.Popen(argv, **popen_kwargs) + except OSError as exc: + report({"kind": "error", + "message": f"could not start server: {exc}"}) + log_handle.close() + if pid_handle is not None: + pid_handle.close() + try: + pid_file.unlink() + except OSError: + pass + return False + log_handle.write(f"\n=== boot {datetime.now():%Y-%m-%d %H:%M:%S} " + f"(pid {proc.pid}) ===\n") + log_handle.flush() log_handle.close() + + # "pid token": the process's start time where the platform provides + # one, so a recycled pid is never mistaken for our server (see + # _pid_owned). Empty token = bare-pid probing. + token = _process_start_token(proc.pid) or "" if pid_handle is not None: - pid_handle.close() try: - pid_file.unlink() + pid_handle.write(f"{proc.pid} {token}\n".strip() + "\n") + pid_handle.close() except OSError: pass - return False - log_handle.write(f"\n=== boot {datetime.now():%Y-%m-%d %H:%M:%S} " - f"(pid {proc.pid}) ===\n") - log_handle.flush() - log_handle.close() - - # "pid token": the process's start time where the platform provides - # one, so a recycled pid is never mistaken for our server (see - # _pid_owned). Empty token = bare-pid probing. - token = _process_start_token(proc.pid) or "" - if pid_handle is not None: - try: - pid_handle.write(f"{proc.pid} {token}\n".strip() + "\n") - pid_handle.close() - except OSError: - pass report({"kind": "starting", "name": spec.name, "argv": " ".join(str(a) for a in argv), "cwd": str(cwd) if cwd is not None else None, |