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 1 commit 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 all commits
Commits
Show all changes
1 commit
Select commit
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);
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;
_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;
_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);
} 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
144 changes: 91 additions & 53 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,64 @@ 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)))
try
{
try
{
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)
{
try { stream.Dispose(); } catch { }
throw;
}
catch when (state >= 3)
var readOperation = stream.BeginRead(buffer, offset, count, null, null);
WaitHandle.WaitAny([readOperation.AsyncWaitHandle, cancellationToken.WaitHandle], timeout);

if (!readOperation.IsCompleted)
{
// a timeout or operation cancelled exception will be thrown instead
try
{
stream.Dispose();
stream.EndRead(readOperation);
}
catch
{
// ignore any exceptions
}

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

if (state == 3) { throw new TimeoutException(); }
if (state == 4) { throw new OperationCanceledException(); }
return stream.EndRead(readOperation);
}
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.

}
}

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);

return bytesRead;
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
{
stream.Dispose();
await readTask.ConfigureAwait(false);
}
catch
{
try { stream.Dispose(); } catch { } // disposing the stream aborts the read attempt
// 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 +102,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 +112,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 +123,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 +173,64 @@ public static async Task ReadBytesAsync(this Stream stream, IByteBuffer buffer,
}
}

public static void Write(this Stream stream, byte[] buffer, int offset, int count, TimeSpan timeout, CancellationToken cancellationToken)
{
try
{
var writeOperation = stream.BeginWrite(buffer, offset, count, null, null);
WaitHandle.WaitAny([writeOperation.AsyncWaitHandle, cancellationToken.WaitHandle], timeout);

if (!writeOperation.IsCompleted)
{
try
{
stream.Dispose();
stream.EndWrite(writeOperation);
}
catch
{
// ignore any exceptions
}

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

stream.EndWrite(writeOperation);
}
catch (ObjectDisposedException ex)
{
throw new EndOfStreamException("The connection was interrupted.", ex);
}
}

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);

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

using (new Timer(_ => ChangeState(3), null, timeout, Timeout.InfiniteTimeSpan))
using (cancellationToken.Register(() => ChangeState(4)))
if (!writeTask.IsCompleted)
{
try
{
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)
{
try { stream.Dispose(); } catch { }
throw;
stream.Dispose();
await writeTask.ConfigureAwait(false);
}
catch when (state >= 3)
catch
{
// a timeout or operation cancelled exception will be thrown instead
// ignore any exceptions
}

if (state == 3) { throw new TimeoutException(); }
if (state == 4) { throw new OperationCanceledException(); }
cancellationToken.ThrowIfCancellationRequested();
throw new TimeoutException();
}

void ChangeState(int to)
{
var from = Interlocked.CompareExchange(ref state, to, 1);
if (from == 1 && to >= 3)
{
try { stream.Dispose(); } catch { } // disposing the stream aborts the write attempt
}
}
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 +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?
offset += bytesToWrite;
count -= bytesToWrite;
}
Expand Down
Loading