Compare commits

...
Author SHA1 Message Date
f1be024f5c perf: cache inspect.signature results in RunnableCallable.__init__
Avoids repeated signature introspection for the same function (e.g. 1000
ChannelWrite instances all inspecting the same _write/_awrite methods).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-07-27 17:08:39 +00:00
f61e9b45b6 perf: cache UntrackedValue isinstance scan in put_writes
The `put_writes` method was scanning all channels with
`any(isinstance(ch, UntrackedValue) ...)` on every call. For the
sequential_1000 benchmark this produced 1M+ isinstance calls through
the ABC machinery, consuming 40% of total runtime.

Cache the result as `_has_untracked_channels` once in __enter__ and
__aenter__, replacing both scan sites (put_writes and checkpoint
sanitization). This yields a 1.8-2.3x speedup on sequential_1000.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-07-27 17:08:11 +00:00
2 changed files with 22 additions and 7 deletions
@@ -142,6 +142,9 @@ ANY_TYPE = object()
ASYNCIO_ACCEPTS_CONTEXT = sys.version_info >= (3, 11)
# Cache for inspect.signature results, keyed by function object
_SIGNATURE_CACHE: dict[Callable, inspect.Signature] = {}
# List of keyword arguments that can be injected into nodes / tasks / tools at runtime.
# A named argument may appear multiple times if it appears with distinct types.
KWARGS_CONFIG_KEYS: tuple[tuple[str, tuple[Any, ...], str, Any], ...] = (
@@ -315,7 +318,16 @@ class RunnableCallable(Runnable):
raise ValueError("At least one of func or afunc must be provided.")
self.func_accepts: dict[str, tuple[str, Any]] = {}
params = inspect.signature(cast(Callable, func or afunc)).parameters
target = cast(Callable, func or afunc)
try:
sig = _SIGNATURE_CACHE[target]
except (KeyError, TypeError):
sig = inspect.signature(target)
try:
_SIGNATURE_CACHE[target] = sig
except TypeError:
pass # unhashable function, skip caching
params = sig.parameters
for kw, typ, runtime_key, default in KWARGS_CONFIG_KEYS:
p = params.get(kw)
+9 -6
View File
@@ -198,6 +198,7 @@ class PregelLoop:
_migrate_checkpoint: Callable[[Checkpoint], None] | None
submit: Submit
channels: Mapping[str, BaseChannel]
_has_untracked_channels: bool
# Futures from `checkpointer.put_writes` calls that produced delta-channel
# writes. `_checkpointer_put_after_previous` drains this list (swap to a
# local `futs` then reset to `[]` and wait/gather) before putting the
@@ -437,9 +438,7 @@ class PregelLoop:
writes_to_save = writes
# check if any writes are to an UntrackedValue channel
if any(
isinstance(channel, UntrackedValue) for channel in self.channels.values()
):
if self._has_untracked_channels:
# we do not persist untracked values in checkpoints
writes_to_save = [
# sanitize UntrackedValues that are nested within Send packets
@@ -1161,9 +1160,7 @@ class PregelLoop:
elif "counters_since_delta_snapshot" in self.checkpoint_metadata:
del self.checkpoint_metadata["counters_since_delta_snapshot"]
# sanitize TASK channel in the checkpoint before saving (durability=="exit")
if TASKS in self.checkpoint["channel_values"] and any(
isinstance(channel, UntrackedValue) for channel in self.channels.values()
):
if TASKS in self.checkpoint["channel_values"] and self._has_untracked_channels:
sanitized_tasks = [
sanitize_untracked_values_in_send(value, self.channels)
if isinstance(value, Send)
@@ -1695,6 +1692,9 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
saver=self.checkpointer,
config=self.checkpoint_config,
)
self._has_untracked_channels = any(
isinstance(ch, UntrackedValue) for ch in self.channels.values()
)
self.stack.push(self._suppress_interrupt)
self.status = "input"
self.step = self.checkpoint_metadata["step"] + 1
@@ -1955,6 +1955,9 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
saver=self.checkpointer,
config=self.checkpoint_config,
)
self._has_untracked_channels = any(
isinstance(ch, UntrackedValue) for ch in self.channels.values()
)
self.stack.push(self._suppress_interrupt)
self.status = "input"
self.step = self.checkpoint_metadata["step"] + 1