From: Forest Date: Tue, 18 Dec 2018 09:56:45 +0000 (-0800) Subject: Got everything pretty stable now X-Git-Tag: 1.0.0~72 X-Git-Url: https://git.deb.at/?a=commitdiff_plain;h=e4fb95265d3ca1a8f2ae881d4ef9b7aaae76a745;p=rhonda%2Fimpostor.hazel.git Got everything pretty stable now --- diff --git a/Hazel.UnitTests/TestHelper.cs b/Hazel.UnitTests/TestHelper.cs index 4fb4374..6cb83a8 100644 --- a/Hazel.UnitTests/TestHelper.cs +++ b/Hazel.UnitTests/TestHelper.cs @@ -16,7 +16,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunServerToClientTest(ConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) + internal static void RunServerToClientTest(NetworkConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) { //Setup meta stuff byte[] data = BuildData(dataSize); @@ -66,7 +66,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunClientToServerTest(ConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) + internal static void RunClientToServerTest(NetworkConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) { //Setup meta stuff byte[] data = BuildData(dataSize); @@ -116,7 +116,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunServerDisconnectTest(ConnectionListener listener, Connection connection) + internal static void RunServerDisconnectTest(NetworkConnectionListener listener, Connection connection) { ManualResetEvent mutex = new ManualResetEvent(false); @@ -142,7 +142,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunClientDisconnectTest(ConnectionListener listener, Connection connection) + internal static void RunClientDisconnectTest(NetworkConnectionListener listener, Connection connection) { ManualResetEvent mutex = new ManualResetEvent(false); ManualResetEvent mutex2 = new ManualResetEvent(false); diff --git a/Hazel/ConnectionListener.cs b/Hazel/ConnectionListener.cs index 51fe187..377e49b 100644 --- a/Hazel/ConnectionListener.cs +++ b/Hazel/ConnectionListener.cs @@ -24,7 +24,7 @@ namespace Hazel /// /// /// - public abstract class ConnectionListener : IDisposable + public abstract class List : IDisposable { /// /// Invoked when a new client connects. diff --git a/Hazel/DataReceivedEventArgs.cs b/Hazel/DataReceivedEventArgs.cs index 69446e4..1d6ea51 100644 --- a/Hazel/DataReceivedEventArgs.cs +++ b/Hazel/DataReceivedEventArgs.cs @@ -16,20 +16,15 @@ namespace Hazel /// /// /// - public class DataReceivedEventArgs : EventArgs, IRecyclable + public class DataReceivedEventArgs : EventArgs { - /// - /// Object pool for this event. - /// - static readonly ObjectPool objectPool = new ObjectPool(() => new DataReceivedEventArgs()); - /// /// Returns an instance of this object from the pool. /// /// A new or recycled DataEventArgs object. internal static DataReceivedEventArgs GetObject() { - return objectPool.GetObject(); + return new DataReceivedEventArgs(); } /// @@ -63,11 +58,5 @@ namespace Hazel this.SendOption = sendOption; this.ReliableId = reliableId; } - - /// - public void Recycle() - { - objectPool.PutObject(this); - } } } diff --git a/Hazel/DisconnectedEventArgs.cs b/Hazel/DisconnectedEventArgs.cs index 36ec2e5..4e950ad 100644 --- a/Hazel/DisconnectedEventArgs.cs +++ b/Hazel/DisconnectedEventArgs.cs @@ -16,20 +16,15 @@ namespace Hazel /// /// /// - public class DisconnectedEventArgs : EventArgs, IRecyclable + public class DisconnectedEventArgs : EventArgs { - /// - /// Object pool for this event. - /// - static readonly ObjectPool objectPool = new ObjectPool(() => new DisconnectedEventArgs()); - /// /// Returns an instance of this object from the pool. /// /// A new or recycled DisconnectedEventArgs object. internal static DisconnectedEventArgs GetObject() { - return objectPool.GetObject(); + return new DisconnectedEventArgs(); } /// @@ -59,11 +54,5 @@ namespace Hazel { this.Exception = e; } - - /// - public void Recycle() - { - objectPool.PutObject(this); - } } } diff --git a/Hazel/MessageReader.cs b/Hazel/MessageReader.cs index 07677e2..6f7bd59 100644 --- a/Hazel/MessageReader.cs +++ b/Hazel/MessageReader.cs @@ -1,12 +1,18 @@ using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; using System.Runtime.CompilerServices; using System.Text; +using System.Threading; namespace Hazel { - /// public class MessageReader : IRecyclable { + public static int Readers = 0; + public int ReaderId = 0; + public static readonly ObjectPool ReaderPool = new ObjectPool(() => new MessageReader()); public byte[] Buffer; @@ -21,14 +27,22 @@ namespace Hazel set { this._position = value; - this.readHead = this._position + Offset; + this.readHead = value + Offset; } } - + private int _position; - private int readHead; + public MessageReader() + { + this.ReaderId = Interlocked.Increment(ref Readers); + } + public override string ToString() + { + return $"{ReaderId}: BL:{Buffer.Length} O:{Offset} L:{Length}"; + } + public static MessageReader GetSized(int minSize) { var output = ReaderPool.GetObject(); @@ -37,9 +51,6 @@ namespace Hazel output.Buffer = new byte[minSize]; } - output.Offset = 0; - output.Position = 0; - output.Length = minSize; output.Tag = byte.MaxValue; return output; } @@ -47,6 +58,7 @@ namespace Hazel public static MessageReader GetRaw(byte[] bytes, int offset, int length) { var output = ReaderPool.GetObject(); + output.Buffer = bytes; output.Offset = offset; output.Position = 0; @@ -58,6 +70,7 @@ namespace Hazel public static MessageReader Get(byte[] buffer) { var output = ReaderPool.GetObject(); + output.Buffer = buffer; output.Offset = 0; output.Position = 0; @@ -67,14 +80,33 @@ namespace Hazel return output; } + public static MessageReader Get(MessageReader source) + { + var output = GetSized(source.Buffer.Length); + System.Buffer.BlockCopy(source.Buffer, 0, output.Buffer, 0, source.Buffer.Length); + + output.Offset = source.Offset; + + output._position = source._position; + output.readHead = source.readHead; + + output.Length = source.Length; + output.Tag = source.Tag; + + return output; + } + public static MessageReader Get(byte[] buffer, int offset) { + // Ensure there is at least a header + if (offset + 3 > buffer.Length) return null; + var output = ReaderPool.GetObject(); + output.Buffer = buffer; output.Offset = offset; output.Position = 0; - if (output.readHead + 3 > output.Buffer.Length) return null; output.Length = output.ReadUInt16(); output.Tag = output.ReadByte(); @@ -87,8 +119,20 @@ namespace Hazel /// public MessageReader ReadMessage() { - var output = MessageReader.Get(this.Buffer, this.readHead); - if (output == null) return null; + // Ensure there is at least a header + if (this.readHead + 3 > this.Buffer.Length) return null; + + var output = new MessageReader(); + + output.Buffer = this.Buffer; + output.Offset = this.readHead; + output.Position = 0; + + output.Length = output.ReadUInt16(); + output.Tag = output.ReadByte(); + + output.Offset += 3; + output.Position = 0; this.Position += output.Length + 3; return output; @@ -97,7 +141,6 @@ namespace Hazel /// public void Recycle() { - this.Position = this.Length = 0; ReaderPool.PutObject(this); } diff --git a/Hazel/NetworkConnectionListener.cs b/Hazel/NetworkConnectionListener.cs index 1e5e237..9e9bb72 100644 --- a/Hazel/NetworkConnectionListener.cs +++ b/Hazel/NetworkConnectionListener.cs @@ -8,10 +8,10 @@ using System.Text; namespace Hazel { /// - /// Abstract base class for a for network based connections. + /// Abstract base class for a for network based connections. /// /// - public abstract class NetworkConnectionListener : ConnectionListener + public abstract class NetworkConnectionListener : List { /// /// The local end point the listener is listening for new clients on. diff --git a/Hazel/NewConnectionEventArgs.cs b/Hazel/NewConnectionEventArgs.cs index b757284..5df102c 100644 --- a/Hazel/NewConnectionEventArgs.cs +++ b/Hazel/NewConnectionEventArgs.cs @@ -6,30 +6,25 @@ using System.Text; namespace Hazel { /// - /// Event arguments for the event. + /// Event arguments for the event. /// /// /// /// This contains the new connection for the client that connection and is passed to subscribers of the - /// event. + /// event. /// /// /// /// - public class NewConnectionEventArgs : EventArgs, IRecyclable + public class NewConnectionEventArgs : EventArgs { - /// - /// Object pool for this event. - /// - static readonly ObjectPool objectPool = new ObjectPool(() => new NewConnectionEventArgs()); - /// /// Returns an instance of this object from the pool. /// /// A new or recycled NewConnectionEventArgs object. internal static NewConnectionEventArgs GetObject() { - return objectPool.GetObject(); + return new NewConnectionEventArgs(); } /// @@ -60,11 +55,5 @@ namespace Hazel this.HandshakeData = msg; this.Connection = connection; } - - /// - public void Recycle() - { - objectPool.PutObject(this); - } } } diff --git a/Hazel/ObjectPool.cs b/Hazel/ObjectPool.cs index fdc1058..b97907c 100644 --- a/Hazel/ObjectPool.cs +++ b/Hazel/ObjectPool.cs @@ -23,6 +23,8 @@ namespace Hazel /// ConcurrentBag pool = new ConcurrentBag(); + private ConcurrentDictionary inuse = new ConcurrentDictionary(); + /// /// The generator for creating new objects. /// @@ -44,11 +46,18 @@ namespace Hazel internal T GetObject() { T item; - if (pool.TryTake(out item)) - return item; + if (!pool.TryTake(out item)) + { + Interlocked.Increment(ref numberCreated); + item = objectFactory.Invoke(); + } + + if (!inuse.TryAdd(item, true)) + { + throw new Exception("Duplicate pull"); + } - Interlocked.Increment(ref numberCreated); - return objectFactory.Invoke(); + return item; } /// @@ -57,7 +66,14 @@ namespace Hazel /// The item to return. internal void PutObject(T item) { - pool.Add(item); + if (inuse.TryRemove(item, out bool b)) + { + pool.Add(item); + } + else + { + throw new Exception("Duplicate add"); + } } } } diff --git a/Hazel/Udp/UdpClientConnection.cs b/Hazel/Udp/UdpClientConnection.cs index 0616299..e987643 100644 --- a/Hazel/Udp/UdpClientConnection.cs +++ b/Hazel/Udp/UdpClientConnection.cs @@ -286,7 +286,7 @@ namespace Hazel.Udp } MessageReader msg = MessageReader.GetRaw(bytes, 0, bytesReceived); - HandleReceive(msg); + HandleReceive(msg, bytesReceived); } /// diff --git a/Hazel/Udp/UdpConnection.Reliable.cs b/Hazel/Udp/UdpConnection.Reliable.cs index 74e2824..8145ce1 100644 --- a/Hazel/Udp/UdpConnection.Reliable.cs +++ b/Hazel/Udp/UdpConnection.Reliable.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Linq; @@ -30,12 +31,12 @@ namespace Hazel.Udp /// /// Holds the last ID allocated. /// - volatile ushort lastIDAllocated; + volatile int lastIDAllocated = ushort.MaxValue + 1; /// /// The packets of data that have been transmitted reliably and not acknowledged. /// - Dictionary reliableDataPacketsSent = new Dictionary(); + ConcurrentDictionary reliableDataPacketsSent = new ConcurrentDictionary(); /// /// The last packets that were received. @@ -94,11 +95,11 @@ namespace Hazel.Udp return objectPool.GetObject(); } + public ushort Id; public byte[] Data; public Timer Timer; public volatile int LastTimeout; public Action AckCallback; - public volatile bool Acknowledged; public volatile int Retransmissions; public Stopwatch Stopwatch = new Stopwatch(); @@ -107,11 +108,12 @@ namespace Hazel.Udp } - internal void Set(byte[] data, Action resendAction, int timeout, Action ackCallback) + internal void Set(ushort id, byte[] data, Action resendAction, int timeout, Action ackCallback) { - Data = data; + this.Id = id; + this.Data = data; - Timer = new Timer( + this.Timer = new Timer( (object obj) => resendAction(this), null, timeout, @@ -120,7 +122,6 @@ namespace Hazel.Udp LastTimeout = timeout; AckCallback = ackCallback; - Acknowledged = false; Retransmissions = 0; Stopwatch.Reset(); @@ -132,8 +133,15 @@ namespace Hazel.Udp /// public void Recycle() { - lock (Timer) - Timer.Dispose(); + lock (this) + { + if (this.Timer != null) + { + this.Id = (ushort)(this.Id - 1); + this.Timer.Dispose(); + this.Timer = null; + } + } objectPool.PutObject(this); } @@ -151,8 +159,15 @@ namespace Hazel.Udp { if (disposing) { - lock (Timer) - Timer.Dispose(); + lock (this) + { + if (this.Timer != null) + { + this.Id = (ushort)(this.Id - 1); + this.Timer.Dispose(); + this.Timer = null; + } + } } } } @@ -171,39 +186,45 @@ namespace Hazel.Udp //Find an ID not used yet. ushort id; + //Create packet object + Packet packet = Packet.GetObject(); + do - id = ++lastIDAllocated; - while (reliableDataPacketsSent.ContainsKey(id)); + { + id = (ushort)Interlocked.Increment(ref lastIDAllocated); + } + while (!reliableDataPacketsSent.TryAdd(id, packet)); //Write ID buffer[offset] = (byte)((id >> 8) & 0xFF); buffer[offset + 1] = (byte)id; - //Create packet object - Packet packet = Packet.GetObject(); packet.Set( + id, buffer, (Packet p) => { - lock (p.Timer) + Packet self; + if (p.Stopwatch.ElapsedMilliseconds > this.disconnectTimeout) { - if (!p.Acknowledged) + if (reliableDataPacketsSent.TryRemove(p.Id, out self)) { - if (p.Stopwatch.ElapsedMilliseconds > this.disconnectTimeout) - { - HandleDisconnect(new HazelException($"Reliable packet {id} was not ack'd after {p.Retransmissions} resends")); + HandleDisconnect(new HazelException($"Reliable packet {self.Id} was not ack'd after {self.Retransmissions} resends")); - //Set acknowledged so we dont change the timer again - p.Acknowledged = true; + self.Recycle(); + } - p.Recycle(); - return; - } + return; + } - // Backoff retry frequency to avoid congestion - p.LastTimeout = (int)Math.Min(p.LastTimeout * 1.5f, this.disconnectTimeout / 2f); - p.Timer.Change(p.LastTimeout, Timeout.Infinite); - } + lock (p) + { + // Callback for a previous packet + if (p.Id != id) return; + + // Backoff retry frequency to avoid congestion + p.LastTimeout = (int)Math.Min(p.LastTimeout * 1.5f, this.disconnectTimeout / 2f); + p.Timer.Change(p.LastTimeout, Timeout.Infinite); } try @@ -222,9 +243,6 @@ namespace Hazel.Udp resendTimeout > 0 ? resendTimeout : (int)Math.Max(40, Math.Min(AveragePingMs * 4, 750)), ackCallback ); - - //Remember packet - reliableDataPacketsSent.Add(id, packet); } } @@ -287,11 +305,17 @@ namespace Hazel.Udp /// Handles a reliable message being received and invokes the data event. /// /// The buffer received. - void ReliableMessageReceive(MessageReader message) + void ReliableMessageReceive(MessageReader message, int bytesReceived) { ushort id; if (ProcessReliableReceive(message.Buffer, 1, out id)) - InvokeDataReceived(SendOption.Reliable, message, 3, id); + { + InvokeDataReceived(SendOption.Reliable, message, 3, bytesReceived, id); + } + else + { + message.Recycle(); + } Statistics.LogReliableReceive(message.Length - 3, message.Length); } @@ -304,11 +328,14 @@ namespace Hazel.Udp /// Whether the packet was a new packet or not. bool ProcessReliableReceive(byte[] bytes, int offset, out ushort id) { + byte b1 = bytes[offset]; + byte b2 = bytes[offset + 1]; + //Get the ID form the packet - id = (ushort)((bytes[offset] << 8) + bytes[offset + 1]); + id = (ushort)((b1 << 8) + b2); //Send an acknowledgement - SendAck(bytes[offset], bytes[offset + 1]); + SendAck(b1, b2); /* * It gets a little complicated here (note the fact I'm actually using a multiline comment for once...) @@ -384,30 +411,21 @@ namespace Hazel.Udp //Get ID ushort id = (ushort)((bytes[1] << 8) + bytes[2]); - lock (reliableDataPacketsSent) + //Dispose of timer and remove from dictionary + Packet packet; + if (reliableDataPacketsSent.TryRemove(id, out packet)) { - //Dispose of timer and remove from dictionary - Packet packet; - if (reliableDataPacketsSent.TryGetValue(id, out packet)) - { - packet.Acknowledged = true; - - if (packet.AckCallback != null) - packet.AckCallback.Invoke(); - - //Add to average ping - packet.Stopwatch.Stop(); - lock (PingLock) - { - this.AveragePingMs = Math.Max(10, this.AveragePingMs * .7f + (float)packet.Stopwatch.Elapsed.TotalMilliseconds * .3f); - } + float rt = packet.Stopwatch.ElapsedMilliseconds; - packet.Recycle(); + packet.AckCallback?.Invoke(); + packet.Recycle(); - reliableDataPacketsSent.Remove(id); + lock (PingLock) + { + this.AveragePingMs = Math.Max(10, this.AveragePingMs * .7f + rt * .3f); } } - + Statistics.LogReliableReceive(0, bytes.Length); } @@ -436,16 +454,14 @@ namespace Hazel.Udp void DisposeReliablePackets() { - lock (this.reliableDataPacketsSent) + var keys = this.reliableDataPacketsSent.Keys.ToArray(); + foreach (var k in keys) { - foreach (var kvp in this.reliableDataPacketsSent) + Packet pkt; + if (this.reliableDataPacketsSent.TryRemove(k, out pkt)) { - Packet pkt = kvp.Value; - pkt.Acknowledged = true; pkt.Recycle(); } - - this.reliableDataPacketsSent.Clear(); } } } diff --git a/Hazel/Udp/UdpConnection.cs b/Hazel/Udp/UdpConnection.cs index 2e82f84..0e98231 100644 --- a/Hazel/Udp/UdpConnection.cs +++ b/Hazel/Udp/UdpConnection.cs @@ -144,37 +144,43 @@ namespace Hazel.Udp /// Handles the receiving of data. /// /// The buffer containing the bytes received. - protected internal void HandleReceive(MessageReader message) + protected internal void HandleReceive(MessageReader message, int bytesReceived) { InvokeDataReceivedRaw(message.Buffer); - + + ushort id; switch (message.Buffer[0]) { //Handle reliable receives case (byte)SendOption.Reliable: - ReliableMessageReceive(message); + ReliableMessageReceive(message, bytesReceived); break; //Handle acknowledgments case (byte)UdpSendOption.Acknowledgement: AcknowledgementMessageReceive(message.Buffer); + message.Recycle(); break; //We need to acknowledge hello and ping messages but dont want to invoke any events! case (byte)UdpSendOption.Ping: + ProcessReliableReceive(message.Buffer, 1, out id); + Statistics.LogHelloReceive(message.Length); + message.Recycle(); + break; case (byte)UdpSendOption.Hello: - ushort id; ProcessReliableReceive(message.Buffer, 1, out id); Statistics.LogHelloReceive(message.Length); break; case (byte)UdpSendOption.Disconnect: - HandleDisconnect(new HazelException("The remote sent a disconnect request")); + HandleDisconnect(new HazelException("The remote sent a disconnect request")); + message.Recycle(); break; //Treat everything else as unreliable default: - InvokeDataReceived(SendOption.None, message, 1, 0); + InvokeDataReceived(SendOption.None, message, 1, bytesReceived, 0); Statistics.LogUnreliableReceive(message.Length - 1, message.Length); break; } @@ -219,10 +225,10 @@ namespace Hazel.Udp /// The send option the message was received with. /// The buffer received. /// The offset of data in the buffer. - void InvokeDataReceived(SendOption sendOption, MessageReader buffer, int dataOffset, ushort reliableId) + void InvokeDataReceived(SendOption sendOption, MessageReader buffer, int dataOffset, int bytesReceived, ushort reliableId) { buffer.Offset = dataOffset; - buffer.Length = buffer.Length - dataOffset; + buffer.Length = bytesReceived - dataOffset; buffer.Position = 0; InvokeDataReceived(buffer, sendOption, reliableId); diff --git a/Hazel/Udp/UdpConnectionListener.cs b/Hazel/Udp/UdpConnectionListener.cs index acb47e6..86a32f9 100644 --- a/Hazel/Udp/UdpConnectionListener.cs +++ b/Hazel/Udp/UdpConnectionListener.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Net; @@ -14,6 +15,8 @@ namespace Hazel.Udp /// public class UdpConnectionListener : NetworkConnectionListener { + public const int BufferSize = ushort.MaxValue / 4; + /// /// The socket listening for connections. /// @@ -22,8 +25,9 @@ namespace Hazel.Udp /// /// The connections we currently hold /// - Dictionary connections = new Dictionary(); + ConcurrentDictionary allConnections = new ConcurrentDictionary(); + public int ConnectionCount { get { return this.allConnections.Count; } } /// /// Creates a new UdpConnectionListener for the given , port and . /// @@ -84,12 +88,14 @@ namespace Hazel.Udp void StartListeningForData() { EndPoint remoteEP = EndPoint; - + + MessageReader message = null; try { - var message = MessageReader.GetSized(ushort.MaxValue); + message = MessageReader.GetSized(BufferSize); + listener.BeginReceiveFrom(message.Buffer, 0, message.Buffer.Length, SocketFlags.None, ref remoteEP, ReadCallback, message); - Interlocked.Increment(ref ActiveThreads); + Interlocked.Increment(ref ActiveListeners); } catch (ObjectDisposedException) { @@ -99,6 +105,7 @@ namespace Hazel.Udp { //Client no longer reachable, pretend it didn't happen //TODO possibly able to disconnect client, see other TODO + message?.Recycle(); StartListeningForData(); return; } @@ -109,17 +116,27 @@ namespace Hazel.Udp /// /// The asyncronous operation's result. - private int ActiveThreads; + public int ActiveListeners; + public int ActiveCallbacks; void ReadCallback(IAsyncResult result) { + var message = (MessageReader)result.AsyncState; + Interlocked.Increment(ref ActiveCallbacks); + int bytesReceived; EndPoint remoteEndPoint = new IPEndPoint(IPMode == IPMode.IPv4 ? IPAddress.Any : IPAddress.IPv6Any, 0); //End the receive operation try { - Interlocked.Decrement(ref ActiveThreads); + Interlocked.Decrement(ref ActiveListeners); bytesReceived = listener.EndReceiveFrom(result, ref remoteEndPoint); + message.Offset = 0; + message.Length = bytesReceived; + } + catch (NullReferenceException) + { + return; } catch (ObjectDisposedException) { @@ -128,44 +145,48 @@ namespace Hazel.Udp } catch (SocketException) { - //Client no longer reachable, pretend it didn't happen - //TODO should this not inform the connection this client is lost??? + // Client no longer reachable, pretend it didn't happen + // TODO should this not inform the connection this client is lost??? - //This thread suggests the IP is not passed out from WinSoc so maybe not possible - //http://stackoverflow.com/questions/2576926/python-socket-error-on-udp-data-receive-10054 + // This thread suggests the IP is not passed out from WinSoc so maybe not possible + // http://stackoverflow.com/questions/2576926/python-socket-error-on-udp-data-receive-10054 + message.Recycle(); StartListeningForData(); return; } - //Exit if no bytes read, we've closed. + // Exit if no bytes read, we've closed. if (bytesReceived == 0) + { + message.Recycle(); return; + } //Begin receiving again StartListeningForData(); - var message = (MessageReader)result.AsyncState; - message.Length = bytesReceived; + bool aware = true; + bool isHello = message.Buffer[0] == (byte)UdpSendOption.Hello; + + //If we're aware of this connection use the one already + //If this is a new client then connect with them! + UdpServerConnection connection = this.allConnections.GetOrAdd( + remoteEndPoint, + key => { aware = false; return new UdpServerConnection(this, key, IPMode); }); - bool aware; - UdpServerConnection connection; - lock (connections) + if (!aware) { - //If we're aware of this connection use the one already - //If this is a new client then connect with them! - if (!(aware = connections.TryGetValue(remoteEndPoint, out connection))) + //Check for malformed connection attempts + if (!isHello) { - //Check for malformed connection attempts - if (message.Buffer[0] != (byte)UdpSendOption.Hello) - return; - - connection = new UdpServerConnection(this, remoteEndPoint, IPMode); - connections.Add(remoteEndPoint, connection); + Interlocked.Decrement(ref ActiveCallbacks); + message.Recycle(); + return; } } //Inform the connection of the buffer (new connections need to send an ack back to client) - connection.HandleReceive(message); + connection.HandleReceive(message, bytesReceived); //If it's a new connection invoke the NewConnection event. if (!aware) @@ -176,6 +197,12 @@ namespace Hazel.Udp message.Position = 0; InvokeNewConnection(message, connection); } + else if (isHello) + { + message.Recycle(); + } + + Interlocked.Decrement(ref ActiveCallbacks); } /// @@ -197,7 +224,11 @@ namespace Hazel.Udp endPoint, delegate (IAsyncResult result) { - listener.EndSendTo(result); + try + { + listener.EndSendTo(result); + } + catch { } }, null ); @@ -247,31 +278,20 @@ namespace Hazel.Udp /// The endpoint of the virtual connection. internal void RemoveConnectionTo(EndPoint endPoint) { - lock (connections) - connections.Remove(endPoint); + this.allConnections.TryRemove(endPoint, out var conn); } /// protected override void Dispose(bool disposing) { - lock (connections) + var keys = this.allConnections.Keys.ToArray(); + foreach (var k in keys) { - var connects = this.connections.ToArray(); - foreach (var kvp in connects) + UdpServerConnection conn; + if (this.allConnections.TryGetValue(k, out conn)) { - if (kvp.Value.State == ConnectionState.Connected) - { - try - { - kvp.Value.SendDisconnect(); - } - catch { } - } - - kvp.Value.Dispose(); + conn.Dispose(); } - - connections.Clear(); } if (listener != null)