problem: no way to cancel receive operation #876
Conversation
| { | ||
| if (TryReceiveBytes(socket, out var bytes)) | ||
| return new ValueTask<byte[]>(bytes); | ||
|
|
||
| // TODO: this is a hack, eventually we need kind of IO ThreadPool for thread-safe socket to wait on asynchronously | ||
| // and probably implement IValueTaskSource | ||
| return new ValueTask<byte[]>(Task.Factory.StartNew(socket.ReceiveBytes, TaskCreationOptions.LongRunning)); | ||
| // TODO: should we avoid lambda here as it cause heap allocation for the environment? | ||
| return new ValueTask<byte[]>(Task.Factory.StartNew(() => socket.ReceiveBytes(cancellationToken), |
There was a problem hiding this comment.
@drewnoakes do you think we would benefit here from de-lambda the action? using a real method with the cancellation token passed as a state object?
There was a problem hiding this comment.
It would avoid a heap allocation for the closure object, yes. You'd need to pass both the socket and token as state, then cast them at the other side.
I played around with this a little:
It'd be interesting to use BenchmarkDotNet to get some data on allocations for send/receive in various scenarios.
There was a problem hiding this comment.
If I have to wrap both the socket and the token I'm not sure I have a benefit of using state instead of lambda.
There was a problem hiding this comment.
With extension method seems interesting:
https://sharplab.io/#gist:e7e76a5ece9d763149b8fb1eeed2587b
The question is, what is __ldftn is doing...
There was a problem hiding this comment.
seems to be allocation free:
https://docs.microsoft.com/en-us/dotnet/api/system.reflection.emit.opcodes.ldftn?view=netcore-3.1
There was a problem hiding this comment.
solved it with private extension method.
There was a problem hiding this comment.
Actually casting CancellationToken as object will box, which allocates. I'm not sure there's a way around allocating here. I'd wondered about making the method async ValueTask<> but didn't test it.
There was a problem hiding this comment.
The main thing is that these methods don't allocate when they complete synchronously. If they need to yield, ValueTask<> will allocate anyway.
There was a problem hiding this comment.
Yes, you are right. I will simplify it back and merge it.
| @@ -993,7 +995,7 @@ public bool TrySend(ref Msg msg, TimeSpan timeout, bool more) | |||
| /// </remarks> | |||
| /// <exception cref="FaultException">the Msg must already have been uninitialised</exception> | |||
| /// <exception cref="TerminatingException">The socket must not already be stopped.</exception> | |||
There was a problem hiding this comment.
Would be good to document OperationCanceledException here too, and a few other places in this file.
There was a problem hiding this comment.
It is not throwing, I only throw from non-Try methods. The Try methods return false.
| { | ||
| if (TryReceiveBytes(socket, out var bytes)) | ||
| return new ValueTask<byte[]>(bytes); | ||
|
|
||
| // TODO: this is a hack, eventually we need kind of IO ThreadPool for thread-safe socket to wait on asynchronously | ||
| // and probably implement IValueTaskSource | ||
| return new ValueTask<byte[]>(Task.Factory.StartNew(socket.ReceiveBytes, TaskCreationOptions.LongRunning)); | ||
| // TODO: should we avoid lambda here as it cause heap allocation for the environment? | ||
| return new ValueTask<byte[]>(Task.Factory.StartNew(() => socket.ReceiveBytes(cancellationToken), |
There was a problem hiding this comment.
It would avoid a heap allocation for the closure object, yes. You'd need to pass both the socket and token as state, then cast them at the other side.
I played around with this a little:
It'd be interesting to use BenchmarkDotNet to get some data on allocations for send/receive in various scenarios.
Solution: add cancellation token to all receive operation (thread safe only, for now)
somdoron
force-pushed
the
thread_safe
branch
2 times, most recently
from
83f7173 to
2658130
Compare
|
@drewnoakes next is IAsyncEnumerable, I wonder if you know of a convention to get the IAsyncEnumerable. One option is for the socket itself to implement it, however, that means we stick to one message type (byte[] probably). BlockingCollection has a method called
Anyway, not a big fan of it either. What do you think? |
|
I haven't had much hands on experience with IAsyncEnumerable yet so don't have any strong opinions. One naming idea to consider is maintaining prefix consistency to help discoverability via IntelliSense. For example:
|
Solution: add cancellation token to all receive operation (thread safe only, for now)