Added blocking tests.

Dtronix · Oct 4, 2016 · 42d8b9a · 42d8b9a
1 parent 3010cb0
commit 42d8b9a
Show file tree

Hide file tree

Showing 5 changed files with 33 additions and 12 deletions.
diff --git a/src/DtronixMessageQueue.Tests.Performance/RpcPerformanceTest.cs b/src/DtronixMessageQueue.Tests.Performance/RpcPerformanceTest.cs
@@ -20,6 +20,8 @@ public RpcPerformanceTest(string[] args) {
 
 			RpcSingleProcessTest(200000, 4, config, RpcTestType.Await);
 
+			RpcSingleProcessTest(100, 4, config, RpcTestType.Block);
+
 			RpcSingleProcessTest(10000, 4, config, RpcTestType.Return);
 
 			RpcSingleProcessTest(10000, 4, config, RpcTestType.Exception);
@@ -76,12 +78,17 @@ private void RpcSingleProcessTest(int runs, int loops, RpcConfig config, RpcTest
 				sw.Restart();
 				for (var i = 0; i < runs; i++) {
 					switch (type) {
+
+						case RpcTestType.Block:
+							service.TestNoReturnBlock();
+							break;
+
 						case RpcTestType.NoRetrun:
 								service.TestNoReturn();
 							break;
 
 						case RpcTestType.Await:
-							service.TestNoReturnBlock();
+							service.TestNoReturnAwait();
 							break;
 
 						case RpcTestType.Return:
@@ -133,6 +140,7 @@ enum RpcTestType {
 		NoRetrun,
 		Return,
 		Exception,
-		Await
+		Await,
+		Block
 	}
 }
diff --git a/src/DtronixMessageQueue.Tests.Performance/Services/Server/TestService.cs b/src/DtronixMessageQueue.Tests.Performance/Services/Server/TestService.cs
@@ -6,6 +6,7 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Xml.Serialization;
 using DtronixMessageQueue.Rpc;
 
 namespace DtronixMessageQueue.Tests.Performance.Services.Server {
@@ -28,13 +29,24 @@ public void TestNoReturn() {
 
 		}
 
-		public async void TestNoReturnBlock() {
+		public async void TestNoReturnAwait() {
 			var number = Interlocked.Increment(ref call_count);
 			await Task.Delay(1000);
+
 			VerifyComplete();
 
 		}
 
+
+		public void TestNoReturnBlock() {
+			Task.Factory.StartNew(() => {
+				var number = Interlocked.Increment(ref call_count);
+
+				Thread.Sleep(1000);
+				VerifyComplete();
+			}, TaskCreationOptions.LongRunning);
+		}
+
 		public int TestIncrement() {
 			call_count++;
 			VerifyComplete();
@@ -69,6 +81,7 @@ private void VerifyComplete() {
 
 	internal interface ITestService : IRemoteService<SimpleRpcSession, RpcConfig> {
 		void TestNoReturn();
+		void TestNoReturnAwait();
 		void TestNoReturnBlock();
 		int TestIncrement();
 		void TestSetup(int calls);

diff --git a/src/DtronixMessageQueue/MqSession.cs b/src/DtronixMessageQueue/MqSession.cs
@@ -79,7 +79,7 @@ protected override void HandleIncomingBytes(byte[] buffer) {
 
 
 			if (inbox_task == null || inbox_task.IsCompleted) {
-				inbox_task = Task.Factory.StartNew(ProcessIncomingQueue);
+				inbox_task = Task.Run((Action)ProcessIncomingQueue);
 			}
 		}
 
@@ -144,7 +144,7 @@ private void ProcessOutbox() {
 
 			lock (outbox_lock) {
 				if (outbox.IsEmpty == false) {
-					outbox_task = Task.Factory.StartNew(ProcessOutbox);
+					outbox_task = Task.Run((Action)ProcessOutbox);
 				}
 			}
 
@@ -221,7 +221,7 @@ private void ProcessIncomingQueue() {
 
 			lock (inbox_lock) {
 				if (inbox_bytes.IsEmpty == false) {
-					inbox_task = Task.Factory.StartNew(ProcessIncomingQueue);
+					inbox_task = Task.Run((Action)ProcessIncomingQueue);
 				}
 			}
 
@@ -302,7 +302,7 @@ public void Send(MqMessage message) {
 			}
 
 			if (outbox_task == null || outbox_task.IsCompleted) {
-				outbox_task = Task.Factory.StartNew(ProcessOutbox);
+				outbox_task = Task.Run((Action)ProcessOutbox);
 			}
 
 

diff --git a/src/DtronixMessageQueue/Rpc/RpcCallMessageHandler.cs b/src/DtronixMessageQueue/Rpc/RpcCallMessageHandler.cs
@@ -112,7 +112,7 @@ public override bool HandleMessage(MqMessage message) {
 		private void ProcessRpcCall(MqMessage message, RpcCallMessageType message_type) {
 
 			// Execute the processing on the worker thread.
-			Task.Factory.StartNew(() => {
+			Task.Run(() => {
 
 				// Retrieve a serialization cache to work with.
 				var serialization = Session.SerializationCache.Get(message);
@@ -247,7 +247,7 @@ private void ProcessRpcCall(MqMessage message, RpcCallMessageType message_type)
 		private void ProcessRpcReturn(MqMessage message) {
 
 			// Execute the processing on the worker thread.
-			Task.Factory.StartNew(() => {
+			Task.Run(() => {
 
 				// Retrieve a serialization cache to work with.
 				var serialization = Session.SerializationCache.Get(message);

diff --git a/src/DtronixMessageQueue/Rpc/RpcSession.cs b/src/DtronixMessageQueue/Rpc/RpcSession.cs
@@ -164,7 +164,7 @@ protected override void ProcessCommand(MqFrame frame) {
 					}
 
 					// Alert the server that this session is ready for usage.
-					Task.Factory.StartNew(() => {
+					Task.Run(() => {
 						Ready?.Invoke(this, new SessionEventArgs<TSession, TConfig>((TSession) this));
 					});
 
@@ -211,7 +211,7 @@ protected override void ProcessCommand(MqFrame frame) {
 						Close(SocketCloseReason.AuthenticationFailure);
 					} else {
 						// Alert the server that this session is ready for usage.
-						Task.Factory.StartNew(() => {
+						Task.Run(() => {
 							Ready?.Invoke(this, new SessionEventArgs<TSession, TConfig>((TSession) this));
 						});
 					}
@@ -240,7 +240,7 @@ protected override void ProcessCommand(MqFrame frame) {
 					AuthenticationResult?.Invoke(this, auth_args);
 
 					// Alert the client that this session is ready for usage.
-					Task.Factory.StartNew(() => {
+					Task.Run(() => {
 						Ready?.Invoke(this, new SessionEventArgs<TSession, TConfig>((TSession) this));
 					});