commit 520db7e9ddc416452bceb18e20ecfba2eaa81846
parent d495b2314b54b3cd04ce467725216f2150f553da
Author: Mikolaj Lenczewski <mikolaj.lenczewski308@gmail.com>
Date: Sun, 26 Apr 2020 14:11:40 +0100
Improved send memory buffer safety, and short-circuited cancellation token setup if the passed token is its default value.
Diffstat:
5 files changed, 261 insertions(+), 190 deletions(-)
diff --git a/NetSharp/NetSharp/NetSharp.xml b/NetSharp/NetSharp/NetSharp.xml
@@ -169,7 +169,7 @@
<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.
+ <see cref="M:NetSharp.Sockets.Datagram.DatagramSocketClient.SendToAsync(System.Net.EndPoint,System.ReadOnlyMemory{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)">
@@ -328,24 +328,50 @@
The cancellation token to observe during the operation.
</param>
</member>
- <member name="T:NetSharp.Sockets.SocketClient.AsyncTransmissionToken">
+ <member name="T:NetSharp.Sockets.SocketClient.AsyncReceiveToken">
<summary>
- A state token for asynchronous network IO operations.
+ A state token for asynchronous incoming network IO operations.
</summary>
</member>
- <member name="F:NetSharp.Sockets.SocketClient.AsyncTransmissionToken.CancellationToken">
+ <member name="F:NetSharp.Sockets.SocketClient.AsyncReceiveToken.CancellationToken">
<summary>
The <see cref="T:System.Threading.CancellationToken" /> associated with the network IO operation.
</summary>
</member>
- <member name="F:NetSharp.Sockets.SocketClient.AsyncTransmissionToken.CompletionSource">
+ <member name="F:NetSharp.Sockets.SocketClient.AsyncReceiveToken.CompletionSource">
<summary>
The completion source which wraps the event-based APM, and provides an awaitable <see cref="T:System.Threading.Tasks.Task" />.
</summary>
</member>
- <member name="M:NetSharp.Sockets.SocketClient.AsyncTransmissionToken.#ctor(System.Threading.Tasks.TaskCompletionSource{NetSharp.Utils.TransmissionResult}@,System.Net.Sockets.SocketAsyncEventArgs@,System.Threading.CancellationToken@)">
+ <member name="M:NetSharp.Sockets.SocketClient.AsyncReceiveToken.#ctor(System.Threading.Tasks.TaskCompletionSource{NetSharp.Utils.TransmissionResult}@,System.Threading.CancellationToken@)">
<summary>
- Constructs a new instance of the <see cref="T:NetSharp.Sockets.SocketClient.AsyncTransmissionToken" /> struct.
+ Constructs a new instance of the <see cref="T:NetSharp.Sockets.SocketClient.AsyncReceiveToken" /> struct.
+ </summary>
+ <param name="completionSource">
+ The completion source to trigger when the IO operation completes.
+ </param>
+ <param name="cancellationToken">
+ The cancellation token to observe during the operation.
+ </param>
+ </member>
+ <member name="T:NetSharp.Sockets.SocketClient.AsyncSendToken">
+ <summary>
+ A state token for asynchronous outgoing network IO operations.
+ </summary>
+ </member>
+ <member name="F:NetSharp.Sockets.SocketClient.AsyncSendToken.CancellationToken">
+ <summary>
+ The <see cref="T:System.Threading.CancellationToken" /> associated with the network IO operation.
+ </summary>
+ </member>
+ <member name="F:NetSharp.Sockets.SocketClient.AsyncSendToken.CompletionSource">
+ <summary>
+ The completion source which wraps the event-based APM, and provides an awaitable <see cref="T:System.Threading.Tasks.Task" />.
+ </summary>
+ </member>
+ <member name="M:NetSharp.Sockets.SocketClient.AsyncSendToken.#ctor(System.Threading.Tasks.TaskCompletionSource{NetSharp.Utils.TransmissionResult}@,System.Byte[]@,System.Threading.CancellationToken@)">
+ <summary>
+ Constructs a new instance of the <see cref="T:NetSharp.Sockets.SocketClient.AsyncSendToken" /> struct.
</summary>
<param name="completionSource">
The completion source to trigger when the IO operation completes.
@@ -564,7 +590,7 @@
<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.
+ <see cref="M:NetSharp.Sockets.Stream.StreamSocketClient.SendAsync(System.ReadOnlyMemory{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)">
diff --git a/NetSharp/NetSharp/Sockets/Datagram/DatagramSocketClient.cs b/NetSharp/NetSharp/Sockets/Datagram/DatagramSocketClient.cs
@@ -2,7 +2,6 @@
using NetSharp.Utils;
using System;
-using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Threading;
@@ -84,13 +83,9 @@ namespace NetSharp.Sockets.Datagram
private void CompleteReceiveFrom(SocketAsyncEventArgs args)
{
- AsyncTransmissionToken receiveToken = (AsyncTransmissionToken)args.UserToken;
+ AsyncReceiveToken receiveToken = (AsyncReceiveToken)args.UserToken;
- if (receiveToken.CancellationToken.IsCancellationRequested)
- {
- Debug.WriteLine("ReceiveFrom completed after being cancelled!");
- return;
- }
+ if (receiveToken.CancellationToken.IsCancellationRequested) return;
switch (args.SocketError)
{
@@ -102,7 +97,6 @@ namespace NetSharp.Sockets.Datagram
break;
case SocketError.OperationAborted:
- Debug.WriteLine("ReceiveFrom received operation aborted!");
break;
default:
@@ -116,7 +110,7 @@ namespace NetSharp.Sockets.Datagram
private void CompleteSendTo(SocketAsyncEventArgs args)
{
- AsyncTransmissionToken sendToken = (AsyncTransmissionToken)args.UserToken;
+ AsyncSendToken sendToken = (AsyncSendToken)args.UserToken;
if (sendToken.CancellationToken.IsCancellationRequested) return;
@@ -138,6 +132,7 @@ namespace NetSharp.Sockets.Datagram
break;
}
+ BufferPool.Return(sendToken.RentedBuffer, true);
TransmissionArgsPool.Return(args);
}
@@ -213,14 +208,20 @@ namespace NetSharp.Sockets.Datagram
args.RemoteEndPoint = remoteEndPoint;
args.SocketFlags = flags;
- args.UserToken = new AsyncTransmissionToken(in tcs, in args, cancellationToken);
+ args.UserToken = new AsyncReceiveToken(in tcs, in cancellationToken);
- // TODO find out why the fricc we leak memory
- CancellationTokenRegistration cancellationRegistration =
- cancellationToken.Register(CancelAsyncTransmissionCallback, args);
+ if (cancellationToken == default)
+ {
+ if (Connection.ReceiveFromAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task);
+ }
+ else
+ {
+ // TODO find out why the fricc we leak memory
+ CancellationTokenRegistration cancellationRegistration =
+ cancellationToken.Register(CancelAsyncTransmissionCallback, args);
- if (Connection.ReceiveFromAsync(args))
- return new ValueTask<TransmissionResult>(
+ if (Connection.ReceiveFromAsync(args))
+ return new ValueTask<TransmissionResult>(
tcs.Task.ContinueWith((task, state) =>
{
((CancellationTokenRegistration)state).Dispose();
@@ -229,7 +230,8 @@ namespace NetSharp.Sockets.Datagram
}, cancellationRegistration, CancellationToken.None)
);
- cancellationRegistration.Dispose();
+ cancellationRegistration.Dispose();
+ }
TransmissionResult result = new TransmissionResult(in args);
@@ -245,37 +247,48 @@ namespace NetSharp.Sockets.Datagram
return new TransmissionResult(in sendBuffer, in sentBytes, in remoteEndPoint);
}
- public ValueTask<TransmissionResult> SendToAsync(EndPoint remoteEndPoint, Memory<byte> sendBuffer,
+ public ValueTask<TransmissionResult> SendToAsync(EndPoint remoteEndPoint, ReadOnlyMemory<byte> sendBuffer,
SocketFlags flags = SocketFlags.None, CancellationToken cancellationToken = default)
{
TaskCompletionSource<TransmissionResult> tcs = new TaskCompletionSource<TransmissionResult>();
SocketAsyncEventArgs args = TransmissionArgsPool.Rent();
+ byte[] transmissionBuffer = BufferPool.Rent(sendBuffer.Length);
+
+ sendBuffer.CopyTo(transmissionBuffer);
- args.SetBuffer(sendBuffer);
+ args.SetBuffer(transmissionBuffer);
args.RemoteEndPoint = remoteEndPoint;
args.SocketFlags = flags;
- args.UserToken = new AsyncTransmissionToken(in tcs, in args, cancellationToken);
+ args.UserToken = new AsyncSendToken(in tcs, in transmissionBuffer, in cancellationToken);
- // TODO find out why the fricc we leak memory
- CancellationTokenRegistration cancellationRegistration =
- cancellationToken.Register(CancelAsyncTransmissionCallback, args);
+ if (cancellationToken == default)
+ {
+ if (Connection.SendToAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task);
+ }
+ else
+ {
+ // TODO find out why the fricc we leak memory
+ CancellationTokenRegistration cancellationRegistration =
+ cancellationToken.Register(CancelAsyncTransmissionCallback, args);
- if (Connection.SendToAsync(args))
- return new ValueTask<TransmissionResult>(
- tcs.Task.ContinueWith((task, state) =>
- {
- ((CancellationTokenRegistration)state).Dispose();
+ if (Connection.SendToAsync(args))
+ return new ValueTask<TransmissionResult>(
+ tcs.Task.ContinueWith((task, state) =>
+ {
+ ((CancellationTokenRegistration)state).Dispose();
- return task.Result;
- }, cancellationRegistration, CancellationToken.None)
- );
+ return task.Result;
+ }, cancellationRegistration, CancellationToken.None)
+ );
- cancellationRegistration.Dispose();
+ cancellationRegistration.Dispose();
+ }
TransmissionResult result = new TransmissionResult(in args);
+ BufferPool.Return(transmissionBuffer, true);
TransmissionArgsPool.Return(args);
return new ValueTask<TransmissionResult>(result);
diff --git a/NetSharp/NetSharp/Sockets/SocketClient.cs b/NetSharp/NetSharp/Sockets/SocketClient.cs
@@ -1,6 +1,4 @@
-using System;
-using System.Diagnostics;
-using NetSharp.Utils;
+using NetSharp.Utils;
using System.Net;
using System.Net.Sockets;
@@ -12,7 +10,7 @@ namespace NetSharp.Sockets
/// <summary>
/// Abstract base class for clients.
/// </summary>
- /// TODO implement cancellation of ReceiveAsync and ReceiveFromAsync methods.
+ /// TODO implement proper memory leak-free cancellation of network IO operations
public abstract class SocketClient : SocketConnection
{
/// <summary>
@@ -54,10 +52,11 @@ namespace NetSharp.Sockets
{
SocketAsyncEventArgs args = (SocketAsyncEventArgs)state;
- AsyncTransmissionToken token = (AsyncTransmissionToken)args.UserToken;
+ AsyncSendToken token = (AsyncSendToken)args.UserToken;
token.CompletionSource.SetResult(TransmissionResult.Timeout);
+ BufferPool.Return(token.RentedBuffer, true);
DestroyTransmissionArgs(args);
}
@@ -90,7 +89,7 @@ namespace NetSharp.Sockets
SocketAsyncEventArgs args = TransmissionArgsPool.Rent();
args.RemoteEndPoint = remoteEndPoint;
- args.UserToken = new AsyncOperationToken(in tcs, CancellationToken.None);
+ args.UserToken = new AsyncOperationToken(in tcs, in cancellationToken);
// TODO find out why the fricc we leak memory
CancellationTokenRegistration cancellationRegistration =
@@ -146,9 +145,9 @@ namespace NetSharp.Sockets
}
/// <summary>
- /// A state token for asynchronous network IO operations.
+ /// A state token for asynchronous incoming network IO operations.
/// </summary>
- protected readonly struct AsyncTransmissionToken
+ protected readonly struct AsyncReceiveToken
{
/// <summary>
/// The <see cref="System.Threading.CancellationToken" /> associated with the network IO operation.
@@ -160,10 +159,42 @@ namespace NetSharp.Sockets
/// </summary>
public readonly TaskCompletionSource<TransmissionResult> CompletionSource;
- public readonly SocketAsyncEventArgs TransmissionArgs;
+ /// <summary>
+ /// Constructs a new instance of the <see cref="AsyncReceiveToken" /> struct.
+ /// </summary>
+ /// <param name="completionSource">
+ /// The completion source to trigger when the IO operation completes.
+ /// </param>
+ /// <param name="cancellationToken">
+ /// The cancellation token to observe during the operation.
+ /// </param>
+ public AsyncReceiveToken(in TaskCompletionSource<TransmissionResult> completionSource, in CancellationToken cancellationToken)
+ {
+ CompletionSource = completionSource;
+
+ CancellationToken = cancellationToken;
+ }
+ }
+
+ /// <summary>
+ /// A state token for asynchronous outgoing network IO operations.
+ /// </summary>
+ protected readonly struct AsyncSendToken
+ {
+ /// <summary>
+ /// The <see cref="System.Threading.CancellationToken" /> associated with the network IO operation.
+ /// </summary>
+ public readonly CancellationToken CancellationToken;
+
+ /// <summary>
+ /// The completion source which wraps the event-based APM, and provides an awaitable <see cref="Task" />.
+ /// </summary>
+ public readonly TaskCompletionSource<TransmissionResult> CompletionSource;
+
+ public readonly byte[] RentedBuffer;
/// <summary>
- /// Constructs a new instance of the <see cref="AsyncTransmissionToken" /> struct.
+ /// Constructs a new instance of the <see cref="AsyncSendToken" /> struct.
/// </summary>
/// <param name="completionSource">
/// The completion source to trigger when the IO operation completes.
@@ -171,11 +202,11 @@ namespace NetSharp.Sockets
/// <param name="cancellationToken">
/// The cancellation token to observe during the operation.
/// </param>
- public AsyncTransmissionToken(in TaskCompletionSource<TransmissionResult> completionSource, in SocketAsyncEventArgs transmissionArgs, in CancellationToken cancellationToken)
+ public AsyncSendToken(in TaskCompletionSource<TransmissionResult> completionSource, in byte[] rentedBuffer, in CancellationToken cancellationToken)
{
CompletionSource = completionSource;
- TransmissionArgs = transmissionArgs;
+ RentedBuffer = rentedBuffer;
CancellationToken = cancellationToken;
}
diff --git a/NetSharp/NetSharp/Sockets/Stream/StreamSocketClient.cs b/NetSharp/NetSharp/Sockets/Stream/StreamSocketClient.cs
@@ -2,7 +2,6 @@
using NetSharp.Utils;
using System;
-using System.Diagnostics;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;
@@ -60,23 +59,22 @@ namespace NetSharp.Sockets.Stream
{
AsyncOperationToken connectToken = (AsyncOperationToken)args.UserToken;
- if (!connectToken.CancellationToken.IsCancellationRequested)
+ if (connectToken.CancellationToken.IsCancellationRequested) return;
+
+ switch (args.SocketError)
{
- switch (args.SocketError)
- {
- case SocketError.Success:
- connectToken.CompletionSource.SetResult(true);
+ case SocketError.Success:
+ connectToken.CompletionSource.SetResult(true);
- break;
+ break;
- case SocketError.OperationAborted:
- break;
+ case SocketError.OperationAborted:
+ break;
- default:
- connectToken.CompletionSource.SetException(new SocketException((int)args.SocketError));
+ default:
+ connectToken.CompletionSource.SetException(new SocketException((int)args.SocketError));
- break;
- }
+ break;
}
TransmissionArgsPool.Return(args);
@@ -86,23 +84,22 @@ namespace NetSharp.Sockets.Stream
{
AsyncOperationToken disconnectToken = (AsyncOperationToken)args.UserToken;
- if (!disconnectToken.CancellationToken.IsCancellationRequested)
+ if (disconnectToken.CancellationToken.IsCancellationRequested) return;
+
+ switch (args.SocketError)
{
- switch (args.SocketError)
- {
- case SocketError.Success:
- disconnectToken.CompletionSource.SetResult(true);
+ case SocketError.Success:
+ disconnectToken.CompletionSource.SetResult(true);
- break;
+ break;
- case SocketError.OperationAborted:
- break;
+ case SocketError.OperationAborted:
+ break;
- default:
- disconnectToken.CompletionSource.SetException(new SocketException((int)args.SocketError));
+ default:
+ disconnectToken.CompletionSource.SetException(new SocketException((int)args.SocketError));
- break;
- }
+ break;
}
TransmissionArgsPool.Return(args);
@@ -110,118 +107,105 @@ namespace NetSharp.Sockets.Stream
private void CompleteReceive(SocketAsyncEventArgs args)
{
- AsyncTransmissionToken receiveToken = (AsyncTransmissionToken)args.UserToken;
+ AsyncReceiveToken receiveToken = (AsyncReceiveToken)args.UserToken;
- if (!receiveToken.CancellationToken.IsCancellationRequested)
- {
- switch (args.SocketError)
- {
- case SocketError.Success:
- Memory<byte> transmissionBuffer = args.MemoryBuffer;
- int expectedBytes = transmissionBuffer.Length;
-
- if (args.BytesTransferred == expectedBytes)
- {
- // buffer was fully received
-
- TransmissionResult result = new TransmissionResult(in args);
-
- receiveToken.CompletionSource.SetResult(result);
+ if (receiveToken.CancellationToken.IsCancellationRequested) return;
- TransmissionArgsPool.Return(args);
- }
- else if (expectedBytes > args.BytesTransferred && args.BytesTransferred > 0)
- {
- // receive the remaining parts of the buffer
-
- int receivedBytes = args.BytesTransferred;
+ switch (args.SocketError)
+ {
+ case SocketError.Success:
+ Memory<byte> transmissionBuffer = args.MemoryBuffer;
+ int expectedBytes = transmissionBuffer.Length;
- args.SetBuffer(receivedBytes, expectedBytes - receivedBytes);
+ if (args.BytesTransferred == expectedBytes)
+ {
+ // buffer was fully received
- Connection.ReceiveAsync(args);
- }
- else
- {
- // no bytes were received, remote socket is dead
+ TransmissionResult result = new TransmissionResult(in args);
- receiveToken.CompletionSource.SetException(new SocketException((int)SocketError.HostDown));
+ receiveToken.CompletionSource.SetResult(result);
+ }
+ else if (expectedBytes > args.BytesTransferred && args.BytesTransferred > 0)
+ {
+ // receive the remaining parts of the buffer
- TransmissionArgsPool.Return(args);
- }
+ int receivedBytes = args.BytesTransferred;
- break;
+ args.SetBuffer(receivedBytes, expectedBytes - receivedBytes);
- case SocketError.OperationAborted:
- TransmissionArgsPool.Return(args);
+ Connection.ReceiveAsync(args);
+ return;
+ }
+ else
+ {
+ // no bytes were received, remote socket is dead
- break;
+ receiveToken.CompletionSource.SetException(new SocketException((int)SocketError.HostDown));
+ }
- default:
- receiveToken.CompletionSource.SetException(new SocketException((int)args.SocketError));
+ break;
- TransmissionArgsPool.Return(args);
+ case SocketError.OperationAborted:
+ break;
- break;
- }
+ default:
+ receiveToken.CompletionSource.SetException(new SocketException((int)args.SocketError));
+ break;
}
+
+ TransmissionArgsPool.Return(args);
}
private void CompleteSend(SocketAsyncEventArgs args)
{
- AsyncTransmissionToken sendToken = (AsyncTransmissionToken)args.UserToken;
+ AsyncSendToken sendToken = (AsyncSendToken)args.UserToken;
- if (!sendToken.CancellationToken.IsCancellationRequested)
- {
- switch (args.SocketError)
- {
- case SocketError.Success:
- Memory<byte> transmissionBuffer = args.MemoryBuffer;
- int remainingBytes = transmissionBuffer.Length;
+ if (sendToken.CancellationToken.IsCancellationRequested) return;
- if (args.BytesTransferred == remainingBytes)
- {
- // buffer was fully sent
-
- TransmissionResult result = new TransmissionResult(in args);
-
- sendToken.CompletionSource.SetResult(result);
-
- TransmissionArgsPool.Return(args);
- }
- else if (remainingBytes > args.BytesTransferred && args.BytesTransferred > 0)
- {
- // send the remaining parts of the buffer
-
- int sentBytes = args.BytesTransferred;
+ switch (args.SocketError)
+ {
+ case SocketError.Success:
+ Memory<byte> transmissionBuffer = args.MemoryBuffer;
+ int remainingBytes = transmissionBuffer.Length;
- args.SetBuffer(sentBytes, remainingBytes - sentBytes);
+ if (args.BytesTransferred == remainingBytes)
+ {
+ // buffer was fully sent
- Connection.SendAsync(args);
- }
- else
- {
- // no bytes were sent, remote socket is dead
+ TransmissionResult result = new TransmissionResult(in args);
- sendToken.CompletionSource.SetException(new SocketException((int)SocketError.HostDown));
+ sendToken.CompletionSource.SetResult(result);
+ }
+ else if (remainingBytes > args.BytesTransferred && args.BytesTransferred > 0)
+ {
+ // send the remaining parts of the buffer
- TransmissionArgsPool.Return(args);
- }
+ int sentBytes = args.BytesTransferred;
- break;
+ args.SetBuffer(sentBytes, remainingBytes - sentBytes);
- case SocketError.OperationAborted:
- TransmissionArgsPool.Return(args);
+ Connection.SendAsync(args);
+ return;
+ }
+ else
+ {
+ // no bytes were sent, remote socket is dead
- break;
+ sendToken.CompletionSource.SetException(new SocketException((int)SocketError.HostDown));
+ }
- default:
- sendToken.CompletionSource.SetException(new SocketException((int)args.SocketError));
+ break;
- TransmissionArgsPool.Return(args);
+ case SocketError.OperationAborted:
+ break;
- break;
- }
+ default:
+ sendToken.CompletionSource.SetException(new SocketException((int)args.SocketError));
+ break;
}
+
+ BufferPool.Return(sendToken.RentedBuffer, true);
+ TransmissionArgsPool.Return(args);
}
/// <inheritdoc />
@@ -295,7 +279,7 @@ namespace NetSharp.Sockets.Stream
SocketAsyncEventArgs args = TransmissionArgsPool.Rent();
args.DisconnectReuseSocket = allowSocketReuse;
- args.UserToken = new AsyncOperationToken(in tcs, CancellationToken.None);
+ args.UserToken = new AsyncOperationToken(in tcs, in cancellationToken);
// TODO find out why the fricc we leak memory
CancellationTokenRegistration cancellationRegistration =
@@ -340,23 +324,30 @@ namespace NetSharp.Sockets.Stream
args.SetBuffer(receiveBuffer);
args.SocketFlags = flags;
- args.UserToken = new AsyncTransmissionToken(in tcs, in args, CancellationToken.None);
+ args.UserToken = new AsyncReceiveToken(in tcs, in cancellationToken);
- // TODO find out why the fricc we leak memory
- CancellationTokenRegistration cancellationRegistration =
- cancellationToken.Register(CancelAsyncTransmissionCallback, args);
+ if (cancellationToken == default)
+ {
+ if (Connection.ReceiveAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task);
+ }
+ else
+ {
+ // TODO find out why the fricc we leak memory
+ CancellationTokenRegistration cancellationRegistration =
+ cancellationToken.Register(CancelAsyncTransmissionCallback, args);
- if (Connection.ReceiveAsync(args))
- return new ValueTask<TransmissionResult>(
- tcs.Task.ContinueWith((task, state) =>
- {
- ((CancellationTokenRegistration)state).Dispose();
+ if (Connection.ReceiveAsync(args))
+ return new ValueTask<TransmissionResult>(
+ tcs.Task.ContinueWith((task, state) =>
+ {
+ ((CancellationTokenRegistration)state).Dispose();
- return task.Result;
- }, cancellationRegistration, CancellationToken.None)
- );
+ return task.Result;
+ }, cancellationRegistration, CancellationToken.None)
+ );
- cancellationRegistration.Dispose();
+ cancellationRegistration.Dispose();
+ }
TransmissionResult result = new TransmissionResult(in args);
@@ -378,32 +369,42 @@ namespace NetSharp.Sockets.Stream
return new TransmissionResult(in buffer, in sentBytes, Connection.RemoteEndPoint);
}
- public ValueTask<TransmissionResult> SendAsync(Memory<byte> sendBuffer, SocketFlags flags = SocketFlags.None, CancellationToken cancellationToken = default)
+ public ValueTask<TransmissionResult> SendAsync(ReadOnlyMemory<byte> sendBuffer, SocketFlags flags = SocketFlags.None, CancellationToken cancellationToken = default)
{
TaskCompletionSource<TransmissionResult> tcs = new TaskCompletionSource<TransmissionResult>();
SocketAsyncEventArgs args = TransmissionArgsPool.Rent();
+ byte[] transmissionBuffer = BufferPool.Rent(sendBuffer.Length);
- args.SetBuffer(sendBuffer);
+ sendBuffer.CopyTo(transmissionBuffer);
+
+ args.SetBuffer(transmissionBuffer);
args.SocketFlags = flags;
- args.UserToken = new AsyncTransmissionToken(in tcs, in args, CancellationToken.None);
+ args.UserToken = new AsyncSendToken(in tcs, in transmissionBuffer, in cancellationToken);
- // TODO find out why the fricc we leak memory
- CancellationTokenRegistration cancellationRegistration =
- cancellationToken.Register(CancelAsyncTransmissionCallback, args);
+ if (cancellationToken == default)
+ {
+ if (Connection.SendAsync(args)) return new ValueTask<TransmissionResult>(tcs.Task);
+ }
+ else
+ {
+ // TODO find out why the fricc we leak memory
+ CancellationTokenRegistration cancellationRegistration =
+ cancellationToken.Register(CancelAsyncTransmissionCallback, args);
- if (Connection.SendToAsync(args))
- return new ValueTask<TransmissionResult>(
- tcs.Task.ContinueWith((task, state) =>
- {
- ((CancellationTokenRegistration)state).Dispose();
+ if (Connection.SendAsync(args))
+ return new ValueTask<TransmissionResult>(
+ tcs.Task.ContinueWith((task, state) =>
+ {
+ ((CancellationTokenRegistration)state).Dispose();
- return task.Result;
- }, cancellationRegistration, CancellationToken.None)
- );
+ return task.Result;
+ }, cancellationRegistration, CancellationToken.None)
+ );
- cancellationRegistration.Dispose();
+ cancellationRegistration.Dispose();
+ }
TransmissionResult result = new TransmissionResult(in args);
diff --git a/NetSharp/NetSharpExamples/Benchmarks/UdpSocketClientAsyncBenchmark.cs b/NetSharp/NetSharpExamples/Benchmarks/UdpSocketClientAsyncBenchmark.cs
@@ -35,9 +35,9 @@ namespace NetSharpExamples.Benchmarks
while (!cancellationToken.IsCancellationRequested)
{
- server.ReceiveFrom(transmissionBuffer, ref remoteEndPoint);
+ int received = server.ReceiveFrom(transmissionBuffer, ref remoteEndPoint);
- server.SendTo(transmissionBuffer, remoteEndPoint);
+ int sent = server.SendTo(transmissionBuffer, remoteEndPoint);
}
server.Close();
@@ -55,7 +55,7 @@ namespace NetSharpExamples.Benchmarks
Console.WriteLine($"{PacketCount} packets will be sent per client. This could take a long time (maybe more than a minute)!");
}
- using CancellationTokenSource serverCts = new CancellationTokenSource(); serverCts.Cancel();
+ using CancellationTokenSource serverCts = new CancellationTokenSource();
Task serverTask = Task.Factory.StartNew(state => ServerTask((CancellationToken)state), serverCts.Token, TaskCreationOptions.LongRunning);
BenchmarkHelper benchmarkHelper = new BenchmarkHelper();
