Performance [P1]: Bound async worker task creation

[berndverst]

Summary

_AsyncWorkerManager._consume_queue() created an asyncio.Task for every dequeued work item and only acquired the concurrency semaphore later, inside _process_work_item(). The semaphore therefore throttled execution but not allocation: during a burst with slow handlers, the number of pending task objects grew with the queue depth rather than with the configured concurrency. The loop also rescanned the entire running_tasks set on every iteration.

Change

• The consumer now acquires a semaphore permit before dequeuing an item and hands that permit off to the task it creates, which releases it once the work item is done. The queue stays the bounded backlog and in-flight work item tasks never exceed the configured ConcurrencyOptions limits.
• Completed tasks are removed from running_tasks via task.add_done_callback(set.discard) instead of the O(n) sweep on each loop iteration.

Semantics are preserved:

• There is still no await between queue.get() returning and asyncio.create_task(), so an item can never be dropped by cancellation in that window (same atomicity as before).
• The permit is released if the dequeue times out, is cancelled, or raises — the loop cannot leak capacity.
• queue.task_done() is still called exactly once per item, before the permit is released, on both the success and failure paths (nested finally so a task_done() error still releases the permit, matching the old async with semaphore: unwinding order).
• Shutdown exit condition, cancellation behavior, exception types, and log messages are unchanged. No public API changed.

Verification

• Added test_async_worker_manager_bounds_task_allocation — submits 25 items with a concurrency limit of 2 and asserts peak allocated work item tasks stays <= 2 while peak running stays at the full limit of 2 (i.e. concurrency is still fully utilized) and all items complete. Confirmed this test fails on main (Expected at most 2 concurrently allocated work item tasks, got 25) and passes with the fix.
• Added test_async_worker_manager_calls_task_done_exactly_once_on_failure — wraps queue.task_done with a counter and asserts exactly one call per item on the failure path, with the cancellation callback invoked for every item.
• Core unit tests: pytest tests/durabletask -q --ignore-glob="*_e2e.py" → 683 passed, 7 skipped. (Baseline on this branch before the change: 680 passed, 1 failed — test_client.py::test_sync_client_recreate_cooldown_prevents_immediate_repeated_recreation, a pre-existing timing-sensitive flake unrelated to this change; it passes in the post-change run.)
• Provider unit tests: pytest tests/durabletask-azuremanaged/test_sandboxes_extension.py tests/durabletask-azuremanaged/test_durabletask_grpc_interceptor.py -q → 42 passed.
• Concurrency test files re-run 3x for flakiness → 11 passed each time.
• flake8 durabletask/worker.py tests/durabletask/test_worker_concurrency_loop_async.py → clean.
• pyright (strict, repo config) on durabletask/worker.py → 0 errors.

Fixes #193

[Copilot]

Pull request overview

This PR fixes a performance issue in the core durabletask async worker manager where asyncio.Task objects were being allocated per queued work item before concurrency throttling applied, causing task allocation (and overhead) to grow with queue depth during bursts. It changes the internal scheduling so task allocation is bounded by ConcurrencyOptions, and improves task-set cleanup efficiency.

Changes:

• Acquire the concurrency semaphore before dequeueing and task creation, and release the permit in the work-item task when finished.
• Replace per-iteration O(n) cleanup of running_tasks with add_done_callback(set.discard) for completed-task removal.
• Add regression tests validating bounded task allocation and exactly-once queue.task_done() behavior on failure, plus a changelog entry under ## Unreleased.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated no comments.

File	Description
durabletask/worker.py	Bounds allocated in-flight work item tasks to concurrency limits and removes completed tasks via done callbacks.
tests/durabletask/test_worker_concurrency_loop_async.py	Adds tests for bounded task allocation and exactly-once task_done() on failure paths.
CHANGELOG.md	Documents the user-visible fix under ## Unreleased → FIXED.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.

[berndverst]

Declining all four instances of the "Statement has no effect" alert on this file (comments 3654773407, 3654773416, 3659153978, 3659153987). The alert is a false positive, and both suggested fixes break the tests. Detail below, since two different fixes have now been proposed and both were verified rather than assumed.

What the name is actually bound to

The helper returns its readiness awaitable already called:

durabletask-python/tests/durabletask/test_worker_concurrency_loop_async.py

Lines 100 to 115 in f75b20e

	def _start_manager_and_wait_for_queues(manager):
	"""Start the manager loop and return an awaitable that resolves once its queues exist.
	The second element of the returned tuple is an already-created coroutine
	object (a readiness awaitable), not a callable -- await it directly.
	"""
	worker_task = asyncio.create_task(manager.run())
	async def wait_for_queues():
	for _ in range(100):
	if manager.activity_queue is not None:
	return
	await asyncio.sleep(0.01)
	raise RuntimeError("Worker manager queues were never initialized")
	return worker_task, wait_for_queues()

Note the trailing () on wait_for_queues(). The second tuple element is therefore an already-created coroutine object — not a callable, and not an asyncio.Event:

type         : coroutine
iscoroutine  : True
callable     : False
has .wait    : False

await <coroutine object> is the correct and only valid form here.

Both proposed fixes fail

suggestion	result
await queues_ready() (original)	TypeError: 'coroutine' object is not callable
await queues_ready.wait() (after rename)	AttributeError: 'coroutine' object has no attribute 'wait'

The first was confirmed twice — once with a standalone reproduction, and once by actually applying the suggestion to this file and running it, which failed at precisely the two flagged lines:

E   TypeError: 'coroutine' object is not callable
tests\durabletask\test_worker_concurrency_loop_async.py:164: TypeError
E   TypeError: 'coroutine' object is not callable
tests\durabletask\test_worker_concurrency_loop_async.py:208: TypeError
2 failed, 1 passed

The second suggestion presumes queues_ready is an asyncio.Event. It is not, so .wait() does not exist on it.

The underlying claim is false

The statement demonstrably does have an effect — awaiting the coroutine runs its body, and the side effects are observable:

the code as written:  await queues_ready
  -> OK, returned 'ready'
  side effects recorded during the await: ['queues became ready']

claim under review: "This statement has no effect."
  observed effects : 1
  claim is         : FALSE

Without that await, the queues are never drained and the test's central assertion — that in-flight work stays bounded by the concurrency limit — would be measuring nothing.

Note

The alert re-fired unchanged after the variable was renamed from wait_for_queues to queues_ready, on new lines, with a different suggested fix. That is the clearest evidence that the rule is matching the syntactic shape await NAME without resolving what NAME is bound to. A rename cannot change whether a statement has an effect.

What was taken from the feedback

The readability concern underneath the alert was fair, so it was addressed: the misleading local variables were renamed wait_for_queues → queues_ready, and the helper's docstring now states the contract explicitly — that the second element of the returned tuple is an already-created coroutine object, not a callable. The helper name and the inner async def wait_for_queues() were deliberately left alone, since those genuinely are callables and their names were already accurate.

No functional change is warranted here.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.