concurrency

Author: berndverst

CHANGELOG.md

@@ -5,11 +5,22 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## v0.4.0
+
+### New
+
+- Added `ConcurrencyOptions` class for fine-grained concurrency control equivalent to .NET DurableTask SDK
+- Enhanced `TaskHubGrpcWorker` with `concurrency_options` parameter supporting separate limits for activities, orchestrations, and entities (100 * processor_count default each)
+
+### Breaking Changes
+
+- Removed `max_workers` parameter from `TaskHubGrpcWorker` constructor in favor of the more flexible `ConcurrencyOptions` approach
+
 ## v0.3.0
 
 ### New
 
-- Added configurable worker concurrency with `max_workers` parameter in `TaskHubGrpcWorker` constructor - allows customization of ThreadPoolExecutor size (default: 16 workers)
+- Added configurable worker concurrency with `max_workers` parameter in `TaskHubGrpcWorker` constructor - allows customization of ThreadPoolExecutor size (default: 16 workers) (NOTE: This parameter was later removed in v0.4.0 in favor of `ConcurrencyOptions`)
 
 ### Fixed
 

durabletask/__init__.py

@@ -3,5 +3,8 @@
 
 """Durable Task SDK for Python"""
 
+from durabletask.worker import ConcurrencyOptions
+
+__all__ = ["ConcurrencyOptions"]
 
 PACKAGE_NAME = "durabletask"

durabletask/worker.py

@@ -3,6 +3,7 @@
 
 import concurrent.futures
 import logging
+import os
 import random
 from datetime import datetime, timedelta
 from threading import Event, Thread
@@ -24,6 +25,53 @@
 TOutput = TypeVar('TOutput')
 
 
+class ConcurrencyOptions:
+    """Configuration options for controlling concurrency of different work item types.
+
+    This class mirrors the .NET DurableTask SDK's ConcurrencyOptions class,
+    providing fine-grained control over concurrent processing limits for
+    activities, orchestrations, and entities.
+    """
+
+    def __init__(self,
+                 maximum_concurrent_activity_work_items: Optional[int] = None,
+                 maximum_concurrent_orchestration_work_items: Optional[int] = None):
+        """Initialize concurrency options.
+
+        Args:
+            maximum_concurrent_activity_work_items: Maximum number of activity work items
+                that can be processed concurrently. Defaults to 100 * processor_count.
+            maximum_concurrent_orchestration_work_items: Maximum number of orchestration work items
+                that can be processed concurrently. Defaults to 100 * processor_count.
+        """
+        processor_count = os.cpu_count() or 1
+        default_concurrency = 100 * processor_count
+
+        self.maximum_concurrent_activity_work_items = (
+            maximum_concurrent_activity_work_items
+            if maximum_concurrent_activity_work_items is not None
+            else default_concurrency
+        )
+
+        self.maximum_concurrent_orchestration_work_items = (
+            maximum_concurrent_orchestration_work_items
+            if maximum_concurrent_orchestration_work_items is not None
+            else default_concurrency
+        )
+
+    @property
+    def max_total_workers(self) -> int:
+        """Calculate the maximum total workers needed for the thread pool.
+
+        Since Python's ThreadPoolExecutor doesn't differentiate between work item types,
+        we use the maximum of all concurrency limits to ensure we have enough workers.
+        """
+        return max(
+            self.maximum_concurrent_activity_work_items,
+            self.maximum_concurrent_orchestration_work_items,
+        )
+
+
 class _Registry:
 
     orchestrators: dict[str, task.Orchestrator]
@@ -93,14 +141,16 @@ def __init__(self, *,
                  log_formatter: Optional[logging.Formatter] = None,
                  secure_channel: bool = False,
                  interceptors: Optional[Sequence[shared.ClientInterceptor]] = None,
-                 max_workers: Optional[int] = None):
+                 concurrency_options: Optional[ConcurrencyOptions] = None):
         self._registry = _Registry()
         self._host_address = host_address if host_address else shared.get_default_host_address()
         self._logger = shared.get_logger("worker", log_handler, log_formatter)
         self._shutdown = Event()
         self._is_running = False
         self._secure_channel = secure_channel
-        self._max_workers = max_workers if max_workers is not None else 16
+
+        # Use provided concurrency options or create default ones
+        self._concurrency_options = concurrency_options if concurrency_options is not None else ConcurrencyOptions()
 
         # Determine the interceptors to use
         if interceptors is not None:
@@ -112,6 +162,11 @@ def __init__(self, *,
         else:
             self._interceptors = None
 
+    @property
+    def concurrency_options(self) -> ConcurrencyOptions:
+        """Get the current concurrency options for this worker."""
+        return self._concurrency_options
+
     def __enter__(self):
         return self
 
@@ -211,7 +266,7 @@ def should_invalidate_connection(rpc_error: grpc.RpcError) -> bool:
 
             # TODO: Investigate whether asyncio could be used to enable greater concurrency for async activity
             #       functions. We'd need to know ahead of time whether a function is async or not.
-            with concurrent.futures.ThreadPoolExecutor(max_workers=self._max_workers, thread_name_prefix="DurableTask") as executor:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=self._concurrency_options.max_total_workers, thread_name_prefix="DurableTask") as executor:
                 while not self._shutdown.is_set():
                     # Ensure we have a valid connection before attempting work
                     if current_stub is None:
@@ -231,7 +286,13 @@ def should_invalidate_connection(rpc_error: grpc.RpcError) -> bool:
                         # Type assertion since we know current_stub is not None at this point
                         assert current_stub is not None, "current_stub should not be None at this point"
                         stub = current_stub  # Local reference for type safety
-                        self._response_stream = stub.GetWorkItems(pb.GetWorkItemsRequest())
+
+                        # Create GetWorkItemsRequest with concurrency limits
+                        get_work_items_request = pb.GetWorkItemsRequest(
+                            maxConcurrentOrchestrationWorkItems=self._concurrency_options.maximum_concurrent_orchestration_work_items,
+                            maxConcurrentActivityWorkItems=self._concurrency_options.maximum_concurrent_activity_work_items
+                        )
+                        self._response_stream = stub.GetWorkItems(get_work_items_request)
                         self._logger.info(f'Successfully connected to {self._host_address}. Waiting for work items...')
 
                         # Process work items concurrently as they arrive

examples/README.md

@@ -24,4 +24,4 @@ In some cases, the sample may require command-line parameters or user inputs. In
 
 - [Activity sequence](./activity_sequence.py): Orchestration that schedules three activity calls in a sequence.
 - [Fan-out/fan-in](./fanout_fanin.py): Orchestration that schedules a dynamic number of activity calls in parallel, waits for all of them to complete, and then performs an aggregation on the results.
-- [Human interaction](./human_interaction.py): Orchestration that waits for a human to approve an order before continuing.
+- [Human interaction](./human_interaction.py): Orchestration that waits for a human to approve an order before continuing.