From d046adacd8c22926e5488dedf17981ed455d9b06 Mon Sep 17 00:00:00 2001 From: JamJar00 Date: Sat, 31 Dec 2016 16:28:47 +0000 Subject: [PATCH] Added fragmented messages --- Hazel.UnitTests/TcpConnectionTests.cs | 4 +- Hazel.UnitTests/TestHelper.cs | 21 ++- Hazel.UnitTests/UdpConnectionTests.cs | 34 ++++- Hazel/Hazel.csproj | 3 +- Hazel/SendOption.cs | 15 +- Hazel/{ => Udp}/SendOptionInternal.cs | 11 +- Hazel/Udp/UdpClientConnection.cs | 4 + Hazel/Udp/UdpConnection.Fragmented.cs | 198 ++++++++++++++++++++++++++ Hazel/Udp/UdpConnection.Reliable.cs | 51 ++++--- Hazel/Udp/UdpConnection.cs | 40 ++++-- Hazel/Udp/UdpConnectionListener.cs | 2 +- 11 files changed, 325 insertions(+), 58 deletions(-) rename Hazel/{ => Udp}/SendOptionInternal.cs (72%) create mode 100644 Hazel/Udp/UdpConnection.Fragmented.cs diff --git a/Hazel.UnitTests/TcpConnectionTests.cs b/Hazel.UnitTests/TcpConnectionTests.cs index fc61c9f..95b48d5 100644 --- a/Hazel.UnitTests/TcpConnectionTests.cs +++ b/Hazel.UnitTests/TcpConnectionTests.cs @@ -94,7 +94,7 @@ namespace Hazel.UnitTests using (TcpConnectionListener listener = new TcpConnectionListener(IPAddress.Any, 4296)) using (TcpConnection connection = new TcpConnection(new NetworkEndPoint(IPAddress.Loopback, 4296))) { - TestHelper.RunServerToClientTest(listener, connection, 4, 5, SendOption.FragmentedReliable); + TestHelper.RunServerToClientTest(listener, connection, 10, SendOption.FragmentedReliable); } } @@ -107,7 +107,7 @@ namespace Hazel.UnitTests using (TcpConnectionListener listener = new TcpConnectionListener(IPAddress.Any, 4296)) using (TcpConnection connection = new TcpConnection(new NetworkEndPoint(IPAddress.Loopback, 4296))) { - TestHelper.RunClientToServerTest(listener, connection, 4, 5, SendOption.FragmentedReliable); + TestHelper.RunClientToServerTest(listener, connection, 10, SendOption.FragmentedReliable); } } diff --git a/Hazel.UnitTests/TestHelper.cs b/Hazel.UnitTests/TestHelper.cs index a66dc6f..0bb1693 100644 --- a/Hazel.UnitTests/TestHelper.cs +++ b/Hazel.UnitTests/TestHelper.cs @@ -16,10 +16,10 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunServerToClientTest(ConnectionListener listener, Connection connection, int headerSize, int totalHandshakeSize, SendOption sendOption) + internal static void RunServerToClientTest(ConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) { //Setup meta stuff - byte[] data = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 }; + byte[] data = BuildData(dataSize); ManualResetEvent mutex = new ManualResetEvent(false); //Setup listener @@ -56,10 +56,10 @@ namespace Hazel.UnitTests /// /// The listener to test. /// The connection to test. - internal static void RunClientToServerTest(ConnectionListener listener, Connection connection, int headerSize, int totalHandshakeSize, SendOption sendOption) + internal static void RunClientToServerTest(ConnectionListener listener, Connection connection, int dataSize, SendOption sendOption) { //Setup meta stuff - byte[] data = new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 }; + byte[] data = BuildData(dataSize); ManualResetEvent mutex = new ManualResetEvent(false); ManualResetEvent mutex2 = new ManualResetEvent(false); @@ -152,5 +152,18 @@ namespace Hazel.UnitTests mutex2.WaitOne(); } + + /// + /// Builds new data of increaseing value bytes. + /// + /// The number of bytes to generate. + /// The data. + static byte[] BuildData(int dataSize) + { + byte[] data = new byte[dataSize]; + for (int i = 0; i < dataSize; i++) + data[i] = (byte)i; + return data; + } } } diff --git a/Hazel.UnitTests/UdpConnectionTests.cs b/Hazel.UnitTests/UdpConnectionTests.cs index 7d7b0a1..878fa5c 100644 --- a/Hazel.UnitTests/UdpConnectionTests.cs +++ b/Hazel.UnitTests/UdpConnectionTests.cs @@ -94,7 +94,7 @@ namespace Hazel.UnitTests using (UdpConnectionListener listener = new UdpConnectionListener(new NetworkEndPoint(IPAddress.Any, 4296))) using (UdpConnection connection = new UdpClientConnection(new NetworkEndPoint(IPAddress.Loopback, 4296))) { - TestHelper.RunServerToClientTest(listener, connection, 1, 4, SendOption.None); + TestHelper.RunServerToClientTest(listener, connection, 10, SendOption.None); } } @@ -107,7 +107,20 @@ namespace Hazel.UnitTests using (UdpConnectionListener listener = new UdpConnectionListener(new NetworkEndPoint(IPAddress.Any, 4296))) using (UdpConnection connection = new UdpClientConnection(new NetworkEndPoint(IPAddress.Loopback, 4296))) { - TestHelper.RunServerToClientTest(listener, connection, 3, 4, SendOption.Reliable); + TestHelper.RunServerToClientTest(listener, connection, 10, SendOption.Reliable); + } + } + + /// + /// Tests server to client reliable communication on the UdpConnection. + /// + [TestMethod] + public void UdpFragmentedServerToClientTest() + { + using (UdpConnectionListener listener = new UdpConnectionListener(new NetworkEndPoint(IPAddress.Any, 4296))) + using (UdpConnection connection = new UdpClientConnection(new NetworkEndPoint(IPAddress.Loopback, 4296))) + { + TestHelper.RunServerToClientTest(listener, connection, (int)(connection.FragmentSize * 9.5), SendOption.FragmentedReliable); } } @@ -120,7 +133,7 @@ namespace Hazel.UnitTests using (UdpConnectionListener listener = new UdpConnectionListener(new NetworkEndPoint(IPAddress.Any, 4296))) using (UdpConnection connection = new UdpClientConnection(new NetworkEndPoint(IPAddress.Loopback, 4296))) { - TestHelper.RunClientToServerTest(listener, connection, 1, 4, SendOption.None); + TestHelper.RunClientToServerTest(listener, connection, 10, SendOption.None); } } @@ -133,7 +146,20 @@ namespace Hazel.UnitTests using (UdpConnectionListener listener = new UdpConnectionListener(new NetworkEndPoint(IPAddress.Any, 4296))) using (UdpConnection connection = new UdpClientConnection(new NetworkEndPoint(IPAddress.Loopback, 4296))) { - TestHelper.RunClientToServerTest(listener, connection, 3, 4, SendOption.Reliable); + TestHelper.RunClientToServerTest(listener, connection, 10, SendOption.Reliable); + } + } + + /// + /// Tests server to client reliable communication on the UdpConnection. + /// + [TestMethod] + public void UdpFragmentedClientToServerTest() + { + using (UdpConnectionListener listener = new UdpConnectionListener(new NetworkEndPoint(IPAddress.Any, 4296))) + using (UdpConnection connection = new UdpClientConnection(new NetworkEndPoint(IPAddress.Loopback, 4296))) + { + TestHelper.RunClientToServerTest(listener, connection, (int)(connection.FragmentSize * 9.5), SendOption.FragmentedReliable); } } diff --git a/Hazel/Hazel.csproj b/Hazel/Hazel.csproj index 151a0aa..f382241 100644 --- a/Hazel/Hazel.csproj +++ b/Hazel/Hazel.csproj @@ -69,7 +69,7 @@ - + @@ -78,6 +78,7 @@ Code + diff --git a/Hazel/SendOption.cs b/Hazel/SendOption.cs index a81fded..e7ebec2 100644 --- a/Hazel/SendOption.cs +++ b/Hazel/SendOption.cs @@ -31,18 +31,7 @@ namespace Hazel /// a larger number of protocol bytes and can be slower than unreliable delivery. /// Reliable = 1, - - /// - /// Requests data be sent so that large messages are fragmented into smaller chunks of - /// data and reassembled when received. - /// - /// - /// Fragmented messages allow large amounts of data to be transmitted in smaller chunks when using connections - /// that do not support the transmission of large messages. Without specifying reliable delivery there is no - /// guarentee that the message will arrive but any incomplete messages will be simply be discarded. - /// - Fragmented = 2, - + /// /// Requests data be sent so that large messages are fragmented into smaller chunks of /// data and reassembled when received. @@ -53,6 +42,6 @@ namespace Hazel /// guaranteed to arrive and to arrive only once but the sending process may require more memory, processing, /// a larger number protocol bytes and may be slower than sending unreliably. /// - FragmentedReliable = 3 + FragmentedReliable = 2 } } diff --git a/Hazel/SendOptionInternal.cs b/Hazel/Udp/SendOptionInternal.cs similarity index 72% rename from Hazel/SendOptionInternal.cs rename to Hazel/Udp/SendOptionInternal.cs index 2a7a707..164ab83 100644 --- a/Hazel/SendOptionInternal.cs +++ b/Hazel/Udp/SendOptionInternal.cs @@ -4,12 +4,12 @@ using System.Linq; using System.Text; -namespace Hazel +namespace Hazel.Udp { /// /// Extra internal states for SendOption enumeration when using UDP. /// - enum SendOptionInternal : byte + enum UdpSendOption : byte { /// /// Hello message for initiating communication. @@ -24,6 +24,11 @@ namespace Hazel /// /// Message acknowledging the receipt of a message. /// - Acknowledgement = 10 + Acknowledgement = 10, + + /// + /// Message that is part of a larger, fragmented message. + /// + Fragment = 11 } } diff --git a/Hazel/Udp/UdpClientConnection.cs b/Hazel/Udp/UdpClientConnection.cs index 26eece7..4ffe424 100644 --- a/Hazel/Udp/UdpClientConnection.cs +++ b/Hazel/Udp/UdpClientConnection.cs @@ -79,6 +79,10 @@ namespace Hazel.Udp lock (socket) socket.EndSendTo(result); } + catch (ObjectDisposedException e) + { + HandleDisconnect(new HazelException("Could not send as the socket was disposed of.", e)); + } catch (SocketException e) { HandleDisconnect(new HazelException("Could not send data as a SocketException occured.", e)); diff --git a/Hazel/Udp/UdpConnection.Fragmented.cs b/Hazel/Udp/UdpConnection.Fragmented.cs new file mode 100644 index 0000000..09248bd --- /dev/null +++ b/Hazel/Udp/UdpConnection.Fragmented.cs @@ -0,0 +1,198 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; + +namespace Hazel.Udp +{ + partial class UdpConnection + { + /// + /// The amount of data that can be put into a fragment. + /// + public int FragmentSize { get; } = 65507 - 1 - 2 - 2 - 2; + + /// + /// The last fragmented message ID that was written. + /// + volatile ushort lastFragmentIDAllocated; + + Dictionary fragmentedMessagesReceived = new Dictionary(); + + /// + /// Sends a message fragmenting it as needed to pass over the network. + /// + /// The send option the message was sent with. + /// The data of the message to send. + void FragmentedSend(byte[] data) + { + //Get an ID not used yet. + ushort id = ++lastFragmentIDAllocated; //TODO manage loop around + + for (ushort i = 0; i < Math.Ceiling(data.Length / (double)FragmentSize); i++) + { + byte[] buffer = new byte[Math.Min(data.Length - (FragmentSize * i), FragmentSize) + 7]; + + //Add send option + buffer[0] = i == 0 ? (byte)SendOption.FragmentedReliable : (byte)UdpSendOption.Fragment; + + //Add fragment message ID + buffer[1] = (byte)((id >> 8) & 0xFF); + buffer[2] = (byte)id; + + //Add length or fragment id + if (i == 0) + { + ushort fragments = (ushort)Math.Ceiling(data.Length / (double)FragmentSize); + buffer[3] = (byte)((fragments >> 8) & 0xFF); + buffer[4] = (byte)fragments; + } + else + { + buffer[3] = (byte)((i >> 8) & 0xFF); + buffer[4] = (byte)i; + } + + //Pass fragment to reliable send code to ensure it will arrive + AttachReliableID(buffer, 5); + + //Copy data into fragment + Buffer.BlockCopy(data, FragmentSize * i, buffer, 7, buffer.Length - 7); + + //Send + WriteBytesToConnection(buffer); + } + } + + /// + /// Gets a message from those we've begun receiving or adds a new one. + /// + /// The Id of the message to find. + /// + FragmentedMessage GetFragmentedMessage(ushort messageId) + { + lock (fragmentedMessagesReceived) + { + FragmentedMessage message; + if (fragmentedMessagesReceived.ContainsKey(messageId)) + { + message = fragmentedMessagesReceived[messageId]; + } + else + { + message = new FragmentedMessage(); + + fragmentedMessagesReceived.Add(messageId, message); + } + + return message; + } + } + + /// + /// Handles a the start message of a fragmented message. + /// + /// The buffer received. + void FragmentedStartMessageReceive(byte[] buffer) + { + //Send to reliable code to send the acknowledgement + if (!ProcessReliableReceive(buffer, 5)) + return; + + ushort id = (ushort)((buffer[1] << 8) + buffer[2]); + + ushort length = (ushort)((buffer[3] << 8) + buffer[4]); + + FragmentedMessage message; + bool messageComplete; + lock (fragmentedMessagesReceived) + { + message = GetFragmentedMessage(id); + message.received.Add(new FragmentedMessage.Fragment(0, buffer, 7)); + message.noFragments = length; + + messageComplete = message.noFragments == message.received.Count; + } + + if (messageComplete) + FinalizeFragmentedMessage(message); + } + + /// + /// Handles a fragment message of a fragmented message. + /// + /// The buffer received. + void FragmentedMessageReceive(byte[] buffer) + { + //Send to reliable code to send the acknowledgement + if (!ProcessReliableReceive(buffer, 5)) + return; + + ushort id = (ushort)((buffer[1] << 8) + buffer[2]); + + ushort fragmentID = (ushort)((buffer[3] << 8) + buffer[4]); + + FragmentedMessage message; + bool messageComplete; + lock (fragmentedMessagesReceived) + { + message = GetFragmentedMessage(id); + message.received.Add(new FragmentedMessage.Fragment(fragmentID, buffer, 7)); + + messageComplete = message.noFragments == message.received.Count; + } + + if (messageComplete) + FinalizeFragmentedMessage(message); + } + + /// + /// Finalizes a completed fragmented message and invokes message received events. + /// + /// The message received. + void FinalizeFragmentedMessage(FragmentedMessage message) + { + IEnumerable orderedFragments = message.received.OrderBy((x) => x.fragmentID); + + byte[] completeData = new byte[(orderedFragments.Count() - 1) * FragmentSize + orderedFragments.Last().data.Length]; + int ptr = 0; + foreach (FragmentedMessage.Fragment fragment in orderedFragments) + { + Buffer.BlockCopy(fragment.data, fragment.offset, completeData, ptr, fragment.data.Length - fragment.offset); + ptr += fragment.data.Length - fragment.offset; + } + + InvokeDataReceived(completeData, SendOption.FragmentedReliable); + } + + /// + /// Holding class for the parts of a fragmented message so far received. + /// + private class FragmentedMessage + { + /// + /// The total number of fragments expected. + /// + public int noFragments = -1; + + /// + /// The fragments received so far. + /// + public List received = new List(); + + public struct Fragment + { + public int fragmentID; + public byte[] data; + public int offset; + + public Fragment(int fragmentID, byte[] data, int offset) + { + this.fragmentID = fragmentID; + this.data = data; + this.offset = offset; + } + } + } + } +} diff --git a/Hazel/Udp/UdpConnection.Reliable.cs b/Hazel/Udp/UdpConnection.Reliable.cs index 2b58683..f87233b 100644 --- a/Hazel/Udp/UdpConnection.Reliable.cs +++ b/Hazel/Udp/UdpConnection.Reliable.cs @@ -177,17 +177,13 @@ namespace Hazel.Udp } /// - /// Sends the bytes reliably and stores the send. + /// Adds a 2 byte ID to the packet at offset and stores the packet reference for retransmission. /// - /// The byte array to write to. + /// The buffer to attach to. + /// The offset to attach at. /// The callback to make once the packet has been acknowledged. - void ReliableSend(byte sendOption, byte[] data, Action ackCallback) + void AttachReliableID(byte[] buffer, int offset, Action ackCallback = null) { - byte[] bytes = new byte[data.Length + 3]; - - //Add message type - bytes[0] = sendOption; - //Find and reliable ID lock (reliableDataPacketsSent) { @@ -199,13 +195,13 @@ namespace Hazel.Udp while (reliableDataPacketsSent.ContainsKey(id)); //Write ID - bytes[1] = (byte)((id >> 8) & 0xFF); - bytes[2] = (byte)id; + buffer[offset] = (byte)((id >> 8) & 0xFF); + buffer[offset + 1] = (byte)id; //Create packet object Packet packet = Packet.GetObject(); packet.Set( - bytes, + buffer, (Packet p) => { //Double packet timeout @@ -217,7 +213,7 @@ namespace Hazel.Udp if (++p.Retransmissions > ResendsBeforeDisconnect) { HandleDisconnect(); - + //Set acknowledged so we dont change the timer again p.Acknowledged = true; @@ -246,6 +242,22 @@ namespace Hazel.Udp //Remember packet reliableDataPacketsSent.Add(id, packet); } + } + + /// + /// Sends the bytes reliably and stores the send. + /// + /// The byte array to write to. + /// The callback to make once the packet has been acknowledged. + void ReliableSend(byte sendOption, byte[] data, Action ackCallback = null) + { + byte[] bytes = new byte[data.Length + 3]; + + //Add message type + bytes[0] = sendOption; + + //Add reliable ID + AttachReliableID(bytes, 1, ackCallback); //Copy data into new array Buffer.BlockCopy(data, 0, bytes, bytes.Length - data.Length, data.Length); @@ -260,9 +272,9 @@ namespace Hazel.Udp /// Handles a reliable message being received and invokes the data event. /// /// The buffer received. - void ReliableReceive(byte[] buffer) + void ReliableMessageReceive(byte[] buffer) { - if (ProcessReliableReceive(buffer)) + if (ProcessReliableReceive(buffer, 1)) InvokeDataReceived(SendOption.Reliable, buffer, 3); Statistics.LogReliableReceive(buffer.Length - 3, buffer.Length); @@ -272,14 +284,15 @@ namespace Hazel.Udp /// Handles receives from reliable packets. /// /// The buffer containing the data. + /// The offset of the reliable header. /// Whether the packet was a new packet or not. - bool ProcessReliableReceive(byte[] bytes) + bool ProcessReliableReceive(byte[] bytes, int offset) { //Get the ID form the packet - ushort id = (ushort)((bytes[1] << 8) + bytes[2]); + ushort id = (ushort)((bytes[offset] << 8) + bytes[offset + 1]); //Send an acknowledgement - SendAck(bytes[1], bytes[2]); + SendAck(bytes[offset], bytes[offset + 1]); /* * It gets a little complicated here (note the fact I'm actually using a multiline comment for once...) @@ -348,7 +361,7 @@ namespace Hazel.Udp /// Handles acknowledgement packets to us. /// /// The buffer containing the data. - void AcknowledgementReceive(byte[] bytes) + void AcknowledgementMessageReceive(byte[] bytes) { //Get ID ushort id = (ushort)((bytes[1] << 8) + bytes[2]); @@ -390,7 +403,7 @@ namespace Hazel.Udp WriteBytesToConnection( //TODO group acks together new byte[] { - (byte)SendOptionInternal.Acknowledgement, + (byte)UdpSendOption.Acknowledgement, byte1, byte2 } diff --git a/Hazel/Udp/UdpConnection.cs b/Hazel/Udp/UdpConnection.cs index 7e7e745..d724be9 100644 --- a/Hazel/Udp/UdpConnection.cs +++ b/Hazel/Udp/UdpConnection.cs @@ -63,13 +63,17 @@ namespace Hazel.Udp { //Handle reliable header and hellos case (byte)SendOption.Reliable: - case (byte)SendOptionInternal.Hello: + case (byte)UdpSendOption.Hello: ReliableSend(sendOption, data, ackCallback); break; + + case (byte)SendOption.FragmentedReliable: + FragmentedSend(data); + break; //Treat all else as unreliable default: - UnreliableSend(sendOption, data); + UnreliableSend(data, sendOption); break; } } @@ -87,24 +91,33 @@ namespace Hazel.Udp { //Handle reliable receives case (byte)SendOption.Reliable: - ReliableReceive(buffer); + ReliableMessageReceive(buffer); break; //Handle acknowledgments - case (byte)SendOptionInternal.Acknowledgement: - AcknowledgementReceive(buffer); + case (byte)UdpSendOption.Acknowledgement: + AcknowledgementMessageReceive(buffer); break; //We need to acknowledge hello messages but dont want to invoke any events! - case (byte)SendOptionInternal.Hello: - ProcessReliableReceive(buffer); + case (byte)UdpSendOption.Hello: + ProcessReliableReceive(buffer, 1); Statistics.LogHelloReceive(buffer.Length); break; - case (byte)SendOptionInternal.Disconnect: + case (byte)UdpSendOption.Disconnect: HandleDisconnect(); break; + //Handle fragmented messages + case (byte)SendOption.FragmentedReliable: + FragmentedStartMessageReceive(buffer); + break; + + case (byte)UdpSendOption.Fragment: + FragmentedMessageReceive(buffer); + break; + //Treat everything else as unreliable default: InvokeDataReceived(SendOption.None, buffer, 1); @@ -113,7 +126,12 @@ namespace Hazel.Udp } } - void UnreliableSend(byte sendOption, byte[] data) + /// + /// Sends bytes using the unreliable UDP protocol. + /// + /// The data. + /// The SendOption to attach. + void UnreliableSend(byte[] data, byte sendOption) { byte[] bytes = new byte[data.Length + 1]; @@ -161,7 +179,7 @@ namespace Hazel.Udp Buffer.BlockCopy(bytes, 0, actualBytes, 1, bytes.Length); } - HandleSend(actualBytes, (byte)SendOptionInternal.Hello, acknowledgeCallback); + HandleSend(actualBytes, (byte)UdpSendOption.Hello, acknowledgeCallback); } /// @@ -175,7 +193,7 @@ namespace Hazel.Udp /// protected void SendDisconnect() { - HandleSend(new byte[0], (byte)SendOptionInternal.Disconnect); //TODO Should disconnect wait for an ack? + HandleSend(new byte[0], (byte)UdpSendOption.Disconnect); //TODO Should disconnect wait for an ack? } /// diff --git a/Hazel/Udp/UdpConnectionListener.cs b/Hazel/Udp/UdpConnectionListener.cs index 22c3c37..5c3c44c 100644 --- a/Hazel/Udp/UdpConnectionListener.cs +++ b/Hazel/Udp/UdpConnectionListener.cs @@ -150,7 +150,7 @@ namespace Hazel.Udp else { //Check for malformed connection attempts - if (buffer[0] != (byte)SendOptionInternal.Hello) + if (buffer[0] != (byte)UdpSendOption.Hello) return; connection = new UdpServerConnection(this, remoteEndPoint, IPMode); -- 2.39.5