Compute ops/s for each parallel workstream (Azure#4)

- More accurate than averaging time

net/Azure.Test.PerfStress/NoOpTest.cs

@@ -3,6 +3,7 @@
 
 namespace Azure.Test.PerfStress
 {
+    // Used for measuring the overhead of the perf framework with the fastest possible test
     public class NoOpTest : PerfStressTest<PerfStressOptions>
     {
         public NoOpTest(PerfStressOptions options) : base(options) { }

net/Azure.Test.PerfStress/PerfStressProgram.cs

@@ -14,14 +14,14 @@ namespace Azure.Test.PerfStress
 {
     public static class PerfStressProgram
     {
-        private static int _completedOperations;
+        private static int[] _completedOperations;
         private static TimeSpan[] _lastCompletionTimes;
 
         public static async Task Main(Assembly assembly, string[] args)
         {
             var testTypes = assembly.ExportedTypes
-                .Where(t => typeof(IPerfStressTest).IsAssignableFrom(t) && !t.IsAbstract)
-                .Append(typeof(NoOpTest));
+                .Concat(typeof(PerfStressProgram).Assembly.ExportedTypes)
+                .Where(t => typeof(IPerfStressTest).IsAssignableFrom(t) && !t.IsAbstract);
 
             var optionTypes = GetOptionTypes(testTypes);
 
@@ -122,26 +122,28 @@ private static async Task Run(Type testType, PerfStressOptions options)
 
         private static async Task RunTestsAsync(IPerfStressTest[] tests, bool sync, int parallel, int durationSeconds, string title)
         {
-            _completedOperations = 0;
+            _completedOperations = new int[parallel];
             _lastCompletionTimes = new TimeSpan[parallel];
 
             var duration = TimeSpan.FromSeconds(durationSeconds);
-            using var cts = new CancellationTokenSource(duration);
-            var cancellationToken = cts.Token;
+            using var testCts = new CancellationTokenSource(duration);
+            var cancellationToken = testCts.Token;
 
             var lastCompleted = 0;
+
+            using var progressStatusCts = new CancellationTokenSource();
             var progressStatusThread = PrintStatus(
                 $"=== {title} ===" + Environment.NewLine +
                 "Current\t\tTotal",
                 () =>
                 {
-                    var totalCompleted = _completedOperations;
+                    var totalCompleted = _completedOperations.Sum();
                     var currentCompleted = totalCompleted - lastCompleted;
                     lastCompleted = totalCompleted;
                     return currentCompleted + "\t\t" + totalCompleted;
                 },
                 newLine: true,
-                cancellationToken);
+                progressStatusCts.Token);
 
             if (sync)
             {
@@ -150,7 +152,7 @@ private static async Task RunTestsAsync(IPerfStressTest[] tests, bool sync, int
                 for (var i = 0; i < parallel; i++)
                 {
                     var j = i;
-                    threads[i] = new Thread(() => RunLoop(tests[j], cancellationToken));
+                    threads[i] = new Thread(() => RunLoop(tests[j], j, cancellationToken));
                     threads[i].Start();
                 }
                 for (var i = 0; i < parallel; i++)
@@ -166,51 +168,53 @@ private static async Task RunTestsAsync(IPerfStressTest[] tests, bool sync, int
                     var j = i;
                     // Call Task.Run() instead of directly calling RunLoopAsync(), to ensure the requested
                     // level of parallelism is achieved even if the test RunAsync() completes synchronously.
-                    tasks[j] = Task.Run(() => RunLoopAsync(tests[j], cancellationToken));
+                    tasks[j] = Task.Run(() => RunLoopAsync(tests[j], j, cancellationToken));
                 }
                 await Task.WhenAll(tasks);
             }
 
+            progressStatusCts.Cancel();
             progressStatusThread.Join();
 
             Console.WriteLine("=== Results ===");
 
-            var averageElapsedSeconds = _lastCompletionTimes.Select(t => t.TotalSeconds).Average();
-            var operationsPerSecond = _completedOperations / averageElapsedSeconds;
+            var totalOperations = _completedOperations.Sum();
+            var operationsPerSecond = _completedOperations.Zip(_lastCompletionTimes, (operations, time) => (operations / time.TotalSeconds)).Sum();
             var secondsPerOperation = 1 / operationsPerSecond;
+            var weightedAverageSeconds = totalOperations / operationsPerSecond;
 
-            Console.WriteLine($"Completed {_completedOperations} operations in an average of {averageElapsedSeconds:N2}s " +
+            Console.WriteLine($"Completed {totalOperations} operations in a weighted-average of {weightedAverageSeconds:N2}s " +
                 $"({operationsPerSecond:N2} ops/s, {secondsPerOperation:N3} s/op)");
             Console.WriteLine();
         }
 
-        private static void RunLoop(IPerfStressTest test, CancellationToken cancellationToken)
+        private static void RunLoop(IPerfStressTest test, int index, CancellationToken cancellationToken)
         {
             var sw = Stopwatch.StartNew();
             while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
                     test.Run(cancellationToken);
-                    var count = Interlocked.Increment(ref _completedOperations);
-                    _lastCompletionTimes[count % _lastCompletionTimes.Length] = sw.Elapsed;
+                    _completedOperations[index]++;
+                    _lastCompletionTimes[index] = sw.Elapsed;
                 }
                 catch (OperationCanceledException)
                 {
                 }
             }
         }
 
-        private static async Task RunLoopAsync(IPerfStressTest test, CancellationToken cancellationToken)
+        private static async Task RunLoopAsync(IPerfStressTest test, int index, CancellationToken cancellationToken)
         {
             var sw = Stopwatch.StartNew();
             while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
                     await test.RunAsync(cancellationToken);
-                    var count = Interlocked.Increment(ref _completedOperations);
-                    _lastCompletionTimes[count % _lastCompletionTimes.Length] = sw.Elapsed;
+                    _completedOperations[index]++;
+                    _lastCompletionTimes[index] = sw.Elapsed;
                 }
                 catch (Exception e)
                 {
@@ -238,21 +242,21 @@ private static Thread PrintStatus(string header, Func<object> status, bool newLi
                     try
                     {
                         Sleep(TimeSpan.FromSeconds(1), token);
+                    }
+                    catch (OperationCanceledException)
+                    {
+                    }
 
-                        var obj = status();
+                    var obj = status();
 
-                        if (newLine)
-                        {
-                            Console.WriteLine(obj);
-                        }
-                        else
-                        {
-                            Console.Write(obj);
-                            needsExtraNewline = true;
-                        }
+                    if (newLine)
+                    {
+                        Console.WriteLine(obj);
                     }
-                    catch (OperationCanceledException)
+                    else
                     {
+                        Console.Write(obj);
+                        needsExtraNewline = true;
                     }
                 }
 

net/Azure.Test.PerfStress/SleepTest.cs

@@ -0,0 +1,38 @@
+using System;
+using System.Numerics;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Azure.Test.PerfStress
+{
+    // Used for verifying the perf framework correctly computes average throughput across parallel tests of different speed
+    public class SleepTest : PerfStressTest<PerfStressOptions>
+    {
+        private static int _instanceCount = 0;
+        private readonly int _secondsPerOperation;
+
+        public SleepTest(PerfStressOptions options) : base(options) {
+            // Each instance of this test completes operations at a different rate, to allow for testing scenarios where
+            // some instances are still waiting when time expires.  The first instance completes in 2 seconds per operation,
+            // the second instance in 4 seconds, the third instance in 8 seconds, and so on. 
+
+            var instanceCount = Interlocked.Increment(ref _instanceCount);
+            _secondsPerOperation = Pow(2, instanceCount);
+        }
+
+        private static int Pow(int value, int exponent)
+        {
+            return (int)BigInteger.Pow(new BigInteger(value), exponent);
+        }
+
+        public override void Run(CancellationToken cancellationToken)
+        {
+            Thread.Sleep(TimeSpan.FromSeconds(_secondsPerOperation));
+        }
+
+        public override Task RunAsync(CancellationToken cancellationToken)
+        {
+            return Task.Delay(TimeSpan.FromSeconds(_secondsPerOperation), cancellationToken);
+        }
+    }
+}

python/.gitignore

@@ -1,4 +1,6 @@
-# https://github.com/github/gitignore/blob/master/Python.gitignore
+.vscode
+
+# https://github.com/github/gitignore/blob/master/Python.gitignore
 
 # Byte-compiled / optimized / DLL files
 __pycache__/