perf: track available channels set to eliminate O(n) scan in apply_writes

Maintain a set of currently-available channel names, updated incrementally
as channels change state, so the step-bump loop in apply_writes only
iterates available channels instead of scanning all channels with
is_available(). For sequential_1000 this reduces function calls by ~54%
and improves overall runtime by ~26%.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

libs/langgraph/langgraph/pregel/_algo.py

@@ -220,6 +220,7 @@ def apply_writes(
     tasks: Iterable[WritesProtocol],
     get_next_version: GetNextVersion | None,
     trigger_to_nodes: Mapping[str, Sequence[str]],
+    available_channels: set[str] | None = None,
 ) -> set[str]:
     """Apply writes from a set of tasks (usually the tasks from a Pregel step)
     to the checkpoint and channels, and return managed values writes to be applied
@@ -275,6 +276,11 @@ def apply_writes(
     }:
         if channels[chan].consume() and next_version is not None:
             checkpoint["channel_versions"][chan] = next_version
+        if available_channels is not None:
+            if channels[chan].is_available():
+                available_channels.add(chan)
+            else:
+                available_channels.discard(chan)
 
     # Group writes by channel
     pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
@@ -298,16 +304,35 @@ def apply_writes(
                 # unavailable channels can't trigger tasks, so don't add them
                 if channels[chan].is_available():
                     updated_channels.add(chan)
+                    if available_channels is not None:
+                        available_channels.add(chan)
+                elif available_channels is not None:
+                    available_channels.discard(chan)
+            elif available_channels is not None:
+                if channels[chan].is_available():
+                    available_channels.add(chan)
+                else:
+                    available_channels.discard(chan)
 
     # Channels that weren't updated in this step are notified of a new step
     if bump_step:
-        for chan in channels:
-            if channels[chan].is_available() and chan not in updated_channels:
-                if channels[chan].update(EMPTY_SEQ) and next_version is not None:
-                    checkpoint["channel_versions"][chan] = next_version
-                    # unavailable channels can't trigger tasks, so don't add them
-                    if channels[chan].is_available():
-                        updated_channels.add(chan)
+        candidates = (
+            available_channels - updated_channels
+            if available_channels is not None
+            else (
+                chan
+                for chan in channels
+                if channels[chan].is_available() and chan not in updated_channels
+            )
+        )
+        for chan in candidates:
+            if channels[chan].update(EMPTY_SEQ) and next_version is not None:
+                checkpoint["channel_versions"][chan] = next_version
+                # unavailable channels can't trigger tasks, so don't add them
+                if channels[chan].is_available():
+                    updated_channels.add(chan)
+                elif available_channels is not None:
+                    available_channels.discard(chan)
 
     # If this is (tentatively) the last superstep, notify all channels of finish
     if bump_step and updated_channels.isdisjoint(trigger_to_nodes):
@@ -317,6 +342,15 @@ def apply_writes(
                 # unavailable channels can't trigger tasks, so don't add them
                 if channels[chan].is_available():
                     updated_channels.add(chan)
+                    if available_channels is not None:
+                        available_channels.add(chan)
+                elif available_channels is not None:
+                    available_channels.discard(chan)
+            elif available_channels is not None:
+                if channels[chan].is_available():
+                    available_channels.add(chan)
+                else:
+                    available_channels.discard(chan)
 
     # Return managed values writes to be applied externally
     return updated_channels

libs/langgraph/langgraph/pregel/_loop.py

@@ -180,6 +180,8 @@ class PregelLoop:
     _migrate_checkpoint: Callable[[Checkpoint], None] | None
     submit: Submit
     channels: Mapping[str, BaseChannel]
+    _has_untracked_channels: bool
+    _available_channels: set[str]
     managed: ManagedValueMapping
     checkpoint: Checkpoint
     checkpoint_id_saved: str
@@ -545,6 +547,7 @@ class PregelLoop:
             self.tasks.values(),
             self.checkpointer_get_next_version,
             self.trigger_to_nodes,
+            available_channels=self._available_channels,
         )
         # produce values output
         if not self.updated_channels.isdisjoint(
@@ -675,6 +678,7 @@ class PregelLoop:
                 [PregelTaskWrites((), INPUT, null_writes, [])],
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
+                available_channels=self._available_channels,
             )
             if updated_channels is not None:
                 updated_channels.update(null_updated_channels)
@@ -717,6 +721,7 @@ class PregelLoop:
                 ],
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
+                available_channels=self._available_channels,
             )
             # save input checkpoint
             self.updated_channels = updated_channels
@@ -844,6 +849,7 @@ class PregelLoop:
                     self.tasks.values(),
                     self.checkpointer_get_next_version,
                     self.trigger_to_nodes,
+                    available_channels=self._available_channels,
                 )
                 if not updated_channels.isdisjoint(
                     (self.output_keys,)
@@ -1114,6 +1120,9 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         self.channels, self.managed = channels_from_checkpoint(
             self.specs, self.checkpoint
         )
+        self._available_channels: set[str] = {
+            k for k, v in self.channels.items() if v.is_available()
+        }
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1
@@ -1295,6 +1304,9 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         self.channels, self.managed = channels_from_checkpoint(
             self.specs, self.checkpoint
         )
+        self._available_channels: set[str] = {
+            k for k, v in self.channels.items() if v.is_available()
+        }
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1