commit da9d92cdd92bfe14ac802c9eda7fc32ddca5c364
parent 0793d3083bc8126a125b6162f42020cbcf9e06a4
Author: Mikolaj Lenczewski <33129490+EnderRifter@users.noreply.github.com>
Date: Thu, 12 Mar 2020 13:07:10 +0000
Went back to basics, with a socket server and socket client.
Diffstat:
8 files changed, 905 insertions(+), 240 deletions(-)
diff --git a/NetSharp/NetSharp/Connection.cs b/NetSharp/NetSharp/Connection.cs
@@ -1,6 +1,7 @@
using System;
using System.Buffers;
using System.Collections.Concurrent;
+using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
@@ -21,7 +22,7 @@ namespace NetSharp
/// </summary>
public sealed partial class Connection : IDisposable
{
- private readonly ConcurrentDictionary<EndPoint, SocketAsyncEventArgs> datagramConnections;
+ private readonly HashSet<EndPoint> datagramConnections;
private readonly Channel<(EndPoint origin, Memory<byte> packet)> incomingPacketChannel;
/// <summary>
@@ -48,7 +49,7 @@ namespace NetSharp
/// </summary>
private readonly CancellationToken ServerShutdownToken;
- private readonly ConcurrentDictionary<EndPoint, SocketAsyncEventArgs> streamConnections;
+ private readonly ConcurrentDictionary<EndPoint, Socket> streamConnections;
/// <summary>
/// A logger object allowing for writing debug messages to an output stream.
@@ -71,8 +72,7 @@ namespace NetSharp
async Task StreamListenerWork(object clientArgsObj)
{
- SocketAsyncEventArgs clientArgs = (SocketAsyncEventArgs)clientArgsObj;
- Socket clientSocket = clientArgs.AcceptSocket;
+ Socket clientSocket = (Socket)clientArgsObj;
EndPoint clientEndPoint = clientSocket.RemoteEndPoint;
logger.LogMessage($"Client handler started for {clientEndPoint}");
@@ -97,6 +97,8 @@ namespace NetSharp
logger.LogMessage($"Client handler stopped for {clientEndPoint}");
+ await DoDisconnectAsync(clientSocket, cancellationToken);
+
clientSocket.Shutdown(SocketShutdown.Both);
clientSocket.Close(1);
}
@@ -105,15 +107,13 @@ namespace NetSharp
while (!cancellationToken.IsCancellationRequested)
{
- SocketAsyncEventArgs clientArgs = await DoAcceptAsync(streamSocket, cancellationToken);
-
- Socket clientSocket = clientArgs.AcceptSocket;
+ Socket clientSocket = await DoAcceptAsync(streamSocket, cancellationToken);
if (!streamConnections.ContainsKey(clientSocket.RemoteEndPoint))
{
- streamConnections[clientSocket.RemoteEndPoint] = clientArgs;
+ streamConnections[clientSocket.RemoteEndPoint] = clientSocket;
- await Task.Factory.StartNew(StreamListenerWork, clientArgs, ServerShutdownToken);
+ await Task.Factory.StartNew(StreamListenerWork, streamConnections[clientSocket.RemoteEndPoint], ServerShutdownToken);
}
else
{
@@ -132,8 +132,6 @@ namespace NetSharp
while (!cancellationToken.IsCancellationRequested)
{
- SocketAsyncEventArgs args = clientSocketArgsPool.Get();
-
// TODO: implement receive buffer pooling
byte[] receiveBuffer = new byte[NetworkPacket.PacketSize];
Memory<byte> receiveBufferMemory = new Memory<byte>(receiveBuffer);
@@ -142,14 +140,12 @@ namespace NetSharp
await DoReceiveFromAsync(datagramSocket, AnyRemoteEndPoint, SocketFlags.None,
receiveBufferMemory, cancellationToken);
- if (!datagramConnections.ContainsKey(result.RemoteEndPoint))
+ if (!datagramConnections.Contains(result.RemoteEndPoint))
{
- datagramConnections[result.RemoteEndPoint] = args;
+ datagramConnections.Add(result.RemoteEndPoint);
}
await incomingPacketChannel.Writer.WriteAsync((result.RemoteEndPoint, receiveBufferMemory), cancellationToken);
-
- clientSocketArgsPool.Return(args);
}
logger.LogMessage("Stopped datagram listener task.");
@@ -190,21 +186,21 @@ namespace NetSharp
Memory<byte> serialisedResponse = outgoingPacketPipeline.ProcessPacket(packet);
- if (datagramConnections.ContainsKey(destination))
+ if (streamConnections.ContainsKey(destination))
{
- await DoSendToAsync(datagramSocket, destination,
- SocketFlags.None, serialisedResponse, cancellationToken);
+ Socket streamConnection = streamConnections[destination];
+
+ await DoSendToAsync(streamConnection, destination, SocketFlags.None,
+ serialisedResponse, cancellationToken);
}
- else if (streamConnections.ContainsKey(destination))
+ else if (datagramConnections.Contains(destination))
{
- SocketAsyncEventArgs streamClientArgs = streamConnections[destination];
-
- await DoSendToAsync(streamClientArgs.AcceptSocket, destination,
- SocketFlags.None, serialisedResponse, cancellationToken);
+ await DoSendToAsync(datagramSocket, destination, SocketFlags.None,
+ serialisedResponse, cancellationToken);
}
else
{
- logger.LogWarning($"Dropping packet destined for unknown destination; {destination}");
+ logger.LogWarning($"Packet destined for unknown destination: {destination}");
}
}
@@ -250,12 +246,14 @@ namespace NetSharp
new LeakTrackingObjectPool<SocketAsyncEventArgs>(
new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(),
objectPoolSize));
-
- acceptArgsPool = new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), objectPoolSize);
- connectArgsPool = new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), objectPoolSize);
- disconnectArgsPool = new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), objectPoolSize);
- receiveArgsPool = new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), objectPoolSize);
- sendArgsPool = new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), objectPoolSize);
+ receiveArgsPool =
+ new LeakTrackingObjectPool<SocketAsyncEventArgs>(
+ new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(),
+ objectPoolSize));
+ sendArgsPool =
+ new LeakTrackingObjectPool<SocketAsyncEventArgs>(
+ new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(),
+ objectPoolSize));
for (int i = 0; i < objectPoolSize; i++)
{
@@ -263,18 +261,6 @@ namespace NetSharp
clientArgs.Completed += HandleIOCompleted;
clientSocketArgsPool.Return(clientArgs);
- SocketAsyncEventArgs acceptArgs = new SocketAsyncEventArgs();
- acceptArgs.Completed += HandleIOCompleted;
- acceptArgsPool.Return(acceptArgs);
-
- SocketAsyncEventArgs connectArgs = new SocketAsyncEventArgs();
- connectArgs.Completed += HandleIOCompleted;
- connectArgsPool.Return(connectArgs);
-
- SocketAsyncEventArgs disconnectArgs = new SocketAsyncEventArgs();
- disconnectArgs.Completed += HandleIOCompleted;
- disconnectArgsPool.Return(disconnectArgs);
-
SocketAsyncEventArgs receiveArgs = new SocketAsyncEventArgs();
receiveArgs.Completed += HandleIOCompleted;
receiveArgsPool.Return(receiveArgs);
@@ -289,8 +275,8 @@ namespace NetSharp
//TODO: Preallocate buffers someday
}
- streamConnections = new ConcurrentDictionary<EndPoint, SocketAsyncEventArgs>();
- datagramConnections = new ConcurrentDictionary<EndPoint, SocketAsyncEventArgs>();
+ streamConnections = new ConcurrentDictionary<EndPoint, Socket>();
+ datagramConnections = new HashSet<EndPoint>();
this.incomingPacketPipeline = incomingPacketPipeline;
BoundedChannelOptions incomingChannelOptions = new BoundedChannelOptions(MaximumPacketBacklog)
diff --git a/NetSharp/NetSharp/ConnectionBase.cs b/NetSharp/NetSharp/ConnectionBase.cs
@@ -23,16 +23,10 @@ namespace NetSharp
/// </summary>
private static readonly EndPoint AnyRemoteEndPoint = new IPEndPoint(IPAddress.Any, 0);
- private readonly ObjectPool<SocketAsyncEventArgs> acceptArgsPool;
-
private readonly ObjectPool<SocketAsyncEventArgs> clientSocketArgsPool;
- private readonly ObjectPool<SocketAsyncEventArgs> connectArgsPool;
-
private readonly Socket datagramSocket;
- private readonly ObjectPool<SocketAsyncEventArgs> disconnectArgsPool;
-
private readonly ObjectPool<SocketAsyncEventArgs> receiveArgsPool;
private readonly ArrayPool<byte> receiveFromBufferPool;
@@ -46,12 +40,6 @@ namespace NetSharp
private readonly Socket streamSocket;
/// <summary>
- /// Socket async event args object used when the connection instance is used as a client. It is set when a
- /// call to <see cref="TryConnectAsync"/> is made.
- /// </summary>
- private volatile SocketAsyncEventArgs? connectionAsyncEventArgs;
-
- /// <summary>
/// Disposes of the managed and unmanaged resources held by this instance.
/// </summary>
/// <param name="disposing">Whether this method is called by <see cref="Dispose()"/> or by the finaliser.</param>
@@ -73,34 +61,47 @@ namespace NetSharp
/// <param name="serverSocket">The socket which should be used to accept an incoming connection attempt.</param>
/// <param name="cancellationToken">The cancellation token to observe for the operation.</param>
/// <returns>The accepted socket.</returns>
- private Task<SocketAsyncEventArgs> DoAcceptAsync(Socket serverSocket, CancellationToken cancellationToken = default)
+ private async Task<Socket> DoAcceptAsync(Socket serverSocket, CancellationToken cancellationToken = default)
{
- TaskCompletionSource<SocketAsyncEventArgs> tcs = new TaskCompletionSource<SocketAsyncEventArgs>();
+ TaskCompletionSource<Socket> tcs = new TaskCompletionSource<Socket>();
+
+ cancellationToken.Register(() => tcs.SetCanceled());
- SocketAsyncEventArgs acceptArgs = acceptArgsPool.Get();
- acceptArgs.AcceptSocket = null;
- acceptArgs.UserToken = new AsyncAcceptToken(tcs, cancellationToken);
+ Task<Socket> task = serverSocket.AcceptAsync();
+ Task<Socket> completedTask = await Task.WhenAny(task, tcs.Task);
- // if the accept operation doesn't complete synchronously, return the awaitable task
- return serverSocket.AcceptAsync(acceptArgs) ? tcs.Task : Task.FromResult(acceptArgs);
+ if (completedTask == task)
+ {
+ Socket result = await task;
+
+ tcs.SetResult(result);
+ }
+
+ return await tcs.Task;
}
/// <summary>
/// Provides an awaitable wrapper around an asynchronous socket connect operation.
/// </summary>
- /// <param name="disconnectedSocket">The socket which should asynchronously connect to the remote endpoint.</param>
+ /// <param name="socket">The socket which should asynchronously connect to the remote endpoint.</param>
/// <param name="remoteEndPoint">The remote endpoint to which the socket should connect.</param>
/// <param name="cancellationToken">The cancellation token to observe for the operation.</param>
- private Task<SocketAsyncEventArgs> DoConnectAsync(Socket disconnectedSocket, EndPoint remoteEndPoint, CancellationToken cancellationToken = default)
+ private async Task DoConnectAsync(Socket socket, EndPoint remoteEndPoint, CancellationToken cancellationToken = default)
{
- TaskCompletionSource<SocketAsyncEventArgs> tcs = new TaskCompletionSource<SocketAsyncEventArgs>();
+ TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
+
+ cancellationToken.Register(() => tcs.SetCanceled());
- SocketAsyncEventArgs connectArgs = connectArgsPool.Get();
- connectArgs.RemoteEndPoint = remoteEndPoint;
- connectArgs.UserToken = new AsyncConnectToken(tcs, cancellationToken);
+ Task task = socket.ConnectAsync(remoteEndPoint);
+ Task completedTask = await Task.WhenAny(task, tcs.Task);
- // if the connect operation doesn't complete synchronously, return the awaitable task
- return disconnectedSocket.ConnectAsync(connectArgs) ? tcs.Task : Task.FromResult(connectArgs);
+ if (completedTask == task)
+ {
+ await task;
+ tcs.SetResult(true);
+ }
+
+ await tcs.Task;
}
/// <summary>
@@ -110,21 +111,10 @@ namespace NetSharp
/// <param name="cancellationToken">The cancellation token to observe for the operation.</param>
private Task DoDisconnectAsync(Socket connectedSocket, CancellationToken cancellationToken = default)
{
- TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
-
- SocketAsyncEventArgs disconnectArgs = disconnectArgsPool.Get();
- disconnectArgs.DisconnectReuseSocket = true;
- disconnectArgs.UserToken = new AsyncDisconnectToken(tcs, cancellationToken);
-
- // if the disconnect operation doesn't complete synchronously, return the awaitable task
- if (connectedSocket.DisconnectAsync(disconnectArgs))
+ return Task.Factory.StartNew(() =>
{
- return tcs.Task;
- }
-
- disconnectArgsPool.Return(disconnectArgs);
-
- return Task.CompletedTask;
+ connectedSocket.Disconnect(true);
+ }, cancellationToken);
}
/// <summary>
@@ -221,74 +211,6 @@ namespace NetSharp
{
switch (args.LastOperation)
{
- case SocketAsyncOperation.Accept:
- AsyncAcceptToken asyncAcceptToken = (AsyncAcceptToken)args.UserToken;
-
- if (asyncAcceptToken.CancellationToken.IsCancellationRequested)
- {
- asyncAcceptToken.CompletionSource.SetCanceled();
- }
- else
- {
- if (args.SocketError != SocketError.Success)
- {
- asyncAcceptToken.CompletionSource.SetException(
- new SocketException((int)args.SocketError));
- }
- else
- {
- asyncAcceptToken.CompletionSource.SetResult(args);
- }
- }
-
- break;
-
- case SocketAsyncOperation.Connect:
- AsyncConnectToken asyncConnectToken = (AsyncConnectToken)args.UserToken;
-
- if (asyncConnectToken.CancellationToken.IsCancellationRequested)
- {
- asyncConnectToken.CompletionSource.SetCanceled();
- }
- else
- {
- if (args.SocketError != SocketError.Success)
- {
- asyncConnectToken.CompletionSource.SetException(
- new SocketException((int)args.SocketError));
- }
- else
- {
- asyncConnectToken.CompletionSource.SetResult(args);
- }
- }
-
- break;
-
- case SocketAsyncOperation.Disconnect:
- AsyncDisconnectToken asyncDisconnectToken = (AsyncDisconnectToken)args.UserToken;
-
- if (asyncDisconnectToken.CancellationToken.IsCancellationRequested)
- {
- asyncDisconnectToken.CompletionSource.SetCanceled();
- }
- else
- {
- if (args.SocketError != SocketError.Success)
- {
- asyncDisconnectToken.CompletionSource.SetException(
- new SocketException((int)args.SocketError));
- }
- else
- {
- asyncDisconnectToken.CompletionSource.SetResult(true);
- }
- }
-
- disconnectArgsPool.Return(args);
-
- break;
-
case SocketAsyncOperation.SendTo:
AsyncWriteToken asyncSendToToken = (AsyncWriteToken)args.UserToken;
@@ -355,42 +277,6 @@ namespace NetSharp
}
}
- private readonly struct AsyncAcceptToken
- {
- public readonly CancellationToken CancellationToken;
- public readonly TaskCompletionSource<SocketAsyncEventArgs> CompletionSource;
-
- public AsyncAcceptToken(TaskCompletionSource<SocketAsyncEventArgs> tcs, CancellationToken cancellationToken = default)
- {
- CompletionSource = tcs;
- CancellationToken = cancellationToken;
- }
- }
-
- private readonly struct AsyncConnectToken
- {
- public readonly CancellationToken CancellationToken;
- public readonly TaskCompletionSource<SocketAsyncEventArgs> CompletionSource;
-
- public AsyncConnectToken(TaskCompletionSource<SocketAsyncEventArgs> tcs, CancellationToken cancellationToken = default)
- {
- CompletionSource = tcs;
- CancellationToken = cancellationToken;
- }
- }
-
- private readonly struct AsyncDisconnectToken
- {
- public readonly CancellationToken CancellationToken;
- public readonly TaskCompletionSource<bool> CompletionSource;
-
- public AsyncDisconnectToken(TaskCompletionSource<bool> tcs, CancellationToken cancellationToken = default)
- {
- CompletionSource = tcs;
- CancellationToken = cancellationToken;
- }
- }
-
private readonly struct AsyncReadToken
{
public readonly CancellationToken CancellationToken;
@@ -450,13 +336,7 @@ namespace NetSharp
using CancellationTokenSource cts =
CancellationTokenSource.CreateLinkedTokenSource(timeoutCancellationTokenSource.Token, ServerShutdownToken);
- if (connectionAsyncEventArgs == null)
- {
- throw new ConstraintException(
- $"{nameof(TryConnectAsync)} has not yet been called, or a valid connection has not been made.");
- }
-
- return DoReceiveFromAsync(connectionAsyncEventArgs.ConnectSocket, connectionAsyncEventArgs.ConnectSocket.RemoteEndPoint, flags, inputBuffer, cts.Token);
+ return DoReceiveFromAsync(streamSocket, streamSocket.RemoteEndPoint, flags, inputBuffer, cts.Token);
}
public Task<TransmissionResult> ReceiveFromAsync(EndPoint remoteEndPoint, Memory<byte> inputBuffer, SocketFlags flags, TimeSpan timeout)
@@ -474,13 +354,7 @@ namespace NetSharp
using CancellationTokenSource cts =
CancellationTokenSource.CreateLinkedTokenSource(timeoutCancellationTokenSource.Token, ServerShutdownToken);
- if (connectionAsyncEventArgs == null)
- {
- throw new ConstraintException(
- $"{nameof(TryConnectAsync)} has not yet been called, or a valid connection has not been made.");
- }
-
- return DoSendToAsync(connectionAsyncEventArgs.ConnectSocket, connectionAsyncEventArgs.ConnectSocket.RemoteEndPoint, flags, outputBuffer, cts.Token);
+ return DoSendToAsync(streamSocket, streamSocket.RemoteEndPoint, flags, outputBuffer, cts.Token);
}
public ValueTask<int> SendToAsync(EndPoint remoteEndPoint, Memory<byte> outputBuffer, SocketFlags flags, TimeSpan timeout)
@@ -548,7 +422,7 @@ namespace NetSharp
try
{
- connectionAsyncEventArgs = await DoConnectAsync(streamSocket, remoteEndPoint, cts.Token);
+ await DoConnectAsync(streamSocket, remoteEndPoint, cts.Token);
return true;
}
@@ -571,13 +445,11 @@ namespace NetSharp
try
{
- if (connectionAsyncEventArgs == null) return false;
+ await DoDisconnectAsync(streamSocket, cts.Token);
streamSocket.Shutdown(SocketShutdown.Both);
streamSocket.Close(1);
- await DoDisconnectAsync(streamSocket, cts.Token);
-
return true;
}
catch (TaskCanceledException)
diff --git a/NetSharp/NetSharp/Extensions/ConnectionExtensions.cs b/NetSharp/NetSharp/Extensions/ConnectionExtensions.cs
@@ -1,5 +1,4 @@
using System;
-using System.ComponentModel;
using System.Net;
using System.Net.Sockets;
using System.Threading;
diff --git a/NetSharp/NetSharp/NetSharp.xml b/NetSharp/NetSharp/NetSharp.xml
@@ -59,12 +59,6 @@
Represents any remote endpoint for datagram operations.
</summary>
</member>
- <member name="F:NetSharp.Connection.connectionAsyncEventArgs">
- <summary>
- Socket async event args object used when the connection instance is used as a client. It is set when a
- call to <see cref="M:NetSharp.Connection.TryConnectAsync(System.Net.EndPoint,System.TimeSpan)"/> is made.
- </summary>
- </member>
<member name="M:NetSharp.Connection.Dispose(System.Boolean)">
<summary>
Disposes of the managed and unmanaged resources held by this instance.
@@ -83,7 +77,7 @@
<summary>
Provides an awaitable wrapper around an asynchronous socket connect operation.
</summary>
- <param name="disconnectedSocket">The socket which should asynchronously connect to the remote endpoint.</param>
+ <param name="socket">The socket which should asynchronously connect to the remote endpoint.</param>
<param name="remoteEndPoint">The remote endpoint to which the socket should connect.</param>
<param name="cancellationToken">The cancellation token to observe for the operation.</param>
</member>
@@ -2048,6 +2042,27 @@
<param name="socket">The socket which should asynchronously disconnect from its remote endpoint.</param>
<param name="cancellationToken">The cancellation token to observe for the operation.</param>
</member>
+ <member name="M:NetSharp.Sockets.SocketClient.Finalize">
+ <summary>
+ Destroys a socket client instance.
+ </summary>
+ </member>
+ <member name="M:NetSharp.Sockets.SocketClient.Dispose(System.Boolean)">
+ <summary>
+ Implementation of dispose pattern.
+ </summary>
+ <param name="disposing">
+ Whether this method is being called by the object finalizer, or by the <see cref="M:NetSharp.Sockets.SocketClient.Dispose"/> method.
+ </param>
+ </member>
+ <member name="M:NetSharp.Sockets.SocketClient.Dispose">
+ <inheritdoc />
+ </member>
+ <member name="T:NetSharp.Sockets.SocketOperations">
+ <summary>
+ Provides helper awaitable functions for wrapping the <see cref="T:System.Net.Sockets.SocketAsyncEventArgs"/> pattern.
+ </summary>
+ </member>
<member name="T:NetSharp.Sockets.SocketReader">
<summary>
Helper class providing awaitable wrappers around asynchronous Receive and ReceiveFrom operations.
@@ -2064,6 +2079,27 @@
<param name="cancellationToken">The cancellation token to observe for the operation.</param>
<returns>The result of the receive operation.</returns>
</member>
+ <member name="M:NetSharp.Sockets.SocketServer.Finalize">
+ <summary>
+ Destroys a socket server instance.
+ </summary>
+ </member>
+ <member name="F:NetSharp.Sockets.SocketServer.listenerSocket">
+ <summary>
+ The socket which should be used to listen for incoming data and to send outgoing data.
+ </summary>
+ </member>
+ <member name="M:NetSharp.Sockets.SocketServer.Dispose(System.Boolean)">
+ <summary>
+ Implementation of dispose pattern.
+ </summary>
+ <param name="disposing">
+ Whether this method is being called by the object finalizer, or by the <see cref="M:NetSharp.Sockets.SocketServer.Dispose"/> method.
+ </param>
+ </member>
+ <member name="M:NetSharp.Sockets.SocketServer.Dispose">
+ <inheritdoc />
+ </member>
<member name="T:NetSharp.Sockets.SocketWriter">
<summary>
Helper class providing awaitable wrappers around asynchronous Send and SendTo operations.
diff --git a/NetSharp/NetSharp/Sockets/SocketClient.cs b/NetSharp/NetSharp/Sockets/SocketClient.cs
@@ -0,0 +1,92 @@
+using System;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.ObjectPool;
+using NetSharp.Utils;
+
+namespace NetSharp.Sockets
+{
+ public class SocketClient : IDisposable
+ {
+ private readonly ObjectPool<SocketAsyncEventArgs> transmissionArgsPool;
+
+ /// <summary>
+ /// Destroys a socket client instance.
+ /// </summary>
+ ~SocketClient()
+ {
+ Dispose(false);
+ }
+
+ protected readonly Socket transmitterSocket;
+
+ /// <summary>
+ /// Implementation of dispose pattern.
+ /// </summary>
+ /// <param name="disposing">
+ /// Whether this method is being called by the object finalizer, or by the <see cref="Dispose()"/> method.
+ /// </param>
+ protected virtual void Dispose(bool disposing)
+ {
+ if (disposing)
+ {
+ transmitterSocket.Dispose();
+ }
+ }
+
+ public SocketClient(AddressFamily transmitterAddressFamily, SocketType transmitterSocketType,
+ ProtocolType transmitterProtocolType)
+ {
+ transmitterSocket = new Socket(transmitterAddressFamily, transmitterSocketType, transmitterProtocolType);
+
+ transmissionArgsPool = new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>());
+ }
+
+ /// <inheritdoc />
+ public void Dispose()
+ {
+ Dispose(true);
+ GC.SuppressFinalize(this);
+ }
+
+ public ValueTask<TransmissionResult> ReceiveAsync(EndPoint remoteEndPoint, SocketFlags receiveFlags, Memory<byte> receiveBuffer,
+ CancellationToken cancellationToken = default)
+ {
+ return SocketOperations.ReceiveFromAsync(transmissionArgsPool, transmitterSocket, remoteEndPoint,
+ receiveFlags, receiveBuffer, cancellationToken);
+ }
+
+ public ValueTask<TransmissionResult> SendAsync(EndPoint remoteEndPoint, SocketFlags sendFlags, Memory<byte> sendBuffer,
+ CancellationToken cancellationToken = default)
+ {
+ return SocketOperations.SendToAsync(transmissionArgsPool, transmitterSocket, remoteEndPoint, sendFlags,
+ sendBuffer, cancellationToken);
+ }
+
+ public Task<bool> TryBindAsync(EndPoint localEndPoint, TimeSpan timeout)
+ {
+ using CancellationTokenSource cts = new CancellationTokenSource(timeout);
+
+ try
+ {
+ return Task.Run(() =>
+ {
+ transmitterSocket.Bind(localEndPoint);
+
+ return true;
+ }, cts.Token);
+ }
+ catch (TaskCanceledException)
+ {
+ return Task.FromResult(false);
+ }
+ catch (SocketException ex)
+ {
+ Console.WriteLine($"Socket exception on binding socket to {localEndPoint}: {ex}");
+ return Task.FromResult(false);
+ }
+ }
+ }
+}
+\ No newline at end of file
diff --git a/NetSharp/NetSharp/Sockets/SocketOperations.cs b/NetSharp/NetSharp/Sockets/SocketOperations.cs
@@ -0,0 +1,427 @@
+using System;
+using System.Buffers;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.ObjectPool;
+using NetSharp.Utils;
+
+namespace NetSharp.Sockets
+{
+ /// <summary>
+ /// Provides helper awaitable functions for wrapping the <see cref="SocketAsyncEventArgs"/> pattern.
+ /// </summary>
+ public static class SocketOperations
+ {
+ private static void HandleIOCompleted(object? sender, SocketAsyncEventArgs args)
+ {
+ args.Completed -= HandleIOCompleted;
+
+ switch (args.LastOperation)
+ {
+ case SocketAsyncOperation.ReceiveFrom:
+ AsyncReadToken asyncReceiveFromToken = (AsyncReadToken)args.UserToken;
+
+ if (asyncReceiveFromToken.CancellationToken.IsCancellationRequested)
+ {
+ asyncReceiveFromToken.CompletionSource.SetCanceled();
+ }
+ else
+ {
+ if (args.SocketError != SocketError.Success)
+ {
+ asyncReceiveFromToken.CompletionSource.SetException(
+ new SocketException((int)args.SocketError));
+ }
+ else
+ {
+ TransmissionResult result = new TransmissionResult(args);
+
+ asyncReceiveFromToken.CompletionSource.SetResult(result);
+ }
+ }
+
+ asyncReceiveFromToken.ReadArgsPool.Return(args);
+
+ break;
+
+ case SocketAsyncOperation.SendTo:
+ AsyncWriteToken asyncSendToToken = (AsyncWriteToken)args.UserToken;
+
+ if (asyncSendToToken.CancellationToken.IsCancellationRequested)
+ {
+ asyncSendToToken.CompletionSource.SetCanceled();
+ }
+ else
+ {
+ if (args.SocketError != SocketError.Success)
+ {
+ asyncSendToToken.CompletionSource.SetException(
+ new SocketException((int)args.SocketError));
+ }
+ else
+ {
+ TransmissionResult result = new TransmissionResult(args);
+
+ asyncSendToToken.CompletionSource.SetResult(result);
+ }
+ }
+
+ asyncSendToToken.WriteArgsPool.Return(args);
+ break;
+
+ case SocketAsyncOperation.Accept:
+ AsyncAcceptToken asyncAcceptToken = (AsyncAcceptToken)args.UserToken;
+
+ if (asyncAcceptToken.CancellationToken.IsCancellationRequested)
+ {
+ asyncAcceptToken.CompletionSource.SetCanceled();
+ }
+ else
+ {
+ if (args.SocketError != SocketError.Success)
+ {
+ asyncAcceptToken.CompletionSource.SetException(
+ new SocketException((int)args.SocketError));
+ }
+ else
+ {
+ asyncAcceptToken.CompletionSource.SetResult(args.AcceptSocket);
+ }
+ }
+
+ asyncAcceptToken.AcceptArgsPool.Return(args);
+
+ break;
+
+ case SocketAsyncOperation.Connect:
+ AsyncConnectToken asyncConnectToken = (AsyncConnectToken)args.UserToken;
+
+ if (asyncConnectToken.CancellationToken.IsCancellationRequested)
+ {
+ asyncConnectToken.CompletionSource.SetCanceled();
+ }
+ else
+ {
+ if (args.SocketError != SocketError.Success)
+ {
+ asyncConnectToken.CompletionSource.SetException(
+ new SocketException((int)args.SocketError));
+ }
+ else
+ {
+ asyncConnectToken.CompletionSource.SetResult(args.ConnectSocket);
+ }
+ }
+
+ asyncConnectToken.ConnectArgsPool.Return(args);
+
+ break;
+
+ case SocketAsyncOperation.Disconnect:
+ AsyncOperationToken asyncDisconnectToken = (AsyncOperationToken)args.UserToken;
+
+ if (asyncDisconnectToken.CancellationToken.IsCancellationRequested)
+ {
+ asyncDisconnectToken.CompletionSource.SetCanceled();
+ }
+ else
+ {
+ if (args.SocketError != SocketError.Success)
+ {
+ asyncDisconnectToken.CompletionSource.SetException(
+ new SocketException((int)args.SocketError));
+ }
+ else
+ {
+ asyncDisconnectToken.CompletionSource.SetResult(true);
+ }
+ }
+
+ asyncDisconnectToken.OperationArgsPool.Return(args);
+
+ break;
+
+ default:
+ throw new InvalidOperationException(
+ $"The {nameof(SocketReader)} class doesn't support the {args.LastOperation} operation.");
+ }
+ }
+
+ private readonly struct AsyncAcceptToken
+ {
+ public readonly ObjectPool<SocketAsyncEventArgs> AcceptArgsPool;
+ public readonly CancellationToken CancellationToken;
+ public readonly TaskCompletionSource<Socket> CompletionSource;
+
+ public AsyncAcceptToken(in ObjectPool<SocketAsyncEventArgs> argsPool, in TaskCompletionSource<Socket> tcs,
+ in CancellationToken cancellationToken = default)
+ {
+ AcceptArgsPool = argsPool;
+
+ CompletionSource = tcs;
+ CancellationToken = cancellationToken;
+ }
+ }
+
+ private readonly struct AsyncConnectToken
+ {
+ public readonly CancellationToken CancellationToken;
+ public readonly TaskCompletionSource<Socket> CompletionSource;
+ public readonly ObjectPool<SocketAsyncEventArgs> ConnectArgsPool;
+
+ public AsyncConnectToken(in ObjectPool<SocketAsyncEventArgs> argsPool, in TaskCompletionSource<Socket> tcs,
+ in CancellationToken cancellationToken = default)
+ {
+ ConnectArgsPool = argsPool;
+
+ CompletionSource = tcs;
+ CancellationToken = cancellationToken;
+ }
+ }
+
+ private readonly struct AsyncOperationToken
+ {
+ public readonly CancellationToken CancellationToken;
+ public readonly TaskCompletionSource<bool> CompletionSource;
+ public readonly ObjectPool<SocketAsyncEventArgs> OperationArgsPool;
+
+ public AsyncOperationToken(in ObjectPool<SocketAsyncEventArgs> argsPool, in TaskCompletionSource<bool> tcs,
+ in CancellationToken cancellationToken = default)
+ {
+ OperationArgsPool = argsPool;
+
+ CompletionSource = tcs;
+ CancellationToken = cancellationToken;
+ }
+ }
+
+ private readonly struct AsyncReadToken
+ {
+ public readonly CancellationToken CancellationToken;
+ public readonly TaskCompletionSource<TransmissionResult> CompletionSource;
+ public readonly ObjectPool<SocketAsyncEventArgs> ReadArgsPool;
+ public readonly Memory<byte> UserBuffer;
+
+ public AsyncReadToken(in ObjectPool<SocketAsyncEventArgs> argsPool, in Memory<byte> userBuffer,
+ in TaskCompletionSource<TransmissionResult> tcs, in CancellationToken cancellationToken = default)
+ {
+ ReadArgsPool = argsPool;
+ UserBuffer = userBuffer;
+
+ CompletionSource = tcs;
+ CancellationToken = cancellationToken;
+ }
+ }
+
+ private readonly struct AsyncWriteToken
+ {
+ public readonly CancellationToken CancellationToken;
+ public readonly TaskCompletionSource<TransmissionResult> CompletionSource;
+ public readonly ObjectPool<SocketAsyncEventArgs> WriteArgsPool;
+
+ public AsyncWriteToken(in ObjectPool<SocketAsyncEventArgs> argsPool,
+ in TaskCompletionSource<TransmissionResult> tcs, in CancellationToken cancellationToken = default)
+ {
+ WriteArgsPool = argsPool;
+
+ CompletionSource = tcs;
+ CancellationToken = cancellationToken;
+ }
+ }
+
+ public static Task<Socket> AcceptAsync(ObjectPool<SocketAsyncEventArgs> acceptArgsPool,
+ Socket socket, CancellationToken cancellationToken = default)
+ {
+ TaskCompletionSource<Socket> tcs = new TaskCompletionSource<Socket>();
+
+ SocketAsyncEventArgs args = acceptArgsPool.Get();
+ args.UserToken = new AsyncAcceptToken(acceptArgsPool, tcs, cancellationToken);
+
+ /*
+ // register cleanup action for when the cancellation token is thrown
+ cancellationToken.Register(() =>
+ {
+ tcs.SetCanceled();
+
+ //TODO this is probably a hideous solution. find a better one
+ args.Completed -= HandleIOCompleted;
+ args.Dispose();
+
+ SocketAsyncEventArgs newArgs = new SocketAsyncEventArgs();
+ newArgs.Completed += HandleIOCompleted;
+ acceptAsyncEventArgsPool.Return(newArgs);
+ });
+ */
+
+ // if the accept operation doesn't complete synchronously, return the awaitable task
+ if (socket.AcceptAsync(args)) return tcs.Task;
+
+ Socket result = args.AcceptSocket;
+ args.Completed -= HandleIOCompleted;
+
+ acceptArgsPool.Return(args);
+
+ return Task.FromResult(result);
+ }
+
+ public static Task<Socket> ConnectAsync(ObjectPool<SocketAsyncEventArgs> connectArgsPool,
+ Socket socket, EndPoint remoteEndPoint, CancellationToken cancellationToken = default)
+ {
+ TaskCompletionSource<Socket> tcs = new TaskCompletionSource<Socket>();
+
+ SocketAsyncEventArgs args = connectArgsPool.Get();
+ args.RemoteEndPoint = remoteEndPoint;
+ args.UserToken = new AsyncConnectToken(connectArgsPool, tcs, cancellationToken);
+
+ /*
+ // register cleanup action for when the cancellation token is thrown
+ cancellationToken.Register(() =>
+ {
+ tcs.SetCanceled();
+
+ //TODO this is probably a hideous solution. find a better one
+ args.Completed -= HandleIOCompleted;
+ args.Dispose();
+
+ SocketAsyncEventArgs newArgs = new SocketAsyncEventArgs();
+ newArgs.Completed += HandleIOCompleted;
+ connectAsyncEventArgsPool.Return(newArgs);
+ });
+ */
+
+ // if the connect operation doesn't complete synchronously, return the awaitable task
+ if (socket.ConnectAsync(args)) return tcs.Task;
+
+ Socket result = args.ConnectSocket;
+ args.Completed -= HandleIOCompleted;
+
+ connectArgsPool.Return(args);
+
+ return Task.FromResult(result);
+ }
+
+ public static Task DisconnectAsync(ObjectPool<SocketAsyncEventArgs> disconnectArgsPool,
+ Socket socket, CancellationToken cancellationToken = default)
+ {
+ TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
+
+ SocketAsyncEventArgs args = disconnectArgsPool.Get();
+ args.UserToken = new AsyncOperationToken(disconnectArgsPool, tcs, cancellationToken);
+
+ /*
+ // register cleanup action for when the cancellation token is thrown
+ cancellationToken.Register(() =>
+ {
+ tcs.SetCanceled();
+
+ //TODO this is probably a hideous solution. find a better one
+ args.Completed -= HandleIOCompleted;
+ args.Dispose();
+
+ SocketAsyncEventArgs newArgs = new SocketAsyncEventArgs();
+ newArgs.Completed += HandleIOCompleted;
+ disconnectAsyncEventArgsPool.Return(newArgs);
+ });
+ */
+
+ // if the disconnect operation doesn't complete synchronously, return the awaitable task
+ if (socket.DisconnectAsync(args)) return tcs.Task;
+
+ args.Completed -= HandleIOCompleted;
+
+ disconnectArgsPool.Return(args);
+
+ return Task.CompletedTask;
+ }
+
+ public static ValueTask<TransmissionResult> ReceiveFromAsync(ObjectPool<SocketAsyncEventArgs> receiveArgsPool,
+ Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, Memory<byte> inputBuffer,
+ CancellationToken cancellationToken = default)
+ {
+ TaskCompletionSource<TransmissionResult> tcs = new TaskCompletionSource<TransmissionResult>();
+
+ SocketAsyncEventArgs args = receiveArgsPool.Get();
+ args.SetBuffer(inputBuffer);
+ args.SocketFlags = socketFlags;
+ args.RemoteEndPoint = remoteEndPoint;
+ args.UserToken = new AsyncReadToken(receiveArgsPool, inputBuffer, tcs, cancellationToken);
+
+ args.Completed += HandleIOCompleted;
+
+ /*
+ // register cleanup action for when the cancellation token is thrown
+ cancellationToken.Register(() =>
+ {
+ tcs.SetCanceled();
+
+ receiveBufferPool.Return(rentedReceiveFromBuffer, true);
+
+ //TODO this is probably a hideous solution. find a better one
+ args.Completed -= HandleIOCompleted;
+ args.Dispose();
+
+ SocketAsyncEventArgs newArgs = new SocketAsyncEventArgs();
+ newArgs.Completed += HandleIOCompleted;
+ receiveAsyncEventArgsPool.Return(newArgs);
+ });
+ */
+
+ // if the receive operation doesn't complete synchronously, returns the awaitable task
+ if (socket.ReceiveFromAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task);
+
+ args.Completed -= HandleIOCompleted;
+
+ TransmissionResult result = new TransmissionResult(args);
+
+ receiveArgsPool.Return(args);
+
+ return new ValueTask<TransmissionResult>(result);
+ }
+
+ public static ValueTask<TransmissionResult> SendToAsync(ObjectPool<SocketAsyncEventArgs> sendArgsPool,
+ Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, Memory<byte> outputBuffer,
+ CancellationToken cancellationToken = default)
+ {
+ TaskCompletionSource<TransmissionResult> tcs = new TaskCompletionSource<TransmissionResult>();
+
+ SocketAsyncEventArgs args = sendArgsPool.Get();
+ args.SetBuffer(outputBuffer);
+ args.SocketFlags = socketFlags;
+ args.RemoteEndPoint = remoteEndPoint;
+ args.UserToken = new AsyncWriteToken(sendArgsPool, tcs, cancellationToken);
+
+ args.Completed += HandleIOCompleted;
+
+ /*
+ // register cleanup action for when the cancellation token is thrown
+ cancellationToken.Register(() =>
+ {
+ tcs.SetCanceled();
+
+ sendBufferPool.Return(rentedSendToBuffer, true);
+
+ //TODO this is probably a hideous solution. find a better one
+ args.Completed -= HandleIOCompleted;
+ args.Dispose();
+
+ SocketAsyncEventArgs newArgs = new SocketAsyncEventArgs();
+ newArgs.Completed += HandleIOCompleted;
+ sendAsyncEventArgsPool.Return(newArgs);
+ });
+ */
+
+ // if the send operation doesn't complete synchronously, return the awaitable task
+ if (socket.SendToAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task);
+
+ args.Completed -= HandleIOCompleted;
+
+ TransmissionResult result = new TransmissionResult(args);
+
+ sendArgsPool.Return(args);
+
+ return new ValueTask<TransmissionResult>(result);
+ }
+ }
+}
+\ No newline at end of file
diff --git a/NetSharp/NetSharp/Sockets/SocketServer.cs b/NetSharp/NetSharp/Sockets/SocketServer.cs
@@ -0,0 +1,95 @@
+using System;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.ObjectPool;
+using NetSharp.Utils;
+
+namespace NetSharp.Sockets
+{
+ public class SocketServer : IDisposable
+ {
+ private readonly ObjectPool<SocketAsyncEventArgs> transmissionArgsPool;
+
+ /// <summary>
+ /// Destroys a socket server instance.
+ /// </summary>
+ ~SocketServer()
+ {
+ Dispose(false);
+ }
+
+ /// <summary>
+ /// The socket which should be used to listen for incoming data and to send outgoing data.
+ /// </summary>
+ protected readonly Socket listenerSocket;
+
+ /// <summary>
+ /// Implementation of dispose pattern.
+ /// </summary>
+ /// <param name="disposing">
+ /// Whether this method is being called by the object finalizer, or by the <see cref="Dispose()"/> method.
+ /// </param>
+ protected virtual void Dispose(bool disposing)
+ {
+ if (disposing)
+ {
+ listenerSocket.Dispose();
+ }
+ }
+
+ public SocketServer(AddressFamily listenerAddressFamily, SocketType listenerSocketType,
+ ProtocolType listenerProtocolType)
+ {
+ listenerSocket = new Socket(listenerAddressFamily, listenerSocketType, listenerProtocolType);
+
+ transmissionArgsPool = new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>());
+ }
+
+ /// <inheritdoc />
+ public void Dispose()
+ {
+ Dispose(true);
+ GC.SuppressFinalize(this);
+ }
+
+ public ValueTask<TransmissionResult> ReceiveAsync(EndPoint remoteEndPoint, SocketFlags receiveFlags, Memory<byte> receiveBuffer,
+ CancellationToken cancellationToken = default)
+ {
+ return SocketOperations.ReceiveFromAsync(transmissionArgsPool, listenerSocket,
+ remoteEndPoint, receiveFlags, receiveBuffer, cancellationToken);
+ }
+
+ public ValueTask<TransmissionResult> SendAsync(EndPoint remoteEndPoint, SocketFlags sendFlags, Memory<byte> sendBuffer,
+ CancellationToken cancellationToken = default)
+ {
+ return SocketOperations.SendToAsync(transmissionArgsPool, listenerSocket,
+ remoteEndPoint, sendFlags, sendBuffer, cancellationToken);
+ }
+
+ public Task<bool> TryBindAsync(EndPoint localEndPoint, TimeSpan timeout)
+ {
+ using CancellationTokenSource cts = new CancellationTokenSource(timeout);
+
+ try
+ {
+ return Task.Run(() =>
+ {
+ listenerSocket.Bind(localEndPoint);
+
+ return true;
+ }, cts.Token);
+ }
+ catch (TaskCanceledException)
+ {
+ return Task.FromResult(false);
+ }
+ catch (SocketException ex)
+ {
+ Console.WriteLine($"Socket exception on binding socket to {localEndPoint}: {ex}");
+ return Task.FromResult(false);
+ }
+ }
+ }
+}
+\ No newline at end of file
diff --git a/NetSharp/NetSharpExamples/Program.cs b/NetSharp/NetSharpExamples/Program.cs
@@ -1,6 +1,8 @@
using System;
+using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
+using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
@@ -10,36 +12,40 @@ using NetSharp;
using NetSharp.Extensions;
using NetSharp.Logging;
using NetSharp.Packets;
+using NetSharp.Sockets;
using NetSharp.Utils;
namespace NetSharpExamples
{
internal class Program
{
- private static int newtorkTimeout = 1_000_000;
- private static IPAddress serverAddress;
-
- private static int serverPort;
+ private const int NetworkTimeout = 1_000_000;
+ private const int ServerPort = 12374;
+ private static readonly IPAddress ServerAddress = IPAddress.Parse("10.4.3.167"); // IPAddress.Parse("192.168.0.31");
+ private static readonly EndPoint ServerEndPoint = new IPEndPoint(ServerAddress, ServerPort);
private static async Task Main()
{
Console.WriteLine("Hello World!");
- serverAddress = IPAddress.Loopback;
- serverPort = 12374;
+ await Task.Factory.StartNew(TestSocketServer);
+ await Task.Factory.StartNew(TestSocketClient).Result;
+
+ Console.ReadLine();
- await Task.Factory.StartNew(TestServer);
- await Task.Factory.StartNew(TestClient).Result;
+ //await Task.Factory.StartNew(TestServer);
+ //await Task.Factory.StartNew(TestClient).Result;
}
+ #region Connection Tests
+
private static async Task TestClient()
{
- TimeSpan socketTimeout = TimeSpan.FromSeconds(newtorkTimeout);
+ TimeSpan socketTimeout = TimeSpan.FromSeconds(NetworkTimeout);
- const int clientCount = 1;
- const long sentPacketCount = 100;
+ const int clientCount = 10;
+ const long sentPacketCount = 10_000;
- EndPoint serverEndPoint = new IPEndPoint(serverAddress, serverPort);
ConnectionBuilder clientBuilder = new ConnectionBuilder();
Console.WriteLine($"Testing client connections...");
@@ -52,7 +58,7 @@ namespace NetSharpExamples
using Connection client = clientBuilder.Build();
await client.TryBindAsync(new IPEndPoint(IPAddress.Any, 0));
- await client.TryConnectAsync(serverEndPoint);
+ await client.TryConnectAsync(ServerEndPoint);
//client.SetLoggingStream(Console.OpenStandardOutput());
//TimeSpan timeout = TimeSpan.FromMilliseconds(100);
Stopwatch stopwatch = new Stopwatch();
@@ -74,21 +80,19 @@ namespace NetSharpExamples
{
stopwatch.Start();
//int sentBytes = await client.SendToAsync(serverEndPoint, requestPacketBuffer, SocketFlags.None);
- int sentBytes = await client.SendAsync(serverEndPoint, requestPacketBuffer, SocketFlags.None);
+ int sentBytes = await client.SendAsync(ServerEndPoint, requestPacketBuffer, SocketFlags.None);
stopwatch.Stop();
Interlocked.Increment(ref sentPackets);
- Console.WriteLine($"[Client {clientId}] Sent {sentBytes} bytes to {serverEndPoint}");
+ //Console.WriteLine($"[Client {clientId}] Sent {sentBytes} bytes to {serverEndPoint}");
stopwatch.Start();
//TransmissionResult result = await client.ReceiveFromAsync(serverEndPoint, responseBuffer, SocketFlags.None);
- TransmissionResult result = await client.ReceiveAsync(serverEndPoint, responseBuffer, SocketFlags.None);
+ TransmissionResult result = await client.ReceiveAsync(ServerEndPoint, responseBuffer, SocketFlags.None);
stopwatch.Stop();
Interlocked.Increment(ref receivedPackets);
- Console.WriteLine($"[Client {clientId}] Received {result.Count} bytes from {result.RemoteEndPoint}");
-
- //await Task.Delay(10);
+ //Console.WriteLine($"[Client {clientId}] Received {result.Count} bytes from {result.RemoteEndPoint}");
}
catch (Exception ex)
{
@@ -99,14 +103,11 @@ namespace NetSharpExamples
long millis = stopwatch.ElapsedMilliseconds;
double megabytes = sentPackets * NetworkPacket.DataSegmentSize / 1_000_000.0;
- Console.WriteLine($"[Client {clientId}] Sent {sentPacketCount} packets to {serverEndPoint} in {millis} milliseconds");
+ Console.WriteLine($"[Client {clientId}] Sent {sentPacketCount} packets to {ServerEndPoint} in {millis} milliseconds");
Console.WriteLine($"[Client {clientId}] Approximate bandwidth: {megabytes / (millis / 1000.0):F3} MBps");
- /*
- Console.WriteLine($"[Client {clientId}] Closing client...");
await client.TryDisconnectAsync();
Console.WriteLine($"[Client {clientId}] Closed client.");
- */
}, i, TaskCreationOptions.LongRunning);
}
@@ -119,11 +120,10 @@ namespace NetSharpExamples
File.Delete(serverLogFile);
await using Stream serverOutputStream = File.OpenWrite(serverLogFile);
- EndPoint serverEndPoint = new IPEndPoint(serverAddress, serverPort);
ConnectionBuilder serverBuilder = new ConnectionBuilder();
using Connection server = serverBuilder.WithLogging(Console.OpenStandardOutput(), LogLevel.Info).Build();
- await server.TryBindAsync(serverEndPoint);
+ await server.TryBindAsync(ServerEndPoint);
//server.SetLoggingStream(Console.OpenStandardOutput());
//server.ChangeLoggingStream(serverOutputStream, LogLevel.Error);
@@ -135,5 +135,160 @@ namespace NetSharpExamples
Console.ReadLine();
}
+
+ #endregion Connection Tests
+
+ #region Socket Tests
+
+ private static async Task TestSocketClient()
+ {
+ const int clientCount = 100;
+ const long packetsToSend = 100_000;
+
+ Thread[] clientThreads = new Thread[clientCount];
+ double[] clientBandwidths = new double[clientCount];
+ HashSet<int> activeThreads = new HashSet<int>(clientCount);
+
+ async void ClientTask(object id)
+ {
+ try
+ {
+ lock (typeof(Console))
+ {
+ Console.WriteLine($"[Client {id}] Starting client...");
+ }
+
+ using SocketClient client = new SocketClient(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+
+ if (!await client.TryBindAsync(new IPEndPoint(IPAddress.Any, 0), TimeSpan.FromMilliseconds(1_000)))
+ {
+ lock (typeof(Console))
+ {
+ Console.WriteLine($"[Client {id}] Could not bind client socket within timeout!");
+ return;
+ }
+ }
+
+ byte[] requestBuffer = new byte[NetworkPacket.PacketSize];
+ Memory<byte> requestBufferMemory = new Memory<byte>(requestBuffer);
+
+ byte[] responseBuffer = new byte[NetworkPacket.PacketSize];
+ Memory<byte> responseBufferMemory = new Memory<byte>(responseBuffer);
+
+ Stopwatch rttStopwatch = new Stopwatch();
+ Stopwatch bandwidthStopwatch = new Stopwatch();
+
+ for (int i = 0; i < packetsToSend; i++)
+ {
+ Encoding.UTF8.GetBytes($"Hello World! (Packet {i})").CopyTo(requestBufferMemory);
+
+ rttStopwatch.Start();
+ bandwidthStopwatch.Start();
+ TransmissionResult sendResult = await client.SendAsync(ServerEndPoint, SocketFlags.None, requestBufferMemory);
+ bandwidthStopwatch.Stop();
+ rttStopwatch.Stop();
+
+#if DEBUG
+ lock (typeof(Console))
+ {
+ //Console.WriteLine($"[Client {id}, Packet {i}] Sent {sendResult.Count} bytes to {sendResult.RemoteEndPoint}");
+ //Console.WriteLine($"[Client {id}, Packet {i}] >>>> {Encoding.UTF8.GetString(sendResult.Buffer.Span)}");
+ }
+#endif
+
+ rttStopwatch.Start();
+ bandwidthStopwatch.Start();
+ TransmissionResult receiveResult = await client.ReceiveAsync(ServerEndPoint, SocketFlags.None, responseBufferMemory);
+ bandwidthStopwatch.Stop();
+ rttStopwatch.Stop();
+
+#if DEBUG
+ lock (typeof(Console))
+ {
+ //Console.WriteLine($"[Client {id}, Packet {i}] Received {receiveResult.Count} bytes from {receiveResult.RemoteEndPoint}");
+ //Console.WriteLine($"[Client {id}, Packet {i}] <<<< {Encoding.UTF8.GetString(receiveResult.Buffer.Span)}");
+ }
+#endif
+
+ lock (typeof(Console))
+ {
+ if (!activeThreads.Contains((int)id))
+ {
+ activeThreads.Add((int)id);
+ }
+
+ //Console.WriteLine($"[Client {id}] Client Round Trip Time: {rttStopwatch.ElapsedMilliseconds} ms");
+ rttStopwatch.Reset();
+ }
+ }
+
+ long millis = bandwidthStopwatch.ElapsedMilliseconds;
+ double megabytes = packetsToSend * NetworkPacket.DataSegmentSize / 1_000_000.0;
+ double bandwidth = megabytes / (millis / 1000.0);
+
+ clientBandwidths[(int)id] = bandwidth;
+
+ lock (typeof(Console))
+ {
+ Console.WriteLine($"[Client {id}] Sent {packetsToSend} packets to {ServerEndPoint} in {millis} milliseconds");
+ Console.WriteLine($"[Client {id}] Approximate bandwidth: {bandwidth:F3} MBps");
+
+ Console.WriteLine($"[Client {id}] Stopping client...");
+ }
+ }
+ catch (Exception ex)
+ {
+ Console.WriteLine(ex);
+ throw;
+ }
+ }
+
+ for (int clientId = 0; clientId < clientCount; clientId++)
+ {
+ clientThreads[clientId] = new Thread(ClientTask) { Name = $"Client thread {clientId:D3}" };
+ clientThreads[clientId].Start(clientId);
+ }
+
+ Console.ReadLine();
+
+ int totalActiveThreads = 0;
+ for (int i = 0; i < clientCount; i++)
+ {
+ Thread clientThread = clientThreads[i];
+
+ bool threadWasActive = activeThreads.Contains(i);
+ if (threadWasActive)
+ {
+ totalActiveThreads++;
+ }
+
+ Console.WriteLine($"{clientThread.Name} bandwidth: {clientBandwidths[i]}, was active? {threadWasActive}");
+ }
+
+ Console.WriteLine($"Total server bandwidth: {clientBandwidths.Sum():F3} MBps, Total Active Threads: {totalActiveThreads}, Total Inactive Threads: {clientCount - totalActiveThreads}");
+ }
+
+ private static async Task TestSocketServer()
+ {
+ using SocketServer server =
+ new SocketServer(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+ await server.TryBindAsync(ServerEndPoint, TimeSpan.FromMilliseconds(1_000));
+
+ byte[] requestBuffer = new byte[NetworkPacket.PacketSize];
+ Memory<byte> requestBufferMemory = new Memory<byte>(requestBuffer);
+
+ while (true)
+ {
+ EndPoint remoteEndPoint = new IPEndPoint(IPAddress.Any, 0);
+
+ TransmissionResult receiveResult =
+ await server.ReceiveAsync(remoteEndPoint, SocketFlags.None, requestBufferMemory);
+
+ TransmissionResult sendResult =
+ await server.SendAsync(receiveResult.RemoteEndPoint, SocketFlags.None, requestBuffer);
+ }
+ }
+
+ #endregion Socket Tests
}
}
\ No newline at end of file
