get performance numbers for message pipe

Author: SirJosh3917

src/StringDB.PerformanceNumbers/InsertRangeFileSize.cs

@@ -23,8 +23,6 @@ void Insert(IDatabase<string, string> db, int c)
 			Console.WriteLine($"Size after 1 elements in an insert range: {size1}");
 			Console.WriteLine($"Size after 50 elements in an insert range: {size2}");
 			Console.WriteLine($"Size after 100 elements in an insert range: {size3}");
-
-			Console.ReadLine();
 		}
 
 		public static long GetSizeAfter(int kvps, Action<IDatabase<string, string>, int> action)

src/StringDB.PerformanceNumbers/MessagePipeMessagesPerSecond.cs

@@ -0,0 +1,56 @@
+using StringDB.Querying.Messaging;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace StringDB.PerformanceNumbers
+{
+	public class MessagePipeMessagesPerSecond
+	{
+		private readonly IMessagePipe<int> _pipe = new SimpleMessagePipe<int>();
+
+		public async Task Run()
+		{
+			var messagesPerSecond = 0;
+
+			var tell = Task.Run(() =>
+			{
+				Console.WriteLine("Tell started");
+				while(true)
+				{
+					System.Threading.Thread.Sleep(1000);
+					var loc = messagesPerSecond;
+					messagesPerSecond = 0;
+					Console.WriteLine($"{nameof(SimpleMessagePipe<int>)} messages per second: {loc:n0}");
+				}
+			});
+
+			var push = Task.Run(() =>
+			{
+				Console.WriteLine("Push started");
+				for (var i = int.MinValue; i < int.MaxValue; i++)
+				{
+					_pipe.Enqueue(i);
+				}
+			});
+
+			var consume = ((Func<Task>)(async () =>
+			{
+				Console.WriteLine("Consume started");
+				while (true)
+				{
+					try
+					{
+						var message = await _pipe.Dequeue();
+						messagesPerSecond++;
+					} catch (Exception ex)
+					{
+						Console.WriteLine("consuming ex: " + ex);
+					}
+				}
+			}))();
+
+			await Task.WhenAll(tell, push, consume).ConfigureAwait(false);
+		}
+	}
+}

src/StringDB.PerformanceNumbers/Program.cs

@@ -16,6 +16,7 @@ private static async Task Main()
 		{
 			new SingleInsertFileSize().Run();
 			new InsertRangeFileSize().Run();
+			await new MessagePipeMessagesPerSecond().Run();
 
 			const BenchmarkToRun benchmark = BenchmarkToRun.YieldOrLinq;
 

src/StringDB.PerformanceNumbers/SingleInsertFileSize.cs

@@ -16,8 +16,6 @@ public void Run()
 			Console.WriteLine($"Size after 1 single insert: {size1}");
 			Console.WriteLine($"Size after 50 single inserts: {size2}");
 			Console.WriteLine($"Size after 100 single inserts: {size3}");
-
-			Console.ReadLine();
 		}
 
 		public static long GetSizeAfter(int times, Action<IDatabase<string, string>> action)

src/StringDB/Querying/Messaging/SimpleMessagePipe.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Concurrent;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -20,16 +21,23 @@ public void Dispose()
 			_mres.Dispose();
 		}
 
+		[MethodImpl(MethodImplOptions.AggressiveInlining)]
 		public void Enqueue(T message)
 		{
 			_queue.Enqueue(message);
+
 			_mres.Set();
 		}
 
 		public Task<T> Dequeue(CancellationToken cancellationToken = default)
 		{
 			if (_queue.TryDequeue(out var result))
 			{
+				if (_mres.IsSet)
+				{
+					_mres.Reset();
+				}
+
 				return Task.FromResult(result);
 			}
 
@@ -61,5 +69,11 @@ private Task<T> AsyncDequeue(CancellationToken cancellationToken)
 
 			return tcs.Task;
 		}
+
+		[MethodImpl(MethodImplOptions.NoInlining)]
+		private static void ThrowDispose()
+		{
+			throw new ObjectDisposedException(nameof(SimpleMessagePipe<T>));
+		}
 	}
 }