commit d652bdc062aefec5267d47284a4db0b651c36fc9
parent b59b0e98dc9f339088acb5fc966143be8a2d9ce1
Author: Mikolaj Lenczewski <mikolaj.lenczewski308@gmail.com>
Date: Fri, 10 Apr 2020 12:08:54 +0100
Made udp server and clients asynchronous, using SocketAsyncEventArgs. Added lightweight ObjectPool implementation
Diffstat:
4 files changed, 90 insertions(+), 74 deletions(-)
diff --git a/NetSharp/NetSharp/Sockets/Datagram/DatagramSocketServer.cs b/NetSharp/NetSharp/Sockets/Datagram/DatagramSocketServer.cs
@@ -21,7 +21,7 @@ namespace NetSharp.Sockets.Datagram
private readonly ConcurrentDictionary<EndPoint, RemoteDatagramClientToken> connectedClientTokens;
- private readonly ObjectPool<SocketAsyncEventArgs> ArgsPool;
+ private readonly MyObjectPool<SocketAsyncEventArgs> ArgsPool;
private readonly struct RemoteDatagramClientToken
{
@@ -70,7 +70,7 @@ namespace NetSharp.Sockets.Datagram
return true;
}
- ArgsPool = new ObjectPool<SocketAsyncEventArgs>(CreateArgs, ResetArgs, DestroyArgs, ReBufferArgsPredicate);
+ ArgsPool = new MyObjectPool<SocketAsyncEventArgs>(CreateArgs, ResetArgs, DestroyArgs, ReBufferArgsPredicate);
}
protected override SocketAsyncEventArgs GenerateConnectionArgs(EndPoint remoteEndPoint)
@@ -229,6 +229,8 @@ namespace NetSharp.Sockets.Datagram
break;
case SocketAsyncOperation.ReceiveFrom:
+ DoReceiveFrom(AnyRemoteEndPoint); // start a new receive from operation immediately, to not drop any packets
+
CompleteReceiveFrom(args);
break;
@@ -281,14 +283,14 @@ namespace NetSharp.Sockets.Datagram
if (!completesAsync)
{
+ DoReceiveFrom(AnyRemoteEndPoint); // start a new receive from operation immediately, to not drop any packets
+
CompleteReceiveFrom(args);
}
}
private void CompleteReceiveFrom(SocketAsyncEventArgs receiveArgs)
{
- DoReceiveFrom(AnyRemoteEndPoint); // start a new receive from operation immediately, to not drop any packets
-
SocketOperationToken receiveToken = (SocketOperationToken)receiveArgs.UserToken;
TransmissionResult receiveResult = new TransmissionResult(receiveArgs);
@@ -535,54 +537,4 @@ namespace NetSharp.Sockets.Datagram
*/
}
}
-
- internal class ObjectPool<T> where T : class
- {
- internal delegate T CreateObjectDelegate();
-
- internal delegate bool KeepObjectPredicate(in T instance);
-
- internal delegate void ResetObjectDelegate(T instance);
-
- internal delegate void DestroyObjectDelegate(T instance);
-
- private readonly CreateObjectDelegate createObjectDelegate;
- private readonly KeepObjectPredicate rebufferObjectPredicate;
- private readonly ResetObjectDelegate resetObjectDelegate;
- private readonly DestroyObjectDelegate destroyObjectDelegate;
-
- private readonly ConcurrentBag<T> objectBuffer;
-
- public ObjectPool(in CreateObjectDelegate createDelegate, in ResetObjectDelegate resetDelegate, in DestroyObjectDelegate destroyDelegate, in KeepObjectPredicate keepObjectPredicate)
- {
- createObjectDelegate = createDelegate;
-
- resetObjectDelegate = resetDelegate;
-
- destroyObjectDelegate = destroyDelegate;
-
- rebufferObjectPredicate = keepObjectPredicate;
-
- objectBuffer = new ConcurrentBag<T>();
- }
-
- public T Rent()
- {
- return objectBuffer.TryTake(out T result) ? result : createObjectDelegate();
- }
-
- public void Return(T instance)
- {
- if (rebufferObjectPredicate(instance))
- {
- resetObjectDelegate(instance);
-
- objectBuffer.Add(instance);
- }
- else
- {
- destroyObjectDelegate(instance);
- }
- }
- }
}
\ No newline at end of file
diff --git a/NetSharp/NetSharp/Sockets/SocketClient.cs b/NetSharp/NetSharp/Sockets/SocketClient.cs
@@ -4,6 +4,8 @@ using System;
using System.Buffers;
using System.Net;
using System.Net.Sockets;
+using System.Threading.Tasks;
+using NetSharp.Utils;
namespace NetSharp.Sockets
{
@@ -11,38 +13,45 @@ namespace NetSharp.Sockets
{
protected readonly ArrayPool<byte> BufferPool;
+ protected readonly SocketAsyncEventArgs Args;
+
protected SocketClient(in AddressFamily connectionAddressFamily, in SocketType connectionSocketType, in ProtocolType connectionProtocolType)
: base(in connectionAddressFamily, in connectionSocketType, in connectionProtocolType)
{
BufferPool = ArrayPool<byte>.Create(NetworkPacket.TotalSize, 10);
+
+ Args = new SocketAsyncEventArgs();
+ Args.Completed += SocketAsyncOperations.HandleIoCompleted;
}
- public int SendBytes(Memory<byte> outgoingDataBuffer)
+ public int SendBytes(Memory<byte> outgoingDataBuffer, SocketFlags flags = SocketFlags.None)
{
- return connection.Send(outgoingDataBuffer.Span);
+ byte[] temporaryBuffer = BufferPool.Rent(NetworkPacket.TotalSize);
+ outgoingDataBuffer.CopyTo(temporaryBuffer);
+ int sentBytes = connection.Send(temporaryBuffer);
+ BufferPool.Return(temporaryBuffer);
+
+ return sentBytes;
}
- public int SendBytesTo(Memory<byte> outgoingDataBuffer, EndPoint remoteEndPoint)
+ public ValueTask<TransmissionResult> SendBytesTo(Memory<byte> outgoingDataBuffer, EndPoint remoteEndPoint, SocketFlags flags = SocketFlags.None)
{
- return connection.SendTo(outgoingDataBuffer.ToArray(), remoteEndPoint);
+ return SocketAsyncOperations.SendToAsync(Args, connection, remoteEndPoint, flags, outgoingDataBuffer);
}
- public int ReceiveBytes(Memory<byte> incomingDataBuffer)
+ public int ReceiveBytes(Memory<byte> incomingDataBuffer, SocketFlags flags = SocketFlags.None)
{
- byte[] temporaryBuffer = new byte[NetworkPacket.TotalSize];
+ byte[] temporaryBuffer = BufferPool.Rent(NetworkPacket.TotalSize);
int receivedBytes = connection.Receive(temporaryBuffer);
temporaryBuffer.CopyTo(incomingDataBuffer);
+ BufferPool.Return(temporaryBuffer);
return receivedBytes;
}
- public int ReceiveBytesFrom(Memory<byte> incomingDataBuffer, ref EndPoint remoteEndPoint)
+ public ValueTask<TransmissionResult> ReceiveBytesFrom(Memory<byte> incomingDataBuffer, ref EndPoint remoteEndPoint, SocketFlags flags = SocketFlags.None)
{
- byte[] temporaryBuffer = new byte[NetworkPacket.TotalSize];
- int receivedBytes = connection.ReceiveFrom(temporaryBuffer, ref remoteEndPoint);
- temporaryBuffer.CopyTo(incomingDataBuffer);
-
- return receivedBytes;
+ return SocketAsyncOperations.ReceiveFromAsync(Args, connection, remoteEndPoint, flags, incomingDataBuffer);
}
}
}
\ No newline at end of file
diff --git a/NetSharp/NetSharp/Utils/MyObjectPool.cs b/NetSharp/NetSharp/Utils/MyObjectPool.cs
@@ -0,0 +1,54 @@
+using System.Collections.Concurrent;
+
+namespace NetSharp.Utils
+{
+ internal class MyObjectPool<T> where T : class
+ {
+ internal delegate T CreateObjectDelegate();
+
+ internal delegate bool KeepObjectPredicate(in T instance);
+
+ internal delegate void ResetObjectDelegate(T instance);
+
+ internal delegate void DestroyObjectDelegate(T instance);
+
+ private readonly CreateObjectDelegate createObjectDelegate;
+ private readonly KeepObjectPredicate rebufferObjectPredicate;
+ private readonly ResetObjectDelegate resetObjectDelegate;
+ private readonly DestroyObjectDelegate destroyObjectDelegate;
+
+ private readonly ConcurrentBag<T> objectBuffer;
+
+ internal MyObjectPool(in CreateObjectDelegate createDelegate, in ResetObjectDelegate resetDelegate, in DestroyObjectDelegate destroyDelegate, in KeepObjectPredicate keepObjectPredicate)
+ {
+ createObjectDelegate = createDelegate;
+
+ resetObjectDelegate = resetDelegate;
+
+ destroyObjectDelegate = destroyDelegate;
+
+ rebufferObjectPredicate = keepObjectPredicate;
+
+ objectBuffer = new ConcurrentBag<T>();
+ }
+
+ internal T Rent()
+ {
+ return objectBuffer.TryTake(out T result) ? result : createObjectDelegate();
+ }
+
+ internal void Return(T instance)
+ {
+ if (rebufferObjectPredicate(instance))
+ {
+ resetObjectDelegate(instance);
+
+ objectBuffer.Add(instance);
+ }
+ else
+ {
+ destroyObjectDelegate(instance);
+ }
+ }
+ }
+}
+\ No newline at end of file
diff --git a/NetSharp/NetSharpExamples/Program.cs b/NetSharp/NetSharpExamples/Program.cs
@@ -13,7 +13,7 @@ using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
-
+using NetSharp.Utils;
using NetworkPacket = NetSharp.Packets.NetworkPacket;
using SocketServer = NetSharp.Sockets.SocketServer;
@@ -44,14 +44,14 @@ namespace NetSharpExamples
private static async Task TestSocketClient()
{
- const int clientCount = 20;
+ const int clientCount = 24;
const long packetsToSend = 100_000;
Task[] clientTasks = new Task[clientCount];
double[] clientBandwidths = new double[clientCount];
HashSet<int> activeTasks = new HashSet<int>(clientCount);
- async void ClientTask(object id)
+ async Task ClientTask(object id)
{
try
{
@@ -103,7 +103,7 @@ namespace NetSharpExamples
#if TCP
int sendResult = client.SendBytes(requestBufferMemory);
#else
- int sendResult = client.SendBytesTo(requestBufferMemory, ServerEndPoint);
+ TransmissionResult sendResult = await client.SendBytesTo(requestBufferMemory, ServerEndPoint);
#endif
bandwidthStopwatch.Stop();
@@ -112,7 +112,7 @@ namespace NetSharpExamples
#if DEBUG
lock (typeof(Console))
{
- Console.WriteLine($"[Client {id}, Packet {i}] Sent {sendResult} bytes to {ServerEndPoint}");
+ Console.WriteLine($"[Client {id}, Packet {i}] Sent {sendResult.Count} bytes to {ServerEndPoint}");
Console.WriteLine($"[Client {id}, Packet {i}] >>>> {Encoding.UTF8.GetString(requestBufferMemory.Span)}");
}
#endif
@@ -125,7 +125,7 @@ namespace NetSharpExamples
#if TCP
int receiveResult = client.ReceiveBytes(responseBufferMemory);
#else
- int receiveResult = client.ReceiveBytesFrom(responseBufferMemory, ref serverEndPoint);
+ TransmissionResult receiveResult = await client.ReceiveBytesFrom(responseBufferMemory, ref serverEndPoint);
#endif
bandwidthStopwatch.Stop();
@@ -134,7 +134,7 @@ namespace NetSharpExamples
#if DEBUG
lock (typeof(Console))
{
- Console.WriteLine($"[Client {id}, Packet {i}] Received {receiveResult} bytes from {serverEndPoint}");
+ Console.WriteLine($"[Client {id}, Packet {i}] Received {receiveResult.Count} bytes from {serverEndPoint}");
Console.WriteLine($"[Client {id}, Packet {i}] <<<< {Encoding.UTF8.GetString(responseBufferMemory.Span)}");
}
#endif
@@ -200,7 +200,7 @@ namespace NetSharpExamples
for (int clientId = 0; clientId < clientCount; clientId++)
{
- clientTasks[clientId] = Task.Factory.StartNew(ClientTask, clientId, TaskCreationOptions.LongRunning);
+ clientTasks[clientId] = Task.Factory.StartNew(ClientTask, clientId, TaskCreationOptions.LongRunning).Result;
}
await Task.WhenAll(clientTasks);
