commit 714643f70efea0a1cf36457a3ad6bbec4733cb7b parent 9ccdaa5fdbc86a7ae839905661948171a0b23eb4 Author: Mikolaj Lenczewski <33129490+EnderRifter@users.noreply.github.com> Date: Thu, 27 Feb 2020 09:05:24 +0000 Started restructuring of solution, utilising SocketAsyncEventArgs and friends! Diffstat:
33 files changed, 1694 insertions(+), 1374 deletions(-)
diff --git a/NetSharp/NetSharp/Clients/TcpClient.cs b/NetSharp/NetSharp/Clients/TcpClient.cs @@ -1,67 +0,0 @@ -using System; -using System.Net.Sockets; -using System.Threading; -using System.Threading.Tasks; -using NetSharp.Packets; -using NetSharp.Packets.Builtin; -using NetSharp.Servers; -using NetSharp.Utils.Socket_Options; - -namespace NetSharp.Clients -{ - /// <summary> - /// Provides methods for TCP communication with a connected <see cref="TcpServer"/> instance. - /// </summary> - public sealed class TcpClient : Client - { - /// <inheritdoc /> - public TcpClient() : base(SocketType.Stream, ProtocolType.Tcp, SocketOptionManager.Tcp) - { - } - - /// <inheritdoc /> - public override async Task<bool> SendBytesAsync(byte[] buffer, TimeSpan timeout) - { - SimpleDataPacket packet = new SimpleDataPacket(buffer); - return await SendSimpleAsync(packet, timeout); - } - - /// <inheritdoc /> - public override async Task<byte[]> SendBytesWithResponseAsync(byte[] buffer, TimeSpan timeout) - { - DataPacket packet = new DataPacket(buffer); - DataResponsePacket response = await SendComplexAsync<DataPacket, DataResponsePacket>(packet, timeout); - - return response.ResponseBuffer.ToArray(); - } - - /// <inheritdoc /> - public override async Task<Rep> SendComplexAsync<Req, Rep>(Req request, TimeSpan timeout) - { - uint packetTypeId = PacketRegistry.GetPacketId<Req>(); - - request.BeforeSerialisation(); - Memory<byte> serialisedRequest = request.Serialise(); - SerialisedPacket rawRequest = new SerialisedPacket(serialisedRequest, packetTypeId); - await DoSendPacketAsync(socket, rawRequest, SocketFlags.None, timeout); - - SerialisedPacket rawResponsePacket = await DoReceivePacketAsync(socket, SocketFlags.None, timeout); - Rep responsePacket = new Rep(); - responsePacket.Deserialise(rawResponsePacket.Contents); - responsePacket.AfterDeserialisation(); - - return responsePacket; - } - - /// <inheritdoc /> - public override async Task<bool> SendSimpleAsync<Req>(Req request, TimeSpan timeout) - { - uint packetTypeId = PacketRegistry.GetPacketId<Req>(); - - request.BeforeSerialisation(); - Memory<byte> serialisedRequest = request.Serialise(); - SerialisedPacket rawRequest = new SerialisedPacket(serialisedRequest, packetTypeId); - return await DoSendPacketAsync(socket, rawRequest, SocketFlags.None, timeout); - } - } -} -\ No newline at end of file diff --git a/NetSharp/NetSharp/Connection.cs b/NetSharp/NetSharp/Connection.cs @@ -5,7 +5,6 @@ using System.Net.Sockets; using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; -using NetSharp.Interfaces; using NetSharp.Logging; using NetSharp.Packets; using NetSharp.Utils; @@ -18,6 +17,11 @@ namespace NetSharp public abstract class Connection : IDisposable { /// <summary> + /// Provides a wrapper around common network operations and enables awaiting for said operations. + /// </summary> + private readonly NetworkOperationsManager networkManager; + + /// <summary> /// The logger to which the server can log messages. /// </summary> protected Logger logger; @@ -27,6 +31,8 @@ namespace NetSharp /// </summary> protected Connection() { + networkManager = new NetworkOperationsManager(); + logger = new Logger(Stream.Null); } @@ -51,7 +57,7 @@ namespace NetSharp /// The timespan within which the packet should be received. After this timespan elapses, the receive task is cancelled. /// </param> /// <param name="cancellationToken">A pre-existing cancellation token that should be observed alongside the timeout.</param> - /// <returns>The packet that was received. <see cref="NullPacket"/> if not received correctly.</returns> + /// <returns>The packet that was received.</returns> protected async Task<SerialisedPacket> DoReceivePacketAsync(Socket socket, SocketFlags socketFlags, TimeSpan timeout, CancellationToken cancellationToken = default) { @@ -59,14 +65,15 @@ namespace NetSharp using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCancellationTokenSource.Token, cancellationToken); + throw new NotImplementedException(); + try { - (SerialisedPacket packet, EndPoint endPoint) = - await NetworkOperations.ReadPacketAsync(socket, socketFlags, cts.Token); + //(SerialisedPacket packet, EndPoint endPoint) = await NetworkOperationsManager.ReadPacketAsync(socket, socketFlags, cts.Token); - OnBytesReceived(endPoint, packet.Contents.Length); + //OnBytesReceived(endPoint, packet.Contents.Length); - return packet; + // return packet; } catch (SocketException ex) { @@ -100,14 +107,15 @@ namespace NetSharp using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCancellationTokenSource.Token, cancellationToken); + throw new NotImplementedException(); + try { - (SerialisedPacket packet, EndPoint endPoint) = - await NetworkOperations.ReadPacketFromAsync(socket, remoteEndPoint, socketFlags, cts.Token); + //(SerialisedPacket packet, EndPoint endPoint) = await NetworkOperationsManager.ReadPacketFromAsync(socket, remoteEndPoint, socketFlags, cts.Token); - OnBytesReceived(endPoint, packet.Contents.Length); + //OnBytesReceived(endPoint, packet.Contents.Length); - return (packet, remoteEndPoint); + //return (packet, remoteEndPoint); } catch (SocketException ex) { @@ -139,9 +147,11 @@ namespace NetSharp using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCancellationTokenSource.Token, cancellationToken); + throw new NotImplementedException(); + try { - await NetworkOperations.WritePacketAsync(remoteSocket, packet, socketFlags, cts.Token); + //await NetworkOperationsManager.WritePacketAsync(remoteSocket, packet, socketFlags, cts.Token); OnBytesSent(remoteSocket.RemoteEndPoint, packet.Contents.Length); @@ -178,9 +188,11 @@ namespace NetSharp using CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(timeoutCancellationTokenSource.Token, cancellationToken); + throw new NotImplementedException(); + try { - await NetworkOperations.WritePacketToAsync(socket, remoteEndPoint, packet, socketFlags, cts.Token); + //await NetworkOperationsManager.WritePacketToAsync(socket, remoteEndPoint, packet, socketFlags, cts.Token); OnBytesSent(remoteEndPoint, packet.Contents.Length); diff --git a/NetSharp/NetSharp/Client.cs b/NetSharp/NetSharp/Deprecated/Client.cs diff --git a/NetSharp/NetSharp/Extensions/ClientExtensions.cs b/NetSharp/NetSharp/Deprecated/ClientExtensions.cs diff --git a/NetSharp/NetSharp/Utils/Socket Options/DefaultSocketOptions.cs b/NetSharp/NetSharp/Deprecated/DefaultSocketOptions.cs diff --git a/NetSharp/NetSharp/Interfaces/IClient.cs b/NetSharp/NetSharp/Deprecated/IClient.cs diff --git a/NetSharp/NetSharp/Interfaces/INetworkSerialisable.cs b/NetSharp/NetSharp/Deprecated/INetworkSerialisable.cs diff --git a/NetSharp/NetSharp/Interfaces/IPacket.cs b/NetSharp/NetSharp/Deprecated/IPacket.cs diff --git a/NetSharp/NetSharp/Deprecated/IPacketHandler.cs b/NetSharp/NetSharp/Deprecated/IPacketHandler.cs @@ -0,0 +1,51 @@ +namespace NetSharp.Interfaces +{ + /// <summary> + /// Describes a class capable of registering and deregistering packet handlers, and capable of + /// handling incoming packets according to the currently registered packet handlers. + /// </summary> + public interface IPacketHandler + { + /// <summary> + /// Attempts to deregister the complex packet handler delegate for all packets of the given type. If a handler + /// method doesn't exist for the given packet type, <paramref name="oldHandlerDelegate"/> will be <c>default</c>. + /// </summary> + /// <typeparam name="Req">The type of request packet for which to deregister the handler delegate.</typeparam> + /// <typeparam name="Rep">The type of response packet that is generated by the delegate method.</typeparam> + /// <param name="oldHandlerDelegate">The old handler delegate method that was previously registered.</param> + /// <returns>Whether the packet handler delegate was successfully deregistered.</returns> + public bool TryDeregisterComplexPacketHandler<Req, Rep>(out ComplexPacketHandler<Req, Rep>? oldHandlerDelegate) + where Req : class, IRequestPacket, new() where Rep : class, IResponsePacket<Req>, new(); + + /// <summary> + /// Attempts to deregister the simple packet handler delegate for all packets of the given type. If a handler + /// method doesn't exist for the given packet type, <paramref name="oldHandlerDelegate"/> will be <c>default</c>. + /// </summary> + /// <typeparam name="Req">The type of request packet for which to deregister the handler delegate.</typeparam> + /// <param name="oldHandlerDelegate">The old handler delegate method that was previously registered.</param> + /// <returns>Whether the packet handler delegate was successfully deregistered.</returns> + public bool TryDeregisterSimplePacketHandler<Req>(out SimplePacketHandler<Req>? oldHandlerDelegate) + where Req : class, IRequestPacket, new(); + + /// <summary> + /// Attempts to register a complex packet handler delegate for all packets of the given type. If a handler + /// method already exists for the given packet type, it will be updated and replaced with the given one. + /// </summary> + /// <typeparam name="Req">The type of request packet for which to register the handler delegate.</typeparam> + /// <typeparam name="Rep">The type of response packet that is generated by the delegate method.</typeparam> + /// <param name="handlerDelegate">The delegate method to register as the complex packet handler.</param> + /// <returns>Whether the packet handler delegate was successfully registered.</returns> + public bool TryRegisterComplexPacketHandler<Req, Rep>(ComplexPacketHandler<Req, Rep> handlerDelegate) + where Req : class, IRequestPacket, new() where Rep : class, IResponsePacket<Req>, new(); + + /// <summary> + /// Attempts to register a simple packet handler delegate for all packets of the given type. If a handler + /// method already exists for the given packet type, it will be updated and replaced with the given one. + /// </summary> + /// <typeparam name="Req">The type of request packet for which to register the handler delegate.</typeparam> + /// <param name="handlerDelegate">The delegate method to register as the simple packet handler.</param> + /// <returns>Whether the packet handler delegate was successfully registered.</returns> + public bool TryRegisterSimplePacketHandler<Req>(SimplePacketHandler<Req> handlerDelegate) + where Req : class, IRequestPacket, new(); + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Interfaces/IRequestPacket.cs b/NetSharp/NetSharp/Deprecated/IRequestPacket.cs diff --git a/NetSharp/NetSharp/Interfaces/IResponsePacket.cs b/NetSharp/NetSharp/Deprecated/IResponsePacket.cs diff --git a/NetSharp/NetSharp/Deprecated/IServer.cs b/NetSharp/NetSharp/Deprecated/IServer.cs @@ -0,0 +1,44 @@ +using System; +using System.Net; +using System.Threading.Tasks; + +namespace NetSharp.Interfaces +{ + /// <summary> + /// Describes a server capable of asynchronously handling multiple <see cref="IClient"/> connections at once. + /// </summary> + public interface IServer + { + /// <summary> + /// Signifies that a connection with a remote endpoint has been made. + /// </summary> + public event Action<EndPoint>? ClientConnected; + + //protected IResponsePacket<IRequestPacket> DeserialiseResponsePacket(in Packet) + /// <summary> + /// Signifies that a connection with a remote endpoint has been lost. + /// </summary> + public event Action<EndPoint>? ClientDisconnected; + + /// <summary> + /// Signifies that the server was started and clients will start being accepted. + /// </summary> + public event Action? ServerStarted; + + /// <summary> + /// Signifies that the server was stopped and clients will stop being accepted. + /// </summary> + public event Action? ServerStopped; + + /// <summary> + /// Starts the server asynchronously and starts accepting client connections. Does not block. + /// </summary> + /// <param name="localEndPoint">The local endpoint to bind to.</param> + public Task RunAsync(EndPoint localEndPoint); + + /// <summary> + /// Shuts down the server. + /// </summary> + public void Shutdown(); + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Packets/SerialisedPacket.cs b/NetSharp/NetSharp/Deprecated/SerialisedPacket.cs diff --git a/NetSharp/NetSharp/Deprecated/Server.cs b/NetSharp/NetSharp/Deprecated/Server.cs @@ -0,0 +1,596 @@ +using System; +using System.Collections.Concurrent; +using System.Net; +using System.Net.Sockets; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using NetSharp.Interfaces; +using NetSharp.Packets; +using NetSharp.Packets.Builtin; +using NetSharp.Utils.Socket_Options; + +namespace NetSharp +{ + /// <summary> + /// Represents a method that receives a request packet of the given type (<typeparamref name="TReq"/>) and + /// handles the request, returning a response packet of the given type (<typeparamref name="TRep"/>). + /// </summary> + /// <typeparam name="TReq">The type of request packet handled by this delegate method.</typeparam> + /// <typeparam name="TRep">The type of response packet returned by this delegate method.</typeparam> + /// <param name="requestPacket">The request packet that should be handled by this delegate method.</param> + /// <param name="remoteEndPoint">The remote endpoint from which the request originated.</param> + /// <returns>The response packet to send back to the remote endpoint from which the request originated.</returns> + public delegate TRep ComplexPacketHandler<in TReq, out TRep>(TReq requestPacket, EndPoint remoteEndPoint) + where TReq : class, IRequestPacket, new() where TRep : class, IResponsePacket<TReq>, new(); + + /// <summary> + /// Represents a method that receives a simple request packet of the given type (<typeparamref name="TReq"/>) and + /// handles the request, not returning any response packets. + /// </summary> + /// <typeparam name="TReq">The type of request packet handled by this delegate method.</typeparam> + /// <param name="requestPacket">The request packet that should be handled by this delegate method.</param> + /// <param name="remoteEndPoint">The remote endpoint from which the request originated.</param> + public delegate void SimplePacketHandler<in TReq>(TReq requestPacket, EndPoint remoteEndPoint) + where TReq : class, IRequestPacket, new(); + + /// <summary> + /// Provides methods for handling connected <see cref="IClient"/> instances. + /// </summary> + public abstract class Server : Connection, IServer, IPacketHandler, IDisposable + { + /// <summary> + /// Maps a packet type id to the complex packet handler for that packet type. + /// </summary> + private readonly ConcurrentDictionary<uint, Func<IRequestPacket, EndPoint, IResponsePacket<IRequestPacket>>> + complexPacketHandlers; + + /// <summary> + /// Maps a packet type id to the raw packet deserialiser that deserialises raw packets to + /// <see cref="IRequestPacket"/> implementors. + /// </summary> + private readonly ConcurrentDictionary<uint, RawRequestPacketDeserialiser> requestPacketDeserialisers; + + /// <summary> + /// Cancellation token source to stop handling client sockets when the server should be shut down. + /// </summary> + private readonly CancellationTokenSource serverShutdownCancellationTokenSource; + + /// <summary> + /// Maps a packet type id to the simple packet handler for that packet type. + /// </summary> + private readonly ConcurrentDictionary<uint, Action<IRequestPacket, EndPoint>> simplePacketHandlers; + + /// <summary> + /// Initialises a new instance of the <see cref="Server"/> class. + /// </summary> + private Server() + { + serverShutdownCancellationTokenSource = new CancellationTokenSource(); + + serverShutdownCancellationToken = serverShutdownCancellationTokenSource.Token; + + requestPacketDeserialisers = new ConcurrentDictionary<uint, RawRequestPacketDeserialiser>(); + + simplePacketHandlers = new ConcurrentDictionary<uint, Action<IRequestPacket, EndPoint>>(); + complexPacketHandlers = + new ConcurrentDictionary<uint, Func<IRequestPacket, EndPoint, IResponsePacket<IRequestPacket>>>(); + + RegisterInternalPacketHandlers(); + + socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); + socketOptions = new DefaultSocketOptions(ref socket); + } + + /// <summary> + /// Destroys an instance of the <see cref="Server"/> class. + /// </summary> + ~Server() + { + Dispose(false); + } + + /// <summary> + /// Represents a method that receives a raw packet, and deserialises it into an <see cref="IRequestPacket"/> implementor. + /// </summary> + /// <param name="rawPacket">The raw packet that was received from the network.</param> + /// <returns>The deserialised instance of the packet.</returns> + private delegate IRequestPacket RawRequestPacketDeserialiser(in SerialisedPacket rawPacket); + + /// <summary> + /// Registers packet handlers for every internal library packet. + /// </summary> + private void RegisterInternalPacketHandlers() + { + TryRegisterSimplePacketHandler((DisconnectPacket packet, EndPoint remoteEndPoint) => + { +#if DEBUG + logger.LogMessage($"Received disconnect packet from {remoteEndPoint}"); +#endif + OnClientDisconnected(remoteEndPoint); + }); + + TryRegisterSimplePacketHandler((SimpleDataPacket packet, EndPoint remoteEndPoint) => + { +#if DEBUG + logger.LogMessage($"Received {packet.RequestBuffer.Length} bytes from {remoteEndPoint}"); +#endif + }); + + TryRegisterComplexPacketHandler((ConnectPacket packet, EndPoint remoteEndPoint) => + { + OnClientConnected(remoteEndPoint); +#if DEBUG + logger.LogMessage($"Received connection request from {remoteEndPoint}"); +#endif + return new ConnectResponsePacket { RequestPacket = packet }; + }); + + TryRegisterComplexPacketHandler((PingPacket packet, EndPoint remoteEndPoint) => + new PingResponsePacket { RequestPacket = packet }); + + TryRegisterComplexPacketHandler((DataPacket packet, EndPoint remoteEndPoint) => + { +#if DEBUG + logger.LogMessage($"Received {packet.RequestBuffer.Length} bytes from {remoteEndPoint}"); + logger.LogMessage($"Sending {packet.RequestBuffer.Length} bytes to {remoteEndPoint}"); +#endif + return new DataResponsePacket { RequestPacket = packet, ResponseBuffer = packet.RequestBuffer }; + }); + } + + /// <summary> + /// The maximum number of connections that are allowed in the connection backlog. + /// </summary> + protected const int PendingConnectionBacklog = 100; + + /// <summary> + /// The default timeout value for all network operations. + /// </summary> + protected static readonly TimeSpan DefaultNetworkOperationTimeout = TimeSpan.FromMilliseconds(10_000); + + /// <summary> + /// The cancellation token that will be set when the server must be shut down. + /// </summary> + protected readonly CancellationToken serverShutdownCancellationToken; + + /// <summary> + /// The <see cref="Socket"/> underlying the connection. + /// </summary> + protected readonly Socket socket; + + /// <summary> + /// Backing field for the <see cref="SocketOptions"/> property. + /// </summary> + protected readonly SocketOptions socketOptions; + + /// <summary> + /// Whether the server should be ran. + /// </summary> + protected volatile bool runServer; + + /// <summary> + /// Initialises a new instance of the <see cref="Server"/> class. + /// </summary> + /// <param name="socketType">The socket type for the underlying socket.</param> + /// <param name="protocolType">The protocol type for the underlying socket.</param> + /// <param name="socketManager">The <see cref="Utils.Socket_Options.SocketOptions"/> implementation to use.</param> + protected Server(SocketType socketType, ProtocolType protocolType, SocketOptions socketManager) + : this(socketType, protocolType, socketManager, DefaultNetworkOperationTimeout) + { + } + + /// <summary> + /// Initialises a new instance of the <see cref="Server"/> class. + /// </summary> + /// <param name="socketType">The socket type for the underlying socket.</param> + /// <param name="protocolType">The protocol type for the underlying socket.</param> + /// <param name="socketManager">The <see cref="Utils.Socket_Options.SocketOptions"/> manager to use.</param> + /// <param name="networkOperationTimeout">The timeout value for send and receive operations over the network.</param> + protected Server(SocketType socketType, ProtocolType protocolType, SocketOptions socketManager, + TimeSpan networkOperationTimeout) : this() + { + socket = new Socket(AddressFamily.InterNetwork, socketType, protocolType); + + socketOptions = socketManager; + + NetworkOperationTimeout = networkOperationTimeout; + } + + /// <summary> + /// Deserialises the given <see cref="NetworkPacket"/> struct into an <see cref="IRequestPacket"/> implementor. + /// </summary> + /// <param name="packetType">The type id of packet that we should deserialise to.</param> + /// <param name="rawRequestPacket">The packet that should be deserialised.</param> + /// <returns>The deserialised packet instance, cast to the <see cref="IRequestPacket"/> interface.</returns> + protected IRequestPacket? DeserialiseRequestPacket(uint packetType, in SerialisedPacket rawRequestPacket) + { + if (requestPacketDeserialisers.TryGetValue(packetType, out RawRequestPacketDeserialiser deserialiser)) + { + return deserialiser.Invoke(rawRequestPacket); + } +#if DEBUG + logger.LogWarning($"No packet deserialiser was registered for packet of type {packetType}"); +#endif + return default; + } + + /// <summary> + /// Disposes of this <see cref="Server"/> instance. + /// </summary> + /// <param name="disposing">Whether this instance is being disposed.</param> + protected override void Dispose(bool disposing) + { + if (disposing) + { + serverShutdownCancellationTokenSource?.Cancel(); + serverShutdownCancellationTokenSource?.Dispose(); + + socket.Dispose(); + } + + base.Dispose(disposing); + } + + /// <summary> + /// Provides a task that represents the handling of a client. Calls the abstract <see cref="HandleClientAsync"/> method. + /// </summary> + /// <param name="clientHandlerArgsObj">The object representing the passed <see cref="ClientHandlerArgs"/> instance.</param> + protected async Task DoHandleClientAsync(object clientHandlerArgsObj) + { + ClientHandlerArgs clientHandlerArgs = (ClientHandlerArgs)clientHandlerArgsObj; + + try + { + await HandleClientAsync(clientHandlerArgs, serverShutdownCancellationTokenSource.Token); + } + catch (TaskCanceledException) + { + logger.LogWarning("Client handling was cancelled via a task cancellation."); + } + catch (OperationCanceledException) + { + logger.LogWarning("Client handling was cancelled via an operation cancellation."); + } + catch (Exception ex) + { + logger.LogException("Exception during client handling", ex); + } + finally + { + if (clientHandlerArgs.ClientSocket != null) + { + logger.LogMessage("Closing and releasing all resources associated with client handler socket"); + + clientHandlerArgs.ClientSocket.Shutdown(SocketShutdown.Both); + clientHandlerArgs.ClientSocket.Disconnect(true); + clientHandlerArgs.ClientSocket.Close(1); + clientHandlerArgs.ClientSocket.Dispose(); + } + } + } + + /// <summary> + /// Handles a client asynchronously. + /// </summary> + /// <param name="args">The client handler arguments that should be passed to the client handler.</param> + /// <param name="cancellationToken">Cancellation token set when the server is shutting down.</param> + protected abstract Task HandleClientAsync(ClientHandlerArgs args, CancellationToken cancellationToken); + + /// <summary> + /// Handles the given request packet with a registered packet handler. In this case, a complex packet handler + /// will override any registered simple packet handlers. + /// </summary> + /// <param name="packetType">The type id of the packet that we should handle.</param> + /// <param name="requestPacket">The packet instance that should be handled.</param> + /// <param name="remoteEndPoint">The remote endpoint from which the request packet originated.</param> + /// <returns>The response packet that should be sent back to the remote endpoint.</returns> + protected IResponsePacket<IRequestPacket>? HandleRequestPacket(uint packetType, in IRequestPacket requestPacket, + in EndPoint remoteEndPoint) + { + try + { + if (complexPacketHandlers.ContainsKey(packetType)) + { + IResponsePacket<IRequestPacket> response = + complexPacketHandlers[packetType].Invoke(requestPacket, remoteEndPoint); + + return response; + } + + if (simplePacketHandlers.ContainsKey(packetType)) + { + simplePacketHandlers[packetType].Invoke(requestPacket, remoteEndPoint); + return null; + } + +#if DEBUG + logger.LogWarning($"No packet handler was registered for packet of type {packetType}"); +#endif + } + catch (Exception ex) + { + logger.LogException( + $"Exception when invoking packet handler for packet (type: {packetType}) received from {remoteEndPoint}", + ex); + } + + return default; + } + + /// <summary> + /// Invokes the <see cref="ClientConnected"/> event. + /// </summary> + /// <param name="remoteEndPoint">The remote endpoint with which a connection was made.</param> + [MethodImpl(MethodImplOptions.AggressiveInlining)] + protected void OnClientConnected(EndPoint remoteEndPoint) => ClientConnected?.Invoke(remoteEndPoint); + + /// <summary> + /// Invokes the <see cref="ClientDisconnected"/> event. + /// </summary> + /// <param name="remoteEndPoint">The remote endpoint with which a connection was lost.</param> + [MethodImpl(MethodImplOptions.AggressiveInlining)] + protected void OnClientDisconnected(EndPoint remoteEndPoint) => ClientDisconnected?.Invoke(remoteEndPoint); + + /// <summary> + /// Invokes the <see cref="ServerStarted"/> event. + /// </summary> + [MethodImpl(MethodImplOptions.AggressiveInlining)] + protected void OnServerStarted() => ServerStarted?.Invoke(); + + /// <summary> + /// Invokes the <see cref="ServerStopped"/> event. + /// </summary> + [MethodImpl(MethodImplOptions.AggressiveInlining)] + protected void OnServerStopped() => ServerStopped?.Invoke(); + + /// <summary> + /// Attempts to synchronously bind the underlying socket to the given local endpoint. Blocks. + /// If the timeout is exceeded the binding attempt is aborted and the method returns false. + /// </summary> + /// <param name="localEndPoint">The local endpoint to bind to.</param> + /// <param name="timeout">The timeout within which to attempt the binding.</param> + /// <returns>Whether the binding was successful or not.</returns> + protected bool TryBind(EndPoint localEndPoint, TimeSpan timeout) => + TryBindAsync(localEndPoint, timeout).Result; + + /// <summary> + /// Attempts to asynchronously bind the underlying socket to the given local endpoint. Does not block. + /// If the timeout is exceeded the binding attempt is aborted and the method returns false. + /// </summary> + /// <param name="localEndPoint">The local endpoint to bind to.</param> + /// <param name="timeout">The timeout within which to attempt the binding.</param> + /// <returns>Whether the binding was successful or not.</returns> + protected async Task<bool> TryBindAsync(EndPoint localEndPoint, TimeSpan timeout) + { + using CancellationTokenSource timeoutCancellationTokenSource = new CancellationTokenSource(timeout); + using CancellationTokenSource cts = + CancellationTokenSource.CreateLinkedTokenSource(timeoutCancellationTokenSource.Token, serverShutdownCancellationToken); + + try + { + return await Task.Run(() => + { + socket.Bind(localEndPoint); + + return true; + }, cts.Token); + } + catch (TaskCanceledException) + { + return false; + } + catch (SocketException ex) + { + logger.LogException($"Socket exception on binding socket to {localEndPoint}:", ex); + return false; + } + } + + /// <summary> + /// Holds information about the arguments passed to every client handler task. + /// </summary> + protected readonly struct ClientHandlerArgs + { + /// <summary> + /// Initialises a new instance of the <see cref="ClientHandlerArgs"/> struct. + /// </summary> + /// <param name="remoteEndPoint">The remote endpoint of the client that should be handled.</param> + /// <param name="handlerSocket">The handler socket of the client that should be handled.</param> + private ClientHandlerArgs(EndPoint remoteEndPoint, Socket? handlerSocket) + { + ClientEndPoint = remoteEndPoint; + + ClientSocket = handlerSocket; + } + + /// <summary> + /// The remote endpoint for the client being handled. + /// </summary> + public readonly EndPoint ClientEndPoint; + + /// <summary> + /// The client handler socket for the client being handled. Is only set if using TCP. + /// </summary> + public readonly Socket? ClientSocket; + + /// <summary> + /// Constructs a new instance of the <see cref="ClientHandlerArgs"/> for a TCP client. + /// </summary> + /// <returns>A new instance of the <see cref="ClientHandlerArgs"/>, setup for a TCP client.</returns> + public static ClientHandlerArgs ForTcpClientHandler(in Socket clientHandlerSocket) + { + return new ClientHandlerArgs(clientHandlerSocket.RemoteEndPoint, clientHandlerSocket); + } + + /// <summary> + /// Constructs a new instance of the <see cref="ClientHandlerArgs"/> for a UDP client. + /// </summary> + /// <returns>A new instance of the <see cref="ClientHandlerArgs"/>, setup for a UDP client.</returns> + public static ClientHandlerArgs ForUdpClientHandler(in EndPoint clientEndPoint) + { + return new ClientHandlerArgs(clientEndPoint, null); + } + } + + /// <summary> + /// Signifies that a connection with a remote endpoint has been made. + /// </summary> + public event Action<EndPoint>? ClientConnected; + + //protected IResponsePacket<IRequestPacket> DeserialiseResponsePacket(in Packet) + /// <summary> + /// Signifies that a connection with a remote endpoint has been lost. + /// </summary> + public event Action<EndPoint>? ClientDisconnected; + + /// <summary> + /// Signifies that the server was started and clients will start being accepted. + /// </summary> + public event Action? ServerStarted; + + /// <summary> + /// Signifies that the server was stopped and clients will stop being accepted. + /// </summary> + public event Action? ServerStopped; + + /// <summary> + /// The timeout value for network operations such as sending bytes or receiving bytes over the network. + /// </summary> + public TimeSpan NetworkOperationTimeout { get; protected set; } + + /// <summary> + /// The configured socket options for the underlying connection. + /// </summary> + public SocketOptions SocketOptions + { + get { return socketOptions; } + } + + /// <inheritdoc /> + public abstract Task RunAsync(EndPoint localEndPoint); + + /// <inheritdoc /> + public void Shutdown() + { + runServer = false; + logger.LogMessage("Signalling server shutdown to all client handlers..."); + serverShutdownCancellationTokenSource.Cancel(); + } + + /// <inheritdoc /> + public bool TryDeregisterComplexPacketHandler<Req, Rep>(out ComplexPacketHandler<Req, Rep>? oldHandlerDelegate) + where Req : class, IRequestPacket, new() where Rep : class, IResponsePacket<Req>, new() + { + uint packetTypeId = PacketRegistry.GetPacketId<Req>(); + oldHandlerDelegate = default; + + try + { + requestPacketDeserialisers.TryRemove(packetTypeId, out _); + + if (!complexPacketHandlers.TryGetValue(packetTypeId, + out Func<IRequestPacket, EndPoint, IResponsePacket<IRequestPacket>> oldDelegate)) + return false; + + oldHandlerDelegate = (p, ep) => (Rep)oldDelegate(p, ep); + return true; + } + catch (Exception ex) + { + logger.LogException("Exception when deregistering complex packet handler", ex); + } + + return false; + } + + /// <inheritdoc /> + public bool TryDeregisterSimplePacketHandler<Req>(out SimplePacketHandler<Req>? oldHandlerDelegate) + where Req : class, IRequestPacket, new() + { + uint packetTypeId = PacketRegistry.GetPacketId<Req>(); + oldHandlerDelegate = default; + + try + { + requestPacketDeserialisers.TryRemove(packetTypeId, out _); + + if (!simplePacketHandlers.TryGetValue(packetTypeId, out Action<IRequestPacket, EndPoint> oldDelegate)) + return false; + + oldHandlerDelegate = (p, ep) => oldDelegate(p, ep); + return true; + } + catch (Exception ex) + { + logger.LogException("Exception when deregistering simple packet handler", ex); + } + + return false; + } + + /// <inheritdoc /> + public bool TryRegisterComplexPacketHandler<Req, Rep>(ComplexPacketHandler<Req, Rep> handlerDelegate) + where Req : class, IRequestPacket, new() where Rep : class, IResponsePacket<Req>, new() + { + uint packetTypeId = PacketRegistry.GetPacketId<Req>(); + + static IRequestPacket PacketDeserialiser(in SerialisedPacket packet) => SerialisedPacket.To<Req>(packet); + + IResponsePacket<IRequestPacket> MappedHandlerDelegate(IRequestPacket p, EndPoint ep) + { + Rep responsePacket = handlerDelegate((Req)p, ep); + return responsePacket; + } + + try + { + requestPacketDeserialisers.AddOrUpdate(packetTypeId, + key => PacketDeserialiser, + (key, oldDeserialiser) => PacketDeserialiser); + + complexPacketHandlers.AddOrUpdate(packetTypeId, + key => MappedHandlerDelegate, + (key, oldDelegate) => MappedHandlerDelegate); + + return true; + } + catch (Exception ex) + { + logger.LogException("Exception when registering complex packet handler", ex); + } + + return false; + } + + /// <inheritdoc /> + public bool TryRegisterSimplePacketHandler<Req>(SimplePacketHandler<Req> handlerDelegate) + where Req : class, IRequestPacket, new() + { + uint packetTypeId = PacketRegistry.GetPacketId<Req>(); + + static IRequestPacket PacketDeserialiser(in SerialisedPacket packet) => SerialisedPacket.To<Req>(packet); + + void MappedHandlerDelegate(IRequestPacket p, EndPoint ep) => handlerDelegate((Req)p, ep); + + try + { + requestPacketDeserialisers.AddOrUpdate(packetTypeId, + key => PacketDeserialiser, + (key, oldDeserialiser) => PacketDeserialiser); + + simplePacketHandlers.AddOrUpdate(packetTypeId, + key => MappedHandlerDelegate, + (key, oldDelegate) => MappedHandlerDelegate); + + return true; + } + catch (Exception ex) + { + logger.LogException("Exception when registering simple packet handler", ex); + } + + return false; + } + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Extensions/ServerExtensions.cs b/NetSharp/NetSharp/Deprecated/ServerExtensions.cs diff --git a/NetSharp/NetSharp/Utils/Socket Options/SocketOptions.cs b/NetSharp/NetSharp/Deprecated/SocketOptions.cs diff --git a/NetSharp/NetSharp/Deprecated/TcpClient.cs b/NetSharp/NetSharp/Deprecated/TcpClient.cs @@ -0,0 +1,67 @@ +using System; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using NetSharp.Packets; +using NetSharp.Packets.Builtin; +using NetSharp.Servers; +using NetSharp.Utils.Socket_Options; + +namespace NetSharp.Clients +{ + /// <summary> + /// Provides methods for TCP communication with a connected <see cref="TcpServer"/> instance. + /// </summary> + public sealed class TcpClient : Client + { + /// <inheritdoc /> + public TcpClient() : base(SocketType.Stream, ProtocolType.Tcp, new TcpSocketOptions(ref socket)) + { + } + + /// <inheritdoc /> + public override async Task<bool> SendBytesAsync(byte[] buffer, TimeSpan timeout) + { + SimpleDataPacket packet = new SimpleDataPacket(buffer); + return await SendSimpleAsync(packet, timeout); + } + + /// <inheritdoc /> + public override async Task<byte[]> SendBytesWithResponseAsync(byte[] buffer, TimeSpan timeout) + { + DataPacket packet = new DataPacket(buffer); + DataResponsePacket response = await SendComplexAsync<DataPacket, DataResponsePacket>(packet, timeout); + + return response.ResponseBuffer.ToArray(); + } + + /// <inheritdoc /> + public override async Task<Rep> SendComplexAsync<Req, Rep>(Req request, TimeSpan timeout) + { + uint packetTypeId = PacketRegistry.GetPacketId<Req>(); + + request.BeforeSerialisation(); + Memory<byte> serialisedRequest = request.Serialise(); + SerialisedPacket rawRequest = new SerialisedPacket(serialisedRequest, packetTypeId); + await DoSendPacketAsync(socket, rawRequest, SocketFlags.None, timeout); + + SerialisedPacket rawResponsePacket = await DoReceivePacketAsync(socket, SocketFlags.None, timeout); + Rep responsePacket = new Rep(); + responsePacket.Deserialise(rawResponsePacket.Contents); + responsePacket.AfterDeserialisation(); + + return responsePacket; + } + + /// <inheritdoc /> + public override async Task<bool> SendSimpleAsync<Req>(Req request, TimeSpan timeout) + { + uint packetTypeId = PacketRegistry.GetPacketId<Req>(); + + request.BeforeSerialisation(); + Memory<byte> serialisedRequest = request.Serialise(); + SerialisedPacket rawRequest = new SerialisedPacket(serialisedRequest, packetTypeId); + return await DoSendPacketAsync(socket, rawRequest, SocketFlags.None, timeout); + } + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Servers/TcpServer.cs b/NetSharp/NetSharp/Deprecated/TcpServer.cs diff --git a/NetSharp/NetSharp/Utils/Socket Options/TcpSocketOptions.cs b/NetSharp/NetSharp/Deprecated/TcpSocketOptions.cs diff --git a/NetSharp/NetSharp/Clients/UdpClient.cs b/NetSharp/NetSharp/Deprecated/UdpClient.cs diff --git a/NetSharp/NetSharp/Servers/UdpServer.cs b/NetSharp/NetSharp/Deprecated/UdpServer.cs diff --git a/NetSharp/NetSharp/Utils/Socket Options/UdpSocketOptions.cs b/NetSharp/NetSharp/Deprecated/UdpSocketOptions.cs diff --git a/NetSharp/NetSharp/Extensions/SocketExtensions.cs b/NetSharp/NetSharp/Extensions/SocketExtensions.cs @@ -1,170 +0,0 @@ -using System; -using System.Net.Sockets; -using System.Runtime.CompilerServices; -using System.Threading; -using System.Threading.Tasks; - -namespace NetSharp.Extensions -{ - /// <summary> - /// Provides additional methods and functionality to the <see cref="Socket"/> class. - /// </summary> - public static class SocketExtensions - { - /// <inheritdoc cref="Socket.ReceiveAsync"/> - public static SocketTask ReceiveAsync(this Socket instance, SocketTask awaitableTask) - { - awaitableTask.Reset(); - if (!instance.ReceiveAsync(awaitableTask.eventArgs)) - { - awaitableTask.wasCompleted = true; - } - - return awaitableTask; - } - - /// <inheritdoc cref="Socket.ReceiveFromAsync"/> - public static SocketTask ReceiveFromAsync(this Socket instance, SocketTask awaitableTask) - { - awaitableTask.Reset(); - if (!instance.ReceiveFromAsync(awaitableTask.eventArgs)) - { - awaitableTask.wasCompleted = true; - } - - return awaitableTask; - } - - /// <inheritdoc cref="Socket.ReceiveMessageFromAsync"/> - public static SocketTask ReceiveMessageFromAsync(this Socket instance, SocketTask awaitableTask) - { - awaitableTask.Reset(); - if (!instance.ReceiveMessageFromAsync(awaitableTask.eventArgs)) - { - awaitableTask.wasCompleted = true; - } - - return awaitableTask; - } - - /// <inheritdoc cref="Socket.SendAsync"/> - public static SocketTask SendAsync(this Socket instance, SocketTask awaitableTask) - { - awaitableTask.Reset(); - if (!instance.SendAsync(awaitableTask.eventArgs)) - { - awaitableTask.wasCompleted = true; - } - - return awaitableTask; - } - - /// <inheritdoc cref="Socket.SendToAsync"/> - public static SocketTask SendToAsync(this Socket instance, SocketTask awaitableTask) - { - awaitableTask.Reset(); - if (!instance.SendToAsync(awaitableTask.eventArgs)) - { - awaitableTask.wasCompleted = true; - } - - return awaitableTask; - } - } - - /// <summary> - /// Custom awaitable to ease the use of sockets with the TAP pattern. - /// Credit goes to https://devblogs.microsoft.com/pfxteam/awaiting-socket-operations/. - /// </summary> - public sealed class SocketTask : INotifyCompletion - { - /// <summary> - /// Representing a null action. - /// </summary> - private static readonly Action SentinelAction = () => { }; - - /// <summary> - /// The action that should be invoked upon the completion of the socket task. - /// </summary> - internal Action? continuationAction; - - /// <summary> - /// The underlying socket event args for this socket task. - /// </summary> - internal SocketAsyncEventArgs eventArgs; - - /// <summary> - /// Whether this socket task was completed. - /// </summary> - internal bool wasCompleted; - - /// <summary> - /// Resets this socket task to its default state, and sets <see cref="continuationAction"/> to <c>default</c>. - /// </summary> - internal void Reset() - { - wasCompleted = false; - continuationAction = default; - } - - /// <summary> - /// Initialises a new instance of the <see cref="SocketTask"/> class. - /// </summary> - /// <param name="asyncEventArgs">The socket event args that this socket task should wrap. Must not be <c>null</c>.</param> - /// <exception cref="ArgumentNullException">Thrown if the given <paramref name="asyncEventArgs"/> were <c>null</c>.</exception> - public SocketTask(SocketAsyncEventArgs? asyncEventArgs) - { - eventArgs = asyncEventArgs ?? - throw new ArgumentNullException(nameof(asyncEventArgs), "The given asynchronous socket event args were null."); - - eventArgs.Completed += delegate - { - Action? previousAction = continuationAction ?? - Interlocked.CompareExchange(ref continuationAction, SentinelAction, default); - - previousAction?.Invoke(); - }; - } - - /// <summary> - /// Whether this socket task has been completed. - /// </summary> - public bool IsCompleted - { - get { return wasCompleted; } - } - - /// <summary> - /// Returns this socket task instance. - /// </summary> - public SocketTask GetAwaiter() - { - return this; - } - - /// <summary> - /// Throws a <see cref="SocketException"/> if the wrapped <see cref="SocketAsyncEventArgs.SocketError"/> - /// is not equal to <see cref="SocketError.Success"/>. - /// </summary> - /// <exception cref="SocketException"> - /// Thrown if the wrapped <see cref="SocketAsyncEventArgs"/> did not complete successfully. - /// </exception> - public void GetResult() - { - if (eventArgs.SocketError != SocketError.Success) - { - throw new SocketException((int)eventArgs.SocketError); - } - } - - /// <inheritdoc /> - public void OnCompleted(Action? continuation) - { - if (continuationAction == SentinelAction || - Interlocked.CompareExchange(ref continuationAction, continuation, default) == SentinelAction) - { - Task.Run(continuation); - } - } - } -} -\ No newline at end of file diff --git a/NetSharp/NetSharp/Interfaces/IServer.cs b/NetSharp/NetSharp/Interfaces/IServer.cs @@ -1,86 +0,0 @@ -using System; -using System.Net; -using System.Threading.Tasks; - -namespace NetSharp.Interfaces -{ - /// <summary> - /// Describes a server capable of asynchronously handling multiple <see cref="IClient"/> connections at once. - /// </summary> - public interface IServer - { - /// <summary> - /// Signifies that a connection with a remote endpoint has been made. - /// </summary> - public event Action<EndPoint>? ClientConnected; - - //protected IResponsePacket<IRequestPacket> DeserialiseResponsePacket(in Packet) - /// <summary> - /// Signifies that a connection with a remote endpoint has been lost. - /// </summary> - public event Action<EndPoint>? ClientDisconnected; - - /// <summary> - /// Signifies that the server was started and clients will start being accepted. - /// </summary> - public event Action? ServerStarted; - - /// <summary> - /// Signifies that the server was stopped and clients will stop being accepted. - /// </summary> - public event Action? ServerStopped; - - /// <summary> - /// Starts the server asynchronously and starts accepting client connections. Does not block. - /// </summary> - /// <param name="localEndPoint">The local endpoint to bind to.</param> - public Task RunAsync(EndPoint localEndPoint); - - /// <summary> - /// Shuts down the server. - /// </summary> - public void Shutdown(); - - /// <summary> - /// Attempts to deregister the complex packet handler delegate for all packets of the given type. If a handler - /// method doesn't exist for the given packet type, <paramref name="oldHandlerDelegate"/> will be <c>default</c>. - /// </summary> - /// <typeparam name="Req">The type of request packet for which to deregister the handler delegate.</typeparam> - /// <typeparam name="Rep">The type of response packet that is generated by the delegate method.</typeparam> - /// <param name="oldHandlerDelegate">The old handler delegate method that was previously registered.</param> - /// <returns>Whether the packet handler delegate was successfully deregistered.</returns> - public bool TryDeregisterComplexPacketHandler<Req, Rep>(out ComplexPacketHandler<Req, Rep>? oldHandlerDelegate) - where Req : class, IRequestPacket, new() where Rep : class, IResponsePacket<Req>, new(); - - /// <summary> - /// Attempts to deregister the simple packet handler delegate for all packets of the given type. If a handler - /// method doesn't exist for the given packet type, <paramref name="oldHandlerDelegate"/> will be <c>default</c>. - /// </summary> - /// <typeparam name="Req">The type of request packet for which to deregister the handler delegate.</typeparam> - /// <param name="oldHandlerDelegate">The old handler delegate method that was previously registered.</param> - /// <returns>Whether the packet handler delegate was successfully deregistered.</returns> - public bool TryDeregisterSimplePacketHandler<Req>(out SimplePacketHandler<Req>? oldHandlerDelegate) - where Req : class, IRequestPacket, new(); - - /// <summary> - /// Attempts to register a complex packet handler delegate for all packets of the given type. If a handler - /// method already exists for the given packet type, it will be updated and replaced with the given one. - /// </summary> - /// <typeparam name="Req">The type of request packet for which to register the handler delegate.</typeparam> - /// <typeparam name="Rep">The type of response packet that is generated by the delegate method.</typeparam> - /// <param name="handlerDelegate">The delegate method to register as the complex packet handler.</param> - /// <returns>Whether the packet handler delegate was successfully registered.</returns> - public bool TryRegisterComplexPacketHandler<Req, Rep>(ComplexPacketHandler<Req, Rep> handlerDelegate) - where Req : class, IRequestPacket, new() where Rep : class, IResponsePacket<Req>, new(); - - /// <summary> - /// Attempts to register a simple packet handler delegate for all packets of the given type. If a handler - /// method already exists for the given packet type, it will be updated and replaced with the given one. - /// </summary> - /// <typeparam name="Req">The type of request packet for which to register the handler delegate.</typeparam> - /// <param name="handlerDelegate">The delegate method to register as the simple packet handler.</param> - /// <returns>Whether the packet handler delegate was successfully registered.</returns> - public bool TryRegisterSimplePacketHandler<Req>(SimplePacketHandler<Req> handlerDelegate) - where Req : class, IRequestPacket, new(); - } -} -\ No newline at end of file diff --git a/NetSharp/NetSharp/NetSharp.csproj b/NetSharp/NetSharp/NetSharp.csproj @@ -21,4 +21,9 @@ <PackageReference Include="Microsoft.Extensions.ObjectPool" Version="3.1.2" /> <PackageReference Include="System.Threading.Channels" Version="4.7.0" /> </ItemGroup> + + <ItemGroup> + <Folder Include="Extensions\" /> + <Folder Include="Interfaces\" /> + </ItemGroup> </Project> \ No newline at end of file diff --git a/NetSharp/NetSharp/Server.cs b/NetSharp/NetSharp/Server.cs @@ -1,601 +0,0 @@ -using System; -using System.Collections.Concurrent; -using System.Net; -using System.Net.Sockets; -using System.Runtime.CompilerServices; -using System.Threading; -using System.Threading.Tasks; -using NetSharp.Interfaces; -using NetSharp.Packets; -using NetSharp.Packets.Builtin; -using NetSharp.Utils.Socket_Options; - -namespace NetSharp -{ - /// <summary> - /// Represents a method that receives a request packet of the given type (<typeparamref name="TReq"/>) and - /// handles the request, returning a response packet of the given type (<typeparamref name="TRep"/>). - /// </summary> - /// <typeparam name="TReq">The type of request packet handled by this delegate method.</typeparam> - /// <typeparam name="TRep">The type of response packet returned by this delegate method.</typeparam> - /// <param name="requestPacket">The request packet that should be handled by this delegate method.</param> - /// <param name="remoteEndPoint">The remote endpoint from which the request originated.</param> - /// <returns>The response packet to send back to the remote endpoint from which the request originated.</returns> - public delegate TRep ComplexPacketHandler<in TReq, out TRep>(TReq requestPacket, EndPoint remoteEndPoint) - where TReq : class, IRequestPacket, new() where TRep : class, IResponsePacket<TReq>, new(); - - /// <summary> - /// Represents a method that receives a simple request packet of the given type (<typeparamref name="TReq"/>) and - /// handles the request, not returning any response packets. - /// </summary> - /// <typeparam name="TReq">The type of request packet handled by this delegate method.</typeparam> - /// <param name="requestPacket">The request packet that should be handled by this delegate method.</param> - /// <param name="remoteEndPoint">The remote endpoint from which the request originated.</param> - public delegate void SimplePacketHandler<in TReq>(TReq requestPacket, EndPoint remoteEndPoint) - where TReq : class, IRequestPacket, new(); - - /// <summary> - /// Provides methods for handling connected <see cref="IClient"/> instances. - /// </summary> - public abstract class Server : Connection, IServer, IDisposable - { - /// <summary> - /// Maps a packet type id to the complex packet handler for that packet type. - /// </summary> - private readonly ConcurrentDictionary<uint, Func<IRequestPacket, EndPoint, IResponsePacket<IRequestPacket>>> - complexPacketHandlers; - - /// <summary> - /// Maps a packet type id to the raw packet deserialiser that deserialises raw packets to - /// <see cref="IRequestPacket"/> implementors. - /// </summary> - private readonly ConcurrentDictionary<uint, RawRequestPacketDeserialiser> requestPacketDeserialisers; - - /// <summary> - /// Cancellation token source to stop handling client sockets when the server should be shut down. - /// </summary> - private readonly CancellationTokenSource serverShutdownCancellationTokenSource; - - /// <summary> - /// Maps a packet type id to the simple packet handler for that packet type. - /// </summary> - private readonly ConcurrentDictionary<uint, Action<IRequestPacket, EndPoint>> simplePacketHandlers; - - /// <summary> - /// Initialises a new instance of the <see cref="Server"/> class. - /// </summary> - private Server() - { - serverShutdownCancellationTokenSource = new CancellationTokenSource(); - - serverShutdownCancellationToken = serverShutdownCancellationTokenSource.Token; - - requestPacketDeserialisers = new ConcurrentDictionary<uint, RawRequestPacketDeserialiser>(); - - simplePacketHandlers = new ConcurrentDictionary<uint, Action<IRequestPacket, EndPoint>>(); - complexPacketHandlers = - new ConcurrentDictionary<uint, Func<IRequestPacket, EndPoint, IResponsePacket<IRequestPacket>>>(); - - RegisterInternalPacketHandlers(); - - socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); - socketOptions = new DefaultSocketOptions(ref socket); - } - - /// <summary> - /// Destroys an instance of the <see cref="Server"/> class. - /// </summary> - ~Server() - { - Dispose(false); - } - - /// <summary> - /// Represents a method that receives a raw packet, and deserialises it into an <see cref="IRequestPacket"/> implementor. - /// </summary> - /// <param name="rawPacket">The raw packet that was received from the network.</param> - /// <returns>The deserialised instance of the packet.</returns> - private delegate IRequestPacket RawRequestPacketDeserialiser(in SerialisedPacket rawPacket); - - /// <summary> - /// Registers packet handlers for every internal library packet. - /// </summary> - private void RegisterInternalPacketHandlers() - { - TryRegisterSimplePacketHandler((DisconnectPacket packet, EndPoint remoteEndPoint) => - { -#if DEBUG - logger.LogMessage($"Received disconnect packet from {remoteEndPoint}"); -#endif - OnClientDisconnected(remoteEndPoint); - }); - - TryRegisterSimplePacketHandler((SimpleDataPacket packet, EndPoint remoteEndPoint) => - { -#if DEBUG - logger.LogMessage($"Received {packet.RequestBuffer.Length} bytes from {remoteEndPoint}"); -#endif - }); - - TryRegisterComplexPacketHandler((ConnectPacket packet, EndPoint remoteEndPoint) => - { - OnClientConnected(remoteEndPoint); -#if DEBUG - logger.LogMessage($"Received connection request from {remoteEndPoint}"); -#endif - return new ConnectResponsePacket { RequestPacket = packet }; - }); - - TryRegisterComplexPacketHandler((PingPacket packet, EndPoint remoteEndPoint) => - new PingResponsePacket { RequestPacket = packet }); - - TryRegisterComplexPacketHandler((DataPacket packet, EndPoint remoteEndPoint) => - { -#if DEBUG - logger.LogMessage($"Received {packet.RequestBuffer.Length} bytes from {remoteEndPoint}"); - logger.LogMessage($"Sending {packet.RequestBuffer.Length} bytes to {remoteEndPoint}"); -#endif - return new DataResponsePacket { RequestPacket = packet, ResponseBuffer = packet.RequestBuffer }; - }); - } - - /// <summary> - /// The maximum number of connections that are allowed in the connection backlog. - /// </summary> - protected const int PendingConnectionBacklog = 100; - - /// <summary> - /// The default timeout value for all network operations. - /// </summary> - protected static readonly TimeSpan DefaultNetworkOperationTimeout = TimeSpan.FromMilliseconds(10_000); - - /// <summary> - /// The cancellation token that will be set when the server must be shut down. - /// </summary> - protected readonly CancellationToken serverShutdownCancellationToken; - - /// <summary> - /// The <see cref="Socket"/> underlying the connection. - /// </summary> - protected readonly Socket socket; - - /// <summary> - /// Backing field for the <see cref="SocketOptions"/> property. - /// </summary> - protected readonly SocketOptions socketOptions; - - /// <summary> - /// Whether the server should be ran. - /// </summary> - protected volatile bool runServer; - - /// <summary> - /// Initialises a new instance of the <see cref="Server"/> class. - /// </summary> - /// <param name="socketType">The socket type for the underlying socket.</param> - /// <param name="protocolType">The protocol type for the underlying socket.</param> - /// <param name="socketManager">The <see cref="Utils.Socket_Options.SocketOptions"/> manager to use.</param> - protected Server(SocketType socketType, ProtocolType protocolType, SocketOptionManager socketManager) - : this(socketType, protocolType, socketManager, DefaultNetworkOperationTimeout) - { - } - - /// <summary> - /// Initialises a new instance of the <see cref="Server"/> class. - /// </summary> - /// <param name="socketType">The socket type for the underlying socket.</param> - /// <param name="protocolType">The protocol type for the underlying socket.</param> - /// <param name="socketManager">The <see cref="Utils.Socket_Options.SocketOptions"/> manager to use.</param> - /// <param name="networkOperationTimeout">The timeout value for send and receive operations over the network.</param> - protected Server(SocketType socketType, ProtocolType protocolType, SocketOptionManager socketManager, - TimeSpan networkOperationTimeout) : this() - { - socket = new Socket(AddressFamily.InterNetwork, socketType, protocolType); - - socketOptions = socketManager switch - { - SocketOptionManager.Tcp => new TcpSocketOptions(ref socket) as SocketOptions, - SocketOptionManager.Udp => new UdpSocketOptions(ref socket) as SocketOptions, - _ => new DefaultSocketOptions(ref socket), - }; - - NetworkOperationTimeout = networkOperationTimeout; - } - - /// <summary> - /// Deserialises the given <see cref="NetworkPacket"/> struct into an <see cref="IRequestPacket"/> implementor. - /// </summary> - /// <param name="packetType">The type id of packet that we should deserialise to.</param> - /// <param name="rawRequestPacket">The packet that should be deserialised.</param> - /// <returns>The deserialised packet instance, cast to the <see cref="IRequestPacket"/> interface.</returns> - protected IRequestPacket? DeserialiseRequestPacket(uint packetType, in SerialisedPacket rawRequestPacket) - { - if (requestPacketDeserialisers.TryGetValue(packetType, out RawRequestPacketDeserialiser deserialiser)) - { - return deserialiser.Invoke(rawRequestPacket); - } -#if DEBUG - logger.LogWarning($"No packet deserialiser was registered for packet of type {packetType}"); -#endif - return default; - } - - /// <summary> - /// Disposes of this <see cref="Server"/> instance. - /// </summary> - /// <param name="disposing">Whether this instance is being disposed.</param> - protected override void Dispose(bool disposing) - { - if (disposing) - { - serverShutdownCancellationTokenSource?.Cancel(); - serverShutdownCancellationTokenSource?.Dispose(); - - socket.Dispose(); - } - - base.Dispose(disposing); - } - - /// <summary> - /// Provides a task that represents the handling of a client. Calls the abstract <see cref="HandleClientAsync"/> method. - /// </summary> - /// <param name="clientHandlerArgsObj">The object representing the passed <see cref="ClientHandlerArgs"/> instance.</param> - protected async Task DoHandleClientAsync(object clientHandlerArgsObj) - { - ClientHandlerArgs clientHandlerArgs = (ClientHandlerArgs)clientHandlerArgsObj; - - try - { - await HandleClientAsync(clientHandlerArgs, serverShutdownCancellationTokenSource.Token); - } - catch (TaskCanceledException) - { - logger.LogWarning("Client handling was cancelled via a task cancellation."); - } - catch (OperationCanceledException) - { - logger.LogWarning("Client handling was cancelled via an operation cancellation."); - } - catch (Exception ex) - { - logger.LogException("Exception during client handling", ex); - } - finally - { - if (clientHandlerArgs.ClientSocket != null) - { - logger.LogMessage("Closing and releasing all resources associated with client handler socket"); - - clientHandlerArgs.ClientSocket.Shutdown(SocketShutdown.Both); - clientHandlerArgs.ClientSocket.Disconnect(true); - clientHandlerArgs.ClientSocket.Close(1); - clientHandlerArgs.ClientSocket.Dispose(); - } - } - } - - /// <summary> - /// Handles a client asynchronously. - /// </summary> - /// <param name="args">The client handler arguments that should be passed to the client handler.</param> - /// <param name="cancellationToken">Cancellation token set when the server is shutting down.</param> - protected abstract Task HandleClientAsync(ClientHandlerArgs args, CancellationToken cancellationToken); - - /// <summary> - /// Handles the given request packet with a registered packet handler. In this case, a complex packet handler - /// will override any registered simple packet handlers. - /// </summary> - /// <param name="packetType">The type id of the packet that we should handle.</param> - /// <param name="requestPacket">The packet instance that should be handled.</param> - /// <param name="remoteEndPoint">The remote endpoint from which the request packet originated.</param> - /// <returns>The response packet that should be sent back to the remote endpoint.</returns> - protected IResponsePacket<IRequestPacket>? HandleRequestPacket(uint packetType, in IRequestPacket requestPacket, - in EndPoint remoteEndPoint) - { - try - { - if (complexPacketHandlers.ContainsKey(packetType)) - { - IResponsePacket<IRequestPacket> response = - complexPacketHandlers[packetType].Invoke(requestPacket, remoteEndPoint); - - return response; - } - - if (simplePacketHandlers.ContainsKey(packetType)) - { - simplePacketHandlers[packetType].Invoke(requestPacket, remoteEndPoint); - return null; - } - -#if DEBUG - logger.LogWarning($"No packet handler was registered for packet of type {packetType}"); -#endif - } - catch (Exception ex) - { - logger.LogException( - $"Exception when invoking packet handler for packet (type: {packetType}) received from {remoteEndPoint}", - ex); - } - - return default; - } - - /// <summary> - /// Invokes the <see cref="ClientConnected"/> event. - /// </summary> - /// <param name="remoteEndPoint">The remote endpoint with which a connection was made.</param> - [MethodImpl(MethodImplOptions.AggressiveInlining)] - protected void OnClientConnected(EndPoint remoteEndPoint) => ClientConnected?.Invoke(remoteEndPoint); - - /// <summary> - /// Invokes the <see cref="ClientDisconnected"/> event. - /// </summary> - /// <param name="remoteEndPoint">The remote endpoint with which a connection was lost.</param> - [MethodImpl(MethodImplOptions.AggressiveInlining)] - protected void OnClientDisconnected(EndPoint remoteEndPoint) => ClientDisconnected?.Invoke(remoteEndPoint); - - /// <summary> - /// Invokes the <see cref="ServerStarted"/> event. - /// </summary> - [MethodImpl(MethodImplOptions.AggressiveInlining)] - protected void OnServerStarted() => ServerStarted?.Invoke(); - - /// <summary> - /// Invokes the <see cref="ServerStopped"/> event. - /// </summary> - [MethodImpl(MethodImplOptions.AggressiveInlining)] - protected void OnServerStopped() => ServerStopped?.Invoke(); - - /// <summary> - /// Attempts to synchronously bind the underlying socket to the given local endpoint. Blocks. - /// If the timeout is exceeded the binding attempt is aborted and the method returns false. - /// </summary> - /// <param name="localEndPoint">The local endpoint to bind to.</param> - /// <param name="timeout">The timeout within which to attempt the binding.</param> - /// <returns>Whether the binding was successful or not.</returns> - protected bool TryBind(EndPoint localEndPoint, TimeSpan timeout) => - TryBindAsync(localEndPoint, timeout).Result; - - /// <summary> - /// Attempts to asynchronously bind the underlying socket to the given local endpoint. Does not block. - /// If the timeout is exceeded the binding attempt is aborted and the method returns false. - /// </summary> - /// <param name="localEndPoint">The local endpoint to bind to.</param> - /// <param name="timeout">The timeout within which to attempt the binding.</param> - /// <returns>Whether the binding was successful or not.</returns> - protected async Task<bool> TryBindAsync(EndPoint localEndPoint, TimeSpan timeout) - { - using CancellationTokenSource timeoutCancellationTokenSource = new CancellationTokenSource(timeout); - using CancellationTokenSource cts = - CancellationTokenSource.CreateLinkedTokenSource(timeoutCancellationTokenSource.Token, serverShutdownCancellationToken); - - try - { - return await Task.Run(() => - { - socket.Bind(localEndPoint); - - return true; - }, cts.Token); - } - catch (TaskCanceledException) - { - return false; - } - catch (SocketException ex) - { - logger.LogException($"Socket exception on binding socket to {localEndPoint}:", ex); - return false; - } - } - - /// <summary> - /// Holds information about the arguments passed to every client handler task. - /// </summary> - protected readonly struct ClientHandlerArgs - { - /// <summary> - /// Initialises a new instance of the <see cref="ClientHandlerArgs"/> struct. - /// </summary> - /// <param name="remoteEndPoint">The remote endpoint of the client that should be handled.</param> - /// <param name="handlerSocket">The handler socket of the client that should be handled.</param> - private ClientHandlerArgs(EndPoint remoteEndPoint, Socket? handlerSocket) - { - ClientEndPoint = remoteEndPoint; - - ClientSocket = handlerSocket; - } - - /// <summary> - /// The remote endpoint for the client being handled. - /// </summary> - public readonly EndPoint ClientEndPoint; - - /// <summary> - /// The client handler socket for the client being handled. Is only set if using TCP. - /// </summary> - public readonly Socket? ClientSocket; - - /// <summary> - /// Constructs a new instance of the <see cref="ClientHandlerArgs"/> for a TCP client. - /// </summary> - /// <returns>A new instance of the <see cref="ClientHandlerArgs"/>, setup for a TCP client.</returns> - public static ClientHandlerArgs ForTcpClientHandler(in Socket clientHandlerSocket) - { - return new ClientHandlerArgs(clientHandlerSocket.RemoteEndPoint, clientHandlerSocket); - } - - /// <summary> - /// Constructs a new instance of the <see cref="ClientHandlerArgs"/> for a UDP client. - /// </summary> - /// <returns>A new instance of the <see cref="ClientHandlerArgs"/>, setup for a UDP client.</returns> - public static ClientHandlerArgs ForUdpClientHandler(in EndPoint clientEndPoint) - { - return new ClientHandlerArgs(clientEndPoint, null); - } - } - - /// <summary> - /// Signifies that a connection with a remote endpoint has been made. - /// </summary> - public event Action<EndPoint>? ClientConnected; - - //protected IResponsePacket<IRequestPacket> DeserialiseResponsePacket(in Packet) - /// <summary> - /// Signifies that a connection with a remote endpoint has been lost. - /// </summary> - public event Action<EndPoint>? ClientDisconnected; - - /// <summary> - /// Signifies that the server was started and clients will start being accepted. - /// </summary> - public event Action? ServerStarted; - - /// <summary> - /// Signifies that the server was stopped and clients will stop being accepted. - /// </summary> - public event Action? ServerStopped; - - /// <summary> - /// The timeout value for network operations such as sending bytes or receiving bytes over the network. - /// </summary> - public TimeSpan NetworkOperationTimeout { get; protected set; } - - /// <summary> - /// The configured socket options for the underlying connection. - /// </summary> - public SocketOptions SocketOptions - { - get { return socketOptions; } - } - - /// <inheritdoc /> - public abstract Task RunAsync(EndPoint localEndPoint); - - /// <inheritdoc /> - public void Shutdown() - { - runServer = false; - logger.LogMessage("Signalling server shutdown to all client handlers..."); - serverShutdownCancellationTokenSource.Cancel(); - } - - /// <inheritdoc /> - public bool TryDeregisterComplexPacketHandler<Req, Rep>(out ComplexPacketHandler<Req, Rep>? oldHandlerDelegate) - where Req : class, IRequestPacket, new() where Rep : class, IResponsePacket<Req>, new() - { - uint packetTypeId = PacketRegistry.GetPacketId<Req>(); - oldHandlerDelegate = default; - - try - { - requestPacketDeserialisers.TryRemove(packetTypeId, out _); - - if (!complexPacketHandlers.TryGetValue(packetTypeId, - out Func<IRequestPacket, EndPoint, IResponsePacket<IRequestPacket>> oldDelegate)) - return false; - - oldHandlerDelegate = (p, ep) => (Rep)oldDelegate(p, ep); - return true; - } - catch (Exception ex) - { - logger.LogException("Exception when deregistering complex packet handler", ex); - } - - return false; - } - - /// <inheritdoc /> - public bool TryDeregisterSimplePacketHandler<Req>(out SimplePacketHandler<Req>? oldHandlerDelegate) - where Req : class, IRequestPacket, new() - { - uint packetTypeId = PacketRegistry.GetPacketId<Req>(); - oldHandlerDelegate = default; - - try - { - requestPacketDeserialisers.TryRemove(packetTypeId, out _); - - if (!simplePacketHandlers.TryGetValue(packetTypeId, out Action<IRequestPacket, EndPoint> oldDelegate)) - return false; - - oldHandlerDelegate = (p, ep) => oldDelegate(p, ep); - return true; - } - catch (Exception ex) - { - logger.LogException("Exception when deregistering simple packet handler", ex); - } - - return false; - } - - /// <inheritdoc /> - public bool TryRegisterComplexPacketHandler<Req, Rep>(ComplexPacketHandler<Req, Rep> handlerDelegate) - where Req : class, IRequestPacket, new() where Rep : class, IResponsePacket<Req>, new() - { - uint packetTypeId = PacketRegistry.GetPacketId<Req>(); - - static IRequestPacket PacketDeserialiser(in SerialisedPacket packet) => SerialisedPacket.To<Req>(packet); - - IResponsePacket<IRequestPacket> MappedHandlerDelegate(IRequestPacket p, EndPoint ep) - { - Rep responsePacket = handlerDelegate((Req)p, ep); - return responsePacket; - } - - try - { - requestPacketDeserialisers.AddOrUpdate(packetTypeId, - key => PacketDeserialiser, - (key, oldDeserialiser) => PacketDeserialiser); - - complexPacketHandlers.AddOrUpdate(packetTypeId, - key => MappedHandlerDelegate, - (key, oldDelegate) => MappedHandlerDelegate); - - return true; - } - catch (Exception ex) - { - logger.LogException("Exception when registering complex packet handler", ex); - } - - return false; - } - - /// <inheritdoc /> - public bool TryRegisterSimplePacketHandler<Req>(SimplePacketHandler<Req> handlerDelegate) - where Req : class, IRequestPacket, new() - { - uint packetTypeId = PacketRegistry.GetPacketId<Req>(); - - static IRequestPacket PacketDeserialiser(in SerialisedPacket packet) => SerialisedPacket.To<Req>(packet); - - void MappedHandlerDelegate(IRequestPacket p, EndPoint ep) => handlerDelegate((Req)p, ep); - - try - { - requestPacketDeserialisers.AddOrUpdate(packetTypeId, - key => PacketDeserialiser, - (key, oldDeserialiser) => PacketDeserialiser); - - simplePacketHandlers.AddOrUpdate(packetTypeId, - key => MappedHandlerDelegate, - (key, oldDelegate) => MappedHandlerDelegate); - - return true; - } - catch (Exception ex) - { - logger.LogException("Exception when registering simple packet handler", ex); - } - - return false; - } - } -} -\ No newline at end of file diff --git a/NetSharp/NetSharp/Sockets/SocketAcceptor.cs b/NetSharp/NetSharp/Sockets/SocketAcceptor.cs @@ -0,0 +1,212 @@ +using System; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.ObjectPool; + +namespace NetSharp.Sockets +{ + public sealed class SocketAcceptor + { + private readonly ObjectPool<SocketAsyncEventArgs> acceptAsyncEventArgsPool; + private readonly ObjectPool<SocketAsyncEventArgs> connectAsyncEventArgsPool; + private readonly ObjectPool<SocketAsyncEventArgs> disconnectAsyncEventArgsPool; + + private void HandleIOCompleted(object? sender, SocketAsyncEventArgs args) + { + 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.AcceptSocket); + } + } + + acceptAsyncEventArgsPool.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(true); + } + } + + connectAsyncEventArgsPool.Return(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); + } + } + + disconnectAsyncEventArgsPool.Return(args); + + break; + + default: + throw new InvalidOperationException( + $"The {nameof(SocketAcceptor)} class doesn't support the {args.LastOperation} operation."); + } + } + + private readonly struct AsyncAcceptToken + { + public readonly CancellationToken CancellationToken; + public readonly TaskCompletionSource<Socket> CompletionSource; + + public AsyncAcceptToken(TaskCompletionSource<Socket> tcs, CancellationToken cancellationToken = default) + { + CompletionSource = tcs; + CancellationToken = cancellationToken; + } + } + + private readonly struct AsyncConnectToken + { + public readonly CancellationToken CancellationToken; + public readonly TaskCompletionSource<bool> CompletionSource; + + public AsyncConnectToken(TaskCompletionSource<bool> 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; + } + } + + internal SocketAcceptor(int maxPooledObjects = 10) + { + acceptAsyncEventArgsPool = new LeakTrackingObjectPool<SocketAsyncEventArgs>( + new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), + maxPooledObjects)); + + connectAsyncEventArgsPool = new LeakTrackingObjectPool<SocketAsyncEventArgs>( + new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), + maxPooledObjects)); + + disconnectAsyncEventArgsPool = new LeakTrackingObjectPool<SocketAsyncEventArgs>( + new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), + maxPooledObjects)); + + for (int i = 0; i < maxPooledObjects; i++) + { + SocketAsyncEventArgs acceptArgs = new SocketAsyncEventArgs(); + acceptArgs.Completed += HandleIOCompleted; + acceptAsyncEventArgsPool.Return(acceptArgs); + + SocketAsyncEventArgs connectArgs = new SocketAsyncEventArgs(); + connectArgs.Completed += HandleIOCompleted; + connectAsyncEventArgsPool.Return(connectArgs); + + SocketAsyncEventArgs disconnectArgs = new SocketAsyncEventArgs(); + disconnectArgs.Completed += HandleIOCompleted; + connectAsyncEventArgsPool.Return(disconnectArgs); + } + } + + public Task<Socket> AcceptAsync(Socket socket, CancellationToken cancellationToken = default) + { + TaskCompletionSource<Socket> tcs = new TaskCompletionSource<Socket>(); + + SocketAsyncEventArgs args = acceptAsyncEventArgsPool.Get(); + args.UserToken = new AsyncAcceptToken(tcs, cancellationToken); + + // if the accept operation doesn't complete synchronously, return the awaitable task + if (socket.AcceptAsync(args)) return tcs.Task; + + Socket result = args.AcceptSocket; + + acceptAsyncEventArgsPool.Return(args); + + return Task.FromResult(result); + } + + public Task ConnectAsync(Socket socket, EndPoint remoteEndPoint, CancellationToken cancellationToken = default) + { + TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>(); + + SocketAsyncEventArgs args = connectAsyncEventArgsPool.Get(); + args.RemoteEndPoint = remoteEndPoint; + args.UserToken = new AsyncConnectToken(tcs, cancellationToken); + + // if the connect operation doesn't complete synchronously, return the awaitable task + if (socket.ConnectAsync(args)) return tcs.Task; + + connectAsyncEventArgsPool.Return(args); + + return Task.CompletedTask; + } + + public Task DisconnectAsync(Socket socket, CancellationToken cancellationToken = default) + { + TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>(); + + SocketAsyncEventArgs args = connectAsyncEventArgsPool.Get(); + args.UserToken = new AsyncDisconnectToken(tcs, cancellationToken); + + // if the disconnect operation doesn't complete synchronously, return the awaitable task + if (socket.DisconnectAsync(args)) return tcs.Task; + + connectAsyncEventArgsPool.Return(args); + + return Task.CompletedTask; + } + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Sockets/SocketReader.cs b/NetSharp/NetSharp/Sockets/SocketReader.cs @@ -0,0 +1,178 @@ +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.Packets; +using NetSharp.Utils; + +namespace NetSharp.Sockets +{ + public sealed class SocketReader + { + private readonly ObjectPool<SocketAsyncEventArgs> receiveAsyncEventArgsPool; + private readonly ArrayPool<byte> receiveBufferPool; + + private void HandleIOCompleted(object? sender, SocketAsyncEventArgs args) + { + switch (args.LastOperation) + { + case SocketAsyncOperation.Receive: + AsyncReadToken asyncReceiveToken = (AsyncReadToken)args.UserToken; + + if (asyncReceiveToken.CancellationToken.IsCancellationRequested) + { + asyncReceiveToken.CompletionSource.SetCanceled(); + } + else + { + if (args.SocketError != SocketError.Success) + { + asyncReceiveToken.CompletionSource.SetException( + new SocketException((int)args.SocketError)); + } + else + { + args.MemoryBuffer.CopyTo(asyncReceiveToken.UserBuffer); + + TransmissionResult result = new TransmissionResult(args); + + asyncReceiveToken.CompletionSource.SetResult(result); + } + } + + receiveBufferPool.Return(asyncReceiveToken.RentedBuffer, true); + receiveAsyncEventArgsPool.Return(args); + + break; + + 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 + { + args.MemoryBuffer.CopyTo(asyncReceiveFromToken.UserBuffer); + + TransmissionResult result = new TransmissionResult(args); + + asyncReceiveFromToken.CompletionSource.SetResult(result); + } + } + + receiveBufferPool.Return(asyncReceiveFromToken.RentedBuffer, true); + receiveAsyncEventArgsPool.Return(args); + + break; + + default: + throw new InvalidOperationException( + $"The {nameof(SocketReader)} class doesn't support the {args.LastOperation} operation."); + } + } + + private readonly struct AsyncReadToken + { + public readonly CancellationToken CancellationToken; + public readonly TaskCompletionSource<TransmissionResult> CompletionSource; + public readonly byte[] RentedBuffer; + public readonly Memory<byte> UserBuffer; + + public AsyncReadToken(byte[] rentedBuffer, Memory<byte> userBuffer, TaskCompletionSource<TransmissionResult> tcs, + CancellationToken cancellationToken = default) + { + RentedBuffer = rentedBuffer; + UserBuffer = userBuffer; + + CompletionSource = tcs; + CancellationToken = cancellationToken; + } + } + + internal SocketReader(int packetBufferLength = NetworkPacket.PacketSize, int maxPooledObjects = 10, + bool preallocateBuffers = false) + { + PacketBufferLength = packetBufferLength; + + receiveBufferPool = ArrayPool<byte>.Create(packetBufferLength, maxPooledObjects); + + receiveAsyncEventArgsPool = new LeakTrackingObjectPool<SocketAsyncEventArgs>( + new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), + maxPooledObjects)); + + for (int i = 0; i < maxPooledObjects; i++) + { + SocketAsyncEventArgs args = new SocketAsyncEventArgs(); + args.Completed += HandleIOCompleted; + receiveAsyncEventArgsPool.Return(args); + } + } + + public int PacketBufferLength { get; } + + public Task<TransmissionResult> ReceiveAsync(Socket socket, SocketFlags socketFlags, + Memory<byte> outputBuffer, CancellationToken cancellationToken = default) + { + TaskCompletionSource<TransmissionResult> tcs = new TaskCompletionSource<TransmissionResult>(); + + byte[] rentedReceiveBuffer = receiveBufferPool.Rent(PacketBufferLength); + Memory<byte> rentedReceiveBufferMemory = new Memory<byte>(rentedReceiveBuffer); + + SocketAsyncEventArgs args = receiveAsyncEventArgsPool.Get(); + args.SetBuffer(rentedReceiveBufferMemory); + args.SocketFlags = socketFlags; + args.UserToken = new AsyncReadToken(rentedReceiveBuffer, outputBuffer, tcs, cancellationToken); + + // if the receive operation doesn't complete synchronously, returns the awaitable task + if (socket.ReceiveAsync(args)) return tcs.Task; + + args.MemoryBuffer.CopyTo(outputBuffer); + + TransmissionResult result = new TransmissionResult(args); + + receiveBufferPool.Return(rentedReceiveBuffer, true); + receiveAsyncEventArgsPool.Return(args); + + return Task.FromResult(result); + } + + public Task<TransmissionResult> ReceiveFromAsync(Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, + Memory<byte> outputBuffer, CancellationToken cancellationToken = default) + { + TaskCompletionSource<TransmissionResult> tcs = new TaskCompletionSource<TransmissionResult>(); + + byte[] rentedReceiveFromBuffer = receiveBufferPool.Rent(PacketBufferLength); + Memory<byte> rentedReceiveFromBufferMemory = new Memory<byte>(rentedReceiveFromBuffer); + + SocketAsyncEventArgs args = receiveAsyncEventArgsPool.Get(); + args.SetBuffer(rentedReceiveFromBufferMemory); + args.SocketFlags = socketFlags; + args.RemoteEndPoint = remoteEndPoint; + args.UserToken = new AsyncReadToken(rentedReceiveFromBuffer, outputBuffer, tcs, cancellationToken); + + // if the receive operation doesn't complete synchronously, returns the awaitable task + if (socket.ReceiveFromAsync(args)) return tcs.Task; + + args.MemoryBuffer.CopyTo(outputBuffer); + + TransmissionResult result = new TransmissionResult(args); + + receiveBufferPool.Return(rentedReceiveFromBuffer, true); + receiveAsyncEventArgsPool.Return(args); + + return Task.FromResult(result); + } + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Sockets/SocketWriter.cs b/NetSharp/NetSharp/Sockets/SocketWriter.cs @@ -0,0 +1,167 @@ +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.Packets; + +namespace NetSharp.Sockets +{ + public sealed class SocketWriter + { + private readonly ObjectPool<SocketAsyncEventArgs> sendAsyncEventArgsPool; + private readonly ArrayPool<byte> sendBufferPool; + + private void HandleIOCompleted(object? sender, SocketAsyncEventArgs args) + { + switch (args.LastOperation) + { + case SocketAsyncOperation.Send: + AsyncWriteToken asyncSendToken = (AsyncWriteToken)args.UserToken; + + if (asyncSendToken.CancellationToken.IsCancellationRequested) + { + asyncSendToken.CompletionSource.SetCanceled(); + } + else + { + if (args.SocketError != SocketError.Success) + { + asyncSendToken.CompletionSource.SetException( + new SocketException((int)args.SocketError)); + } + else + { + asyncSendToken.CompletionSource.SetResult(args.BytesTransferred); + } + } + + sendBufferPool.Return(asyncSendToken.RentedBuffer, true); + sendAsyncEventArgsPool.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 + { + asyncSendToToken.CompletionSource.SetResult(args.BytesTransferred); + } + } + + sendBufferPool.Return(asyncSendToToken.RentedBuffer, true); + sendAsyncEventArgsPool.Return(args); + + break; + + default: + throw new InvalidOperationException( + $"The {nameof(SocketWriter)} class doesn't support the {args.LastOperation} operation."); + } + } + + private readonly struct AsyncWriteToken + { + public readonly CancellationToken CancellationToken; + public readonly TaskCompletionSource<int> CompletionSource; + public readonly byte[] RentedBuffer; + + public AsyncWriteToken(byte[] rentedBuffer, TaskCompletionSource<int> tcs, + CancellationToken cancellationToken = default) + { + RentedBuffer = rentedBuffer; + + CompletionSource = tcs; + CancellationToken = cancellationToken; + } + } + + internal SocketWriter(int packetBufferLength = NetworkPacket.PacketSize, int maxPooledObjects = 10, + bool preallocateBuffers = false) + { + PacketBufferLength = packetBufferLength; + + sendBufferPool = ArrayPool<byte>.Create(packetBufferLength, maxPooledObjects); + + sendAsyncEventArgsPool = new LeakTrackingObjectPool<SocketAsyncEventArgs>( + new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), + maxPooledObjects)); + + for (int i = 0; i < maxPooledObjects; i++) + { + SocketAsyncEventArgs args = new SocketAsyncEventArgs(); + args.Completed += HandleIOCompleted; + sendAsyncEventArgsPool.Return(args); + } + } + + public int PacketBufferLength { get; } + + public Task<int> SendAsync(Socket socket, SocketFlags socketFlags, Memory<byte> outputBuffer, + CancellationToken cancellationToken = default) + { + TaskCompletionSource<int> tcs = new TaskCompletionSource<int>(); + + byte[] rentedSendBuffer = sendBufferPool.Rent(PacketBufferLength); + Memory<byte> rentedSendBufferMemory = new Memory<byte>(rentedSendBuffer); + + outputBuffer.CopyTo(rentedSendBufferMemory); + + SocketAsyncEventArgs args = sendAsyncEventArgsPool.Get(); + args.SetBuffer(rentedSendBufferMemory); + args.SocketFlags = socketFlags; + args.UserToken = new AsyncWriteToken(rentedSendBuffer, tcs, cancellationToken); + + // if the send operation doesn't complete synchronously, return the awaitable task + if (socket.SendAsync(args)) return tcs.Task; + + int result = args.BytesTransferred; + + sendBufferPool.Return(rentedSendBuffer, true); + sendAsyncEventArgsPool.Return(args); + + return Task.FromResult(result); + } + + public Task<int> SendToAsync(Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, + Memory<byte> outputBuffer, CancellationToken cancellationToken = default) + { + TaskCompletionSource<int> tcs = new TaskCompletionSource<int>(); + + byte[] rentedSendToBuffer = sendBufferPool.Rent(PacketBufferLength); + Memory<byte> rentedSendToBufferMemory = new Memory<byte>(rentedSendToBuffer); + + outputBuffer.CopyTo(rentedSendToBufferMemory); + + SocketAsyncEventArgs args = sendAsyncEventArgsPool.Get(); + args.SetBuffer(rentedSendToBufferMemory); + args.SocketFlags = socketFlags; + args.RemoteEndPoint = remoteEndPoint; + args.UserToken = new AsyncWriteToken(rentedSendToBuffer, tcs, cancellationToken); + + // if the send operation doesn't complete synchronously, return the awaitable task + if (socket.SendToAsync(args)) return tcs.Task; + + int result = args.BytesTransferred; + + sendBufferPool.Return(rentedSendToBuffer, true); + sendAsyncEventArgsPool.Return(args); + + return Task.FromResult(result); + } + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Utils/NetworkOperations.cs b/NetSharp/NetSharp/Utils/NetworkOperations.cs @@ -1,400 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Net; -using System.Net.Sockets; -using System.Threading; -using System.Threading.Tasks; -using Microsoft.Extensions.ObjectPool; -using NetSharp.Extensions; -using NetSharp.Packets; -using NetSharp.Utils.Conversion; - -namespace NetSharp.Utils -{ - /// <summary> - /// Helper class for asynchronously performing common network operations, for both the UDP and TCP protocols. - /// </summary> - /// TODO: Implement cancellation support for network operations - /// TODO: Somehow dont run into the exception below - /// [Excep] Socket exception while reading bytes from 0.0.0.0:0: System.Net.Sockets.SocketException (10040): Komunikat wysłany na gniazdo datagramu był większy niż wewnętrzny bufor lub przekraczał inny sieciowy limit albo bufor używany do odbierania datagramów był mniejszy niż sam datagram. - /// at NetSharp.Extensions.SocketTask.GetResult() in G:\Git Repos\EnderRifter\NetSharp\NetSharp\NetSharp\Extensions\SocketExtensions.cs:line 158 - /// at NetSharp.Utils.NetworkOperations.ReadFromAsync(Socket socket, Int32 count, EndPoint remoteEndPoint, SocketFlags socketFlags) in G:\Git Repos\EnderRifter\NetSharp\NetSharp\NetSharp\Utils\NetworkOperations.cs:line 78 - /// at NetSharp.Utils.NetworkOperations.ReadPacketFromAsync(Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, CancellationToken cancellationToken) in G:\Git Repos\EnderRifter\NetSharp\NetSharp\NetSharp\Utils\NetworkOperations.cs:line 225 - /// at NetSharp.Connection.DoReceivePacketFromAsync(Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, TimeSpan timeout, CancellationToken cancellationToken) in G:\Git Repos\EnderRifter\NetSharp\NetSharp\NetSharp\Connection.cs:line 114 - internal static class NetworkOperations - { - /// <summary> - /// Reads the specified amount of data from the network, via the given socket. - /// The given <see cref="SocketFlags"/> are associated with the read. - /// </summary> - /// <param name="socket">The socket which should read data from the network.</param> - /// <param name="count">The number of bytes to read from the network.</param> - /// <param name="socketFlags">The socket flags associated with the receive operation.</param> - /// <returns>The result of the receive operation.</returns> - private static Task<TransmissionResult> ReadAsync(Socket socket, int count, SocketFlags socketFlags, CancellationToken cancellationToken = default) - { - /* - SocketAsyncEventArgs args = new SocketAsyncEventArgs(); - args.SetBuffer(new byte[count], 0, count); - args.SocketFlags = socketFlags; - SocketTask awaitableTask = new SocketTask(args); - - while (count > args.BytesTransferred) - { - await socket.ReceiveAsync(awaitableTask); - } - - return new TransmissionResult(args.MemoryBuffer, args.BytesTransferred, socket.RemoteEndPoint); - */ - - return Task.Factory.StartNew(() => - { - byte[] byteBuffer = new byte[count]; - int receivedBytesCount = 0; - - while (count > receivedBytesCount) - { - Span<byte> receivedBytes = - new Span<byte>(byteBuffer, receivedBytesCount, count - receivedBytesCount); - - receivedBytesCount += socket.Receive(receivedBytes, socketFlags); - } - - return new TransmissionResult(byteBuffer, receivedBytesCount, socket.RemoteEndPoint); - }, cancellationToken); - } - - /// <summary> - /// Reads a datagram segment of the given length from the given remote endpoint, via the given socket. - /// The given <see cref="SocketFlags"/> are associated with the read. - /// </summary> - /// <param name="socket">The socket which should read data from the network.</param> - /// <param name="count">The number of bytes to read from the network.</param> - /// <param name="remoteEndPoint">The remote endpoint from which data should be read.</param> - /// <param name="socketFlags">The socket flags associated with the receive operation.</param> - /// <returns>The result of the receive operation.</returns> - private static Task<TransmissionResult> ReadFromAsync(Socket socket, int count, EndPoint remoteEndPoint, SocketFlags socketFlags, CancellationToken cancellationToken = default) - { - /* - SocketAsyncEventArgs args = new SocketAsyncEventArgs(); - args.SetBuffer(new byte[count], 0, count); - args.SocketFlags = socketFlags; - args.RemoteEndPoint = remoteEndPoint; - SocketTask awaitableTask = new SocketTask(args); - - while (count > args.BytesTransferred) - { - await socket.ReceiveMessageFromAsync(awaitableTask); - } - - return new TransmissionResult(args.MemoryBuffer, args.BytesTransferred, args.RemoteEndPoint); - */ - - return Task.Factory.StartNew(() => - { - byte[] byteBuffer = new byte[count]; - EndPoint actualRemoteEndPoint = remoteEndPoint; - int receivedBytesCount = 0; - - while (count > receivedBytesCount) - { - receivedBytesCount += - socket.ReceiveMessageFrom(byteBuffer, receivedBytesCount, count - receivedBytesCount, - ref socketFlags, ref actualRemoteEndPoint, out IPPacketInformation packetInformation); - } - - return new TransmissionResult(byteBuffer, receivedBytesCount, actualRemoteEndPoint); - }, cancellationToken); - } - - /// <summary> - /// Writes the given data buffer to the network, via the given socket. - /// The given <see cref="SocketFlags"/> are associated with the write, and the given <see cref="CancellationToken"/> - /// is used to allow for asynchronous task cancellation. - /// </summary> - /// <param name="socket">The socket which should write data to the network.</param> - /// <param name="buffer">The buffer that should be written to the network.</param> - /// <param name="socketFlags">The socket flags associated with the send operation.</param> - private static Task WriteAsync(Socket socket, Memory<byte> buffer, SocketFlags socketFlags, CancellationToken cancellationToken = default) - { - /* - SocketAsyncEventArgs args = new SocketAsyncEventArgs(); - args.SetBuffer(buffer); - args.SocketFlags = socketFlags; - SocketTask awaitableTask = new SocketTask(args); - - int bytesToSend = buffer.Length; - while (bytesToSend > args.BytesTransferred) - { - await socket.SendAsync(awaitableTask); - } - */ - - return Task.Factory.StartNew(() => - { - int bytesToSend = buffer.Length; - int sentBytesCount = 0; - - while (bytesToSend > sentBytesCount) - { - ReadOnlySpan<byte> bufferSegment = buffer.Span.Slice(sentBytesCount, bytesToSend - sentBytesCount); - - sentBytesCount += socket.Send(bufferSegment, socketFlags); - } - }, cancellationToken); - } - - /// <summary> - /// Writes the given data buffer to the given remote endpoint, via the given socket. - /// The given <see cref="SocketFlags"/> are associated with the write, and the given <see cref="CancellationToken"/> - /// is used to allow for asynchronous task cancellation. - /// </summary> - /// <param name="socket">The socket which should write data to the network.</param> - /// <param name="remoteEndPoint">The remote endpoint to which data should be written.</param> - /// <param name="buffer">The buffer that should be written to the network.</param> - /// <param name="socketFlags">The socket flags associated with the send operation.</param> - private static Task WriteToAsync(Socket socket, EndPoint remoteEndPoint, Memory<byte> buffer, SocketFlags socketFlags, CancellationToken cancellationToken = default) - { - /* - SocketAsyncEventArgs args = new SocketAsyncEventArgs(); - args.SetBuffer(buffer); - args.SocketFlags = socketFlags; - args.RemoteEndPoint = remoteEndPoint; - SocketTask awaitableTask = new SocketTask(args); - - int bytesToSend = buffer.Length; - while (bytesToSend > args.BytesTransferred) - { - await socket.SendToAsync(awaitableTask); - } - */ - - return Task.Factory.StartNew(() => - { - int bytesToSend = buffer.Length; - int sentBytesCount = 0; - - while (bytesToSend > sentBytesCount) - { - ReadOnlySpan<byte> bufferSegment = buffer.Span.Slice(sentBytesCount, bytesToSend - sentBytesCount); - - sentBytesCount += socket.SendTo(bufferSegment.ToArray(), socketFlags, remoteEndPoint); - } - }, cancellationToken); - } - - /// <summary> - /// Reads a packet from network, via the given socket. - /// The given <see cref="SocketFlags"/> are associated with the read. - /// </summary> - /// <param name="socket">The socket which should read the packet from the network.</param> - /// <param name="socketFlags">The socket flags associated with the receive operation.</param> - /// <param name="cancellationToken">The cancellation token that should be observed for the duration of the task.</param> - /// <returns>The read packet, and the endpoint from which it was read.</returns> - internal static async Task<(SerialisedPacket packet, EndPoint remoteEndPoint)> ReadPacketAsync(Socket socket, SocketFlags socketFlags, - CancellationToken cancellationToken = default) - { - List<ReadOnlyMemory<byte>> userDataBuffer = new List<ReadOnlyMemory<byte>>(1); - int receivedBytes = 0; - - EndPoint remoteEndPoint; - NetworkPacket receivedPacket; - uint receivedPacketType; - - do - { - TransmissionResult result = await ReadAsync(socket, NetworkPacket.PacketSize, socketFlags, cancellationToken); - - remoteEndPoint = result.RemoteEndPoint; - receivedPacket = NetworkPacket.Deserialise(result.Buffer); - receivedPacketType = receivedPacket.Header.Type; - - // TODO: try to remove this extra allocation - userDataBuffer.Add(receivedPacket.DataBuffer); - receivedBytes += receivedPacket.Header.DataLength; - } while (receivedPacket.Footer.HasSucceedingPacket); - - Memory<byte> finalBuffer = new byte[receivedBytes]; - int writtenBytes = 0; - - foreach (ReadOnlyMemory<byte> bufferSegment in userDataBuffer) - { - bufferSegment.CopyTo(finalBuffer.Slice(writtenBytes, bufferSegment.Length)); - writtenBytes += bufferSegment.Length; - } - - SerialisedPacket finalPacket = new SerialisedPacket(finalBuffer, receivedPacketType); - - return (finalPacket, remoteEndPoint); - - /* - TransmissionResult packetHeaderResult = - await ReadAsync(socket, NetworkPacket.HeaderSize, socketFlags, cancellationToken); - - int packetSize = EndianAwareBitConverter.ToInt32(packetHeaderResult.Buffer.Span.Slice(0, sizeof(int))); - - if (packetSize == 0) - { - return NetworkPacket.Deserialise(packetHeaderResult.Buffer); - } - - TransmissionResult packetDataResult = - await ReadAsync(socket, packetSize, socketFlags, cancellationToken); - - byte[] serialisedPacket = new byte[NetworkPacket.HeaderSize + packetSize]; - - Memory<byte> serialisedPacketHeader = new Memory<byte>(serialisedPacket, 0, NetworkPacket.HeaderSize); - packetHeaderResult.Buffer.CopyTo(serialisedPacketHeader); - - Memory<byte> serialisedPacketData = new Memory<byte>(serialisedPacket, NetworkPacket.HeaderSize, packetSize); - packetDataResult.Buffer.CopyTo(serialisedPacketData); - - return NetworkPacket.Deserialise(serialisedPacket); - */ - } - - /// <summary> - /// Reads a packet from the given remote endpoint, via the given socket. - /// The given <see cref="SocketFlags"/> are associated with the read. - /// </summary> - /// <param name="socket">The socket which should read the packet from the network.</param> - /// <param name="remoteEndPoint">The remote endpoint from which a packet should be read.</param> - /// <param name="socketFlags">The socket flags associated with the receive operation.</param> - /// <param name="cancellationToken">The cancellation token that should be observed for the duration of the task.</param> - /// <returns>The read packet, and the endpoint from which it was read.</returns> - internal static async Task<(SerialisedPacket packet, EndPoint remoteEndPoint)> ReadPacketFromAsync( - Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, CancellationToken cancellationToken = default) - { - List<ReadOnlyMemory<byte>> userDataBuffer = new List<ReadOnlyMemory<byte>>(1); - int receivedBytes = 0; - - NetworkPacket receivedPacket; - uint receivedPacketType; - - do - { - TransmissionResult result = - await ReadFromAsync(socket, NetworkPacket.PacketSize, remoteEndPoint, socketFlags, cancellationToken); - - remoteEndPoint = result.RemoteEndPoint; - receivedPacket = NetworkPacket.Deserialise(result.Buffer); - receivedPacketType = receivedPacket.Header.Type; - - // TODO: try to remove this extra allocation - userDataBuffer.Add(receivedPacket.DataBuffer); - receivedBytes += receivedPacket.Header.DataLength; - } while (receivedPacket.Footer.HasSucceedingPacket); - - Memory<byte> finalBuffer = new byte[receivedBytes]; - int writtenBytes = 0; - - foreach (ReadOnlyMemory<byte> bufferSegment in userDataBuffer) - { - bufferSegment.CopyTo(finalBuffer.Slice(writtenBytes, bufferSegment.Length)); - writtenBytes += bufferSegment.Length; - } - - SerialisedPacket finalPacket = new SerialisedPacket(finalBuffer, receivedPacketType); - - return (finalPacket, remoteEndPoint); - - /* - TransmissionResult packetHeaderResult = - await ReadFromAsync(socket, NetworkPacket.HeaderSize, remoteEndPoint, socketFlags, cancellationToken); - - int packetSize = EndianAwareBitConverter.ToInt32(packetHeaderResult.Buffer.Span.Slice(0, sizeof(int))); - - if (packetSize == 0) - { - return (NetworkPacket.Deserialise(packetHeaderResult.Buffer), packetHeaderResult); - } - - TransmissionResult packetDataResult = - await ReadFromAsync(socket, packetSize, packetHeaderResult.RemoteEndPoint, socketFlags, cancellationToken); - - byte[] serialisedPacket = new byte[NetworkPacket.HeaderSize + packetSize]; - - Memory<byte> serialisedPacketHeader = new Memory<byte>(serialisedPacket, 0, NetworkPacket.HeaderSize); - packetHeaderResult.Buffer.CopyTo(serialisedPacketHeader); - - Memory<byte> serialisedPacketData = new Memory<byte>(serialisedPacket, NetworkPacket.HeaderSize, packetSize); - packetDataResult.Buffer.CopyTo(serialisedPacketData); - - return (NetworkPacket.Deserialise(serialisedPacket), packetDataResult); - */ - } - - /// <summary> - /// Writes the given packet to the network, via the given socket. - /// The given <see cref="SocketFlags"/> are associated with the write. - /// </summary> - /// <param name="socket">The socket which should write data to the network.</param> - /// <param name="serialisedPacket">The packet that should be written to the network.</param> - /// <param name="socketFlags">The socket flags associated with the send operation.</param> - /// <param name="cancellationToken">The cancellation token that should be observed for the duration of the task.</param> - internal static async Task WritePacketAsync(Socket socket, SerialisedPacket serialisedPacket, SocketFlags socketFlags, - CancellationToken cancellationToken = default) - { - int packetCount = serialisedPacket.Contents.Length / NetworkPacket.PacketSize; - - for (int i = 0; i < packetCount; i++) - { - ReadOnlyMemory<byte> packetDataSegment = - serialisedPacket.Contents.Slice(NetworkPacket.PacketSize * i, NetworkPacket.PacketSize); - - NetworkPacket packet = new NetworkPacket(packetDataSegment, packetDataSegment.Length, - serialisedPacket.Type, NetworkErrorCode.Ok, true); - - await WriteAsync(socket, NetworkPacket.Serialise(packet), socketFlags, cancellationToken); - } - - if (serialisedPacket.Contents.Length % NetworkPacket.PacketSize != 0) - { - ReadOnlyMemory<byte> packetDataSegment = - serialisedPacket.Contents.Slice(NetworkPacket.PacketSize * packetCount, serialisedPacket.Contents.Length % NetworkPacket.PacketSize); - - NetworkPacket packet = new NetworkPacket(packetDataSegment, packetDataSegment.Length, - serialisedPacket.Type, NetworkErrorCode.Ok, true); - - await WriteAsync(socket, NetworkPacket.Serialise(packet), socketFlags, cancellationToken); - } - } - - /// <summary> - /// Writes the given packet to the given remote endpoint, via the given socket. - /// The given <see cref="SocketFlags"/> are associated with the write. - /// </summary> - /// <param name="socket">The socket which should write data to the network.</param> - /// <param name="remoteEndPoint">The remote endpoint to which data should be written.</param> - /// <param name="serialisedPacket">The packet that should be written to the remote endpoint.</param> - /// <param name="socketFlags">The socket flags associated with the send operation.</param> - /// <param name="cancellationToken">The cancellation token that should be observed for the duration of the task.</param> - internal static async Task WritePacketToAsync(Socket socket, EndPoint remoteEndPoint, SerialisedPacket serialisedPacket, SocketFlags socketFlags, - CancellationToken cancellationToken = default) - { - int packetCount = serialisedPacket.Contents.Length / NetworkPacket.PacketSize; - - for (int i = 0; i < packetCount; i++) - { - ReadOnlyMemory<byte> packetDataSegment = - serialisedPacket.Contents.Slice(NetworkPacket.PacketSize * i, NetworkPacket.PacketSize); - - NetworkPacket packet = new NetworkPacket(packetDataSegment, packetDataSegment.Length, - serialisedPacket.Type, NetworkErrorCode.Ok, true); - - await WriteToAsync(socket, remoteEndPoint, NetworkPacket.Serialise(packet), socketFlags, cancellationToken); - } - - if (serialisedPacket.Contents.Length % NetworkPacket.PacketSize != 0) - { - ReadOnlyMemory<byte> packetDataSegment = - serialisedPacket.Contents.Slice(NetworkPacket.PacketSize * packetCount, serialisedPacket.Contents.Length % NetworkPacket.PacketSize); - - NetworkPacket packet = new NetworkPacket(packetDataSegment, packetDataSegment.Length, - serialisedPacket.Type, NetworkErrorCode.Ok, true); - - await WriteToAsync(socket, remoteEndPoint, NetworkPacket.Serialise(packet), socketFlags, cancellationToken); - } - } - } -} -\ No newline at end of file diff --git a/NetSharp/NetSharp/Utils/NetworkOperationsManager.cs b/NetSharp/NetSharp/Utils/NetworkOperationsManager.cs @@ -0,0 +1,330 @@ +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.Packets; + +namespace NetSharp.Utils +{ + /// <summary> + /// Helper class for asynchronously performing common network operations in an efficient manner, for both the UDP and TCP protocols. + /// </summary> + public sealed class NetworkOperationsManager + { + private readonly ArrayPool<byte> receiveBufferPool; + private readonly ArrayPool<byte> receiveFromBufferPool; + private readonly ArrayPool<byte> sendBufferPool; + private readonly ArrayPool<byte> sendToBufferPool; + private readonly ObjectPool<SocketAsyncEventArgs> socketAsyncEventArgsPool; + + private void HandleIOCompleted(object? sender, SocketAsyncEventArgs eventArgs) + { + bool closed = false; + + switch (eventArgs.LastOperation) + { + case SocketAsyncOperation.Send: + AsyncWriteToken asyncSendToken = (AsyncWriteToken)eventArgs.UserToken; + + if (asyncSendToken.CancellationToken.IsCancellationRequested) + { + asyncSendToken.CompletionSource.SetCanceled(); + } + else + { + if (eventArgs.SocketError != SocketError.Success) + { + asyncSendToken.CompletionSource.SetException( + new SocketException((int)eventArgs.SocketError)); + } + else + { + asyncSendToken.CompletionSource.SetResult(eventArgs.BytesTransferred); + } + } + + sendBufferPool.Return(asyncSendToken.RentedBuffer, true); + socketAsyncEventArgsPool.Return(eventArgs); + + break; + + case SocketAsyncOperation.SendTo: + AsyncWriteToken asyncSendToToken = (AsyncWriteToken)eventArgs.UserToken; + + if (asyncSendToToken.CancellationToken.IsCancellationRequested) + { + asyncSendToToken.CompletionSource.SetCanceled(); + } + else + { + if (eventArgs.SocketError != SocketError.Success) + { + asyncSendToToken.CompletionSource.SetException( + new SocketException((int)eventArgs.SocketError)); + } + else + { + asyncSendToToken.CompletionSource.SetResult(eventArgs.BytesTransferred); + } + } + + sendToBufferPool.Return(asyncSendToToken.RentedBuffer, true); + socketAsyncEventArgsPool.Return(eventArgs); + + break; + + case SocketAsyncOperation.Receive: + AsyncReadToken asyncReceiveToken = (AsyncReadToken)eventArgs.UserToken; + + if (asyncReceiveToken.CancellationToken.IsCancellationRequested) + { + asyncReceiveToken.CompletionSource.SetCanceled(); + } + else + { + if (eventArgs.SocketError != SocketError.Success) + { + asyncReceiveToken.CompletionSource.SetException( + new SocketException((int)eventArgs.SocketError)); + } + else + { + eventArgs.MemoryBuffer.CopyTo(asyncReceiveToken.UserBuffer); + + TransmissionResult result = new TransmissionResult(eventArgs); + + asyncReceiveToken.CompletionSource.SetResult(result); + } + } + + receiveBufferPool.Return(asyncReceiveToken.RentedBuffer, true); + + break; + + case SocketAsyncOperation.ReceiveFrom: + AsyncReadToken asyncReceiveFromToken = (AsyncReadToken)eventArgs.UserToken; + + if (asyncReceiveFromToken.CancellationToken.IsCancellationRequested) + { + asyncReceiveFromToken.CompletionSource.SetCanceled(); + } + else + { + if (eventArgs.SocketError != SocketError.Success) + { + asyncReceiveFromToken.CompletionSource.SetException( + new SocketException((int)eventArgs.SocketError)); + } + else + { + eventArgs.MemoryBuffer.CopyTo(asyncReceiveFromToken.UserBuffer); + TransmissionResult result = new TransmissionResult(eventArgs); + + asyncReceiveFromToken.CompletionSource.SetResult(result); + } + } + + receiveFromBufferPool.Return(asyncReceiveFromToken.RentedBuffer, true); + + break; + + case SocketAsyncOperation.Disconnect: + closed = true; + break; + + case SocketAsyncOperation.Accept: + case SocketAsyncOperation.Connect: + case SocketAsyncOperation.ReceiveMessageFrom: + case SocketAsyncOperation.SendPackets: + case SocketAsyncOperation.None: + throw new NotImplementedException(); + + default: + throw new ArgumentOutOfRangeException(); + } + + if (closed) + { + // handle the client closing the connection on tcp servers at some point + } + } + + private readonly struct AsyncReadToken + { + public readonly CancellationToken CancellationToken; + public readonly TaskCompletionSource<TransmissionResult> CompletionSource; + public readonly byte[] RentedBuffer; + public readonly Memory<byte> UserBuffer; + + public AsyncReadToken(byte[] rentedBuffer, Memory<byte> userBuffer, TaskCompletionSource<TransmissionResult> tcs, + CancellationToken cancellationToken = default) + { + RentedBuffer = rentedBuffer; + UserBuffer = userBuffer; + + CompletionSource = tcs; + CancellationToken = cancellationToken; + } + } + + private readonly struct AsyncWriteToken + { + public readonly CancellationToken CancellationToken; + public readonly TaskCompletionSource<int> CompletionSource; + public readonly byte[] RentedBuffer; + + public AsyncWriteToken(byte[] rentedBuffer, TaskCompletionSource<int> tcs, + CancellationToken cancellationToken = default) + { + RentedBuffer = rentedBuffer; + + CompletionSource = tcs; + CancellationToken = cancellationToken; + } + } + + public const int DefaultMaximumPooledObjects = 10; + + public NetworkOperationsManager(int bufferSize = NetworkPacket.PacketSize, + int maxPooledObjectCount = DefaultMaximumPooledObjects, bool preallocateBuffers = false) + { + MaxPooledObjects = maxPooledObjectCount; + BufferSize = bufferSize; + + sendBufferPool = ArrayPool<byte>.Create(bufferSize, maxPooledObjectCount); + receiveBufferPool = ArrayPool<byte>.Create(bufferSize, maxPooledObjectCount); + + sendToBufferPool = ArrayPool<byte>.Create(bufferSize, maxPooledObjectCount); + receiveFromBufferPool = ArrayPool<byte>.Create(bufferSize, maxPooledObjectCount); + + socketAsyncEventArgsPool = new LeakTrackingObjectPool<SocketAsyncEventArgs>( + new DefaultObjectPool<SocketAsyncEventArgs>(new DefaultPooledObjectPolicy<SocketAsyncEventArgs>(), + maxPooledObjectCount)); + + for (int i = 0; i < MaxPooledObjects; i++) + { + SocketAsyncEventArgs args = new SocketAsyncEventArgs(); + args.Completed += HandleIOCompleted; + + socketAsyncEventArgsPool.Return(args); + } + + if (preallocateBuffers) + { + // TODO: Allocate and return array pool buffers + } + } + + public int BufferSize { get; } + + public int MaxPooledObjects { get; } + + public Task<TransmissionResult> ReceiveAsync(Socket socket, SocketFlags socketFlags, Memory<byte> outputBuffer, + CancellationToken cancellationToken = default) + { + TaskCompletionSource<TransmissionResult> tcs = new TaskCompletionSource<TransmissionResult>(); + + byte[] rentedReceiveBuffer = receiveBufferPool.Rent(BufferSize); + Memory<byte> rentedReceiveBufferMemory = new Memory<byte>(rentedReceiveBuffer); + + SocketAsyncEventArgs args = socketAsyncEventArgsPool.Get(); + + args.SetBuffer(rentedReceiveBufferMemory); + args.SocketFlags = socketFlags; + args.UserToken = new AsyncReadToken(rentedReceiveBuffer, outputBuffer, tcs, cancellationToken); + + // if the receive operation doesn't complete synchronously, returns the awaitable task + if (socket.ReceiveAsync(args)) return tcs.Task; + + args.MemoryBuffer.CopyTo(outputBuffer); + + TransmissionResult result = new TransmissionResult(args); + + receiveBufferPool.Return(rentedReceiveBuffer, true); + + return Task.FromResult(result); + } + + public Task<TransmissionResult> ReceiveFromAsync(Socket socket, EndPoint remoteEndPoint, SocketFlags socketFlags, + Memory<byte> outputBuffer, CancellationToken cancellationToken = default) + { + TaskCompletionSource<TransmissionResult> tcs = new TaskCompletionSource<TransmissionResult>(); + + byte[] rentedReceiveFromBuffer = receiveFromBufferPool.Rent(BufferSize); + Memory<byte> rentedReceiveFromBufferMemory = new Memory<byte>(rentedReceiveFromBuffer); + + SocketAsyncEventArgs args = socketAsyncEventArgsPool.Get(); + + args.SetBuffer(rentedReceiveFromBufferMemory); + args.SocketFlags = socketFlags; + args.RemoteEndPoint = remoteEndPoint; + args.UserToken = new AsyncReadToken(rentedReceiveFromBuffer, outputBuffer, tcs, cancellationToken); + + // if the receive operation doesn't complete synchronously, returns the awaitable task + if (socket.ReceiveFromAsync(args)) return tcs.Task; + + args.MemoryBuffer.CopyTo(outputBuffer); + + TransmissionResult result = new TransmissionResult(args); + + receiveFromBufferPool.Return(rentedReceiveFromBuffer, true); + + return Task.FromResult(result); + } + + public Task<int> SendAsync(Socket socket, SocketAsyncEventArgs receiveArgs, SocketFlags socketFlags, + Memory<byte> inputBuffer, CancellationToken cancellationToken = default) + { + TaskCompletionSource<int> tcs = new TaskCompletionSource<int>(); + + byte[] rentedSendBuffer = sendBufferPool.Rent(BufferSize); + Memory<byte> rentedSendBufferMemory = new Memory<byte>(rentedSendBuffer); + + inputBuffer.CopyTo(rentedSendBufferMemory); + + SocketAsyncEventArgs args = receiveArgs; + args.SetBuffer(rentedSendBufferMemory); + args.SocketFlags = socketFlags; + args.UserToken = new AsyncWriteToken(rentedSendBuffer, tcs, cancellationToken); + + // if the send operation doesn't complete synchronously, return the awaitable task + if (socket.SendAsync(args)) return tcs.Task; + + int result = args.BytesTransferred; + + sendBufferPool.Return(rentedSendBuffer, true); + socketAsyncEventArgsPool.Return(args); + + return Task.FromResult(result); + } + + public Task<int> SendToAsync(Socket socket, SocketAsyncEventArgs receiveFromArgs, SocketFlags socketFlags, + Memory<byte> inputBuffer, CancellationToken cancellationToken = default) + { + TaskCompletionSource<int> tcs = new TaskCompletionSource<int>(); + + byte[] rentedSendToBuffer = sendToBufferPool.Rent(BufferSize); + Memory<byte> rentedSendToBufferMemory = new Memory<byte>(rentedSendToBuffer); + + inputBuffer.CopyTo(rentedSendToBufferMemory); + + SocketAsyncEventArgs args = receiveFromArgs; + args.SetBuffer(rentedSendToBufferMemory); + args.SocketFlags = socketFlags; + args.UserToken = new AsyncWriteToken(rentedSendToBuffer, tcs, cancellationToken); + + // if the send operation doesn't complete synchronously, return the awaitable task + if (socket.SendToAsync(args)) return tcs.Task; + + int result = args.BytesTransferred; + + sendToBufferPool.Return(rentedSendToBuffer, true); + socketAsyncEventArgsPool.Return(args); + + return Task.FromResult(result); + } + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Utils/Socket Options/SocketOptionManager.cs b/NetSharp/NetSharp/Utils/Socket Options/SocketOptionManager.cs @@ -1,25 +0,0 @@ -namespace NetSharp.Utils.Socket_Options -{ - /// <summary> - /// Enumerates the possible socket option manager types to instantiate for a <see cref="Client"/> and <see cref="Server"/> - /// instance. - /// </summary> - public enum SocketOptionManager - { - /// <summary> - /// Causes a <see cref="DefaultSocketOptions"/> instance to be created as the socket option manager. - /// This means that certain socket options will throw an error, as the socket type is not specified. - /// </summary> - Default, - - /// <summary> - /// Causes a <see cref="TcpSocketOptions"/> instance to be created as the socket option manager. - /// </summary> - Tcp, - - /// <summary> - /// Causes a <see cref="UdpSocketOptions"/> instance to be created as the socket option manager. - /// </summary> - Udp, - } -} -\ No newline at end of file diff --git a/NetSharp/NetSharp/Utils/TransmissionResult.cs b/NetSharp/NetSharp/Utils/TransmissionResult.cs @@ -1,5 +1,6 @@ using System; using System.Net; +using System.Net.Sockets; namespace NetSharp.Utils { @@ -24,16 +25,20 @@ namespace NetSharp.Utils public readonly EndPoint RemoteEndPoint; /// <summary> + /// Socket arguments and other data associated with the transmission. + /// </summary> + public readonly SocketAsyncEventArgs TransmissionArgs; + + /// <summary> /// Initialises a new instance of the <see cref="TransmissionResult"/> struct. /// </summary> - /// <param name="buffer">The byte buffer that was transmitted.</param> - /// <param name="count">The number of bytes that were transmitted.</param> - /// <param name="remoteEndPoint">The remote endpoint to which the buffer was transmitted.</param> - public TransmissionResult(Memory<byte> buffer, int count, EndPoint remoteEndPoint) + /// <param name="args">The socket arguments associated with the transmission.</param> + public TransmissionResult(SocketAsyncEventArgs args) { - Buffer = buffer; - Count = count; - RemoteEndPoint = remoteEndPoint; + TransmissionArgs = args; + Buffer = args.MemoryBuffer; + Count = args.BytesTransferred; + RemoteEndPoint = args.RemoteEndPoint; } } } \ No newline at end of file