From f9536f3d5911d681b0fafb49da4a014bcdeed399 Mon Sep 17 00:00:00 2001 From: Bar Moshe <1barmoshe1@gmail.com> Date: Thu, 30 Jul 2026 13:46:01 +0300 Subject: [PATCH 1/2] Shut down the executor and manager in hello_activity_multiprocess The sample passed a freshly constructed ProcessPoolExecutor and multiprocessing.Manager straight into the Worker and never shut either down, so the sample was teaching a resource leak. Both are now entered as context managers, outside the worker so they outlive it. Co-Authored-By: Claude Opus 5 --- hello/hello_activity_multiprocess.py | 65 ++++++++++++++++------------ 1 file changed, 37 insertions(+), 28 deletions(-) diff --git a/hello/hello_activity_multiprocess.py b/hello/hello_activity_multiprocess.py index c7793c2c1..8c17094f0 100644 --- a/hello/hello_activity_multiprocess.py +++ b/hello/hello_activity_multiprocess.py @@ -48,36 +48,45 @@ async def main(): config.setdefault("target_host", "localhost:7233") client = await Client.connect(**config) - # Run a worker for the workflow - async with Worker( - client, - task_queue="hello-activity-multiprocess-task-queue", - workflows=[GreetingWorkflow], - activities=[compose_greeting], - # Synchronous activities are not allowed unless we provide some kind of - # executor. Here we are giving a process pool executor which means the - # activity will actually run in a separate process. This same executor - # could be passed to multiple workers if desired. - activity_executor=ProcessPoolExecutor(5), - # Since we are using an executor that is not a thread pool executor, - # Temporal needs some kind of manager to share state such as - # cancellation info and heartbeat info between the host and the - # activity. Therefore, we must provide a shared_state_manager here. A - # helper is provided to create it from a multiprocessing manager. - shared_state_manager=SharedStateManager.create_from_multiprocessing( - multiprocessing.Manager() - ), + # Synchronous activities are not allowed unless we provide some kind of + # executor. Here we are giving a process pool executor which means the + # activity will actually run in a separate process. This same executor could + # be passed to multiple workers if desired. + # + # Since we are using an executor that is not a thread pool executor, Temporal + # needs some kind of manager to share state such as cancellation info and + # heartbeat info between the host and the activity. Therefore, we must + # provide a shared_state_manager below. A helper is provided to create it + # from a multiprocessing manager. + # + # Both are used as context managers so their processes are shut down when we + # are done with them. They are entered outside the worker so that they + # outlive it. + with ( + ProcessPoolExecutor(5) as activity_executor, + multiprocessing.Manager() as manager, ): - # While the worker is running, use the client to run the workflow and - # print out its result. Note, in many production setups, the client - # would be in a completely separate process from the worker. - result = await client.execute_workflow( - GreetingWorkflow.run, - "World", - id="hello-activity-multiprocess-workflow-id", + # Run a worker for the workflow + async with Worker( + client, task_queue="hello-activity-multiprocess-task-queue", - ) - print(f"Result on PID {os.getpid()}: {result}") + workflows=[GreetingWorkflow], + activities=[compose_greeting], + activity_executor=activity_executor, + shared_state_manager=SharedStateManager.create_from_multiprocessing( + manager + ), + ): + # While the worker is running, use the client to run the workflow and + # print out its result. Note, in many production setups, the client + # would be in a completely separate process from the worker. + result = await client.execute_workflow( + GreetingWorkflow.run, + "World", + id="hello-activity-multiprocess-workflow-id", + task_queue="hello-activity-multiprocess-task-queue", + ) + print(f"Result on PID {os.getpid()}: {result}") if __name__ == "__main__": From 655081011bd7d24847b272c1b078b06691d0c34e Mon Sep 17 00:00:00 2001 From: Bar Moshe <1barmoshe1@gmail.com> Date: Thu, 30 Jul 2026 13:50:02 +0300 Subject: [PATCH 2/2] Own the heartbeat poller executor and fix teardown order Two follow-ups on the previous commit: - The manager is now entered first so it is torn down last. It owns the heartbeat queue, and the SDK's heartbeat poller can still be blocked reading that queue for up to its 0.5s poll timeout after the worker exits. - create_from_multiprocessing builds its own ThreadPoolExecutor when no queue poller executor is passed, and nothing ever shuts that one down. Pass an explicit one so it is shut down with the rest. Co-Authored-By: Claude Opus 5 --- hello/hello_activity_multiprocess.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/hello/hello_activity_multiprocess.py b/hello/hello_activity_multiprocess.py index 8c17094f0..934b653d5 100644 --- a/hello/hello_activity_multiprocess.py +++ b/hello/hello_activity_multiprocess.py @@ -2,7 +2,7 @@ import multiprocessing import os import time -from concurrent.futures import ProcessPoolExecutor +from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from dataclasses import dataclass from datetime import timedelta @@ -57,14 +57,18 @@ async def main(): # needs some kind of manager to share state such as cancellation info and # heartbeat info between the host and the activity. Therefore, we must # provide a shared_state_manager below. A helper is provided to create it - # from a multiprocessing manager. + # from a multiprocessing manager. That helper also needs a thread pool + # executor to poll the heartbeat queue on; it creates one itself if we do not + # pass one, so we pass our own to be able to shut it down. # - # Both are used as context managers so their processes are shut down when we - # are done with them. They are entered outside the worker so that they - # outlive it. + # All three are used as context managers so they are shut down when we are + # done with them. They are entered outside the worker so that they outlive + # it, and the manager is entered first so that it is torn down last: it owns + # the heartbeat queue that the other two are still using as they stop. with ( - ProcessPoolExecutor(5) as activity_executor, multiprocessing.Manager() as manager, + ThreadPoolExecutor(1) as queue_poller_executor, + ProcessPoolExecutor(5) as activity_executor, ): # Run a worker for the workflow async with Worker( @@ -74,7 +78,7 @@ async def main(): activities=[compose_greeting], activity_executor=activity_executor, shared_state_manager=SharedStateManager.create_from_multiprocessing( - manager + manager, queue_poller_executor ), ): # While the worker is running, use the client to run the workflow and