Pregel algorithm layer: interrupt, write-back, scheduling

源码版本1.2.9

Responsibilities

_algo.py is the "mathematical kernel" of PregelLoop and PregelRunner. The loop is responsible for "how many iterations to run", the runner for "how to execute concurrently", and algo for "the semantics of each step": when to interrupt, how to merge node writes back into channels, and which nodes to trigger next. It has no state and no IO — it is all pure functions, taking checkpoint + channels + tasks as input and producing a new channel version table / updated_channels set / a new task dict.

The three core functions correspond exactly to the three calls in PregelLoop.tick + after_tick: should_interrupt (should_interrupt:155) is called mid-tick to decide whether to raise GraphInterrupt before execution; apply_writes (apply_writes:232) is called at the start of after_tick to merge all task writes from this step into channels and return updated_channels; prepare_next_tasks (prepare_next_tasks:392) is called at the start of tick to compute which tasks to execute this step from the current checkpoint.

In other words, these three functions decide "what the BSP loop sees at each step, what it does, and what the channels look like after wrap-up". Understanding them means understanding Pregel's "state transition rules".

Design motivation

Why split the algorithmic logic into pure functions?

• Testability: should_interrupt / apply_writes / prepare_next_tasks are all pure functions, taking a checkpoint dict + channel mapping as input and producing data structures. They can be unit-tested in isolation from Pregel's IO, submit, and checkpointer — which is exactly what most tests in libs/langgraph/tests/unit/ look like.
• Sync/async sharing: PregelLoop has both SyncPregelLoop and AsyncPregelLoop subclasses, but the algo layer is completely side-effect free, so both subclasses share the same logic (tick calls prepare_next_tasks:612). Otherwise the algorithm would have to be written twice for sync and async, with bugs fixed on both sides.
• updated_channels optimization: apply_writes returns the updated_channels set, which prepare_next_tasks accepts as a hint to skip "scanning all nodes to check whether their triggers fire" (updated_channels hint:475-486). On large graphs this reduces an O(N×M) trigger scan to O(updated×triggered) — the single most important performance optimization.
• should_interrupt dedupes via versions_seen: the versions_seen of the INTERRUPT channel (versions_seen INTERRUPT:163) records the channel versions seen at the last interrupt. Only "channels with new updates since the last interrupt" are considered for another interrupt — preventing a death loop of trigger-interrupt-resume-interrupt.
• The finish signal at the end of apply_writes: when bump_step and updated_channels.isdisjoint(trigger_to_nodes), channels[chan].finish() is called (finish:336-342) to send all channels the "this is the last superstep" signal. Permanent channels (e.g. LastValue) use this opportunity to mark themselves unavailable so prepare_next_tasks triggers no further nodes — this is the graph's "natural death".

Key files

• should_interrupt:155-185 — checks whether the graph should interrupt; first looks at versions_seen[INTERRUPT] to decide whether there are new updates, then whether the triggered tasks fall in the interrupt_nodes list.
• any_updates_since_prev_interrupt:161-168 — compares channel versions to decide "whether there have been updates since the last interrupt"; the core of interrupt deduplication.
• interrupt_nodes filter:170-184 — interrupt_nodes == "*" means all non-hidden tasks count; otherwise filter by task.name in interrupt_nodes.
• apply_writes:232-345 — merges this step's task writes into channels and returns the updated_channels set.
• bump_step:256-259 — any task with triggers sets bump_step=True, meaning this step consumes channels and advances the step counter.
• update seen versions:262-269 — each task records the version of the trigger channels it read into versions_seen, so the next _triggers check becomes "has there been a new update since the last execution".
• consume channels:284-292 — calls channels[chan].consume() to mark channels read this step as consumed; the implementation of the PULL channel "read once then clear" semantics.
• apply writes to channels:315-323 — channels[chan].update(vals) actually merges writes into the channel; only is_available() channels are added to updated_channels.
• bump_step notify:326-333 — when bump_step=True, available channels not updated this step also get update(EMPTY_SEQ), advancing their version — so nodes subscribed to those channels but not triggered this step will not be erroneously fired on the next tick.
• finish signal:336-342 — when no node is triggered by this step's updates, all channels get finish(); permanent channels become unavailable and the graph ends naturally.
• prepare_next_tasks:392-513 — computes which tasks to run next; both PUSH (from Send fan-out) and PULL (from edge triggers) originate here.
• updated_channels optimization:475-486 — uses the previous step's updated_channels + trigger_to_nodes to reverse-lookup the triggered node set, skipping the full scan.
• prepare_single_task:524 — single task construction; PUSH goes through prepare_push_task_functional / prepare_push_task_send, PULL goes through the _triggers check + _proc_input to fetch input.
• _triggers:1260-1277 — checks whether a node's trigger channels have "an unread new version"; the core condition for PULL scheduling.
• local_read:188-224 — entry point for node functions reading channel values; supports fresh=True to read a local view of "what I just wrote but has not been merged into the channel", for the write-then-read semantics of conditional edges.

Data flow

should_interrupt is the simplest of the three; its logic is "there are new updates + the current task hits the list":

def should_interrupt(
    checkpoint: Checkpoint,
    interrupt_nodes: All | Sequence[str],
    tasks: Iterable[PregelExecutableTask],
) -> list[PregelExecutableTask]:
    """Check if the graph should be interrupted based on current state."""
    version_type = type(next(iter(checkpoint["channel_versions"].values()), None))
    null_version = version_type()  # type: ignore[misc]
    seen = checkpoint["versions_seen"].get(INTERRUPT, {})
    # interrupt if any channel has been updated since last interrupt
    any_updates_since_prev_interrupt = any(
        version > seen.get(chan, null_version)  # type: ignore[operator]
        for chan, version in checkpoint["channel_versions"].items()
    )
    # and any triggered node is in interrupt_nodes list
    return (
        [task for task in tasks if ...]
        if any_updates_since_prev_interrupt
        else []
    )

This comes from should_interrupt:155-185. The key is the versions_seen[INTERRUPT] record — it is not the version a node has read, but "the version snapshot of all channels at the time of the last interrupt". any_updates_since_prev_interrupt compares this snapshot with the current channel_versions; as long as any channel has a higher version, it counts as "there is something new", and only then is the interrupt check entered. This way, even if a resume triggers a node with the same name on the next tick, it will not immediately interrupt again — it must wait until a channel genuinely has a new write.

apply_writes is the core of the wrap-up phase: first update versions_seen, then consume the channels that were read, and finally write the new values:

# update seen versions
for task in tasks:
    checkpoint["versions_seen"].setdefault(task.name, {}).update(
        {
            chan: checkpoint["channel_versions"][chan]
            for chan in task.triggers
            if chan in checkpoint["channel_versions"]
        }
    )

# Consume all channels that were read
for chan in {
    chan
    for task in tasks
    for chan in task.triggers
    if chan not in RESERVED and chan in channels
}:
    if channels[chan].consume() and next_version is not None:
        checkpoint["channel_versions"][chan] = next_version

# Apply writes to channels
updated_channels: set[str] = set()
for chan, vals in pending_writes_by_channel.items():
    if chan in channels:
        if channels[chan].update(vals) and next_version is not None:
            checkpoint["channel_versions"][chan] = next_version
            if channels[chan].is_available():
                updated_channels.add(chan)

This comes from apply_writes core:262-323. The three phases have clear semantics:

1. Update versions_seen — record each task's current trigger channel versions as evidence that "this node has seen this version of the channel". The next _triggers check compares against future versions; only an update will trigger again.
2. consume() channels that were read — channels like Topic with "read once then clear" semantics empty their internal buffer in consume(), keeping only the version advance. This is why a Send-fanned-out message does not trigger other nodes after being consumed.
3. update(vals) writes new values — all writes from all tasks this step are collected per channel and merged in at once, invoking the merge logic of reducers / LastValue and similar channels. Only is_available() channels (with values and not finished) end up in updated_channels — channels after finish() advance their version but trigger no nodes.

prepare_next_tasks is called at the start of tick and decides which tasks to run this step:

# Consume pending tasks
tasks_channel = cast(Topic[Send] | None, channels.get(TASKS))
if tasks_channel and tasks_channel.is_available():
    for idx, _ in enumerate(tasks_channel.get()):
        if task := prepare_single_task(
            (PUSH, idx), None, ..., for_execution=for_execution, ...
        ):
            tasks.append(task)

# This section is an optimization that allows which nodes will be active
# during the next step.
if updated_channels and trigger_to_nodes:
    triggered_nodes: set[str] = set()
    for channel in updated_channels:
        if node_ids := trigger_to_nodes.get(channel):
            triggered_nodes.update(node_ids)
    candidate_nodes: Iterable[str] = sorted(triggered_nodes)
elif not checkpoint["channel_versions"]:
    candidate_nodes = ()
else:
    candidate_nodes = processes.keys()

for name in candidate_nodes:
    if task := prepare_single_task((PULL, name), None, ..., for_execution=for_execution, ...):
        tasks.append(task)
return {t.id: t for t in tasks}

This comes from prepare_next_tasks core:441-513. Two kinds of tasks converge here:

• PUSH task: pulls Send objects from the TASKS channel — these were fanned out by some node return Send(...) in the previous step; each Send corresponds to a PUSH task.
• PULL task: reverse-looks-up updated_channels through trigger_to_nodes — this mapping is computed at compile time as "channel -> list of subscribed nodes", and taking the union gives the candidate nodes to trigger this step. sorted ensures deterministic task order, which does not affect execution results but does affect checkpoint replay.

Each candidate node then goes through _triggers for a second confirmation (_triggers call:606-612) — "the channel has a new version and the node has not seen this version" — before a task is actually generated. This is the key to PULL node deduplication: even if updated_channels contains a node's trigger channel, if versions_seen[name] has already recorded this version, the task will not be rescheduled.

The position of the algo layer within the Pregel loop:

Boundaries and failure modes

• versions_seen[INTERRUPT] not updating causes a death loop: should_interrupt only returns a non-empty list when any_updates_since_prev_interrupt is true, and the update of versions_seen[INTERRUPT] happens in the implicit logic of _put_checkpoint after apply_writes. If the INTERRUPT channel's seen version is not correctly advanced on resume, the same update is repeatedly judged as "new", causing an interrupt-resume-interrupt death loop (versions_seen INTERRUPT:163-168).
• bump_step is the only signal for step advancement: when any(t.triggers for t in tasks) is false, no bump happens — this is the "null task writes channels but does not advance the step" semantics (bump_step:259). Null tasks are used to inject input writes before tick; they should not bump the step counter.
• finish() is called only when "no node is triggered": bump_step and updated_channels.isdisjoint(trigger_to_nodes) (finish condition:336) — as long as any channel in updated_channels is subscribed to by a node, do not finish. This means finish is only sent "when the graph dies on the last step", not on every step.
• Unknown channel writes only warn, not error: when pending_writes_by_channel contains a channel name not in channels, the code just logger.warning (unknown channel warn:310-313). This is for fault tolerance — e.g. a node writing to a channel that has already been finished should not crash the whole graph.
• local_read with fresh=True copies the channel: fresh reads do channels[k].copy() per channel (fresh read copy:213-219), applying this task's writes on the copy — so conditional edge nodes can read their own just-written local view before apply_writes without affecting the global state seen by other tasks.
• The task id hash algorithm in prepare_single_task: when checkpoint["v"] > 1, _xxhash_str is used; otherwise _uuid5_str (task_id_func:550) — this is checkpoint-version compatibility: old checkpoints use task ids generated by uuid5 and must be replayable.

Summary

_algo.py's three pure functions — should_interrupt / apply_writes / prepare_next_tasks — cleanly abstract Pregel's semantic layer: when to interrupt, how to merge writes back, and who to run next. They are stateless and IO-free, driven by PregelLoop.tick + after_tick to power the entire BSP loop. See /pregel/loop for the loop driver layer, /pregel/runner for the execution layer, and /pregel/pregel for the overall assembly. The semantics of the channel methods update / consume / finish / is_available are in /channel/base-channel.

See official docs: LangGraph docs · README.