diff --git a/docs/DETAILS.md b/docs/DETAILS.md index 1217072e428..6e55930bb39 100644 --- a/docs/DETAILS.md +++ b/docs/DETAILS.md @@ -296,6 +296,27 @@ Terminal chat: `.venv/bin/python chat.py`. --- +## Sharing the GPU with other programs (optional) + +By default the model stays loaded until you close Strata. On a PC that also games, renders or runs another model +server, three server options (all off by default; also as keys in `strata-.json`) give the VRAM back: + +| Option | Config key | What it does | +| --- | --- | --- | +| `--idle-unload 600` | `"idle_unload_s": 600` | unload the model after 600 s without requests; the next request loads it again | +| `--min-free-vram-mib 11000` | `"min_free_vram_mib": 11000` | load an unloaded model only when that much VRAM is free (it waits up to 15 s for memory being given back), else answer **503** "the GPU is in use by another program" instead of starting into what a game left. With several GPUs only the first one (the engine's main card) is checked | +| `--before-load "cmd"` | `"before_load": "cmd"` or `["cmd", "arg"]` | a command run before the model is loaded again, e.g. one that unloads another server's model | + +`POST /unload` unloads it now (`409` while a request is running) and `POST /load` loads it ahead of a request; +`/health` says `"loaded"`, `/v1/models` lists it as `unloaded` (like llama.cpp's router), `/props` sets +`is_sleeping` and the Monitor shows the state. Unloading ends the engine process - and the image encoder, when images +are on; it is started again first, as at a start - so their VRAM and RAM go straight back. The model files stay in +the OS file cache, so loading again takes seconds while that RAM is not needed elsewhere. Measured on an RTX 5060 Ti +16 GB with Q2_0 in the low-RAM mode: unloading takes ~0.5-0.8 s, and a request to an unloaded model answered after +4.6 s (text) or 14.7 s (a picture, image encoder on the CPU). + +--- + ## Using it The server listens on `http://127.0.0.1:8080` (change with `--port` in setup, or edit the run script). diff --git a/serve/server.py b/serve/server.py index 9393f2d58cf..257175dcfc4 100644 --- a/serve/server.py +++ b/serve/server.py @@ -26,6 +26,7 @@ import hashlib import hmac import codecs +import ctypes import json import os import queue @@ -95,6 +96,11 @@ class EngineDied(RuntimeError): """The engine process ended in the middle of a request (issue #27: on Linux, the out-of-memory killer).""" +class GpuBusy(RuntimeError): + """The model is unloaded and the GPU has less free VRAM than min_free_vram_mib: something else (a game, another + model server) is using it, so the engine is not started into the little that is left.""" + + def narrate_start(log_path: str, offset: int, args: list, done: threading.Event, heartbeat=20.0) -> None: """While the engine starts, say in the server window what it is doing, from its log: the start reads tens of GB into RAM and locks part of it for the GPU, and on many PCs everything is slow or frozen for a minute or more - @@ -170,6 +176,7 @@ def __init__(self, exe: str, args: list[str], cwd: str | None = None, log: str | stderr=self.log, text=True, encoding="utf-8", bufsize=1, env=env) contain(self.proc) # ends with the server, however it ends (Windows) self.max_context = 0 + self.unloaded = False # stopped on purpose (idle unload, POST /unload), not crashed self.can_stop = False # the engine honours a STOP line mid-request (READY stop) self.last = {} self.info = {} # INFO key=value facts (engine 0.1.8+): kv, expert slots, ... (Monitor tab) @@ -238,6 +245,27 @@ def exit_code(self): except subprocess.TimeoutExpired: return None + def unload(self): + """Stop the engine process so its VRAM and RAM go back to the system (idle unload, POST /unload); the next + request starts it again with restart(). Only between requests: the caller holds the service's fifo. + It is asked to QUIT first, as close() does; terminated, then killed, only if it does not end by itself.""" + try: + self.proc.stdin.write("QUIT\n") + self.proc.stdin.flush() + self.proc.wait(timeout=10) + except Exception: + try: + self.proc.terminate() + self.proc.wait(timeout=10) + except Exception: + self.proc.kill() + try: + self.proc.wait(timeout=10) + except Exception: + pass + self.ended = True + self.unloaded = True + def restart(self): """Start the engine again (the same command) after it died; the new process has its own line queue.""" try: @@ -397,14 +425,37 @@ def __init__(self, cfg: dict, log=None, env: dict | None = None): if cfg.get("max_tokens"): args += ["--max-tokens", str(cfg["max_tokens"])] self.dir = Path(tempfile.mkdtemp(prefix="strata-vision-")) + self.spawn = (args, log, env) # to start it again after an unload + self.stopped = False + self._start() + self.lock = threading.Lock() + self.cache: dict[str, tuple[Path, int]] = {} + + def _start(self): + args, log, env = self.spawn self.proc = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=log or subprocess.DEVNULL, text=True, encoding="utf-8", bufsize=1, env=env) contain(self.proc) line = self.proc.stdout.readline() if not line.startswith("READY"): raise RuntimeError("the vision encoder did not start: " + line.strip()) - self.lock = threading.Lock() - self.cache: dict[str, tuple[Path, int]] = {} + self.stopped = False + + def alive(self) -> bool: + return not self.stopped and self.proc.poll() is None + + def unload(self): + """Stop the encoder process (its VRAM or RAM goes back); the encoded images stay cached on disk.""" + self.close() + self.stopped = True + + def restart(self): + """Start the encoder again after an unload (or if it died); the cache of encoded images is kept.""" + try: + self.proc.kill() + except OSError: + pass + self._start() @staticmethod def load(source: str) -> bytes: @@ -585,9 +636,114 @@ def __init__(self, engine: Engine, tokenizer, template: ChatTemplate, model_name self.started_at = time.time() self.status_lock = threading.Lock() self.mcp = None # serve/mcp.py's McpHub when MCP servers are configured + # sharing the GPU with other programs (all off by default): unload the engine after this many idle seconds, + # only start it again when this much VRAM is free, and run this command first (e.g. to unload another + # server's model); the next request after an unload starts the engine again + self.idle_unload_s = 0 + self.min_free_vram_mib = 0 + self.before_load = None self.stop_ids = set(tokenizer.encode(IM_END, parse_special=True) + tokenizer.encode("<|endoftext|>", parse_special=True)) + def loaded(self) -> bool: + return not hasattr(self.engine, "alive") or self.engine.alive() + + def _vision_down(self) -> bool: + return self.vision is not None and hasattr(self.vision, "alive") and not self.vision.alive() + + def free_vram_mib(self) -> int | None: + """Free VRAM on the engine's (first) GPU, from NVML; None when it can't be read (then nothing is refused).""" + try: + from serve.telemetry import _Nvml + nv = _Nvml(int(getattr(self, "gpu_index", 0) or 0)) + if not nv.ok(): + return None + m = nv.Mem() + if nv.lib.nvmlDeviceGetMemoryInfo(nv.dev, ctypes.byref(m)) != 0: + return None + return int(m.free >> 20) + except Exception: + return None + + def ensure_loaded(self): + """Start the engine if it is not running (unloaded, or it died - issue #27), after the before_load hook and + the free-VRAM check. The caller holds self.fifo.""" + if self.loaded() and not self._vision_down(): + return + if self.before_load: + cmd = self.before_load + print(f"[strata] before loading: {cmd if isinstance(cmd, str) else ' '.join(map(str, cmd))}", flush=True) + try: + subprocess.run(cmd, shell=isinstance(cmd, str), timeout=120, stdin=subprocess.DEVNULL) + except (OSError, subprocess.SubprocessError) as e: + print(f"[strata] the before_load command failed ({e}); loading anyway", flush=True) + if self.min_free_vram_mib: + free = self.free_vram_mib() + deadline = time.time() + 15 # memory another process just gave back can take a moment + while free is not None and free < self.min_free_vram_mib and time.time() < deadline: + time.sleep(0.5) + free = self.free_vram_mib() + if free is not None and free < self.min_free_vram_mib: + raise GpuBusy(f"the GPU is in use by another program: {free} MiB of VRAM free, the model needs " + f"{self.min_free_vram_mib} (min_free_vram_mib) - it stays unloaded until that is free") + if self._vision_down(): # first, as at a start: a GPU encoder takes its VRAM before + print("[strata] starting the vision encoder again ...", flush=True) # the engine sizes its cache + self.vision.restart() + if self.loaded(): + return + if getattr(self.engine, "unloaded", False): + print("[strata] loading the model again (it was unloaded) ...", flush=True) + else: + code = self.engine.exit_code() if hasattr(self.engine, "exit_code") else None + print(f"[strata] the engine had stopped (exit code {code}); starting it again " + "(a minute or two) ...", flush=True) + self.engine.restart() + print("[strata] the engine is running again", flush=True) + + def load(self): + """POST /load and every generation request: start the engine now if it is unloaded (raises GpuBusy).""" + with self.status_lock: # a request is on its way: the idle thread must not unload it in between + self.last_request_at = time.time() + if self.loaded() and not self._vision_down(): + return + with self.fifo: + self.ensure_loaded() + + def unload(self, idle_for: float | None = None) -> str: + """Stop the engine between requests: "unloaded", "not loaded", "busy" (a request is running or waiting, or + with idle_for: one ran more recently than that) or "unsupported".""" + if not hasattr(self.engine, "unload"): + return "unsupported" + if not self.fifo.acquire(blocking=False): + return "busy" + try: + if not self.engine.alive(): + return "not loaded" + with self.status_lock: + if self.status.get("busy") or self.status.get("queued"): + return "busy" + if idle_for is not None and time.time() - (self.last_request_at or self.started_at) < idle_for: + return "busy" + self.engine.unload() + if self.vision is not None and hasattr(self.vision, "unload"): + self.vision.unload() + print(f"[strata] model unloaded{f' after {idle_for:.0f} s idle' if idle_for else ''}; " + "the next request loads it again", flush=True) + return "unloaded" + finally: + self.fifo.release() + + def start_idle_unload(self): + if not self.idle_unload_s or not hasattr(self.engine, "unload"): + return + print(f"[strata] the model unloads after {self.idle_unload_s} s without requests", flush=True) + + def loop(): + while True: + time.sleep(max(1.0, min(30.0, self.idle_unload_s / 4))) + self.unload(idle_for=self.idle_unload_s) + threading.Thread(target=loop, daemon=True).start() + def set_shared(self, defaults) -> dict: """The Chat settings every client gets for what it leaves out; {} / None = clients use their own again.""" self.shared = clean_shared_defaults(defaults) @@ -673,6 +829,8 @@ def metrics(self, all_requests=False) -> dict: state = "reading" elif s.get("busy"): state = "generating" + elif getattr(self.engine, "unloaded", False): # the flag, not alive(): /metrics never probes the process + state = "unloaded" else: state = "idle" live = {"state": state, "queued": s.get("queued", 0), "phase": s.get("phase") if s.get("busy") else None, @@ -838,13 +996,8 @@ def run(self, ids, thinking, tools, max_new, sampling, cancel) -> Iterator[tuple with self.fifo: with self.status_lock: self.status["queued"] -= 1 - if hasattr(self.engine, "alive") and not self.engine.alive(): - # issue #27: it died in an earlier request - start it again instead of failing every request - code = self.engine.exit_code() if hasattr(self.engine, "exit_code") else None - print(f"[strata] the engine had stopped (exit code {code}); starting it again " - "(a minute or two) ...", flush=True) - self.engine.restart() - print("[strata] the engine is running again", flush=True) + # issue #27: it died in an earlier request (or was unloaded) - start it again instead of failing + self.ensure_loaded() with self.status_lock: self.status.update(busy=True, phase="reading the prompt", prompt_tokens=len(ids), generated=0, started=time.time(), first_token=None, tool=None, tail="", max_tokens=max_new) @@ -1330,7 +1483,8 @@ def do_GET(self): self.wfile.write(body) elif path == "/health": self._json(200, {"status": "ok", "max_context": svc.engine.max_context, "model": svc.model, - "images": svc.vision is not None, "api_key": bool(svc.api_key)}) + "images": svc.vision is not None, "api_key": bool(svc.api_key), + "loaded": svc.loaded()}) elif path == "/status": if not self._authorized(): # #212: it shows the end of the last answer return @@ -1347,11 +1501,14 @@ def do_GET(self): self._json(200, s) elif path in ("/v1/models", "/models"): if self._authorized(): - loaded = not hasattr(svc.engine, "alive") or svc.engine.alive() + loaded = svc.loaded() model = {"id": svc.model, "object": "model", "status": {"value": "loaded"}, "meta": {"n_ctx": svc.engine.max_context}, "architecture": {"input_modalities": ["text", "image"] if svc.vision is not None else ["text"], "output_modalities": ["text"]}} + if not loaded and (svc.idle_unload_s or getattr(svc.engine, "unloaded", False)): + model["status"] = {"value": "unloaded"} # like llama-server's router: listed, loads on use + loaded = True self._json(200, {"object": "list", "data": [model] if loaded else []}) elif path == "/props": if self._authorized(): @@ -1376,8 +1533,21 @@ def do_POST(self): if path == "/settings": self._settings() return + if path == "/unload": # give the GPU back now (between requests) + r = svc.unload() + self._json(409 if r == "busy" else 200, {"status": r}) + return + if path == "/load": # load now, e.g. ahead of a request + try: + svc.load() + self._json(200, {"status": "loaded"}) + except GpuBusy as e: + self._json(503, {"error": {"type": "server_error", "message": str(e)}}) + return try: req = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}") + if path in ("/v1/chat/completions", "/v1/messages"): + svc.load() # unloaded: load first (or 503 while the GPU is busy) if path == "/v1/chat/completions": self._openai(req) elif path == "/v1/messages": @@ -1386,6 +1556,8 @@ def do_POST(self): self._json(404, {"error": {"message": "not found"}}) except ValueError as e: self._json(400, {"error": {"type": "invalid_request_error", "message": str(e)}}) + except GpuBusy as e: + self._json(503, {"error": {"type": "server_error", "message": str(e)}}) except EngineDied as e: # before the answer started (not streamed) self._json(503, {"error": {"type": "server_error", "message": f"{e}; the next request restarts it"}}) @@ -1394,7 +1566,7 @@ def _props(self): if model != svc.model: self._json(404, {"error": {"message": "model not found"}}) return - if hasattr(svc.engine, "alive") and not svc.engine.alive(): + if not svc.loaded() and not getattr(svc.engine, "unloaded", False): self._json(503, {"error": {"message": "the engine is not running"}}) return defaults = {**svc.sampling_defaults, **svc.shared} @@ -1406,7 +1578,7 @@ def _props(self): props = {"default_generation_settings": {"n_ctx": svc.engine.max_context, "params": params}, "total_slots": 1, "model_alias": svc.model, "chat_template": svc.template.source, "modalities": {"vision": svc.vision is not None}, "models_autoload": False, - "is_sleeping": False} + "is_sleeping": not svc.loaded()} if getattr(svc.engine, "model_path", None): props["model_path"] = svc.engine.model_path version = getattr(svc.engine, "info", {}).get("version") @@ -1699,6 +1871,15 @@ def main() -> int: ap.add_argument("--mcp-config", help="a JSON file with MCP servers in Claude Desktop's format ({\"mcpServers\": " "{...}}); the web app's chat can use their tools (also \"mcp_servers\" in " "the config)") + ap.add_argument("--idle-unload", type=float, default=None, metavar="SECONDS", + help="unload the model after this many seconds without requests, so other programs (games, other " + "model servers) can use the VRAM; the next request loads it again (also \"idle_unload_s\" " + "in the config; default: never)") + ap.add_argument("--min-free-vram-mib", type=int, default=None, + help="load an unloaded model only when this much VRAM is free, else answer 503 (also " + "\"min_free_vram_mib\" in the config; default: always load)") + ap.add_argument("--before-load", help="a command run before the model is loaded again (e.g. to unload another " + "server's model; also \"before_load\" in the config, a string or a list)") a = ap.parse_args() cfg = json.loads(Path(a.config).read_text(encoding="utf-8-sig")) if a.config else {} # Notepad adds a BOM if a.gpu is not None: @@ -1759,6 +1940,10 @@ def main() -> int: file=sys.stderr) return 2 svc.api_key = a.api_key or cfg.get("api_key", "") + svc.idle_unload_s = a.idle_unload if a.idle_unload is not None else float(cfg.get("idle_unload_s") or 0) + svc.min_free_vram_mib = a.min_free_vram_mib if a.min_free_vram_mib is not None else \ + int(cfg.get("min_free_vram_mib") or 0) + svc.before_load = a.before_load or cfg.get("before_load") or None svc.gpu_index = (gpu_list(cfg) or [0])[0] # the Monitor reads the card the engine runs on (issue #51) svc.gpu_indices = gpu_list(cfg) # ... or every card of a layer split (issue #112) if a.config: # the Chat settings shared with other apps, from last time @@ -1778,6 +1963,7 @@ def main() -> int: hub.start() atexit.register(hub.close) # the servers Strata started end with it httpd = serve(svc, host=a.host, port=a.port) + svc.start_idle_unload() here = "127.0.0.1" if a.host in ("0.0.0.0", "", "::") else a.host print(f"ready: http://{here}:{a.port}/v1 (OpenAI: /v1/chat/completions, Anthropic: /v1/messages, " f"context {engine.max_context} tokens{', images on' if vision else ''}" diff --git a/serve/test_server.py b/serve/test_server.py index 76aa9aa2133..9942f78a7a0 100644 --- a/serve/test_server.py +++ b/serve/test_server.py @@ -7,6 +7,7 @@ import json import os import sys +import tempfile import threading import time import unittest @@ -16,7 +17,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from serve.frontend import ChatTemplate # noqa: E402 -from serve.server import CTX_SLACK, ByteTokenizer, EngineDied, MockEngine, Service, StrataEngine, request_timings, serve # noqa: E402 +from serve.server import CTX_SLACK, ByteTokenizer, EngineDied, GpuBusy, MockEngine, Service, StrataEngine, request_timings, serve # noqa: E402 ROOT = Path(__file__).resolve().parents[1] CTX = 4096 @@ -932,5 +933,176 @@ def test_draft_fields(self): self.assertNotIn("draft_n", request_timings(24, 20, base)) self.assertIsNone(request_timings(24, 20, {})) + +class UnloadableEngine(MockEngine): + """A mock engine that can be stopped and started again like StrataEngine (alive / unload / restart).""" + + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self.running, self.unloaded, self.starts = True, False, 0 + + def alive(self): + return self.running + + def unload(self): + self.running, self.unloaded = False, True + + def restart(self): + self.running, self.unloaded = True, False + self.starts += 1 + + +class SharingTheGpu(unittest.TestCase): + """Idle unload, POST /unload and /load, the free-VRAM guard and the before_load hook (all off by default).""" + + def setUp(self): + tok = ByteTokenizer() + self.engine = UnloadableEngine(tok, "\n\nok", max_context=CTX) + self.svc = Service(self.engine, tok, ChatTemplate(ROOT / "serve/chat_template.jinja")) + self.httpd = serve(self.svc, port=0) + self.base = f"http://127.0.0.1:{self.httpd.server_address[1]}" + + def tearDown(self): + self.httpd.shutdown() + self.httpd.server_close() + + def req(self, path, body=None): + r = urllib.request.Request(self.base + path, data=None if body is None else json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, method="GET" if body is None else "POST") + try: + with urllib.request.urlopen(r, timeout=30) as resp: + return resp.status, json.loads(resp.read()) + except urllib.error.HTTPError as e: + with e: + return e.code, json.loads(e.read()) + + def chat(self): + return self.req("/v1/chat/completions", {"model": "m", "messages": [{"role": "user", "content": "hi"}]}) + + def test_unload_then_the_next_request_loads(self): + self.assertEqual(self.req("/unload", {}), (200, {"status": "unloaded"})) + self.assertFalse(self.engine.alive()) + self.assertEqual(self.req("/health")[1]["loaded"], False) + self.assertEqual(self.req("/v1/models")[1]["data"][0]["status"]["value"], "unloaded") + self.assertEqual(self.req("/unload", {}), (200, {"status": "not loaded"})) + s, b = self.chat() + self.assertEqual(s, 200) + self.assertEqual(b["choices"][0]["message"]["content"], "ok") + self.assertEqual(self.engine.starts, 1) + self.assertEqual(self.req("/health")[1]["loaded"], True) + + def test_load_endpoint(self): + self.svc.unload() + self.assertEqual(self.req("/load", {}), (200, {"status": "loaded"})) + self.assertTrue(self.engine.alive()) + self.assertEqual(self.req("/load", {}), (200, {"status": "loaded"})) + self.assertEqual(self.engine.starts, 1) + + def test_unload_refused_while_a_request_runs(self): + with self.svc.fifo: + self.assertEqual(self.svc.unload(), "busy") + self.assertTrue(self.engine.alive()) + + def test_idle_unload(self): + self.svc.idle_unload_s = 1 + self.svc.last_request_at = time.time() + self.assertEqual(self.svc.unload(idle_for=1), "busy") # a request just now: not idle yet + self.svc.start_idle_unload() + deadline = time.time() + 10 + while self.engine.alive() and time.time() < deadline: + time.sleep(0.1) + self.assertFalse(self.engine.alive()) + self.assertEqual(self.chat()[0], 200) + + def test_min_free_vram_refuses_to_load(self): + self.svc.min_free_vram_mib = 8000 + self.svc.free_vram_mib = lambda: 2000 + self.svc.unload() + t0 = time.time() + s, b = self.chat() + self.assertEqual(s, 503) + self.assertIn("in use by another program", b["error"]["message"]) + self.assertFalse(self.engine.alive()) + self.assertGreater(time.time() - t0, 10) # waited for memory being given back first + self.svc.free_vram_mib = lambda: 9000 + self.assertEqual(self.chat()[0], 200) + + def test_min_free_vram_unreadable_loads(self): + self.svc.min_free_vram_mib = 8000 + self.svc.free_vram_mib = lambda: None # no NVML: never refuse + self.svc.unload() + self.assertEqual(self.chat()[0], 200) + + def test_before_load_runs_first(self): + mark = Path(tempfile.mkdtemp()) / "ran" + self.svc.before_load = [sys.executable, "-c", f"open({str(mark)!r}, 'w').close()"] + self.svc.unload() + self.assertFalse(mark.exists()) + self.assertEqual(self.chat()[0], 200) + self.assertTrue(mark.exists()) + + def test_vision_encoder_unloads_and_starts_first(self): + order = [] + + class FakeVision: + running = True + + def alive(self): + return self.running + + def unload(self): + self.running = False + + def restart(self): + order.append("vision") + self.running = True + + engine_restart = self.engine.restart + self.engine.restart = lambda: (order.append("engine"), engine_restart()) + self.svc.vision = FakeVision() + self.assertEqual(self.svc.unload(), "unloaded") + self.assertFalse(self.svc.vision.alive()) + self.assertEqual(self.chat()[0], 200) + self.assertEqual(order, ["vision", "engine"]) # the encoder first, as at a start + self.assertTrue(self.svc.vision.alive()) + + def test_load_counts_as_activity(self): + """The idle thread must not unload between load() and the request it was called for (PR review).""" + self.svc.last_request_at = time.time() - 3600 + self.svc.load() + self.assertEqual(self.svc.unload(idle_for=60), "busy") + self.assertTrue(self.engine.alive()) + + def test_off_by_default(self): + self.assertEqual((self.svc.idle_unload_s, self.svc.min_free_vram_mib, self.svc.before_load), (0, 0, None)) + self.assertEqual(self.req("/health")[1]["loaded"], True) + + +@unittest.skipIf(os.name == "nt", "the fake engine is a script with a shebang") +class EngineUnloadQuits(unittest.TestCase): + """StrataEngine.unload() asks the engine to QUIT (as close() does) instead of terminating it straight away.""" + + def test_quit_before_terminate(self): + d = Path(tempfile.mkdtemp()) + fake, mark = d / "fake-engine", d / "how-it-ended" + fake.write_text("#!" + sys.executable + "\n" + "import signal, sys\n" + "mark = sys.argv[-1]\n" + "signal.signal(signal.SIGTERM, lambda *a: (open(mark, 'w').write('terminated'), sys.exit(1)))\n" + "print('READY 4096', flush=True)\n" + "for line in sys.stdin:\n" + " if line.strip() == 'QUIT':\n" + " open(mark, 'w').write('quit')\n" + " sys.exit(0)\n") + fake.chmod(0o755) + eng = StrataEngine(str(fake), [str(mark)]) + self.assertTrue(eng.alive()) + t0 = time.time() + eng.unload() + self.assertFalse(eng.alive()) + self.assertTrue(eng.unloaded) + self.assertEqual(mark.read_text(), "quit") + self.assertLess(time.time() - t0, 5) + if __name__ == "__main__": unittest.main()