-
Notifications
You must be signed in to change notification settings - Fork 1.3k
CSHARP-3662: MongoClientSettings.SocketTimeout does not work for values under 500ms on Windows for sync code #1690
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Conversation
…es under 500ms on Windows for sync code
} | ||
catch (ObjectDisposedException ex) | ||
{ | ||
throw new EndOfStreamException("The connection was interrupted.", ex); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Have to catch ObjectDisposedException
and throw something that will make WrapExceptionIfRequired
to throw MongoConnectionException
. Otherwise we will fail to re-try on connection pool closing in-use connections.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How would we ever reach this line?
Any ObjectDisposedException
thrown by line 50 is discarded by line 55.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is another code path. We could get ObjectDisposedException
here when connection pool was stopped with close in-use connection option. In such case connection pool will dispose all connections.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've re-wrote the code to avoid the confusion.
try | ||
{ | ||
socket.Dispose(); | ||
await connectTask.ConfigureAwait(false); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This probably deserves a comment, as it might be counter-intuitive. Same for sync version.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Agreed.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Will add the comment. But the reason to await here is to catch the ObjectDisposedException to avoid triggering UnobservedException.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Comments were added.
} | ||
catch when (state >= 3) | ||
var readOperation = stream.BeginRead(buffer, offset, count, null, null); | ||
WaitHandle.WaitAny([readOperation.AsyncWaitHandle, cancellationToken.WaitHandle], timeout); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Evaluate the option of using ManualResetEventSlim
instead.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
try | ||
{ | ||
stream.Dispose(); | ||
await readTask.ConfigureAwait(false); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can waiting for readTask
be blocking for long and cause significant deviation from timeout?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It should not. Because stream.Dispose closes socket, what should abort the task.
@@ -204,7 +242,7 @@ public static void WriteBytes(this Stream stream, IByteBuffer buffer, int offset | |||
cancellationToken.ThrowIfCancellationRequested(); | |||
var backingBytes = buffer.AccessBackingBytes(offset); | |||
var bytesToWrite = Math.Min(count, backingBytes.Count); | |||
stream.Write(backingBytes.Array, backingBytes.Offset, bytesToWrite); // TODO: honor cancellationToken? | |||
stream.Write(backingBytes.Array, backingBytes.Offset, bytesToWrite, timeout, cancellationToken); // TODO: honor cancellationToken? |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Remove the TODO comment
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
|
||
public static async Task WriteAsync(this Stream stream, byte[] buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken) | ||
{ | ||
var state = 1; // 1 == writing, 2 == done writing, 3 == timedout, 4 == cancelled | ||
var timeoutTask = Task.Delay(timeout); | ||
var writeTask = stream.WriteAsync(buffer, offset, count, cancellationToken); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
cancellationToken
is passed in WriteAsync
but not in ReadAsync
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed.
.Returns(task); | ||
mockStream | ||
.Setup(s => s.EndRead(It.IsAny<IAsyncResult>())) | ||
.Returns<IAsyncResult>(x => ((Task<int>)x).GetAwaiter().GetResult()); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good opportunity to switch to Record.ExceptionAsync
below.
.Setup(s => s.Read(It.IsAny<byte[]>(), It.IsAny<int>(), It.IsAny<int>())) | ||
.Throws(new SocketException()); | ||
.Setup(s => s.BeginRead(It.IsAny<byte[]>(), It.IsAny<int>(), It.IsAny<int>(), null, null)) | ||
.Returns(task); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does this mimic BeginRead
that is implemented with Tasks?
If so, it that's not probably case for NetworkStream
, at least for older frameworks.
Are we ok with loosing .Throws(new SocketException());
test case?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It will throw the same SocketException. Because the task was created from the exception. See above.
Exception exception; | ||
if (async) | ||
{ | ||
exception = Record.Exception(() => _subject.ReceiveMessageAsync(1, encoderSelector, _messageEncoderSettings, CancellationToken.None).GetAwaiter().GetResult()); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ExceptionAsync
?
} | ||
catch (ObjectDisposedException ex) | ||
{ | ||
throw new EndOfStreamException("The connection was interrupted.", ex); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How would we ever reach this line?
Any ObjectDisposedException
thrown by line 50 is discarded by line 55.
@@ -34,7 +34,7 @@ public void Compress(Stream input, Stream output) | |||
{ | |||
var uncompressedSize = (int)(input.Length - input.Position); | |||
var uncompressedBytes = new byte[uncompressedSize]; // does not include uncompressed message headers | |||
input.ReadBytes(uncompressedBytes, offset: 0, count: uncompressedSize, CancellationToken.None); | |||
input.ReadBytes(uncompressedBytes, offset: 0, count: uncompressedSize, Timeout.InfiniteTimeSpan, CancellationToken.None); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Curious why this change is needed.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Because this code uses helper method that I've adjusted with new parameter, so I should pass something from here too.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Will stream.ReadTimeout
still be respected by BeginRead
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
As far as I understood Compress/Decompress works only with in-memory kind of streams. Not sure if timeout makes sense in such case.
@@ -333,14 +333,15 @@ private IByteBuffer ReceiveBuffer(CancellationToken cancellationToken) | |||
try | |||
{ | |||
var messageSizeBytes = new byte[4]; | |||
_stream.ReadBytes(messageSizeBytes, 0, 4, cancellationToken); | |||
var readTimeout = _stream.CanTimeout ? TimeSpan.FromMilliseconds(_stream.ReadTimeout) : Timeout.InfiniteTimeSpan; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Seems weird to be getting the readTimeout
from the Stream
. Shouldn't it come from settings somewhere?
Also, why do we care about _stream.CanTimeout
? The new code will support timeouts on any Stream
no matter what the value of _stream.CanTimeout
is.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This code is inhereted/copied from the RecieveBufferAsync
. I can double-check when/which streams could have CanTimeout = false.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I agree that _stream.CanTimeout
seems irrelevant here.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ok. Should I remove checking the _stream.CanTimeout
for both sync and async versions? Async version used to have that check before.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, I think we can ignore this now as we are doing our own timeouts.
@@ -535,7 +536,8 @@ private void SendBuffer(IByteBuffer buffer, CancellationToken cancellationToken) | |||
|
|||
try | |||
{ | |||
_stream.WriteBytes(buffer, 0, buffer.Length, cancellationToken); | |||
var writeTimeout = _stream.CanTimeout ? TimeSpan.FromMilliseconds(_stream.WriteTimeout) : Timeout.InfiniteTimeSpan; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Similar question here as for readTimeout
above.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Reasonable changes. My major concern is accessing objects after disposal, which happens in a few places. This will typically raise an ObjectDisposedException
. Are these calls reversed intentionally (e.g. dispose then a further cleanup call) and if so why?
try | ||
{ | ||
socket.Dispose(); | ||
socket.EndConnect(connectOperation); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Shouldn't these calls be reversed? You shouldn't be calling methods on an object after disposal.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've removed EndConnect call. We do have to await on task returned for async code path to observe the exception. I was under impression that same logic is applicable for Begin/End approach, but it is not.
try | ||
{ | ||
socket.Dispose(); | ||
await connectTask.ConfigureAwait(false); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Agreed.
try | ||
{ | ||
stream.Dispose(); | ||
stream.EndRead(readOperation); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Same comment here about using an object after disposal.
{ | ||
try | ||
using var manualResetEvent = new ManualResetEventSlim(); | ||
readOperation = stream.BeginRead(buffer, offset, count, state => |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Consider placing each argument in a separate line.
Also just state => ((ManualResetEventSlim)state.AsyncState).Set()
is sufficient.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
|
||
namespace MongoDB.Driver.Core.Misc | ||
{ | ||
public class TaskExtensionsTests |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This test collection testing our pollyfil of Task.WaitAsync
for net472 and netstandard2.1. Technically it do not have to run for net6, but I decided to run it for now, to demonstrate that our implementation behave the same way as native.
@@ -333,14 +333,15 @@ private IByteBuffer ReceiveBuffer(CancellationToken cancellationToken) | |||
try | |||
{ | |||
var messageSizeBytes = new byte[4]; | |||
_stream.ReadBytes(messageSizeBytes, 0, 4, cancellationToken); | |||
var readTimeout = _stream.CanTimeout ? TimeSpan.FromMilliseconds(_stream.ReadTimeout) : Timeout.InfiniteTimeSpan; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I agree that _stream.CanTimeout
seems irrelevant here.
} | ||
catch | ||
{ | ||
// ignore any exceptions |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: Capital i
for consistency.
await HandleTaskCancellation().ConfigureAwait(false); | ||
throw; | ||
} | ||
catch (TimeoutException) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why not: catch (Exception e) when (e is OperationCanceledException || e is TimeoutException)
And no need for HandleTaskCancellation
then.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I was about to propose the same thing, slightly differently:
catch (Exception ex) when (ex is OperationCanceledException or TimeoutException)
var readTask = stream.ReadAsync(buffer, offset, count); | ||
try | ||
{ | ||
return await readTask.WaitAsync(timeout, cancellationToken).ConfigureAwait(false); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Nice.
{ | ||
try | ||
using var manualResetEvent = new ManualResetEventSlim(); | ||
writeOperation = stream.BeginWrite(buffer, offset, count, state => |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Not need for brackets:
writeOperation = stream.BeginWrite(buffer, offset, count, state => ((ManualResetEventSlim)state.AsyncState).Set(), manualResetEvent);
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
var bytesRead = 0; | ||
using (new Timer(_ => ChangeState(3), null, timeout, Timeout.InfiniteTimeSpan)) | ||
using (cancellationToken.Register(() => ChangeState(4))) | ||
IAsyncResult readOperation; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can define readOperation
inside the scope (write as well)?
var readOperation = ...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done
} | ||
catch when (state == 1) | ||
catch (OperationCanceledException) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Would handling +
in the outer catch be readable/result in shorter code?
} | ||
catch (ObjectDisposedException ex) | ||
{ | ||
throw new EndOfStreamException("The connection was interrupted.", ex); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
So we don't have EndOfStreamException
in async case?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good question. Will double-check.
} | ||
} | ||
|
||
#endif |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
minor: Add and empty line after endif
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
And remove the empty line before it?
await HandleTaskCancellation().ConfigureAwait(false); | ||
throw; | ||
} | ||
catch (TimeoutException) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I was about to propose the same thing, slightly differently:
catch (Exception ex) when (ex is OperationCanceledException or TimeoutException)
await HandleTaskCancellation().ConfigureAwait(false); | ||
throw; | ||
} | ||
catch (TimeoutException) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Same comment as for ReadAsync
:
catch (Exception ex) when (ex is OperationCanceledException or TimeoutException)
|
||
public static async Task WaitAsync(this Task task, TimeSpan timeout, CancellationToken cancellationToken) | ||
{ | ||
ValidateTimeout(timeout); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
These extension methods are nice!
But there is a subtle issue here that we may or may not want to address.
The way this is currently coded when passing an invalid timeout the exception will NOT be thrown immediately. Instead a Task will be returned and the exception will be thrown later and observed by the code that awaits the Task.
To immediately throw an exception without returning a Task requires some tricks involving an async helper method.
I leave it up to you to decide if you want to be more precise about when the exception is thrown.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For one example of how to throw an exception immediately instead of returning a Task see:
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Which in turn links to:
which has an example of using a static local method to separate the sync argument checking from the async part that does the work. This looks like a nice approach.
@@ -333,14 +333,15 @@ private IByteBuffer ReceiveBuffer(CancellationToken cancellationToken) | |||
try | |||
{ | |||
var messageSizeBytes = new byte[4]; | |||
_stream.ReadBytes(messageSizeBytes, 0, 4, cancellationToken); | |||
var readTimeout = _stream.CanTimeout ? TimeSpan.FromMilliseconds(_stream.ReadTimeout) : Timeout.InfiniteTimeSpan; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, I think we can ignore this now as we are doing our own timeouts.
|
||
await task.WaitAsync(Timeout.InfiniteTimeSpan, CancellationToken.None); | ||
|
||
task.IsCompleted.Should().BeTrue(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think I understand why you duplicated this test, but I think it's ok to just have one:
WaitAsync_should_work_for_task_with_inifite_timeout
.
|
||
public static async Task WaitAsync(this Task task, TimeSpan timeout, CancellationToken cancellationToken) | ||
{ | ||
ValidateTimeout(timeout); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
|
||
public static async Task WaitAsync(this Task task, TimeSpan timeout, CancellationToken cancellationToken) | ||
{ | ||
ValidateTimeout(timeout); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Which in turn links to:
which has an example of using a static local method to separate the sync argument checking from the async part that does the work. This looks like a nice approach.
} | ||
} | ||
|
||
#endif |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
And remove the empty line before it?
throw new TimeoutException(); | ||
} | ||
|
||
private static void ValidateTimeout(TimeSpan timeout) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You can keep this name if you want, but normally we would have named a method like this EnsureTimeoutIsValid
.
No description provided.