symfonic.kernel.contracts.run_spawn

run_spawn

The lineage a task keeps when it is spawned with a stripped context.

asyncio.create_task(coro, context=contextvars.Context()) inside a kernel run starts coro with no lineage at all. Owner ruling (CORE-199): if that coroutine begins a new top-level run, the run descends from the run that spawned it, so it inherits the spawner's erase captures instead of observing fresh ones that would launder what it was handed after a forget.

The kernel wraps the running loop's task factory when it issues a run. The wrapper runs in the spawning context, so it records the spawner's lineage -- or, for a task spawned by a stripped task, that task's recorded lineage -- against the task its inner factory returns. It keys the task, not the coroutine: a host factory can wrap the coroutine before building the task (Sentry's patch_asyncio does), so task.get_coro() is not the coroutine the kernel saw. An eager inner factory runs the task's first step before it returns; for that step the lineage is the build in progress, which only the stack running the coroutine the kernel was handed claims (run_build). The kernel consults the record only when it issues a run with no lineage in context.

contextvars.Context().run(asyncio.create_task, coro) calls the factory itself under an empty context, so the spawner's lineage is not in context there either. The kernel therefore also records each run it issues against the task that issued it, until that issuance is reset; the factory falls back to the innermost such run of the current task. A task that issued no live run -- or whose run has ended -- hands nothing on.

The same record covers a coroutine the issuing task drives itself under an empty context, Context().run(coro.send, None): a run issued there finds no lineage in context and descends from the task's innermost live run.

A task built without the kernel's factory -- the factory replaced, or asyncio.Task constructed directly -- has no record; run_census finds it and fails closed.

Work handed to the loop's default executor -- loop.run_in_executor(None, ...) and asyncio.to_thread -- runs on another thread, which must run its own loop to issue a run, and run_in_executor copies no context. The kernel installs a default executor that records the submitter's lineage, resolved as for a task (context, issuing run, spawn record), and hands it to the function for exactly its call: a run issued on that thread with no other lineage descends from it. Submitted outside any run, it hands nothing on. A default executor the host set before the loop's first run is wrapped, not replaced; one set after it, or any explicitly passed executor, a raw thread or another process, is not covered.

A loop callback runs in no task. The kernel hands it the lineage of the code that scheduled it -- resolved as for a task -- for exactly its call (run_callbacks), so a task it creates descends from the scheduler; a callback scheduled outside any run, as a server's per-connection callbacks are, hands nothing on.

enter_issuing

enter_issuing(lineage: Any) -> Any

Record lineage as a live run of the current task; return that task.

Source code in src/symfonic/kernel/contracts/run_spawn.py

def enter_issuing(lineage: Any) -> Any:
    """Record ``lineage`` as a live run of the current task; return that task."""
    task = _current_task()
    if task is not None:
        _ISSUING.setdefault(task, []).append(lineage)
    return task

exit_issuing

exit_issuing(task: Any, lineage: Any) -> None

Forget the record :func:enter_issuing made, whatever task resets it.

Source code in src/symfonic/kernel/contracts/run_spawn.py

def exit_issuing(task: Any, lineage: Any) -> None:
    """Forget the record :func:`enter_issuing` made, whatever task resets it."""
    stack = None if task is None else _ISSUING.get(task)
    if not stack:
        return
    for index in range(len(stack) - 1, -1, -1):
        if stack[index] is lineage:
            del stack[index]
            break
    if not stack:
        _ISSUING.pop(task, None)

handed_lineage

handed_lineage() -> Any

The lineage of the submitter whose function this thread is running, if any.

Source code in src/symfonic/kernel/contracts/run_spawn.py

def handed_lineage() -> Any:
    """The lineage of the submitter whose function this thread is running, if any."""
    return getattr(_HANDED, "lineage", None)

issuing_lineage

issuing_lineage() -> Any

The innermost live run the current task issued, if any.

Source code in src/symfonic/kernel/contracts/run_spawn.py

def issuing_lineage() -> Any:
    """The innermost live run the current task issued, if any."""
    task = _current_task()
    stack = None if task is None else _ISSUING.get(task)
    return stack[-1] if stack else None

spawned_lineage

spawned_lineage() -> Any

The lineage the running task was spawned under; fails closed if unseen.

Source code in src/symfonic/kernel/contracts/run_spawn.py

def spawned_lineage() -> Any:
    """The lineage the running task was spawned under; fails closed if unseen."""
    task = _current_task()
    if task is None:
        return _called() or handed_lineage()
    return _spawned_of(task)

track_spawns

track_spawns() -> None

Wrap the running loop's task factory, once; outside a loop, nothing.

Raises ContractViolationError (:data:UNCARRIED), before changing the loop, if the loop refuses the callback carrier (run_callbacks).

Source code in src/symfonic/kernel/contracts/run_spawn.py

def track_spawns() -> None:
    """Wrap the running loop's task factory, once; outside a loop, nothing.

    Raises ``ContractViolationError`` (:data:`UNCARRIED`), before changing the
    loop, if the loop refuses the callback carrier (``run_callbacks``).
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return
    if not track_callbacks(loop, _scheduler_origin):
        from symfonic.core.contracts.errors import ContractViolationError

        raise ContractViolationError(UNCARRIED)
    from symfonic.kernel.contracts.run_census import census

    census(loop, _seen)
    _hand_default_executor(loop)
    current = loop.get_task_factory()
    if type(current) is not _SpawnFactory:
        loop.set_task_factory(_SpawnFactory(current))