JosephFerano/flan
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

A restart is taken only after every job queued before it was accepted has run, and a choice waiting for an outer break survives a nested one, so a fix loaded just before retry is the body the retry runs.

Browse Source
This commit is contained in:
Joseph Ferano 2026-09-25 22:01:08 +07:00
parent d77d541360
commit c9b556a539
6 changed files with 141 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
TODO.org
View File

@ -1350,13 +1350,9 @@ poll would, and test/agent_hooks.c uses it to choose at an outer break and then
nest a break on top before the outer one looks. The inner break turns past the
choice and resumes only on its own. Rules out a sleep-timed socket test for this.
** TODO A choice made at an outer break is lost to a nested one
=chosen_index=, =chosen_gen= and =chosen_ready= are one slot. A choice validated
against an outer break and met by a nested one survives the nested break's
turns, but the nested break can only resume on a choice of its own, which
overwrites it — so the outer break stays stopped after the listener answered ok
for it. test/agent_hooks.c's =stale= mode pins this as it is. A slot per
snapshot is the likely fix.
** DONE A choice made at an outer break survives a nested one
CLOSED: [2026-09-25]
A nested break keeps the choice pending for the break below it and puts it back when it is left, so the outer break takes it without being asked again; and a restart is taken only after every job queued before it was accepted has run. Rules out a scripted fix-then-retry running the old body.
** DONE SNAP_MAX and SNAP_NAMES are read rather than tested
CLOSED: [2026-09-25]

16
test/agent_hooks.c
View File

@ -134,10 +134,9 @@ static int snapnames(void) {
* not addressed to it, for as many turns as it is left alone, and resumes
* only on a choice made against its own list.
*
* What happens after that is pinned as it is, not as it ought to be: the
* choice slot is one slot, so the inner choice overwrote the outer one, and
* the outer break has to be asked again. TODO.org, "A choice made at an outer
* break is lost to a nested one". */
* The choice slot is one slot, so the inner choice overwrites the outer one;
* the inner break puts the outer one back when it is left, and the outer
* break takes it without being asked again. */
static int level, inner_turns, outer_turns, reasked;
static void *outer_a, *outer_b, *inner;
@ -195,10 +194,9 @@ static int stale(void) {
/* ── A restart accepted with a read queued behind it ────────────────
*
* The inspector's [dyn] read is a job for the stopped thread, named for the
* stop. One queued before a restart is accepted, and one asked for after,
* must neither run in the break being left: the first stays in the ring and
* is dropped at the next poll, the second is refused at the door. The break
* takes the restart without draining the ring first. */
* stop. Requests take effect in the order they were sent: one queued before a
* restart is accepted runs in the stop, and then the restart is taken; one
* asked for after acceptance is refused at the door. */
extern uint64_t flan_dynword(void *xfer) __asm__("flan.dynword");
int32_t flan_agent_poll(void);
static int resuming_done;
@ -239,6 +237,8 @@ static int resuming(void) {
pthread_join(resumer, NULL);
printf("queued %stake %slate %s", r_queued, r_take, r_late);
printf("returned %d\n", v);
/* The read wrote the result buffer, whose generation starts at zero. */
printf("read %s\n", strtol(ask("result"), NULL, 10) > 0 ? "ran" : "did not run");
before = refusal_count();
flan_agent_poll();
printf("dropped %ld\n", refusal_count() - before);

19
test/programs/dev-fix-retry.flan Normal file
View File

@ -0,0 +1,19 @@
;;;; Fix and retry, from a script: [f] errors, a client loads a new [f] and
;;;; takes [retry] at once, and the retry must run the new body. The load is a
;;;; job in the agent's ring and the restart a choice; the break loop runs what
;;;; was queued before the choice was accepted, then takes it.
(import agent "vendor:agent")
(defstruct Boom [why i32])
(defn f [] i32 (error (Boom {.why 3})) 1)
(defn g [] i32
(restart-case (f)
(retry [] (g))))
(defn main [] i32
(agent/start "/tmp/flan-dev-fix-retry-fallback.sock")
(let [r (g)] (println "RESULT " r))
(dotimes [i 4000] (agent/wait 5))
0)

20
test/test_agent.ml
View File

@ -944,26 +944,26 @@ let () =
(* A choice made against the outer break, then a break nested on top of it
before the outer one looks. The inner break turns past it five times
and resumes only on its own choice, into its own frame; index 1 read
without its generation would have sent it to outer-b. The last two lines
are the outer break needing to be asked again, because the choice slot
is one slot — TODO.org, "A choice made at an outer break is lost to a
nested one". *)
without its generation would have sent it to outer-b. The inner break
puts the outer choice back when it is left, so the outer break takes it
without being asked again. *)
let code, out, err = hook_mode "stale" in
let want =
"outer choice ok\nstatus stopped Inner\ninner choice ok\ninner turns 5\n\
inner resumed into inner\nouter re-asked 1\nouter resumed into outer-a\n"
inner resumed into inner\nouter re-asked 0\nouter resumed into outer-a\n"
in
if code <> 0 || out <> want then
fail "a choice addressed to an outer break, met by a nested one\n got: %S (exit %d, err %S)\n wanted: %S"
out code err want;
(* A restart accepted with an inspector read queued behind it, and a
second asked for after. Neither runs in the break being left: the
break takes the restart without draining the ring, the queued job is
dropped at the next poll, and the late one is refused at the door. *)
(* An inspector read queued, then a restart accepted, then a second read.
Requests take effect in the order they were sent: the first read runs
in the stop it was sent to, the restart is taken after it, and the
late read is refused at the door rather than run in the stop being
left. *)
let code, out, err = hook_mode "resuming" in
let want =
"queued ok\ntake ok\nlate err the program is resuming: a restart was taken\n\
returned 0\ndropped 1\n"
returned 0\nread ran\ndropped 0\n"
in
if code <> 0 || out <> want then
fail "a restart accepted with a read queued behind it\n got: %S (exit %d, err %S)\n wanted: %S"

55
test/test_dev.ml
View File

@ -6755,6 +6755,61 @@ let () =
List.iter (fun f -> try Sys.remove f with Sys_error _ -> ()) [ psock; pout ])
[ "--llvm"; "--x86" ];
(* ── Fix and retry, sent back to back ───────────────────────────── *)
(* A client that loads a new body and takes [retry] in the next breath —
a script, or an editor command that does both — must get the new body.
The load is queued before the restart is accepted, so it runs first.
Both backends. *)
List.iter
(fun backend ->
let fsock = tmp ("fixretry" ^ backend ^ ".sock")
and fout = tmp ("fixretry" ^ backend ^ ".out") in
(try Sys.remove fsock with Sys_error _ -> ());
let ffd =
Unix.openfile fout [ Unix.O_WRONLY; Unix.O_CREAT; Unix.O_TRUNC ] 0o600
in
let fpid =
Unix.create_process flan
[| flan; "dev"; "programs/dev-fix-retry.flan"; "-s"; fsock; backend |]
Unix.stdin ffd Unix.stderr
in
Unix.close ffd;
if not (listening ~pid:fpid fsock) then begin
fail "the %s fix-retry daemon %s" backend !listen_why;
(try Unix.kill fpid Sys.sigkill with Unix.Unix_error _ -> ())
end
else begin
let c = connect fsock in
let stopped r =
match Wire.field r "stopped" with
| Some { Form.v = Form.Sym "t"; _ } -> true
| _ -> false
in
if not (await (fun () -> stopped (request c "(:op \"describe\")"))) then
fail "the %s fix-retry program never stopped" backend
else begin
Buffer.clear output;
let r = request c "(:op \"eval\" :code \"(defn f [] i32 2)\")" in
if status r <> "ok" then fail "%s: loading the fix was refused" backend;
let r = request c "(:op \"restart\" :name \"retry\")" in
if status r <> "ok" then fail "%s: retry was refused" backend;
if not
(await (fun () ->
ignore (request c "(:op \"describe\")");
contains_sub (Buffer.contents output) "RESULT "))
then fail "%s: the retried program printed nothing" backend
else if not (contains_sub (Buffer.contents output) "RESULT 2") then
fail "%s: the retry ran the old body: %S" backend
(Buffer.contents output)
end;
ignore (request c "(:op \"close\")");
(try Unix.close c with Unix.Unix_error _ -> ());
(try ignore (Unix.waitpid [] fpid) with Unix.Unix_error _ -> ())
end;
List.iter (fun f -> try Sys.remove f with Sys_error _ -> ()) [ fsock; fout ])
[ "--llvm"; "--x86" ];
(* ── A dyn value that traps or faults while it is printed ─────────── *)
(* The reader hands a dyn word to the program's own thread to render,

53
vendor/agent/flan_agent.c vendored
View File

@ -589,6 +589,11 @@ static _Atomic int chosen_index;
* take it; the inner loop simply does not claim what is not addressed to it. */
static _Atomic int chosen_gen;
static _Atomic int chosen_ready;
/* The ring's head when the choice was accepted. Requests take effect in the
* order they were sent: every job queued before the restart runs in the stop
* it was sent to — a redefinition loaded and then [retry] is the fix-and-retry
* loop — and only then is the restart taken. */
static _Atomic unsigned chosen_head;
static _Atomic int aborting;
/* -- The snapshot ---------------------------------------------------- */
@ -1082,6 +1087,19 @@ static _Noreturn void die_now(void) {
* it is the same interleaving on every run. */
void (*flan_agent_break_poll_hook)(void);
static int32_t poll_upto(int bounded, unsigned limit);
/* Put back a choice that was waiting for an outer break when a nested one
* started, unless it was this break's own. */
static void restore_choice(int ready, int index, int gen, unsigned head_at,
int32_t mine) {
if (!ready || gen == mine) return;
atomic_store(&chosen_index, index);
atomic_store(&chosen_gen, gen);
atomic_store(&chosen_head, head_at);
atomic_store(&chosen_ready, 1);
}
static void break_loop_at(const uint8_t *name, int64_t namelen, void *condition,
void *xfer, int resumable) {
struct timespec step = { 0, 2000000 }; /* 2ms */
@ -1121,6 +1139,14 @@ static void break_loop_at(const uint8_t *name, int64_t namelen, void *condition,
fflush(stderr);
die_now();
}
/* A choice already accepted for a break below this one — this break was
* pushed by a job drained ahead of it. The slot is one slot, so it is kept
* here and put back when this break is left, and the outer break takes it
* then. */
int outer_ready = atomic_load(&chosen_ready);
int outer_index = atomic_load(&chosen_index);
int outer_gen = atomic_load(&chosen_gen);
unsigned outer_head = atomic_load(&chosen_head);
{
snapshot *s = snap_top();
my_gen = s->gen;
@ -1192,13 +1218,15 @@ static void break_loop_at(const uint8_t *name, int64_t namelen, void *condition,
fflush(stderr);
die_now();
}
/* The ring is not drained while a choice for this break is waiting. A job
* queued after a restart was accepted belongs to the stop being left; run
* first, it would run in that stop and could push a break of its own that
* takes the restart meant for this one. Left in the ring, it meets the gate
* in [flan_agent_poll] at the next boundary and is dropped. */
/* With a choice for this break waiting, the ring is drained only up to
* where it stood when the choice was accepted: what the client sent before
* the restart runs in this stop, in order, and then the restart is taken.
* A job that needs this stop and arrives after acceptance is refused at the
* listener; anything else queued after it runs at the next boundary. */
for (;;) {
if (!(atomic_load(&chosen_ready) && atomic_load(&chosen_gen) == my_gen)) {
if (atomic_load(&chosen_ready) && atomic_load(&chosen_gen) == my_gen)
poll_upto(1, atomic_load(&chosen_head));
else {
flan_agent_poll();
if (flan_agent_break_poll_hook != NULL) flan_agent_break_poll_hook();
}
@ -1235,6 +1263,7 @@ static void break_loop_at(const uint8_t *name, int64_t namelen, void *condition,
atomic_store(&aborting, 0);
snap_pop();
atomic_fetch_sub(&depth, 1);
restore_choice(outer_ready, outer_index, outer_gen, outer_head, my_gen);
siglongjmp(*eval_escape, 1);
}
if (ok) {
@ -1264,6 +1293,7 @@ static void break_loop_at(const uint8_t *name, int64_t namelen, void *condition,
* arriving in it is validated against a list nobody is looking at. */
snap_pop();
atomic_fetch_sub(&depth, 1);
restore_choice(outer_ready, outer_index, outer_gen, outer_head, my_gen);
return;
}
/* The listener checks all of this before answering ok, so reaching here
@ -1305,7 +1335,13 @@ static void trap_stop(const uint8_t *name, int64_t namelen) {
* consumed, and running a C-x C-e thunk a second time is the one thing the
* whole dev loop is careful never to do. Still single-consumer: only the game
* thread writes tail, nesting included. */
int32_t flan_agent_poll(void) {
static int32_t poll_upto(int bounded, unsigned limit);
int32_t flan_agent_poll(void) { return poll_upto(0, 0); }
/* [bounded]: stop at ring position [limit] — the head when a restart was
* accepted — rather than at the head now. Signed difference, because a
* nested break's own poll may already have taken the ring past it. */
static int32_t poll_upto(int bounded, unsigned limit) {
int32_t n = 0;
/* A poll the game loop makes is a frame boundary, and a dev build wipes the
* temp allocator there, as the program's own (free-temp) would. Not a poll
@ -1317,6 +1353,7 @@ int32_t flan_agent_poll(void) {
unsigned t = atomic_load_explicit(&tail, memory_order_relaxed);
unsigned h = atomic_load_explicit(&head, memory_order_acquire);
if (t == h) return n;
if (bounded && (int)(limit - t) <= 0) return n;
job j = queue[t % QUEUE];
atomic_store_explicit(&tail, t + 1, memory_order_relaxed);
/* The gate the [job] comment argues for, asked at the only moment whose
@ -1837,6 +1874,7 @@ static void handle_line(char *line, sink *o) {
if (unarmed(s, (int32_t)idx)) { reply_unarmed(o, s, (int32_t)idx); return; }
atomic_store(&chosen_index, (int)idx);
atomic_store(&chosen_gen, s->gen);
atomic_store(&chosen_head, atomic_load(&head));
/* Published last, so the game thread never reads an index that is about
* to change, or one whose generation has not arrived yet. */
atomic_store(&chosen_ready, 1);
@ -1886,6 +1924,7 @@ static void handle_line(char *line, sink *o) {
if (unarmed(s, at)) { reply_unarmed(o, s, at); return; }
atomic_store(&chosen_index, at);
atomic_store(&chosen_gen, s->gen);
atomic_store(&chosen_head, atomic_load(&head));
atomic_store(&chosen_ready, 1);
/* The same two answers as [restart-at], because this verb is defined as
* that one on the first index offering the name. Two verbs that resolve to