Skip to content

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.

Sign up for GitHub

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

Open
wants to merge 8 commits into
base: main
Choose a base branch
from
Open

CSHARP-3662: MongoClientSettings.SocketTimeout does not work for values under 500ms on Windows for sync code #1690

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
e60210b
CSHARP-3662: MongoClientSettings.SocketTimeout does not work for valu…
May 9, 2025
ea9d7d1
PR
May 23, 2025
f276888
Fix tests
May 23, 2025
e7095f9
PR
May 27, 2025
fe81697
pr
May 27, 2025
8f3f0ac
Use Task.WaitAsync on async code-path.
May 28, 2025
940c764
PR
May 28, 2025
fdd35f5
PR
May 29, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/MongoDB.Driver/Core/Compression/SnappyCompressor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Copy link
Contributor

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.

Copy link
Member Author

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.

Copy link
Contributor

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?

Copy link
Member Author

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.

var maxCompressedSize = Snappy.GetMaxCompressedLength(uncompressedSize);
var compressedBytes = new byte[maxCompressedSize];
var compressedSize = Snappy.Compress(uncompressedBytes, compressedBytes);
Expand All @@ -50,7 +50,7 @@ public void Decompress(Stream input, Stream output)
{
var compressedSize = (int)(input.Length - input.Position);
var compressedBytes = new byte[compressedSize];
input.ReadBytes(compressedBytes, offset: 0, count: compressedSize, CancellationToken.None);
input.ReadBytes(compressedBytes, offset: 0, count: compressedSize, Timeout.InfiniteTimeSpan, CancellationToken.None);
var uncompressedSize = Snappy.GetUncompressedLength(compressedBytes);
var decompressedBytes = new byte[uncompressedSize];
var decompressedSize = Snappy.Decompress(compressedBytes, decompressedBytes);
Expand Down
8 changes: 5 additions & 3 deletions src/MongoDB.Driver/Core/Connections/BinaryConnection.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Copy link
Contributor

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.

Copy link
Member Author

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.

Copy link
Contributor

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.

Copy link
Member Author

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.

Copy link
Contributor

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.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As it was discussed offline - will keep it for now. It will be replaced by CSOT timeout on the later stages of implementation.

_stream.ReadBytes(messageSizeBytes, 0, 4, readTimeout, cancellationToken);
var messageSize = BinaryPrimitives.ReadInt32LittleEndian(messageSizeBytes);
EnsureMessageSizeIsValid(messageSize);
var inputBufferChunkSource = new InputBufferChunkSource(BsonChunkPool.Default);
var buffer = ByteBufferFactory.Create(inputBufferChunkSource, messageSize);
buffer.Length = messageSize;
buffer.SetBytes(0, messageSizeBytes, 0, 4);
_stream.ReadBytes(buffer, 4, messageSize - 4, cancellationToken);
_stream.ReadBytes(buffer, 4, messageSize - 4, readTimeout, cancellationToken);
_lastUsedAtUtc = DateTime.UtcNow;
buffer.MakeReadOnly();
return buffer;
Expand Down Expand Up @@ -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;
Copy link
Contributor

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.

_stream.WriteBytes(buffer, 0, buffer.Length, writeTimeout, cancellationToken);
_lastUsedAtUtc = DateTime.UtcNow;
}
catch (Exception ex)
Expand Down
12 changes: 10 additions & 2 deletions src/MongoDB.Driver/Core/Connections/TcpStreamFactory.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,11 @@ private void Connect(Socket socket, EndPoint endPoint, CancellationToken cancell

if (!connectOperation.IsCompleted)
{
try { socket.Dispose(); } catch { }
try
{
socket.Dispose();
socket.EndConnect(connectOperation);
Copy link
Contributor

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.

Copy link
Member Author

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.

} catch { }

cancellationToken.ThrowIfCancellationRequested();
throw new TimeoutException($"Timed out connecting to {endPoint}. Timeout was {_settings.ConnectTimeout}.");
Expand All @@ -164,7 +168,11 @@ private async Task ConnectAsync(Socket socket, EndPoint endPoint, CancellationTo

if (!connectTask.IsCompleted)
{
try { socket.Dispose(); } catch { }
try
{
socket.Dispose();
await connectTask.ConfigureAwait(false);
} catch { }

cancellationToken.ThrowIfCancellationRequested();
throw new TimeoutException($"Timed out connecting to {endPoint}. Timeout was {_settings.ConnectTimeout}.");
Expand Down
160 changes: 111 additions & 49 deletions src/MongoDB.Driver/Core/Misc/StreamExtensionMethods.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,46 +36,76 @@ public static void EfficientCopyTo(this Stream input, Stream output)
}
}

public static async Task<int> ReadAsync(this Stream stream, byte[] buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
public static int Read(this Stream stream, byte[] buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
{
var state = 1; // 1 == reading, 2 == done reading, 3 == timedout, 4 == cancelled

var bytesRead = 0;
using (new Timer(_ => ChangeState(3), null, timeout, Timeout.InfiniteTimeSpan))
using (cancellationToken.Register(() => ChangeState(4)))
IAsyncResult readOperation;
try
{
try
using var manualResetEvent = new ManualResetEventSlim();
readOperation = stream.BeginRead(buffer, offset, count, state =>
{
bytesRead = await stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
ChangeState(2); // note: might not actually go to state 2 if already in state 3 or 4
}
catch when (state == 1)
((ManualResetEventSlim)state.AsyncState).Set();
}, manualResetEvent);

try
{
try { stream.Dispose(); } catch { }
throw;
if (readOperation.IsCompleted || manualResetEvent.Wait(timeout, cancellationToken))
{
return stream.EndRead(readOperation);
}
}
catch when (state >= 3)
catch (OperationCanceledException)
{
// a timeout or operation cancelled exception will be thrown instead
// Have to suppress OperationCanceledException here, it will be thrown after the stream will be disposed.
}
}
catch (ObjectDisposedException ex)
{
throw new EndOfStreamException("The connection was interrupted.", ex);
Copy link
Member Author

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.

Copy link
Contributor

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.

Copy link
Member Author

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.

Copy link
Member Author

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.

}

if (state == 3) { throw new TimeoutException(); }
if (state == 4) { throw new OperationCanceledException(); }
try
{
stream.Dispose();
stream.EndRead(readOperation);
}
catch
{
// ignore any exceptions
Copy link
Contributor

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.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

}

return bytesRead;
cancellationToken.ThrowIfCancellationRequested();
throw new TimeoutException();
}

public static async Task<int> ReadAsync(this Stream stream, byte[] buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
{
var timeoutTask = Task.Delay(timeout, cancellationToken);
var readTask = stream.ReadAsync(buffer, offset, count);

await Task.WhenAny(readTask, timeoutTask).ConfigureAwait(false);

void ChangeState(int to)
if (!readTask.IsCompleted)
{
var from = Interlocked.CompareExchange(ref state, to, 1);
if (from == 1 && to >= 3)
try
{
try { stream.Dispose(); } catch { } // disposing the stream aborts the read attempt
stream.Dispose();
// should await in the read task to avoid UnobservedTaskException
await readTask.ConfigureAwait(false);
}
catch
{
// ignore any exceptions
}

cancellationToken.ThrowIfCancellationRequested();
throw new TimeoutException();
}

return await readTask.ConfigureAwait(false);
}

public static void ReadBytes(this Stream stream, byte[] buffer, int offset, int count, CancellationToken cancellationToken)
public static void ReadBytes(this Stream stream, byte[] buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
{
Ensure.IsNotNull(stream, nameof(stream));
Ensure.IsNotNull(buffer, nameof(buffer));
Expand All @@ -84,7 +114,7 @@ public static void ReadBytes(this Stream stream, byte[] buffer, int offset, int

while (count > 0)
{
var bytesRead = stream.Read(buffer, offset, count); // TODO: honor cancellationToken?
var bytesRead = stream.Read(buffer, offset, count, timeout, cancellationToken);
if (bytesRead == 0)
{
throw new EndOfStreamException();
Expand All @@ -94,7 +124,7 @@ public static void ReadBytes(this Stream stream, byte[] buffer, int offset, int
}
}

public static void ReadBytes(this Stream stream, IByteBuffer buffer, int offset, int count, CancellationToken cancellationToken)
public static void ReadBytes(this Stream stream, IByteBuffer buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
{
Ensure.IsNotNull(stream, nameof(stream));
Ensure.IsNotNull(buffer, nameof(buffer));
Expand All @@ -105,7 +135,7 @@ public static void ReadBytes(this Stream stream, IByteBuffer buffer, int offset,
{
var backingBytes = buffer.AccessBackingBytes(offset);
var bytesToRead = Math.Min(count, backingBytes.Count);
var bytesRead = stream.Read(backingBytes.Array, backingBytes.Offset, bytesToRead); // TODO: honor cancellationToken?
var bytesRead = stream.Read(backingBytes.Array, backingBytes.Offset, bytesToRead, timeout, cancellationToken);
if (bytesRead == 0)
{
throw new EndOfStreamException();
Expand Down Expand Up @@ -155,44 +185,76 @@ public static async Task ReadBytesAsync(this Stream stream, IByteBuffer buffer,
}
}


public static async Task WriteAsync(this Stream stream, byte[] buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
public static void Write(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

using (new Timer(_ => ChangeState(3), null, timeout, Timeout.InfiniteTimeSpan))
using (cancellationToken.Register(() => ChangeState(4)))
IAsyncResult writeOperation;
try
{
try
using var manualResetEvent = new ManualResetEventSlim();
writeOperation = stream.BeginWrite(buffer, offset, count, state =>
{
await stream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
ChangeState(2); // note: might not actually go to state 2 if already in state 3 or 4
}
catch when (state == 1)
((ManualResetEventSlim)state.AsyncState).Set();
}, manualResetEvent);

try
{
try { stream.Dispose(); } catch { }
throw;
if (writeOperation.IsCompleted || manualResetEvent.Wait(timeout, cancellationToken))
{
stream.EndWrite(writeOperation);
return;
}
}
catch when (state >= 3)
catch (OperationCanceledException)
{
// a timeout or operation cancelled exception will be thrown instead
// Have to suppress OperationCanceledException here, it will be thrown after the stream will be disposed.
}
}
catch (ObjectDisposedException ex)
{
throw new EndOfStreamException("The connection was interrupted.", ex);
Copy link
Contributor

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?

Copy link
Member Author

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.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As per our discussion: async code-path do not require catching ObjectDisposedException. Also we agreed to replace EndOfStreamException with IOException, to match the throwed exception from async code path.

}

if (state == 3) { throw new TimeoutException(); }
if (state == 4) { throw new OperationCanceledException(); }
try
{
stream.Dispose();
stream.EndWrite(writeOperation);
}
catch
{
// ignore any exceptions
}

void ChangeState(int to)
cancellationToken.ThrowIfCancellationRequested();
throw new TimeoutException();
}

public static async Task WriteAsync(this Stream stream, byte[] buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
{
var timeoutTask = Task.Delay(timeout, cancellationToken);
var writeTask = stream.WriteAsync(buffer, offset, count);

await Task.WhenAny(writeTask, timeoutTask).ConfigureAwait(false);

if (!writeTask.IsCompleted)
{
var from = Interlocked.CompareExchange(ref state, to, 1);
if (from == 1 && to >= 3)
try
{
stream.Dispose();
await writeTask.ConfigureAwait(false);
}
catch
{
try { stream.Dispose(); } catch { } // disposing the stream aborts the write attempt
// ignore any exceptions
}

cancellationToken.ThrowIfCancellationRequested();
throw new TimeoutException();
}

await writeTask.ConfigureAwait(false);
}

public static void WriteBytes(this Stream stream, IByteBuffer buffer, int offset, int count, CancellationToken cancellationToken)
public static void WriteBytes(this Stream stream, IByteBuffer buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
{
Ensure.IsNotNull(stream, nameof(stream));
Ensure.IsNotNull(buffer, nameof(buffer));
Expand All @@ -204,7 +266,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?
offset += bytesToWrite;
count -= bytesToWrite;
}
Expand Down
Loading