From: JamJar00 Date: Thu, 28 Apr 2016 21:18:38 +0000 (+0100) Subject: Fixed reliable sends and added keepalive X-Git-Tag: 1.0.0~164 X-Git-Url: https://git.deb.at/?a=commitdiff_plain;h=2acbb246b9743d8c004dc777b4af770dcdf8bebc;p=rhonda%2Fimpostor.hazel.git Fixed reliable sends and added keepalive --- diff --git a/Hazel.UnitTests/TcpConnectionTests.cs b/Hazel.UnitTests/TcpConnectionTests.cs index cde4049..290eeae 100644 --- a/Hazel.UnitTests/TcpConnectionTests.cs +++ b/Hazel.UnitTests/TcpConnectionTests.cs @@ -40,7 +40,7 @@ namespace Hazel.UnitTests using (TcpConnectionListener listener = new TcpConnectionListener(IPAddress.Any, 4296)) using (TcpConnection connection = new TcpConnection()) { - TestHelper.RunServerToClientTest(listener, connection, 4, 0, 0, SendOption.OrderedFragmentedReliable); + TestHelper.RunServerToClientTest(listener, connection, 4, 0, SendOption.OrderedFragmentedReliable); } } @@ -53,7 +53,7 @@ namespace Hazel.UnitTests using (TcpConnectionListener listener = new TcpConnectionListener(IPAddress.Any, 4296)) using (TcpConnection connection = new TcpConnection()) { - TestHelper.RunClientToServerTest(listener, connection, 4, 0, 0, SendOption.OrderedFragmentedReliable); + TestHelper.RunClientToServerTest(listener, connection, 4, 0, SendOption.OrderedFragmentedReliable); } } } diff --git a/Hazel.UnitTests/TestHelper.cs b/Hazel.UnitTests/TestHelper.cs index 1c4214f..02cc113 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 headerSize, int handshakeSize, int totalHandshakeSize, SendOption sendOption) + internal static void RunServerToClientTest(ConnectionListener listener, Connection connection, int headerSize, int totalHandshakeSize, SendOption sendOption) { //Setup meta stuff byte[] data = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 }; @@ -25,11 +25,13 @@ namespace Hazel.UnitTests //Setup listener listener.NewConnection += delegate(object sender, NewConnectionEventArgs args) { + Assert.AreEqual(0, args.Connection.Statistics.DataBytesReceived); + Assert.AreEqual(0, args.Connection.Statistics.TotalBytesReceived); + args.Connection.WriteBytes(data, sendOption); + Assert.AreEqual(data.Length, args.Connection.Statistics.DataBytesSent); - Assert.AreEqual(0, args.Connection.Statistics.DataBytesReceived); Assert.AreEqual(data.Length + headerSize, args.Connection.Statistics.TotalBytesSent); - Assert.AreEqual(0, args.Connection.Statistics.TotalBytesReceived); }; listener.Start(); @@ -54,7 +56,7 @@ namespace Hazel.UnitTests //Wait until data is received mutex.WaitOne(); - Assert.AreEqual(handshakeSize, connection.Statistics.DataBytesSent); + Assert.AreEqual(0, connection.Statistics.DataBytesSent); Assert.AreEqual(data.Length, connection.Statistics.DataBytesReceived); Assert.AreEqual(totalHandshakeSize, connection.Statistics.TotalBytesSent); Assert.AreEqual(data.Length + headerSize, connection.Statistics.TotalBytesReceived); @@ -65,7 +67,7 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunClientToServerTest(ConnectionListener listener, Connection connection, int headerSize, int handshakeSize, int totalHandshakeSize, SendOption sendOption) + internal static void RunClientToServerTest(ConnectionListener listener, Connection connection, int headerSize, int totalHandshakeSize, SendOption sendOption) { //Setup meta stuff byte[] data = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 }; @@ -103,10 +105,10 @@ namespace Hazel.UnitTests //Wait until data is received mutex.WaitOne(); - Assert.AreEqual(data.Length + handshakeSize, connection.Statistics.DataBytesSent); + Assert.AreEqual(data.Length, connection.Statistics.DataBytesSent); Assert.AreEqual(0, connection.Statistics.DataBytesReceived); Assert.AreEqual(totalHandshakeSize + data.Length + headerSize, connection.Statistics.TotalBytesSent); - Assert.AreEqual(sendOption == SendOption.Reliable ? 3 : 0, connection.Statistics.TotalBytesReceived); + Assert.AreEqual(0, connection.Statistics.TotalBytesReceived); } } } diff --git a/Hazel.UnitTests/UdpConnectionTests.cs b/Hazel.UnitTests/UdpConnectionTests.cs index 0134cdc..e2abd18 100644 --- a/Hazel.UnitTests/UdpConnectionTests.cs +++ b/Hazel.UnitTests/UdpConnectionTests.cs @@ -1,6 +1,7 @@ using System; using Microsoft.VisualStudio.TestTools.UnitTesting; using System.Net; +using System.Threading; namespace Hazel.UnitTests { @@ -26,7 +27,7 @@ namespace Hazel.UnitTests //UdpConnection fields Assert.AreEqual(new IPEndPoint(IPAddress.Loopback, 4296), connection.RemoteEndPoint); - Assert.AreEqual(1, connection.Statistics.DataBytesSent); + Assert.AreEqual(0, connection.Statistics.DataBytesSent); Assert.AreEqual(0, connection.Statistics.DataBytesReceived); } } @@ -40,7 +41,7 @@ namespace Hazel.UnitTests using (UdpConnectionListener listener = new UdpConnectionListener(IPAddress.Any, 4296)) using (UdpConnection connection = new UdpClientConnection()) { - TestHelper.RunServerToClientTest(listener, connection, 1, 1, 2, SendOption.None); + TestHelper.RunServerToClientTest(listener, connection, 1, 3, SendOption.None); } } @@ -53,7 +54,7 @@ namespace Hazel.UnitTests using (UdpConnectionListener listener = new UdpConnectionListener(IPAddress.Any, 4296)) using (UdpConnection connection = new UdpClientConnection()) { - TestHelper.RunServerToClientTest(listener, connection, 3, 1, 2, SendOption.Reliable); + TestHelper.RunServerToClientTest(listener, connection, 3, 3, SendOption.Reliable); } } @@ -66,7 +67,7 @@ namespace Hazel.UnitTests using (UdpConnectionListener listener = new UdpConnectionListener(IPAddress.Any, 4296)) using (UdpConnection connection = new UdpClientConnection()) { - TestHelper.RunClientToServerTest(listener, connection, 1, 1, 2, SendOption.None); + TestHelper.RunClientToServerTest(listener, connection, 1, 3, SendOption.None); } } @@ -79,7 +80,56 @@ namespace Hazel.UnitTests using (UdpConnectionListener listener = new UdpConnectionListener(IPAddress.Any, 4296)) using (UdpConnection connection = new UdpClientConnection()) { - TestHelper.RunClientToServerTest(listener, connection, 3, 1, 2, SendOption.Reliable); + TestHelper.RunClientToServerTest(listener, connection, 3, 3, SendOption.Reliable); + } + } + + /// + /// Tests the keepalive functionality from the client, + /// + [TestMethod] + public void KeepAliveClientTest() + { + using (UdpConnectionListener listener = new UdpConnectionListener(IPAddress.Any, 4296)) + using (UdpConnection connection = new UdpClientConnection()) + { + listener.Start(); + + connection.Connect(new NetworkEndPoint(IPAddress.Loopback, 4296)); + connection.KeepAliveInterval = 100; + + System.Threading.Thread.Sleep(1100); //Enough time for 10 keep alive packets + + Assert.AreEqual(33, connection.Statistics.TotalBytesSent); + } + } + + /// + /// Tests the keepalive functionality from the client, + /// + [TestMethod] + public void KeepAliveServerTest() + { + ManualResetEvent mutex = new ManualResetEvent(false); + + using (UdpConnectionListener listener = new UdpConnectionListener(IPAddress.Any, 4296)) + using (UdpConnection connection = new UdpClientConnection()) + { + listener.NewConnection += delegate(object sender, NewConnectionEventArgs args) + { + ((UdpConnection)args.Connection).KeepAliveInterval = 100; + + Thread.Sleep(1100); //Enough time for 10 keep alive packets + + Assert.AreEqual(30, args.Connection.Statistics.TotalBytesSent); + mutex.Set(); + }; + + listener.Start(); + + connection.Connect(new NetworkEndPoint(IPAddress.Loopback, 4296)); + + mutex.WaitOne(); } } } diff --git a/Hazel/Connection.cs b/Hazel/Connection.cs index 7350be6..d6c4c83 100644 --- a/Hazel/Connection.cs +++ b/Hazel/Connection.cs @@ -4,6 +4,7 @@ using System.Linq; using System.Text; using System.Net.Sockets; using System.Net; +using System.Threading; /* @@ -43,9 +44,30 @@ namespace Hazel /// /// The state of this connection. /// - public ConnectionState State { get { return state; } protected set { state = value; } } + public ConnectionState State + { + get + { + return state; + } + + protected set + { + state = value; + + if (state == ConnectionState.Connected) + connectWaitLock.Set(); + else + connectWaitLock.Reset(); + } + } volatile ConnectionState state; + /// + /// Reset event that is triggered when the connection is marked Connected. + /// + ManualResetEvent connectWaitLock = new ManualResetEvent(false); + /// /// Constructor that initializes the ConnecitonStatistics object. /// @@ -97,6 +119,14 @@ namespace Hazel handler(this, args); } + /// + /// Blocks until the Connection is connected. + /// + protected void WaitOnConnect() + { + connectWaitLock.WaitOne(); + } + /// /// Closes this connections safely. /// diff --git a/Hazel/Hazel.csproj b/Hazel/Hazel.csproj index 38f4f3d..efadf84 100644 --- a/Hazel/Hazel.csproj +++ b/Hazel/Hazel.csproj @@ -65,6 +65,7 @@ Code + diff --git a/Hazel/SendOptionInternal.cs b/Hazel/SendOptionInternal.cs index 335d581..5e4f621 100644 --- a/Hazel/SendOptionInternal.cs +++ b/Hazel/SendOptionInternal.cs @@ -11,6 +11,14 @@ namespace Hazel /// enum SendOptionInternal : byte { + /// + /// Hello message for initiating communication. + /// + Hello = 254, + + /// + /// Message acknowledging the receipt of a message. + /// Acknowledgement = 255 } } diff --git a/Hazel/TcpConnection.cs b/Hazel/TcpConnection.cs index 4ad2db2..f39514b 100644 --- a/Hazel/TcpConnection.cs +++ b/Hazel/TcpConnection.cs @@ -63,6 +63,22 @@ namespace Hazel Socket.NoDelay = true; } + /// + /// Internal call to start listening once this socket has been constructed and is ready. + /// + internal void StartListening() + { + //Start receiving data + try + { + StartWaitingForHeader(); + } + catch (SocketException e) + { + throw new HazelException("A Socket exception occured while initiating a receive operation.", e); + } + } + /// /// Connects this TCP connection to the endpoint. /// diff --git a/Hazel/TcpConnectionListener.cs b/Hazel/TcpConnectionListener.cs index 1c01ad6..09af39e 100644 --- a/Hazel/TcpConnectionListener.cs +++ b/Hazel/TcpConnectionListener.cs @@ -98,6 +98,8 @@ namespace Hazel NewConnectionEventArgs args = new NewConnectionEventArgs(tcpConnection); FireNewConnectionEvent(args); + + tcpConnection.StartListening(); } } diff --git a/Hazel/UdpClientConnection.cs b/Hazel/UdpClientConnection.cs index fcdc917..b5516c4 100644 --- a/Hazel/UdpClientConnection.cs +++ b/Hazel/UdpClientConnection.cs @@ -25,6 +25,7 @@ namespace Hazel /// Creates a new UdpClientConnection. /// public UdpClientConnection() + : base() { socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); } @@ -36,8 +37,11 @@ namespace Hazel /// The option this data is requested to send with. public override void WriteBytes(byte[] bytes, SendOption sendOption = SendOption.None) { + if (State != ConnectionState.Connected) + throw new InvalidOperationException("Could not send data as this Connection is not connected. Did you disconnect?"); + //Add header information and send - HandleSend(bytes, sendOption); + HandleSend(bytes, (byte)sendOption); } /// @@ -53,8 +57,8 @@ namespace Hazel lock (socket) { - if (State != ConnectionState.Connected) - throw new InvalidOperationException("Could not send data as this Connection is not connected. Did you disconnect?"); + if (State != ConnectionState.Connected && State != ConnectionState.Connecting) + throw new InvalidOperationException("Could not send data as this Connection is not connected and is not connecting. Did you disconnect?"); try { @@ -82,7 +86,7 @@ namespace Hazel NetworkEndPoint nep = remoteEndPoint as NetworkEndPoint; if (nep == null) { - throw new ArgumentException("The remote end point of a TCP connection must be a NetworkEndPoint."); + throw new ArgumentException("The remote end point of a UDP connection must be a NetworkEndPoint."); } this.EndPoint = nep; @@ -118,12 +122,14 @@ namespace Hazel { throw new HazelException("A Socket exception occured while initiating a receive operation.", e); } - - State = ConnectionState.Connected; } - //Write bytes to the server to tell it hi (and to punch a hole in our NAT, if present). - WriteBytes(new byte[] { 0 }, SendOption.None); //TODO special hello message + //Write bytes to the server to tell it hi (and to punch a hole in our NAT, if present) + //When acknowledged set the state to connected + SendHello(() => State = ConnectionState.Connected); + + //Wait till hello packet is acknowledged and the state is set to Connected + WaitOnConnect(); } /// diff --git a/Hazel/UdpConnection.KeepAlive.cs b/Hazel/UdpConnection.KeepAlive.cs new file mode 100644 index 0000000..1531874 --- /dev/null +++ b/Hazel/UdpConnection.KeepAlive.cs @@ -0,0 +1,87 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace Hazel +{ + /// + /// UdpConnection part which handles keepalive packets. + /// + partial class UdpConnection + { + /// + /// The interval from data being received or transmitted to a keepalive packet being sent. + /// + /// + /// Set to System.Threading.Timeout.Infinite to disable keepalive packets. + /// + public int KeepAliveInterval + { + get + { + return keepAliveInterval; + } + + set + { + keepAliveInterval = value; + + //Update timer + ResetKeepAliveTimer(); + } + } + int keepAliveInterval = 10000; + + /// + /// The timer creating keepalive pulses. + /// + Timer keepAliveTimer; + + /// + /// Lock for keep alive timer. + /// + Object keepAliveTimerLock = new Object(); + + /// + /// Starts the keepalive timer. + /// + void InitializeKeepAliveTimer() + { + lock (keepAliveTimerLock) + { + keepAliveTimer = new Timer( + (o) => + { + Trace.WriteLine("Keepalive packet sent."); + SendHello(null); + }, + null, + keepAliveInterval, + keepAliveInterval + ); + } + } + + /// + /// Resets the keepalive timer to zero. + /// + void ResetKeepAliveTimer() + { + lock (keepAliveTimerLock) + keepAliveTimer.Change(keepAliveInterval, keepAliveInterval); + } + + /// + /// Disposes of the keep alive timer. + /// + void DisposeKeepAliveTimer() + { + lock(keepAliveTimerLock) + keepAliveTimer.Dispose(); + } + } +} diff --git a/Hazel/UdpConnection.Reliable.cs b/Hazel/UdpConnection.Reliable.cs index 36baf3c..fd53ab7 100644 --- a/Hazel/UdpConnection.Reliable.cs +++ b/Hazel/UdpConnection.Reliable.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Text; using System.Threading; @@ -56,8 +57,9 @@ namespace Hazel public byte[] Data; public Timer Timer; public int LastTimeout; + public Action AckCallback; - public Packet(byte[] data, Action resendAction, int timeout) + public Packet(byte[] data, Action resendAction, int timeout, Action ackCallback) { Data = data; @@ -65,10 +67,11 @@ namespace Hazel (object obj) => resendAction(this), null, timeout, - timeout + Timeout.Infinite ); LastTimeout = timeout; + AckCallback = ackCallback; } } @@ -76,7 +79,7 @@ namespace Hazel /// Writes the bytes neccessary for a reliable send and stores the send. /// /// The byte array to write to. - void WriteReliableSendHeader(byte[] bytes) + void WriteReliableSendHeader(byte[] bytes, Action ackCallback) { lock (reliableDataPacketsSent) { @@ -99,9 +102,13 @@ namespace Hazel WriteBytesToConnection(p.Data); //Double packet timeout - p.Timer.Change(0, p.LastTimeout *= 2); + lock (p.Timer) + p.Timer.Change(p.LastTimeout *= 2, Timeout.Infinite); + + Trace.WriteLine("Resend."); }, - resendTimeout + resendTimeout, + ackCallback ); //Remember packet @@ -119,15 +126,8 @@ namespace Hazel //Get the ID form the packet ushort id = (ushort)((bytes[1] << 8) + bytes[2]); - //Always reply with acknowledgement in order to stop the sender repeatedly sending it - WriteBytesToConnection( //TODO group acks together - new byte[] - { - (byte)SendOptionInternal.Acknowledgement, - bytes[1], - bytes[2] - } - ); + //Send an acknowledgement + SendAck(bytes[1], bytes[2]); //Handle reliableness! lock (reliableDataPacketsMissing) @@ -174,10 +174,30 @@ namespace Hazel //Dispose of timer and remove from dictionary if (reliableDataPacketsSent.ContainsKey(id)) { - reliableDataPacketsSent[id].Timer.Dispose(); + Packet packet = reliableDataPacketsSent[id]; + + lock (packet.Timer) + packet.Timer.Dispose(); + + if (packet.AckCallback != null) + packet.AckCallback.Invoke(); + reliableDataPacketsSent.Remove(id); } } } + + internal void SendAck(byte byte1, byte byte2) + { + //Always reply with acknowledgement in order to stop the sender repeatedly sending it + WriteBytesToConnection( //TODO group acks together + new byte[] + { + (byte)SendOptionInternal.Acknowledgement, + byte1, + byte2 + } + ); + } } } diff --git a/Hazel/UdpConnection.cs b/Hazel/UdpConnection.cs index 09936fb..557e7dc 100644 --- a/Hazel/UdpConnection.cs +++ b/Hazel/UdpConnection.cs @@ -31,20 +31,32 @@ namespace Hazel /// The bytes to write. protected abstract void WriteBytesToConnection(byte[] bytes); + protected UdpConnection() + { + InitializeKeepAliveTimer(); + } + /// /// Handles the reliable/fragmented/ordered sending from this connection. /// /// The data being sent. - /// The send option. + /// The send option as a byte. /// The bytes that should actually be sent. - protected void HandleSend(byte[] data, SendOption sendOption) + protected void HandleSend(byte[] data, byte sendOption, Action ackCallback = null) { byte[] bytes; switch (sendOption) { - case SendOption.Reliable: + //Handle reliable header + case (byte)SendOption.Reliable: bytes = new byte[data.Length + 3]; - WriteReliableSendHeader(bytes); + WriteReliableSendHeader(bytes, ackCallback); + break; + + //Handle hellos (ignore data) + case (byte)SendOptionInternal.Hello: + bytes = new byte[3]; + WriteReliableSendHeader(bytes, ackCallback); break; default: @@ -53,14 +65,17 @@ namespace Hazel } //Add message type - bytes[0] = (byte)sendOption; + bytes[0] = sendOption; //Copy data into new array Buffer.BlockCopy(data, 0, bytes, bytes.Length - data.Length, data.Length); + //Inform keepalive not to send for a while + ResetKeepAliveTimer(); //TODO keepalive tests + //Write to connection WriteBytesToConnection(bytes); - + Statistics.LogSend(data.Length, bytes.Length); } @@ -72,9 +87,13 @@ namespace Hazel /// The bytes of data received. protected byte[] HandleReceive(byte[] buffer, int bytesReceived) { + //Inform keepalive not to send for a while + ResetKeepAliveTimer(); + int headerSize = 1; switch (buffer[0]) { + //Handle reliable receives case (byte)SendOption.Reliable: headerSize = 3; @@ -82,11 +101,17 @@ namespace Hazel return null; break; + //Handle acknowledgments case (byte)SendOptionInternal.Acknowledgement: HandleAcknowledgement(buffer); - - Statistics.LogReceive(0, bytesReceived); - + + return null; + + //We need to acknowledge hello messages so just use the same reliable receive + //method + case (byte)SendOptionInternal.Hello: + HandleReliableReceive(buffer); + return null; } @@ -97,5 +122,28 @@ namespace Hazel return dataBytes; } + + /// + /// Sends a hello packet to the remote endpoint. + /// + /// The callback to invoke when the hello packet is acknowledged. + protected void SendHello(Action acknowledgeCallback) + { + HandleSend(new byte[0], (byte)SendOptionInternal.Hello, acknowledgeCallback); + } + + /// + /// Called when things are being disposed of + /// + /// + protected override void Dispose(bool disposing) + { + if (disposing) + { + DisposeKeepAliveTimer(); + } + + base.Dispose(disposing); + } } } diff --git a/Hazel/UdpConnectionListener.cs b/Hazel/UdpConnectionListener.cs index f820e76..cd362f7 100644 --- a/Hazel/UdpConnectionListener.cs +++ b/Hazel/UdpConnectionListener.cs @@ -142,6 +142,9 @@ namespace Hazel { connection = new UdpServerConnection(this, remoteEndPoint); connections.Add(remoteEndPoint, connection); + + //Then ping back an ack to make sure they're happy + connection.SendAck(buffer[1], buffer[2]); } } diff --git a/Hazel/UdpServerConnection.cs b/Hazel/UdpServerConnection.cs index ee9f95f..07735bf 100644 --- a/Hazel/UdpServerConnection.cs +++ b/Hazel/UdpServerConnection.cs @@ -31,6 +31,7 @@ namespace Hazel /// /// internal UdpServerConnection(UdpConnectionListener listener, EndPoint endPoint) + : base() { this.Listener = listener; this.RemoteEndPoint = endPoint; @@ -46,7 +47,7 @@ namespace Hazel /// The option this data is requested to send with. public override void WriteBytes(byte[] bytes, SendOption sendOption = SendOption.None) { - HandleSend(bytes, sendOption); + HandleSend(bytes, (byte)sendOption); } ///