From 4e02ab31b0d9276e1e0c06a0028ad5fa18c8baf3 Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Sat, 29 Aug 2026 16:22:18 +0200 Subject: [PATCH] Add the protocols and state machines pages Every message, frame and status code now has one page saying who sends it, what answers it and which files must change together, and every long-lived process and thread has its states, triggers and timers written down next to the invariants a change must keep. --- docs/architecture.md | 6 +- docs/protocols.md | 247 +++++++++++++++++++++++++++++++++++++++++ docs/state-machines.md | 191 +++++++++++++++++++++++++++++++ rebar.config | 4 + 4 files changed, 445 insertions(+), 3 deletions(-) create mode 100644 docs/protocols.md create mode 100644 docs/state-machines.md diff --git a/docs/architecture.md b/docs/architecture.md index 841a2a7..37588ae 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -127,9 +127,9 @@ In isolated mode there is one path: a status-3 frame on the socket, answered by a process the `py_isolated` state machine spawns; nested calls into the same context are dispatched immediately because they come from that process. -The frame format shared by the pipe and the socket, and the ETF conventions, -will get their own page (protocols); until then `c_src/py_convert.c` (type -mapping) and `priv/_erlang_impl/_etf.py` are the reference. +Every message and frame, with the rules for changing them, is in +[protocols](protocols.md); the states each process and thread moves through +are in [state machines](state-machines.md). ## asyncio diff --git a/docs/protocols.md b/docs/protocols.md new file mode 100644 index 0000000..3308a40 --- /dev/null +++ b/docs/protocols.md @@ -0,0 +1,247 @@ +# Protocols + +Every message and frame that crosses a boundary in erlang_python: between +an Erlang process and the context process, between the context thread in C +and Erlang, and between the VM and an isolated child. Read this when you +change a message shape, add a request kind, or debug a hang with +`sys:trace/2` or `strace`. The code is the reference; this page tells you +where each piece is and what must stay consistent. + +## Boundaries + +``` + caller process --(1) Erlang messages--> py_context / py_isolated process + | + embedded modes | isolated mode + (2) NIF request queue | (4) Unix socket frames + (3) callback pipe / thread pipe | + v + context thread in C child OS process +``` + +1. Erlang messages: `src/py_context.erl` (API side and embedded loop), + `src/py_isolated.erl` (isolated loop). +2. Request queue: `ctx_request_t` in `c_src/py_nif.h`, enqueued by + `nif_context_call_async`, answered with `{py_result, Ref, Result}`. +3. Callbacks: `erlang_call_impl` in `c_src/py_callback.c`, + `c_src/py_thread_worker.c`, `src/py_thread_handler.erl`. +4. Socket: `src/py_isolated.erl`, `priv/_erlang_impl/_isolated.py`, + `priv/_erlang_impl/_etf.py`. + +## 1. Caller to context process + +The public API (`py:call/3`, `py_context:call/4`, ...) is a plain message +plus a monitor. Both context loops accept the same messages, so a caller +never knows the mode. + +```erlang +%% py_context:submit/5 and friends +MRef = erlang:monitor(process, Ctx), +Ctx ! {call, self(), MRef, Module, Func, Args, Kwargs}, +%% reply +{MRef, Result} %% {ok, Term} | {error, Reason} +``` + +| Message | Reply | Notes | +|---|---|---| +| `{call, From, MRef, Module, Func, Args, Kwargs}` | `{ok, R}` / `{error, E}` | 8-tuple variant adds `EnvRef` for process-local envs | +| `{eval, From, MRef, Code, Locals}` | same | 6-tuple variant adds `EnvRef` | +| `{exec, From, MRef, Code}` | `ok` / `{error, E}` | 5-tuple variant adds `EnvRef` | +| `{submit, From, MRef, TaskRef, Module, Func, Args, Kwargs}` | `ok`, then `{async_result, TaskRef, Result}` to `From` | asyncio coroutine on the context loop | +| `{start_loop, From, MRef, Owner}` | `ok` / `{error, already_running}` | runs `run_forever` on the context thread; `Owner` is monitored and gets `{py_loop_exit, Ctx, Result}` when the loop ends | +| `{stop_loop, From, MRef, GraceMs}` | `ok` / `{error, no_loop}` | cooperative stop, interrupt after `GraceMs`, then kill | +| `{loop_ref, From, MRef}` | `{ok, LoopRef}` / `{error, no_loop}` | isolated: `{error, not_supported_in_isolated}` | +| `{interrupt, From, MRef}` | `ok` / `not_running` | | +| `{kill, From, MRef}` | `ok` | isolated only: SIGKILL, answered once the new child is up | +| `{pass_fd, From, MRef, Fd}` | `ok` / `{error, E}` | isolated only: SCM_RIGHTS | +| `{child_info, From, MRef}` | `{ok, Map}` | isolated only | +| `{stop, From, MRef}` | `ok` | | + +Timeouts are the caller's business: `await_reply/3` waits `Timeout`, then +sends `{interrupt_request, MRef}` so the request executing now is +interrupted, waits a short grace for the late reply, and returns +`{error, timeout}`. Loop control messages use `await_ctrl_reply/3`, which +must not interrupt (it would stop the loop it manages) and instead sends +`{cancel_ctrl, MRef}` so an isolated context drops the pending entry. + +Rules: + +- One request at a time per context. The embedded loop blocks in the + request; `py_isolated` postpones with `gen_statem` `postpone` while in + `{busy, Id}`. +- A request from a process that is running a callback for this context is + nested and served at once (both loops track those pids). +- Replies always go to `From`, tagged with `MRef`; a late reply after a + timeout is flushed by `demonitor(MRef, [flush])`. + +## 2. Request queue (embedded modes) + +`nif_context_call_async(Ctx, Kind, Data, RequestId)` allocates a +`ctx_request_t`, copies the terms into `request_env`, appends it under +`queue_mutex` and returns. The context thread dequeues, runs the request, +and sends: + +```erlang +{py_result, RequestId, Result} +``` + +to `caller_pid` through `msg_env`. Cancelled requests (context destroyed +while queued) are answered with `{py_result, Id, {error, cancelled}}`. +Request kinds are `ctx_request_type_t` in `c_src/py_nif.h`; the execute +functions read them from the request mirror on `py_context_t` +(`request_type`, `request_term`, ...). + +A result that needs a callback served before completion comes back as +`{suspended, CallbackId, StateRef, {Name, Args}}` in place of the final +result (see 3.1); the process resumes with `resume_callback/2` and waits +for the next `py_result`. + +## 3. Python calling Erlang + +`erlang.call(Name, Args...)` picks a path in `erlang_call_impl` +(`c_src/py_callback.c`); the comment above that function is authoritative. +Precedence: suspension when the calling thread is a context thread with +suspension enabled, then the context's blocking pipe when a handler is set, +then the thread worker for every other Python thread. + +### 3.1 Suspension + +The Python call raises `SuspensionRequired` and sets thread-local pending +state; the execute function sees the flag before the exception type and +returns the `{suspended, ...}` result above. `py_context` runs the fun +(`execute/2` in `py_callback`), serving nested requests meanwhile +(`wait_for_callback/2`), and calls `resume_callback(StateRef, Response)` +with the response body of 3.4. + +### 3.2 Context callback pipe + +Set up with `context_set_callback_handler(Ref, Pid)`. The context thread +sends: + +```erlang +{erlang_callback, CallbackId, FuncName, Args} +``` + +to the handler process and blocks (GIL released) reading the pipe with a +30 s timeout. The handler runs the fun and answers with +`context_write_callback_response(Ref, Body)`, which writes +`<>` on the pipe. There is no id on this pipe: +one context thread, one outstanding call. + +### 3.3 Thread worker pipe + +Any other Python thread (a `threading.Thread`, an executor worker) sends: + +```erlang +{thread_callback, WorkerId, CallbackId, FuncName, Args} +``` + +to the `py_thread_handler` coordinator, which gives each `WorkerId` a +handler process (`{thread_worker_spawn, WorkerId, WriteFd}`) and replies +with `thread_worker_write_response(Fd, CallbackId, Body)`. The frame on the +response pipe is: + +``` +<> +``` + +The reader discards frames whose id is not the one it waits for; a short +read or a timeout poisons the worker so the pipe is never read out of +phase. `erlang.async_call` uses the same frame on a per-interpreter +non-blocking pipe (`async_callback_pipe`, `{async_callback, CallbackId, +FuncName, Args, WriteFd}`), parsed incrementally by the reader tick. + +### 3.4 Response body + +Every path answers with the same body: + +``` +<<2, ETF/binary>> %% ok, term_to_binary(Result) +<<1, Message/binary>> %% error, UTF-8 text, raised as RuntimeError +``` + +Built in `handle_blocking_callback/3` (`py_context`), `py_thread_handler` +and the isolated loop; parsed by `parse_callback_response` in C and +`_isolated.py` in the child. Keep the three writers and two parsers in +step. + +## 4. Isolated socket + +The child (`priv/py_isolated_child.py`) connects to a Unix socket in the +private directory (`py_isolated:sock_dir/0`) and both sides exchange frames: + +``` +<> +Body = <> +``` + +| Status | Direction | Meaning | Payload | +|---|---|---|---| +| 0 | Erlang to child | request | `{call, M, F, Args, Kwargs}`, `{eval, Code, Locals}`, `{exec, Code}`, `{submit, TaskRef, ...}`, `start_loop`, `{pass_fd, ...}`, `ping`, `shutdown`, `{init, Opts}` | +| 1 | either | error reply | `{Class, Message, Traceback}` or an atom | +| 2 | either | ok reply | the result term | +| 3 | child to Erlang | request from Python | `{call, Name, Args}`, `{send, Pid, Msg}`, `{whereis, Name}` | +| 4 | child to Erlang | event, `Id` = 0 | `{ready, Info}`, `{startup_error, Problems}`, `{memory_limit, Rss}`, `{log, Level, Msg}`, `{async_result, TaskRef, R}`, `{loop_exit, R}` | +| 5 | Erlang to child | control, `Id` = 0 | `{interrupt, Target}`, `{cancel, Id}`, `stop_loop`, `{shm_close, Id}` | + +Ids: Erlang numbers its requests from 1 (`next_id`); the child numbers its +status-3 requests independently; replies carry the id of the request they +answer. Control and events use id 0 and never get a reply. + +ETF: `_etf.py` encodes what `py_convert.c` would produce for the same +Python value, with two differences worth knowing: `Pid`, `Ref` and `Port` +are opaque and keep their raw bytes, and the child can create atoms +(`binary_to_term/1` is not called with `safe`). Shared handles +(`{'$py_shm', Id, Path, Size}`, `{'$py_buffer', Id, Path, Ring}`) are plain +tuples on the wire and become `SharedMemory` / `SharedBuffer` on arrival +(`_shm.from_term`); NIF resources cannot cross and the API answers +`{error, not_supported_in_isolated}`. + +Handshake (`py_isolated:handshake/1`): the child sends `{ready, Info}`, +Erlang sends `{init, Opts}` (status 0, id 1) and the preload `exec`, +serving status-3 requests that arrive meanwhile with a bounded callback +runner. Anything else during the handshake fails the start. + +Interrupt: `{interrupt, Target}` is read by the child's reader thread, which +sends SIGUSR1 to the main thread when `Target` is the request executing +now (top of `_exec_stack`); the signal handler raises `_Interrupted` +(a `KeyboardInterrupt` subclass) only while a request runs. `{cancel, Id}` +marks a queued request so it is answered `{error, cancelled}` instead of +run. Erlang arms a SIGKILL backstop bound to the request id when it sends +an interrupt; a reply for that id cancels it. + +Flow control: the socket buffers are 1 MB each way; a frame larger than +that is written in pieces by both sides. Bulk data goes through shared +memory, not the socket (see [isolated](isolated.md)). + +## 5. Shared buffer ring + +`py_buffer:new(#{shared => true})` allocates a `py_shm` region used as a +ring. The first page is a header written by Erlang after each `write/2` +(write position, closed flag, ring size) and read by Python; the +`py_shm` server keeps the consumed position from the callbacks below, not +from the header. Notifications are callbacks, so they work in every mode: + +| Callback | Called by | Returns | +|---|---|---| +| `_py_buffer_wait(Id, ReadPos)` | reader, before blocking | `{WPos, Closed}` once `WPos > ReadPos` or the buffer is closed | +| `_py_buffer_consumed(Id, N)` | reader, after a read | `ok`; a writer blocked on space wakes up | +| `_py_buffer_state(Id)` | reader, on (re)map | `{WPos, Closed}` without waiting | + +All three answer `{error, closed}` for an unknown id. The header is only +read after a callback returned, so ordering follows the round trip and the +Python side needs no fence. + +## Changing a protocol + +- Add a request kind: the message in `py_context` (API and embedded loop), + the `ctx_request_type_t` and execute function in C, the `request/6` + clause in `py_isolated`, and `_on_request` in `_isolated.py`. +- Add a control: `send_frame(Child, 0, ?STATUS_CONTROL, Term)` in + `py_isolated` and `_on_control` in `_isolated.py`; controls must be safe + to handle on the reader thread. +- Add a child event: `self.event(...)` in `_isolated.py` and an + `{ok, {0, ?STATUS_EVENT, ...}}` clause in `drain_socket` or `handshake`. +- Change the response body: three writers and two parsers listed in 3.4. +- Never change the frame header: the child, the callback pipe reader and + the async pipe parser share it. diff --git a/docs/state-machines.md b/docs/state-machines.md new file mode 100644 index 0000000..871bab1 --- /dev/null +++ b/docs/state-machines.md @@ -0,0 +1,191 @@ +# State machines + +The states each long-lived thing in erlang_python moves through, what moves +it, and what is allowed in each state. Read this before changing a loop, +a shutdown path or a restart policy; the invariants at the end are the ones +a change must keep. Messages and frames are in [protocols](protocols.md). + +## The Python runtime (C) + +`g_runtime_state` in `c_src/py_nif.h`, moved with compare-and-swap so only +one thread wins each transition. + +``` +UNINIT --init--> INITING --ok--> RUNNING --finalize--> SHUTTING_DOWN --> STOPPED + | ^ + +--------------------- failure ------------------------+ +``` + +- `runtime_is_running()` gates every NIF that touches Python; a NIF that + finds `SHUTTING_DOWN` or `STOPPED` returns `not_running` or an error. +- `STOPPED` may re-enter `INITING`: the runtime can be finalized and + initialized again in one VM (the suites do this). + +## An embedded context (`py_context` process + context thread) + +Two cooperating machines: the Erlang process and the pthread in C. + +Context thread (`worker_context_thread_main`, `owngil_context_thread_main` +in `c_src/py_nif.c`): + +``` +starting --namespaces created--> waiting --dequeue--> executing --reply--> waiting + | | + +-- init_error -----> exited +-- shutdown_requested ----> exited +``` + +- `waiting` blocks on `queue_not_empty` under `queue_mutex`. +- `executing` is bracketed by `py_context_exec_enter` / `exec_leave` + (interrupt bookkeeping) around the GIL; the request mirror on + `py_context_t` is valid only here. +- `exited` sets `worker_running = false`; `nif_context_destroy` joins with a + timeout and, if the join fails, marks the context `leaked` and pins the + resource instead of freeing it. + +Erlang process (`loop/1` in `py_context`): + +``` +idle --{call|eval|exec|submit}--> in_request --{py_result}--> idle + | | + | +--{suspended, ...}--> in_callback --resume--> in_request + | + +--{start_loop}--> loop_running --{py_result, LoopReq}--> idle + | + +--{stop_loop, GraceMs}--> stopping --grace--> interrupt --deadline--> idle +``` + +- `in_request` is a blocking receive for the reply; nested callbacks + arrive as `{erlang_callback, ...}` (pipe) or `{suspended, ...}` and are + served inline, so the process never deadlocks with its own thread. +- In `loop_running` the loop request `LoopReq` occupies the thread: `call`, + `eval`, `exec` and `call_method` answer `{error, loop_running}`; `stop` + first stops the loop. The owner is monitored and its `DOWN` stops the loop. +- `stopping` arms `loop_stop_deadline` (cooperative) then + `loop_interrupt_deadline`; the loop exit is the `py_result` for `LoopReq`, + and the owner gets `{py_loop_exit, Ctx, Result}`. + +## A request (`ctx_request_t`) + +``` +created (refcount 1) --enqueue--> queued (2) --dequeue--> running --done--> completed + | | + +-- cancelled (destroy, timeout) -------+ + v + freed when refcount hits 0 +``` + +The queue and the caller each hold one reference; whoever releases last +frees. `cancelled` is checked by the thread before running, so a request +cancelled while queued is answered `{error, cancelled}` without touching +Python. + +## An isolated context (`py_isolated`, `gen_statem`) + +States are the `state()` type in `src/py_isolated.erl`; `sys:get_state/1` +shows the current one and `sys:trace/2` prints transitions. + +``` + start_child + handshake + | + v + +--------------> idle <----------------------------------+ + | | | + | main request | start_loop | + | v | + | {busy, Id} ---reply Id---> idle | + | | | + | | looping --stop_loop-----> stopping_loop + | | | (grace: interrupt, then kill) + | | +--loop_exit event--------+ + | | + +-- child exit / kill / socket error --> {restarting, Reason} --new child--> idle + | + +-- budget exhausted --> stop +``` + +Per state: + +| State | Main requests (`call`, `eval`, `exec`, `start_loop`) | Other requests | Timers armed | +|---|---|---|---| +| `idle` | dispatched, go to `{busy, Id}` (`start_loop` goes to `looping`) | dispatched | none | +| `{busy, Id}` | postponed, unless from a process running a callback for this context (nested, dispatched) | dispatched | `{timeout, kill}` bound to `Id` once an interrupt was sent | +| `looping` | `{error, loop_running}` | dispatched (`submit`, `pass_fd`, ...) | none | +| `stopping_loop` | postponed | dispatched | `state_timeout` for the interrupt, then `{timeout, kill}` bound to `loop` | +| `{restarting, R}` | postponed | postponed | `state_timeout` waiting for the port's `exit_status` | + +Transitions and their triggers: + +- `{busy, Id}` to `idle`: a status-1/2 frame with `Id`. Frames for other + ids (nested requests) do not change state. +- Anything to `{restarting, Reason}`: `{Port, {exit_status, S}}`, a socket + `abort`, a `{memory_limit, Rss}` event, `kill/1`, or a request the state + machine cannot deliver. In-flight requests, submitted tasks and a running + loop fail with `Reason` (`fail_pending/2`); postponed requests are kept + and served by the next child. +- `{restarting, _}` to `idle`: the port reported the exit and a new child + passed the handshake. `restart_allowed/1` counts restarts in + `restart_period`; over `max_restarts` (or with `restart => false`) the + process stops with `{child_exited, Reason}`. +- `looping` to `stopping_loop`: `stop_loop/2` or the owner's `DOWN`. + `stopping_loop` to `idle`: the `{loop_exit, R}` event. The interrupt + `state_timeout` and the kill backstop escalate if the loop does not exit. + +Interrupt timing: `interrupt/1` in `{busy, Id}` sends `{interrupt, Id}` and +arms `{{timeout, kill}, KillAfter, Id}`. The reply for `Id` cancels the +timer; if it fires, the child gets SIGKILL and the machine goes to +`{restarting, killed}`. An interrupt in any other state answers +`not_running`. + +## The child (`_isolated.py`) + +The main thread runs one request at a time but can nest: + +``` +idle --request--> executing [stack: Id1] + | + +-- erlang.call --> waiting for reply, serving nested requests [Id1, Id2] ... + | + +-- SIGUSR1 while running --> _Interrupted raised in the request on top +``` + +- `_exec_stack` holds the ids being executed, innermost last. An + `{interrupt, Target}` control is honoured only if `Target` is the top of + the stack; otherwise it is stale and dropped. The signal handler raises + only while `running` is true, so an interrupt between requests cannot + leak into the next one. +- The reader thread never runs Python code: it parses frames, resolves + waiters, and pushes requests and interrupts to the main thread's inbox. +- `broken` (EOF or a hard error on the socket) is terminal: the reader + calls `os._exit`, since nothing useful can happen in the process any more + and a main thread stuck in a C call must not keep it alive. Erlang sees + the port's `exit_status` and runs the restart policy above. +- With a loop: `start_loop` runs `run_forever` on the main thread; requests + that need the main thread are refused with `loop_running`, `submit` goes + through `call_soon_threadsafe`, and `stop_loop` calls `loop.stop()` from + the reader thread. + +## A shared region (`py_shm`) + +``` +new --> open --close/1 or owner DOWN--> closed (file unlinked, handle closed) +``` + +Mappings in Python outlive `closed` until the wrapper is closed or +collected; a later access raises `ValueError`, never a fault. A shared +buffer adds `closed = true` in its header at `py_buffer:close/1`, and +readers waiting in `_py_buffer_wait` are answered with the closed flag. + +## Invariants + +- A context executes one top-level request at a time, in every mode. + Embedded: one thread, one dequeue. Isolated: `{busy, Id}` plus `postpone`. +- Nested requests only come from a process serving a callback of the same + context. Anything else waits. +- A restart never loses a queued request, only the ones in flight, and + callers of those get an error naming the cause. +- Interrupts target the request executing now; a stale interrupt is + dropped on both sides (kill timer bound to the id in Erlang, stack check + in the child, `interrupt_pending` cleared in `exec_leave` for embedded + contexts). +- Shutdown never frees memory a thread may still use: embedded contexts + leak on a failed join, the child is reaped through the port. diff --git a/rebar.config b/rebar.config index d3ee4e1..6e23e77 100644 --- a/rebar.config +++ b/rebar.config @@ -79,6 +79,8 @@ <<"docs/architecture.md">>, <<"docs/code-map.md">>, <<"docs/glossary.md">>, + <<"docs/protocols.md">>, + <<"docs/state-machines.md">>, <<"docs/preload.md">>, <<"docs/owngil_internals.md">>, <<"docs/event_loop_architecture.md">> @@ -117,6 +119,8 @@ <<"docs/architecture.md">>, <<"docs/code-map.md">>, <<"docs/glossary.md">>, + <<"docs/protocols.md">>, + <<"docs/state-machines.md">>, <<"docs/preload.md">>, <<"docs/owngil_internals.md">>, <<"docs/event_loop_architecture.md">>