Skip to content

Reduce send locking and add parallel-pipelining #50

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
willardf wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Reduce send locking and add parallel-pipelining #50

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
74ef777
Attempt to remove the lock around peer AES instances while sending me…
willardf Nov 1, 2022
0ea6f94
Fix up instability with thread static making inauthentic packets
willardf Nov 1, 2022
803c4fe
Make sure peers are cleaned up
willardf Nov 2, 2022
796f154
Remove a stray thread static and readd send parallelization
willardf Nov 2, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions Hazel.UnitTests/Dtls/AesGcmRecordProtectedTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,42 @@ public void ServerCanEncryptAndDecryptData()
}
}


[TestMethod]
public void ServerCanEncryptWithDuplicateInstances()
{
using (Aes128GcmRecordProtection recordProtection = new Aes128GcmRecordProtection(this.masterSecret, this.serverRandom, this.clientRandom))
using (IRecordProtection duplicateProtection = recordProtection.Duplicate())
{
byte[] messageAsBytes = Encoding.UTF8.GetBytes(TestMessage);

// I want to see that if instances are used at different rates, the outputs are deterministic
Record unusedRecord = new Record();
unusedRecord.Length = (ushort)recordProtection.GetEncryptedSize(messageAsBytes.Length);
ByteSpan unused = new byte[unusedRecord.Length];
recordProtection.EncryptClientPlaintext(unused, messageAsBytes, ref unusedRecord);

Record record = new Record();
record.ContentType = ContentType.ApplicationData;
record.ProtocolVersion = ProtocolVersion.DTLS1_2;
record.Epoch = 1;
record.SequenceNumber = 124;
record.Length = (ushort)recordProtection.GetEncryptedSize(messageAsBytes.Length);

ByteSpan encrypted1 = new byte[record.Length];
recordProtection.EncryptClientPlaintext(encrypted1, messageAsBytes, ref record);

ByteSpan encrypted2 = new byte[record.Length];
duplicateProtection.EncryptClientPlaintext(encrypted2, messageAsBytes, ref record);

Assert.AreEqual(encrypted1.Length, encrypted2.Length);
for (int i = 0; i < encrypted1.Length; i++)
{
Assert.AreEqual(encrypted1[i], encrypted2[i]);
}
}
}

[TestMethod]
public void ClientCanEncryptAndDecryptData()
{
Expand Down
52 changes: 24 additions & 28 deletions Hazel.UnitTests/Dtls/DtlsConnectionTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ private static X509Certificate2Collection GetCertificateForClient()

protected DtlsConnectionListener CreateListener(int numWorkers, IPEndPoint endPoint, ILogger logger, IPMode ipMode = IPMode.IPv4)
{
DtlsConnectionListener listener = new DtlsConnectionListener(2, endPoint, logger, ipMode);
DtlsConnectionListener listener = new DtlsConnectionListener(numWorkers, endPoint, logger, ipMode);
listener.SetCertificate(GetCertificateForServer());
return listener;

Expand All @@ -105,42 +105,38 @@ public void DtlsServerDisposeDisconnectsTest()
{
IPEndPoint ep = new IPEndPoint(IPAddress.Loopback, 27510);

bool serverConnected = false;
ManualResetEventSlim serverConnected = new ManualResetEventSlim();
bool serverDisconnected = false;
bool clientDisconnected = false;

Semaphore signal = new Semaphore(0, int.MaxValue);
ManualResetEventSlim clientDisconnected = new ManualResetEventSlim();

using (var listener = (DtlsConnectionListener)CreateListener(2, new IPEndPoint(IPAddress.Any, ep.Port), new TestLogger()))
using (var listener = CreateListener(2, new IPEndPoint(IPAddress.Any, ep.Port), new TestLogger()))
using (var connection = CreateConnection(ep, new TestLogger()))
{
listener.NewConnection += (evt) =>
{
serverConnected = true;
signal.Release();
evt.Connection.Disconnected += (o, et) => {
serverConnected.Set();
evt.Connection.Disconnected += (o, et) =>
{
serverDisconnected = true;
};
};
connection.Disconnected += (o, evt) => {
clientDisconnected = true;
signal.Release();

connection.Disconnected += (o, evt) =>
{
clientDisconnected.Set();
};

listener.Start();
connection.Connect();

// wait for the client to connect
signal.WaitOne(10);
Assert.IsTrue(serverConnected.Wait(1000), "Server never connected");

listener.Dispose();

// wait for the client to disconnect
signal.WaitOne(100);
Assert.IsTrue(clientDisconnected.Wait(5000), "Client never disconnected");
Assert.IsFalse(serverDisconnected, "Server shouldn't see a disconnect event when disposing.");

Assert.IsTrue(serverConnected);
Assert.IsTrue(clientDisconnected);
Assert.IsFalse(serverDisconnected);
Assert.AreEqual(0, listener.PeerCount);
}
}
Expand Down Expand Up @@ -221,7 +217,7 @@ public void TestMalformedApplicationData()
IPEndPoint ep = new IPEndPoint(IPAddress.Loopback, 27510);

IPEndPoint connectionEndPoint = ep;
DtlsConnectionListener.ConnectionId connectionId = new ThreadLimitedUdpConnectionListener.ConnectionId();
ConnectionId connectionId = new ConnectionId();

Semaphore signal = new Semaphore(0, int.MaxValue);

Expand Down Expand Up @@ -277,7 +273,7 @@ public void TestMalformedConnectionData()
IPEndPoint ep = new IPEndPoint(IPAddress.Loopback, 27510);

IPEndPoint connectionEndPoint = ep;
DtlsConnectionListener.ConnectionId connectionId = new ThreadLimitedUdpConnectionListener.ConnectionId();
ConnectionId connectionId = new ConnectionId();

Semaphore signal = new Semaphore(0, int.MaxValue);

Expand Down Expand Up @@ -963,8 +959,8 @@ public void DtlsIPv6ConnectionTest()
[TestMethod]
public void DtlsUnreliableServerToClientTest()
{
using (ThreadLimitedUdpConnectionListener listener = this.CreateListener(2, new IPEndPoint(IPAddress.Any, 4296), new TestLogger()))
using (UdpConnection connection = this.CreateConnection(new IPEndPoint(IPAddress.Loopback, 4296), new TestLogger()))
using (ThreadLimitedUdpConnectionListener listener = this.CreateListener(2, new IPEndPoint(IPAddress.Any, 4296), new TestLogger("Server")))
using (UdpConnection connection = this.CreateConnection(new IPEndPoint(IPAddress.Loopback, 4296), new TestLogger("Client")))
{
TestHelper.RunServerToClientTest(listener, connection, 10, SendOption.None);
}
Expand Down Expand Up @@ -1081,30 +1077,30 @@ public void ClientDisconnectTest()
using (var listener = this.CreateListener(2, new IPEndPoint(IPAddress.Any, 4296), new TestLogger("Server")))
using (var connection = this.CreateConnection(new IPEndPoint(IPAddress.Loopback, 4296), new TestLogger("Client")))
{
ManualResetEvent mutex = new ManualResetEvent(false);
ManualResetEvent mutex2 = new ManualResetEvent(false);
ManualResetEvent clientConnected = new ManualResetEvent(false);
ManualResetEvent clientDisconnected = new ManualResetEvent(false);

listener.NewConnection += delegate (NewConnectionEventArgs args)
{
args.Connection.Disconnected += delegate (object sender2, DisconnectedEventArgs args2)
{
mutex2.Set();
clientDisconnected.Set();
};

mutex.Set();
clientConnected.Set();
};

listener.Start();

connection.Connect();

Assert.AreEqual(ConnectionState.Connected, connection.State);
mutex.WaitOne(1000);
Assert.IsTrue(clientConnected.WaitOne(1000), "Client never connected");
Assert.AreEqual(ConnectionState.Connected, connection.State);

connection.Disconnect("Testing");

mutex2.WaitOne(1000);
Assert.IsTrue(clientDisconnected.WaitOne(1000), "Client never disconnected");
Assert.AreEqual(ConnectionState.NotConnected, connection.State);
}
}
Expand Down
156 changes: 24 additions & 132 deletions Hazel.UnitTests/TestHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,8 @@
using System;
using Hazel.Udp.FewerThreads;
using Microsoft.VisualStudio.TestTools.UnitTesting;

using Hazel;
using System.Net;
using System.Threading;
using System;
using System.Diagnostics;
using Hazel.Udp.FewerThreads;
using System.Threading;

namespace Hazel.UnitTests
{
Expand All @@ -17,90 +14,41 @@ public static class TestHelper
/// </summary>
/// <param name="listener">The listener to test.</param>
/// <param name="connection">The connection to test.</param>
internal static void RunServerToClientTest(ThreadLimitedUdpConnectionListener listener, Connection connection, int dataSize, SendOption sendOption)
internal static void RunServerToClientTest(NetworkConnectionListener listener, Connection connection, int dataSize, SendOption sendOption)
{
//Setup meta stuff
MessageWriter data = BuildData(sendOption, dataSize);
ManualResetEvent mutex = new ManualResetEvent(false);
ManualResetEvent clientConnected = new ManualResetEvent(false);
ManualResetEvent dataReceived = new ManualResetEvent(false);

Connection conn = null;
//Setup listener
listener.NewConnection += delegate (NewConnectionEventArgs ncArgs)
{
ncArgs.Connection.Send(data);
conn = ncArgs.Connection;
clientConnected.Set();
};

listener.Start();

//Setup connection
DataReceivedEventArgs? result = null;
//Setup conneciton
connection.DataReceived += delegate (DataReceivedEventArgs a)
{
Trace.WriteLine("Data was received correctly.");

try
{
result = a;
}
finally
{
mutex.Set();
}
};

connection.Connect();

//Wait until data is received
mutex.WaitOne();

var dataReader = ConvertToMessageReader(data);
Assert.AreEqual(dataReader.Length, result.Value.Message.Length);
for (int i = 0; i < dataReader.Length; i++)
{
Assert.AreEqual(dataReader.ReadByte(), result.Value.Message.ReadByte());
}

Assert.AreEqual(sendOption, result.Value.SendOption);
}

/// <summary>
/// Runs a general test on the given listener and connection.
/// </summary>
/// <param name="listener">The listener to test.</param>
/// <param name="connection">The connection to test.</param>
internal static void RunServerToClientTest(NetworkConnectionListener listener, Connection connection, int dataSize, SendOption sendOption)
{
//Setup meta stuff
MessageWriter data = BuildData(sendOption, dataSize);
ManualResetEvent mutex = new ManualResetEvent(false);

//Setup listener
listener.NewConnection += delegate (NewConnectionEventArgs ncArgs)
{
ncArgs.Connection.Send(data);
Console.WriteLine("Data was received correctly.");
result = a;
dataReceived.Set();
};

listener.Start();
connection.Connect();

DataReceivedEventArgs? result = null;
//Setup conneciton
connection.DataReceived += delegate (DataReceivedEventArgs a)
{
Trace.WriteLine("Data was received correctly.");
// Wait until data is received
Assert.IsTrue(clientConnected.WaitOne(1000), "Client never connected");

try
{
result = a;
}
finally
{
mutex.Set();
}
};
Assert.AreEqual(SendErrors.None, conn.Send(data));
Console.WriteLine("Data was sent correctly.");

connection.Connect();

//Wait until data is received
mutex.WaitOne();
// Wait until data is received
Assert.IsTrue(dataReceived.WaitOne(1000), "Data was never received");

var dataReader = ConvertToMessageReader(data);
Assert.AreEqual(dataReader.Length, result.Value.Message.Length);
Expand Down Expand Up @@ -130,27 +78,22 @@ internal static void RunClientToServerTest(NetworkConnectionListener listener, C
{
args.Connection.DataReceived += delegate (DataReceivedEventArgs innerArgs)
{
Trace.WriteLine("Data was received correctly.");

Console.WriteLine("Data was received correctly.");
result = innerArgs;

mutex2.Set();
};

mutex.Set();
};

listener.Start();

//Connect
connection.Connect();

mutex.WaitOne();
Assert.IsTrue(mutex.WaitOne(1000), "Client never connected");

connection.Send(data);
Assert.AreEqual(SendErrors.None, connection.Send(data));

//Wait until data is received
mutex2.WaitOne();
Assert.IsTrue(mutex2.WaitOne(1000), "Data was never received");

var dataReader = ConvertToMessageReader(data);
Assert.AreEqual(dataReader.Length, result.Value.Message.Length);
Expand All @@ -162,57 +105,6 @@ internal static void RunClientToServerTest(NetworkConnectionListener listener, C
Assert.AreEqual(sendOption, result.Value.SendOption);
}


/// <summary>
/// Runs a general test on the given listener and connection.
/// </summary>
/// <param name="listener">The listener to test.</param>
/// <param name="connection">The connection to test.</param>
internal static void RunClientToServerTest(ThreadLimitedUdpConnectionListener listener, Connection connection, int dataSize, SendOption sendOption)
{
//Setup meta stuff
MessageWriter data = BuildData(sendOption, dataSize);
ManualResetEvent mutex = new ManualResetEvent(false);
ManualResetEvent mutex2 = new ManualResetEvent(false);

//Setup listener
DataReceivedEventArgs? result = null;
listener.NewConnection += delegate (NewConnectionEventArgs args)
{
args.Connection.DataReceived += delegate (DataReceivedEventArgs innerArgs)
{
Trace.WriteLine("Data was received correctly.");

result = innerArgs;

mutex2.Set();
};

mutex.Set();
};

listener.Start();

//Connect
connection.Connect();

Assert.IsTrue(mutex.WaitOne(100), "Timeout while connecting");

connection.Send(data);

//Wait until data is received
Assert.IsTrue(mutex2.WaitOne(100), "Timeout while sending data");

var dataReader = ConvertToMessageReader(data);
Assert.AreEqual(dataReader.Length, result.Value.Message.Length);
for (int i = 0; i < dataReader.Length; i++)
{
Assert.AreEqual(dataReader.ReadByte(), result.Value.Message.ReadByte());
}

Assert.AreEqual(sendOption, result.Value.SendOption);
}

/// <summary>
/// Runs a server disconnect test on the given listener and connection.
/// </summary>
Expand Down
Loading