commit bf7b75413f7c02b06edde80b5a5d6aae892614a2 parent f57577263557f721bd905e829e248129c5ccfdf1 Author: Mikolaj Lenczewski <mikolaj.lenczewski308@gmail.com> Date: Fri, 24 Apr 2020 21:01:10 +0100 Cancellation for Socket Clients WIP. We have memory leaks and shit Diffstat:
15 files changed, 602 insertions(+), 176 deletions(-)
diff --git a/NetSharp/NetSharp/NetSharp.xml b/NetSharp/NetSharp/NetSharp.xml @@ -156,6 +156,30 @@ The memory buffer to write the serialised packet header instance to. </param> </member> + <member name="T:NetSharp.Sockets.Datagram.DatagramSocketClientOptions"> + <summary> + Provides additional configuration options for a <see cref="T:NetSharp.Sockets.Datagram.DatagramSocketClient" /> instance. + </summary> + </member> + <member name="F:NetSharp.Sockets.Datagram.DatagramSocketClientOptions.Defaults"> + <summary> + The default configuration. + </summary> + </member> + <member name="F:NetSharp.Sockets.Datagram.DatagramSocketClientOptions.PreallocatedTransmissionArgs"> + <summary> + The number of <see cref="T:System.Net.Sockets.SocketAsyncEventArgs" /> instances that should be preallocated for use in the + <see cref="M:NetSharp.Sockets.Datagram.DatagramSocketClient.SendToAsync(System.Net.EndPoint,System.Memory{System.Byte},System.Net.Sockets.SocketFlags,System.Threading.CancellationToken)" /> and <see cref="M:NetSharp.Sockets.Datagram.DatagramSocketClient.ReceiveFromAsync(System.Net.EndPoint,System.Memory{System.Byte},System.Net.Sockets.SocketFlags,System.Threading.CancellationToken)" /> methods. + </summary> + </member> + <member name="M:NetSharp.Sockets.Datagram.DatagramSocketClientOptions.#ctor(System.UInt16)"> + <summary> + Constructs a new instance of the <see cref="T:NetSharp.Sockets.Datagram.DatagramSocketClientOptions" /> struct. + </summary> + <param name="preallocatedTransmissionArgs"> + The number of <see cref="T:System.Net.Sockets.SocketAsyncEventArgs" /> instances to preallocate. + </param> + </member> <member name="M:NetSharp.Sockets.Datagram.DatagramSocketClient.CanTransmissionArgsBeReused(System.Net.Sockets.SocketAsyncEventArgs@)"> <inheritdoc /> </member> @@ -171,6 +195,39 @@ <member name="M:NetSharp.Sockets.Datagram.DatagramSocketClient.ResetTransmissionArgs(System.Net.Sockets.SocketAsyncEventArgs)"> <inheritdoc /> </member> + <member name="T:NetSharp.Sockets.Datagram.DatagramSocketServerOptions"> + <summary> + Provides additional configuration options for a <see cref="T:NetSharp.Sockets.Datagram.DatagramSocketServer" /> instance. + </summary> + </member> + <member name="F:NetSharp.Sockets.Datagram.DatagramSocketServerOptions.Defaults"> + <summary> + The default configuration. + </summary> + </member> + <member name="F:NetSharp.Sockets.Datagram.DatagramSocketServerOptions.ConcurrentReceiveFromCalls"> + <summary> + The number of <see cref="M:System.Net.Sockets.Socket.ReceiveFromAsync(System.Net.Sockets.SocketAsyncEventArgs)" /> calls that will be 'in-flight' at any one time, and ready to service incoming client + packets. This should be set to the number of client which will be connected at once. + </summary> + </member> + <member name="F:NetSharp.Sockets.Datagram.DatagramSocketServerOptions.PreallocatedTransmissionArgs"> + <summary> + The number of <see cref="T:System.Net.Sockets.SocketAsyncEventArgs" /> instances that should be preallocated for use in the <see cref="M:System.Net.Sockets.Socket.SendToAsync(System.Net.Sockets.SocketAsyncEventArgs)" /> + and <see cref="M:System.Net.Sockets.Socket.ReceiveFromAsync(System.Net.Sockets.SocketAsyncEventArgs)" /> methods. + </summary> + </member> + <member name="M:NetSharp.Sockets.Datagram.DatagramSocketServerOptions.#ctor(System.Int32,System.UInt16)"> + <summary> + Constructs a new instance of the <see cref="T:NetSharp.Sockets.Datagram.DatagramSocketServerOptions" /> struct. + </summary> + <param name="concurrentReceiveFromCalls"> + The number of <see cref="M:System.Net.Sockets.Socket.ReceiveFromAsync(System.Net.Sockets.SocketAsyncEventArgs)" /> calls which should be 'in-flight' at any one time. + </param> + <param name="preallocatedTransmissionArgs"> + The number of <see cref="T:System.Net.Sockets.SocketAsyncEventArgs" /> instances to preallocate. + </param> + </member> <member name="M:NetSharp.Sockets.Datagram.DatagramSocketServer.#ctor(System.Net.Sockets.AddressFamily@,System.Net.Sockets.ProtocolType@,NetSharp.Sockets.SocketServerPacketHandler@,System.Nullable{NetSharp.Sockets.Datagram.DatagramSocketServerOptions}@)"> <summary> Constructs a new instance of the <see cref="T:NetSharp.Sockets.Datagram.DatagramSocketServer" /> class. @@ -602,6 +659,30 @@ A <see cref="T:System.Threading.Tasks.Task" /> representing the server's execution. </returns> </member> + <member name="T:NetSharp.Sockets.Stream.StreamSocketClientOptions"> + <summary> + Provides additional configuration options for a <see cref="T:NetSharp.Sockets.Stream.StreamSocketClient" /> instance. + </summary> + </member> + <member name="F:NetSharp.Sockets.Stream.StreamSocketClientOptions.Defaults"> + <summary> + The default configuration. + </summary> + </member> + <member name="F:NetSharp.Sockets.Stream.StreamSocketClientOptions.PreallocatedTransmissionArgs"> + <summary> + The number of <see cref="T:System.Net.Sockets.SocketAsyncEventArgs" /> instances that should be preallocated for use in the + <see cref="M:NetSharp.Sockets.Stream.StreamSocketClient.SendAsync(System.Memory{System.Byte},System.Net.Sockets.SocketFlags,System.Threading.CancellationToken)" /> and <see cref="M:NetSharp.Sockets.Stream.StreamSocketClient.ReceiveAsync(System.Memory{System.Byte},System.Net.Sockets.SocketFlags,System.Threading.CancellationToken)" /> methods. + </summary> + </member> + <member name="M:NetSharp.Sockets.Stream.StreamSocketClientOptions.#ctor(System.UInt16)"> + <summary> + Constructs a new instance of the <see cref="T:NetSharp.Sockets.Stream.StreamSocketClientOptions" /> struct. + </summary> + <param name="preallocatedTransmissionArgs"> + The number of <see cref="T:System.Net.Sockets.SocketAsyncEventArgs" /> instances to preallocate. + </param> + </member> <member name="M:NetSharp.Sockets.Stream.StreamSocketClient.CanTransmissionArgsBeReused(System.Net.Sockets.SocketAsyncEventArgs@)"> <inheritdoc /> </member> @@ -617,6 +698,39 @@ <member name="M:NetSharp.Sockets.Stream.StreamSocketClient.ResetTransmissionArgs(System.Net.Sockets.SocketAsyncEventArgs)"> <inheritdoc /> </member> + <member name="T:NetSharp.Sockets.Stream.StreamSocketServerOptions"> + <summary> + Provides additional configuration options for a <see cref="T:NetSharp.Sockets.Stream.StreamSocketServer" /> instance. + </summary> + </member> + <member name="F:NetSharp.Sockets.Stream.StreamSocketServerOptions.Defaults"> + <summary> + The default configuration. + </summary> + </member> + <member name="F:NetSharp.Sockets.Stream.StreamSocketServerOptions.ConcurrentAcceptCalls"> + <summary> + The number of <see cref="M:System.Net.Sockets.Socket.AcceptAsync(System.Net.Sockets.SocketAsyncEventArgs)" /> calls that will be 'in-flight' at any one time, and ready to service incoming client + connection requests. This should be set according to the number of client which will be attempting to connect at once. + </summary> + </member> + <member name="F:NetSharp.Sockets.Stream.StreamSocketServerOptions.PreallocatedTransmissionArgs"> + <summary> + The number of <see cref="T:System.Net.Sockets.SocketAsyncEventArgs" /> instances that should be preallocated for use in the <see cref="M:System.Net.Sockets.Socket.SendAsync(System.Net.Sockets.SocketAsyncEventArgs)" /> and + <see cref="M:System.Net.Sockets.Socket.ReceiveAsync(System.Net.Sockets.SocketAsyncEventArgs)" /> methods. + </summary> + </member> + <member name="M:NetSharp.Sockets.Stream.StreamSocketServerOptions.#ctor(System.Int32,System.UInt16)"> + <summary> + Constructs a new instance of the <see cref="T:NetSharp.Sockets.Stream.StreamSocketServerOptions" /> struct. + </summary> + <param name="concurrentAcceptCalls"> + The number of <see cref="M:System.Net.Sockets.Socket.AcceptAsync(System.Net.Sockets.SocketAsyncEventArgs)" /> calls which should be 'in-flight' at any one time. + </param> + <param name="preallocatedTransmissionArgs"> + The number of <see cref="T:System.Net.Sockets.SocketAsyncEventArgs" /> instances to preallocate. + </param> + </member> <member name="M:NetSharp.Sockets.Stream.StreamSocketServer.#ctor(System.Net.Sockets.AddressFamily@,System.Net.Sockets.ProtocolType@,NetSharp.Sockets.SocketServerPacketHandler@,System.Nullable{NetSharp.Sockets.Stream.StreamSocketServerOptions}@)"> <summary> Constructs a new instance of the <see cref="T:NetSharp.Sockets.Stream.StreamSocketServer" /> class. @@ -1003,6 +1117,11 @@ Represents the result of a socket transmission. </summary> </member> + <member name="F:NetSharp.Utils.TransmissionResult.Timeout"> + <summary> + Represents an asynchronous transmission which timed out. + </summary> + </member> <member name="F:NetSharp.Utils.TransmissionResult.Buffer"> <summary> The byte buffer that was transmitted across the network. @@ -1040,5 +1159,13 @@ The remote end point associated with the transmission. </param> </member> + <member name="M:NetSharp.Utils.TransmissionResult.TimedOut"> + <summary> + Checks whether this instance represents a timed out transmission. + </summary> + <returns> + Whether this instance has timed out. + </returns> + </member> </members> </doc> diff --git a/NetSharp/NetSharp/Sockets/Datagram/DatagramSocketClient.cs b/NetSharp/NetSharp/Sockets/Datagram/DatagramSocketClient.cs @@ -2,6 +2,7 @@ using NetSharp.Utils; using System; +using System.Diagnostics; using System.Net; using System.Net.Sockets; using System.Threading; @@ -9,14 +10,29 @@ using System.Threading.Tasks; namespace NetSharp.Sockets.Datagram { - //TODO document + /// <summary> + /// Provides additional configuration options for a <see cref="DatagramSocketClient" /> instance. + /// </summary> public readonly struct DatagramSocketClientOptions { + /// <summary> + /// The default configuration. + /// </summary> public static readonly DatagramSocketClientOptions Defaults = new DatagramSocketClientOptions(0); + /// <summary> + /// The number of <see cref="SocketAsyncEventArgs" /> instances that should be preallocated for use in the + /// <see cref="DatagramSocketClient.SendToAsync" /> and <see cref="DatagramSocketClient.ReceiveFromAsync" /> methods. + /// </summary> public readonly ushort PreallocatedTransmissionArgs; + /// <summary> + /// Constructs a new instance of the <see cref="DatagramSocketClientOptions" /> struct. + /// </summary> + /// <param name="preallocatedTransmissionArgs"> + /// The number of <see cref="SocketAsyncEventArgs" /> instances to preallocate. + /// </param> public DatagramSocketClientOptions(ushort preallocatedTransmissionArgs) { PreallocatedTransmissionArgs = preallocatedTransmissionArgs; @@ -49,13 +65,27 @@ namespace NetSharp.Sockets.Datagram { connectToken.CompletionSource.SetCanceled(); } - else if (args.SocketError == SocketError.Success) - { - connectToken.CompletionSource.SetResult(true); - } else { - connectToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + switch (args.SocketError) + { + case SocketError.Success: + connectToken.CompletionSource.SetResult(true); + + break; + + case SocketError.OperationAborted: + Debug.WriteLine("CompleteConnect experienced SocketError.OperationAborted!"); + + connectToken.CompletionSource.SetResult(false); + + break; + + default: + connectToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + + break; + } } TransmissionArgsPool.Return(args); @@ -65,19 +95,26 @@ namespace NetSharp.Sockets.Datagram { AsyncTransmissionToken receiveToken = (AsyncTransmissionToken)args.UserToken; - if (receiveToken.CancellationToken.IsCancellationRequested) - { - receiveToken.CompletionSource.SetCanceled(); - } - else if (args.SocketError == SocketError.Success) + switch (args.SocketError) { - TransmissionResult result = new TransmissionResult(in args); + case SocketError.Success: + TransmissionResult result = new TransmissionResult(in args); - receiveToken.CompletionSource.SetResult(result); - } - else - { - receiveToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + receiveToken.CompletionSource.SetResult(result); + + break; + + case SocketError.OperationAborted: + Debug.WriteLine("CompleteReceiveFrom experienced SocketError.OperationAborted!"); + + receiveToken.CompletionSource.SetResult(TransmissionResult.Timeout); + + break; + + default: + receiveToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + + break; } TransmissionArgsPool.Return(args); @@ -91,15 +128,29 @@ namespace NetSharp.Sockets.Datagram { sendToken.CompletionSource.SetCanceled(); } - else if (args.SocketError == SocketError.Success) - { - TransmissionResult result = new TransmissionResult(in args); - - sendToken.CompletionSource.SetResult(result); - } else { - sendToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + switch (args.SocketError) + { + case SocketError.Success: + TransmissionResult result = new TransmissionResult(in args); + + sendToken.CompletionSource.SetResult(result); + + break; + + case SocketError.OperationAborted: + Debug.WriteLine("CompleteSendTo experienced SocketError.OperationAborted!"); + + sendToken.CompletionSource.SetResult(TransmissionResult.Timeout); + + break; + + default: + sendToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + + break; + } } TransmissionArgsPool.Return(args); @@ -155,6 +206,12 @@ namespace NetSharp.Sockets.Datagram } /// <inheritdoc /> + protected override void ResetSocketOnAsyncCancellationEx() + { + throw new NotImplementedException(); + } + + /// <inheritdoc /> protected override void ResetTransmissionArgs(SocketAsyncEventArgs args) { } @@ -179,12 +236,9 @@ namespace NetSharp.Sockets.Datagram args.SocketFlags = flags; args.UserToken = new AsyncTransmissionToken(in tcs, in cancellationToken); - // TODO implement cancellation for client sockets - cancellationToken.Register(token => - { - AsyncTransmissionCancellationToken transmissionCancellationToken = - (AsyncTransmissionCancellationToken)token; - }, new AsyncTransmissionCancellationToken(in Connection, in args, in TransmissionArgsPool, in tcs)); + // TODO implement cancellation for client socket receiveFromAsync + cancellationToken.Register(CancelAsyncTransmission, + new AsyncTransmissionCancellationToken(in Connection, in args, in TransmissionArgsPool, in tcs)); if (Connection.ReceiveFromAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task); @@ -215,12 +269,9 @@ namespace NetSharp.Sockets.Datagram args.SocketFlags = flags; args.UserToken = new AsyncTransmissionToken(in tcs, in cancellationToken); - // TODO implement cancellation for client sockets - cancellationToken.Register(token => - { - AsyncTransmissionCancellationToken transmissionCancellationToken = - (AsyncTransmissionCancellationToken)token; - }, new AsyncTransmissionCancellationToken(in Connection, in args, in TransmissionArgsPool, in tcs)); + // TODO implement cancellation for client socket sendToAsync + cancellationToken.Register(CancelAsyncTransmission, + new AsyncTransmissionCancellationToken(in Connection, in args, in TransmissionArgsPool, in tcs)); if (Connection.SendToAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task); diff --git a/NetSharp/NetSharp/Sockets/Datagram/DatagramSocketServer.cs b/NetSharp/NetSharp/Sockets/Datagram/DatagramSocketServer.cs @@ -8,15 +8,38 @@ using System.Threading.Tasks; namespace NetSharp.Sockets.Datagram { - //TODO document + /// <summary> + /// Provides additional configuration options for a <see cref="DatagramSocketServer" /> instance. + /// </summary> public readonly struct DatagramSocketServerOptions { + /// <summary> + /// The default configuration. + /// </summary> public static readonly DatagramSocketServerOptions Defaults = new DatagramSocketServerOptions(Environment.ProcessorCount, 0); + /// <summary> + /// The number of <see cref="Socket.ReceiveFromAsync" /> calls that will be 'in-flight' at any one time, and ready to service incoming client + /// packets. This should be set to the number of client which will be connected at once. + /// </summary> public readonly int ConcurrentReceiveFromCalls; + + /// <summary> + /// The number of <see cref="SocketAsyncEventArgs" /> instances that should be preallocated for use in the <see cref="Socket.SendToAsync" /> + /// and <see cref="Socket.ReceiveFromAsync" /> methods. + /// </summary> public readonly ushort PreallocatedTransmissionArgs; + /// <summary> + /// Constructs a new instance of the <see cref="DatagramSocketServerOptions" /> struct. + /// </summary> + /// <param name="concurrentReceiveFromCalls"> + /// The number of <see cref="Socket.ReceiveFromAsync" /> calls which should be 'in-flight' at any one time. + /// </param> + /// <param name="preallocatedTransmissionArgs"> + /// The number of <see cref="SocketAsyncEventArgs" /> instances to preallocate. + /// </param> public DatagramSocketServerOptions(int concurrentReceiveFromCalls, ushort preallocatedTransmissionArgs) { ConcurrentReceiveFromCalls = concurrentReceiveFromCalls; @@ -204,7 +227,7 @@ namespace NetSharp.Sockets.Datagram { serverShutdownToken = cancellationToken; - for (int i = 0; i < ServerOptions.ConcurrentReceiveFromCalls; i++) + for (int i = 0; i < serverOptions.ConcurrentReceiveFromCalls; i++) { SocketAsyncEventArgs newReceiveArgs = TransmissionArgsPool.Rent(); newReceiveArgs.RemoteEndPoint = AnyRemoteEndPoint; diff --git a/NetSharp/NetSharp/Sockets/SocketClient.cs b/NetSharp/NetSharp/Sockets/SocketClient.cs @@ -1,4 +1,6 @@ -using NetSharp.Utils; +using System; +using System.Diagnostics; +using NetSharp.Utils; using System.Net; using System.Net.Sockets; @@ -36,6 +38,41 @@ namespace NetSharp.Sockets { } + private void ResetSocketOnAsyncCancellation() + { + // TODO actually implement cancellation of Socket.XXXAsync methods + + // backup socket options + } + + protected void CancelAsyncOperation(object tokenObj) + { + AsyncOperationCancellationToken token = (AsyncOperationCancellationToken)tokenObj; + + Debug.WriteLine("Cancelling asynchronous operation!"); + + DestroyTransmissionArgs(token.TransmissionArgs); + + // TODO consider replacing the destroyed args + + token.CompletionSource.SetResult(false); + } + + protected void CancelAsyncTransmission(object tokenObj) + { + AsyncTransmissionCancellationToken token = (AsyncTransmissionCancellationToken)tokenObj; + + Debug.WriteLine("Cancelling asynchronous transmission!"); + + DestroyTransmissionArgs(token.TransmissionArgs); + + // TODO consider replacing the destroyed args + + token.CompletionSource.SetResult(TransmissionResult.Timeout); + } + + protected abstract void ResetSocketOnAsyncCancellationEx(); + /// <summary> /// Connects the client to the specified end point. If called on a <see cref="SocketType.Dgram" />-based client, this method configures the /// default remote host, and the client will ignore any packets not coming from this default host (i.e the given <paramref name="remoteEndPoint" />). @@ -70,13 +107,14 @@ namespace NetSharp.Sockets args.RemoteEndPoint = remoteEndPoint; args.UserToken = new AsyncOperationToken(in tcs, in cancellationToken); + //TODO implement cancellation for client socket connectAsync cancellationToken.Register(token => { AsyncOperationCancellationToken operationCancellationToken = (AsyncOperationCancellationToken)token; Socket.CancelConnectAsync(operationCancellationToken.TransmissionArgs); - operationCancellationToken.CompletionSource.SetCanceled(); + operationCancellationToken.CompletionSource.SetResult(false); operationCancellationToken.TransmissionArgsPool.Return(operationCancellationToken.TransmissionArgs); }, new AsyncOperationCancellationToken(in Connection, in args, in TransmissionArgsPool, in tcs)); diff --git a/NetSharp/NetSharp/Sockets/SocketConnection.cs b/NetSharp/NetSharp/Sockets/SocketConnection.cs @@ -10,7 +10,7 @@ namespace NetSharp.Sockets /// <summary> /// Abstract base class for clients and servers. /// </summary> - /// TODO implement better protections for accessing socket options + /// TODO implement access for socket options in conjunction to cancellable Socket.XXXAsync calls public abstract class SocketConnection : IDisposable { /// <summary> @@ -140,30 +140,6 @@ namespace NetSharp.Sockets /// </param> protected abstract void ResetTransmissionArgs(SocketAsyncEventArgs args); - /// <inheritdoc cref="Socket.SetSocketOption(SocketOptionLevel,SocketOptionName,bool)" /> - protected void SetSocketOption(SocketOptionLevel optionLevel, SocketOptionName optionName, bool optionValue) - { - Connection.SetSocketOption(optionLevel, optionName, optionValue); - } - - /// <inheritdoc cref="Socket.SetSocketOption(SocketOptionLevel,SocketOptionName,byte[])" /> - protected void SetSocketOption(SocketOptionLevel optionLevel, SocketOptionName optionName, byte[] optionValue) - { - Connection.SetSocketOption(optionLevel, optionName, optionValue); - } - - /// <inheritdoc cref="Socket.SetSocketOption(SocketOptionLevel,SocketOptionName,int)" /> - protected void SetSocketOption(SocketOptionLevel optionLevel, SocketOptionName optionName, int optionValue) - { - Connection.SetSocketOption(optionLevel, optionName, optionValue); - } - - /// <inheritdoc cref="Socket.SetSocketOption(SocketOptionLevel,SocketOptionName,object)" /> - protected void SetSocketOption(SocketOptionLevel optionLevel, SocketOptionName optionName, object optionValue) - { - Connection.SetSocketOption(optionLevel, optionName, optionValue); - } - /// <summary> /// Binds the underlying socket. /// </summary> @@ -182,24 +158,6 @@ namespace NetSharp.Sockets GC.SuppressFinalize(this); } - /// <inheritdoc cref="Socket.GetSocketOption(SocketOptionLevel,SocketOptionName)" /> - public object GetSocketOption(SocketOptionLevel optionLevel, SocketOptionName optionName) - { - return Connection.GetSocketOption(optionLevel, optionName); - } - - /// <inheritdoc cref="Socket.GetSocketOption(SocketOptionLevel,SocketOptionName,int)" /> - public byte[] GetSocketOption(SocketOptionLevel optionLevel, SocketOptionName optionName, int optionLength) - { - return Connection.GetSocketOption(optionLevel, optionName, optionLength); - } - - /// <inheritdoc cref="Socket.GetSocketOption(SocketOptionLevel,SocketOptionName,byte[])" /> - public void GetSocketOption(SocketOptionLevel optionLevel, SocketOptionName optionName, byte[] optionValue) - { - Connection.GetSocketOption(optionLevel, optionName, optionValue); - } - /// <summary> /// Shuts down the underlying socket. /// </summary> diff --git a/NetSharp/NetSharp/Sockets/Stream/StreamSocketClient.cs b/NetSharp/NetSharp/Sockets/Stream/StreamSocketClient.cs @@ -2,20 +2,36 @@ using NetSharp.Utils; using System; +using System.Diagnostics; using System.Net.Sockets; using System.Threading; using System.Threading.Tasks; namespace NetSharp.Sockets.Stream { - //TODO document + /// <summary> + /// Provides additional configuration options for a <see cref="StreamSocketClient" /> instance. + /// </summary> public readonly struct StreamSocketClientOptions { + /// <summary> + /// The default configuration. + /// </summary> public static readonly StreamSocketClientOptions Defaults = new StreamSocketClientOptions(0); + /// <summary> + /// The number of <see cref="SocketAsyncEventArgs" /> instances that should be preallocated for use in the + /// <see cref="StreamSocketClient.SendAsync" /> and <see cref="StreamSocketClient.ReceiveAsync" /> methods. + /// </summary> public readonly ushort PreallocatedTransmissionArgs; + /// <summary> + /// Constructs a new instance of the <see cref="StreamSocketClientOptions" /> struct. + /// </summary> + /// <param name="preallocatedTransmissionArgs"> + /// The number of <see cref="SocketAsyncEventArgs" /> instances to preallocate. + /// </param> public StreamSocketClientOptions(ushort preallocatedTransmissionArgs) { PreallocatedTransmissionArgs = preallocatedTransmissionArgs; @@ -44,19 +60,24 @@ namespace NetSharp.Sockets.Stream { AsyncOperationToken connectToken = (AsyncOperationToken)args.UserToken; - if (connectToken.CancellationToken.IsCancellationRequested) + switch (args.SocketError) { - Connection.Disconnect(true); + case SocketError.Success: + connectToken.CompletionSource.SetResult(true); - connectToken.CompletionSource.SetCanceled(); - } - else if (args.SocketError == SocketError.Success) - { - connectToken.CompletionSource.SetResult(true); - } - else - { - connectToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + break; + + case SocketError.OperationAborted: + Debug.WriteLine("CompleteConnect experienced SocketError.OperationAborted!"); + + connectToken.CompletionSource.SetResult(false); + + break; + + default: + connectToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + + break; } TransmissionArgsPool.Return(args); @@ -66,17 +87,24 @@ namespace NetSharp.Sockets.Stream { AsyncOperationToken disconnectToken = (AsyncOperationToken)args.UserToken; - if (disconnectToken.CancellationToken.IsCancellationRequested) + switch (args.SocketError) { - disconnectToken.CompletionSource.SetCanceled(); - } - else if (args.SocketError == SocketError.Success) - { - disconnectToken.CompletionSource.SetResult(true); - } - else - { - disconnectToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + case SocketError.Success: + disconnectToken.CompletionSource.SetResult(true); + + break; + + case SocketError.OperationAborted: + Debug.WriteLine("CompleteDisconnect experienced SocketError.OperationAborted!"); + + disconnectToken.CompletionSource.SetResult(false); + + break; + + default: + disconnectToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + + break; } TransmissionArgsPool.Return(args); @@ -86,51 +114,58 @@ namespace NetSharp.Sockets.Stream { AsyncTransmissionToken receiveToken = (AsyncTransmissionToken)args.UserToken; - if (receiveToken.CancellationToken.IsCancellationRequested) + switch (args.SocketError) { - receiveToken.CompletionSource.SetCanceled(); + case SocketError.Success: + Memory<byte> transmissionBuffer = args.MemoryBuffer; + int expectedBytes = transmissionBuffer.Length; - TransmissionArgsPool.Return(args); - } - else if (args.SocketError == SocketError.Success) - { - Memory<byte> transmissionBuffer = args.MemoryBuffer; - int expectedBytes = transmissionBuffer.Length; + if (args.BytesTransferred == expectedBytes) + { + // buffer was fully received - if (args.BytesTransferred == expectedBytes) - { - // buffer was fully received + TransmissionResult result = new TransmissionResult(in args); - TransmissionResult result = new TransmissionResult(in args); + receiveToken.CompletionSource.SetResult(result); - receiveToken.CompletionSource.SetResult(result); + TransmissionArgsPool.Return(args); + } + else if (expectedBytes > args.BytesTransferred && args.BytesTransferred > 0) + { + // receive the remaining parts of the buffer - TransmissionArgsPool.Return(args); - } - else if (expectedBytes > args.BytesTransferred && args.BytesTransferred > 0) - { - // receive the remaining parts of the buffer + int receivedBytes = args.BytesTransferred; - int receivedBytes = args.BytesTransferred; + args.SetBuffer(receivedBytes, expectedBytes - receivedBytes); - args.SetBuffer(receivedBytes, expectedBytes - receivedBytes); + Connection.ReceiveAsync(args); + } + else + { + // no bytes were received, remote socket is dead - Connection.ReceiveAsync(args); - } - else - { - // no bytes were received, remote socket is dead + receiveToken.CompletionSource.SetException(new SocketException((int)SocketError.HostDown)); - receiveToken.CompletionSource.SetException(new SocketException((int)SocketError.HostDown)); + TransmissionArgsPool.Return(args); + } + + break; + + case SocketError.OperationAborted: + Debug.WriteLine("CompleteReceive experienced SocketError.OperationAborted!"); + + receiveToken.CompletionSource.SetResult(TransmissionResult.Timeout); TransmissionArgsPool.Return(args); - } - } - else - { - receiveToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); - TransmissionArgsPool.Return(args); + break; + + default: + receiveToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + + TransmissionArgsPool.Return(args); + + break; } } @@ -138,51 +173,58 @@ namespace NetSharp.Sockets.Stream { AsyncTransmissionToken sendToken = (AsyncTransmissionToken)args.UserToken; - if (sendToken.CancellationToken.IsCancellationRequested) + switch (args.SocketError) { - sendToken.CompletionSource.SetCanceled(); + case SocketError.Success: + Memory<byte> transmissionBuffer = args.MemoryBuffer; + int remainingBytes = transmissionBuffer.Length; - TransmissionArgsPool.Return(args); - } - else if (args.SocketError == SocketError.Success) - { - Memory<byte> transmissionBuffer = args.MemoryBuffer; - int remainingBytes = transmissionBuffer.Length; + if (args.BytesTransferred == remainingBytes) + { + // buffer was fully sent - if (args.BytesTransferred == remainingBytes) - { - // buffer was fully sent + TransmissionResult result = new TransmissionResult(in args); - TransmissionResult result = new TransmissionResult(in args); + sendToken.CompletionSource.SetResult(result); - sendToken.CompletionSource.SetResult(result); + TransmissionArgsPool.Return(args); + } + else if (remainingBytes > args.BytesTransferred && args.BytesTransferred > 0) + { + // send the remaining parts of the buffer - TransmissionArgsPool.Return(args); - } - else if (remainingBytes > args.BytesTransferred && args.BytesTransferred > 0) - { - // send the remaining parts of the buffer + int sentBytes = args.BytesTransferred; + + args.SetBuffer(sentBytes, remainingBytes - sentBytes); + + Connection.SendAsync(args); + } + else + { + // no bytes were sent, remote socket is dead - int sentBytes = args.BytesTransferred; + sendToken.CompletionSource.SetException(new SocketException((int)SocketError.HostDown)); - args.SetBuffer(sentBytes, remainingBytes - sentBytes); + TransmissionArgsPool.Return(args); + } + + break; - Connection.SendAsync(args); - } - else - { - // no bytes were sent, remote socket is dead + case SocketError.OperationAborted: + Debug.WriteLine("CompleteSend experienced SocketError.OperationAborted!"); - sendToken.CompletionSource.SetException(new SocketException((int)SocketError.HostDown)); + sendToken.CompletionSource.SetResult(TransmissionResult.Timeout); TransmissionArgsPool.Return(args); - } - } - else - { - sendToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); - TransmissionArgsPool.Return(args); + break; + + default: + sendToken.CompletionSource.SetException(new SocketException((int)args.SocketError)); + + TransmissionArgsPool.Return(args); + + break; } } @@ -241,6 +283,12 @@ namespace NetSharp.Sockets.Stream } /// <inheritdoc /> + protected override void ResetSocketOnAsyncCancellationEx() + { + throw new NotImplementedException(); + } + + /// <inheritdoc /> protected override void ResetTransmissionArgs(SocketAsyncEventArgs args) { } @@ -259,6 +307,18 @@ namespace NetSharp.Sockets.Stream args.DisconnectReuseSocket = allowSocketReuse; args.UserToken = new AsyncOperationToken(in tcs, in cancellationToken); + //TODO implement cancellation for client socket disconnectAsync + cancellationToken.Register(token => + { + AsyncOperationCancellationToken operationCancellationToken = (AsyncOperationCancellationToken)token; + + Socket.CancelConnectAsync(operationCancellationToken.TransmissionArgs); + + operationCancellationToken.CompletionSource.SetCanceled(); + + operationCancellationToken.TransmissionArgsPool.Return(operationCancellationToken.TransmissionArgs); + }, new AsyncOperationCancellationToken(in Connection, in args, in TransmissionArgsPool, in tcs)); + if (Connection.DisconnectAsync(args)) return new ValueTask(tcs.Task); TransmissionArgsPool.Return(args); @@ -291,6 +351,13 @@ namespace NetSharp.Sockets.Stream args.SocketFlags = flags; args.UserToken = new AsyncTransmissionToken(in tcs, in cancellationToken); + // TODO implement cancellation for client socket receiveAsync + cancellationToken.Register(token => + { + AsyncTransmissionCancellationToken transmissionCancellationToken = + (AsyncTransmissionCancellationToken)token; + }, new AsyncTransmissionCancellationToken(in Connection, in args, in TransmissionArgsPool, in tcs)); + if (Connection.ReceiveAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task); TransmissionResult result = new TransmissionResult(in args); @@ -325,6 +392,13 @@ namespace NetSharp.Sockets.Stream args.SocketFlags = flags; args.UserToken = new AsyncTransmissionToken(in tcs, in cancellationToken); + // TODO implement cancellation for client socket sendAsync + cancellationToken.Register(token => + { + AsyncTransmissionCancellationToken transmissionCancellationToken = + (AsyncTransmissionCancellationToken)token; + }, new AsyncTransmissionCancellationToken(in Connection, in args, in TransmissionArgsPool, in tcs)); + if (Connection.SendAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task); TransmissionResult result = new TransmissionResult(in args); diff --git a/NetSharp/NetSharp/Sockets/Stream/StreamSocketServer.cs b/NetSharp/NetSharp/Sockets/Stream/StreamSocketServer.cs @@ -7,15 +7,38 @@ using System.Threading.Tasks; namespace NetSharp.Sockets.Stream { - //TODO document + /// <summary> + /// Provides additional configuration options for a <see cref="StreamSocketServer" /> instance. + /// </summary> public readonly struct StreamSocketServerOptions { + /// <summary> + /// The default configuration. + /// </summary> public static readonly StreamSocketServerOptions Defaults = new StreamSocketServerOptions(Environment.ProcessorCount, 0); + /// <summary> + /// The number of <see cref="Socket.AcceptAsync" /> calls that will be 'in-flight' at any one time, and ready to service incoming client + /// connection requests. This should be set according to the number of client which will be attempting to connect at once. + /// </summary> public readonly int ConcurrentAcceptCalls; + + /// <summary> + /// The number of <see cref="SocketAsyncEventArgs" /> instances that should be preallocated for use in the <see cref="Socket.SendAsync" /> and + /// <see cref="Socket.ReceiveAsync" /> methods. + /// </summary> public readonly ushort PreallocatedTransmissionArgs; + /// <summary> + /// Constructs a new instance of the <see cref="StreamSocketServerOptions" /> struct. + /// </summary> + /// <param name="concurrentAcceptCalls"> + /// The number of <see cref="Socket.AcceptAsync" /> calls which should be 'in-flight' at any one time. + /// </param> + /// <param name="preallocatedTransmissionArgs"> + /// The number of <see cref="SocketAsyncEventArgs" /> instances to preallocate. + /// </param> public StreamSocketServerOptions(int concurrentAcceptCalls, ushort preallocatedTransmissionArgs) { ConcurrentAcceptCalls = concurrentAcceptCalls; diff --git a/NetSharp/NetSharp/Utils/TransmissionResult.cs b/NetSharp/NetSharp/Utils/TransmissionResult.cs @@ -1,6 +1,7 @@ using System; using System.Net; using System.Net.Sockets; +using System.Runtime.CompilerServices; namespace NetSharp.Utils { @@ -10,6 +11,11 @@ namespace NetSharp.Utils public readonly struct TransmissionResult { /// <summary> + /// Represents an asynchronous transmission which timed out. + /// </summary> + internal static readonly TransmissionResult Timeout = new TransmissionResult(); + + /// <summary> /// The byte buffer that was transmitted across the network. /// </summary> public readonly Memory<byte> Buffer; @@ -55,5 +61,14 @@ namespace NetSharp.Utils Count = count; RemoteEndPoint = remoteEndPoint; } + + /// <summary> + /// Checks whether this instance represents a timed out transmission. + /// </summary> + /// <returns> + /// Whether this instance has timed out. + /// </returns> + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public bool TimedOut() => Equals(Timeout); } } \ No newline at end of file diff --git a/NetSharp/NetSharpExamples/Examples/TcpSocketClientExample.cs b/NetSharp/NetSharpExamples/Examples/TcpSocketClientExample.cs @@ -33,7 +33,10 @@ namespace NetSharpExamples.Examples /* a cancellable asynchronous version also exists. client.ConnectAsync(in remoteEndPoint, CancellationToken.None); */ - while (true) + + Console.WriteLine("Starting TCP Socket Client!"); + + for (int i = 0; i < 10; i++) { string data = $"Hello World from {client.LocalEndPoint}!"; dataEncoding.GetBytes(data).CopyTo(sendBuffer, 0); @@ -63,7 +66,7 @@ namespace NetSharpExamples.Examples // lock is not necessary, but means that console output is clean and not interleaved lock (typeof(Console)) { - Console.WriteLine($"[Client] Received response with contents \'{dataEncoding.GetString(receiveBuffer)}\' from {remoteEndPoint}"); + Console.WriteLine($"[Client] Received response with contents \'{dataEncoding.GetString(receiveBuffer).TrimEnd('\0', ' ')}\' from {remoteEndPoint}"); } } } diff --git a/NetSharp/NetSharpExamples/Examples/TcpSocketServerExample.cs b/NetSharp/NetSharpExamples/Examples/TcpSocketServerExample.cs @@ -23,7 +23,7 @@ namespace NetSharpExamples.Examples // lock is not necessary, but means that console output is clean and not interleaved lock (typeof(Console)) { - Console.WriteLine($"[Server] Received request with contents \'{ServerEncoding.GetString(request.Data.Span)}\' from {remoteEndPoint}"); + Console.WriteLine($"[Server] Received request with contents \'{ServerEncoding.GetString(request.Data.Span).TrimEnd('\0', ' ')}\' from {remoteEndPoint}"); Console.WriteLine($"[Server] Echoing back request to {remoteEndPoint}"); } @@ -43,6 +43,8 @@ namespace NetSharpExamples.Examples server.Bind(in ServerEndPoint); + Console.WriteLine("Starting TCP Socket Server!"); + return server.RunAsync(CancellationToken.None); // we run forever. alternatively, pass in a cancellation token to ensure that the server terminates } } diff --git a/NetSharp/NetSharpExamples/Examples/UdpSocketClientCancellationExample.cs b/NetSharp/NetSharpExamples/Examples/UdpSocketClientCancellationExample.cs @@ -0,0 +1,93 @@ +using NetSharp.Packets; +using NetSharp.Sockets.Datagram; +using NetSharp.Utils; + +using System; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; + +namespace NetSharpExamples.Examples +{ + public class UdpSocketClientCancellationExample : INetSharpExample + { + private static readonly EndPoint ServerEndPoint = new IPEndPoint(IPAddress.Loopback, 12377); + + /// <inheritdoc /> + public string Name { get; } = "UDP Socket Client Cancellation Example"; + + /// <summary> + /// A read only server. Never sends out data! + /// </summary> + private Task ServerTask(CancellationToken cancellationToken) + { + Socket server = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); + + server.Bind(ServerEndPoint); + + byte[] transmissionBuffer = new byte[NetworkPacket.TotalSize]; + + EndPoint remoteEndPoint = new IPEndPoint(IPAddress.Any, 0); + + while (!cancellationToken.IsCancellationRequested) + { + server.ReceiveFrom(transmissionBuffer, ref remoteEndPoint); + } + + server.Close(); + + return Task.CompletedTask; + } + + /// <inheritdoc /> + public async Task RunAsync() + { + using CancellationTokenSource serverCts = new CancellationTokenSource(); + Task serverTask = Task.Factory.StartNew(state => ServerTask((CancellationToken)state), serverCts.Token, TaskCreationOptions.LongRunning); + + DatagramSocketClientOptions clientOptions = new DatagramSocketClientOptions((ushort)2); + using DatagramSocketClient client = new DatagramSocketClient(AddressFamily.InterNetwork, ProtocolType.Udp, clientOptions); + + byte[] sendBuffer = new byte[NetworkPacket.TotalSize]; + byte[] receiveBuffer = new byte[NetworkPacket.TotalSize]; + + EndPoint remoteEndPoint = ServerEndPoint; + + TimeSpan timeout = TimeSpan.FromMilliseconds(500); + + Console.WriteLine("Starting UDP Socket Client!"); + + for (int i = 0; i < 10; i++) + { + CancellationTokenSource sendCts = new CancellationTokenSource(); + + sendCts.CancelAfter(timeout); + TransmissionResult sendResult = await client.SendToAsync(remoteEndPoint, sendBuffer, SocketFlags.None, sendCts.Token); + + if (sendResult.TimedOut()) + { + Console.WriteLine("Send timed out!"); + + continue; + } + + Console.WriteLine($"Sent {sendResult.Count} bytes of data!"); + + CancellationTokenSource receiveCts = new CancellationTokenSource(); + + receiveCts.CancelAfter(timeout); + TransmissionResult receiveResult = await client.ReceiveFromAsync(remoteEndPoint, receiveBuffer, SocketFlags.None, receiveCts.Token); + + if (receiveResult.TimedOut()) + { + Console.WriteLine("Receive timed out!"); + + continue; + } + + Console.WriteLine($"Received {receiveResult.Count} bytes of data!"); + } + } + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharpExamples/Examples/UdpSocketClientExample.cs b/NetSharp/NetSharpExamples/Examples/UdpSocketClientExample.cs @@ -28,7 +28,9 @@ namespace NetSharpExamples.Examples EndPoint remoteEndPoint = UdpSocketServerExample.ServerEndPoint; - while (true) + Console.WriteLine("Starting UDP Socket Client!"); + + for (int i = 0; i < 10; i++) { string data = $"Hello World from {client.LocalEndPoint}!"; dataEncoding.GetBytes(data).CopyTo(sendBuffer, 0); @@ -58,7 +60,7 @@ namespace NetSharpExamples.Examples // lock is not necessary, but means that console output is clean and not interleaved lock (typeof(Console)) { - Console.WriteLine($"[Client] Received response with contents \'{dataEncoding.GetString(receiveBuffer)}\' from {remoteEndPoint}"); + Console.WriteLine($"[Client] Received response with contents \'{dataEncoding.GetString(receiveBuffer).TrimEnd('\0', ' ')}\' from {remoteEndPoint}"); } } } diff --git a/NetSharp/NetSharpExamples/Examples/UdpSocketServerExample.cs b/NetSharp/NetSharpExamples/Examples/UdpSocketServerExample.cs @@ -23,7 +23,7 @@ namespace NetSharpExamples.Examples // lock is not necessary, but means that console output is clean and not interleaved lock (typeof(Console)) { - Console.WriteLine($"[Server] Received request with contents \'{ServerEncoding.GetString(request.Data.Span)}\' from {remoteEndPoint}"); + Console.WriteLine($"[Server] Received request with contents \'{ServerEncoding.GetString(request.Data.Span).TrimEnd('\0', ' ')}\' from {remoteEndPoint}"); Console.WriteLine($"[Server] Echoing back request to {remoteEndPoint}"); } @@ -43,6 +43,8 @@ namespace NetSharpExamples.Examples server.Bind(in ServerEndPoint); + Console.WriteLine("Starting UDP Socket Server!"); + return server.RunAsync(CancellationToken.None); // we run forever. alternatively, pass in a cancellation token to ensure that the server terminates } } diff --git a/NetSharp/NetSharpExamples/NetSharpExamples.xml b/NetSharp/NetSharpExamples/NetSharpExamples.xml @@ -82,6 +82,17 @@ <member name="M:NetSharpExamples.Examples.TcpSocketServerExample.RunAsync"> <inheritdoc /> </member> + <member name="P:NetSharpExamples.Examples.UdpSocketClientCancellationExample.Name"> + <inheritdoc /> + </member> + <member name="M:NetSharpExamples.Examples.UdpSocketClientCancellationExample.ServerTask(System.Threading.CancellationToken)"> + <summary> + A read only server. Never sends out data! + </summary> + </member> + <member name="M:NetSharpExamples.Examples.UdpSocketClientCancellationExample.RunAsync"> + <inheritdoc /> + </member> <member name="P:NetSharpExamples.Examples.UdpSocketClientExample.Name"> <inheritdoc /> </member> diff --git a/NetSharp/NetSharpExamples/Program.cs b/NetSharp/NetSharpExamples/Program.cs @@ -16,6 +16,7 @@ namespace NetSharpExamples new UdpSocketClientSyncBenchmark(), new UdpSocketClientAsyncBenchmark(), new UdpSocketClientExample(), + new UdpSocketClientCancellationExample(), // TCP socket server and client examples new TcpSocketServerBenchmark(), @@ -31,6 +32,8 @@ namespace NetSharpExamples while (true) { + GC.Collect(); + PickExample(); } }