[Backport release-1.19] workflow: run orchestrator work items on their own thread pool

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.
 

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,40 +2685,47 @@ 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)
         else:
             # 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."""

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.

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()