From: Forest Date: Thu, 7 Mar 2019 04:00:02 +0000 (-0800) Subject: Clean up some stuff, fix a thing where 0 retries wasn't infinite X-Git-Tag: 1.0.0~54 X-Git-Url: https://git.deb.at/?a=commitdiff_plain;h=adf9b30b2c74f382c226055a80d616b5016bad77;p=rhonda%2Fimpostor.hazel.git Clean up some stuff, fix a thing where 0 retries wasn't infinite --- diff --git a/Hazel/Connection.cs b/Hazel/Connection.cs index 998fe6a..964e82d 100644 --- a/Hazel/Connection.cs +++ b/Hazel/Connection.cs @@ -51,6 +51,8 @@ namespace Hazel public event Action DataReceived; public int TestLagMs = -1; + public int TestDropRate = 0; + protected int testDropCount = 0; /// /// Called when the end point disconnects or an error occurs. diff --git a/Hazel/MessageWriter.cs b/Hazel/MessageWriter.cs index bd00eb5..54f0dd1 100644 --- a/Hazel/MessageWriter.cs +++ b/Hazel/MessageWriter.cs @@ -57,8 +57,8 @@ namespace Hazel } case SendOption.Tcp: { - byte[] output = new byte[this.Length - 4]; - System.Buffer.BlockCopy(this.Buffer, 4, output, 0, this.Length - 4); + byte[] output = new byte[this.Length]; + System.Buffer.BlockCopy(this.Buffer, 0, output, 0, this.Length); return output; } } @@ -113,9 +113,8 @@ namespace Hazel public void Clear(SendOption sendOption) { - this.Position = this.Length = 0; + this.messageStarts.Clear(); this.SendOption = sendOption; - this.Buffer[0] = (byte)sendOption; switch (sendOption) { @@ -126,6 +125,7 @@ namespace Hazel this.Length = this.Position = 3; break; case SendOption.Tcp: + this.Length = this.Position = 0; break; } } diff --git a/Hazel/Tcp/TcpConnection.cs b/Hazel/Tcp/TcpConnection.cs index 8c350d9..fc63ae7 100644 --- a/Hazel/Tcp/TcpConnection.cs +++ b/Hazel/Tcp/TcpConnection.cs @@ -84,7 +84,9 @@ namespace Hazel.Tcp //Start receiving data try { - ListenForData(InvokeAndListen); + var msg = MessageReader.GetSized(ushort.MaxValue); + + ListenForData(msg, InvokeAndListen); } catch (Exception e) { @@ -170,31 +172,39 @@ namespace Hazel.Tcp { this.State = ConnectionState.Connected; + var buffer = MessageReader.GetSized(ushort.MaxValue); try { - ListenForData( - delegate (MessageReader msg) - { - ListenForData(InvokeAndListen); - - //Remove version byte - msg.Offset = 1; - msg.Length -= 1; - msg.Position = 0; + buffer.Offset = 0; + buffer.Length = 4; + buffer.Position = 0; - callback.Invoke(msg); - } + ListenForData( + buffer, + m => ReadHeader(m, + delegate (MessageReader msg) + { + ListenForData(); + + //Remove version byte + msg.Offset = 1; + msg.Length -= 1; + msg.Position = 0; + + callback.Invoke(msg); + }) ); } catch (Exception e) { + buffer.Recycle(); Disconnect("An exception occured while initiating the first receive operation: " + e.Message); } } private void InvokeAndListen(MessageReader msg) { - this.ListenForData(InvokeAndListen); + this.ListenForData(); try { @@ -202,58 +212,42 @@ namespace Hazel.Tcp } catch { } } - - private void ListenForData(Action callback) + + private void ListenForData() { - if (State == ConnectionState.Disconnecting || State == ConnectionState.NotConnected) - throw new HazelException("Not connected"); - var msg = MessageReader.GetSized(ushort.MaxValue); - try - { - socket.BeginReceive(msg.Buffer, 0, 4, SocketFlags.None, o => HeaderReadCallback(callback, o), msg); - } - catch (SocketException s) - { - Disconnect("SocketException while reading header: " + s.Message); - } + msg.Offset = 0; + msg.Length = 4; + msg.Position = 0; + + ListenForData(msg, m => ReadHeader(m, null)); } - private void HeaderReadCallback(Action callback, IAsyncResult result) + private void ReadHeader(MessageReader msg, Action callback) { - int bytesRead; - try - { - bytesRead = socket.EndReceive(result); - if (bytesRead == 0) - { - Disconnect("Received 0 bytes"); - return; - } - - Statistics.LogFragmentedReceive(0, bytesRead); - } - catch (SocketException s) - { - Disconnect("SocketException while reading header: " + s.Message); - return; - } - - // TODO: Could possibly fragment here... - var msg = (MessageReader)result.AsyncState; msg.Length = GetLengthFromBytes(msg.Buffer); + msg.Position = 0; + + ListenForData(msg, callback ?? InvokeAndListen); + } + private void ListenForData(MessageReader msg, Action callback) + { + if (State == ConnectionState.Disconnecting || State == ConnectionState.NotConnected) + throw new HazelException("Not connected"); + try { - socket.BeginReceive(msg.Buffer, 0, msg.Length, SocketFlags.None, o => BodyReadCallback(callback, o), msg); + socket.BeginReceive(msg.Buffer, msg.Position, msg.Length, SocketFlags.None, o => ReadUntilFull(callback, o), msg); } catch (SocketException s) { - Disconnect("SocketException while reading body: " + s.Message); + msg.Recycle(); + Disconnect("SocketException while reading header: " + s.Message); } } - - private void BodyReadCallback(Action callback, IAsyncResult result) + + private void ReadUntilFull(Action callback, IAsyncResult result) { int bytesRead; try @@ -278,14 +272,7 @@ namespace Hazel.Tcp if (msg.Position < bytesRead) { - try - { - socket.BeginReceive(msg.Buffer, msg.Position, msg.Length - msg.Position, SocketFlags.None, o => BodyReadCallback(callback, o), msg); - } - catch (SocketException s) - { - Disconnect("SocketException while reading body: " + s.Message); - } + ListenForData(msg, callback); } else { diff --git a/Hazel/Udp/UdpBroadcastListener.cs b/Hazel/Udp/UdpBroadcastListener.cs index 5304332..fb8c52a 100644 --- a/Hazel/Udp/UdpBroadcastListener.cs +++ b/Hazel/Udp/UdpBroadcastListener.cs @@ -97,7 +97,7 @@ namespace Hazel.Udp } IPEndPoint ipEnd = (IPEndPoint)endpt; - string data = ASCIIEncoding.ASCII.GetString(buffer, 2, numBytes - 2); + string data = UTF8Encoding.UTF8.GetString(buffer, 2, numBytes - 2); int dataHash = data.GetHashCode(); lock (packets) @@ -139,7 +139,21 @@ namespace Hazel.Udp { if (this.socket != null) { - this.socket.Close(); + try + { + this.socket.Shutdown(SocketShutdown.Both); + } + catch { } + try + { + this.socket.Close(); + } + catch { } + try + { + this.socket.Dispose(); + } + catch { } this.socket = null; } } diff --git a/Hazel/Udp/UdpBroadcaster.cs b/Hazel/Udp/UdpBroadcaster.cs index 872e685..d0a9f84 100644 --- a/Hazel/Udp/UdpBroadcaster.cs +++ b/Hazel/Udp/UdpBroadcaster.cs @@ -28,12 +28,12 @@ namespace Hazel.Udp /// public void SetData(string data) { - int len = ASCIIEncoding.ASCII.GetByteCount(data); + int len = UTF8Encoding.UTF8.GetByteCount(data); this.data = new byte[len + 2]; this.data[0] = 4; this.data[1] = 2; - ASCIIEncoding.ASCII.GetBytes(data, 0, data.Length, this.data, 2); + UTF8Encoding.UTF8.GetBytes(data, 0, data.Length, this.data, 2); } /// @@ -52,7 +52,21 @@ namespace Hazel.Udp { if (this.socket != null) { - this.socket.Close(); + try + { + this.socket.Shutdown(SocketShutdown.Both); + } + catch { } + try + { + this.socket.Close(); + } + catch { } + try + { + this.socket.Dispose(); + } + catch { } this.socket = null; } } diff --git a/Hazel/Udp/UdpClientConnection.cs b/Hazel/Udp/UdpClientConnection.cs index a76bdeb..58a16be 100644 --- a/Hazel/Udp/UdpClientConnection.cs +++ b/Hazel/Udp/UdpClientConnection.cs @@ -18,14 +18,14 @@ namespace Hazel.Udp /// /// The socket we're connected via. /// - Socket socket; + private Socket socket; /// /// The buffer to store incomming data in. /// - byte[] dataBuffer = new byte[ushort.MaxValue]; + private byte[] dataBuffer = new byte[ushort.MaxValue]; - Timer reliablePacketTimer; + private Timer reliablePacketTimer; /// /// Creates a new UdpClientConnection. @@ -76,7 +76,6 @@ namespace Hazel.Udp } } - public event Action DataSentRaw; public event Action DataReceivedRaw; @@ -267,6 +266,14 @@ namespace Hazel.Udp Thread.Sleep(this.TestLagMs); } + if (this.TestDropRate > 0) + { + if ((this.testDropCount++ % this.TestDropRate) == 0) + { + return; + } + } + DataReceivedRaw?.Invoke(bytes); MessageReader msg = MessageReader.GetRaw(bytes, 0, bytesReceived); HandleReceive(msg, bytesReceived); diff --git a/Hazel/Udp/UdpConnection.Reliable.cs b/Hazel/Udp/UdpConnection.Reliable.cs index 68e1d92..1b0c342 100644 --- a/Hazel/Udp/UdpConnection.Reliable.cs +++ b/Hazel/Udp/UdpConnection.Reliable.cs @@ -25,11 +25,17 @@ namespace Hazel.Udp /// resulting in a more dynamic resend that responds to endpoints on slower or faster connections. /// /// - public int ResendTimeout { get { return resendTimeout; } set { resendTimeout = value; } } - private volatile int resendTimeout = 0; + public volatile int ResendTimeout = 0; + /// + /// Max number of times to resend. 0 == no limit + /// public volatile int ResendLimit = 0; + /// + /// A compounding multiplier to back off resend timeout. + /// Applied to ping before first timeout when ResendTimeout == 0. + /// public volatile float ResendPingMultiplier = 3; /// @@ -40,24 +46,24 @@ namespace Hazel.Udp /// /// The packets of data that have been transmitted reliably and not acknowledged. /// - ConcurrentDictionary reliableDataPacketsSent = new ConcurrentDictionary(); + internal ConcurrentDictionary reliableDataPacketsSent = new ConcurrentDictionary(); /// /// The last packets that were received. /// - HashSet reliableDataPacketsMissing = new HashSet(); + private HashSet reliableDataPacketsMissing = new HashSet(); /// /// The packet id that was received last. /// - volatile ushort reliableReceiveLast = 0; - + private volatile ushort reliableReceiveLast = 0; + /// /// Has the connection received anything yet /// - volatile bool hasReceivedSomething = false; + private volatile bool hasReceivedSomething = false; - object PingLock = new object(); + private object PingLock = new object(); /// /// Returns the average ping to this endpoint. @@ -77,7 +83,7 @@ namespace Hazel.Udp /// connection will be marked as disconnected and the Disconnected event /// will be invoked. /// - public volatile int DisconnectTimeout = 2500; + public volatile int DisconnectTimeout = 5000; /// /// Class to hold packet data @@ -110,11 +116,11 @@ namespace Hazel.Udp public int Retransmissions; public Stopwatch Stopwatch = new Stopwatch(); - + Packet() { } - + internal void Set(ushort id, UdpConnection connection, byte[] data, int length, int timeout, Action ackCallback) { this.Id = id; @@ -151,7 +157,9 @@ namespace Hazel.Udp if (lifetime >= this.NextTimeout) { - if (++this.Retransmissions > connection.ResendLimit) + ++this.Retransmissions; + if (connection.ResendLimit != 0 + && this.Retransmissions > connection.ResendLimit) { if (connection.reliableDataPacketsSent.TryRemove(this.Id, out Packet self)) { @@ -163,7 +171,7 @@ namespace Hazel.Udp return 0; } - this.NextTimeout = (int)Math.Min(this.NextTimeout * 3f, connection.DisconnectTimeout); + this.NextTimeout = (int)Math.Min(this.NextTimeout * connection.ResendPingMultiplier, connection.DisconnectTimeout); try { connection.WriteBytesToConnection(this.Data, this.Length); @@ -190,13 +198,12 @@ namespace Hazel.Udp PacketPool.PutObject(this); } } - + internal int ManageReliablePackets() { int output = 0; if (this.reliableDataPacketsSent.Count > 0) { - double minTimeout = int.MaxValue; foreach (var kvp in this.reliableDataPacketsSent) { Packet pkt = kvp.Value; @@ -206,8 +213,6 @@ namespace Hazel.Udp output += pkt.Resend(); } catch { } - - minTimeout = Math.Min(pkt.NextTimeout, minTimeout); } } @@ -238,7 +243,7 @@ namespace Hazel.Udp this, buffer, sendLength, - resendTimeout > 0 ? resendTimeout : (int)Math.Max(300, Math.Min(AveragePingMs * this.ResendPingMultiplier, 2000)), + ResendTimeout > 0 ? ResendTimeout : (int)Math.Max(300, Math.Min(AveragePingMs * this.ResendPingMultiplier, 2000)), ackCallback); if (!reliableDataPacketsSent.TryAdd(id, packet)) diff --git a/Hazel/Udp/UdpConnectionListener.cs b/Hazel/Udp/UdpConnectionListener.cs index 1058980..b38530d 100644 --- a/Hazel/Udp/UdpConnectionListener.cs +++ b/Hazel/Udp/UdpConnectionListener.cs @@ -67,21 +67,15 @@ namespace Hazel.Udp { this.Dispose(false); } - - public float AveragePacketsTime = 1; - - Stopwatch stopwatch = new Stopwatch(); + private void ManageReliablePackets(object state) { - stopwatch.Restart(); foreach (var kvp in this.allConnections) { var sock = kvp.Value; sock.ManageReliablePackets(); } - - this.AveragePacketsTime = this.AveragePacketsTime * .7f + stopwatch.ElapsedMilliseconds * .3f; - + this.reliablePacketTimer.Change(100, Timeout.Infinite); } @@ -336,6 +330,11 @@ namespace Hazel.Udp if (this.socket != null) { + try + { + this.socket.Shutdown(SocketShutdown.Both); + } + catch { } this.socket.Close(); this.socket.Dispose(); this.socket = null;