Merge pull request #1 from firdaus/master

take rebase from main repo

Author: meetchandan

README.md

@@ -42,21 +42,22 @@ This should be considered EXPERIMENTAL at the moment. At the moment, all I can s
 - [x]  Workflow client - invoking queries
 
 1.1
-- [ ] ActivityStub and Workflow.newUntypedActivityStub
-- [ ] Classes as arguments and return values to/from activity and workflow methods
+- [x] ActivityStub and Workflow.newUntypedActivityStub
+- [ ] Classes as arguments and return values to/from activity and workflow methods (DataConverter)
 - [ ] WorkflowStub and WorkflowClient.newUntypedWorkflowStub
 - [ ] Custom workflow ids through start() and new_workflow_stub()
 - [ ] ContinueAsNew
+- [ ] Parallel activity execution (STATUS: there's a working but not finalized API).
 - [ ] Compatibility with Java client
 - [ ] Compatibility with Golang client
+- [ ] Remove threading, use coroutines for everything all concurrency 
 
 2.0
 - [ ] Sticky workflows
 
 Post 2.0:
 - [ ] sideEffect/mutableSideEffect
 - [ ] Local activity
-- [ ] Parallel activity execution
 - [ ] Timers
 - [ ] Cancellation Scopes
 - [ ] Child Workflows

temporal/activity.py

@@ -144,7 +144,6 @@ async def complete_exceptionally(service, task_token, ex: Exception):
     respond.task_token = task_token
     respond.identity = get_identity()
     respond.failure = serialize_exception(ex)
-    # TODO: error handling
     await service.respond_activity_task_failed(request=respond)
 
 
@@ -153,5 +152,4 @@ async def complete(service, task_token, return_value: object):
     respond.task_token = task_token
     respond.result = to_payloads([return_value])
     respond.identity = get_identity()
-    # TODO: error handling
     await service.respond_activity_task_completed(request=respond)

temporal/activity_loop.py

@@ -5,9 +5,12 @@
 import inspect
 from typing import List
 
+from grpclib import GRPCError
+
 from temporal.activity import ActivityContext, ActivityTask, complete_exceptionally, complete
 from temporal.api.taskqueue.v1 import TaskQueue, TaskQueueMetadata
 from temporal.conversions import from_payloads
+from temporal.retry import retry
 from temporal.service_helpers import create_workflow_service, get_identity
 from temporal.worker import Worker, StopRequestedException
 from temporal.api.workflowservice.v1 import WorkflowServiceStub as WorkflowService, PollActivityTaskQueueRequest, \
@@ -20,6 +23,7 @@ def activity_task_loop(worker: Worker):
     asyncio.run(activity_task_loop_func(worker))
 
 
+@retry(logger=logger)
 async def activity_task_loop_func(worker: Worker):
     service: WorkflowService = create_workflow_service(worker.host, worker.port, timeout=worker.get_timeout())
     worker.manage_service(service)
@@ -43,14 +47,9 @@ async def activity_task_loop_func(worker: Worker):
                 logger.debug("PollActivityTaskQueue: %dms", (polling_end - polling_start).total_seconds() * 1000)
             except StopRequestedException:
                 return
-            except Exception as ex:
-                logger.error("PollActivityTaskQueue error: %s", ex)
+            except GRPCError as ex:
+                logger.error("Error invoking poll_activity_task_queue: %s", ex, exc_info=True)
                 continue
-            # -----
-            # if err:
-            #     logger.error("PollActivityTaskQueue failed: %s", err)
-            #     continue
-            # -----
             task_token = task.task_token
             if not task_token:
                 logger.debug("PollActivityTaskQueue has no task_token (expected): %s", task)
@@ -78,20 +77,20 @@ async def activity_task_loop_func(worker: Worker):
                 if activity_context.do_not_complete:
                     logger.info(f"Not completing activity {task.activity_type.name}({str(args)[1:-1]})")
                     continue
-                await complete(service, task_token, return_value)
-                # -----
-                # if error:
-                #     logger.error("Error invoking RespondActivityTaskCompleted: %s", error)
-                # -----
+
                 logger.info(
                     f"Activity {task.activity_type.name}({str(args)[1:-1]}) returned {json.dumps(return_value)}")
+
+                try:
+                    await complete(service, task_token, return_value)
+                except GRPCError as ex:
+                    logger.error("Error invoking respond_activity_task_completed: %s", ex, exc_info=True)
             except Exception as ex:
                 logger.error(f"Activity {task.activity_type.name} failed: {type(ex).__name__}({ex})", exc_info=True)
-                await complete_exceptionally(service, task_token, ex)
-                # -----
-                # if error:
-                #     logger.error("Error invoking RespondActivityTaskFailed: %s", error)
-                # -----
+                try:
+                    await complete_exceptionally(service, task_token, ex)
+                except GRPCError as ex2:
+                    logger.error("Error invoking respond_activity_task_failed: %s", ex2, exc_info=True)
             finally:
                 ActivityContext.set(None)
                 process_end = datetime.datetime.now()
@@ -103,3 +102,4 @@ async def activity_task_loop_func(worker: Worker):
         except Exception:
             logger.warning("service.close() failed", exc_info=True)
         worker.notify_thread_stopped()
+        logger.info("Activity loop ended")

temporal/activity_method.py

@@ -1,11 +1,10 @@
-import copy
 import inspect
 from dataclasses import dataclass, field
 from datetime import timedelta
 from typing import Callable, List
 
 from temporal.api.common.v1 import RetryPolicy, ActivityType, Payloads
-from temporal.conversions import to_payloads
+
 
 
 def get_activity_method_name(method: Callable):
@@ -49,17 +48,10 @@ def activity_method(func: Callable = None, name: str = "", schedule_to_close_tim
     def wrapper(fn: Callable):
         # noinspection PyProtectedMember
         async def stub_activity_fn(self, *args):
-            assert self._decision_context
-            assert stub_activity_fn._execute_parameters
-            parameters: ExecuteActivityParameters = copy.deepcopy(stub_activity_fn._execute_parameters)
-            if hasattr(self, "_activity_options") and self._activity_options:
-                self._activity_options.fill_execute_activity_parameters(parameters)
-            if self._retry_parameters:
-                parameters.retry_parameters = self._retry_parameters
-            parameters.input = to_payloads(args)
-            from temporal.decision_loop import DecisionContext
-            decision_context: DecisionContext = self._decision_context
-            return await decision_context.schedule_activity_task(parameters=parameters)
+            from .async_activity import Async
+            from .decision_loop import ActivityFuture
+            future: ActivityFuture = Async.function_with_self(stub_activity_fn, self, *args)
+            return await future.wait_for_result()
 
         if not task_queue:
             raise Exception("task_queue parameter is mandatory")
@@ -102,3 +94,21 @@ def fill_execute_activity_parameters(self, execute_parameters: ExecuteActivityPa
             execute_parameters.heartbeat_timeout = self.heartbeat_timeout
         if self.task_queue is not None:
             execute_parameters.task_queue = self.task_queue
+
+
+@dataclass
+class UntypedActivityStub:
+    _decision_context: object = None
+    _retry_parameters: RetryParameters = None
+    _activity_options: ActivityOptions = None
+
+    async def execute(self, activity_name: str, *args):
+        f = await self.execute_async(activity_name, *args)
+        return await f.wait_for_result()
+
+    async def execute_async(self, activity_name: str, *args):
+        from .async_activity import Async
+        execute_parameters = ExecuteActivityParameters()
+        execute_parameters.activity_type = ActivityType()
+        execute_parameters.activity_type.name = activity_name
+        return Async.call(self, execute_parameters, args)

temporal/async_activity.py

@@ -0,0 +1,65 @@
+import copy
+from asyncio import Future
+from typing import List, Union
+
+from temporal.activity_method import ExecuteActivityParameters
+from temporal.conversions import to_payloads
+from temporal.decision_loop import ActivityFuture
+
+
+class Async:
+    @staticmethod
+    def function(method, *args) -> ActivityFuture:
+        return Async.function_with_self(method, method.__self__, *args)
+
+    @staticmethod
+    def function_with_self(method, self, *args):
+        assert self._decision_context
+        assert method._execute_parameters
+        parameters: ExecuteActivityParameters = copy.deepcopy(method._execute_parameters)
+        return Async.call(self, parameters, args)
+
+    @staticmethod
+    def call(self, parameters, args: List[object]):
+        if hasattr(self, "_activity_options") and self._activity_options:
+            self._activity_options.fill_execute_activity_parameters(parameters)
+        if self._retry_parameters:
+            parameters.retry_parameters = self._retry_parameters
+        parameters.input = to_payloads(args)
+        from temporal.decision_loop import DecisionContext
+        decision_context: DecisionContext = self._decision_context
+        return decision_context.schedule_activity_task(parameters=parameters)
+
+    @staticmethod
+    async def any_of(futures: List[Union[ActivityFuture, Future]], timeout_seconds=0):
+        done, pending = [], []
+
+        def condition():
+            done[:] = []
+            pending[:] = []
+            for f in futures:
+                if f.done():
+                    done.append(f)
+                else:
+                    pending.append(f)
+            if done:
+                return True
+            else:
+                return False
+
+        await Workflow.await_till(condition, timeout_seconds=timeout_seconds)
+        return done, pending
+
+    @staticmethod
+    async def all_of(futures: List[Union[ActivityFuture, Future]], timeout_seconds=0):
+
+        def condition():
+            for f in futures:
+                if not f.done():
+                    return False
+            return True
+
+        await Workflow.await_till(condition, timeout_seconds=timeout_seconds)
+
+
+from temporal.workflow import Workflow