-
Notifications
You must be signed in to change notification settings - Fork 20
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
15 changed files
with
474 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| namespace TemporalioSamples.Mutex; | ||
|
|
||
| using Temporalio.Activities; | ||
|
|
||
| public record NotifyLockedInput(string ResourceId, string ReleaseSignalName); | ||
|
|
||
| public record UseApiThatCantBeCalledInParallelInput(TimeSpan SleepFor); | ||
|
|
||
| public record NotifyUnlockedInput(string ResourceId); | ||
|
|
||
| public static class Activities | ||
| { | ||
| [Activity] | ||
| public static void NotifyLocked(NotifyLockedInput input) | ||
| { | ||
| ActivityExecutionContext.Current.Logger.LogInformation( | ||
| "Lock for resource '{ResourceId}' acquired, release signal name '{ReleaseSignalName}'", input.ResourceId, input.ReleaseSignalName); | ||
| } | ||
|
|
||
| [Activity] | ||
| public static async Task UseApiThatCantBeCalledInParallelAsync(UseApiThatCantBeCalledInParallelInput input) | ||
| { | ||
| var logger = ActivityExecutionContext.Current.Logger; | ||
|
|
||
| logger.LogInformation("Sleeping for '{SleepFor}'...", input.SleepFor); | ||
|
|
||
| await Task.Delay(input.SleepFor); | ||
|
|
||
| logger.LogInformation("Done sleeping!"); | ||
| } | ||
|
|
||
| [Activity] | ||
| public static void NotifyUnlocked(NotifyUnlockedInput input) | ||
| { | ||
| ActivityExecutionContext.Current.Logger.LogInformation( | ||
| "Lock for resource '{ResourceId}' released", input.ResourceId); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| namespace TemporalioSamples.Mutex.Impl; | ||
|
|
||
| public interface ILockHandle : IAsyncDisposable | ||
| { | ||
| public string LockInitiatorId { get; } | ||
|
|
||
| public string ResourceId { get; } | ||
|
|
||
| public string ReleaseSignalName { get; } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| namespace TemporalioSamples.Mutex.Impl; | ||
|
|
||
| internal interface ILockHandler | ||
| { | ||
| public string? CurrentOwnerId { get; } | ||
|
|
||
| public Task HandleAsync(LockRequest lockRequest); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| namespace TemporalioSamples.Mutex.Impl; | ||
|
|
||
| internal record LockRequest(string InitiatorId, string AcquireLockSignalName, TimeSpan? Timeout = null); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| namespace TemporalioSamples.Mutex.Impl; | ||
|
|
||
| using Temporalio.Activities; | ||
| using Temporalio.Client; | ||
| using Temporalio.Workflows; | ||
|
|
||
| internal record SignalWithStartMutexWorkflowInput(string MutexWorkflowId, string ResourceId, string AcquireLockSignalName, TimeSpan? LockTimeout = null); | ||
|
|
||
| internal class MutexActivities | ||
| { | ||
| private static readonly string RequestLockSignalName = | ||
| WorkflowSignalDefinition.FromMethod( | ||
| typeof(MutexWorkflow).GetMethod(nameof(MutexWorkflow.RequestLockAsync)) | ||
| ?? throw new InvalidOperationException($"Method {nameof(MutexWorkflow.RequestLockAsync)} not found on type {typeof(MutexWorkflow)}")) | ||
| .Name ?? throw new InvalidOperationException("Signal name is null."); | ||
|
|
||
| private readonly ITemporalClient client; | ||
|
|
||
| public MutexActivities(ITemporalClient client) | ||
| { | ||
| this.client = client; | ||
| } | ||
|
|
||
| [Activity] | ||
| public async Task SignalWithStartMutexWorkflowAsync(SignalWithStartMutexWorkflowInput input) | ||
| { | ||
| var activityInfo = ActivityExecutionContext.Current.Info; | ||
|
|
||
| await this.client.StartWorkflowAsync( | ||
| (MutexWorkflow mw) => mw.RunAsync(MutexWorkflowInput.Empty), | ||
| new WorkflowOptions(input.MutexWorkflowId, activityInfo.TaskQueue) | ||
| { | ||
| StartSignal = RequestLockSignalName, | ||
| StartSignalArgs = new object[] { new LockRequest(activityInfo.WorkflowId, input.AcquireLockSignalName, input.LockTimeout), }, | ||
| }); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| namespace TemporalioSamples.Mutex.Impl; | ||
|
|
||
| using Temporalio.Workflows; | ||
|
|
||
| internal record MutexWorkflowInput(IReadOnlyCollection<LockRequest> InitialRequests) | ||
| { | ||
| public static readonly MutexWorkflowInput Empty = new(Array.Empty<LockRequest>()); | ||
| } | ||
|
|
||
| [Workflow] | ||
| internal class MutexWorkflow | ||
| { | ||
| private readonly ILockHandler lockHandler = WorkflowMutex.CreateLockHandler(); | ||
| private readonly Queue<LockRequest> requests = new(); | ||
|
|
||
| [WorkflowRun] | ||
| public async Task RunAsync(MutexWorkflowInput input) | ||
| { | ||
| var logger = Workflow.Logger; | ||
|
|
||
| foreach (var request in input.InitialRequests) | ||
| { | ||
| requests.Enqueue(request); | ||
| } | ||
|
|
||
| while (!Workflow.ContinueAsNewSuggested) | ||
| { | ||
| if (requests.Count == 0) | ||
| { | ||
| logger.LogInformation("No lock requests, waiting for more..."); | ||
|
|
||
| await Workflow.WaitConditionAsync(() => requests.Count > 0); | ||
| } | ||
|
|
||
| while (requests.TryDequeue(out var lockRequest)) | ||
| { | ||
| await lockHandler.HandleAsync(lockRequest); | ||
| } | ||
| } | ||
|
|
||
| if (requests.Count > 0) | ||
| { | ||
| var newInput = new MutexWorkflowInput(requests); | ||
| throw Workflow.CreateContinueAsNewException((MutexWorkflow x) => x.RunAsync(newInput)); | ||
| } | ||
| } | ||
|
|
||
| [WorkflowQuery] | ||
| public string? CurrentOwnerId => lockHandler.CurrentOwnerId; | ||
|
|
||
| [WorkflowSignal] | ||
| public Task RequestLockAsync(LockRequest request) | ||
| { | ||
| requests.Enqueue(request); | ||
|
|
||
| Workflow.Logger.LogInformation("Received lock request. (InitiatorId='{InitiatorId}')", request.InitiatorId); | ||
|
|
||
| return Task.CompletedTask; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| namespace TemporalioSamples.Mutex.Impl; | ||
|
|
||
| using Temporalio.Client; | ||
| using Temporalio.Worker; | ||
|
|
||
| public static class TemporalWorkerOptionsExtensions | ||
| { | ||
| public static TemporalWorkerOptions AddWorkflowMutex(this TemporalWorkerOptions options, ITemporalClient client) | ||
| { | ||
| var mutexActivities = new MutexActivities(client); | ||
|
|
||
| options | ||
| .AddAllActivities(mutexActivities) | ||
| .AddWorkflow<MutexWorkflow>(); | ||
|
|
||
| return options; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| namespace TemporalioSamples.Mutex.Impl; | ||
|
|
||
| using Temporalio.Workflows; | ||
|
|
||
| internal record AcquireLockInput(string ReleaseSignalName); | ||
|
|
||
| /// <summary> | ||
| /// Represents a mutual exclusion mechanism for Workflows. | ||
| /// This part contains API for acquiring locks. | ||
| /// </summary> | ||
| public static class WorkflowMutex | ||
| { | ||
| private const string MutexWorkflowIdPrefix = "__wm-lock:"; | ||
|
|
||
| public static async Task<ILockHandle> LockAsync(string resourceId, TimeSpan? lockTimeout = null) | ||
| { | ||
| if (!Workflow.InWorkflow) | ||
| { | ||
| throw new InvalidOperationException("Cannot acquire a lock outside of a workflow."); | ||
| } | ||
|
|
||
| var initiatorId = Workflow.Info.WorkflowId; | ||
| var lockStarted = Workflow.UtcNow; | ||
|
|
||
| string? releaseSignalName = null; | ||
| var acquireLockSignalName = Workflow.NewGuid().ToString(); | ||
| var signalDefinition = WorkflowSignalDefinition.CreateWithoutAttribute(acquireLockSignalName, (AcquireLockInput input) => | ||
| { | ||
| releaseSignalName = input.ReleaseSignalName; | ||
| return Task.CompletedTask; | ||
| }); | ||
| Workflow.Signals[acquireLockSignalName] = signalDefinition; | ||
| try | ||
| { | ||
| var startMutexWorkflowInput = new SignalWithStartMutexWorkflowInput($"{MutexWorkflowIdPrefix}{resourceId}", resourceId, acquireLockSignalName, lockTimeout); | ||
| await Workflow.ExecuteActivityAsync<MutexActivities>( | ||
| act => act.SignalWithStartMutexWorkflowAsync(startMutexWorkflowInput), | ||
| new ActivityOptions { StartToCloseTimeout = TimeSpan.FromMinutes(1), }); | ||
|
|
||
| await Workflow.WaitConditionAsync(() => releaseSignalName != null); | ||
|
|
||
| var elapsed = Workflow.UtcNow - lockStarted; | ||
| Workflow.Logger.LogInformation( | ||
| "Lock for resource '{ResourceId}' acquired in {AcquireTime}ms by '{LockInitiatorId}', release signal name '{ReleaseSignalName}'", | ||
| resourceId, | ||
| (int)elapsed.TotalMilliseconds, | ||
| initiatorId, | ||
| releaseSignalName); | ||
|
|
||
| return new LockHandle(initiatorId, startMutexWorkflowInput.MutexWorkflowId, resourceId, releaseSignalName!); | ||
| } | ||
| finally | ||
| { | ||
| Workflow.Signals.Remove(acquireLockSignalName); | ||
| } | ||
| } | ||
|
|
||
| internal static ILockHandler CreateLockHandler() | ||
| { | ||
| if (!Workflow.InWorkflow) | ||
| { | ||
| throw new InvalidOperationException("Cannot acquire a lock outside of a workflow."); | ||
| } | ||
|
|
||
| return new LockHandler(); | ||
| } | ||
|
|
||
| internal sealed class LockHandle : ILockHandle | ||
| { | ||
| private readonly string mutexWorkflowId; | ||
|
|
||
| public LockHandle(string lockInitiatorId, string mutexWorkflowId, string resourceId, string releaseSignalId) | ||
| { | ||
| LockInitiatorId = lockInitiatorId; | ||
| this.mutexWorkflowId = mutexWorkflowId; | ||
| ResourceId = resourceId; | ||
| ReleaseSignalName = releaseSignalId; | ||
| } | ||
|
|
||
| /// <inheritdoc /> | ||
| public string LockInitiatorId { get; } | ||
|
|
||
| /// <inheritdoc /> | ||
| public string ResourceId { get; } | ||
|
|
||
| /// <inheritdoc /> | ||
| public string ReleaseSignalName { get; } | ||
|
|
||
| /// <inheritdoc /> | ||
| public async ValueTask DisposeAsync() | ||
| { | ||
| var mutexHandle = Workflow.GetExternalWorkflowHandle(mutexWorkflowId); | ||
| await mutexHandle.SignalAsync(ReleaseSignalName, Array.Empty<object?>()); | ||
| } | ||
| } | ||
|
|
||
| internal sealed class LockHandler : ILockHandler | ||
| { | ||
| /// <inheritdoc /> | ||
| public string? CurrentOwnerId { get; private set; } | ||
|
|
||
| /// <inheritdoc /> | ||
| public async Task HandleAsync(LockRequest lockRequest) | ||
| { | ||
| var releaseSignalName = Workflow.NewGuid().ToString(); | ||
|
|
||
| var initiator = Workflow.GetExternalWorkflowHandle(lockRequest.InitiatorId); | ||
| await initiator.SignalAsync(lockRequest.AcquireLockSignalName, new[] { new AcquireLockInput(releaseSignalName) }); | ||
|
|
||
| var released = false; | ||
| Workflow.Signals[releaseSignalName] = WorkflowSignalDefinition.CreateWithoutAttribute(releaseSignalName, () => | ||
| { | ||
| released = true; | ||
| return Task.CompletedTask; | ||
| }); | ||
| CurrentOwnerId = lockRequest.InitiatorId; | ||
|
|
||
| if (!await Workflow.WaitConditionAsync(() => released, lockRequest.Timeout ?? Timeout.InfiniteTimeSpan)) | ||
| { | ||
| Workflow.Logger.LogWarning( | ||
| "Lock for resource '{ResourceId}' has been timed out after '{Timeout}'. (LockInitiatorId='{LockInitiatorId}')", | ||
| Workflow.Info.WorkflowId[MutexWorkflowIdPrefix.Length..], | ||
| lockRequest.Timeout, | ||
| lockRequest.InitiatorId); | ||
| } | ||
|
|
||
| CurrentOwnerId = null; | ||
| Workflow.Signals.Remove(releaseSignalName); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.