Add shared memory regions and shared buffers over iommap

CHANGELOG.md

@@ -26,6 +26,15 @@
 - **Pure-Python ETF codec** (`priv/_erlang_impl/_etf.py`) with the type
   mapping of `py_convert.c`; the child needs no C extension. Integers beyond
   64 bits round-trip exactly in isolated mode.
+- **Shared memory** - `py_shm:new/1,2`, `write/3`, `read/3`, `binary/3`
+  (no copy), `close/1`: fixed-size regions over
+  [iommap](https://hex.pm/packages/iommap) (optional dependency) that any
+  context mode maps as `erlang.SharedMemory` (buffer protocol, numpy
+  friendly). `py_buffer:new(#{shared => true})` is a streaming buffer over
+  such a region with ring backpressure, usable as `wsgi.input` in isolated
+  contexts. Handles are plain terms and travel inside any argument or result;
+  `py_shm:read_only/1` and `new(Size, #{writable => false})` hand Python a
+  read-only mapping.
 - `py:python_executable/0`, `py:kill/1`, `py_nif:os_kill/2`.
 - `py_isolated` is a `gen_statem` (states `idle`, `{busy, Id}`, `looping`,
   `stopping_loop`, `{restarting, Reason}`): `sys:get_state/1` and

README.md

@@ -623,7 +623,9 @@ When creating Python contexts, you can choose the execution mode:
 **Isolated mode** is the only mode with a hard bound: `py_context:interrupt/1`
 stops a blocking C call, and `SIGKILL` is the backstop. It costs a process per
 context (about 16 MB and 40 ms to start) and roughly twice the call latency.
-See [Isolated Contexts](docs/isolated.md).
+Bulk data crosses through shared memory (`py_shm`, with the optional
+[iommap](https://hex.pm/packages/iommap) dependency). See
+[Isolated Contexts](docs/isolated.md).
 
 **Worker mode is recommended** because it works with any Python version and automatically benefits from free-threaded Python (3.13t+) when available. Each context owns a dedicated pthread, providing stable thread affinity for libraries with thread-local state (numpy, torch, tensorflow).
 

c_src/py_convert.c

@@ -404,6 +404,18 @@ static ERL_NIF_TERM py_to_term_d(ErlNifEnv *env, PyObject *obj, int depth) {
         return enif_make_atom(env, atom_obj->name);
     }
 
+    /* Shared memory wrappers (_erlang_impl._shm) travel as their handle tuple */
+    if (PyObject_HasAttrString(obj, "to_term") &&
+        PyObject_HasAttrString(obj, "_mmap")) {
+        PyObject *term = PyObject_CallMethod(obj, "to_term", NULL);
+        if (term != NULL) {
+            ERL_NIF_TERM result = py_to_term_d(env, term, depth + 1);
+            Py_DECREF(term);
+            return result;
+        }
+        PyErr_Clear();
+    }
+
     /* Handle NumPy arrays by converting to Python list first */
     if (is_numpy_ndarray(obj)) {
         PyObject *tolist = PyObject_CallMethod(obj, "tolist", NULL);
@@ -601,6 +613,37 @@ static PyObject *term_to_py_d(ErlNifEnv *env, ERL_NIF_TERM term, int depth) {
         }
     }
 
+    /* {'$py_shm', Id, Path, Size} / {'$py_buffer', Id, Path, Ring}: a shared
+     * region handle, turned into the Python wrapper (mapped once per
+     * interpreter, see priv/_erlang_impl/_shm.py). */
+    {
+        int arity4;
+        const ERL_NIF_TERM *el;
+        if (enif_get_tuple(env, term, &arity4, &el) && arity4 == 4) {
+            char tag_buf[16];
+            if (enif_get_atom(env, el[0], tag_buf, sizeof(tag_buf), ERL_NIF_LATIN1) &&
+                (strcmp(tag_buf, "$py_shm") == 0 || strcmp(tag_buf, "$py_shm_ro") == 0 ||
+                 strcmp(tag_buf, "$py_buffer") == 0)) {
+                PyObject *mod = PyImport_ImportModule("_erlang_impl._shm");
+                if (mod == NULL) {
+                    return NULL;
+                }
+                PyObject *id = term_to_py_d(env, el[1], depth + 1);
+                PyObject *path = term_to_py_d(env, el[2], depth + 1);
+                PyObject *size = term_to_py_d(env, el[3], depth + 1);
+                PyObject *result = NULL;
+                if (id != NULL && path != NULL && size != NULL) {
+                    result = PyObject_CallMethod(mod, "from_term", "sOOO", tag_buf, id, path, size);
+                }
+                Py_XDECREF(id);
+                Py_XDECREF(path);
+                Py_XDECREF(size);
+                Py_DECREF(mod);
+                return result;
+            }
+        }
+    }
+
     /* Check list (must come after binary to preserve structure) */
     if (enif_get_list_length(env, term, &list_len)) {
         PyObject *list = PyList_New(list_len);

docs/isolated.md

@@ -202,6 +202,73 @@ socket with `SCM_RIGHTS`:
 Inside a coroutine, `await erlang.async_call(name, *args)` keeps the loop
 running while Erlang answers.
 
+## Bulk data with shared memory
+
+Arguments and results cross the socket as a copy. For large payloads use a
+shared region: a file mapped `MAP_SHARED` on both sides through
+[iommap](https://hex.pm/packages/iommap). Add it to your deps:
+
+```erlang
+{deps, [{iommap, "1.1.3"}]}.
+```
+
+A region is a fixed-size handle you pass like any other argument, in any
+context mode:
+
+```erlang
+{ok, Shm} = py_shm:new(64 * 1024 * 1024),
+ok = py_shm:write(Shm, 0, Floats),                       %% one copy
+{ok, Sum} = py_context:call(Ctx, myapp, sum_floats, [Shm]),
+Out = py_shm:binary(Shm, 0, 1024),                       %% no copy
+ok = py_shm:close(Shm).
+```
+
+```python
+import numpy
+
+def sum_floats(shm):                        # erlang.SharedMemory
+    a = numpy.frombuffer(shm.buffer, dtype=numpy.float32)   # no copy
+    a[:1024] = 0                            # Erlang sees it
+    return float(a.sum())
+```
+
+Python-produced data is zero-copy in both directions (write into the
+region, read it in Erlang with `py_shm:binary/3`); Erlang-produced data
+costs one `write/3` copy instead of encode, socket, decode and copy. A
+region is mapped once per interpreter and reused across calls; it is
+closed by `close/1` or when its owner process exits.
+
+Streaming bodies use the same mechanism through `py_buffer`:
+
+```erlang
+{ok, Buf} = py_buffer:new(#{shared => true}),            %% 4 MB ring
+ok = py_buffer:write(Buf, Chunk),                        %% blocks when full
+ok = py_buffer:close(Buf),
+py_context:call(Ctx, myapp, handle, [#{<<"wsgi.input">> => Buf}]).
+```
+
+The Python side gets `erlang.SharedBuffer` with the `read`, `readline`,
+`readlines`, iteration and `read_nonblock` of the native buffer. Flow
+control is a callback round trip per blocking read, so in an embedded
+context the native `py_buffer:new/0,1` is still the cheaper choice; use
+`shared => true` when the buffer may reach an isolated context or a pool
+mixing modes. A native buffer cannot cross into a child.
+
+Rules: regions are never resized (a truncated file would be a `SIGBUS`, so
+the size is checked when mapping); mapped pages count against the child's
+`as` limit and its resident set; `/dev/shm` is used when present, else a
+private directory under `TMPDIR`; while a call holds a handle the child
+owns the region, and a concurrent `write/3` is a caller error.
+
+What sharing changes about isolation: the child can still only crash or
+exhaust itself, but it can write anything into a region it holds, at any
+time, and `py_shm:binary/3` sees those bytes (a binary that changes under
+you; take it once the callee is done, or copy with `read/3`). Hand the child
+a read-only handle when it only needs to read: `py_shm:new(Size, #{writable => false})`
+or `py_shm:read_only(Shm)` map it `PROT_READ` in Python, and Erlang keeps
+writing. A child that runs as your user could still truncate the region
+file on purpose; sealing and syscall filtering are separate hardening work.
+
 ## Process model
 
 - One child per context, started with `open_port` so the VM reaps it and
@@ -252,12 +319,7 @@ running while Erlang answers.
   seccomp (Linux) or Capsicum (FreeBSD) sandbox is a separate hardening step.
 - Each call copies its arguments and result through the socket: a 1 MB
   binary round-trips in about 1.3 ms, 16 MB in about 27 ms (worker mode:
-  0.2 ms and 3 ms). For bulk data prefer a file, a socket the child reads
-  itself, or a shared mapping: `erlang-iommap` opens a file `MAP_SHARED` and
-  `region_binary/3` gives a binary over it, while the child maps the same
-  file with `mmap`; a 64 MB region costs 3 us on the Erlang side and 5 ms to
-  map in Python, against 12 ms to copy it through ETF. `py_buffer` is not
-  ported to that yet.
+  0.2 ms and 3 ms). For bulk data use shared memory (below).
 
 ## See also
 

priv/_erlang_impl/_etf.py

@@ -42,9 +42,17 @@
 
 __all__ = [
     'Atom', 'Pid', 'Ref', 'Port',
-    'encode', 'decode', 'DecodeError',
+    'encode', 'decode', 'DecodeError', 'register_encoder',
 ]
 
+# (predicate, to_term) pairs consulted before the generic fallback
+_encoders = []
+
+
+def register_encoder(predicate, to_term):
+    """Encode objects matching `predicate` as the term `to_term(obj)` returns."""
+    _encoders.append((predicate, to_term))
+
 VERSION = 131
 
 # Tags
@@ -234,6 +242,10 @@ def _encode(obj, out):
     elif isinstance(obj, (set, frozenset)):
         _encode(list(obj), out)
     else:
+        for predicate, to_term in _encoders:
+            if predicate(obj):
+                _encode(to_term(obj), out)
+                return
         # Same fallback as py_to_term: the string representation as a binary
         _encode(str(obj), out)
 

priv/_erlang_impl/_isolated.py

@@ -58,8 +58,12 @@
 import traceback
 
 from . import _etf
+from . import _shm
 from ._etf import Atom, Pid, Ref, Port, DecodeError
 
+# Shared memory wrappers returned to Erlang travel as their handle tuple
+_etf.register_encoder(_shm.is_shared, lambda obj: obj.to_term())
+
 __all__ = ['Runtime', 'install_erlang_module']
 
 STATUS_REQUEST = 0
@@ -379,6 +383,8 @@ def _on_control(self, term):
                 # raises when the nested request finishes
                 self.inbox.put(('interrupt', target))
             # else: already finished (or still queued: cancel handles that)
+        elif isinstance(term, tuple) and len(term) == 2 and term[0] == 'shm_close':
+            _shm.forget(term[1])
         elif isinstance(term, tuple) and len(term) == 2 and term[0] == 'cancel':
             with self._cancel_lock:
                 self._cancelled.add(term[1])
@@ -425,7 +431,8 @@ def _on_submit(self, frame_id, term):
         def schedule():
             try:
                 fn = _resolve(module, func)
-                result = fn(*_as_list(args), **_as_dict(kwargs))
+                result = fn(*_shm.convert_args(_as_list(args)),
+                            **_shm.convert_args(_as_dict(kwargs)))
                 if inspect.isawaitable(result):
                     task = asyncio.ensure_future(result)
                     task.add_done_callback(
@@ -554,11 +561,12 @@ def _dispatch(self, tag, term):
         if tag == 'call':
             _, module, func, args, kwargs = term
             fn = _resolve(module, func, self.globals)
-            result = fn(*_as_list(args), **_as_dict(kwargs))
+            result = fn(*_shm.convert_args(_as_list(args)),
+                        **_shm.convert_args(_as_dict(kwargs)))
         elif tag == 'eval':
             _, code, locals_ = term
             loc = dict(self.globals)
-            loc.update(_as_dict(locals_))
+            loc.update(_shm.convert_args(_as_dict(locals_)))
             result = eval(compile(_as_text(code), '<erlang>', 'eval'), self.globals, loc)
         elif tag == 'exec':
             _, code = term
@@ -792,12 +800,14 @@ def __getattr__(name):
         return Function(name)
 
     from . import _server as server
+    from ._shm import SharedMemory, SharedBuffer
 
     ns = dict(
         call=call, async_call=async_call, send=send, whereis=whereis,
         self=self_, atom=atom, Atom=Atom, Pid=Pid, Ref=Ref, Port=Port,
         ProcessError=ProcessError, SuspensionRequired=SuspensionRequired,
         Function=Function, is_isolated=is_isolated, run=run,
+        SharedMemory=SharedMemory, SharedBuffer=SharedBuffer,
         new_event_loop=new_event_loop, get_event_loop_policy=get_event_loop_policy,
         install=install, spawn_task=spawn_task, sleep=sleep, log=log,
         server=server, __getattr__=__getattr__,