Extend timeout for mpsc buffer test (#411)

* spin

* yield

* ws

* rem timeout

* task ext

* detect full

* wait much longer

* log cpu count

* simpler

---------

Author: bitfaster

BitFaster.Caching.UnitTests/Buffers/MpscBoundedBufferTests.cs

@@ -1,19 +1,23 @@
 using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Runtime.InteropServices;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using BitFaster.Caching.Buffers;
 using FluentAssertions;
 using Xunit;
-
+using Xunit.Abstractions;
+
 namespace BitFaster.Caching.UnitTests.Buffers
 {
     public class MpscBoundedBufferTests
     {
-        private readonly MpscBoundedBuffer<string> buffer = new MpscBoundedBuffer<string>(10);
+        private readonly ITestOutputHelper testOutputHelper;
+        private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30);
+        private readonly MpscBoundedBuffer<string> buffer = new MpscBoundedBuffer<string>(10);
+
+        public MpscBoundedBufferTests(ITestOutputHelper testOutputHelper)
+        {
+            this.testOutputHelper = testOutputHelper;
+        }
 
         [Fact]
         public void WhenSizeIsLessThan1CtorThrows()
@@ -173,9 +177,11 @@ await Threaded.Run(4, () =>
             });
         }
 
-        [Fact(Timeout = 5000)]
+        [Fact]
         public async Task WhileBufferIsFilledItemsCanBeTaken()
         {
+            this.testOutputHelper.WriteLine($"ProcessorCount={Environment.ProcessorCount}.");
+
             var buffer = new MpscBoundedBuffer<string>(1024);
 
             var fill = Threaded.Run(4, () =>
@@ -184,34 +190,42 @@ public async Task WhileBufferIsFilledItemsCanBeTaken()
                 int count = 0;
                 while (count < 256)
                 {
-                    while (true)
-                    {
-                        if (buffer.TryAdd("hello") == BufferStatus.Success)
-                        {
-                            break;
-                        }
-                        spin.SpinOnce();
+                    while (true)
+                    {
+                        if (buffer.TryAdd("hello") == BufferStatus.Success)
+                        {
+                            break;
+                        }
+                        spin.SpinOnce();
                     }
                     count++;
                 }
             });
 
-            int taken = 0;
-
-            while (taken < 1024)
-            {
-                if (buffer.TryTake(out var _) == BufferStatus.Success) 
-                {
-                    taken++;
+            var take = Task.Run(() => 
+            {
+                int taken = 0;
+
+                while (taken < 1024)
+                {
+                    var spin = new SpinWait();
+                    if (buffer.TryTake(out var _) == BufferStatus.Success)
+                    {
+                        taken++;
+                    }
+                    spin.SpinOnce();
                 }
-            }
+            });
 
-            await fill;
+            await fill.TimeoutAfter(Timeout, $"fill timed out");
+            await take.TimeoutAfter(Timeout, "take timed out");
         }
 
-        [Fact(Timeout = 5000)]
+        [Fact]
         public async Task WhileBufferIsFilledBufferCanBeDrained()
         {
+            this.testOutputHelper.WriteLine($"ProcessorCount={Environment.ProcessorCount}.");
+
             var buffer = new MpscBoundedBuffer<string>(1024);
 
             var fill = Threaded.Run(4, () =>
@@ -232,15 +246,19 @@ public async Task WhileBufferIsFilledBufferCanBeDrained()
                 }
             });
 
-            int drained = 0;
-            var drainBuffer = new ArraySegment<string>(new string[1024]);
-
-            while (drained < 1024)
-            {
-                drained += buffer.DrainTo(drainBuffer);
-            }
+            var drain = Task.Run(() =>
+            {
+                int drained = 0;
+                var drainBuffer = new ArraySegment<string>(new string[1024]);
+
+                while (drained < 1024)
+                {
+                    drained += buffer.DrainTo(drainBuffer);
+                }
+            });
 
-            await fill;
+            await fill.TimeoutAfter(Timeout, "fill timed out");
+            await drain.TimeoutAfter(Timeout, "drain timed out");
         }
     }
 }

BitFaster.Caching.UnitTests/TaskExtensions.cs

@@ -0,0 +1,16 @@
+using System;
+using System.Threading.Tasks;
+
+namespace BitFaster.Caching.UnitTests
+{
+    public static class TaskExtensions
+    {
+        public static async Task TimeoutAfter(this Task task, TimeSpan timeout, string message)
+        {
+            if (task == await Task.WhenAny(task, Task.Delay(timeout)))
+                await task;
+            else
+                throw new TimeoutException(message);
+        }
+    }
+}

BitFaster.Caching/Buffers/MpscBoundedBuffer.cs

@@ -2,7 +2,6 @@
 using System.Diagnostics;
 using System.Runtime.CompilerServices;
 using System.Threading;
-using BitFaster.Caching.Lfu;
 
 namespace BitFaster.Caching.Buffers
 {
@@ -97,7 +96,7 @@ public BufferStatus TryAdd(T item)
 
             if (Interlocked.CompareExchange(ref this.headAndTail.Tail, tail + 1, tail) == tail)
             {
-                int index = (int)(tail & mask);
+                int index = tail & mask;
                 Volatile.Write(ref buffer[index], item);
 
                 return BufferStatus.Success;