<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="UdpConnectionTests.cs" />
<Compile Include="MessageWriterTests.cs" />
+ <Compile Include="UnitTest1.cs" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Hazel\Hazel.csproj">
ManualResetEvent mutex = new ManualResetEvent(false);
//Setup listener
- listener.NewConnection += delegate(object sender, NewConnectionEventArgs args)
+ listener.NewConnection += delegate(object sender, NewConnectionEventArgs ncArgs)
{
- args.Connection.SendBytes(data, sendOption);
+ ncArgs.Connection.SendBytes(data, sendOption);
};
listener.Start();
+ DataReceivedEventArgs args = null;
//Setup conneciton
- connection.DataReceived += delegate(object sender, DataReceivedEventArgs args)
+ connection.DataReceived += delegate(object sender, DataReceivedEventArgs a)
{
Trace.WriteLine("Data was received correctly.");
- Assert.AreEqual(data.Length, args.Message.Length);
-
- for (int i = 0; i < data.Length; i++)
+ try
{
- Assert.AreEqual(data[i], args.Message.ReadByte());
+ args = a;
+ }
+ finally
+ {
+ mutex.Set();
}
-
- Assert.AreEqual(sendOption, args.SendOption);
-
- mutex.Set();
};
connection.Connect();
//Wait until data is received
mutex.WaitOne();
+
+ Assert.AreEqual(data.Length, args.Message.Length);
+
+ for (int i = 0; i < data.Length; i++)
+ {
+ Assert.AreEqual(data[i], args.Message.ReadByte());
+ }
+
+ Assert.AreEqual(sendOption, args.SendOption);
}
/// <summary>
ManualResetEvent mutex2 = new ManualResetEvent(false);
//Setup listener
+ DataReceivedEventArgs result = null;
listener.NewConnection += delegate(object sender, NewConnectionEventArgs args)
{
args.Connection.DataReceived += delegate(object innerSender, DataReceivedEventArgs innerArgs)
{
Trace.WriteLine("Data was received correctly.");
- Assert.AreEqual(data.Length, innerArgs.Message.Length);
-
- for (int i = 0; i < data.Length; i++)
- {
- Assert.AreEqual(data[i], innerArgs.Message.ReadByte());
- }
-
- Assert.AreEqual(sendOption, innerArgs.SendOption);
+ result = innerArgs;
mutex2.Set();
};
//Wait until data is received
mutex2.WaitOne();
+
+ Assert.AreEqual(data.Length, result.Message.Length);
+
+ for (int i = 0; i < data.Length; i++)
+ {
+ Assert.AreEqual(data[i], result.Message.ReadByte());
+ }
+
+ Assert.AreEqual(sendOption, result.SendOption);
}
/// <summary>
TestHelper.RunServerToClientTest(listener, connection, 10, SendOption.Reliable);
}
}
-
- /// <summary>
- /// Tests server to client reliable communication on the UdpConnection.
- /// </summary>
- [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)(UdpConnection.FragmentSize * 9.5), SendOption.FragmentedReliable);
- }
- }
-
+
/// <summary>
/// Tests server to client unreliable communication on the UdpConnection.
/// </summary>
TestHelper.RunClientToServerTest(listener, connection, 10, SendOption.Reliable);
}
}
-
- /// <summary>
- /// Tests server to client reliable communication on the UdpConnection.
- /// </summary>
- [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)(UdpConnection.FragmentSize * 9.5), SendOption.FragmentedReliable);
- }
- }
-
+
/// <summary>
/// Tests the keepalive functionality from the client,
/// </summary>
--- /dev/null
+using System;
+using System.Net;
+using System.Threading;
+using System.Threading.Tasks;
+using Hazel.Udp;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Hazel.UnitTests
+{
+ [TestClass]
+ public class UnitTest1
+ {
+ // [TestMethod]
+ public void StressTest()
+ {
+ Parallel.For(0, 10000,
+ new ParallelOptions { MaxDegreeOfParallelism = 16 },
+ (i) =>
+ {
+ var ep = new NetworkEndPoint(IPAddress.Loopback, 22023);
+ using (var connection = new UdpClientConnection(ep))
+ {
+ connection.Connect();
+ Thread.Sleep(100);
+ }
+ });
+ }
+ }
+}
/// </remarks>
protected void InvokeDataReceived(MessageReader msg, SendOption sendOption, ushort reliableId)
{
- DataReceivedEventArgs args = DataReceivedEventArgs.GetObject();
- args.Set(msg, sendOption, reliableId);
-
//Make a copy to avoid race condition between null check and invocation
EventHandler<DataReceivedEventArgs> handler = DataReceived;
- if (handler != null) handler.Invoke(this, args);
+ if (handler != null)
+ {
+ DataReceivedEventArgs args = DataReceivedEventArgs.GetObject();
+ args.Set(msg, sendOption, reliableId);
+ handler.Invoke(this, args);
+ }
+ else
+ {
+ msg.Recycle();
+ }
}
/// <summary>
/// </remarks>
protected void InvokeDisconnected(Exception e = null)
{
- DisconnectedEventArgs args = DisconnectedEventArgs.GetObject();
- args.Set(e);
-
//Make a copy to avoid race condition between null check and invocation
EventHandler<DisconnectedEventArgs> handler = Disconnected;
- if (handler != null) handler.Invoke(this, args);
+ if (handler != null)
+ {
+ DisconnectedEventArgs args = DisconnectedEventArgs.GetObject();
+ args.Set(e);
+ handler.Invoke(this, args);
+ }
}
/// <summary>
/// </remarks>
protected void InvokeNewConnection(MessageReader msg, Connection connection)
{
- //Get new args
- NewConnectionEventArgs args = NewConnectionEventArgs.GetObject();
- args.Set(msg, connection);
-
//Make a copy to avoid race condition between null check and invocation
EventHandler<NewConnectionEventArgs> handler = NewConnection;
if (handler != null)
+ {
+ NewConnectionEventArgs args = NewConnectionEventArgs.GetObject();
+ args.Set(msg, connection);
handler(this, args);
+ }
+ else
+ {
+ msg.Recycle();
+ }
}
/// <summary>
<Compile Include="Udp\UdpConnection.cs">
<SubType>Code</SubType>
</Compile>
- <Compile Include="Udp\UdpConnection.Fragmented.cs" />
<Compile Include="Udp\UdpConnection.KeepAlive.cs" />
<Compile Include="Udp\UdpConnection.Reliable.cs" />
<Compile Include="Udp\UdpConnectionListener.cs" />
public byte Tag;
public int Length;
+ public int Offset;
- public int Offset { get; private set; }
public int Position
{
get { return this._position; }
private int _position;
private int readHead;
+
+ public static MessageReader GetSized(int minSize)
+ {
+ var output = ReaderPool.GetObject();
+ if (output.Buffer == null || output.Buffer.Length < minSize)
+ {
+ output.Buffer = new byte[minSize];
+ }
+
+ output.Offset = 0;
+ output.Position = 0;
+ output.Length = minSize;
+ output.Tag = byte.MaxValue;
+ return output;
+ }
public static MessageReader GetRaw(byte[] bytes, int offset, int length)
{
output.Offset = offset;
output.Position = 0;
output.Length = length;
+ output.Tag = byte.MaxValue;
return output;
}
private Stack<int> messageStarts = new Stack<int>();
+ public MessageWriter(byte[] buffer)
+ {
+ this.Buffer = buffer;
+ this.Length = this.Buffer.Length;
+ }
+
///
public MessageWriter(int bufferSize)
{
case SendOption.Reliable:
this.Length = this.Position = 3;
break;
- case SendOption.FragmentedReliable:
- throw new NotImplementedException("Sry bruh");
}
}
}
#region WriteMethods
+
+ public void CopyFrom(MessageReader target)
+ {
+ int offset, length;
+ if (target.Tag == byte.MaxValue)
+ {
+ offset = target.Offset;
+ length = target.Length;
+ }
+ else
+ {
+ offset = target.Offset - 3;
+ length = target.Length + 3;
+ }
+
+ System.Buffer.BlockCopy(target.Buffer, offset, this.Buffer, this.Position, length);
+ this.Position += length;
+ if (this.Position > this.Length) this.Length = this.Position;
+ }
+
public void Write(bool value)
{
this.Buffer[this.Position++] = (byte)(value ? 1 : 0);
/// a larger number of protocol bytes and can be slower than unreliable delivery.
/// </remarks>
Reliable = 1,
-
- /// <summary>
- /// Requests data be sent so that large messages are fragmented into smaller chunks of
- /// data and reassembled when received.
- /// </summary>
- /// <remarks>
- /// 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. By specifying reliable delivery messages are
- /// 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.
- /// </remarks>
- FragmentedReliable = 2
}
}
Thread.Sleep(this.TestLagMs);
}
- HandleReceive(bytes);
+ MessageReader msg = MessageReader.GetRaw(bytes, 0, bytesReceived);
+ HandleReceive(msg);
}
/// <inheritdoc />
+++ /dev/null
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-
-namespace Hazel.Udp
-{
- partial class UdpConnection
- {
- /// <summary>
- /// The amount of data that can be put into a fragment.
- /// </summary>
- public const int FragmentSize = 65507 - 1 - 2 - 2 - 2;
-
- /// <summary>
- /// The last fragmented message ID that was written.
- /// </summary>
- volatile ushort lastFragmentIDAllocated;
-
- Dictionary<ushort, FragmentedMessage> fragmentedMessagesReceived = new Dictionary<ushort, FragmentedMessage>();
-
- /// <summary>
- /// Sends a message fragmenting it as needed to pass over the network.
- /// </summary>
- /// <param name="sendOption">The send option the message was sent with.</param>
- /// <param name="data">The data of the message to send.</param>
- void FragmentedSend(byte[] data)
- {
- //Get an ID not used yet.
- ushort id = ++lastFragmentIDAllocated; //TODO is extra code needed to 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, buffer.Length);
-
- //Copy data into fragment
- Buffer.BlockCopy(data, FragmentSize * i, buffer, 7, buffer.Length - 7);
-
- //Send
- WriteBytesToConnection(buffer, buffer.Length);
- }
- }
-
- /// <summary>
- /// Gets a message from those we've begun receiving or adds a new one.
- /// </summary>
- /// <param name="messageId">The Id of the message to find.</param>
- /// <returns></returns>
- 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;
- }
- }
-
- /// <summary>
- /// Handles a the start message of a fragmented message.
- /// </summary>
- /// <param name="buffer">The buffer received.</param>
- void FragmentedStartMessageReceive(byte[] buffer)
- {
- //Send to reliable code to send the acknowledgement
- ushort reliableId;
- if (!ProcessReliableReceive(buffer, 5, out reliableId))
- 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);
- }
-
- /// <summary>
- /// Handles a fragment message of a fragmented message.
- /// </summary>
- /// <param name="buffer">The buffer received.</param>
- void FragmentedMessageReceive(byte[] buffer)
- {
- //Send to reliable code to send the acknowledgement
- ushort reliableId;
- if (!ProcessReliableReceive(buffer, 5, out reliableId))
- 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);
- }
-
- /// <summary>
- /// Finalizes a completed fragmented message and invokes message received events.
- /// </summary>
- /// <param name="message">The message received.</param>
- void FinalizeFragmentedMessage(FragmentedMessage message)
- {
- IEnumerable<FragmentedMessage.Fragment> orderedFragments = message.received.OrderBy((x) => x.fragmentID);
- FragmentedMessage.Fragment last = orderedFragments.Last();
-
- byte[] completeData = new byte[(orderedFragments.Count() - 1) * FragmentSize + last.data.Length - last.offset];
- 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;
- }
-
- var reader = MessageReader.GetRaw(completeData, 0, completeData.Length);
- try
- {
- InvokeDataReceived(reader, SendOption.FragmentedReliable, 0);
- }
- finally
- {
- reader.Recycle();
- }
- }
-
- /// <summary>
- /// Holding class for the parts of a fragmented message so far received.
- /// </summary>
- private class FragmentedMessage
- {
- /// <summary>
- /// The total number of fragments expected.
- /// </summary>
- public int noFragments = -1;
-
- /// <summary>
- /// The fragments received so far.
- /// </summary>
- public List<Fragment> received = new List<Fragment>();
-
- 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;
- }
- }
- }
- }
-}
/// <summary>
/// Handles a reliable message being received and invokes the data event.
/// </summary>
- /// <param name="buffer">The buffer received.</param>
- void ReliableMessageReceive(byte[] buffer)
+ /// <param name="message">The buffer received.</param>
+ void ReliableMessageReceive(MessageReader message)
{
ushort id;
- if (ProcessReliableReceive(buffer, 1, out id))
- InvokeDataReceived(SendOption.Reliable, buffer, 3, id);
+ if (ProcessReliableReceive(message.Buffer, 1, out id))
+ InvokeDataReceived(SendOption.Reliable, message, 3, id);
- Statistics.LogReliableReceive(buffer.Length - 3, buffer.Length);
+ Statistics.LogReliableReceive(message.Length - 3, message.Length);
}
/// <summary>
Statistics.LogReliableSend(buffer.Length - 3, buffer.Length);
break;
- case SendOption.FragmentedReliable:
- throw new NotImplementedException("Not yet");
-
default:
WriteBytesToConnection(buffer, buffer.Length);
Statistics.LogUnreliableSend(buffer.Length - 1, buffer.Length);;
case SendOption.Reliable:
ReliableSend((byte)sendOption, bytes, offset, length);
break;
-
- case SendOption.FragmentedReliable:
- throw new NotImplementedException();
- // FragmentedSend(data);
- // break;
-
+
//Treat all else as unreliable
default:
UnreliableSend((byte)sendOption, bytes, offset, length);
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);
/// <summary>
/// Handles the receiving of data.
/// </summary>
- /// <param name="buffer">The buffer containing the bytes received.</param>
- protected internal void HandleReceive(byte[] buffer)
+ /// <param name="message">The buffer containing the bytes received.</param>
+ protected internal void HandleReceive(MessageReader message)
{
- InvokeDataReceivedRaw(buffer);
+ InvokeDataReceivedRaw(message.Buffer);
- switch (buffer[0])
+ switch (message.Buffer[0])
{
//Handle reliable receives
case (byte)SendOption.Reliable:
- ReliableMessageReceive(buffer);
+ ReliableMessageReceive(message);
break;
//Handle acknowledgments
case (byte)UdpSendOption.Acknowledgement:
- AcknowledgementMessageReceive(buffer);
+ AcknowledgementMessageReceive(message.Buffer);
break;
//We need to acknowledge hello and ping messages but dont want to invoke any events!
case (byte)UdpSendOption.Ping:
case (byte)UdpSendOption.Hello:
ushort id;
- ProcessReliableReceive(buffer, 1, out id);
- Statistics.LogHelloReceive(buffer.Length);
+ ProcessReliableReceive(message.Buffer, 1, out id);
+ Statistics.LogHelloReceive(message.Length);
break;
case (byte)UdpSendOption.Disconnect:
HandleDisconnect(new HazelException("The remote sent a disconnect request"));
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, 0);
- Statistics.LogUnreliableReceive(buffer.Length - 1, buffer.Length);
+ InvokeDataReceived(SendOption.None, message, 1, 0);
+ Statistics.LogUnreliableReceive(message.Length - 1, message.Length);
break;
}
}
/// <param name="sendOption">The send option the message was received with.</param>
/// <param name="buffer">The buffer received.</param>
/// <param name="dataOffset">The offset of data in the buffer.</param>
- void InvokeDataReceived(SendOption sendOption, byte[] buffer, int dataOffset, ushort reliableId)
+ void InvokeDataReceived(SendOption sendOption, MessageReader buffer, int dataOffset, ushort reliableId)
{
- var reader = MessageReader.GetRaw(buffer, dataOffset, buffer.Length - dataOffset);
- try
- {
- InvokeDataReceived(reader, sendOption, reliableId);
- }
- finally
- {
- reader.Recycle();
- }
+ buffer.Offset = dataOffset;
+ buffer.Length = buffer.Length - dataOffset;
+ buffer.Position = 0;
+
+ InvokeDataReceived(buffer, sendOption, reliableId);
}
/// <summary>
using System.Net;
using System.Net.Sockets;
using System.Text;
-
+using System.Threading;
namespace Hazel.Udp
{
/// The socket listening for connections.
/// </summary>
Socket listener;
-
- /// <summary>
- /// Buffer to store incoming data in.
- /// </summary>
- byte[] dataBuffer = new byte[ushort.MaxValue];
-
+
/// <summary>
/// The connections we currently hold
/// </summary>
try
{
- listener.BeginReceiveFrom(dataBuffer, 0, dataBuffer.Length, SocketFlags.None, ref remoteEP, ReadCallback, dataBuffer);
+ var message = MessageReader.GetSized(ushort.MaxValue);
+ listener.BeginReceiveFrom(message.Buffer, 0, message.Buffer.Length, SocketFlags.None, ref remoteEP, ReadCallback, message);
+ Interlocked.Increment(ref ActiveThreads);
}
catch (ObjectDisposedException)
{
/// Called when data has been received by the listener.
/// </summary>
/// <param name="result">The asyncronous operation's result.</param>
+
+ private int ActiveThreads;
void ReadCallback(IAsyncResult result)
{
int bytesReceived;
//End the receive operation
try
{
+ Interlocked.Decrement(ref ActiveThreads);
bytesReceived = listener.EndReceiveFrom(result, ref remoteEndPoint);
}
catch (ObjectDisposedException)
//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
-
StartListeningForData();
return;
}
//Exit if no bytes read, we've closed.
if (bytesReceived == 0)
return;
-
- //Copy to new buffer
- byte[] buffer = new byte[bytesReceived];
- Buffer.BlockCopy((byte[])result.AsyncState, 0, buffer, 0, bytesReceived);
-
+
//Begin receiving again
StartListeningForData();
+ var message = (MessageReader)result.AsyncState;
+ message.Length = bytesReceived;
+
bool aware;
UdpServerConnection connection;
lock (connections)
if (!(aware = connections.TryGetValue(remoteEndPoint, out connection)))
{
//Check for malformed connection attempts
- if (buffer[0] != (byte)UdpSendOption.Hello)
+ if (message.Buffer[0] != (byte)UdpSendOption.Hello)
return;
connection = new UdpServerConnection(this, remoteEndPoint, IPMode);
}
//Inform the connection of the buffer (new connections need to send an ack back to client)
- connection.HandleReceive(buffer);
-
+ connection.HandleReceive(message);
+
//If it's a new connection invoke the NewConnection event.
if (!aware)
{
- var reader = MessageReader.GetRaw(buffer, 4, buffer.Length - 4);
- try
- {
- InvokeNewConnection(reader, connection);
- }
- finally
- {
- reader.Recycle();
- }
+ // Skip header and hello byte;
+ message.Offset = 4;
+ message.Length = bytesReceived - 4;
+ message.Position = 0;
+ InvokeNewConnection(message, connection);
}
}
projects / rhonda / impostor.hazel.git / commitdiff