diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 5bea7251..30aeb53c 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -14,6 +14,11 @@ the synchronous client into synchronous functions and the asynchronous client into coroutine functions. Both clients support scheduled-task and history-export APIs without an async-to-sync bridge. +CHANGED + +- Reused host-driven orchestration and entity workers across invocations, +avoiding repeated allocation of unused worker resources. + ## 2.0.0b1 First preview (beta) release of `azure-functions-durable` 2.x — a ground-up diff --git a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py index 17a5664b..6747186b 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -233,6 +233,7 @@ def decorator(entity_func: task.Entity[Any, Any]) -> FunctionBuilder: if entity_name is not None: entity_func.__durable_entity_name__ = entity_name # type: ignore[union-attr] # Construct an orchestrator based on the end-user code + worker = DurableFunctionsWorker() # TODO: Because this handle method is the one actually exposed to the Functions SDK decorator, # the parameter name will always be "context" here, even if the user specified a different name. @@ -243,7 +244,7 @@ def decorator(entity_func: task.Entity[Any, Any]) -> FunctionBuilder: # binding converter to accept it; at runtime the host passes that # transport context (exposing ``.body``). def handle(context: func.EntityContext) -> str: - return DurableFunctionsWorker().execute_entity_batch_request(entity_func, context) + return worker.execute_entity_batch_request(entity_func, context) handle.entity_function = entity_func # pyright: ignore[reportFunctionMemberAccess] diff --git a/azure-functions-durable/azure/durable_functions/orchestrator.py b/azure-functions-durable/azure/durable_functions/orchestrator.py index 63d0b1b7..0e8e147c 100644 --- a/azure-functions-durable/azure/durable_functions/orchestrator.py +++ b/azure-functions-durable/azure/durable_functions/orchestrator.py @@ -20,7 +20,11 @@ class Orchestrator: function. """ - def __init__(self, orchestrator_func: Callable[..., Any]): + def __init__( + self, + orchestrator_func: Callable[..., Any], + worker: DurableFunctionsWorker | None = None, + ): """Create a new orchestrator wrapper for a user orchestrator function. The wrapped function may be a durabletask-native two-argument @@ -30,6 +34,7 @@ def __init__(self, orchestrator_func: Callable[..., Any]): :param orchestrator_func: The user's orchestrator function to run. """ self.fn: Callable[..., Any] = orchestrator_func + self._worker = worker if worker is not None else DurableFunctionsWorker() def handle(self, context: func.OrchestrationContext) -> str: """Handle the orchestration of the user defined generator function. @@ -49,7 +54,7 @@ def handle(self, context: func.OrchestrationContext) -> str: Durable Functions host wire format) produced by this invocation. """ self.durable_context = context - return DurableFunctionsWorker().execute_orchestration_request(self.fn, context) + return self._worker.execute_orchestration_request(self.fn, context) @classmethod def create(cls, fn: Callable[..., Any]) -> Callable[[Any], str]: @@ -67,13 +72,15 @@ def create(cls, fn: Callable[..., Any]) -> Callable[[Any], str]: Handle function of the newly created orchestration client """ + worker = DurableFunctionsWorker() + # The generated handle is the function registered with the Azure # Functions host. Its ``context`` parameter must be annotated with # ``azure.functions.OrchestrationContext`` so the host's # orchestrationTrigger binding converter accepts it; at runtime the # host passes that transport context (exposing ``.body``). def handle(context: func.OrchestrationContext) -> str: - return Orchestrator(fn).handle(context) + return Orchestrator(fn, worker).handle(context) handle.orchestrator_function = fn # pyright: ignore[reportFunctionMemberAccess] diff --git a/azure-functions-durable/azure/durable_functions/worker.py b/azure-functions-durable/azure/durable_functions/worker.py index 8f2bd727..a27a0eaf 100644 --- a/azure-functions-durable/azure/durable_functions/worker.py +++ b/azure-functions-durable/azure/durable_functions/worker.py @@ -2,6 +2,7 @@ # Licensed under the MIT License. import base64 +from threading import Lock from typing import Any, Optional from durabletask import task @@ -41,10 +42,40 @@ def __init__(self) -> None: # azure-functions codec (df_dumps/df_loads) so user types round-trip in # the wire format the Durable Functions host extension expects. super().__init__(data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER) + self._registration_lock = Lock() + self._registered_orchestrator_functions: dict[ + str, task.Orchestrator[Any, Any] + ] = {} + self._registered_entity_functions: dict[ + str, task.Entity[Any, Any] + ] = {} def add_named_orchestrator(self, name: str, func: task.Orchestrator[Any, Any]) -> None: self._registry.add_named_orchestrator(name, func) + def _register_orchestrator_once( + self, + name: str, + func: task.Orchestrator[Any, Any], + ) -> None: + with self._registration_lock: + if self._registered_orchestrator_functions.get(name) is func: + return + if self._registry.get_orchestrator(name) is not None: + raise ValueError(f"A '{name}' orchestrator already exists.") + self.add_named_orchestrator(name, wrap_orchestrator(func)) + self._registered_orchestrator_functions[name] = func + + def _register_entity_once(self, func: task.Entity[Any, Any]) -> None: + name = task.get_entity_name(func).lower() + with self._registration_lock: + if self._registered_entity_functions.get(name) is func: + return + if self._registry.get_entity(name) is not None: + raise ValueError(f"A '{name}' entity already exists.") + self.add_entity(wrap_entity(func), name=name) + self._registered_entity_functions[name] = func + def execute_orchestration_request(self, func: task.Orchestrator[Any, Any], context: Any) -> str: context_body = getattr(context, "body", None) if context_body is None: @@ -70,7 +101,7 @@ def stub_complete(stub_response: OrchestratorResponse) -> None: raise ValueError("No ExecutionStarted event found in orchestration request.") function_name = execution_started_events[-1].executionStarted.name - self.add_named_orchestrator(function_name, wrap_orchestrator(func)) + self._register_orchestrator_once(function_name, func) super()._execute_orchestrator(request, stub, None) if response is None: @@ -94,7 +125,7 @@ def stub_complete(stub_response: EntityBatchResult) -> None: response = stub_response stub.CompleteEntityTask = stub_complete - self.add_entity(wrap_entity(func)) + self._register_entity_once(func) super()._execute_entity_batch(request, stub, None) if response is None: diff --git a/tests/azure-functions-durable/test_decorator_compat.py b/tests/azure-functions-durable/test_decorator_compat.py index 898a277c..f0f1ab2c 100644 --- a/tests/azure-functions-durable/test_decorator_compat.py +++ b/tests/azure-functions-durable/test_decorator_compat.py @@ -4,6 +4,7 @@ import azure.durable_functions as df import inspect import pytest +from unittest.mock import MagicMock, patch from azure.durable_functions.constants import ( ACTIVITY_TRIGGER, DURABLE_CLIENT, @@ -104,6 +105,32 @@ def my_entity(context): assert trigger.entity_name == "MyEntity" +def test_entity_trigger_reuses_worker_across_invocations(): + app = df.DFApp() + + def my_entity(context): + return None + + with patch( + "azure.durable_functions.decorators.durable_app.DurableFunctionsWorker" + ) as worker_cls: + worker = worker_cls.return_value + worker.execute_entity_batch_request.return_value = "encoded" + fb = app.entity_trigger(context_name="context")(my_entity) + handle = fb._function._func + first_context = MagicMock() + second_context = MagicMock() + first_result = handle(first_context) + second_result = handle(second_context) + + assert first_result == second_result == "encoded" + worker_cls.assert_called_once_with() + assert worker.execute_entity_batch_request.call_args_list == [ + ((my_entity, first_context),), + ((my_entity, second_context),), + ] + + # --------------------------------------------------------------------------- # durable_client_input # --------------------------------------------------------------------------- diff --git a/tests/azure-functions-durable/test_orchestrator_compat.py b/tests/azure-functions-durable/test_orchestrator_compat.py index a3bc136b..252cda0f 100644 --- a/tests/azure-functions-durable/test_orchestrator_compat.py +++ b/tests/azure-functions-durable/test_orchestrator_compat.py @@ -5,7 +5,7 @@ ``Orchestrator`` is the thin adapter registered with the Azure Functions host: its generated ``handle`` receives the host's transport context and delegates to -a fresh :class:`DurableFunctionsWorker` per invocation. +a shared :class:`DurableFunctionsWorker`. """ from unittest.mock import MagicMock, patch @@ -21,11 +21,12 @@ def user_orchestrator(context): with patch( "azure.durable_functions.orchestrator.DurableFunctionsWorker" ) as worker_cls: - worker_cls.return_value.execute_orchestration_request.return_value = "encoded" + worker = worker_cls.return_value + worker.execute_orchestration_request.return_value = "encoded" result = Orchestrator(user_orchestrator).handle(context) assert result == "encoded" - worker_cls.return_value.execute_orchestration_request.assert_called_once_with( + worker.execute_orchestration_request.assert_called_once_with( user_orchestrator, context) @@ -34,8 +35,8 @@ def user_orchestrator(context): return None context = MagicMock() - orchestrator = Orchestrator(user_orchestrator) with patch("azure.durable_functions.orchestrator.DurableFunctionsWorker"): + orchestrator = Orchestrator(user_orchestrator) orchestrator.handle(context) assert orchestrator.durable_context is context @@ -53,14 +54,20 @@ def test_created_handle_delegates_to_worker(): def user_orchestrator(context): return None - handle = Orchestrator.create(user_orchestrator) - context = MagicMock() with patch( "azure.durable_functions.orchestrator.DurableFunctionsWorker" ) as worker_cls: - worker_cls.return_value.execute_orchestration_request.return_value = "encoded" - result = handle(context) - - assert result == "encoded" - worker_cls.return_value.execute_orchestration_request.assert_called_once_with( - user_orchestrator, context) + worker = worker_cls.return_value + worker.execute_orchestration_request.return_value = "encoded" + handle = Orchestrator.create(user_orchestrator) + first_context = MagicMock() + second_context = MagicMock() + first_result = handle(first_context) + second_result = handle(second_context) + + assert first_result == second_result == "encoded" + worker_cls.assert_called_once_with() + assert worker.execute_orchestration_request.call_args_list == [ + ((user_orchestrator, first_context),), + ((user_orchestrator, second_context),), + ] diff --git a/tests/azure-functions-durable/test_worker_compat.py b/tests/azure-functions-durable/test_worker_compat.py index 227c1ef9..dc15a507 100644 --- a/tests/azure-functions-durable/test_worker_compat.py +++ b/tests/azure-functions-durable/test_worker_compat.py @@ -12,6 +12,7 @@ import base64 import json +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace import pytest @@ -131,6 +132,40 @@ def orchestrator(context): assert "boom" in completion.failureDetails.errorMessage +def test_execute_orchestration_request_supports_concurrent_reinvocation(): + def orchestrator(context): + return context.instance_id + + worker = DurableFunctionsWorker() + encoded = _encode_orchestrator_request("concurrent-orch") + with ThreadPoolExecutor(max_workers=4) as executor: + results = list(executor.map( + lambda _: worker.execute_orchestration_request(orchestrator, encoded), + range(8), + )) + + assert all( + json.loads(_get_completion_action( + _decode_orchestrator_response(result)).result.value) == TEST_INSTANCE_ID + for result in results + ) + + +def test_execute_orchestration_request_rejects_different_function_with_same_name(): + def first_orchestrator(context): + return "first" + + def second_orchestrator(context): + return "second" + + worker = DurableFunctionsWorker() + encoded = _encode_orchestrator_request("same-name") + worker.execute_orchestration_request(first_orchestrator, encoded) + + with pytest.raises(ValueError, match="A 'same-name' orchestrator already exists"): + worker.execute_orchestration_request(second_orchestrator, encoded) + + # --------------------------------------------------------------------------- # execute_entity_batch_request # --------------------------------------------------------------------------- @@ -197,3 +232,40 @@ def entity(context): response = _decode_entity_response(result) assert response.results[0].HasField("failure") assert "entity failed" in response.results[0].failure.failureDetails.errorMessage + + +def test_execute_entity_batch_request_supports_concurrent_reinvocation(): + def entity(context): + context.set_result("handled") + + entity.__durable_entity_name__ = "Counter" + worker = DurableFunctionsWorker() + encoded = _encode_entity_batch_request("@counter@key1", "op") + with ThreadPoolExecutor(max_workers=4) as executor: + results = list(executor.map( + lambda _: worker.execute_entity_batch_request(entity, encoded), + range(8), + )) + + assert all( + json.loads(_decode_entity_response( + result).results[0].success.result.value) == "handled" + for result in results + ) + + +def test_execute_entity_batch_request_rejects_different_function_with_same_name(): + def first_entity(context): + context.set_result("first") + + def second_entity(context): + context.set_result("second") + + first_entity.__durable_entity_name__ = "Counter" + second_entity.__durable_entity_name__ = "Counter" + worker = DurableFunctionsWorker() + encoded = _encode_entity_batch_request("@counter@key1", "op") + worker.execute_entity_batch_request(first_entity, encoded) + + with pytest.raises(ValueError, match="A 'counter' entity already exists"): + worker.execute_entity_batch_request(second_entity, encoded)