Module

_subinterpreter_bootstrap

Subinterpreter worker bootstrap.

This module providesbootstrap(), the entry point executed inside each subinterpreter worker. The supervisor injects IIC-safe values via interp.prepare_main()and then runs::

from pounce._subinterpreter_bootstrap import bootstrap
bootstrap(ctrl_queue, status_queue, config_json,
          app_import_path, sock_fd, worker_id, parent_sys_path)

Design

Rather than duplicating Worker internals, the bootstrap:

  1. ReconstructsServerConfigfrom JSON
  2. Imports the ASGI app by module path
  3. Reconstructs the socket from a dup'd file descriptor
  4. Creates a standardWorker with shutdown_event=None
  5. Monkey-patchesWorker._serveto inject an IIC shutdown bridge that pollsctrl_queue and sets _async_shutdownwhen commanded

This keeps Worker as the single source of truth for request handling.

7Functions

Functions

bootstrap 9 None ▼
Bootstrap a Worker inside a subinterpreter. All arguments are IIC-safe types i…
def bootstrap(ctrl_queue: Any, status_queue: Any, config_json: str, lifespan_state_json: str, app_import_path: str, sock_fd: int, sock_family: int, worker_id: int, parent_sys_path: tuple[str, ...]) -> None

Bootstrap a Worker inside a subinterpreter.

All arguments are IIC-safe types injected by the supervisor via interp.prepare_main(). sock_familyis the parent socket's address family (AF_INET / AF_INET6 / AF_UNIX, as an int) so the reconstructed socket matches the bound listener.

Parameters
Name Type Description
ctrl_queue Any
status_queue Any
config_json str
lifespan_state_json str
app_import_path str
sock_fd int
sock_family int
worker_id int
parent_sys_path tuple[str, ...]
_run_worker_with_iic 3 None ▼
Run Worker._serve() with an IIC-based shutdown bridge. This replicates the set…
async
async def _run_worker_with_iic(worker: Any, ctrl_queue: Any, status_queue: Any) -> None

Run Worker._serve() with an IIC-based shutdown bridge.

This replicates the setup that Worker._serve() does, then injects an IIC polling task alongside the normal accept loop.

Parameters
Name Type Description
worker Any
ctrl_queue Any
status_queue Any
_iic_bridge 3 None ▼
Poll the IIC ctrl_queue and translate commands to Worker state changes.
async
async def _iic_bridge(worker: Any, ctrl_queue: Any, status_queue: Any) -> None
Parameters
Name Type Description
worker Any
ctrl_queue Any
status_queue Any
_import_app 1 Any ▼
Import an ASGI app by dotted path (e.g. ``'myapp.main:app'``). Supports factor…
def _import_app(app_path: str) -> Any

Import an ASGI app by dotted path (e.g.'myapp.main:app').

Supports factory syntax:'myapp.main:create_app()'.

Parameters
Name Type Description
app_path str
Returns
Any
_try_get 1 tuple[Any, ...] | None ▼
Non-blocking get from an IIC queue. Returns None if empty.
def _try_get(queue: Any) -> tuple[Any, ...] | None
Parameters
Name Type Description
queue Any
Returns
tuple[Any, ...] | None
_noop_receive 0 dict[str, str] ▼
Noop receive for lifecycle hooks.
async
async def _noop_receive() -> dict[str, str]
Returns
dict[str, str]
_noop_send 1 None ▼
Noop send for lifecycle hooks.
async
async def _noop_send(message: dict[str, Any]) -> None
Parameters
Name Type Description
message dict[str, Any]