Rename the workflows for clarity, improve README

Author: nagl-temporal

resource_locking/README.md

@@ -1,7 +1,8 @@
-# Semaphore Sample
+# Resource Locking Sample
 
-This sample shows how to use a long-lived `semaphore_workflow` to ensure that each `resource` is used by at most one
-`load_workflow` at a time. `load_workflow` runs several activities while it has ownership of a resource. 
+This sample shows how to use a long-lived `LockManagerWorkflow` to ensure that each `resource` is used by at most one
+`ResourceLockingWorkflow` at a time. `ResourceLockingWorkflow` runs several activities while it has ownership of a
+resource.
 
 Run the following from this directory to start the worker:
 
@@ -11,21 +12,36 @@ This will start the worker. Then, in another terminal, run the following to exec
 
     poetry run python starter.py
 
-You should see output indicating that the semaphore workflow serialized access to each resource.
+You should see output indicating that the LockManagerWorkflow serialized access to each resource.
 
 You can query the set of current lock holders with:
 
-    tctl wf query -w semaphore --qt get_current_holders
+    tctl wf query -w lock_manager --qt get_current_holders
+
+# Other approaches
+
+There are simpler ways to manage concurrent access to resources. Consider using resource-specific workers/task queues,
+and limiting the number of activity slots on the workers. The golang SDK also [sessions](https://docs.temporal.io/develop/go/sessions)
+that allow workflows to pin themselves to workers.
+
+The technique in this sample is capable of more complex resource locking than the options above, but it doesn't scale
+as well. Specifically, it can:
+- Manage access to a set of resources that is decoupled from the set of workers and task queues
+- Run arbitrary code to place workloads on resources as they become available
 
 # Caveats
 
 This sample uses true locking (not leasing!) to avoid complexity and scaling concerns associated with heartbeating via
-signals. Locking carries the risk of a "leak" (failure to unlock) permanently removing a resource from the pool. With
-Temporal's durabile execution guarantees, this can only happen if:
+signals. Locking carries a risk where failure to unlock permanently removing a resource from the pool. However, with
+Temporal's durable execution guarantees, this can only happen if:
 
 - A LoadWorkflow times out (prohibited in the sample code)
-- You shut down your workers, and never restart them (unhandled, but probably irrelevant)
+- You shut down your workers and never restart them (unhandled, but irrelevant)
+
+If a leak were to happen, you could discover the identity of the leaker using the query above, then:
 
-If a leak were to happen in the wild, you could discover the identity of the leaker using the query above, then:
+    tctl wf signal -w lock_manager --name release_resource --input '{ "resource": "the resource", "workflow_id": "holder workflow id", "run_id": "holder run id" }'
 
-    tctl wf signal -w semaphore --name release_resource --input '{ "resource": "the resource", "workflow_id": "holder workflow id", "run_id": "holder run id" }'
+Performance: A single LockManagerWorkflow scales to tens, but not hundreds, of lock/unlock events per second. It is
+best suited for locking resources during long-running workflows. Actual performance will depend on your temporal
+server's persistence layer.

resource_locking/sem_workflow.py resource_locking/lock_manager_workflow.pyresource_locking/sem_workflow.py renamed to resource_locking/lock_manager_workflow.py

@@ -25,20 +25,17 @@ class HandoffRequest:
     old_run_id: str
     new_run_id: str
 
-SEMAPHORE_WORKFLOW_TYPE = "semaphore-wf"
-SEMAPHORE_WORKFLOW_ID = "semaphore"
+LOCK_MANAGER_WORKFLOW_ID = "lock_manager"
 
 @dataclass
-class SemaphoreWorkflowInput:
+class LockManagerWorkflowInput:
     # Key is resource, value is users of the resource. The first item in each list is the current holder of the lease
     # on that resource. A similar data structure could allow for multiple holders (perhaps the first n items are the
     # current holders).
     resource_queues: dict[str, list[AcquireRequest]]
 
-@workflow.defn(
-    name=SEMAPHORE_WORKFLOW_TYPE,
-)
-class SemaphoreWorkflow:
+@workflow.defn
+class LockManagerWorkflow:
     def __init__(self):
         self.resource_queues: dict[str, list[AcquireRequest]] = {}
 
@@ -119,7 +116,7 @@ def get_current_holders(self) -> dict[str, AcquireRequest]:
         return { k: v[0] if v else None for k, v in self.resource_queues.items() }
 
     @workflow.run
-    async def run(self, input: SemaphoreWorkflowInput) -> None:
+    async def run(self, input: LockManagerWorkflowInput) -> None:
         self.resource_queues = input.resource_queues
 
         # Continue as new either when temporal tells us to, or every 12 hours (so it occurs semi-frequently)
@@ -128,4 +125,4 @@ async def run(self, input: SemaphoreWorkflowInput) -> None:
             timeout=timedelta(hours=12),
         )
 
-        workflow.continue_as_new(SemaphoreWorkflowInput(self.resource_queues))
+        workflow.continue_as_new(LockManagerWorkflowInput(self.resource_queues))

resource_locking/load_workflow.py …rce_locking/resource_locking_workflow.pyresource_locking/load_workflow.py renamed to resource_locking/resource_locking_workflow.py resource_locking/load_workflow.py renamed to resource_locking/resource_locking_workflow.py

@@ -5,24 +5,23 @@
 
 from temporalio import activity, workflow
 
-from resource_locking.sem_workflow import AssignedResource, SEMAPHORE_WORKFLOW_ID, \
+from resource_locking.lock_manager_workflow import AssignedResource, LOCK_MANAGER_WORKFLOW_ID, \
     ReleaseRequest, AcquireRequest, HandoffRequest
 
-
 @dataclass
-class LoadActivityInput:
+class UseResourceActivityInput:
     resource: str
     iteration: str
 
 @activity.defn
-async def load(input: LoadActivityInput) -> None:
-    workflow_id = activity.info().workflow_id
-    print(f"Workflow {workflow_id} starts using {input.resource} the {input.iteration} time")
-    await asyncio.sleep(5)
-    print(f"Workflow {workflow_id} finishes using {input.resource} the {input.iteration} time")
+async def use_resource(input: UseResourceActivityInput) -> None:
+    info = activity.info()
+    activity.logger.info(f"{info.workflow_id} starts using {input.resource} the {input.iteration} time")
+    await asyncio.sleep(3)
+    activity.logger.info(f"{info.workflow_id} done using {input.resource} the {input.iteration} time")
 
 @dataclass
-class LoadWorkflowInput:
+class ResourceLockingWorkflowInput:
     # If set, this workflow will fail after the "first", "second", or "third" activity.
     iteration_to_fail_after: Optional[str]
 
@@ -36,15 +35,13 @@ class LoadWorkflowInput:
 class FailWorkflowException(Exception):
     pass
 
+# Wait this long for a resource before giving up
 MAX_RESOURCE_WAIT_TIME = timedelta(minutes=5)
 
-def has_timeout(timeout: Optional[timedelta]) -> bool:
-    return timeout is not None and timeout > timedelta(0)
-
 @workflow.defn(
   failure_exception_types=[FailWorkflowException]
 )
-class LoadWorkflow:
+class ResourceLockingWorkflow:
 
     def __init__(self):
         self.assigned_resource: Optional[str] = None
@@ -54,15 +51,15 @@ def handle_assign_resource(self, input: AssignedResource):
         self.assigned_resource = input.resource
 
     @workflow.run
-    async def run(self, input: LoadWorkflowInput):
+    async def run(self, input: ResourceLockingWorkflowInput):
         workflow.info()
         if has_timeout(workflow.info().run_timeout):
             # See "locking" comment below for rationale
-            raise FailWorkflowException(f"LoadWorkflow cannot have a run_timeout (found {workflow.info().run_timeout})")
+            raise FailWorkflowException(f"ResourceLockingWorkflow cannot have a run_timeout (found {workflow.info().run_timeout})")
         if has_timeout(workflow.info().execution_timeout):
-            raise FailWorkflowException(f"LoadWorkflow cannot have an execution_timeout (found {workflow.info().execution_timeout})")
+            raise FailWorkflowException(f"ResourceLockingWorkflow cannot have an execution_timeout (found {workflow.info().execution_timeout})")
 
-        sem_handle = workflow.get_external_workflow_handle(SEMAPHORE_WORKFLOW_ID)
+        sem_handle = workflow.get_external_workflow_handle(LOCK_MANAGER_WORKFLOW_ID)
 
         info = workflow.info()
         if input.already_owned_resource is None:
@@ -78,13 +75,13 @@ async def run(self, input: LoadWorkflowInput):
             raise FailWorkflowException(f"No resource was assigned after {MAX_RESOURCE_WAIT_TIME}")
 
         # From this point forward, we own the resource. Note that this is a lock, not a lease! Our finally block will
-        # free up the resource if an activity fails. This is why we asserted the lack of workflow-level timeouts
+        # release the resource if an activity fails. This is why we asserted the lack of workflow-level timeouts
         # above - the finally block wouldn't run if there was a timeout.
         try:
             for iteration in ["first", "second", "third"]:
                 await workflow.execute_activity(
-                    load,
-                    LoadActivityInput(self.assigned_resource, iteration),
+                    use_resource,
+                    UseResourceActivityInput(self.assigned_resource, iteration),
                     start_to_close_timeout=timedelta(seconds=10),
                 )
 
@@ -93,15 +90,20 @@ async def run(self, input: LoadWorkflowInput):
                     raise FailWorkflowException()
 
             if input.should_continue_as_new:
-                next_input = LoadWorkflowInput(
+                next_input = ResourceLockingWorkflowInput(
                     iteration_to_fail_after=input.iteration_to_fail_after,
                     should_continue_as_new=False,
                     already_owned_resource=self.assigned_resource,
                 )
                 workflow.continue_as_new(next_input)
         finally:
             # Only release the resource if we didn't continue-as-new. workflow.continue_as_new raises to halt workflow
-            # execution, but the code in this finally block will still run. It wouldn't successfully send the signal...
+            # execution, but this code in this finally block will still run. It wouldn't successfully send the signal...
             # the if statement just avoids some warnings in the log.
             if not input.should_continue_as_new:
-                await sem_handle.signal("release_resource", ReleaseRequest(self.assigned_resource, info.workflow_id, info.run_id))
+                await sem_handle.signal("release_resource", ReleaseRequest(self.assigned_resource, info.workflow_id, info.run_id))
+
+def has_timeout(timeout: Optional[timedelta]) -> bool:
+    # After continue_as_new, timeouts are 0, even if they were None before continue_as_new (and were not set in the
+    # continue_as_new call).
+    return timeout is not None and timeout > timedelta(0)

resource_locking/starter.py

@@ -3,49 +3,50 @@
 
 from temporalio.client import Client, WorkflowHandle, WorkflowFailureError
 
-from resource_locking.load_workflow import LoadWorkflow, LoadWorkflowInput
-from resource_locking.sem_workflow import SemaphoreWorkflow, SemaphoreWorkflowInput, SEMAPHORE_WORKFLOW_ID
+from resource_locking.resource_locking_workflow import ResourceLockingWorkflow, ResourceLockingWorkflowInput
+from resource_locking.lock_manager_workflow import LockManagerWorkflow, LockManagerWorkflowInput, LOCK_MANAGER_WORKFLOW_ID
 
 
 async def main():
     # Connect client
     client = await Client.connect("localhost:7233")
 
-    # Run the semaphore workflow
-    sem_handle = await client.start_workflow(
-        workflow=SemaphoreWorkflow.run,
-        arg=SemaphoreWorkflowInput({
+    # Start the LockManagerWorkflow
+    lock_manager_handle = await client.start_workflow(
+        workflow=LockManagerWorkflow.run,
+        arg=LockManagerWorkflowInput({
             "resource_a": [],
             "resource_b": [],
         }),
-        id=SEMAPHORE_WORKFLOW_ID,
+        id=LOCK_MANAGER_WORKFLOW_ID,
         task_queue="default",
     )
 
-    load_handles: list[WorkflowHandle[Any, Any]] = []
+    # Start the ResourceLockingWorkflows
+    resource_locking_handles: list[WorkflowHandle[Any, Any]] = []
     for i in range(0, 4):
-        input = LoadWorkflowInput(iteration_to_fail_after=None, should_continue_as_new=False, already_owned_resource=None)
+        input = ResourceLockingWorkflowInput(iteration_to_fail_after=None, should_continue_as_new=False, already_owned_resource=None)
         if i == 0:
             input.should_continue_as_new = True
         if i == 1:
             input.iteration_to_fail_after = "first"
 
-        load_handle = await client.start_workflow(
-            workflow=LoadWorkflow.run,
+        resource_locking_handle = await client.start_workflow(
+            workflow=ResourceLockingWorkflow.run,
             arg=input,
-            id=f"load-workflow-{i}",
+            id=f"resource-locking-workflow-{i}",
             task_queue="default",
         )
-        load_handles.append(load_handle)
+        resource_locking_handles.append(resource_locking_handle)
 
-    for load_handle in load_handles:
+    for resource_locking_handle in resource_locking_handles:
         try:
-            await load_handle.result()
+            await resource_locking_handle.result()
         except WorkflowFailureError:
             pass
 
-    await sem_handle.terminate()
-
+    # Clean up after ourselves. In the real world, the lock manager workflow would run forever.
+    await lock_manager_handle.terminate()
 
 if __name__ == "__main__":
     asyncio.run(main())

resource_locking/worker.py

@@ -4,8 +4,8 @@
 from temporalio.client import Client
 from temporalio.worker import Worker
 
-from resource_locking.load_workflow import LoadWorkflow, load
-from resource_locking.sem_workflow import SemaphoreWorkflow
+from resource_locking.resource_locking_workflow import ResourceLockingWorkflow, use_resource
+from resource_locking.lock_manager_workflow import LockManagerWorkflow
 
 async def main():
     # Uncomment the line below to see logging
@@ -18,8 +18,8 @@ async def main():
     worker = Worker(
         client,
         task_queue="default",
-        workflows=[SemaphoreWorkflow, LoadWorkflow],
-        activities=[load],
+        workflows=[LockManagerWorkflow, ResourceLockingWorkflow],
+        activities=[use_resource],
     )
 
     await worker.run()