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 12e3b4c0e..a9f5b3b0d 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 @@ -2520,6 +2522,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 @@ -2536,6 +2542,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 @@ -2616,11 +2629,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() @@ -2648,7 +2670,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 @@ -2663,31 +2685,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) @@ -2695,8 +2724,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) @@ -2711,6 +2740,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()