From cd8bb1db7cc808fe3638610be75fd51525d426b3 Mon Sep 17 00:00:00 2001 From: Forest Date: Tue, 18 Dec 2018 19:20:21 -0800 Subject: [PATCH] Clean up some enumerators, try a single timer per connection for packets --- Hazel/Udp/UdpConnection.Reliable.cs | 188 +++++++++++++++------------- Hazel/Udp/UdpConnection.cs | 1 + Hazel/Udp/UdpConnectionListener.cs | 21 ++-- 3 files changed, 107 insertions(+), 103 deletions(-) diff --git a/Hazel/Udp/UdpConnection.Reliable.cs b/Hazel/Udp/UdpConnection.Reliable.cs index 316990b..39619ab 100644 --- a/Hazel/Udp/UdpConnection.Reliable.cs +++ b/Hazel/Udp/UdpConnection.Reliable.cs @@ -79,7 +79,7 @@ namespace Hazel.Udp /// /// Class to hold packet data /// - public class Packet : IRecyclable, IDisposable + public class Packet : IRecyclable { /// /// Object pool for this event. @@ -97,77 +97,86 @@ namespace Hazel.Udp public ushort Id; public byte[] Data; - public Timer Timer; - public volatile int LastTimeout; + + public DateTime LastSend; + public int LastTimeout; + public volatile bool Acknowledged; + + private Action ResendAction; public Action AckCallback; + public volatile int Retransmissions; public Stopwatch Stopwatch = new Stopwatch(); Packet() { - } internal void Set(ushort id, byte[] data, Action resendAction, int timeout, Action ackCallback) { this.Id = id; this.Data = data; - - this.Timer = new Timer( - (object obj) => resendAction(this), - null, - timeout, - Timeout.Infinite - ); - - LastTimeout = timeout; + + this.Acknowledged = false; + this.LastSend = DateTime.Now; + this.LastTimeout = timeout; + this.ResendAction = resendAction; AckCallback = ackCallback; Retransmissions = 0; - Stopwatch.Reset(); - Stopwatch.Start(); + Stopwatch.Restart(); } - /// - /// Returns this object back to the object pool from whence it came. - /// - public void Recycle() + public void Resend() { - lock (this) + var evt = this.ResendAction; + if (!this.Acknowledged) { - if (this.Timer != null) + if (evt != null) { - this.Id = (ushort)(this.Id - 1); - this.Timer.Dispose(); - this.Timer = null; + evt(this); } } - - PacketPool.PutObject(this); } /// - /// Disposes of this object. + /// Returns this object back to the object pool from whence it came. /// - public void Dispose() + public void Recycle() { - Dispose(true); - GC.SuppressFinalize(this); + this.Acknowledged = true; + + PacketPool.PutObject(this); } + } + + private Timer reliableTimer; + private int activePackets; - protected void Dispose(bool disposing) + private void InitializeReliableTimer() + { + reliableTimer = new Timer(ManageReliablePackets, null, 100, 100); + } + + private void ManageReliablePackets(object state) + { + if (this.reliableDataPacketsSent.Count > 0) { - if (disposing) + double minTimeout = int.MaxValue; + foreach (var kvp in this.reliableDataPacketsSent) { - lock (this) + Packet pkt = kvp.Value; + double timeSinceLast = (DateTime.Now - pkt.LastSend).TotalMilliseconds; + if (timeSinceLast >= pkt.LastTimeout) { - if (this.Timer != null) + try { - this.Id = (ushort)(this.Id - 1); - this.Timer.Dispose(); - this.Timer = null; + pkt.Resend(); } + catch { } } + + minTimeout = Math.Min(pkt.LastTimeout, minTimeout); } } } @@ -180,70 +189,67 @@ namespace Hazel.Udp /// The callback to make once the packet has been acknowledged. void AttachReliableID(byte[] buffer, int offset, int sendLength, Action ackCallback = null) { - //Find and reliable ID - lock (reliableDataPacketsSent) + //Find an ID not used yet. + ushort id; + + //Create packet object + Packet packet = Packet.GetObject(); + + do { - //Find an ID not used yet. - ushort id; + id = (ushort)Interlocked.Increment(ref lastIDAllocated); + } + while (!reliableDataPacketsSent.TryAdd(id, packet)); + + int timeout = resendTimeout > 0 ? resendTimeout : (int)Math.Max(50, Math.Min(AveragePingMs * 2, 1000)); - //Create packet object - Packet packet = Packet.GetObject(); + //Write ID + buffer[offset] = (byte)((id >> 8) & 0xFF); + buffer[offset + 1] = (byte)id; - do + packet.Set( + id, + buffer, + (Packet p) => { - id = (ushort)Interlocked.Increment(ref lastIDAllocated); - } - while (!reliableDataPacketsSent.TryAdd(id, packet)); + // Callback for a previous packet + if (p.Acknowledged) return; - //Write ID - buffer[offset] = (byte)((id >> 8) & 0xFF); - buffer[offset + 1] = (byte)id; + p.LastSend = DateTime.Now; - packet.Set( - id, - buffer, - (Packet p) => + Packet self; + if (p.Stopwatch.ElapsedMilliseconds > this.disconnectTimeout) { - Packet self; - if (p.Stopwatch.ElapsedMilliseconds > this.disconnectTimeout) + if (reliableDataPacketsSent.TryRemove(p.Id, out self)) { - if (reliableDataPacketsSent.TryRemove(p.Id, out self)) - { - HandleDisconnect(new HazelException($"Reliable packet {self.Id} was not ack'd after {self.Retransmissions} resends")); - - self.Recycle(); - } + Interlocked.Decrement(ref this.activePackets); + HandleDisconnect(new HazelException($"Reliable packet {self.Id} was not ack'd after {self.Retransmissions} resends")); - return; + self.Recycle(); } - lock (p) - { - // Callback for a previous packet - if (p.Id != id) return; + return; + } - // Backoff retry frequency to avoid congestion - p.LastTimeout = (int)Math.Min(p.LastTimeout * 1.25f, 2000); - p.Timer.Change(p.LastTimeout, Timeout.Infinite); - } + // Backoff retry frequency to avoid congestion + p.LastTimeout = (int)Math.Min(p.LastTimeout * 1.25f, 1000); - try - { - WriteBytesToConnection(p.Data, sendLength); - p.Retransmissions++; - } - catch (InvalidOperationException e) - { - //No longer connected - HandleDisconnect(new HazelException("Could not resend data as connection is no longer connected", e)); - } + try + { + WriteBytesToConnection(p.Data, sendLength); + p.Retransmissions++; + } + catch (InvalidOperationException e) + { + //No longer connected + HandleDisconnect(new HazelException("Could not resend data as connection is no longer connected", e)); + } - Trace.WriteLine("Resend."); - }, - resendTimeout > 0 ? resendTimeout : (int)Math.Max(500, Math.Min(AveragePingMs * 4, 2000)), - ackCallback - ); - } + Trace.WriteLine("Resend."); + }, + timeout, + ackCallback + ); } /// @@ -415,6 +421,7 @@ namespace Hazel.Udp Packet packet; if (reliableDataPacketsSent.TryRemove(id, out packet)) { + Interlocked.Decrement(ref this.activePackets); float rt = packet.Stopwatch.ElapsedMilliseconds; packet.AckCallback?.Invoke(); @@ -454,11 +461,12 @@ namespace Hazel.Udp void DisposeReliablePackets() { - var keys = this.reliableDataPacketsSent.Keys.ToArray(); - foreach (var k in keys) + this.reliableTimer.Dispose(); + + foreach (var kvp in reliableDataPacketsSent) { Packet pkt; - if (this.reliableDataPacketsSent.TryRemove(k, out pkt)) + if (this.reliableDataPacketsSent.TryRemove(kvp.Key, out pkt)) { pkt.Recycle(); } diff --git a/Hazel/Udp/UdpConnection.cs b/Hazel/Udp/UdpConnection.cs index 0e98231..ca6a986 100644 --- a/Hazel/Udp/UdpConnection.cs +++ b/Hazel/Udp/UdpConnection.cs @@ -21,6 +21,7 @@ namespace Hazel.Udp protected UdpConnection() { InitializeKeepAliveTimer(); + InitializeReliableTimer(); } /// diff --git a/Hazel/Udp/UdpConnectionListener.cs b/Hazel/Udp/UdpConnectionListener.cs index 6252e58..b4d6261 100644 --- a/Hazel/Udp/UdpConnectionListener.cs +++ b/Hazel/Udp/UdpConnectionListener.cs @@ -15,6 +15,9 @@ namespace Hazel.Udp /// public class UdpConnectionListener : NetworkConnectionListener { + public long BytesReceived; + public long BytesSent; + public const int BufferSize = ushort.MaxValue / 4; public int MinConnectionLength = 0; @@ -32,25 +35,14 @@ namespace Hazel.Udp ConcurrentDictionary allConnections = new ConcurrentDictionary(); public int ConnectionCount { get { return this.allConnections.Count; } } - /// - /// Creates a new UdpConnectionListener for the given , port and . - /// - /// The IPAddress to listen on. - /// The port to listen on. - /// The to listen with. - [Obsolete("Temporary constructor in beta only, use NetworkEndPoint constructor instead.")] - public UdpConnectionListener(IPAddress IPAddress, int port, Action logger, IPMode mode = IPMode.IPv4) - : this (new NetworkEndPoint(IPAddress, port, mode)) - { - this.Logger = logger; - } /// /// Creates a new UdpConnectionListener for the given , port and . /// /// The endpoint to listen on. - public UdpConnectionListener(NetworkEndPoint endPoint) + public UdpConnectionListener(NetworkEndPoint endPoint, Action logger = null) { + this.Logger = logger; this.EndPoint = endPoint.EndPoint; this.IPMode = endPoint.IPMode; @@ -135,6 +127,8 @@ namespace Hazel.Udp { Interlocked.Decrement(ref ActiveListeners); bytesReceived = listener.EndReceiveFrom(result, ref remoteEndPoint); + Interlocked.Add(ref BytesReceived, bytesReceived); + message.Offset = 0; message.Length = bytesReceived; } @@ -240,6 +234,7 @@ namespace Hazel.Udp internal void SendData(byte[] bytes, int length, EndPoint endPoint) { if (length > bytes.Length) return; + Interlocked.Add(ref BytesSent, length); try { -- 2.39.5