symfonic.kernel.contracts.run_census

run_census

Tasks the kernel's task factory never saw: they fail closed.

run_spawn records the spawner's lineage only for a task built through the kernel's factory. A host can build one without it: replace the loop's factory (and even restore it before the task's run), or construct asyncio.Task directly. Such a task, spawned with a stripped context inside a run, would start a fresh root and launder what it was handed after a forget (CORE-199).

Nothing records when such a task was created, so the kernel takes a census of the loop's tasks: when a task it never saw asks for the lineage it was spawned under, and at least every :data:CENSUS_EVERY runs issued on the loop. Each unseen task the census finds descends from every run that could have been live when it was created -- every run alive at the previous census, held until this one, and every run issued since. That over-approximates on purpose (the owner's fail-closed rule): a forget of any scope those runs captured blocks the unseen task, as it blocks a detached worker. A host that wants exact lineage keeps the kernel's factory or chains to it -- wrapping the coroutine first, as Sentry's patch_asyncio does, keeps it exact too, since the kernel records the task its inner factory returns. A task the census misses -- not enumerated, or gone from its registry before the lookup -- still descends from a window, never starts a root.

When a loop's first run is issued from a task -- which predates the kernel's factory -- the census runs before that run exists, so tasks already on the loop start as roots. If a run precedes the loop's first census (one issued from a callback outside any task), earlier tasks fail closed like the rest.

Retention: a run's lineage, and every fence it captured, stays alive until the next census on its loop -- forced within :data:CENSUS_EVERY further runs there; a loop that issues no more runs keeps it until it closes and another loop begins -- so a task built during the run can still inherit it. A run issued meanwhile that names the ended run as its parent or root descends from it too.

Cost (measured on one CPython 3.12 host): a census walks asyncio.all_tasks(loop) once, about 0.45 us per live task -- 0.5 ms at 2,000 tasks, 3 ms at 12,000, 23 ms at 52,000. A loop pays it once every :data:CENSUS_EVERY runs, which adds about 1.5-2.5 us per run up to 1,000 live tasks and about 12 us at 10,000, and again whenever an unseen task the last census missed asks for its lineage. Building a task through the kernel's factory costs about 4.5 us more on a CPython 3.12 dev host (1.9 us before the build mark), since every task is recorded and every build is marked where the new task runs, for an eager first step (run_build). A run in an unseen task re-proves every fence its window holds on each mutation -- one generation read per distinct fence -- so the bypass path, not the default one, pays for the over-approximation.

census

census(loop: Any, seen: Callable[[Any], bool]) -> _Census

The census of loop, begun now if this is its first run.

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

def census(loop: Any, seen: Callable[[Any], bool]) -> _Census:
    """The census of ``loop``, begun now if this is its first run."""
    found = _of(loop)
    if found is None:
        for key, other in tuple(_CENSUSES.items()):
            ended = other.loop()
            if ended is None or ended.is_closed():
                _CENSUSES.pop(key, None)  # another loop's thread may sweep too
        found = _CENSUSES[id(loop)] = _Census(loop, seen)
    return found

note_issued

note_issued(lineage: Any) -> None

Hold a run issued on the running loop until a census has seen it.

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

def note_issued(lineage: Any) -> None:
    """Hold a run issued on the running loop until a census has seen it."""
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        return
    found = _of(loop)
    if found is None:
        return
    found.alive.add(lineage)
    found.held.append(lineage)
    found.issued += 1
    if found.issued >= CENSUS_EVERY:
        found.take(loop)

take_census

take_census(loop: Any) -> None

Classify every task on loop the kernel has not seen; started loops only.

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

def take_census(loop: Any) -> None:
    """Classify every task on ``loop`` the kernel has not seen; started loops only."""
    found = _of(loop)
    if found is not None:
        found.take(loop)

unseen_lineage

unseen_lineage(task: Any) -> Any

The lineage an unseen task descends from; None for a root.

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

def unseen_lineage(task: Any) -> Any:
    """The lineage an unseen ``task`` descends from; ``None`` for a root."""
    found = _of(task.get_loop())
    if found is None:
        return None
    if task not in found.unseen:
        found.take(task.get_loop(), task)
    if task in found.unseen:
        return found.unseen[task]
    # Gone from the registry since the census classified it: every run still
    # held -- a superset of that census's window -- never a root.
    return found.unseen.setdefault(task, _window(tuple(found.alive)))