remove spin wait from is or becomes true

test execution speed improvements

Author: condron

src/ReactiveDomain.Foundation.Tests/when_using_read_model_base.cs

@@ -37,7 +37,8 @@ public when_using_read_model_base(StreamStoreConnectionFixture fixture)
 
             AppendEvents(10, _conn, _stream1, 2);
             AppendEvents(10, _conn, _stream2, 3);
-            _conn.TryConfirmStream(_stream1, 10);
+
+            _conn.TryConfirmStream(_stream1, 10);          
             _conn.TryConfirmStream(_stream2, 10);
             _conn.TryConfirmStream(Namer.GenerateForCategory(typeof(TestAggregate)), 20);
         }
@@ -116,7 +117,7 @@ public void can_wait_for_two_streams_to_go_live()
 
             Start(_stream2, null, true);
             AssertEx.IsModelVersion(this, 21, 100, msg: $"Expected 21 got {Version}");
-            AssertEx.IsOrBecomesTrue(() => Sum == 50, 100);
+            AssertEx.IsOrBecomesTrue(() => Sum == 50, 150);
         }
         [Fact]
         public void can_listen_to_one_stream()

src/ReactiveDomain.Testing/AssertEx.cs

@@ -1,7 +1,10 @@
 using System;
+using System.Data.SqlTypes;
 using System.Threading;
+using System.Threading.Tasks;
 using ReactiveDomain.Foundation;
 using ReactiveDomain.Messaging.Bus;
+using ReactiveDomain.Util;
 using Xunit;
 
 
@@ -40,47 +43,64 @@ public static void ArraySegmentEqual<T>(
         }
 
         /// <summary>
-        /// Asserts the given function will return false before the timeout expires.
-        /// Repeatedly evaluates the function until false is returned or the timeout expires.
-        /// Will return immediately when the condition is false.
-        /// Evaluates the timeout every 10 msec until expired.
-        /// Will not yield the thread by default, if yielding is required to resolve deadlocks set yieldThread to true.
+        /// Asserts the given function will return false before the timeout expires.  
+        /// Repeatedly evaluates the function until false is returned or the timeout expires.  
+        /// Will return immediately when the condition is false.  
+        /// Evaluates the condition on an exponential back off up to 250 ms until timeout.  
+        /// Will yield the thread after each evaluation, use the hint yieldThread if it is known the function should yield before evaluation.  
         /// </summary>
         /// <param name="func">The function to evaluate.</param>
         /// <param name="timeout">A timeout in milliseconds. If not specified, defaults to 1000.</param>
         /// <param name="msg">A message to display if the condition is not satisfied.</param>
-        /// <param name="yieldThread">If true, the thread relinquishes the remainder of its time
-        /// slice to any thread of equal priority that is ready to run.</param>
+        /// <param name="yieldThread">Execution hint to yield thread before evaluation</param>
         public static void IsOrBecomesFalse(Func<bool> func, int? timeout = null, string msg = null, bool yieldThread = false)
         {
             IsOrBecomesTrue(() => !func(), timeout, msg, yieldThread);
         }
 
         /// <summary>
-        /// Asserts the given function will return true before the timeout expires.
-        /// Repeatedly evaluates the function until true is returned or the timeout expires.
-        /// Will return immediately when the condition is true.
-        /// Evaluates the timeout every 10 msec until expired.
-        /// Will not yield the thread by default, if yielding is required to resolve deadlocks set yieldThread to true.
+        /// Asserts the given function will return true before the timeout expires.  
+        /// Repeatedly evaluates the function until true is returned or the timeout expires.  
+        /// Will return immediately when the condition is true.  
+        /// Evaluates the condition on an exponential back off up to 250 ms until timeout.  
+        /// Will yield the thread after each evaluation, use the hint yieldThread if it is known the function should yield before evaluation.  
         /// </summary>
         /// <param name="func">The function to evaluate.</param>
         /// <param name="timeout">A timeout in milliseconds. If not specified, defaults to 1000.</param>
         /// <param name="msg">A message to display if the condition is not satisfied.</param>
-        /// <param name="yieldThread">If true, the thread relinquishes the remainder of its time
-        /// slice to any thread of equal priority that is ready to run.</param>
+        /// <param name="yieldThread">Execution hint to yield thread before evaluation</param>
         public static void IsOrBecomesTrue(Func<bool> func, int? timeout = null, string msg = null, bool yieldThread = false)
         {
-            if (yieldThread) Thread.Sleep(0);
-            if (!timeout.HasValue) timeout = 1000;
-            var waitLoops = timeout / 10;
+            if (!yieldThread && func() == true)
+            {
+                Assert.True(true, msg ?? "");
+                return;
+            }
+
             var result = false;
-            for (int i = 0; i < waitLoops; i++)
+            var startTime = Environment.TickCount; //returns MS since machine start
+            var endTime = startTime + (timeout ?? 1000);
+
+
+            var delay = 1;
+            while (true)
             {
-                if (SpinWait.SpinUntil(func, 10))
+                using (var task = EvaluateAfterDelay(func, TimeSpan.FromMilliseconds(delay)))
                 {
-                    result = true;
-                    break;
+                    task.Wait();
+                    if (task.Result == true)
+                    {
+                        result = true;
+                        break;
+                    }
                 }
+                var now = Environment.TickCount;
+                if ((endTime - now) <= 0) { break; }
+                if (delay < 250)
+                {
+                    delay = delay << 1;
+                }    
+                delay = Math.Min(delay, endTime - now);
             }
             Assert.True(result, msg ?? "");
         }
@@ -116,5 +136,17 @@ public static void AtLeastModelVersion(ReadModelBase readModel, int expectedVers
         {
             IsOrBecomesTrue(() => readModel.Version >= expectedVersion, timeout, msg, yieldThread);
         }
+        /// <summary>
+        /// Evaluates the given function after the supplied delay.
+        /// The current thread will yield at the start of the delay.  
+        /// </summary>
+        /// <param name="func">The function to evaluate.</param>
+        /// <param name="delay">A delay timeSpan.</param>
+        /// <param name="cancellationToken">A token to cancel evaluation, TaskCanceledException will be thrown.</param>
+        public static async Task<bool> EvaluateAfterDelay(Func<bool> func, TimeSpan delay, CancellationToken cancellationToken = default)
+        {
+            await Task.Delay(delay, cancellationToken);
+            return func();
+        }
     }
 }

src/ReactiveDomain.Testing/ConnectionUtil.cs

@@ -7,7 +7,7 @@ public static bool TryConfirmStream(this IStreamStoreConnection conn, string str
             while (true) {
                 try {
                     var slice = conn.ReadStreamForward(streamTypeName, StreamPosition.Start, 500);
-                    if (slice.IsEndOfStream && slice.Events.Length == expectedEventCount) {
+                    if (slice.IsEndOfStream && slice.Events.Length >= expectedEventCount) {
                         return true;
                     }
                     Thread.Sleep(10);

src/ReactiveDomain.Testing/Specifications/TestQueue.cs

@@ -49,7 +49,7 @@ public TestQueue(
         {
             _isFiltered = messageTypeFilter != null;
             _msgTypeFilter = messageTypeFilter ?? new Type[] { };
-            
+
             _trackTypes = trackTypes;
 
             Messages = new ConcurrentMessageQueue<IMessage>("Messages");
@@ -108,11 +108,11 @@ public void Clear()
         /// <param name="timeout">How long to wait before timing out.</param>
         public void WaitFor<T>(TimeSpan timeout) where T : IMessage
         {
-               WaitForMultiple<T>(1, timeout);       
+            WaitForMultiple<T>(1, timeout);
         }
 
         /// <summary>
-        /// Wait for some number of messages of type T to appear in the queue.
+        /// Wait for some number of messages compatible with T (via the 'is' operator) to appear in the queue.
         /// </summary>
         /// <typeparam name="T">The type to wait for.</typeparam>
         /// <param name="num">The number of messages of the given type to wait for.</param>
@@ -121,22 +121,27 @@ public void WaitForMultiple<T>(uint num, TimeSpan timeout) where T : IMessage
         {
             if (_disposed) { throw new ObjectDisposedException(nameof(TestQueue)); }
             if (!_trackTypes) { throw new InvalidOperationException("Type tracking is disabled for this instance."); }
-            var deadline = DateTime.Now + timeout;
-            do
+
+            var startTime = Environment.TickCount; //returns MS since machine start
+            var endTime = startTime + (int)timeout.TotalMilliseconds;
+
+            var delay = 1;
+            while (true)
             {
-                if (SpinWait.SpinUntil(() => Messages.Count(x => x is T) >= num, 50))
-                {
-                    return;
-                }
-                if (DateTime.Now > deadline)
+                if (_disposed) { throw new ObjectDisposedException(nameof(TestQueue)); }
+                //Evaluating the entire queue is a bit heavy, but is required to support waiting on base types, interfaces, etc. 
+                using (var task = AssertEx.EvaluateAfterDelay(() => Messages.Count(x => x is T) >= num, TimeSpan.FromMilliseconds(delay)))
                 {
-                    throw new TimeoutException();
+                    task.Wait();
+                    if (task.Result == true) { break; }
                 }
-                if (_disposed) { throw new ObjectDisposedException(nameof(TestQueue)); }
+                var now = Environment.TickCount;
+                if ((endTime - now) <= 0) { throw new TimeoutException(); }
 
-            } while (true);
+                if (delay < 250) { delay = delay << 1; }
+                delay = Math.Min(delay, endTime - now);
+            }
         }
-
         /// <summary>
         /// Wait for a message with a specific MsgId to appear in the queue.
         /// </summary>