Control queue partitioning/scale-out (static)

.gitignore

@@ -43,4 +43,5 @@ sign.snk
 .Trashes
 Icon?
 ehthumbs.db
-Thumbs.db
+Thumbs.db
+.vs/

Test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs

@@ -14,7 +14,6 @@
 namespace DurableTask.AzureStorage.Tests
 {
     using System;
-    using System.Configuration;
     using System.IO;
     using System.Linq;
     using System.Runtime.Serialization;
@@ -33,7 +32,7 @@ public class AzureStorageScenarioTests
         [TestMethod]
         public async Task HelloWorldOrchestration_Inline()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -54,7 +53,7 @@ public async Task HelloWorldOrchestration_Inline()
         [TestMethod]
         public async Task HelloWorldOrchestration_Activity()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -75,7 +74,7 @@ public async Task HelloWorldOrchestration_Activity()
         [TestMethod]
         public async Task SequentialOrchestration()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -97,7 +96,7 @@ public async Task SequentialOrchestration()
         [TestMethod]
         public async Task ParallelOrchestration()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -118,7 +117,7 @@ public async Task ParallelOrchestration()
         [TestMethod]
         public async Task ActorOrchestration()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -170,7 +169,7 @@ public async Task ActorOrchestration()
         [TestMethod]
         public async Task TerminateOrchestration()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -198,7 +197,7 @@ public async Task TerminateOrchestration()
         [TestMethod]
         public async Task TimerCancellation()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -224,7 +223,7 @@ public async Task TimerCancellation()
         [TestMethod]
         public async Task TimerExpiration()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -251,7 +250,7 @@ public async Task TimerExpiration()
         [TestMethod]
         public async Task OrchestrationConcurrency()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -288,7 +287,7 @@ public async Task OrchestrationConcurrency()
         [TestMethod]
         public async Task HandledActivityException()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -309,7 +308,7 @@ public async Task HandledActivityException()
         [TestMethod]
         public async Task UnhandledOrchestrationException()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -330,7 +329,7 @@ public async Task UnhandledOrchestrationException()
         [TestMethod]
         public async Task UnhandledActivityException()
         {
-            using (TestOrchestrationHost host = GetTestOrchestrationHost())
+            using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost())
             {
                 await host.StartAsync();
 
@@ -345,37 +344,6 @@ public async Task UnhandledActivityException()
             }
         }
 
-        static TestOrchestrationHost GetTestOrchestrationHost()
-        {
-            string storageConnectionString = GetTestSetting("StorageConnectionString");
-            if (string.IsNullOrEmpty(storageConnectionString))
-            {
-                throw new ArgumentNullException("A Storage connection string must be defined in either an environment variable or in configuration.");
-            }
-
-            var service = new AzureStorageOrchestrationService(
-                new AzureStorageOrchestrationServiceSettings
-                {
-                    StorageConnectionString = storageConnectionString,
-                    TaskHubName = ConfigurationManager.AppSettings.Get("TaskHubName"),
-                });
-
-            service.CreateAsync().GetAwaiter().GetResult();
-
-            return new TestOrchestrationHost(service);
-        }
-
-        static string GetTestSetting(string name)
-        {
-            string value = Environment.GetEnvironmentVariable("DurableTaskTest" + name);
-            if (string.IsNullOrEmpty(value))
-            {
-                value = ConfigurationManager.AppSettings.Get(name);
-            }
-
-            return value;
-        }
-
         static class Orchestrations
         {
             internal class SayHelloInline : TaskOrchestration<string, string>

Test/DurableTask.AzureStorage.Tests/TestHelpers.cs

@@ -0,0 +1,50 @@
+using System;
+using System.Collections.Generic;
+using System.Configuration;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace DurableTask.AzureStorage.Tests
+{
+    static class TestHelpers
+    {
+        public static TestOrchestrationHost GetTestOrchestrationHost()
+        {
+            string storageConnectionString = GetTestStorageAccountConnectionString();
+
+            var service = new AzureStorageOrchestrationService(
+                new AzureStorageOrchestrationServiceSettings
+                {
+                    StorageConnectionString = storageConnectionString,
+                    TaskHubName = ConfigurationManager.AppSettings.Get("TaskHubName"),
+                });
+
+            service.CreateAsync().GetAwaiter().GetResult();
+
+            return new TestOrchestrationHost(service);
+        }
+
+        public static string GetTestStorageAccountConnectionString()
+        {
+            string storageConnectionString = GetTestSetting("StorageConnectionString");
+            if (string.IsNullOrEmpty(storageConnectionString))
+            {
+                throw new ArgumentNullException("A Storage connection string must be defined in either an environment variable or in configuration.");
+            }
+
+            return storageConnectionString;
+        }
+
+        static string GetTestSetting(string name)
+        {
+            string value = Environment.GetEnvironmentVariable("DurableTaskTest" + name);
+            if (string.IsNullOrEmpty(value))
+            {
+                value = ConfigurationManager.AppSettings.Get(name);
+            }
+
+            return value;
+        }
+    }
+}

Test/DurableTask.AzureStorage.Tests/TestOrchestrationHost.cs

@@ -14,10 +14,8 @@
 namespace DurableTask.AzureStorage.Tests
 {
     using System;
-    using System.Collections;
     using System.Collections.Generic;
     using System.Diagnostics;
-    using System.Linq;
     using System.Runtime.Serialization;
     using System.Threading.Tasks;
     using DurableTask.Core;
@@ -49,7 +47,7 @@ public Task StartAsync()
 
         public Task StopAsync()
         {
-            return this.worker.StopAsync();
+            return this.worker.StopAsync(isForced: true);
         }
 
         public async Task<TestOrchestrationClient> StartOrchestrationAsync(Type orchestrationType, object input)

src/DurableTask.AzureStorage/AnalyticsEventSource.cs

@@ -118,5 +118,32 @@ public void AppendedInstanceState(string InstanceId, string ExecutionId, int New
             EnsureLogicalTraceActivityId();
             this.WriteEvent(111, InstanceId, ExecutionId, NewEventCount, TotalEventCount, NewEvents, LatencyMs);
         }
+
+        [Event(120, Level = EventLevel.Informational)]
+        public void LogPartitionInfo(string Details)
+        {
+            EnsureLogicalTraceActivityId();
+            this.WriteEvent(120, Details);
+        }
+
+        [Event(121, Level = EventLevel.Error)]
+        public void LogPartitionWarning(string Details)
+        {
+            EnsureLogicalTraceActivityId();
+            this.WriteEvent(121, Details);
+        }
+
+        [NonEvent]
+        public void LogPartitionException(Exception e)
+        {
+            this.LogPartitionError(e.ToString());
+        }
+
+        [Event(122, Level = EventLevel.Error)]
+        public void LogPartitionError(string Details)
+        {
+            EnsureLogicalTraceActivityId();
+            this.WriteEvent(122, Details);
+        }
     }
 }

src/DurableTask.AzureStorage/AssemblyInfo.cs

@@ -0,0 +1,16 @@
+//  ----------------------------------------------------------------------------------
+//  Copyright Microsoft Corporation
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//  http://www.apache.org/licenses/LICENSE-2.0
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//  ----------------------------------------------------------------------------------
+
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("DurableTask.AzureStorage.Tests")]

src/DurableTask.AzureStorage/AsyncAutoResetEvent.cs

@@ -20,7 +20,7 @@ namespace DurableTask.AzureStorage
     using System.Threading.Tasks;
 
     // Inspired by https://blogs.msdn.microsoft.com/pfxteam/2012/02/11/building-async-coordination-primitives-part-2-asyncautoresetevent/
-    public class AsyncAutoResetEvent
+    class AsyncAutoResetEvent
     {
         readonly LinkedList<TaskCompletionSource<bool>> waiters = 
             new LinkedList<TaskCompletionSource<bool>>();