okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

dawnkernel: swap stale kernels at idle and run kernels on remote hosts over sshf835864

main
branchesmainall refs →
3.1 KiB95 linesraw
1"""Checked group termination for the erlang supervisor.
2
3Erlang runs this as its own process, so supervision keeps making progress when
4the kernel cannot (a native cell holding the GIL, a wedged interpreter), and so
5the supervisor never parses kill(1) error text or guesses an exit status. One
6argv carries the request (or `-` and one stdin line, the form run over ssh
7for a remote kernel's processes); one JSON object on stdout carries the
8verdict.
9"""
10
11from __future__ import annotations
12
13import signal
14
15import asyncio
16import json
17import os
18import sys
19
20import albedo_proc
21
22TERM_MS = 250
23KILL_MS = 1000  # the supervisor's remaining budget is short: this is the last resort
24
25
26def budget(request: dict, name: str, default: int) -> float:
27    """Seconds for one ladder step, clamped to something a caller can wait for."""
28    value = request.get(name, default)
29    return (
30        min(max(value, 10), default) / 1000
31        if isinstance(value, int)
32        else default / 1000
33    )
34
35
36def target(spec: object) -> albedo_proc.Group:
37    """One requested target: a process group, or one process that leads none."""
38    if not isinstance(spec, dict):
39        raise ValueError("target must be an object")
40    pgid = spec.get("pgid")
41    operand = pgid if isinstance(pgid, int) else spec.get("pid")
42    if not isinstance(operand, int) or operand <= 1:
43        raise ValueError("target needs a pgid or pid above 1")
44    leader = spec.get("leader")
45    whole = isinstance(pgid, int)
46    if not whole:
47        try:
48            whole = os.getpgid(operand) == operand
49        except (ProcessLookupError, PermissionError):
50            pass
51    return albedo_proc.Group(
52        operand, leader if isinstance(leader, str) else None, whole
53    )
54
55
56def supervise(request: object) -> dict[str, object]:
57    if not isinstance(request, dict):
58        raise ValueError("request must be an object")
59    specs = request.get("targets")
60    if not isinstance(specs, list):
61        raise ValueError("request needs a target list")
62    targets = [target(spec) for spec in specs]
63    endings = asyncio.run(
64        albedo_proc.terminate(
65            targets,
66            term=budget(request, "term_ms", TERM_MS),
67            kill=budget(request, "kill_ms", KILL_MS),
68        )
69    )
70    return {
71        "targets": [
72            {**ending.as_json(), "label": spec.get("label")}
73            for spec, ending in zip(specs, endings)
74        ]
75    }
76
77
78def main(request: str) -> int:
79    try:
80        answer: dict[str, object] = {"ok": True, **supervise(json.loads(request))}
81        status = 0
82    except Exception as error:
83        answer = {"ok": False, "error": f"{type(error).__name__}: {error}"}
84        status = 2
85    print(json.dumps(answer))
86    return status
87
88
89if __name__ == "__main__":
90    # Default SIGALRM terminates even a wedged interpreter; no Python handler.
91    # The owning port waits longer than this before reporting a helper timeout.
92    signal.signal(signal.SIGALRM, signal.SIG_DFL)
93    signal.alarm(3)
94    request = sys.argv[1] if len(sys.argv) > 1 else "{}"
95    sys.exit(main(sys.stdin.readline() if request == "-" else request))