From 1e7bfd119c2f5698f202bd83ce2d54ea7139cc7a Mon Sep 17 00:00:00 2001 From: Josh van Leeuwen Date: Tue, 13 Oct 2026 14:00:06 +0000 Subject: [PATCH] workflow: run orchestrator work items on their own thread pool (#1264) * workflow: run orchestrator work items on their own thread pool The worker ran sync activities and orchestrator work items on one thread pool. An activity that schedules a workflow waits until that workflow has started, which needs the new workflow's first orchestrator work item to run on the same worker. With enough such activities in flight to take every pool thread, the orchestrator work items queued behind them and the worker deadlocked. Give orchestrator work items their own pool of the same size. Signed-off-by: joshvanl * Review comments Signed-off-by: joshvanl --------- Signed-off-by: joshvanl (cherry picked from commit 6d8fb6e71006dd2557e8f2edb301a6d711df176d) Signed-off-by: dapr-bot --- dapr/ext/workflow/AGENTS.md | 2 +- dapr/ext/workflow/_durabletask/worker.py | 74 +++++++++++++------ dapr/ext/workflow/docs/concurrency.md | 10 ++- .../test_worker_orchestration_starvation.py | 52 +++++++++++++ 4 files changed, 113 insertions(+), 25 deletions(-) create mode 100644 tests/ext/workflow/durabletask/test_worker_orchestration_starvation.py diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index 0a58a89fa..e2606e8e9 100644 --- a/dapr/ext/workflow/AGENTS.md +++ b/dapr/ext/workflow/AGENTS.md @@ -127,7 +127,7 @@ Workflow (orchestrator) functions must remain generators (`def` with `yield`). T **Decorator ordering gotcha.** Wrapping `@wfr.activity` over `@alternate_name(...)` over `async def` works because `@alternate_name` now emits an `async def innerfn` when the wrapped function is async. A user-written decorator that wraps an async function in a sync `def` (without `@functools.wraps` exposing `__wrapped__`) defeats `_is_async_callable`, routes the activity to the sync path, and produces an un-awaited coroutine. Such decorators should use `@functools.wraps(fn)` so the unwrap walks through them. -**`maximum_thread_pool_workers` covers both paths.** This knob sizes the worker thread pool used for sync-activity bodies and for async-activity gRPC response sends. Mixed workloads with long-running sync activities can starve async response delivery (and vice versa) since they share the pool — size to the sum of peak sync activity concurrency and peak in-flight async response sends. +**`maximum_thread_pool_workers` covers both paths.** This knob sizes the worker thread pool used for sync-activity bodies and for async-activity gRPC response sends. Mixed workloads with long-running sync activities can starve async response delivery (and vice versa) since they share the pool — size to the sum of peak sync activity concurrency and peak in-flight async response sends. Orchestrator work items run on a separate pool of the same size, so activities that block (for example one that schedules a workflow and waits for it to start, which needs that workflow's first orchestrator work item to run on this worker) cannot starve them. **Concurrency sizing and load characterization.** See `docs/concurrency.md` for sizing recommendations (`maximum_concurrent_activity_work_items`, `maximum_thread_pool_workers`) and an async-vs-sync decision tree. `tests/ext/workflow/durabletask/test_async_dispatch_regression.py` (marked `perf`) guards the core invariant: a batch of async activities overlaps on the event loop instead of serializing through the thread pool. diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py index 090eb747a..c1fd3ccae 100644 --- a/dapr/ext/workflow/_durabletask/worker.py +++ b/dapr/ext/workflow/_durabletask/worker.py @@ -71,7 +71,8 @@ class ConcurrencyOptions: """Concurrency limits for the worker. ``maximum_thread_pool_workers`` sizes the pool used to run sync activities and to - deliver async-activity responses to the sidecar. + deliver async-activity responses to the sidecar. Orchestrator work items run on a + second pool of the same size. """ def __init__( @@ -87,9 +88,10 @@ def __init__( Defaults to ``100 * cpu_count``. maximum_concurrent_orchestration_work_items: Cap on concurrent orchestration work items. Defaults to ``100 * cpu_count``. - maximum_thread_pool_workers: Size of the worker thread pool. Sync activities run - on this pool, and async-activity gRPC response sends also borrow a thread - from it. Defaults to ``cpu_count + 4``. + maximum_thread_pool_workers: Size of each of the worker's two thread pools. Sync + activities and async-activity gRPC response sends share one; orchestrator + work items run on the other, so a worker can start up to twice this many + threads. Defaults to ``cpu_count + 4``. """ processor_count = os.cpu_count() or 1 default_concurrency = 100 * processor_count @@ -2518,6 +2520,10 @@ def _is_suspendable(event: pb.HistoryEvent) -> bool: ] +# A queued work item: the callable and the arguments it is called with. +_WorkItem = tuple[Callable[..., Any], tuple[Any, ...], dict[str, Any]] + + class _AsyncWorkerManager: def __init__(self, concurrency_options: ConcurrencyOptions, logger: logging.Logger): self.concurrency_options = concurrency_options @@ -2534,6 +2540,13 @@ def __init__(self, concurrency_options: ConcurrencyOptions, logger: logging.Logg max_workers=concurrency_options.maximum_thread_pool_workers, thread_name_prefix='DurableTask', ) + # Orchestrations get their own pool: an activity that schedules a workflow + # blocks until that workflow's first orchestration work item has run here, + # so sharing the activity pool deadlocks once activities hold every thread. + self.orchestration_thread_pool = ThreadPoolExecutor( + max_workers=concurrency_options.maximum_thread_pool_workers, + thread_name_prefix='DurableTask-Orchestration', + ) self._shutdown = False self._logger = logger @@ -2614,11 +2627,20 @@ async def run(self): # Start background consumers for each work type if self.activity_queue is not None and self.orchestration_queue is not None: await asyncio.gather( - self._consume_queue(self.activity_queue, self.activity_semaphore), - self._consume_queue(self.orchestration_queue, self.orchestration_semaphore), + self._consume_queue(self.activity_queue, self.activity_semaphore, self.thread_pool), + self._consume_queue( + self.orchestration_queue, + self.orchestration_semaphore, + self.orchestration_thread_pool, + ), ) - async def _consume_queue(self, queue: asyncio.Queue, semaphore: asyncio.Semaphore): + async def _consume_queue( + self, + queue: asyncio.Queue[_WorkItem], + semaphore: asyncio.Semaphore, + executor: ThreadPoolExecutor, + ) -> None: # List to track running tasks running_tasks: set[asyncio.Task] = set() @@ -2646,7 +2668,7 @@ async def _consume_queue(self, queue: asyncio.Queue, semaphore: asyncio.Semaphor func, args, kwargs = work # Create a concurrent task for processing task = asyncio.create_task( - self._process_work_item(semaphore, queue, func, args, kwargs) + self._process_work_item(semaphore, queue, executor, func, args, kwargs) ) running_tasks.add(task) # handle the cancellation bubbled up from the loop @@ -2661,31 +2683,38 @@ async def _consume_queue(self, queue: asyncio.Queue, semaphore: asyncio.Semaphor raise async def _process_work_item( - self, semaphore: asyncio.Semaphore, queue: asyncio.Queue, func, args, kwargs - ): + self, + semaphore: asyncio.Semaphore, + queue: asyncio.Queue[_WorkItem], + executor: ThreadPoolExecutor, + func: Callable[..., Any], + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: async with semaphore: try: - await self._run_func(func, *args, **kwargs) + await self._run_func_on(executor, func, *args, **kwargs) finally: queue.task_done() - async def _run_func(self, func, *args, **kwargs): + async def _run_func(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + return await self._run_func_on(self.thread_pool, func, *args, **kwargs) + + async def _run_func_on( + self, executor: ThreadPoolExecutor, func: Callable[..., Any], *args: Any, **kwargs: Any + ) -> Any: if is_async_callable(func): return await func(*args, **kwargs) else: loop = asyncio.get_running_loop() # Avoid submitting to executor after shutdown - if ( - getattr(self, '_shutdown', False) - and getattr(self, 'thread_pool', None) - and getattr(self.thread_pool, '_shutdown', False) - ): + if getattr(self, '_shutdown', False) and getattr(executor, '_shutdown', False): return None - result = await loop.run_in_executor(self.thread_pool, lambda: func(*args, **kwargs)) + result = await loop.run_in_executor(executor, lambda: func(*args, **kwargs)) return result - def submit_activity(self, func, *args, **kwargs): - work_item = (func, args, kwargs) + def submit_activity(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None: + work_item: _WorkItem = (func, args, kwargs) self._ensure_queues_for_current_loop() if self.activity_queue is not None: self.activity_queue.put_nowait(work_item) @@ -2693,8 +2722,8 @@ def submit_activity(self, func, *args, **kwargs): # No event loop running, store in pending list self._pending_activity_work.append(work_item) - def submit_orchestration(self, func, *args, **kwargs): - work_item = (func, args, kwargs) + def submit_orchestration(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> None: + work_item: _WorkItem = (func, args, kwargs) self._ensure_queues_for_current_loop() if self.orchestration_queue is not None: self.orchestration_queue.put_nowait(work_item) @@ -2709,6 +2738,7 @@ def shutdown(self): # ThreadPoolExecutor.shutdown(wait=True) doesn't support a timeout, but with proper # cancellation in place, threads should exit promptly, otherwise this will hang and block shutdown for the application. self.thread_pool.shutdown(wait=True) + self.orchestration_thread_pool.shutdown(wait=True) def reset_for_new_run(self): """Reset the manager state for a new run.""" diff --git a/dapr/ext/workflow/docs/concurrency.md b/dapr/ext/workflow/docs/concurrency.md index 50353dc64..666fd2a91 100644 --- a/dapr/ext/workflow/docs/concurrency.md +++ b/dapr/ext/workflow/docs/concurrency.md @@ -8,7 +8,7 @@ Sizing notes for the worker's concurrency knobs. | --- | --- | --- | | `maximum_concurrent_activity_work_items` | `100 × cpu_count` | Async semaphore cap on in-flight activity work items. | | `maximum_concurrent_orchestration_work_items` | `100 × cpu_count` | Same, for orchestrations. | -| `maximum_thread_pool_workers` | `cpu_count + 4` | Worker thread pool size. Sync activities run on this pool, and async-activity gRPC response sends also borrow a thread from it. | +| `maximum_thread_pool_workers` | `cpu_count + 4` | Size of each of the worker's two thread pools. Sync activities and async-activity gRPC response sends share one; orchestrator work items run on the other. | A `def` activity consumes a semaphore slot **and** a thread pool worker. An `async def` activity consumes only a semaphore slot. @@ -46,7 +46,13 @@ backpressure. ## Sizing the thread pool -The worker thread pool, sized by `maximum_thread_pool_workers`, has two uses. +`maximum_thread_pool_workers` sizes two pools: the activity pool described below, +and a separate pool for orchestrator work items, so blocked activities cannot starve +orchestrations. A worker can therefore start up to `2 × maximum_thread_pool_workers` +threads (`2 × (cpu_count + 4)` by default); count both when sizing for memory or a +thread limit. + +The activity pool has two uses. **Sync activity execution.** Each `def` activity holds one thread for its duration. Size to peak concurrent sync-activity count. diff --git a/tests/ext/workflow/durabletask/test_worker_orchestration_starvation.py b/tests/ext/workflow/durabletask/test_worker_orchestration_starvation.py new file mode 100644 index 000000000..d130a9736 --- /dev/null +++ b/tests/ext/workflow/durabletask/test_worker_orchestration_starvation.py @@ -0,0 +1,52 @@ +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""An activity that schedules a workflow waits until the sidecar reports that +workflow started, which needs the new workflow's orchestration work item to run +on this worker. Activities holding every pool thread must not stop it.""" + +import asyncio +import logging +import threading + +from dapr.ext.workflow._durabletask.worker import ConcurrencyOptions, _AsyncWorkerManager + + +def test_orchestration_runs_while_activities_hold_every_pool_thread() -> None: + manager = _AsyncWorkerManager( + ConcurrencyOptions(maximum_thread_pool_workers=2), logger=logging.getLogger() + ) + child_started = threading.Event() + activities_running = threading.Barrier(3) + + def activity_waiting_for_child() -> None: + activities_running.wait(timeout=5) + child_started.wait(timeout=5) + + def child_orchestration() -> None: + child_started.set() + + async def scenario() -> bool: + runner = asyncio.create_task(manager.run()) + manager.submit_activity(activity_waiting_for_child) + manager.submit_activity(activity_waiting_for_child) + # Both pool threads are held before the orchestration is submitted. + await asyncio.to_thread(activities_running.wait, 5) + manager.submit_orchestration(child_orchestration) + started = await asyncio.to_thread(child_started.wait, 3) + manager._shutdown = True + await runner + return started + + try: + assert asyncio.run(scenario()), 'orchestration starved by blocked activities' + finally: + manager.shutdown()