From fbcd4e8c8ee2035f66a5657dd03606d6fcfa982b Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Fri, 2 Oct 2026 11:27:51 +0300 Subject: [PATCH] kernel: a restarted daemon re-counts held heavy slots and ends a dead kernel's jobs #59: the hello reports jobs holding a granted slot, and the new pool claims them; kernel_links records the job groups a kernel owns, so an attach that finds the kernel gone still ends them. Darwin now gets a leader token (start time via sysctl), so a reused pid is never signalled. kernel.md spells out why replayed kernel frames are harmless after a crash restart. Signed-off-by: dawn <90008@klbr.net> --- priv/python/albedo_kernel.py | 1 + priv/python/albedo_link.py | 12 +++- priv/python/albedo_proc.py | 20 +++++- robot-docs/kernel.md | 27 +++++--- .../extensions/python/albedo_job_slots.erl | 13 +++- .../extensions/python/albedo_python.erl | 65 +++++++++++++++---- .../harness/extensions/python/kernel.gleam | 3 + .../harness/extensions/python/link.gleam | 53 +++++++++++---- test/e2e/kernel_detach_test.py | 50 +++++++++++++- .../harness/albedo_job_slots_test_support.erl | 4 +- test/harness/kernel_link_test.gleam | 1 + test/harness/process_supervision_test.py | 9 ++- 12 files changed, 210 insertions(+), 48 deletions(-) diff --git a/priv/python/albedo_kernel.py b/priv/python/albedo_kernel.py index a522ac81..f46da576 100644 --- a/priv/python/albedo_kernel.py +++ b/priv/python/albedo_kernel.py @@ -223,6 +223,7 @@ def deliver(message: albedo_api.Incoming) -> None: if entry is not None and not entry[0].done(): slot, on_queued = entry if message.get("ok") is True: + LINK.jobs.hold(message["id"]) slot.set_result(None) elif message.get("queued") is True: on_queued() diff --git a/priv/python/albedo_link.py b/priv/python/albedo_link.py index a9c74bd6..fc92dbec 100644 --- a/priv/python/albedo_link.py +++ b/priv/python/albedo_link.py @@ -160,14 +160,20 @@ class JobBook: """Job groups this kernel still owns, read from the frames it sends. The same bookkeeping the daemon keeps: a started job's group is owned - until a job frame proves it gone. A reattaching daemon learns them from - the hello, and the kernel stays alive past its grace while any are live. + until a job frame proves it gone, and so is the heavy-job slot it was + granted. A reattaching daemon learns both from the hello, and the kernel + stays alive past its grace while any job is live. """ def __init__(self) -> None: self.started: dict[str, dict[str, object]] = {} + self.held: set[str] = set() self.external = 0 + def hold(self, id: str) -> None: + """The daemon granted this job a slot.""" + self.held.add(id) + def observe(self, frame: dict[str, object]) -> None: kind, id = frame.get("type"), frame.get("id") if kind == "job_start" and isinstance(id, str): @@ -178,6 +184,7 @@ class JobBook: cleanup = frame.get("cleanup") if isinstance(cleanup, dict) and cleanup.get("gone") is True: self.started.pop(id, None) + self.held.discard(id) elif kind == "jobs": live = frame.get("live") if isinstance(live, int) and not isinstance(live, bool) and live >= 0: @@ -324,6 +331,7 @@ class SocketLink: "epoch": self.epoch, "ack": self.inbound.last, "jobs": list(self.jobs.started.values()), + "held": sorted(self.jobs.held), "external": self.jobs.external, "dropped": self.outbox.dropped, } diff --git a/priv/python/albedo_proc.py b/priv/python/albedo_proc.py index 520487b2..c03441e4 100644 --- a/priv/python/albedo_proc.py +++ b/priv/python/albedo_proc.py @@ -12,14 +12,17 @@ from collections.abc import Callable, Sequence from dataclasses import dataclass from typing import Any import asyncio +import ctypes import os import signal +import struct import subprocess import sys TERM_GRACE = 0.25 # seconds a group gets to honour TERM KILL_GRACE = 2.0 # seconds a group gets to die after KILL PROBE_INTERVAL = 0.02 +LIBC = ctypes.CDLL(None) def reap_stopped_safely(loop: asyncio.AbstractEventLoop) -> None: @@ -66,7 +69,7 @@ def reap_stopped_safely(loop: asyncio.AbstractEventLoop) -> None: class Group: """A target to end: a whole process group, or one process that leads none. - `leader` identifies the leader where /proc exists, so a reused id is refused. + `leader` identifies the leader by its start time, so a reused id is refused. """ pgid: int @@ -116,12 +119,25 @@ def start_token(stat: bytes) -> str | None: def leader_token(pid: int) -> str | None: - """Identity token for a live pid, where the platform exposes /proc.""" + """Identity token for a live pid: its start time, from /proc or, on + Darwin, its process table entry. None where neither can tell.""" try: with open(f"/proc/{pid}/stat", "rb") as stat: return start_token(stat.read()) except OSError: + return darwin_start(pid) if sys.platform == "darwin" else None + + +def darwin_start(pid: int) -> str | None: + """A Darwin process's start time: the timeval that opens its kinfo_proc + (sysctl CTL_KERN, KERN_PROC, KERN_PROC_PID). None for a pid not in use.""" + mib = (ctypes.c_int * 4)(1, 14, 1, pid) + buffer = ctypes.create_string_buffer(1024) + size = ctypes.c_size_t(len(buffer)) + if LIBC.sysctl(mib, 4, buffer, ctypes.byref(size), None, 0) != 0 or size.value < 12: return None + seconds, micros = struct.unpack_from("=qi", buffer.raw) + return f"{seconds}.{micros:06d}" def live_members(pgid: int) -> list[int] | None: diff --git a/robot-docs/kernel.md b/robot-docs/kernel.md index b24a9577..0845e487 100644 --- a/robot-docs/kernel.md +++ b/robot-docs/kernel.md @@ -38,13 +38,15 @@ the daemon writes `{"attach": {"kernel", "token", "ack", "grace"}}` first. the kernel checks the token (kept in the daemon's sqlite), bumps its epoch, cuts off any older connection (newest attach wins), and answers `{"hello": {protocol, bundle, epoch, ack, pid, pgid, leader, ready, jobs, -external, slots, dropped}}`, or `{"refused": reason}`. `protocol` (1) and the +held, external, slots, dropped}}`, or `{"refused": reason}`. `protocol` (1) and the hello/snapshot/shutdown frames never change shape. `bundle` is `albedo_bundle.digest()`, the same content hash the remote plugin stages under; a hello whose bundle or protocol differs from the bridge's makes the -kernel stale (see version skew). `jobs` (live `job_start` frames), `external` -and `slots` (job slots still waited for) let a fresh port owner take over job -ownership and admission. +kernel stale (see version skew). `jobs` (live `job_start` frames), `external`, +`held` (jobs granted a heavy slot whose end isn't proven yet) and `slots` (job +slots still waited for) let a fresh port owner take over job ownership and +admission: held slots count in its pool at once (`albedo_job_slots:claim`, +even past the limit), waited-for ones are asked for again. ## session layer @@ -61,7 +63,7 @@ a bare `{"ack": m}`. `ack` is cumulative: everything up to it arrived. - **daemon side** (`python/link.gleam`, tables from the python extension's `kernel_links` schema migration): `kernel_links` (one row per session: kernel id, token, run dir, workspace, modules, pid/pgid/leader, epoch, last seq - sent), `kernel_outbox` keyed (session, kernel, seq), bounded at 2048 frames / + sent, and the job groups it owns, rewritten as they change), `kernel_outbox` keyed (session, kernel, seq), bounded at 2048 frames / 64 MiB per kernel, and `kernel_calls`. the port owner persists each frame before writing it, deletes what the kernel acknowledges, and acknowledges kernel frames within 100 ms. its own last-seen seq lives in memory: after a @@ -78,12 +80,21 @@ dedupe: - execute, by cell id: a finished cell (the newest 16) resends its `done`; a queued or running one is ignored. callers must not reuse cell ids. - interrupt, reply, release are idempotent already. +- what the kernel resends after a restart: job wakes are host calls + (`jobs.completed`), so the call ledger covers them; `job_start`, `job`, + `job_cancel` and `jobs` only update ownership and slot bookkeeping, which + converges; `job_acquire` asks the new pool, which never saw it; a `done` + nobody waits for rewrites the same outcome; traces save by cell id. ## drops - the bridge dies: the port owner reattaches with backoff (50 ms doubling to 2 s, 40 tries); exit 3 or a refusal means the kernel is gone and the old - abandon path reaps it. the cell keeps running meanwhile. + abandon path reaps it. a port owner attaching after a restart starts from + the record's pid/pgid/leader and groups, so a kernel that died hard while + nobody was attached still has its process group and its jobs' groups + ended. the leader token (start time, from `/proc` or Darwin's `sysctl`) + keeps a reused pid from being signalled. the cell keeps running meanwhile. - the cell deadline passes, or an interrupt goes unanswered for 2 s, while detached: the interrupt waits in the outbox and the caller gets `kernel.Detached`, which leaves the journaled cell `started` (the model is @@ -267,9 +278,5 @@ away, or the session moves. ## not yet -- after a daemon restart, job groups of a kernel that died hard (SIGKILL) while - nobody was attached are not reaped: the recorded pid could be reused. -- job slots a reattached kernel's running jobs held are not re-counted in the - new daemon's pool; only slots still waited for are asked for again. - profile output a remote login shell prints reaches the daemon's own stderr log on every remote command. diff --git a/src/albedo/harness/extensions/python/albedo_job_slots.erl b/src/albedo/harness/extensions/python/albedo_job_slots.erl index eebe2166..0c664e79 100644 --- a/src/albedo/harness/extensions/python/albedo_job_slots.erl +++ b/src/albedo/harness/extensions/python/albedo_job_slots.erl @@ -18,7 +18,7 @@ %% same. -module(albedo_job_slots). -behaviour(gen_server). --export([ensure/0, ensure/2, acquire/2, release/2, release_owner/1, limit/0]). +-export([ensure/0, ensure/2, acquire/2, claim/2, release/2, release_owner/1, limit/0]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2]). -record(state, {base, running = 0, entries = #{}, waiting = [], load = false, ticking = false, hosts = #{}}). @@ -37,6 +37,9 @@ ensure() -> ensure(Host, Cpus) -> gen_server:call(ensure(), {host, Host, Cpus}). acquire(Pool, Id) -> gen_server:cast(Pool, {acquire, self(), Id}). +%% A slot a running job already holds, granted by an earlier daemon: it counts +%% at once, even past the limit, and nothing is sent back. +claim(Pool, Id) -> gen_server:cast(Pool, {claim, self(), Id}). release(Pool, Id) -> gen_server:cast(Pool, {release, {self(), Id}}). release_owner(Pool) -> gen_server:cast(Pool, {release_owner, self()}). @@ -100,6 +103,14 @@ handle_cast({acquire, Owner, Id}, S = #state{entries = Entries}) -> end, {noreply, tick(S1)} end; +handle_cast({claim, Owner, Id}, S = #state{entries = Entries, running = Running}) -> + Key = {Owner, Id}, + case maps:is_key(Key, Entries) of + true -> {noreply, S}; + false -> + Mon = monitor(process, Owner), + {noreply, S#state{entries = Entries#{Key => {active, Mon}}, running = Running + 1}} + end; handle_cast({release, Key}, S) -> {noreply, drain(drop(Key, S))}; handle_cast({base, Cpus}, S) -> {noreply, drain(S#state{base = remote_base(Cpus)})}; handle_cast({release_owner, Owner}, S) -> diff --git a/src/albedo/harness/extensions/python/albedo_python.erl b/src/albedo/harness/extensions/python/albedo_python.erl index 9990238c..62ffa401 100644 --- a/src/albedo/harness/extensions/python/albedo_python.erl +++ b/src/albedo/harness/extensions/python/albedo_python.erl @@ -40,7 +40,7 @@ -define(GRACE_MARGIN_S, 60). %% past a remote kernel's grace before it is given up %% Boot is the Gleam kernel.Boot record; Link is kernel.Link. -start({boot, Owner, Python, Bridge, Cwd, Host, Modules, Link, RunDir, Kernel, Token, Grace, OutSeq, Fresh, Remote}) -> +start({boot, Owner, Python, Bridge, Cwd, Host, Modules, Link, RunDir, Kernel, Token, Grace, OutSeq, Owned, Fresh, Remote}) -> Parent = self(), Ref = make_ref(), {Pid, Mon} = spawn_monitor(fun() -> process_flag(trap_exit, true), @@ -50,8 +50,11 @@ start({boot, Owner, Python, Bridge, Cwd, Host, Modules, Link, RunDir, Kernel, To none -> albedo_job_slots:ensure() end, monitor(process, Pool), + %% What the record says the kernel owned, so a kernel found dead at + %% this attach still has its groups ended. + {Target, Groups} = owned(Owned), S = #{port => none, host => Host, link => Link, owner => Owner, active => none, events => [], - groups => #{}, external => 0, pool => Pool, slots => #{}, target => none, + groups => Groups, external => 0, pool => Pool, slots => #{}, target => Target, python => Python, bridge => Bridge, cwd => Cwd, remote => Remote, modules => Modules, run_dir => RunDir, kernel => Kernel, token => Token, grace => Grace, in => 0, acked => 0, kack => 0, out => OutSeq, @@ -414,9 +417,12 @@ hello(Hello, S0) -> Live when is_integer(Live), Live >= 0 -> S1#{external => Live}; _ -> S1 end, - Waiting = [Id || Id <- maps:get(<<"slots">>, Hello, []), is_binary(Id), - not maps:is_key(Id, maps:get(slots, S2))], - lists:foldl(fun acquire_slot/2, S2, Waiting). + %% Slots its running jobs hold count again in this daemon's pool; the + %% ones still waited for are asked for again. + Untracked = fun(Field) -> [Id || Id <- maps:get(Field, Hello, []), is_binary(Id), + not maps:is_key(Id, maps:get(slots, S2))] end, + S3 = lists:foldl(fun claim_slot/2, S2, Untracked(<<"held">>)), + lists:foldl(fun acquire_slot/2, S3, Untracked(<<"slots">>)). %% A kernel speaking another protocol, or running another bundle than the %% bridge that reached it, is stale: the session swaps it at its next idle @@ -485,6 +491,10 @@ acquire_slot(Id, S = #{pool := Pool, slots := Slots}) -> albedo_job_slots:acquire(Pool, Id), S#{slots => Slots#{Id => true}}. +claim_slot(Id, S = #{pool := Pool, slots := Slots}) -> + albedo_job_slots:claim(Pool, Id), + S#{slots => Slots#{Id => true}}. + slot_reply(Id, Result, S = #{slots := Slots}) -> case maps:is_key(Id, Slots) of false -> S; %% cancelled before the grant reached us @@ -513,10 +523,36 @@ host_call(Host, Message) -> %% shutdown drains: a started job's group is recorded; a job proven gone %% releases its slot and its group. start_job(Message, Id, S = #{groups := Groups}) -> - S#{groups => Groups#{Id => group_of(Message)}}. + owns(S, Groups#{Id => group_of(Message)}). gone(Id, S = #{groups := Groups}) -> - (release_slot(Id, S))#{groups => maps:remove(Id, Groups)}. + owns(release_slot(Id, S), maps:remove(Id, Groups)). + +%% The groups are recorded as they change, so a daemon that finds the kernel +%% dead after a restart can still end them. +owns(S = #{groups := Groups}, Groups) -> S; +owns(S, Groups) -> + link_own(S, json:encode(maps:map(fun(_, Spec) -> maps:map(fun nil_null/2, Spec) end, Groups))), + S#{groups => Groups}. + +nil_null(_, nil) -> null; +nil_null(_, V) -> V. + +%% The record's {pid, pgid, leader, groups}: the kernel's identity from its +%% last hello and the job groups it owned. +owned(Owned) -> + M = try json:decode(Owned) catch _:_ -> #{} end, + Spec = fun(Fields) -> maps:map(fun(_, null) -> nil; (_, V) -> V end, + maps:with([<<"pid">>, <<"pgid">>, <<"leader">>], Fields)) end, + Target = case maps:get(<<"pid">>, M, null) of + Pid when is_integer(Pid), Pid > 1 -> target_of(Spec(M)); + _ -> none + end, + Groups = case maps:get(<<"groups">>, M, #{}) of + G when is_map(G) -> #{Id => group_of(Spec(Fields)) || Id := Fields <- G, is_map(Fields)}; + _ -> #{} + end, + {Target, Groups}. %% The newest 100 job and trace frames, what events/1 hands out. journal(Message, S = #{events := Events}) -> @@ -552,19 +588,20 @@ answer_detached(S) -> S. %% The Gleam link: the durable outbox and host-call ledger. A storage failure %% is logged and never stops the kernel; the frame still goes out. Once the %% store is gone (its owner died) there is nothing left to record. -link_persist(S = #{link := {link, Persist, _, _, _, _, _, _}}, Seq, Frame) -> guarded(S, fun() -> Persist(Seq, Frame) end). -link_ack(S = #{link := {link, _, Ack, _, _, _, _, _}}, Upto) -> guarded(S, fun() -> Ack(Upto) end). -link_pending(S = #{link := {link, _, _, Pending, _, _, _, _}}) -> +link_persist(S = #{link := {link, Persist, _, _, _, _, _, _, _}}, Seq, Frame) -> guarded(S, fun() -> Persist(Seq, Frame) end). +link_ack(S = #{link := {link, _, Ack, _, _, _, _, _, _}}, Upto) -> guarded(S, fun() -> Ack(Upto) end). +link_pending(S = #{link := {link, _, _, Pending, _, _, _, _, _}}) -> case guarded(S, Pending) of Frames when is_list(Frames) -> Frames; _ -> [] end. -link_call(S = #{link := {link, _, _, _, Call, _, _, _}}, Id) -> +link_call(S = #{link := {link, _, _, _, Call, _, _, _, _}}, Id) -> case guarded(S, fun() -> Call(Id) end) of State when State =:= fresh; State =:= answered; State =:= unknown -> State; _ -> fresh end. -link_reply(S = #{link := {link, _, _, _, _, Reply, _, _}}, Id, Seq, Frame) -> guarded(S, fun() -> Reply(Id, Seq, Frame) end). -link_record(S = #{link := {link, _, _, _, _, _, Record, _}}, Fields) -> +link_reply(S = #{link := {link, _, _, _, _, Reply, _, _, _}}, Id, Seq, Frame) -> guarded(S, fun() -> Reply(Id, Seq, Frame) end). +link_record(S = #{link := {link, _, _, _, _, _, Record, _, _}}, Fields) -> guarded(S, fun() -> Record(iolist_to_binary(json:encode(maps:filter(fun(_, V) -> V =/= nil end, Fields)))) end). -link_forget(S = #{link := {link, _, _, _, _, _, _, Forget}}) -> guarded(S, Forget). +link_forget(S = #{link := {link, _, _, _, _, _, _, Forget, _}}) -> guarded(S, Forget). +link_own(S = #{link := {link, _, _, _, _, _, _, _, Own}}, Groups) -> guarded(S, fun() -> Own(iolist_to_binary(Groups)) end). guarded(#{owner := Owner}, Fun) -> case is_process_alive(Owner) of diff --git a/src/albedo/harness/extensions/python/kernel.gleam b/src/albedo/harness/extensions/python/kernel.gleam index ddda8d26..2d8ee47f 100644 --- a/src/albedo/harness/extensions/python/kernel.gleam +++ b/src/albedo/harness/extensions/python/kernel.gleam @@ -77,6 +77,7 @@ type Boot { token: String, grace: Int, out_seq: Int, + owned: String, fresh: Bool, remote: Option(Remote), ) @@ -405,6 +406,7 @@ fn boot_fresh( cwd: cwd, modules: modules, out_seq: 0, + owned: "{}", ) use _ <- result.try( link.create(store, record) |> result.map_error(Unavailable), @@ -469,6 +471,7 @@ fn booter( token: record.token, grace: grace(), out_seq: record.out_seq, + owned: record.owned, fresh: fresh, remote: remote, )) diff --git a/src/albedo/harness/extensions/python/link.gleam b/src/albedo/harness/extensions/python/link.gleam index 789d5bd9..68b20d60 100644 --- a/src/albedo/harness/extensions/python/link.gleam +++ b/src/albedo/harness/extensions/python/link.gleam @@ -1,12 +1,13 @@ //// The daemon's half of a detached kernel's durable session layer. //// //// One row per session names its kernel: where its run directory is, the -//// token an attach presents, the workspace and modules it booted with, and -//// the last sequence number the daemon sent it. Frames the daemon sent and -//// the kernel has not acknowledged stay in `kernel_outbox`, so a reattach -//// (or a restarted daemon) resends them. `kernel_calls` remembers every host -//// call the kernel made until the kernel has its reply, so a replayed call -//// never runs twice. +//// token an attach presents, the workspace and modules it booted with, the +//// last sequence number the daemon sent it, and the process groups it owns, +//// so a kernel found dead after a restart still has its jobs ended. Frames +//// the daemon sent and the kernel has not acknowledged stay in +//// `kernel_outbox`, so a reattach (or a restarted daemon) resends them. +//// `kernel_calls` remembers every host call the kernel made until the kernel +//// has its reply, so a replayed call never runs twice. import albedo/daemon/store import gleam/dynamic/decode @@ -31,6 +32,9 @@ pub type Record { cwd: String, modules: String, out_seq: Int, + /// `{pid, pgid, leader, groups}`: the kernel's identity from its last + /// hello and its job groups by job id, as JSON. + owned: String, ) } @@ -54,17 +58,18 @@ pub type Link { reply: fn(String, Int, String) -> Nil, record: fn(String) -> Nil, forget: fn() -> Nil, + own: fn(String) -> Nil, ) } pub fn apply(db: sqlight.Connection) -> Result(Nil, String) { - store.exec( + use _ <- result.try(store.exec( db, "CREATE TABLE IF NOT EXISTS kernel_links ( session TEXT PRIMARY KEY, kernel TEXT NOT NULL UNIQUE, token TEXT NOT NULL, run_dir TEXT NOT NULL, cwd TEXT NOT NULL, modules TEXT NOT NULL, pid INTEGER, pgid INTEGER, leader TEXT, epoch INTEGER NOT NULL DEFAULT 0, - out_seq INTEGER NOT NULL DEFAULT 0, + out_seq INTEGER NOT NULL DEFAULT 0, groups TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))); CREATE TABLE IF NOT EXISTS kernel_outbox ( session TEXT NOT NULL, kernel TEXT NOT NULL, seq INTEGER NOT NULL, @@ -72,7 +77,10 @@ pub fn apply(db: sqlight.Connection) -> Result(Nil, String) { CREATE TABLE IF NOT EXISTS kernel_calls ( kernel TEXT NOT NULL, call TEXT NOT NULL, reply_seq INTEGER, PRIMARY KEY (kernel, call));", - ) + )) + store.add_columns(db, "kernel_links", [ + #("groups", "TEXT NOT NULL DEFAULT '{}'"), + ]) } /// Record a kernel about to boot, replacing whatever the session had. @@ -107,10 +115,20 @@ fn record_decoder() -> decode.Decoder(Record) { use cwd <- decode.field(4, decode.string) use modules <- decode.field(5, decode.string) use out_seq <- decode.field(6, decode.int) - decode.success(Record(session, kernel, token, run_dir, cwd, modules, out_seq)) + use owned <- decode.field(7, decode.string) + decode.success(Record( + session, + kernel, + token, + run_dir, + cwd, + modules, + out_seq, + owned, + )) } -const columns = "session,kernel,token,run_dir,cwd,modules,out_seq" +const columns = "session,kernel,token,run_dir,cwd,modules,out_seq,json_object('pid',pid,'pgid',pgid,'leader',leader,'groups',json(groups))" /// The kernel a session last booted, if one is recorded. pub fn find(storage: store.Store, session: String) -> Option(Record) { @@ -189,9 +207,22 @@ pub fn bind(storage: store.Store, record: Record) -> Link { reply: fn(id, seq, frame) { log(reply(storage, record, id, seq, frame)) }, record: fn(fields) { log(identify(storage, record, fields)) }, forget: fn() { log(forget(storage, record)) }, + own: fn(groups) { log(own(storage, record, groups)) }, ) } +/// The job groups the kernel owns now, by job id. +fn own( + storage: store.Store, + record: Record, + groups: String, +) -> Result(Nil, String) { + store.write(storage, "UPDATE kernel_links SET groups=? WHERE kernel=?", [ + sqlight.text(groups), + sqlight.text(record.kernel), + ]) +} + fn log(outcome: Result(Nil, String)) -> Nil { case outcome { Ok(_) -> Nil diff --git a/test/e2e/kernel_detach_test.py b/test/e2e/kernel_detach_test.py index 5cd14a67..f37bfd58 100644 --- a/test/e2e/kernel_detach_test.py +++ b/test/e2e/kernel_detach_test.py @@ -3,6 +3,8 @@ The kernel runs detached behind a bridge process; the daemon reattaches to it after a restart (graceful or crash) and after the bridge dies mid-cell, so the namespace, the kernel process, and an in-flight cell's result all survive. +What the kernel owns survives with it: a reattached kernel's running job keeps +its heavy-job slot, and one found dead after a restart has its jobs ended. """ import json @@ -27,6 +29,14 @@ def bridges(home): ] +def alive(pid): + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + return True + + @exclusive class KernelDetachTests(unittest.TestCase): def setUp(self): @@ -39,7 +49,16 @@ class KernelDetachTests(unittest.TestCase): self.provider = Provider(script) self.addCleanup(self.provider.close) - self.app = Albedo(self.provider, protocol="responses") + + def prepare(app): + # One heavy slot, granted to a job half a second in. + app.daemon.env.update( + ALBEDO_MAX_LOCAL_JOBS="1", + ALBEDO_JOB_LOAD="0", + ALBEDO_JOB_GRACE_SECONDS="0.5", + ) + + self.app = Albedo(self.provider, protocol="responses", prepare=prepare) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session() @@ -91,6 +110,35 @@ class KernelDetachTests(unittest.TestCase): (replacement,) = bridges(self.app.home) self.assertNotEqual(replacement, bridge) + def test_a_reattached_kernels_running_job_keeps_its_slot(self): + held = self.cell( + "import asyncio\na = run('sleep', '60')\nawait asyncio.sleep(1.5)\na.queued" + ) + self.assertEqual(held["value"], "False") + self.app.restart() + # The new daemon's pool counts the slot a's job already holds. + waiting = self.cell( + "b = run('sleep', '60')\nawait asyncio.sleep(1.5)\n(a.queued, b.queued)" + ) + self.assertEqual(waiting["value"], "(False, True)") + freed = self.cell("await a.stop()\nawait asyncio.sleep(1)\nb.queued") + self.assertEqual(freed["value"], "False") + self.cell("await b.stop()") + + def test_a_kernel_killed_while_the_daemon_was_down_has_its_jobs_ended(self): + started = self.cell( + "import asyncio, os\nj = run('sleep', '300')\nawait asyncio.sleep(0.5)\n(os.getpid(), j.process.pid)" + ) + kernel, job = (int(part) for part in started["value"].strip("()").split(",")) + + # Graceful stop leaves the kernel running; it dies hard before the + # next daemon reaches it, and its job's group outlives it. + self.app.restart(prepare=lambda _: os.kill(kernel, signal.SIGKILL)) + deadline = time.monotonic() + 15 + while alive(job) and time.monotonic() < deadline: + time.sleep(0.1) + self.assertFalse(alive(job), "the dead kernel's job is still running") + if __name__ == "__main__": unittest.main() diff --git a/test/harness/albedo_job_slots_test_support.erl b/test/harness/albedo_job_slots_test_support.erl index bc82d44d..860a06a0 100644 --- a/test/harness/albedo_job_slots_test_support.erl +++ b/test/harness/albedo_job_slots_test_support.erl @@ -107,13 +107,13 @@ kernels() -> %% tests never reattach, so the link only has to answer. start(Python, Script, Host, Modules) -> Link = {link, fun(_, _) -> nil end, fun(_) -> nil end, fun() -> [] end, - fun(_) -> fresh end, fun(_, _, _) -> nil end, fun(_) -> nil end, fun() -> nil end}, + fun(_) -> fresh end, fun(_, _, _) -> nil end, fun(_) -> nil end, fun() -> nil end, fun(_) -> nil end}, Id = binary:part(albedo_native:new_id(), 0, 16), RunDir = filename:join(os:getenv("TMPDIR", "/tmp"), <<"albedo-run-", Id/binary>>), Bridge = filename:join(filename:dirname(Script), <<"albedo_bridge.py">>), albedo_python:start({boot, self(), Python, Bridge, <<"/tmp">>, Host, iolist_to_binary(json:encode(Modules)), Link, unicode:characters_to_binary(RunDir), - Id, albedo_native:new_id(), 20, 0, true, none}). + Id, albedo_native:new_id(), 20, 0, <<"{}">>, true, none}). cell(Kernel, Code) -> %% A kernel answers a cell id it already ran with that run's result. diff --git a/test/harness/kernel_link_test.gleam b/test/harness/kernel_link_test.gleam index 644b1a03..d9a4ef5a 100644 --- a/test/harness/kernel_link_test.gleam +++ b/test/harness/kernel_link_test.gleam @@ -19,6 +19,7 @@ fn recorded() -> #(store.Store, link.Record) { cwd: "/tmp", modules: "[]", out_seq: 0, + owned: "{}", ) let assert Ok(_) = link.create(ledger, record) #(ledger, record) diff --git a/test/harness/process_supervision_test.py b/test/harness/process_supervision_test.py index caf5481a..db779ab5 100644 --- a/test/harness/process_supervision_test.py +++ b/test/harness/process_supervision_test.py @@ -181,12 +181,11 @@ class SupervisionTest(unittest.TestCase): asyncio.create_subprocess_shell("sleep 30", start_new_session=True) ) try: + # Linux and Darwin both give a process an identity beyond its pid. + self.assertIsNotNone(albedo_proc.leader_token(leader.pid)) group = albedo_proc.Group(leader.pid, "not-the-leader-we-created") - if ( - albedo_proc.leader_token(leader.pid) is not None - ): # identity source present - self.assertFalse(albedo_proc.alive(group)) - self.assertTrue(present(leader.pid)) # and it was left alone + self.assertFalse(albedo_proc.alive(group)) + self.assertTrue(present(leader.pid)) # and it was left alone finally: run(albedo_proc.terminate([albedo_proc.Group(leader.pid)])) run(leader.wait()) -- knotmirror 0.0.1