<Compile Include="Properties\AssemblyInfo.cs" />
<Compile Include="UdpConnectionTests.cs" />
<Compile Include="MessageWriterTests.cs" />
- <Compile Include="UnitTest1.cs" />
+ <Compile Include="StressTests.cs" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Hazel\Hazel.csproj">
}
+ [TestMethod]
+ public void CopySubMessage()
+ {
+ const byte Test1 = 12;
+ const byte Test2 = 146;
+
+ var msg = new MessageWriter(2048);
+ msg.StartMessage(1);
+
+ msg.StartMessage(2);
+ msg.Write(Test1);
+ msg.Write(Test2);
+ msg.EndMessage();
+
+ msg.EndMessage();
+
+ MessageReader handleMessage = MessageReader.Get(msg.Buffer, 0);
+ Assert.AreEqual(1, handleMessage.Tag);
+
+ var parentReader = MessageReader.Get(handleMessage);
+
+ handleMessage.Recycle();
+ SetZero(handleMessage);
+
+ Assert.AreEqual(1, parentReader.Tag);
+
+ for (int i = 0; i < 5; ++i)
+ {
+
+ var reader = parentReader.ReadMessage();
+ Assert.AreEqual(2, reader.Tag);
+ Assert.AreEqual(Test1, reader.ReadByte());
+ Assert.AreEqual(Test2, reader.ReadByte());
+
+ var temp = parentReader;
+ parentReader = MessageReader.CopyMessageIntoParent(reader);
+
+ temp.Recycle();
+ SetZero(temp);
+ SetZero(reader);
+ }
+ }
+
[TestMethod]
public void ReadMessageLength()
{
{
Assert.IsTrue(MessageWriter.IsLittleEndian());
}
+
+ private void SetZero(MessageReader reader)
+ {
+ for (int i = 0; i < reader.Buffer.Length; ++i)
+ reader.Buffer[i] = 0;
+ }
}
+
}
\ No newline at end of file
Assert.AreEqual(68000u, reader.ReadPackedUInt32());
}
-
[TestMethod]
public void WritePackedInt()
{
msg.WritePacked(8);
msg.WritePacked(250);
msg.WritePacked(68000);
+ msg.WritePacked(60168000);
msg.WritePacked(-68000);
msg.WritePacked(-250);
msg.WritePacked(-8);
+
+ msg.WritePacked(0);
+ msg.WritePacked(-1);
+ msg.WritePacked(int.MinValue);
+ msg.WritePacked(int.MaxValue);
msg.EndMessage();
- Assert.AreEqual(3 + 1 + 2 + 3 + 5 + 5 + 5, msg.Position);
+ Assert.AreEqual(3 + 1 + 2 + 3 + 4 + 5 + 5 + 5 + 1 + 5 + 5 + 5, msg.Position);
Assert.AreEqual(msg.Length, msg.Position);
MessageReader reader = MessageReader.Get(msg.Buffer, 0);
Assert.AreEqual(8, reader.ReadPackedInt32());
Assert.AreEqual(250, reader.ReadPackedInt32());
Assert.AreEqual(68000, reader.ReadPackedInt32());
-
-
+ Assert.AreEqual(60168000, reader.ReadPackedInt32());
+
Assert.AreEqual(-68000, reader.ReadPackedInt32());
Assert.AreEqual(-250, reader.ReadPackedInt32());
Assert.AreEqual(-8, reader.ReadPackedInt32());
+
+ Assert.AreEqual(0, reader.ReadPackedInt32());
+ Assert.AreEqual(-1, reader.ReadPackedInt32());
+ Assert.AreEqual(int.MinValue, reader.ReadPackedInt32());
+ Assert.AreEqual(int.MaxValue, reader.ReadPackedInt32());
}
[TestMethod]
--- /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 StressTests
+ {
+ // [TestMethod]
+ public void StressTestOpeningConnections()
+ {
+ // Start a listener in another process, or even better,
+ // adjust the target IP and start listening on another computer.
+ var ep = new IPEndPoint(IPAddress.Loopback, 22023);
+ Parallel.For(0, 10000,
+ new ParallelOptions { MaxDegreeOfParallelism = 64 },
+ (i) => {
+
+ var connection = new UdpClientConnection(ep);
+ connection.KeepAliveInterval = 50;
+
+ connection.Connect(new byte[5]);
+ });
+ }
+ }
+}
[TestClass]
public class UdpConnectionTests
{
+ [TestMethod]
+ public void ServerDisposeDisconnectsTest()
+ {
+ IPEndPoint ep = new IPEndPoint(IPAddress.Loopback, 4296);
+
+ bool serverConnected = false;
+ bool serverDisconnected = false;
+ bool clientDisconnected = false;
+
+ using (UdpConnectionListener listener = new UdpConnectionListener(new IPEndPoint(IPAddress.Any, 4296)))
+ using (UdpConnection connection = new UdpClientConnection(ep))
+ {
+ listener.NewConnection += (evt) =>
+ {
+ serverConnected = true;
+ evt.Connection.Disconnected += (o, et) => serverDisconnected = true;
+ };
+ connection.Disconnected += (o, evt) => clientDisconnected = true;
+
+ listener.Start();
+ connection.Connect();
+
+ listener.Dispose();
+ Thread.Sleep(10);
+
+ Assert.IsTrue(serverConnected);
+ Assert.IsTrue(clientDisconnected);
+ Assert.IsFalse(serverDisconnected);
+ }
+ }
+
+ [TestMethod]
+ public void ClientServerDisposeDisconnectsTest()
+ {
+ IPEndPoint ep = new IPEndPoint(IPAddress.Loopback, 4296);
+
+ bool serverConnected = false;
+ bool serverDisconnected = false;
+ bool clientDisconnected = false;
+
+ using (UdpConnectionListener listener = new UdpConnectionListener(new IPEndPoint(IPAddress.Any, 4296)))
+ using (UdpConnection connection = new UdpClientConnection(ep))
+ {
+ listener.NewConnection += (evt) =>
+ {
+ serverConnected = true;
+ evt.Connection.Disconnected += (o, et) => serverDisconnected = true;
+ };
+
+ connection.Disconnected += (o, et) => clientDisconnected = true;
+
+ listener.Start();
+ connection.Connect();
+ connection.Dispose();
+
+ Thread.Sleep(10);
+
+ Assert.IsTrue(serverConnected);
+ Assert.IsTrue(serverDisconnected);
+ Assert.IsFalse(clientDisconnected);
+ }
+ }
+
/// <summary>
/// Tests the fields on UdpConnection.
/// </summary>
TestHelper.RunClientToServerTest(listener, connection, 10, SendOption.Reliable);
}
}
-
+
+ /// <summary>
+ /// Tests the keepalive functionality from the client,
+ /// </summary>
+ [TestMethod]
+ public void PingDisconnectClientTest()
+ {
+#if DEBUG
+ using (UdpConnectionListener listener = new UdpConnectionListener(new IPEndPoint(IPAddress.Any, 4296)))
+ using (UdpConnection connection = new UdpClientConnection(new IPEndPoint(IPAddress.Loopback, 4296)))
+ {
+ listener.Start();
+
+ connection.Connect();
+
+ // After connecting, quietly stop responding to all messages to fake connection loss.
+ Thread.Sleep(10);
+ listener.TestDropRate = 1;
+
+ connection.KeepAliveInterval = 100;
+
+ Thread.Sleep(1050); //Enough time for ~10 keep alive packets
+
+ Assert.AreEqual(ConnectionState.NotConnected, connection.State);
+ Assert.AreEqual(3 * connection.MissingPingsUntilDisconnect + 4, connection.Statistics.TotalBytesSent); // + 4 for connecting overhead
+ }
+#else
+ Assert.Inconclusive("Only works in DEBUG");
+#endif
+ }
+
/// <summary>
/// Tests the keepalive functionality from the client,
/// </summary>
System.Threading.Thread.Sleep(1050); //Enough time for ~10 keep alive packets
+ Assert.AreEqual(ConnectionState.Connected, connection.State);
Assert.IsTrue(
connection.Statistics.TotalBytesSent >= 30 &&
connection.Statistics.TotalBytesSent <= 50,
using (UdpConnectionListener listener = new UdpConnectionListener(new IPEndPoint(IPAddress.Any, 4296)))
using (UdpConnection connection = new UdpClientConnection(new IPEndPoint(IPAddress.Loopback, 4296)))
{
- listener.NewConnection += delegate(NewConnectionEventArgs args)
+ UdpConnection client = null;
+ listener.NewConnection += delegate (NewConnectionEventArgs args)
{
- ((UdpConnection)args.Connection).KeepAliveInterval = 100;
+ client = (UdpConnection)args.Connection;
+ client.KeepAliveInterval = 100;
Thread.Sleep(1050); //Enough time for ~10 keep alive packets
- try
- {
- Assert.IsTrue(
- args.Connection.Statistics.TotalBytesSent >= 30 &&
- args.Connection.Statistics.TotalBytesSent <= 50,
- "Sent: " + args.Connection.Statistics.TotalBytesSent
- );
- }
- finally
- {
- mutex.Set();
- }
+ mutex.Set();
};
listener.Start();
connection.Connect();
mutex.WaitOne();
+
+ Assert.AreEqual(ConnectionState.Connected, client.State);
+
+ Assert.IsTrue(
+ client.Statistics.TotalBytesSent >= 27 &&
+ client.Statistics.TotalBytesSent <= 50,
+ "Sent: " + client.Statistics.TotalBytesSent
+ );
}
}
+++ /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()
- {
- var ep = new IPEndPoint(IPAddress.Loopback, 22023);
- Parallel.For(0, 10000,
- new ParallelOptions { MaxDegreeOfParallelism = 64 },
- (i) => {
-
- var connection = new UdpClientConnection(ep);
- connection.KeepAliveInterval = 50;
-
- connection.Connect(new byte[5]);
- });
- }
- }
-}
/// <summary>
/// Reset event that is triggered when the connection is marked Connected.
/// </summary>
- ManualResetEvent connectWaitLock = new ManualResetEvent(false);
+ private ManualResetEvent connectWaitLock = new ManualResetEvent(false);
/// <summary>
/// Constructor that initializes the ConnecitonStatistics object.
{
this.DataReceived = null;
this.Disconnected = null;
+ this.connectWaitLock.Dispose();
}
}
}
/// </summary>
int acknowledgementMessagesReceived;
+ /// <summary>
+ /// The number of ping messages received.
+ /// </summary>
+ /// <remarks>
+ /// This is the number of hello messages that were received by the <see cref="Connection"/>, incremented
+ /// each time that LogHelloReceive is called by the Connection. Messages are counted before the receive event is invoked.
+ /// </remarks>
+ public int PingMessagesReceived
+ {
+ get
+ {
+ return pingMessagesReceived;
+ }
+ }
+
+ /// <summary>
+ /// The number of hello messages received.
+ /// </summary>
+ int pingMessagesReceived;
+
/// <summary>
/// The number of hello messages received.
/// </summary>
Interlocked.Add(ref totalBytesReceived, totalLength);
}
+ /// <summary>
+ /// Logs the receiving of a hello data packet in the statistics.
+ /// </summary>
+ /// <param name="totalLength">The total number of bytes received.</param>
+ /// <remarks>
+ /// This should be called before the received event is invoked so it is up to date for subscribers to that event.
+ /// </remarks>
+ internal void LogPingReceive(int totalLength)
+ {
+ Interlocked.Increment(ref pingMessagesReceived);
+ Interlocked.Add(ref totalBytesReceived, totalLength);
+ }
+
/// <summary>
/// Logs the receiving of a hello data packet in the statistics.
/// </summary>
return output;
}
+ public static MessageReader CopyMessageIntoParent(MessageReader source)
+ {
+ var output = MessageReader.GetSized(source.Length + 3);
+ System.Buffer.BlockCopy(source.Buffer, source.Offset - 3, output.Buffer, 0, source.Length + 3);
+
+ output.Offset = 0;
+ output.Position = 0;
+ output.Length = source.Length + 3;
+
+ return output;
+ }
+
public static MessageReader Get(MessageReader source)
{
var output = GetSized(source.Buffer.Length);
throw new InvalidOperationException("IPV6 not supported!");
socket = new Socket(AddressFamily.InterNetworkV6, SocketType.Dgram, ProtocolType.Udp);
- socket.SetSocketOption(SocketOptionLevel.IPv6, (SocketOptionName)27, false); //TODO these lines shouldn't be needed anymore
+ socket.SetSocketOption(SocketOptionLevel.IPv6, SocketOptionName.IPv6Only, false);
}
reliablePacketTimer = new Timer(ManageReliablePacketsInternal, null, 100, Timeout.Infinite);
length,
SocketFlags.None,
RemoteEndPoint,
- delegate (IAsyncResult result)
- {
- try
- {
- socket.EndSendTo(result);
- }
- catch (NullReferenceException) { }
- catch (ObjectDisposedException)
- {
- Disconnect("Could not send as the socket was disposed of.");
- }
- catch (SocketException)
- {
- Disconnect("Could not send data as a SocketException occured.");
- }
- },
- null
- );
+ HandleSendTo,
+ null);
}
+ catch (NullReferenceException) { }
catch (ObjectDisposedException)
{
- //User probably called Disconnect in between this method starting and here so report the issue
- throw new InvalidOperationException("Could not send data as this Connection is not connected. Did you disconnect?");
+ // Already disposed and disconnected...
}
- catch (SocketException)
+ catch (SocketException ex)
{
- Disconnect("Could not send data as a SocketException occured.");
- throw;
+ Disconnect("Could not send data as a SocketException occured: " + ex.Message, true);
}
}
- protected override void WriteBytesToConnectionSync(byte[] bytes, int length)
+ private void HandleSendTo(IAsyncResult result)
{
- DataSentRaw?.Invoke(bytes, length);
-
try
{
- socket.SendTo(
- bytes,
- 0,
- length,
- SocketFlags.None,
- RemoteEndPoint);
+ socket.EndSendTo(result);
}
+ catch (NullReferenceException) { }
catch (ObjectDisposedException)
{
- //User probably called Disconnect in between this method starting and here so report the issue
- throw new InvalidOperationException("Could not send data as this Connection is not connected. Did you disconnect?");
+ // Already disposed and disconnected...
}
- catch (SocketException)
+ catch (SocketException ex)
{
- Disconnect("Could not send data as a SocketException occured.");
- throw;
+ Disconnect("Could not send data as a SocketException occured: " + ex.Message, true);
}
}
{
try
{
- WriteBytesToConnectionSync(DisconnectBytes, 1);
+ socket.SendTo(
+ DisconnectBytes,
+ 0,
+ 1,
+ SocketFlags.None,
+ RemoteEndPoint);
}
catch { }
}
if (this._state == ConnectionState.Connected
|| this._state == ConnectionState.Disconnecting)
{
- SendDisconnect();
this._state = ConnectionState.NotConnected;
+ SendDisconnect();
}
}
using System;
+using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
{
partial class UdpConnection
{
+
+ /// <summary>
+ /// Class to hold packet data
+ /// </summary>
+ public class PingPacket : IRecyclable
+ {
+ private static readonly ObjectPool<PingPacket> PacketPool = new ObjectPool<PingPacket>(() => new PingPacket());
+
+ public readonly Stopwatch Stopwatch = new Stopwatch();
+
+ internal static PingPacket GetObject()
+ {
+ return PacketPool.GetObject();
+ }
+
+ public void Recycle()
+ {
+ Stopwatch.Stop();
+ PacketPool.PutObject(this);
+ }
+ }
+
+ internal ConcurrentDictionary<ushort, PingPacket> activePingPackets = new ConcurrentDictionary<ushort, PingPacket>();
+
/// <summary>
/// The interval from data being received or transmitted to a keepalive packet being sent in milliseconds.
/// </summary>
set
{
keepAliveInterval = value;
-
+
//Update timer
ResetKeepAliveTimer();
}
}
- int keepAliveInterval = 2000;
+ int keepAliveInterval = 1500;
+
+ public int MissingPingsUntilDisconnect { get; set; } = 6;
+ int pingsSinceAck = 0;
/// <summary>
/// The timer creating keepalive pulses.
keepAliveTimer = new Timer(
(o) =>
{
+ if (this.pingsSinceAck >= this.MissingPingsUntilDisconnect)
+ {
+ this.Disconnect($"Sent {this.pingsSinceAck} pings that remote has not responded to.");
+ return;
+ }
+
try
{
- ReliableSend((byte)UdpSendOption.Ping);
+ SendPing();
+ this.pingsSinceAck++;
}
catch
{
);
}
+ // Pings are special, quasi-reliable packets.
+ // We send them to trigger responses that validate our connection is alive
+ // They should never be the *cause* of a disconnect.
+ // Rather, the responses will reset our
+ void SendPing()
+ {
+ ushort id = (ushort)Interlocked.Increment(ref lastIDAllocated);
+
+ byte[] bytes = new byte[3];
+ bytes[0] = (byte)UdpSendOption.Ping;
+ bytes[1] = (byte)(id >> 8);
+ bytes[2] = (byte)id;
+
+ PingPacket pkt;
+ if (!this.activePingPackets.TryGetValue(id, out pkt))
+ {
+ pkt = PingPacket.GetObject();
+ if (!this.activePingPackets.TryAdd(id, pkt))
+ {
+ throw new Exception("This shouldn't be possible");
+ }
+ }
+
+ pkt.Stopwatch.Restart();
+
+ WriteBytesToConnection(bytes, bytes.Length);
+
+ Statistics.LogReliableSend(0, bytes.Length);
+ }
+
/// <summary>
/// Resets the keepalive timer to zero.
/// </summary>
}
}
}
-}
+}
\ No newline at end of file
void AttachReliableID(byte[] buffer, int offset, int sendLength, Action ackCallback = null)
{
//Find an ID not used yet.
- ushort id;
-
- //Create packet object
- Packet packet = Packet.GetObject();
+ ushort id = (ushort)Interlocked.Increment(ref lastIDAllocated);
- id = (ushort)Interlocked.Increment(ref lastIDAllocated);
-
- buffer[offset] = (byte)((id >> 8) & 0xFF);
+ buffer[offset] = (byte)(id >> 8);
buffer[offset + 1] = (byte)id;
+ //Create packet object
+ Packet packet = Packet.GetObject();
packet.Set(
id,
this,
Statistics.LogReliableSend(length, bytes.Length);
}
- void ReliableSend(byte sendOption)
- {
- byte[] bytes = new byte[3];
- bytes[0] = sendOption;
-
- //Add reliable ID
- AttachReliableID(bytes, 1, bytes.Length, null);
-
- //Write to connection
- WriteBytesToConnection(bytes, bytes.Length);
-
- Statistics.LogReliableSend(0, bytes.Length);
- }
-
/// <summary>
/// Handles a reliable message being received and invokes the data event.
/// </summary>
/// <param name="bytes">The buffer containing the data.</param>
void AcknowledgementMessageReceive(byte[] bytes)
{
+ this.pingsSinceAck = 0;
+
//Get ID
ushort id = (ushort)((bytes[1] << 8) + bytes[2]);
this.AveragePingMs = Math.Max(50, this.AveragePingMs * .7f + rt * .3f);
}
}
+ else if (this.activePingPackets.TryRemove(id, out PingPacket pingPkt))
+ {
+ float rt = pingPkt.Stopwatch.ElapsedMilliseconds;
+ lock (PingLock)
+ {
+ this.AveragePingMs = Math.Max(50, this.AveragePingMs * .7f + rt * .3f);
+ }
+
+ pingPkt.Recycle();
+ }
Statistics.LogReliableReceive(0, bytes.Length);
}
/// <param name="bytes">The bytes to write.</param>
protected abstract void WriteBytesToConnection(byte[] bytes, int length);
- /// <summary>
- /// Writes the given bytes to the connection synchronously.
- /// </summary>
- /// <param name="bytes">The bytes to write.</param>
- protected abstract void WriteBytesToConnectionSync(byte[] bytes, int length);
-
/// <inheritdoc/>
public override void Send(MessageWriter msg)
{
throw new HazelException("IPV6 not supported!");
this.socket = new Socket(AddressFamily.InterNetworkV6, SocketType.Dgram, ProtocolType.Udp);
- this.socket.SetSocketOption(SocketOptionLevel.IPv6, (SocketOptionName)27, false);
+ this.socket.SetSocketOption(SocketOptionLevel.IPv6, SocketOptionName.IPv6Only, false);
}
socket.ReceiveBufferSize = BufferSize;
var sock = kvp.Value;
sock.ManageReliablePackets();
}
-
- this.reliablePacketTimer.Change(100, Timeout.Infinite);
+
+ try
+ {
+ this.reliablePacketTimer.Change(100, Timeout.Infinite);
+ }
+ catch { }
}
/// <inheritdoc />
endPoint
);
}
- catch (SocketException e)
- {
- throw new HazelException("Could not send data as a SocketException occured.", e);
- }
- catch (ObjectDisposedException)
- {
- //Keep alive timer probably ran, ignore
- return;
- }
+ catch { }
}
/// <summary>
Listener.SendData(bytes, length, RemoteEndPoint);
}
- /// <inheritdoc />
- protected override void WriteBytesToConnectionSync(byte[] bytes, int length)
- {
- // No throw: As an internal interface, I want to try sending bytes whenever the I feel like it.
-
- Listener.SendDataSync(bytes, length, RemoteEndPoint);
- }
-
/// <inheritdoc />
/// <remarks>
/// This will always throw a HazelException.
{
try
{
- WriteBytesToConnection(DisconnectBytes, 1);
+ Listener.SendDataSync(DisconnectBytes, 1, RemoteEndPoint);
}
catch { }
}
projects / rhonda / impostor.hazel.git / commitdiff