Fixed internal thread stealing by TaskCompletionSource

Src/Neon.Networking/Tcp/TcpConfigurationPeer.cs

@@ -149,7 +149,7 @@ internal virtual void Validate()
 
         public TcpConfigurationPeer()
         {
-            synchronizationContext = new DummySynchronizationContext();
+            synchronizationContext = new NeonSynchronizationContext();
             contextSynchronizationMode = ContextSynchronizationMode.Post;
 
             memoryManager = Neon.Util.Pooling.MemoryManager.Shared;

Src/Neon.Networking/Tcp/TcpConnection.cs

@@ -586,7 +586,7 @@ async Task SendMessageSkipSimulationAsync(TcpMessage message)
                 sendTask = sendTask.ContinueWith(
                         (task, msg) =>
                         {
-                            return SendMessageInternalAsync(msg as TcpMessage);
+                            return SendMessageImmediatelyInternalAsync(msg as TcpMessage);
                         }, message, TaskContinuationOptions.ExecuteSynchronously)
                     .Unwrap();
 
@@ -596,7 +596,7 @@ async Task SendMessageSkipSimulationAsync(TcpMessage message)
             await newSendTask.ConfigureAwait(false);
         }
 
-        async Task SendMessageInternalAsync(TcpMessage message)
+        async Task SendMessageImmediatelyInternalAsync(TcpMessage message)
         {
             await sendSemaphore.WaitAsync().ConfigureAwait(false);
             try

Src/Neon.Networking/Udp/Channels/DelayedPacket.cs

@@ -14,7 +14,7 @@ struct DelayedPacket : IDisposable
         public DelayedPacket(Datagram datagram)
         {
             this.Datagram = datagram;
-            this.tcs = new TaskCompletionSource<object>();
+            this.tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
         }
 
         public void Dispose()

Src/Neon.Networking/Udp/UdpConfigurationPeer.cs

@@ -156,7 +156,7 @@ public UdpConfigurationPeer()
             sendBufferSize = 65535;
             receiveBufferSize = 1048676;
             reuseAddress = true;
-            synchronizationContext = new DummySynchronizationContext();
+            synchronizationContext = new NeonSynchronizationContext();
             tooLargeMessageBehaviour = TooLargeMessageBehaviour.RaiseException;
             contextSynchronizationMode = ContextSynchronizationMode.Post;
             networkReceiveThreads = Math.Max(1, Environment.ProcessorCount - 1);

Src/Neon.Networking/Udp/UdpConnection.cs

@@ -90,8 +90,8 @@ public UdpConnection(UdpPeer peer)
             if (peer == null)
                 throw new ArgumentNullException(nameof(peer));
             Random rnd = new Random();
-            this.connectingTcs = new TaskCompletionSource<object>();
-            this.disconnectingTcs = new TaskCompletionSource<object>();
+            this.connectingTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
+            this.disconnectingTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
             this.connectionCancellationToken = new CancellationTokenSource();
             this.Id = peer.GetNextConnectionId();
             this.status = UdpConnectionStatus.InitialWaiting;

Src/Neon.Rpc/Authorization/AuthSessionClient.cs

@@ -27,7 +27,7 @@ internal async Task<object> Start(object arg, CancellationToken cancellationToke
             if (this.tcs != null)
                 throw new InvalidOperationException($"{nameof(Start)} could be called only once");
 
-            this.tcs = new TaskCompletionSource<object>();
+            this.tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             AuthenticationRequest request = new AuthenticationRequest();
             request.Argument = arg;

Src/Neon.Rpc/Middleware/EncryptionMiddleware.cs

@@ -65,7 +65,7 @@ public void Set(ICipher cipher)
             Reset();
             this.cipher = cipher ?? throw new ArgumentNullException(nameof(cipher));
             this.dh = new DiffieHellmanImpl(cipher);
-            this.tcs = new TaskCompletionSource<object>();
+            this.tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
         }
 
         void CheckCipher()

Src/Neon.Rpc/Net/RpcConfiguration.cs

@@ -131,7 +131,7 @@ public RpcConfiguration()
             Assembly assembly = Assembly.GetEntryAssembly();
             serializer = new RpcSerializer(MemoryManager.Shared);
             remotingInvocationRules = RemotingInvocationRules.Default;
-            synchronizationContext = new DummySynchronizationContext();
+            synchronizationContext = new NeonSynchronizationContext();
             contextSynchronizationMode = ContextSynchronizationMode.Post;
             taskScheduler = TaskScheduler.Default;
             logManager = Logging.LogManager.Dummy;

Src/Neon.Util/NeonSynchronizationContext.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Neon.Util
+{
+    /// <summary>
+    /// A placeholder for a synchronization context.
+    /// </summary>
+    public class NeonSynchronizationContext : SynchronizationContext
+    {
+        Task task = Task.CompletedTask;
+        object mutex = new object();
+
+        public override void Post(SendOrPostCallback d, object state)
+        {
+            lock (mutex)
+            {
+                task = task.ContinueWith(
+                    (task, args_) =>
+                    {
+                        Tuple<SendOrPostCallback, object> args__ = (Tuple<SendOrPostCallback, object>) args_;
+                        args__.Item1(args__.Item2);
+                    }, new Tuple<SendOrPostCallback, object>(d, state), TaskContinuationOptions.RunContinuationsAsynchronously);
+            }
+        }
+
+        public override void Send(SendOrPostCallback d, object state)
+        {
+            d(state);
+        }
+    }
+}