From 756786089b5d25b4cb200b5cb02e9da521022934 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Sun, 26 Jul 2026 23:10:45 -0700 Subject: [PATCH 1/2] Make entity method signature caching effective _execute_entity_batch() constructed a new _EntityExecutor inside its per-operation loop, so the executor's _entity_method_cache was always cold and inspect.signature() was re-run for every single entity operation. The executor is now constructed once per batch and shares a bounded, thread-safe signature cache owned by the worker, so a given (entity type, operation) pair is reflected once for the lifetime of the worker rather than once per operation. The cache is bounded (_EntityMethodSignatureCache, 1024 entries, oldest evicted first) because entity types and operation names come from user code and are unbounded in principle. Writes are serialized under a lock so concurrently processed entity work items stay safe. Behavior is unchanged: the dispatch logic in _EntityExecutor.execute() is untouched, so the no-argument versus input-argument distinction and all error messages for unknown or invalid operations are preserved. _EntityExecutor's existing three-argument constructor still works; the shared cache is an optional fourth argument. Fixes #186 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 2265f7b9-f8ad-4b50-a25d-19bd6f1ac96c --- durabletask/worker.py | 65 +++++- tests/durabletask/test_entity_executor.py | 238 +++++++++++++++++++++- 2 files changed, 298 insertions(+), 5 deletions(-) diff --git a/durabletask/worker.py b/durabletask/worker.py index 7efb732c..2c75b357 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -534,6 +534,9 @@ def __init__( data_converter: DataConverter | None = None, ): self._registry = _Registry() + # Shared by every entity batch this worker processes so that reflected + # entity method signatures are computed once instead of per operation. + self._entity_method_cache = _EntityMethodSignatureCache() self._host_address = ( host_address if host_address else shared.get_default_host_address() ) @@ -1370,9 +1373,17 @@ def _execute_entity_batch( raise RuntimeError(f"Invalid entity instance ID '{instance_id}' in entity operation request.") results: list[pb.OperationResult] = [] + # A single executor serves the whole batch and shares the worker's + # bounded signature cache, so entity method reflection is not repeated + # for every operation. + executor = _EntityExecutor( + self._registry, + self._logger, + self._data_converter, + entity_method_cache=self._entity_method_cache, + ) for operation in req.operations: start_time = datetime.now(timezone.utc) - executor = _EntityExecutor(self._registry, self._logger, self._data_converter) operation_result = None @@ -3101,13 +3112,61 @@ def execute( return encoded_output +class _EntityMethodSignatureCache: + """Bounded, thread-safe cache of reflected entity method signatures. + + Dispatching an operation on a class-based entity requires knowing whether + the target method declares a required parameter, which is answered with an + ``inspect.signature()`` call. The answer depends only on the entity type + and the operation name, so it is cached here and shared across entity + batches (and therefore across the worker threads that process them + concurrently). + + The cache is bounded because entity types and operation names originate + from user code and are unbounded in principle; once the limit is reached + the oldest entries are evicted instead of growing without limit. + """ + + DEFAULT_MAX_SIZE = 1024 + + def __init__(self, max_size: int = DEFAULT_MAX_SIZE): + self._max_size = max(1, max_size) + self._lock = Lock() + self._entries: dict[tuple[type, str], bool] = {} + + def get(self, key: tuple[type, str]) -> bool | None: + """Returns the cached result for ``key``, or ``None`` when not cached.""" + return self._entries.get(key) + + def __setitem__(self, key: tuple[type, str], value: bool) -> None: + with self._lock: + self._entries[key] = value + # Dicts preserve insertion order, so the first key is the oldest. + while len(self._entries) > self._max_size: + self._entries.pop(next(iter(self._entries)), None) + + def __contains__(self, key: object) -> bool: + return key in self._entries + + def __len__(self) -> int: + return len(self._entries) + + class _EntityExecutor: def __init__(self, registry: _Registry, logger: logging.Logger, - data_converter: DataConverter): + data_converter: DataConverter, + entity_method_cache: _EntityMethodSignatureCache | None = None): self._registry = registry self._logger = logger self._data_converter = data_converter - self._entity_method_cache: dict[tuple[type, str], bool] = {} + # When the caller supplies a cache (the worker does), reflected entity + # method signatures are shared with it and therefore survive beyond the + # lifetime of this executor. + self._entity_method_cache: _EntityMethodSignatureCache = ( + entity_method_cache + if entity_method_cache is not None + else _EntityMethodSignatureCache() + ) def execute( self, diff --git a/tests/durabletask/test_entity_executor.py b/tests/durabletask/test_entity_executor.py index 42774050..e02e5f66 100644 --- a/tests/durabletask/test_entity_executor.py +++ b/tests/durabletask/test_entity_executor.py @@ -2,13 +2,23 @@ # Licensed under the MIT License. """Unit tests for the _EntityExecutor class in durabletask.worker.""" +import inspect import json import logging +import threading -from durabletask import entities +from google.protobuf import wrappers_pb2 + +import durabletask.internal.orchestrator_service_pb2 as pb +from durabletask import entities, worker from durabletask.internal.entity_state_shim import StateShim from durabletask.serialization import JsonDataConverter -from durabletask.worker import _EntityExecutor, _Registry +from durabletask.worker import ( + TaskHubGrpcWorker, + _EntityExecutor, + _EntityMethodSignatureCache, + _Registry, +) def _make_executor(*entity_args) -> _EntityExecutor: @@ -289,3 +299,227 @@ def test_falsy_serialized_state_is_not_dropped(self): state = StateShim("0", JsonDataConverter(), is_serialized=True) assert state.get_state(int) == 0 assert state.encode_state() == "0" + + +class _CountingInspect: + """Delegates to :mod:`inspect` while counting ``signature()`` calls.""" + + def __init__(self): + self.signature_calls: list[object] = [] + + def signature(self, obj, *args, **kwargs): + self.signature_calls.append(obj) + return inspect.signature(obj, *args, **kwargs) + + def __getattr__(self, name): + return getattr(inspect, name) + + +class _RecordingEntityStub: + """Captures the batch results the worker would send to the sidecar.""" + + def __init__(self): + self.completed: list[pb.EntityBatchResult] = [] + + def CompleteEntityTask(self, batch_result): + self.completed.append(batch_result) + + +def _entity_batch_request(instance_id: str, operations) -> pb.EntityBatchRequest: + """Builds an entity batch request from (operation, encoded_input) pairs.""" + return pb.EntityBatchRequest( + instanceId=instance_id, + operations=[ + pb.OperationRequest( + operation=name, + requestId=str(index), + input=( + wrappers_pb2.StringValue(value=encoded_input) + if encoded_input is not None + else None + ), + ) + for index, (name, encoded_input) in enumerate(operations) + ], + ) + + +class Counter(entities.DurableEntity): + """Entity exercising both input-argument and no-argument dispatch.""" + + def add(self, value: int): + current = self.get_state(int, 0) + self.set_state(current + value) + return current + value + + def get(self): + return self.get_state(int, 0) + + +class TestEntityMethodSignatureCaching: + """Reflection of an entity method signature must happen only once.""" + + def test_repeated_operations_reflect_once_per_executor(self, monkeypatch): + executor = _make_executor(Counter) + counting = _CountingInspect() + monkeypatch.setattr(worker, "inspect", counting) + + entity_id = entities.EntityInstanceId("Counter", "test-key") + state = StateShim(None, JsonDataConverter()) + for _ in range(5): + executor.execute("test-orch", entity_id, "add", state, "1") + state.commit() + + assert len(counting.signature_calls) == 1 + assert state.get_state(int) == 5 + + def test_cache_is_shared_between_executors(self, monkeypatch): + shared_cache = _EntityMethodSignatureCache() + registry = _Registry() + registry.add_entity(Counter) + counting = _CountingInspect() + monkeypatch.setattr(worker, "inspect", counting) + + entity_id = entities.EntityInstanceId("Counter", "test-key") + for _ in range(3): + executor = _EntityExecutor( + registry, logging.getLogger("test"), JsonDataConverter(), + entity_method_cache=shared_cache, + ) + executor.execute( + "test-orch", entity_id, "add", + StateShim(None, JsonDataConverter()), "1") + + assert len(counting.signature_calls) == 1 + + def test_executor_without_shared_cache_still_caches(self, monkeypatch): + # Constructing the executor with the historical three-argument form + # keeps working and still avoids repeated reflection. + registry = _Registry() + registry.add_entity(Counter) + executor = _EntityExecutor( + registry, logging.getLogger("test"), JsonDataConverter()) + counting = _CountingInspect() + monkeypatch.setattr(worker, "inspect", counting) + + entity_id = entities.EntityInstanceId("Counter", "test-key") + state = StateShim(None, JsonDataConverter()) + executor.execute("test-orch", entity_id, "add", state, "1") + executor.execute("test-orch", entity_id, "add", state, "1") + + assert len(counting.signature_calls) == 1 + + +class TestEntityBatchSignatureCaching: + """The worker must not recompute signatures for every batch operation.""" + + @staticmethod + def _make_worker() -> TaskHubGrpcWorker: + instance = TaskHubGrpcWorker() + instance.add_entity(Counter) + return instance + + def test_batch_reflects_once_for_repeated_operations(self, monkeypatch): + instance = self._make_worker() + stub = _RecordingEntityStub() + counting = _CountingInspect() + monkeypatch.setattr(worker, "inspect", counting) + + request = _entity_batch_request("@counter@key", [("add", "1")] * 5) + result = instance._execute_entity_batch(request, stub, b"token") + + assert [r.WhichOneof("resultType") for r in result.results] == ["success"] * 5 + assert json.loads(result.entityState.value) == 5 + assert len(counting.signature_calls) == 1 + assert len(stub.completed) == 1 + + def test_cache_survives_across_batches(self, monkeypatch): + instance = self._make_worker() + stub = _RecordingEntityStub() + counting = _CountingInspect() + monkeypatch.setattr(worker, "inspect", counting) + + for _ in range(3): + instance._execute_entity_batch( + _entity_batch_request("@counter@key", [("add", "1")]), stub, b"token") + + assert len(counting.signature_calls) == 1 + + def test_worker_cache_is_populated_by_batch(self): + instance = self._make_worker() + instance._execute_entity_batch( + _entity_batch_request("@counter@key", [("add", "1")]), + _RecordingEntityStub(), b"token") + + assert (Counter, "add") in instance._entity_method_cache + + def test_no_arg_and_input_arg_dispatch_preserved(self, monkeypatch): + instance = self._make_worker() + stub = _RecordingEntityStub() + counting = _CountingInspect() + monkeypatch.setattr(worker, "inspect", counting) + + request = _entity_batch_request( + "@counter@key", + [("add", "2"), ("get", None), ("add", "3"), ("get", None)], + ) + result = instance._execute_entity_batch(request, stub, b"token") + + assert [r.success.result.value for r in result.results] == ["2", "2", "5", "5"] + # One reflection per distinct operation, regardless of repeat count. + assert len(counting.signature_calls) == 2 + + def test_unknown_operation_failure_is_unchanged(self): + instance = self._make_worker() + result = instance._execute_entity_batch( + _entity_batch_request("@counter@key", [("missing", None)]), + _RecordingEntityStub(), b"token") + + failure = result.results[0] + assert failure.WhichOneof("resultType") == "failure" + details = failure.failure.failureDetails + assert details.errorType == "builtins.AttributeError" + assert "does not have operation 'missing'" in details.errorMessage + + +class TestEntityMethodSignatureCacheBounds: + """The shared signature cache must stay bounded and thread-safe.""" + + def test_get_returns_none_for_unknown_key(self): + cache = _EntityMethodSignatureCache() + assert cache.get((int, "missing")) is None + + def test_false_values_are_distinguishable_from_misses(self): + cache = _EntityMethodSignatureCache() + cache[(int, "op")] = False + assert cache.get((int, "op")) is False + assert (int, "op") in cache + + def test_evicts_oldest_entries_when_full(self): + cache = _EntityMethodSignatureCache(max_size=2) + cache[(int, "op")] = True + cache[(str, "op")] = True + cache[(float, "op")] = True + + assert len(cache) == 2 + assert cache.get((int, "op")) is None + assert cache.get((str, "op")) is True + assert cache.get((float, "op")) is True + + def test_concurrent_writes_stay_within_bound(self): + cache = _EntityMethodSignatureCache(max_size=8) + entity_types = [type(f"Entity{index}", (), {}) for index in range(64)] + barrier = threading.Barrier(4) + + def writer(worker_index: int) -> None: + barrier.wait() + for index, entity_type in enumerate(entity_types): + cache[(entity_type, f"op{worker_index}")] = index % 2 == 0 + + threads = [threading.Thread(target=writer, args=(index,)) for index in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert len(cache) == 8 From 03339f272dbe3bb1e70c21b2876a6db9e0b627cb Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Mon, 27 Jul 2026 09:37:27 -0700 Subject: [PATCH 2/2] Test that a non-positive signature cache bound is clamped _EntityMethodSignatureCache clamps its bound with max(1, max_size), but nothing covered that clamp, so the PR description overstated the test coverage. Add regression tests for it. The clamp is not cosmetic. Without it a bound of 0 evicts each entry immediately after inserting it, so every lookup misses and the entity method reflection this cache exists to avoid runs on every operation again -- silently reintroducing the very problem this change fixes. A negative bound is worse: the eviction loop keeps popping past an already-empty dict and raises StopIteration. The new tests cover a zero and negative bound, that a clamped cache still behaves as a working size-one cache rather than a no-op, and that an executor handed a clamped cache still reflects only once. Verified red before green: with the clamp removed all five fail, including 'assert 3 == 1' on the reflection count. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 2265f7b9-f8ad-4b50-a25d-19bd6f1ac96c --- tests/durabletask/test_entity_executor.py | 55 +++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/tests/durabletask/test_entity_executor.py b/tests/durabletask/test_entity_executor.py index e02e5f66..3e2b75a7 100644 --- a/tests/durabletask/test_entity_executor.py +++ b/tests/durabletask/test_entity_executor.py @@ -7,6 +7,7 @@ import logging import threading +import pytest from google.protobuf import wrappers_pb2 import durabletask.internal.orchestrator_service_pb2 as pb @@ -523,3 +524,57 @@ def writer(worker_index: int) -> None: thread.join() assert len(cache) == 8 + + @pytest.mark.parametrize("max_size", [0, -1, -1024]) + def test_non_positive_max_size_is_clamped_to_one(self, max_size: int): + """A non-positive bound must not produce a zero-capacity cache. + + Without clamping, a bound of 0 would evict every entry immediately + after inserting it, so every lookup would miss and the signature + reflection this cache exists to avoid would run on every operation + again. A negative bound is worse still: the eviction loop would keep + popping past an already-empty dict. + """ + cache = _EntityMethodSignatureCache(max_size=max_size) + + cache[(int, "op")] = True + + assert len(cache) == 1 + assert cache.get((int, "op")) is True + assert (int, "op") in cache + + def test_clamped_cache_still_evicts_and_serves_hits(self): + """A clamped cache behaves as a working size-1 cache, not a no-op.""" + cache = _EntityMethodSignatureCache(max_size=0) + + cache[(int, "op")] = True + cache[(str, "op")] = False + + # The newest entry is retained and readable; the older one is evicted + # because the effective bound is one, not because caching is disabled. + assert len(cache) == 1 + assert cache.get((str, "op")) is False + assert cache.get((int, "op")) is None + + def test_executor_with_clamped_cache_still_caches_reflection(self, monkeypatch): + """An executor handed a clamped cache must still avoid re-reflecting.""" + counting = _CountingInspect() + monkeypatch.setattr(worker, "inspect", counting) + + registry = _Registry() + registry.add_entity(Counter) + executor = _EntityExecutor( + registry, + logging.getLogger("test"), + JsonDataConverter(), + entity_method_cache=_EntityMethodSignatureCache(max_size=0), + ) + + entity_id = entities.EntityInstanceId("Counter", "test-key") + state = StateShim(None, JsonDataConverter()) + for _ in range(3): + executor.execute("test-orch", entity_id, "add", state, "1") + state.commit() + + assert len(counting.signature_calls) == 1 + assert state.get_state(int) == 3