Allowing for multiple events of same name to be raised in parallel

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## v1.0.4
+
+### Microsoft.DurableTask.Worker
+
+- Fix handling of concurrent external events with the same name (https://github.com/microsoft/durabletask-dotnet/pull/194)
+
 ## v1.0.3
 
 ### Microsoft.DurableTask.Worker

eng/targets/Release.props

@@ -17,7 +17,7 @@
   </PropertyGroup>
 
   <PropertyGroup>
-    <VersionPrefix>1.0.3</VersionPrefix>
+    <VersionPrefix>1.0.4</VersionPrefix>
   </PropertyGroup>
 
 </Project>

src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs

@@ -21,11 +21,6 @@
         /// </summary>
         Type EventType { get; }
 
-        /// <summary>
-        /// Gets or sets the next task completion source in the stack.
-        /// </summary>
-        IEventSource? Next { get; set; }
-
         /// <summary>
         /// Tries to set the result on tcs.
         /// </summary>
@@ -37,9 +32,7 @@
     {
         /// <inheritdoc/>
         public Type EventType => typeof(T);
 
-        /// <inheritdoc/>
-        public IEventSource? Next { get; set; }
 
         /// <inheritdoc/>
         void IEventSource.TrySetResult(object result) => this.TrySetResult((T)result);

src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs

@@ -14,7 +14,7 @@ namespace Microsoft.DurableTask.Worker.Shims;
 /// </summary>
 sealed partial class TaskOrchestrationContextWrapper : TaskOrchestrationContext
 {
-    readonly Dictionary<string, IEventSource> externalEventSources = new(StringComparer.OrdinalIgnoreCase);
+    readonly Dictionary<string, Queue<IEventSource>> externalEventSources = new(StringComparer.OrdinalIgnoreCase);
     readonly NamedQueue<string> externalEventBuffer = new();
     readonly OrchestrationContext innerContext;
     readonly OrchestrationInvocationContext invocationContext;
@@ -195,19 +195,21 @@ public override Task<T> WaitForExternalEvent<T>(string eventName, CancellationTo
 
         // Create a task completion source that will be set when the external event arrives.
         EventTaskCompletionSource<T> eventSource = new();
-        if (this.externalEventSources.TryGetValue(eventName, out IEventSource? existing))
+        if (this.externalEventSources.TryGetValue(eventName, out Queue<IEventSource>? existing))
         {
-            if (existing.EventType != typeof(T))
+            if (existing.Count > 0 && existing.Peek().EventType != typeof(T))
             {
                 throw new ArgumentException("Events with the same name must have the same type argument. Expected"
-                    + $" {existing.EventType.FullName}.");
+                    + $" {existing.Peek().GetType().FullName} but was requested {typeof(T).FullName}.");
             }
 
-            existing.Next = eventSource;
+            existing.Enqueue(eventSource);
         }
         else
         {
-            this.externalEventSources.Add(eventName, eventSource);
+            Queue<IEventSource> eventSourceQueue = new();
+            eventSourceQueue.Enqueue(eventSource);
+            this.externalEventSources.Add(eventName, eventSourceQueue);
         }
 
         // TODO: this needs to be tracked and disposed appropriately.
@@ -307,19 +309,16 @@ static void SwapByteArrayElements(byte[] byteArray, int left, int right)
     /// <param name="rawEventPayload">The serialized event payload.</param>
     internal void CompleteExternalEvent(string eventName, string rawEventPayload)
     {
-        if (this.externalEventSources.TryGetValue(eventName, out IEventSource? waiter))
+        if (this.externalEventSources.TryGetValue(eventName, out Queue<IEventSource>? waiters))
         {
+            IEventSource waiter = waiters.Dequeue();
             object? value = this.DataConverter.Deserialize(rawEventPayload, waiter.EventType);
 
             // Events are completed in FIFO order. Remove the key if the last event was delivered.
-            if (waiter.Next == null)
+            if (waiters.Count == 0)
             {
                 this.externalEventSources.Remove(eventName);
             }
-            else
-            {
-                this.externalEventSources[eventName] = waiter.Next;
-            }
 
             waiter.TrySetResult(value);
         }

test/Grpc.IntegrationTests/OrchestrationPatterns.cs

@@ -311,6 +311,49 @@ public async Task ExternalEvents(int eventCount)
         Assert.Equal<int>(expected, metadata.ReadOutputAs<int[]>());
     }
 
+    [Theory]
+    [InlineData(1)]
+    [InlineData(5)]
+    public async Task ExternalEventsInParallel(int eventCount)
+    {
+        TaskName orchestratorName = nameof(ExternalEvents);
+        await using HostTestLifetime server = await this.StartWorkerAsync(b =>
+        {
+            b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, async ctx =>
+            {
+                List<Task<int>> events = new();
+                for (int i = 0; i < eventCount; i++)
+                {
+                    events.Add(ctx.WaitForExternalEvent<int>("Event"));
+                }
+
+                return await Task.WhenAll(events);
+            }));
+        });
+
+        string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName);
+
+        // To ensure consistency, wait for the instance to start before sending the events
+        OrchestrationMetadata metadata = await server.Client.WaitForInstanceStartAsync(
+            instanceId,
+            this.TimeoutToken);
+
+        // Send events one-at-a-time to that we can better ensure ordered processing.
+        for (int i = 0; i < eventCount; i++)
+        {
+            await server.Client.RaiseEventAsync(metadata.InstanceId, "Event", eventPayload: i);
+        }
+
+        // Once the orchestration receives all the events it is expecting, it should complete.
+        metadata = await server.Client.WaitForInstanceCompletionAsync(
+            instanceId, getInputsAndOutputs: true);
+        Assert.NotNull(metadata);
+        Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
+
+        int[] expected = Enumerable.Range(0, eventCount).ToArray();
+        Assert.Equal<int>(expected, metadata.ReadOutputAs<int[]>());
+    }
+
     [Fact]
     public async Task Termination()
     {