|
| 1 | +# Protocols |
| 2 | + |
| 3 | +Every message and frame that crosses a boundary in erlang_python: between |
| 4 | +an Erlang process and the context process, between the context thread in C |
| 5 | +and Erlang, and between the VM and an isolated child. Read this when you |
| 6 | +change a message shape, add a request kind, or debug a hang with |
| 7 | +`sys:trace/2` or `strace`. The code is the reference; this page tells you |
| 8 | +where each piece is and what must stay consistent. |
| 9 | + |
| 10 | +## Boundaries |
| 11 | + |
| 12 | +``` |
| 13 | + caller process --(1) Erlang messages--> py_context / py_isolated process |
| 14 | + | |
| 15 | + embedded modes | isolated mode |
| 16 | + (2) NIF request queue | (4) Unix socket frames |
| 17 | + (3) callback pipe / thread pipe | |
| 18 | + v |
| 19 | + context thread in C child OS process |
| 20 | +``` |
| 21 | + |
| 22 | +1. Erlang messages: `src/py_context.erl` (API side and embedded loop), |
| 23 | + `src/py_isolated.erl` (isolated loop). |
| 24 | +2. Request queue: `ctx_request_t` in `c_src/py_nif.h`, enqueued by |
| 25 | + `nif_context_call_async`, answered with `{py_result, Ref, Result}`. |
| 26 | +3. Callbacks: `erlang_call_impl` in `c_src/py_callback.c`, |
| 27 | + `c_src/py_thread_worker.c`, `src/py_thread_handler.erl`. |
| 28 | +4. Socket: `src/py_isolated.erl`, `priv/_erlang_impl/_isolated.py`, |
| 29 | + `priv/_erlang_impl/_etf.py`. |
| 30 | + |
| 31 | +## 1. Caller to context process |
| 32 | + |
| 33 | +The public API (`py:call/3`, `py_context:call/4`, ...) is a plain message |
| 34 | +plus a monitor. Both context loops accept the same messages, so a caller |
| 35 | +never knows the mode. |
| 36 | + |
| 37 | +```erlang |
| 38 | +%% py_context:submit/5 and friends |
| 39 | +MRef = erlang:monitor(process, Ctx), |
| 40 | +Ctx ! {call, self(), MRef, Module, Func, Args, Kwargs}, |
| 41 | +%% reply |
| 42 | +{MRef, Result} %% {ok, Term} | {error, Reason} |
| 43 | +``` |
| 44 | + |
| 45 | +| Message | Reply | Notes | |
| 46 | +|---|---|---| |
| 47 | +| `{call, From, MRef, Module, Func, Args, Kwargs}` | `{ok, R}` / `{error, E}` | 8-tuple variant adds `EnvRef` for process-local envs | |
| 48 | +| `{eval, From, MRef, Code, Locals}` | same | 6-tuple variant adds `EnvRef` | |
| 49 | +| `{exec, From, MRef, Code}` | `ok` / `{error, E}` | 5-tuple variant adds `EnvRef` | |
| 50 | +| `{submit, From, MRef, TaskRef, Module, Func, Args, Kwargs}` | `ok`, then `{async_result, TaskRef, Result}` to `From` | asyncio coroutine on the context loop | |
| 51 | +| `{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 | |
| 52 | +| `{stop_loop, From, MRef, GraceMs}` | `ok` / `{error, no_loop}` | cooperative stop, interrupt after `GraceMs`, then kill | |
| 53 | +| `{loop_ref, From, MRef}` | `{ok, LoopRef}` / `{error, no_loop}` | isolated: `{error, not_supported_in_isolated}` | |
| 54 | +| `{interrupt, From, MRef}` | `ok` / `not_running` | | |
| 55 | +| `{kill, From, MRef}` | `ok` | isolated only: SIGKILL, answered once the new child is up | |
| 56 | +| `{pass_fd, From, MRef, Fd}` | `ok` / `{error, E}` | isolated only: SCM_RIGHTS | |
| 57 | +| `{child_info, From, MRef}` | `{ok, Map}` | isolated only | |
| 58 | +| `{stop, From, MRef}` | `ok` | | |
| 59 | + |
| 60 | +Timeouts are the caller's business: `await_reply/3` waits `Timeout`, then |
| 61 | +sends `{interrupt_request, MRef}` so the request executing now is |
| 62 | +interrupted, waits a short grace for the late reply, and returns |
| 63 | +`{error, timeout}`. Loop control messages use `await_ctrl_reply/3`, which |
| 64 | +must not interrupt (it would stop the loop it manages) and instead sends |
| 65 | +`{cancel_ctrl, MRef}` so an isolated context drops the pending entry. |
| 66 | + |
| 67 | +Rules: |
| 68 | + |
| 69 | +- One request at a time per context. The embedded loop blocks in the |
| 70 | + request; `py_isolated` postpones with `gen_statem` `postpone` while in |
| 71 | + `{busy, Id}`. |
| 72 | +- A request from a process that is running a callback for this context is |
| 73 | + nested and served at once (both loops track those pids). |
| 74 | +- Replies always go to `From`, tagged with `MRef`; a late reply after a |
| 75 | + timeout is flushed by `demonitor(MRef, [flush])`. |
| 76 | + |
| 77 | +## 2. Request queue (embedded modes) |
| 78 | + |
| 79 | +`nif_context_call_async(Ctx, Kind, Data, RequestId)` allocates a |
| 80 | +`ctx_request_t`, copies the terms into `request_env`, appends it under |
| 81 | +`queue_mutex` and returns. The context thread dequeues, runs the request, |
| 82 | +and sends: |
| 83 | + |
| 84 | +```erlang |
| 85 | +{py_result, RequestId, Result} |
| 86 | +``` |
| 87 | + |
| 88 | +to `caller_pid` through `msg_env`. Cancelled requests (context destroyed |
| 89 | +while queued) are answered with `{py_result, Id, {error, cancelled}}`. |
| 90 | +Request kinds are `ctx_request_type_t` in `c_src/py_nif.h`; the execute |
| 91 | +functions read them from the request mirror on `py_context_t` |
| 92 | +(`request_type`, `request_term`, ...). |
| 93 | + |
| 94 | +A result that needs a callback served before completion comes back as |
| 95 | +`{suspended, CallbackId, StateRef, {Name, Args}}` in place of the final |
| 96 | +result (see 3.1); the process resumes with `resume_callback/2` and waits |
| 97 | +for the next `py_result`. |
| 98 | + |
| 99 | +## 3. Python calling Erlang |
| 100 | + |
| 101 | +`erlang.call(Name, Args...)` picks a path in `erlang_call_impl` |
| 102 | +(`c_src/py_callback.c`); the comment above that function is authoritative. |
| 103 | +Precedence: suspension when the calling thread is a context thread with |
| 104 | +suspension enabled, then the context's blocking pipe when a handler is set, |
| 105 | +then the thread worker for every other Python thread. |
| 106 | + |
| 107 | +### 3.1 Suspension |
| 108 | + |
| 109 | +The Python call raises `SuspensionRequired` and sets thread-local pending |
| 110 | +state; the execute function sees the flag before the exception type and |
| 111 | +returns the `{suspended, ...}` result above. `py_context` runs the fun |
| 112 | +(`execute/2` in `py_callback`), serving nested requests meanwhile |
| 113 | +(`wait_for_callback/2`), and calls `resume_callback(StateRef, Response)` |
| 114 | +with the response body of 3.4. |
| 115 | + |
| 116 | +### 3.2 Context callback pipe |
| 117 | + |
| 118 | +Set up with `context_set_callback_handler(Ref, Pid)`. The context thread |
| 119 | +sends: |
| 120 | + |
| 121 | +```erlang |
| 122 | +{erlang_callback, CallbackId, FuncName, Args} |
| 123 | +``` |
| 124 | + |
| 125 | +to the handler process and blocks (GIL released) reading the pipe with a |
| 126 | +30 s timeout. The handler runs the fun and answers with |
| 127 | +`context_write_callback_response(Ref, Body)`, which writes |
| 128 | +`<<Len:32/native, Body/binary>>` on the pipe. There is no id on this pipe: |
| 129 | +one context thread, one outstanding call. |
| 130 | + |
| 131 | +### 3.3 Thread worker pipe |
| 132 | + |
| 133 | +Any other Python thread (a `threading.Thread`, an executor worker) sends: |
| 134 | + |
| 135 | +```erlang |
| 136 | +{thread_callback, WorkerId, CallbackId, FuncName, Args} |
| 137 | +``` |
| 138 | + |
| 139 | +to the `py_thread_handler` coordinator, which gives each `WorkerId` a |
| 140 | +handler process (`{thread_worker_spawn, WorkerId, WriteFd}`) and replies |
| 141 | +with `thread_worker_write_response(Fd, CallbackId, Body)`. The frame on the |
| 142 | +response pipe is: |
| 143 | + |
| 144 | +``` |
| 145 | +<<CallbackId:64/native, Len:32/native, Body:Len/binary>> |
| 146 | +``` |
| 147 | + |
| 148 | +The reader discards frames whose id is not the one it waits for; a short |
| 149 | +read or a timeout poisons the worker so the pipe is never read out of |
| 150 | +phase. `erlang.async_call` uses the same frame on a per-interpreter |
| 151 | +non-blocking pipe (`async_callback_pipe`, `{async_callback, CallbackId, |
| 152 | +FuncName, Args, WriteFd}`), parsed incrementally by the reader tick. |
| 153 | + |
| 154 | +### 3.4 Response body |
| 155 | + |
| 156 | +Every path answers with the same body: |
| 157 | + |
| 158 | +``` |
| 159 | +<<2, ETF/binary>> %% ok, term_to_binary(Result) |
| 160 | +<<1, Message/binary>> %% error, UTF-8 text, raised as RuntimeError |
| 161 | +``` |
| 162 | + |
| 163 | +Built in `handle_blocking_callback/3` (`py_context`), `py_thread_handler` |
| 164 | +and the isolated loop; parsed by `parse_callback_response` in C and |
| 165 | +`_isolated.py` in the child. Keep the three writers and two parsers in |
| 166 | +step. |
| 167 | + |
| 168 | +## 4. Isolated socket |
| 169 | + |
| 170 | +The child (`priv/py_isolated_child.py`) connects to a Unix socket in the |
| 171 | +private directory (`py_isolated:sock_dir/0`) and both sides exchange frames: |
| 172 | + |
| 173 | +``` |
| 174 | +<<Id:64/native, Len:32/native, Body:Len/binary>> |
| 175 | +Body = <<Status:8, ETF/binary>> |
| 176 | +``` |
| 177 | + |
| 178 | +| Status | Direction | Meaning | Payload | |
| 179 | +|---|---|---|---| |
| 180 | +| 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}` | |
| 181 | +| 1 | either | error reply | `{Class, Message, Traceback}` or an atom | |
| 182 | +| 2 | either | ok reply | the result term | |
| 183 | +| 3 | child to Erlang | request from Python | `{call, Name, Args}`, `{send, Pid, Msg}`, `{whereis, Name}` | |
| 184 | +| 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}` | |
| 185 | +| 5 | Erlang to child | control, `Id` = 0 | `{interrupt, Target}`, `{cancel, Id}`, `stop_loop`, `{shm_close, Id}` | |
| 186 | + |
| 187 | +Ids: Erlang numbers its requests from 1 (`next_id`); the child numbers its |
| 188 | +status-3 requests independently; replies carry the id of the request they |
| 189 | +answer. Control and events use id 0 and never get a reply. |
| 190 | + |
| 191 | +ETF: `_etf.py` encodes what `py_convert.c` would produce for the same |
| 192 | +Python value, with two differences worth knowing: `Pid`, `Ref` and `Port` |
| 193 | +are opaque and keep their raw bytes, and the child can create atoms |
| 194 | +(`binary_to_term/1` is not called with `safe`). Shared handles |
| 195 | +(`{'$py_shm', Id, Path, Size}`, `{'$py_buffer', Id, Path, Ring}`) are plain |
| 196 | +tuples on the wire and become `SharedMemory` / `SharedBuffer` on arrival |
| 197 | +(`_shm.from_term`); NIF resources cannot cross and the API answers |
| 198 | +`{error, not_supported_in_isolated}`. |
| 199 | + |
| 200 | +Handshake (`py_isolated:handshake/1`): the child sends `{ready, Info}`, |
| 201 | +Erlang sends `{init, Opts}` (status 0, id 1) and the preload `exec`, |
| 202 | +serving status-3 requests that arrive meanwhile with a bounded callback |
| 203 | +runner. Anything else during the handshake fails the start. |
| 204 | + |
| 205 | +Interrupt: `{interrupt, Target}` is read by the child's reader thread, which |
| 206 | +sends SIGUSR1 to the main thread when `Target` is the request executing |
| 207 | +now (top of `_exec_stack`); the signal handler raises `_Interrupted` |
| 208 | +(a `KeyboardInterrupt` subclass) only while a request runs. `{cancel, Id}` |
| 209 | +marks a queued request so it is answered `{error, cancelled}` instead of |
| 210 | +run. Erlang arms a SIGKILL backstop bound to the request id when it sends |
| 211 | +an interrupt; a reply for that id cancels it. |
| 212 | + |
| 213 | +Flow control: the socket buffers are 1 MB each way; a frame larger than |
| 214 | +that is written in pieces by both sides. Bulk data goes through shared |
| 215 | +memory, not the socket (see [isolated](isolated.md)). |
| 216 | + |
| 217 | +## 5. Shared buffer ring |
| 218 | + |
| 219 | +`py_buffer:new(#{shared => true})` allocates a `py_shm` region used as a |
| 220 | +ring. The first page is a header written by Erlang after each `write/2` |
| 221 | +(write position, closed flag, ring size) and read by Python; the |
| 222 | +`py_shm` server keeps the consumed position from the callbacks below, not |
| 223 | +from the header. Notifications are callbacks, so they work in every mode: |
| 224 | + |
| 225 | +| Callback | Called by | Returns | |
| 226 | +|---|---|---| |
| 227 | +| `_py_buffer_wait(Id, ReadPos)` | reader, before blocking | `{WPos, Closed}` once `WPos > ReadPos` or the buffer is closed | |
| 228 | +| `_py_buffer_consumed(Id, N)` | reader, after a read | `ok`; a writer blocked on space wakes up | |
| 229 | +| `_py_buffer_state(Id)` | reader, on (re)map | `{WPos, Closed}` without waiting | |
| 230 | + |
| 231 | +All three answer `{error, closed}` for an unknown id. The header is only |
| 232 | +read after a callback returned, so ordering follows the round trip and the |
| 233 | +Python side needs no fence. |
| 234 | + |
| 235 | +## Changing a protocol |
| 236 | + |
| 237 | +- Add a request kind: the message in `py_context` (API and embedded loop), |
| 238 | + the `ctx_request_type_t` and execute function in C, the `request/6` |
| 239 | + clause in `py_isolated`, and `_on_request` in `_isolated.py`. |
| 240 | +- Add a control: `send_frame(Child, 0, ?STATUS_CONTROL, Term)` in |
| 241 | + `py_isolated` and `_on_control` in `_isolated.py`; controls must be safe |
| 242 | + to handle on the reader thread. |
| 243 | +- Add a child event: `self.event(...)` in `_isolated.py` and an |
| 244 | + `{ok, {0, ?STATUS_EVENT, ...}}` clause in `drain_socket` or `handshake`. |
| 245 | +- Change the response body: three writers and two parsers listed in 3.4. |
| 246 | +- Never change the frame header: the child, the callback pipe reader and |
| 247 | + the async pipe parser share it. |
0 commit comments