Task Syncronization context patch to aviod WPF deadlock, Holepunch extension .

Author: ReferenceType

NetworkLibrary/TCP/Base/AsyncTcpClient.cs

@@ -34,9 +34,9 @@ public override void Connect(string IP, int port)
 
         public override async Task<bool> ConnectAsyncAwaitable(string IP, int port)
         {
-            connectedCompletionSource = new TaskCompletionSource<bool>();
+            connectedCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
             ConnectAsync(IP, port);
-            return await connectedCompletionSource.Task;
+            return await connectedCompletionSource.Task.ConfigureAwait(false);
         }
 
         public override void ConnectAsync(string IP, int port)

NetworkLibrary/TCP/SSL/SslClient.cs

@@ -61,7 +61,7 @@ public override async Task<bool> ConnectAsyncAwaitable(string ip, int port)
                 IsConnecting = true;
                 var clientSocket = GetSocket();
 
-                await clientSocket.ConnectAsync(new IPEndPoint(IPAddress.Parse(ip), port));
+                await clientSocket.ConnectAsync(new IPEndPoint(IPAddress.Parse(ip), port)).ConfigureAwait(false);
 
                 Connected(ip,clientSocket);
                 return true;
@@ -82,7 +82,7 @@ public override void ConnectAsync(string IP, int port)
                 bool result = false;
                 try
                 {
-                    result = await ConnectAsyncAwaitable(IP, port);
+                    result = await ConnectAsyncAwaitable(IP, port).ConfigureAwait(false);
 
                 }
                 catch (Exception ex) 

NetworkLibrary/UDP/AsyncUdpClient.cs

@@ -129,7 +129,8 @@ private void Receive()
             {
                 try
                 {
-                    activeRec = clientSocket.BeginReceiveFrom(recieveBuffer, 0, recieveBuffer.Length, SocketFlags.None, ref remoteEndPoint, EndRecieveFrom, null);
+                    EndPoint rr = new IPEndPoint(IPAddress.Any, 0);
+                    activeRec = clientSocket.BeginReceiveFrom(recieveBuffer, 0, recieveBuffer.Length, SocketFlags.None, ref rr, EndRecieveFrom, null);
 
                 }
                 catch (Exception e)
@@ -213,6 +214,36 @@ public void SendAsync(byte[] bytes)
             SendAsync(bytes, 0, bytes.Length);
         }
 
+        public void SendTo(byte[] bytes, int offset, int count, EndPoint endpoint)
+        {
+            clientSocket.SendTo(bytes, offset, count, SocketFlags.None, endpoint);
+        }
+
+        public void ReceiveOnceFrom(EndPoint endPoint, Action<byte[],int,int> OnReceived)
+        {
+            var buffer = BufferPool.RentBuffer(64000);
+            clientSocket.BeginReceiveFrom(buffer, 0, 62000, SocketFlags.None, ref endPoint, OnReceived_, buffer);
+
+            void OnReceived_(IAsyncResult ar)
+            {
+                try
+                {
+                    int amount = clientSocket.EndReceiveFrom(ar,ref endPoint);
+                    byte[] bytes = ar.AsyncState as byte[];
+                    OnReceived?.Invoke(bytes, 0, amount);
+                    BufferPool.ReturnBuffer(bytes);
+                }
+                catch(Exception e) 
+                {
+                    MiniLogger.Log(MiniLogger.LogLevel.Error,"ReceiveOnceFrom Error: "+e.Message);
+                    OnReceived?.Invoke(null, 0, 0);
+                }
+            
+            }
+        }
+
+     
+
         public void JoinMulticastGroup(IPAddress multicastAddr)
         {
             MulticastOption mcOpt = new MulticastOption(multicastAddr, ((IPEndPoint)clientSocket.LocalEndPoint).Address);

NetworkLibrary/UDP/Secure/SecureUdpClient.cs

@@ -11,7 +11,7 @@ namespace NetworkLibrary.UDP.Secure
 {
     public class SecureUdpClient:AsyncUdpClient
     {
-        ConcurrentAesAlgorithm algorithm;
+        public ConcurrentAesAlgorithm algorithm;
         public SecureUdpClient(ConcurrentAesAlgorithm algorithm, int port) : base(port)
         {
             this.algorithm = algorithm;

Protobuff/Components/ConcurrentProtoSerialiser.cs

@@ -78,7 +78,7 @@ public T Deserialize<T>(byte[] data) where T : IProtoMessage
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public T Deserialize<T>(byte[] data, int offset, int count) where T : IProtoMessage
         {
-            if (null == data)
+            if (data == null || count == 0)
                 return default;
 
             ReadOnlySpan<byte> seq = new ReadOnlySpan<byte>(data, offset, count);

Protobuff/Components/MessageAwaiter.cs

@@ -192,14 +192,14 @@ public MessageAwaiter()
             };
         }
 
-        public async ValueTask<MessageEnvelope> RegisterWait(Guid messageId, int timeoutMs)
+        public async Task<MessageEnvelope> RegisterWait(Guid messageId, int timeoutMs)
         {
-            awaitingMessages[messageId] = new TaskCompletionSource<MessageEnvelope>();
+            awaitingMessages[messageId] = new TaskCompletionSource<MessageEnvelope>(TaskCreationOptions.RunContinuationsAsynchronously);
             var pending = awaitingMessages[messageId].Task;
             MessageEnvelope returnMessage;
 
             var delay = Task.Delay(timeoutMs);
-            if (await Task.WhenAny(pending, delay) == pending)
+            if (await Task.WhenAny(pending, delay).ConfigureAwait(false) == pending)
             {
                 // Task completed within timeout.
                  returnMessage = pending.Result;
@@ -217,8 +217,12 @@ public async ValueTask<MessageEnvelope> RegisterWait(Guid messageId, int timeout
 
         public void ResponseArrived(MessageEnvelope envelopedMessage)
         {
-            if(envelopedMessage.MessageId!=null && awaitingMessages.TryGetValue(envelopedMessage.MessageId, out var completionSource))
+            if (envelopedMessage.MessageId != null && awaitingMessages.TryGetValue(envelopedMessage.MessageId, out var completionSource))
+            {
+                // lock(do a copy) because continiation will be async.
+                envelopedMessage.LockBytes();
                 completionSource.TrySetResult(envelopedMessage);
+            }
         }
 
         internal bool IsWaiting(in Guid messageId)

Protobuff/P2P/HolePunch/ClientHolepunchState.cs

@@ -1,7 +1,12 @@
 using NetworkLibrary.Components;
 using NetworkLibrary.Utils;
+using ProtoBuf;
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.ComponentModel;
+using System.Drawing;
+using System.Linq;
 using System.Net;
 using System.Net.Sockets;
 using System.Text;
@@ -12,22 +17,25 @@ namespace Protobuff.P2P.HolePunch
 {
     internal class ClientHolepunchState
     {
-        public TaskCompletionSource<EncryptedUdpProtoClient> Completion = new TaskCompletionSource<EncryptedUdpProtoClient>();
+        public TaskCompletionSource<EncryptedUdpProtoClient> Completion = new TaskCompletionSource<EncryptedUdpProtoClient>(TaskCreationOptions.RunContinuationsAsynchronously);
         private int Success = 0;
 
         internal EncryptedUdpProtoClient holepunchClient;
         private readonly RelayClient client;
         internal readonly Guid stateId;
         internal readonly Guid DestinationId;
+        internal bool encrypted = true;
 
         // initiator client is the one who generated the state id.
         // this id traveled though relay to here.
-        public ClientHolepunchState(RelayClient client, Guid stateId, Guid To)
+        public ClientHolepunchState(RelayClient client, Guid stateId, Guid To, int timeoutMs = 5000, bool encrypted = true)
         {
-            StartLifetimeCounter(5000);
+            StartLifetimeCounter(timeoutMs);
             this.client = client;
             this.stateId = stateId;
             DestinationId = To;
+            this.encrypted= encrypted;
+            MiniLogger.Log(MiniLogger.LogLevel.Info,"---------- Encryption:  " + encrypted.ToString());
         }
 
         public void HandleMessage(MessageEnvelope message)
@@ -51,9 +59,11 @@ public void HandleMessage(MessageEnvelope message)
 
         private async void StartLifetimeCounter(int lifeSpanMs)
         {
-            await Task.Delay(lifeSpanMs);
+            await Task.Delay(lifeSpanMs).ConfigureAwait(false);
             if (!Completion.Task.IsCompleted)
             {
+                Interlocked.Exchange(ref cancelSends, 1);
+                Interlocked.Exchange(ref endReceives, 1);
                 holepunchClient.Dispose();
                 Completion.TrySetResult(null);
             }
@@ -69,7 +79,7 @@ private void CreateUdpChannel(MessageEnvelope message)
             holepunchClient = new EncryptedUdpProtoClient(aesAlgorithm);
             holepunchClient.SocketSendBufferSize = 12800000;
             holepunchClient.ReceiveBufferSize = 12800000;
-            holepunchClient.OnMessageReceived += HolePunchPeerMsgReceived;
+            //holepunchClient.OnMessageReceived += HolePunchPeerMsgReceived;
             holepunchClient.Bind();
 
             // tricky point: its disaster when udp client receives from 2 endpoints.. corruption
@@ -86,45 +96,134 @@ private void SendUdpEndpointMessage()
             MiniLogger.Log(MiniLogger.LogLevel.Info, "Sending Endpoint");
 
             MessageEnvelope envelope = GetEnvelope("");
+            EndpointTransferMessage innerMsg  =  new EndpointTransferMessage();
+            innerMsg.LocalEndpoints = GetLocalEndpoints();
             envelope.From = client.sessionId;
 
-            holepunchClient.SendAsyncMessage(envelope);
+            holepunchClient.SendAsyncMessage(envelope, innerMsg);
         }
+        private List<EndpointTransferMessage> GetLocalEndpoints()
+        {
+            List<EndpointTransferMessage> endpoints = new List<EndpointTransferMessage>();
+            var lep = (IPEndPoint)holepunchClient.LocalEndpoint;
 
+            var host = Dns.GetHostEntry(Dns.GetHostName());
+            foreach (var ip in host.AddressList)
+            {
+                if (ip.AddressFamily == AddressFamily.InterNetwork)
+                {
+                    if (ip.ToString() == "0.0.0.0")
+                        continue;
+                    endpoints.Add(new EndpointTransferMessage()
+                    {
+                        IpRemote = ip.ToString(),
+                        PortRemote=lep.Port
+                    });
+                }
+            }
+            return endpoints;
+        }
         int cancelSends;
+        ConcurrentProtoSerialiser seri = new ConcurrentProtoSerialiser();
         private void StartHolepunch(MessageEnvelope message)
         {
             var endPoint = message.UnpackPayload<EndpointTransferMessage>();
-            MiniLogger.Log(MiniLogger.LogLevel.Info, client.sessionId.ToString() + " --- punching  towards " + endPoint.PortRemote);
-           
-            holepunchClient?.SetRemoteEnd(endPoint.IpRemote, endPoint.PortRemote);
+            MiniLogger.Log(MiniLogger.LogLevel.Info, client.sessionId.ToString() + " --- punching  towards " + endPoint.IpRemote + " - " + endPoint.PortRemote);
+            foreach (var item in endPoint.LocalEndpoints)
+            {
+                MiniLogger.Log(MiniLogger.LogLevel.Info, client.sessionId.ToString() + " --- punching  towards " +item.IpRemote+" - "+item.PortRemote);
+            }
+
+            message.From = client.sessionId;
+            message.MessageId = stateId;
+
+            var any = new IPEndPoint(IPAddress.Any, endPoint.PortRemote);
+            holepunchClient.ReceiveOnceFrom(any, OnBytesReceived);
+
+            IPEndPoint ep = new IPEndPoint(IPAddress.Parse(endPoint.IpRemote).MapToIPv4(), endPoint.PortRemote);
+            var bytes = seri.SerializeMessageEnvelope(message, endPoint);
+          
+            PunchAlgorithm(bytes, 0, bytes.Length, ep);
+
+            foreach (var endpointMsg in endPoint.LocalEndpoints)
+            {
+                IPEndPoint epl = new IPEndPoint(IPAddress.Parse(endpointMsg.IpRemote).MapToIPv4(), endpointMsg.PortRemote);
+                var bytes_ = seri.SerializeMessageEnvelope(message, endpointMsg);
+                PunchAlgorithm(bytes_, 0, bytes_.Length, epl);
+            }
+
+         
+        }
+        private void PunchAlgorithm(byte[] bytes_,int offset,int count, EndPoint epl)
+        {
+
             //message.Payload = null;
 
-            // now we punch a hole through nat, this is exp[erimentally optimized.
+            // now we punch a hole through nat, this is experimentally optimized.
+            if (Interlocked.CompareExchange(ref cancelSends, 0, 0) == 1)
+                return;
+            holepunchClient.SendTo(bytes_, 0, count, epl);
+
+            if (Interlocked.CompareExchange(ref cancelSends, 0, 0) == 1)
+                return;
+            holepunchClient.SendTo(bytes_, 0, count, epl);
 
-            holepunchClient?.SendAsyncMessage(message);
-            holepunchClient?.SendAsyncMessage(message);
-            holepunchClient?.SendAsyncMessage(message);
+            if (Interlocked.CompareExchange(ref cancelSends, 0, 0) == 1)
+                return;
+           
 
             Task.Run(async () =>
             {
-                for (int i = 0; i < 30; i++)
+                for (int i = 0; i < 10; i++)
                 {
                     if(Interlocked.CompareExchange(ref cancelSends,0,0)==1)
                         return;
-                    holepunchClient?.SendAsyncMessage(message);
-                    if (i > 10)
-                        await Task.Delay(i - 10);
+                    holepunchClient.SendTo(bytes_, 0, count, epl);
+
+                        await Task.Delay(i).ConfigureAwait(false);
 
                 }
             });
         }
+        ConcurrentDictionary<EndpointTransferMessage,bool> successes = new ConcurrentDictionary<EndpointTransferMessage,bool>();
+        int endReceives = 0;
+        int msgSent = 0;
+        private void OnBytesReceived(byte[] arg1, int arg2, int arg3)
+        {
+            if (Interlocked.CompareExchange(ref endReceives, 0, 0) == 1)
+                return;
+            if(arg1 == null)
+            {
+                var any = new IPEndPoint(IPAddress.Any, 0);
+                holepunchClient.ReceiveOnceFrom(any, OnBytesReceived);
+                return;
+            }
+
+            if (Interlocked.CompareExchange(ref msgSent, 1,0) == 1)
+            {
+                var any = new IPEndPoint(IPAddress.Any, 0);
+                holepunchClient.ReceiveOnceFrom(any, OnBytesReceived);
+                return;
+            }
+
+            MessageEnvelope msg = seri.DeserialiseEnvelopedMessage(arg1, 0, arg3);
+            var succesfullEp= msg.UnpackPayload<EndpointTransferMessage>();
+            if (successes.TryAdd(succesfullEp,true))
+            {
+                var envelope = GetEnvelope(HolepunchHeaders.SuccesAck);
+                envelope.From = client.sessionId;
+                envelope.MessageId = stateId;
+                client.SendAsyncMessage(DestinationId, envelope,succesfullEp);
+            }
+            var any1 = new IPEndPoint(IPAddress.Any, 0);
+            holepunchClient.ReceiveOnceFrom(any1, OnBytesReceived);
+        }
 
         private void HolePunchPeerMsgReceived(MessageEnvelope obj)
         {
             if (Interlocked.CompareExchange(ref Success, 1, 0) == 0)
             {
-                Interlocked.Increment(ref cancelSends);
+                Interlocked.Exchange(ref cancelSends,1);
                 MiniLogger.Log(MiniLogger.LogLevel.Info, "got hp feedback yay!" + client.sessionId.ToString());
                 SendAckToRelay();
             }
@@ -140,11 +239,41 @@ private void SendAckToRelay()
         // done.
         private void HandleSuccess(MessageEnvelope message)
         {
-            MiniLogger.Log(MiniLogger.LogLevel.Info, "success!!");
+            Interlocked.Exchange(ref cancelSends, 1); 
             message.LockBytes();
-           holepunchClient.SwapAlgorith(new ConcurrentAesAlgorithm(message.Payload, message.Payload));
-           // holepunchClient.SwapAlgorith(null);
-            Completion.TrySetResult(holepunchClient);
+            if (message.KeyValuePairs != null)
+            {
+                var ip = message.KeyValuePairs["IP"];
+                var port = message.KeyValuePairs["Port"];
+                ThreadPool.UnsafeQueueUserWorkItem(async(s) =>
+                {
+
+                    await Task.Delay(200);
+                    Interlocked.Exchange(ref endReceives, 1);
+
+                    if(encrypted)
+                        holepunchClient.SwapAlgorith(new ConcurrentAesAlgorithm(message.Payload, message.Payload));
+                    else
+                        holepunchClient.SwapAlgorith(null);
+
+
+                    holepunchClient.SetRemoteEnd(ip, int.Parse(port));
+
+                    Completion.TrySetResult(holepunchClient);
+                    MiniLogger.Log(MiniLogger.LogLevel.Info, "HP Complete!:" + ip);
+
+
+                }, null);
+                MiniLogger.Log(MiniLogger.LogLevel.Info, "Success Selected IP:" + ip);
+
+            }
+            else
+            {
+                holepunchClient.SwapAlgorith(new ConcurrentAesAlgorithm(message.Payload, message.Payload));
+                // holepunchClient.SwapAlgorith(null);
+                Completion.TrySetResult(holepunchClient);
+            }
+         
         }
 
         private MessageEnvelope GetEnvelope(string Header)