commit a6c70d65bc41477b7fa578a519dfd1a530fec13d parent bd3d3a56ae0835746dd9089083ca26e61a14de62 Author: Mikolaj Lenczewski <mikolaj.lenczewski308@gmail.com> Date: Wed, 10 Jun 2020 13:59:43 +0100 Finished reimplementation of RawStreamNetworkWriter. Rewrite of stream network connections is now complete! Diffstat:
14 files changed, 359 insertions(+), 161 deletions(-)
diff --git a/NetSharp/NetSharp/AssemblyAttributes.cs b/NetSharp/NetSharp/AssemblyAttributes.cs @@ -0,0 +1,10 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("NetSharpExamples", AllInternalsVisible = true)] + +namespace NetSharp +{ + internal static class AssemblyAttributes + { + } +} +\ No newline at end of file diff --git a/NetSharp/NetSharp/Raw/Datagram/RawDatagramNetworkReader.cs b/NetSharp/NetSharp/Raw/Datagram/RawDatagramNetworkReader.cs @@ -17,17 +17,17 @@ namespace NetSharp.Raw.Datagram private readonly RawDatagramRequestHandler requestHandler; /// <inheritdoc /> - public RawDatagramNetworkReader(ref Socket rawConnection, RawDatagramRequestHandler? requestHandler, EndPoint defaultEndPoint, int messageSize, - int pooledBuffersPerBucket = 50, uint preallocatedStateObjects = 0) : base(ref rawConnection, defaultEndPoint, messageSize, + public RawDatagramNetworkReader(ref Socket rawConnection, RawDatagramRequestHandler? requestHandler, EndPoint defaultEndPoint, int datagramSize, + int pooledBuffersPerBucket = 50, uint preallocatedStateObjects = 0) : base(ref rawConnection, defaultEndPoint, datagramSize, pooledBuffersPerBucket, preallocatedStateObjects) { - if (messageSize <= 0 || MaxDatagramSize < messageSize) + if (datagramSize <= 0 || MaxDatagramSize < datagramSize) { - throw new ArgumentOutOfRangeException(nameof(messageSize), messageSize, + throw new ArgumentOutOfRangeException(nameof(datagramSize), datagramSize, $"The datagram size must be greater than 0 and less than {MaxDatagramSize}"); } - this.messageSize = messageSize; + messageSize = datagramSize; this.requestHandler = requestHandler ?? DefaultRequestHandler; } @@ -62,18 +62,14 @@ namespace NetSharp.Raw.Datagram break; default: - BufferPool.Return(receiveBuffer, true); - ArgsPool.Return(args); + CleanupTransmissionBufferAndState(args); break; } } private void CompleteSendTo(SocketAsyncEventArgs args) { - byte[] sendBuffer = args.Buffer; - BufferPool.Return(sendBuffer, true); - - ArgsPool.Return(args); + CleanupTransmissionBufferAndState(args); } [MethodImpl(MethodImplOptions.AggressiveInlining)] @@ -120,11 +116,7 @@ namespace NetSharp.Raw.Datagram { if (ShutdownToken.IsCancellationRequested) { - byte[] sendBuffer = args.Buffer; - BufferPool.Return(sendBuffer, true); - - ArgsPool.Return(args); - + CleanupTransmissionBufferAndState(args); return; } diff --git a/NetSharp/NetSharp/Raw/Datagram/RawDatagramNetworkWriter.cs b/NetSharp/NetSharp/Raw/Datagram/RawDatagramNetworkWriter.cs @@ -26,7 +26,7 @@ namespace NetSharp.Raw.Datagram private void CompleteConnect(SocketAsyncEventArgs args) { - AsyncOperationToken token = (AsyncOperationToken) args.UserToken; + OperationToken token = (OperationToken) args.UserToken; switch (args.SocketError) { @@ -39,8 +39,7 @@ namespace NetSharp.Raw.Datagram break; default: - int errorCode = (int) args.SocketError; - token.CompletionSource.SetException(new SocketException(errorCode)); + token.CompletionSource.SetException(new SocketException((int) args.SocketError)); break; } @@ -49,7 +48,7 @@ namespace NetSharp.Raw.Datagram private void CompleteReceiveFrom(SocketAsyncEventArgs args) { - AsyncDatagramReadToken token = (AsyncDatagramReadToken) args.UserToken; + PacketReadToken token = (PacketReadToken) args.UserToken; byte[] receiveBuffer = args.Buffer; @@ -65,8 +64,7 @@ namespace NetSharp.Raw.Datagram break; default: - int errorCode = (int) args.SocketError; - token.CompletionSource.SetException(new SocketException(errorCode)); + token.CompletionSource.SetException(new SocketException((int) args.SocketError)); break; } @@ -76,7 +74,7 @@ namespace NetSharp.Raw.Datagram private void CompleteSendTo(SocketAsyncEventArgs args) { - AsyncDatagramWriteToken token = (AsyncDatagramWriteToken) args.UserToken; + PacketWriteToken token = (PacketWriteToken) args.UserToken; byte[] sendBuffer = args.Buffer; @@ -91,8 +89,7 @@ namespace NetSharp.Raw.Datagram break; default: - int errorCode = (int) args.SocketError; - token.CompletionSource.SetException(new SocketException(errorCode)); + token.CompletionSource.SetException(new SocketException((int) args.SocketError)); break; } @@ -159,7 +156,7 @@ namespace NetSharp.Raw.Datagram args.RemoteEndPoint = remoteEndPoint; - AsyncOperationToken token = new AsyncOperationToken(tcs); + OperationToken token = new OperationToken(tcs); args.UserToken = token; if (Connection.ConnectAsync(args)) @@ -216,7 +213,7 @@ namespace NetSharp.Raw.Datagram args.RemoteEndPoint = remoteEndPoint; args.SocketFlags = flags; - AsyncDatagramReadToken token = new AsyncDatagramReadToken(tcs, in readBuffer); + PacketReadToken token = new PacketReadToken(tcs, in readBuffer); args.UserToken = token; if (Connection.ReceiveFromAsync(args)) @@ -229,8 +226,7 @@ namespace NetSharp.Raw.Datagram transmissionBuffer.CopyTo(readBuffer); - BufferPool.Return(transmissionBuffer, true); - ArgsPool.Return(args); + CleanupTransmissionBufferAndState(args); // transmissionBuffer was assigned to args.Buffer earlier, so this call is safe return new ValueTask<int>(result); } @@ -281,7 +277,7 @@ namespace NetSharp.Raw.Datagram args.RemoteEndPoint = remoteEndPoint; args.SocketFlags = flags; - AsyncDatagramWriteToken token = new AsyncDatagramWriteToken(tcs); + PacketWriteToken token = new PacketWriteToken(tcs); args.UserToken = token; if (Connection.SendToAsync(args)) @@ -292,18 +288,17 @@ namespace NetSharp.Raw.Datagram // inlining CompleteSendTo(SocketAsyncEventArgs) for performance int result = args.BytesTransferred; - BufferPool.Return(transmissionBuffer, true); - ArgsPool.Return(args); + CleanupTransmissionBufferAndState(args); // transmissionBuffer was assigned to args.Buffer earlier, so this call is safe return new ValueTask<int>(result); } - private readonly struct AsyncDatagramReadToken + private readonly struct PacketReadToken { public readonly TaskCompletionSource<int> CompletionSource; public readonly Memory<byte> UserBuffer; - public AsyncDatagramReadToken(TaskCompletionSource<int> completionSource, in Memory<byte> userBuffer) + public PacketReadToken(TaskCompletionSource<int> completionSource, in Memory<byte> userBuffer) { CompletionSource = completionSource; @@ -311,11 +306,11 @@ namespace NetSharp.Raw.Datagram } } - private readonly struct AsyncDatagramWriteToken + private readonly struct PacketWriteToken { public readonly TaskCompletionSource<int> CompletionSource; - public AsyncDatagramWriteToken(TaskCompletionSource<int> completionSource) + public PacketWriteToken(TaskCompletionSource<int> completionSource) { CompletionSource = completionSource; } diff --git a/NetSharp/NetSharp/Raw/RawNetworkConnectionBase.cs b/NetSharp/NetSharp/Raw/RawNetworkConnectionBase.cs @@ -2,6 +2,7 @@ using System.Buffers; using System.Net; using System.Net.Sockets; +using System.Runtime.CompilerServices; using NetSharp.Utils; @@ -40,6 +41,13 @@ namespace NetSharp.Raw protected abstract bool CanReuseStateObject(ref SocketAsyncEventArgs instance); + [MethodImpl(MethodImplOptions.AggressiveInlining)] + protected void CleanupTransmissionBufferAndState(SocketAsyncEventArgs args) + { + BufferPool.Return(args.Buffer, true); + ArgsPool.Return(args); + } + protected abstract SocketAsyncEventArgs CreateStateObject(); protected abstract void DestroyStateObject(SocketAsyncEventArgs instance); diff --git a/NetSharp/NetSharp/Raw/RawNetworkWriterBase.cs b/NetSharp/NetSharp/Raw/RawNetworkWriterBase.cs @@ -35,11 +35,11 @@ namespace NetSharp.Raw public abstract ValueTask<int> WriteAsync(EndPoint remoteEndPoint, ReadOnlyMemory<byte> writeBuffer, SocketFlags flags = SocketFlags.None); - protected readonly struct AsyncOperationToken + protected readonly struct OperationToken { public readonly TaskCompletionSource<bool> CompletionSource; - public AsyncOperationToken(TaskCompletionSource<bool> completionSource) + public OperationToken(TaskCompletionSource<bool> completionSource) { CompletionSource = completionSource; } diff --git a/NetSharp/NetSharp/Raw/Stream/RawStreamNetworkReader.cs b/NetSharp/NetSharp/Raw/Stream/RawStreamNetworkReader.cs @@ -12,13 +12,13 @@ namespace NetSharp.Raw.Stream private readonly RawStreamRequestHandler RequestHandler; /// <inheritdoc /> - public RawStreamNetworkReader(ref Socket rawConnection, RawStreamRequestHandler? requestHandler, EndPoint defaultEndPoint, int maxMessageSize, - int pooledBuffersPerBucket = 50, uint preallocatedStateObjects = 0) : base(ref rawConnection, defaultEndPoint, maxMessageSize, + public RawStreamNetworkReader(ref Socket rawConnection, RawStreamRequestHandler? requestHandler, EndPoint defaultEndPoint, int maxPooledMessageSize, + int pooledBuffersPerBucket = 50, uint preallocatedStateObjects = 0) : base(ref rawConnection, defaultEndPoint, maxPooledMessageSize, pooledBuffersPerBucket, preallocatedStateObjects) { - if (maxMessageSize <= 0) + if (maxPooledMessageSize <= 0) { - throw new ArgumentOutOfRangeException(nameof(maxMessageSize), maxMessageSize, + throw new ArgumentOutOfRangeException(nameof(maxPooledMessageSize), maxPooledMessageSize, $"The maximum message size must be greater than 0"); } @@ -54,7 +54,7 @@ namespace NetSharp.Raw.Stream // usually we will usually be returning the ResponseDataBuffer). args.SetBuffer(Array.Empty<byte>(), 0, 0); - ConfigureReceiveHeader(args); + ConfigureAsyncReceiveHeader(args); StartReceive(args); break; @@ -66,9 +66,9 @@ namespace NetSharp.Raw.Stream private void CompleteReceive(SocketAsyncEventArgs args) { - RequestReadToken readToken = (RequestReadToken) args.UserToken; + PacketReadToken readToken = (PacketReadToken) args.UserToken; - void CompleteReceiveHeader(SocketAsyncEventArgs args, in RequestReadToken readToken) + void CompleteReceiveHeader(SocketAsyncEventArgs args, in PacketReadToken readToken) { Memory<byte> headerBuffer = args.Buffer; @@ -81,7 +81,7 @@ namespace NetSharp.Raw.Stream { RawStreamPacketHeader header = RawStreamPacketHeader.Deserialise(in headerBuffer); - ConfigureReceiveData(args, in header); + ConfigureAsyncReceiveData(args, in header); StartReceive(args); } else if (0 < totalReceivedBytes && totalReceivedBytes < expectedBytes) // transmission not complete @@ -96,7 +96,7 @@ namespace NetSharp.Raw.Stream } } - void CompleteReceiveData(SocketAsyncEventArgs args, in RequestReadToken readToken) + void CompleteReceiveData(SocketAsyncEventArgs args, in PacketReadToken readToken) { Memory<byte> dataBuffer = args.Buffer; @@ -109,13 +109,13 @@ namespace NetSharp.Raw.Stream { EndPoint clientEndPoint = args.AcceptSocket.RemoteEndPoint; - // TODO use user-supplied delegate to get response packet size - int responseBufferDataSize = expectedBytes; - int responseBufferSize = RawStreamPacket.TotalPacketSize(responseBufferDataSize); + // TODO use user-supplied delegate to generate response packet header + RawStreamPacketHeader responseHeader = new RawStreamPacketHeader(expectedBytes); + int responseBufferSize = RawStreamPacket.TotalPacketSize(in responseHeader); byte[] responseBuffer = BufferPool.Rent(responseBufferSize); - Memory<byte> responseBufferMemory = responseBuffer.AsMemory(RawStreamPacketHeader.TotalSize, responseBufferDataSize); + Memory<byte> responseBufferMemory = responseBuffer.AsMemory(RawStreamPacketHeader.TotalSize, responseHeader.DataSize); // TODO rework request handler bool responseExists = RequestHandler(clientEndPoint, dataBuffer, totalReceivedBytes, responseBufferMemory); @@ -123,9 +123,7 @@ namespace NetSharp.Raw.Stream switch (responseExists) { case true: - RawStreamPacket response = new RawStreamPacket(in responseBufferMemory, responseBufferDataSize); - - ConfigureSendResponse(args, ref responseBuffer, in response); + ConfigureAsyncSendPacket(args, ref responseBuffer, in responseHeader, responseBufferMemory); StartSend(args); break; @@ -134,7 +132,7 @@ namespace NetSharp.Raw.Stream // packet we can reuse it as a packet header buffer in the below ConfigureReceiveHeader() call BufferPool.Return(responseBuffer, true); - ConfigureReceiveHeader(args); + ConfigureAsyncReceiveHeader(args); StartReceive(args); break; } @@ -176,7 +174,7 @@ namespace NetSharp.Raw.Stream private void CompleteSend(SocketAsyncEventArgs args) { - ResponseWriteToken writeToken = (ResponseWriteToken) args.UserToken; + PacketWriteToken writeToken = (PacketWriteToken) args.UserToken; int sentBytes = args.BytesTransferred, previousSentBytes = args.Offset, @@ -188,7 +186,7 @@ namespace NetSharp.Raw.Stream case SocketError.Success: if (totalSentBytes == expectedBytes) // transmission complete { - ConfigureReceiveHeader(args); + ConfigureAsyncReceiveHeader(args); StartReceive(args); } else if (0 < totalSentBytes && totalSentBytes < expectedBytes) // transmission not complete @@ -209,34 +207,36 @@ namespace NetSharp.Raw.Stream } } - private void ConfigureReceiveData(SocketAsyncEventArgs args, in RawStreamPacketHeader receivedPacketHeader) + private void ConfigureAsyncReceiveData(SocketAsyncEventArgs args, in RawStreamPacketHeader receivedPacketHeader) { BufferPool.Return(args.Buffer, true); // return and clear the requestHeaderBuffer (as it was already parsed) byte[] pendingPacketDataBuffer = BufferPool.Rent(receivedPacketHeader.DataSize); args.SetBuffer(pendingPacketDataBuffer, 0, receivedPacketHeader.DataSize); - args.UserToken = new RequestReadToken(receivedPacketHeader.DataSize, receivedPacketHeader); + args.UserToken = new PacketReadToken(receivedPacketHeader.DataSize, receivedPacketHeader); } - private void ConfigureReceiveHeader(SocketAsyncEventArgs args) + private void ConfigureAsyncReceiveHeader(SocketAsyncEventArgs args) { BufferPool.Return(args.Buffer, true); // return and clear the responseDataBuffer (or requestDataBuffer if no response was generated) byte[] pendingPacketHeaderBuffer = BufferPool.Rent(RawStreamPacketHeader.TotalSize); args.SetBuffer(pendingPacketHeaderBuffer, 0, RawStreamPacketHeader.TotalSize); - args.UserToken = new RequestReadToken(RawStreamPacketHeader.TotalSize, null); + args.UserToken = new PacketReadToken(RawStreamPacketHeader.TotalSize, null); } - private void ConfigureSendResponse(SocketAsyncEventArgs args, ref byte[] pendingPacketBuffer, in RawStreamPacket pendingPacket) + private void ConfigureAsyncSendPacket(SocketAsyncEventArgs args, ref byte[] pendingPacketBuffer, in RawStreamPacketHeader pendingPacketHeader, + in ReadOnlyMemory<byte> pendingPacketData) { BufferPool.Return(args.Buffer, true); // return and clear the requestDataBuffer (as it was already parsed) - pendingPacket.Serialise(pendingPacketBuffer); + RawStreamPacket.Serialise(pendingPacketBuffer, in pendingPacketHeader, in pendingPacketData); - args.SetBuffer(pendingPacketBuffer, 0, pendingPacket.TotalSize); - args.UserToken = new ResponseWriteToken(pendingPacket.TotalSize); + int totalPacketSize = RawStreamPacket.TotalPacketSize(in pendingPacketHeader); + args.SetBuffer(pendingPacketBuffer, 0, totalPacketSize); + args.UserToken = new PacketWriteToken(totalPacketSize); } private void ContinueReceive(SocketAsyncEventArgs args) @@ -394,12 +394,12 @@ namespace NetSharp.Raw.Stream } } - private readonly struct RequestReadToken + private readonly struct PacketReadToken { public readonly int BytesToTransfer; public readonly RawStreamPacketHeader? Header; - public RequestReadToken(int bytesToTransfer, in RawStreamPacketHeader? header) + public PacketReadToken(int bytesToTransfer, in RawStreamPacketHeader? header) { BytesToTransfer = bytesToTransfer; @@ -407,11 +407,11 @@ namespace NetSharp.Raw.Stream } } - private readonly struct ResponseWriteToken + private readonly struct PacketWriteToken { public readonly int BytesToTransfer; - public ResponseWriteToken(int bytesToTransfer) + public PacketWriteToken(int bytesToTransfer) { BytesToTransfer = bytesToTransfer; } diff --git a/NetSharp/NetSharp/Raw/Stream/RawStreamNetworkWriter.cs b/NetSharp/NetSharp/Raw/Stream/RawStreamNetworkWriter.cs @@ -9,19 +9,19 @@ namespace NetSharp.Raw.Stream public sealed class RawStreamNetworkWriter : RawNetworkWriterBase { /// <inheritdoc /> - public RawStreamNetworkWriter(ref Socket rawConnection, EndPoint defaultEndPoint, int maxMessageSize, int pooledBuffersPerBucket = 50, - uint preallocatedStateObjects = 0) : base(ref rawConnection, defaultEndPoint, maxMessageSize, pooledBuffersPerBucket, preallocatedStateObjects) + public RawStreamNetworkWriter(ref Socket rawConnection, EndPoint defaultEndPoint, int maxPooledMessageSize, int pooledBuffersPerBucket = 50, + uint preallocatedStateObjects = 0) : base(ref rawConnection, defaultEndPoint, maxPooledMessageSize, pooledBuffersPerBucket, preallocatedStateObjects) { - if (maxMessageSize <= 0) + if (maxPooledMessageSize <= 0) { - throw new ArgumentOutOfRangeException(nameof(maxMessageSize), maxMessageSize, + throw new ArgumentOutOfRangeException(nameof(maxPooledMessageSize), maxPooledMessageSize, $"The message size must be greater than 0"); } } private void CompleteConnect(SocketAsyncEventArgs args) { - AsyncOperationToken token = (AsyncOperationToken) args.UserToken; + OperationToken token = (OperationToken) args.UserToken; switch (args.SocketError) { @@ -34,8 +34,7 @@ namespace NetSharp.Raw.Stream break; default: - int errorCode = (int) args.SocketError; - token.CompletionSource.SetException(new SocketException(errorCode)); + token.CompletionSource.SetException(new SocketException((int) args.SocketError)); break; } @@ -44,7 +43,7 @@ namespace NetSharp.Raw.Stream private void CompleteDisconnect(SocketAsyncEventArgs args) { - AsyncOperationToken token = (AsyncOperationToken) args.UserToken; + OperationToken token = (OperationToken) args.UserToken; switch (args.SocketError) { @@ -57,8 +56,7 @@ namespace NetSharp.Raw.Stream break; default: - int errorCode = (int) args.SocketError; - token.CompletionSource.SetException(new SocketException(errorCode)); + token.CompletionSource.SetException(new SocketException((int) args.SocketError)); break; } @@ -67,70 +65,175 @@ namespace NetSharp.Raw.Stream private void CompleteReceive(SocketAsyncEventArgs args) { - void CompleteReceiveHeader(SocketAsyncEventArgs args) - { - } + PacketReadToken readToken = (PacketReadToken) args.UserToken; - void CompleteReceiveData(SocketAsyncEventArgs args) + void CompleteReceiveHeader(SocketAsyncEventArgs args, in PacketReadToken readToken) { + int receivedBytes = args.BytesTransferred, + previousReceivedBytes = args.Offset, + totalReceivedBytes = previousReceivedBytes + receivedBytes, + expectedBytes = readToken.BytesToTransfer; + + if (totalReceivedBytes == expectedBytes) // transmission complete + { + Memory<byte> headerBuffer = args.Buffer.AsMemory(0, RawStreamPacketHeader.TotalSize); + RawStreamPacketHeader header = RawStreamPacketHeader.Deserialise(in headerBuffer); + + ConfigureAsyncReceiveData(args, in header, in readToken.UserDataBuffer, readToken.CompletionSource); + + StartReceive(args); + } + else if (0 < totalReceivedBytes && totalReceivedBytes < expectedBytes) // transmission not complete + { + args.SetBuffer(totalReceivedBytes, expectedBytes - totalReceivedBytes); + + ContinueReceive(args); + } + else if (receivedBytes == 0) // connection is dead + { + readToken.CompletionSource.SetException(new SocketException((int) SocketError.HostDown)); + + CleanupTransmissionBufferAndState(args); + } } - } - private void CompleteSend(SocketAsyncEventArgs args) - { - void CompleteSendHeader(SocketAsyncEventArgs args) + void CompleteReceiveData(SocketAsyncEventArgs args, in PacketReadToken readToken) { + int receivedBytes = args.BytesTransferred, + previousReceivedBytes = args.Offset, + totalReceivedBytes = previousReceivedBytes + receivedBytes, + expectedBytes = readToken.BytesToTransfer; + + if (totalReceivedBytes == expectedBytes) // transmission complete + { + args.Buffer.AsMemory(0, readToken.UserDataBuffer.Length).CopyTo(readToken.UserDataBuffer); + + // we only return the number of bytes of user data that were read + readToken.CompletionSource.SetResult(totalReceivedBytes); + + CleanupTransmissionBufferAndState(args); + } + else if (0 < totalReceivedBytes && totalReceivedBytes < expectedBytes) // transmission not complete + { + args.SetBuffer(totalReceivedBytes, expectedBytes - totalReceivedBytes); + + ContinueReceive(args); + } + else if (receivedBytes == 0) // connection is dead + { + readToken.CompletionSource.SetException(new SocketException((int) SocketError.HostDown)); + + CleanupTransmissionBufferAndState(args); + } } - void CompleteSendData(SocketAsyncEventArgs args) + bool receivingHeader = readToken.BytesToTransfer == RawStreamPacketHeader.TotalSize; + + switch (args.SocketError) { + case SocketError.OperationAborted: + readToken.CompletionSource.SetCanceled(); + + CleanupTransmissionBufferAndState(args); + break; + + case SocketError.Success: + switch (receivingHeader) + { + case true: + CompleteReceiveHeader(args, in readToken); + break; + + case false: + CompleteReceiveData(args, in readToken); + break; + } + break; + + default: + readToken.CompletionSource.SetException(new SocketException((int) args.SocketError)); + + CleanupTransmissionBufferAndState(args); + break; } } - private void ConfigureReceiveData(SocketAsyncEventArgs args, in RawStreamPacketHeader receivedPacketHeader) + private void CompleteSend(SocketAsyncEventArgs args) { - BufferPool.Return(args.Buffer, true); // return and clear the requestHeaderBuffer (as it was already parsed) + PacketWriteToken writeToken = (PacketWriteToken) args.UserToken; - byte[] pendingPacketDataBuffer = BufferPool.Rent(receivedPacketHeader.DataSize); + int sentBytes = args.BytesTransferred, + previousSentBytes = args.Offset, + totalSentBytes = previousSentBytes + sentBytes, + expectedBytes = writeToken.BytesToTransfer; - args.SetBuffer(pendingPacketDataBuffer, 0, receivedPacketHeader.DataSize); + switch (args.SocketError) + { + case SocketError.OperationAborted: + writeToken.CompletionSource.SetCanceled(); - // TODO add transmission state token - args.UserToken = new RequestReadToken(receivedPacketHeader.DataSize, receivedPacketHeader); - } + CleanupTransmissionBufferAndState(args); + break; - private void ConfigureReceiveHeader(SocketAsyncEventArgs args) - { - byte[] pendingPacketHeaderBuffer = BufferPool.Rent(RawStreamPacketHeader.TotalSize); + case SocketError.Success: + if (totalSentBytes == expectedBytes) // transmission complete + { + // we only return the number of bytes of user data that were written + writeToken.CompletionSource.SetResult(totalSentBytes - RawStreamPacketHeader.TotalSize); + + CleanupTransmissionBufferAndState(args); + } + else if (0 < totalSentBytes && totalSentBytes < expectedBytes) // transmission not complete + { + args.SetBuffer(totalSentBytes, expectedBytes - totalSentBytes); + + ContinueSend(args); + } + else if (sentBytes == 0) // connection is dead + { + writeToken.CompletionSource.SetException(new SocketException((int) SocketError.HostDown)); + + CleanupTransmissionBufferAndState(args); + } + break; - args.SetBuffer(pendingPacketHeaderBuffer, 0, RawStreamPacketHeader.TotalSize); + default: + writeToken.CompletionSource.SetException(new SocketException((int) args.SocketError)); - // TODO add transmission state token - args.UserToken = new RequestReadToken(RawStreamPacketHeader.TotalSize, null); + CleanupTransmissionBufferAndState(args); + break; + } } - private void ConfigureSendData(SocketAsyncEventArgs args, in RawStreamPacketHeader pendingPacketHeader, in ReadOnlyMemory<byte> pendingPacketData) + private void ConfigureAsyncReceiveData(SocketAsyncEventArgs args, in RawStreamPacketHeader receivedPacketHeader, in Memory<byte> userDataBuffer, + TaskCompletionSource<int> tcs) { - BufferPool.Return(args.Buffer, true); // return and clear the requestHeaderBuffer (as it was already sent) + BufferPool.Return(args.Buffer, true); // return and clear the requestHeaderBuffer (as it was already parsed) - byte[] pendingPacketDataBuffer = BufferPool.Rent(pendingPacketHeader.DataSize); + byte[] pendingPacketDataBuffer = BufferPool.Rent(receivedPacketHeader.DataSize); - args.SetBuffer(pendingPacketDataBuffer, 0, pendingPacketHeader.DataSize); + args.SetBuffer(pendingPacketDataBuffer, 0, receivedPacketHeader.DataSize); // TODO add transmission state token - args.UserToken = new ResponseWriteToken(pendingPacketHeader.DataSize); + args.UserToken = new PacketReadToken(receivedPacketHeader.DataSize, receivedPacketHeader, in userDataBuffer, tcs); } - private void ConfigureSendHeader(SocketAsyncEventArgs args, in RawStreamPacketHeader pendingPacketHeader) + private void ConfigureAsyncReceiveHeader(SocketAsyncEventArgs args, in Memory<byte> userDataBuffer, TaskCompletionSource<int> tcs) { byte[] pendingPacketHeaderBuffer = BufferPool.Rent(RawStreamPacketHeader.TotalSize); - pendingPacketHeader.Serialise(pendingPacketHeaderBuffer); - args.SetBuffer(pendingPacketHeaderBuffer, 0, RawStreamPacketHeader.TotalSize); + args.UserToken = new PacketReadToken(RawStreamPacketHeader.TotalSize, null, in userDataBuffer, tcs); + } - // TODO add transmission state token - args.UserToken = new ResponseWriteToken(RawStreamPacketHeader.TotalSize); + private void ConfigureAsyncSendPacket(SocketAsyncEventArgs args, ref byte[] pendingPacketBuffer, in RawStreamPacketHeader pendingPacketHeader, + in ReadOnlyMemory<byte> userDataBuffer, TaskCompletionSource<int> tcs) + { + RawStreamPacket.Serialise(pendingPacketBuffer, in pendingPacketHeader, in userDataBuffer); + + int totalPacketSize = RawStreamPacket.TotalPacketSize(in pendingPacketHeader); + args.SetBuffer(pendingPacketBuffer, 0, totalPacketSize); + args.UserToken = new PacketWriteToken(totalPacketSize, tcs); } private void ContinueReceive(SocketAsyncEventArgs args) @@ -175,6 +278,26 @@ namespace NetSharp.Raw.Stream } } + private void StartReceive(SocketAsyncEventArgs args) + { + if (Connection.ReceiveAsync(args)) + { + return; + } + + CompleteReceive(args); + } + + private void StartSend(SocketAsyncEventArgs args) + { + if (Connection.SendAsync(args)) + { + return; + } + + CompleteSend(args); + } + /// <inheritdoc /> protected override bool CanReuseStateObject(ref SocketAsyncEventArgs instance) { @@ -216,7 +339,7 @@ namespace NetSharp.Raw.Stream args.RemoteEndPoint = remoteEndPoint; - AsyncOperationToken token = new AsyncOperationToken(tcs); + OperationToken token = new OperationToken(tcs); args.UserToken = token; if (Connection.ConnectAsync(args)) @@ -241,7 +364,7 @@ namespace NetSharp.Raw.Stream args.DisconnectReuseSocket = reuseSocket; - AsyncOperationToken token = new AsyncOperationToken(tcs); + OperationToken token = new OperationToken(tcs); args.UserToken = token; if (Connection.DisconnectAsync(args)) @@ -256,44 +379,124 @@ namespace NetSharp.Raw.Stream public override int Read(ref EndPoint remoteEndPoint, Memory<byte> readBuffer, SocketFlags flags = SocketFlags.None) { - throw new NotImplementedException(); + static int ReadBytesIntoBuffer(Socket connection, ref byte[] buffer, int count, SocketFlags flags) + { + int readBytes = 0; + + do + { + readBytes += connection.Receive(buffer, readBytes, count - readBytes, flags); + } while (readBytes < count && readBytes > 0); + + return readBytes; + } + + byte[] pendingHeaderBuffer = BufferPool.Rent(RawStreamPacketHeader.TotalSize); + + int _ = ReadBytesIntoBuffer(Connection, ref pendingHeaderBuffer, RawStreamPacketHeader.TotalSize, flags); + + RawStreamPacketHeader packetHeader = RawStreamPacketHeader.Deserialise(pendingHeaderBuffer); + BufferPool.Return(pendingHeaderBuffer, true); // return and clear the pendingHeaderBuffer (as it was already parsed) + + byte[] pendingPacketDataBuffer = BufferPool.Rent(packetHeader.DataSize); + + int bodyBytes = ReadBytesIntoBuffer(Connection, ref pendingPacketDataBuffer, packetHeader.DataSize, flags); + + pendingPacketDataBuffer.AsMemory(0, readBuffer.Length).CopyTo(readBuffer); + BufferPool.Return(pendingPacketDataBuffer, true); // return and clear the pendingDataBuffer (as it was already copied) + + return bodyBytes; // we only return the number of bytes of user data that were read } public override ValueTask<int> ReadAsync(EndPoint remoteEndPoint, Memory<byte> readBuffer, SocketFlags flags = SocketFlags.None) { - throw new NotImplementedException(); + TaskCompletionSource<int> tcs = new TaskCompletionSource<int>(); + SocketAsyncEventArgs args = ArgsPool.Rent(); + + ConfigureAsyncReceiveHeader(args, in readBuffer, tcs); + + args.RemoteEndPoint = remoteEndPoint; + args.SocketFlags = flags; + + StartReceive(args); + + return new ValueTask<int>(tcs.Task); } public override int Write(EndPoint remoteEndPoint, ReadOnlyMemory<byte> writeBuffer, SocketFlags flags = SocketFlags.None) { - throw new NotImplementedException(); + static int WriteBytesFromBuffer(Socket connection, ref byte[] buffer, int count, SocketFlags flags) + { + int writtenBytes = 0; + + do + { + writtenBytes += connection.Send(buffer, writtenBytes, count - writtenBytes, flags); + } while (writtenBytes < count && writtenBytes > 0); + + return writtenBytes; + } + + RawStreamPacketHeader pendingPacketHeader = new RawStreamPacketHeader(writeBuffer.Length); + int totalPacketSize = RawStreamPacket.TotalPacketSize(in pendingPacketHeader); + byte[] pendingPacketBuffer = BufferPool.Rent(totalPacketSize); + + RawStreamPacket.Serialise(pendingPacketBuffer, in pendingPacketHeader, in writeBuffer); + + int _ = WriteBytesFromBuffer(Connection, ref pendingPacketBuffer, totalPacketSize, flags); + BufferPool.Return(pendingPacketBuffer, true); // return and clear the pendingPacketBuffer (as it was already cleared) + + return pendingPacketHeader.DataSize; // we only return the number of bytes of user data that were written } public override ValueTask<int> WriteAsync(EndPoint remoteEndPoint, ReadOnlyMemory<byte> writeBuffer, SocketFlags flags = SocketFlags.None) { - throw new NotImplementedException(); + TaskCompletionSource<int> tcs = new TaskCompletionSource<int>(); + SocketAsyncEventArgs args = ArgsPool.Rent(); + + RawStreamPacketHeader pendingPacketHeader = new RawStreamPacketHeader(writeBuffer.Length); + int totalPacketSize = RawStreamPacket.TotalPacketSize(in pendingPacketHeader); + byte[] pendingPacketBuffer = BufferPool.Rent(totalPacketSize); + + ConfigureAsyncSendPacket(args, ref pendingPacketBuffer, in pendingPacketHeader, in writeBuffer, tcs); + + args.RemoteEndPoint = remoteEndPoint; + args.SocketFlags = flags; + + StartSend(args); + + return new ValueTask<int>(tcs.Task); } - private readonly struct RequestReadToken + private readonly struct PacketReadToken { public readonly int BytesToTransfer; + public readonly TaskCompletionSource<int> CompletionSource; public readonly RawStreamPacketHeader? Header; + public readonly Memory<byte> UserDataBuffer; - public RequestReadToken(int bytesToTransfer, in RawStreamPacketHeader? header) + public PacketReadToken(int bytesToTransfer, in RawStreamPacketHeader? header, in Memory<byte> userDataBuffer, TaskCompletionSource<int> tcs) { BytesToTransfer = bytesToTransfer; Header = header; + + UserDataBuffer = userDataBuffer; + + CompletionSource = tcs; } } - private readonly struct ResponseWriteToken + private readonly struct PacketWriteToken { public readonly int BytesToTransfer; + public readonly TaskCompletionSource<int> CompletionSource; - public ResponseWriteToken(int bytesToTransfer) + public PacketWriteToken(int bytesToTransfer, TaskCompletionSource<int> tcs) { BytesToTransfer = bytesToTransfer; + + CompletionSource = tcs; } } } diff --git a/NetSharp/NetSharp/Raw/Stream/RawStreamPacket.cs b/NetSharp/NetSharp/Raw/Stream/RawStreamPacket.cs @@ -7,44 +7,25 @@ namespace NetSharp.Raw.Stream { public readonly struct RawStreamPacket { - public readonly Memory<byte> Data; - public readonly RawStreamPacketHeader Header; - - private RawStreamPacket(in RawStreamPacketHeader packetHeader, in Memory<byte> packetData) - { - Header = packetHeader; - - Data = packetData; - } - - public RawStreamPacket(in Memory<byte> packetData, int packetDataSize) - { - Header = new RawStreamPacketHeader(packetDataSize); - - Data = packetData; - } - - public int TotalSize => RawStreamPacketHeader.TotalSize + Data.Length; - - public static RawStreamPacket Deserialise(in Memory<byte> buffer) + public static (RawStreamPacketHeader packetHeader, ReadOnlyMemory<byte> packetData) Deserialise(in Memory<byte> buffer) { Memory<byte> serialisedHeader = buffer.Slice(0, RawStreamPacketHeader.TotalSize); RawStreamPacketHeader header = RawStreamPacketHeader.Deserialise(in serialisedHeader); - return new RawStreamPacket(in header, in buffer); + return (header, buffer.Slice(RawStreamPacketHeader.TotalSize)); } - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static int TotalPacketSize(int packetDataSize) + public static void Serialise(in Memory<byte> buffer, in RawStreamPacketHeader packetHeader, in ReadOnlyMemory<byte> packetData) { - return RawStreamPacketHeader.TotalSize + packetDataSize; + packetHeader.Serialise(buffer.Slice(0, RawStreamPacketHeader.TotalSize)); + + packetData.CopyTo(buffer.Slice(RawStreamPacketHeader.TotalSize, packetData.Length)); } - public void Serialise(in Memory<byte> buffer) + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public static int TotalPacketSize(in RawStreamPacketHeader packetHeader) { - Header.Serialise(buffer.Slice(0, RawStreamPacketHeader.TotalSize)); - - Data.CopyTo(buffer.Slice(RawStreamPacketHeader.TotalSize, Data.Length)); + return RawStreamPacketHeader.TotalSize + packetHeader.DataSize; } } @@ -72,5 +53,10 @@ namespace NetSharp.Raw.Stream Span<byte> serialisedDataSize = EndianAwareBitConverter.GetBytes(DataSize); serialisedDataSize.CopyTo(buffer.Slice(0, sizeof(int)).Span); } + + public override string ToString() + { + return $"[Data Segment Size: {DataSize}]"; + } } } \ No newline at end of file diff --git a/NetSharp/NetSharpExamples/Benchmarks/Stream Network Connection Benchmarks/StreamNetworkReaderBenchmark.cs b/NetSharp/NetSharpExamples/Benchmarks/Stream Network Connection Benchmarks/StreamNetworkReaderBenchmark.cs @@ -23,7 +23,7 @@ namespace NetSharpExamples.Benchmarks.Stream_Network_Connection_Benchmarks public static readonly ManualResetEventSlim ServerReadyEvent = new ManualResetEventSlim(); /// <inheritdoc /> - public string Name { get; } = "Raw Variable Packet-size Stream Network Reader Benchmark"; + public string Name { get; } = "Raw Stream Network Reader Benchmark"; private static bool RequestHandler(EndPoint remoteEndPoint, in ReadOnlyMemory<byte> requestBuffer, int receivedRequestBytes, in Memory<byte> responseBuffer) @@ -60,8 +60,8 @@ namespace NetSharpExamples.Benchmarks.Stream_Network_Connection_Benchmarks { ServerEncoding.GetBytes($"[Client {id}] Hello World! (Packet {i})").CopyTo(packetBuffer, 0); - RawStreamPacket streamPacket = new RawStreamPacket(packetBuffer, PacketSize); - streamPacket.Serialise(sendBuffer); + RawStreamPacketHeader streamPacketHeader = new RawStreamPacketHeader(PacketSize); + RawStreamPacket.Serialise(sendBuffer, in streamPacketHeader, packetBuffer); benchmarkHelper.StartStopwatch(); diff --git a/NetSharp/NetSharpExamples/Benchmarks/Stream Network Connection Benchmarks/StreamNetworkWriterAsyncBenchmark.cs b/NetSharp/NetSharpExamples/Benchmarks/Stream Network Connection Benchmarks/StreamNetworkWriterAsyncBenchmark.cs @@ -21,7 +21,7 @@ namespace NetSharpExamples.Benchmarks.Stream_Network_Connection_Benchmarks public static readonly ManualResetEventSlim ServerReadyEvent = new ManualResetEventSlim(); /// <inheritdoc /> - public string Name { get; } = "Raw Variable Packet-size Stream Network Writer Benchmark (Asynchronous)"; + public string Name { get; } = "Raw Stream Network Writer Benchmark (Asynchronous)"; private static Task ServerTask(CancellationToken cancellationToken) { @@ -30,7 +30,8 @@ namespace NetSharpExamples.Benchmarks.Stream_Network_Connection_Benchmarks server.Bind(ServerEndPoint); ServerReadyEvent.Set(); - byte[] transmissionBuffer = new byte[PacketSize]; + RawStreamPacketHeader archetypalHeader = new RawStreamPacketHeader(PacketSize); + byte[] transmissionBuffer = new byte[RawStreamPacket.TotalPacketSize(in archetypalHeader)]; server.Listen(1); Socket clientSocket = server.Accept(); diff --git a/NetSharp/NetSharpExamples/Benchmarks/Stream Network Connection Benchmarks/StreamNetworkWriterSyncBenchmark.cs b/NetSharp/NetSharpExamples/Benchmarks/Stream Network Connection Benchmarks/StreamNetworkWriterSyncBenchmark.cs @@ -21,7 +21,7 @@ namespace NetSharpExamples.Benchmarks.Stream_Network_Connection_Benchmarks public static readonly ManualResetEventSlim ServerReadyEvent = new ManualResetEventSlim(); /// <inheritdoc /> - public string Name { get; } = "Raw Variable Packet-size Stream Network Writer Benchmark (Synchronous)"; + public string Name { get; } = "Raw Stream Network Writer Benchmark (Synchronous)"; private static Task ServerTask(CancellationToken cancellationToken) { @@ -30,7 +30,9 @@ namespace NetSharpExamples.Benchmarks.Stream_Network_Connection_Benchmarks server.Bind(ServerEndPoint); ServerReadyEvent.Set(); - byte[] transmissionBuffer = new byte[PacketSize]; + // all the headers should have the same packet size, so will fit in the transmission buffer + RawStreamPacketHeader archetypalHeader = new RawStreamPacketHeader(PacketSize); + byte[] transmissionBuffer = new byte[RawStreamPacket.TotalPacketSize(in archetypalHeader)]; server.Listen(1); Socket clientSocket = server.Accept(); @@ -43,7 +45,7 @@ namespace NetSharpExamples.Benchmarks.Stream_Network_Connection_Benchmarks do { receivedBytes += clientSocket.Receive(transmissionBuffer, receivedBytes, expectedBytes - receivedBytes, SocketFlags.None); - } while (receivedBytes != 0 && receivedBytes < expectedBytes); + } while (receivedBytes < expectedBytes && receivedBytes > 0); if (receivedBytes == 0) { @@ -54,7 +56,7 @@ namespace NetSharpExamples.Benchmarks.Stream_Network_Connection_Benchmarks do { sentBytes += clientSocket.Send(transmissionBuffer, sentBytes, expectedBytes - sentBytes, SocketFlags.None); - } while (sentBytes != 0 && sentBytes < expectedBytes); + } while (sentBytes < expectedBytes && sentBytes > 0); if (sentBytes == 0) { diff --git a/NetSharp/NetSharpExamples/Examples/Stream Network Connection Examples/StreamNetworkReaderExample.cs b/NetSharp/NetSharpExamples/Examples/Stream Network Connection Examples/StreamNetworkReaderExample.cs @@ -15,7 +15,7 @@ namespace NetSharpExamples.Examples.Stream_Network_Connection_Examples public static readonly EndPoint ServerEndPoint = new IPEndPoint(IPAddress.Loopback, 12377); /// <inheritdoc /> - public string Name { get; } = "Raw Variable Packet-size Stream Network Reader Example"; + public string Name { get; } = "Raw Stream Network Reader Example"; private static bool RequestHandler(EndPoint remoteEndPoint, in ReadOnlyMemory<byte> requestBuffer, int receivedRequestBytes, in Memory<byte> responseBuffer) diff --git a/NetSharp/NetSharpExamples/Examples/Stream Network Connection Examples/StreamNetworkWriterAsyncExample.cs b/NetSharp/NetSharpExamples/Examples/Stream Network Connection Examples/StreamNetworkWriterAsyncExample.cs @@ -18,7 +18,7 @@ namespace NetSharpExamples.Examples.Stream_Network_Connection_Examples public static readonly EndPoint ServerEndPoint = StreamNetworkReaderExample.ServerEndPoint; /// <inheritdoc /> - public string Name { get; } = "Raw Variable Packet-size Stream Network Writer Example (Asynchronous)"; + public string Name { get; } = "Raw Stream Network Writer Example (Asynchronous)"; /// <inheritdoc /> public async Task RunAsync() diff --git a/NetSharp/NetSharpExamples/Examples/Stream Network Connection Examples/StreamNetworkWriterSyncExample.cs b/NetSharp/NetSharpExamples/Examples/Stream Network Connection Examples/StreamNetworkWriterSyncExample.cs @@ -18,7 +18,7 @@ namespace NetSharpExamples.Examples.Stream_Network_Connection_Examples public static readonly EndPoint ServerEndPoint = StreamNetworkReaderExample.ServerEndPoint; /// <inheritdoc /> - public string Name { get; } = "Raw Variable Packet-size Stream Network Writer Example (Synchronous)"; + public string Name { get; } = "Raw Stream Network Writer Example (Synchronous)"; /// <inheritdoc /> public Task RunAsync()