From ee53ef1f41cc7cba5b6b406c8e5ebbe56ad1dd76 Mon Sep 17 00:00:00 2001 From: Chad Retz Date: Fri, 4 Nov 2022 09:53:17 -0500 Subject: [PATCH 1/2] Clarify client thread safety Fixes #147 --- temporalio/client.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/temporalio/client.py b/temporalio/client.py index d65d3de72..89ba18372 100644 --- a/temporalio/client.py +++ b/temporalio/client.py @@ -65,6 +65,12 @@ class Client: :py:attr:`service` property provides access to a raw gRPC client. To create another client, like for a different namespace, :py:func:`Client` may be directly instantiated with a :py:attr:`service` of another. + + Clients are not thread-safe and should only be used in the event loop they + are first connected in. If a client needs to be used from another thread + than where it was created, make sure the event loop where it was created is + captured, and then call :py:func:`asyncio.run_coroutine_threadsafe` with the + client call and that event loop. """ @staticmethod From 0515439a3dd58dbbd7d715023df0d1809125e35d Mon Sep 17 00:00:00 2001 From: Chad Retz Date: Fri, 4 Nov 2022 10:11:29 -0500 Subject: [PATCH 2/2] Support for raising CancelledError out of sync activities --- README.md | 5 ++-- temporalio/worker/activity.py | 11 ++++++-- tests/worker/test_activity.py | 51 +++++++++++++++++++++++++++++++++++ 3 files changed, 63 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index da0b9343f..5fd40a4e4 100644 --- a/README.md +++ b/README.md @@ -782,8 +782,9 @@ Synchronous activities, i.e. functions that do not have `async def`, can be used activities. Cancellation for synchronous activities is done in the background and the activity must choose to listen for it and -react appropriately. An activity must heartbeat to receive cancellation and there are other ways to be notified about -cancellation (see "Activity Context" and "Heartbeating and Cancellation" later). +react appropriately. If after cancellation is obtained an unwrapped `temporalio.exceptions.CancelledError` is raised, +the activity will be marked cancelled. An activity must heartbeat to receive cancellation and there are other ways to be +notified about cancellation (see "Activity Context" and "Heartbeating and Cancellation" later). Note, all calls from an activity to functions in the `temporalio.activity` package are powered by [contextvars](https://docs.python.org/3/library/contextvars.html). Therefore, new threads starting _inside_ of diff --git a/temporalio/worker/activity.py b/temporalio/worker/activity.py index 37f56b220..52813946d 100644 --- a/temporalio/worker/activity.py +++ b/temporalio/worker/activity.py @@ -402,6 +402,7 @@ async def _run_activity( except ( Exception, asyncio.CancelledError, + temporalio.exceptions.CancelledError, temporalio.activity._CompleteAsyncError, ) as err: try: @@ -409,7 +410,10 @@ async def _run_activity( temporalio.activity.logger.debug("Completing asynchronously") completion.result.will_complete_async.SetInParent() elif ( - isinstance(err, asyncio.CancelledError) + isinstance( + err, + (asyncio.CancelledError, temporalio.exceptions.CancelledError), + ) and running_activity.cancelled_due_to_heartbeat_error ): err = running_activity.cancelled_due_to_heartbeat_error @@ -422,7 +426,10 @@ async def _run_activity( completion.result.failed.failure, ) elif ( - isinstance(err, asyncio.CancelledError) + isinstance( + err, + (asyncio.CancelledError, temporalio.exceptions.CancelledError), + ) and running_activity.cancelled_by_request ): temporalio.activity.logger.debug("Completing as cancelled") diff --git a/tests/worker/test_activity.py b/tests/worker/test_activity.py index a0a387512..67cea2831 100644 --- a/tests/worker/test_activity.py +++ b/tests/worker/test_activity.py @@ -283,6 +283,31 @@ def wait_cancel() -> str: assert result.result == "Cancelled" +async def test_sync_activity_thread_cancel_uncaught( + client: Client, worker: ExternalWorker +): + @activity.defn + def wait_cancel() -> str: + while not activity.is_cancelled(): + time.sleep(1) + activity.heartbeat() + raise CancelledError("Cancelled") + + with pytest.raises(WorkflowFailureError) as err: + with concurrent.futures.ThreadPoolExecutor() as executor: + await _execute_workflow_with_activity( + client, + worker, + wait_cancel, + cancel_after_ms=100, + wait_for_cancellation=True, + heartbeat_timeout_ms=3000, + worker_config={"activity_executor": executor}, + ) + assert isinstance(err.value.cause, ActivityError) + assert isinstance(err.value.cause.cause, CancelledError) + + @activity.defn def picklable_activity_wait_cancel() -> str: while not activity.is_cancelled(): @@ -305,6 +330,32 @@ async def test_sync_activity_process_cancel(client: Client, worker: ExternalWork assert result.result == "Cancelled" +@activity.defn +def picklable_activity_raise_cancel() -> str: + while not activity.is_cancelled(): + time.sleep(1) + activity.heartbeat() + raise CancelledError("Cancelled") + + +async def test_sync_activity_process_cancel_uncaught( + client: Client, worker: ExternalWorker +): + with pytest.raises(WorkflowFailureError) as err: + with concurrent.futures.ProcessPoolExecutor() as executor: + result = await _execute_workflow_with_activity( + client, + worker, + picklable_activity_raise_cancel, + cancel_after_ms=100, + wait_for_cancellation=True, + heartbeat_timeout_ms=3000, + worker_config={"activity_executor": executor}, + ) + assert isinstance(err.value.cause, ActivityError) + assert isinstance(err.value.cause.cause, CancelledError) + + async def test_activity_does_not_exist(client: Client, worker: ExternalWorker): @activity.defn async def say_hello(name: str) -> str: