Skip to content

CSHARP-5608: CSOT: Command Execution #1716

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

Merged
merged 5 commits into from
Jul 2, 2025
Merged

CSHARP-5608: CSOT: Command Execution #1716

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
7e00c43
CSHARP-5608: CSOT: Command Execution
Jul 1, 2025
1db6d65
PR
Jul 2, 2025
e723254
PR
Jul 2, 2025
d5004eb
PR
Jul 2, 2025
18f01ef
fix tests
Jul 2, 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
6 changes: 4 additions & 2 deletions src/MongoDB.Driver/Authentication/MongoDBX509Authenticator.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,8 @@ public void Authenticate(IConnection connection, ConnectionDescription descripti
try
{
var protocol = CreateAuthenticateProtocol();
protocol.Execute(connection, cancellationToken);
// TODO: CSOT: implement operationContext support for Auth.
protocol.Execute(new OperationContext(Timeout.InfiniteTimeSpan, cancellationToken), connection);
}
catch (MongoCommandException ex)
{
Expand All @@ -79,7 +80,8 @@ public async Task AuthenticateAsync(IConnection connection, ConnectionDescriptio
try
{
var protocol = CreateAuthenticateProtocol();
await protocol.ExecuteAsync(connection, cancellationToken).ConfigureAwait(false);
// TODO: CSOT: implement operationContext support for Auth.
await protocol.ExecuteAsync(new OperationContext(Timeout.InfiniteTimeSpan, cancellationToken), connection).ConfigureAwait(false);
}
catch (MongoCommandException ex)
{
Expand Down
6 changes: 4 additions & 2 deletions src/MongoDB.Driver/Authentication/SaslAuthenticator.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,8 @@ public void Authenticate(IConnection connection, ConnectionDescription descripti
try
{
var protocol = CreateCommandProtocol(command);
result = protocol.Execute(connection, cancellationToken);
// TODO: CSOT: implement operationContext support for Auth.
result = protocol.Execute(new OperationContext(Timeout.InfiniteTimeSpan, cancellationToken), connection);
conversationId ??= result?.GetValue("conversationId").AsInt32;
}
catch (MongoException ex)
Expand Down Expand Up @@ -172,7 +173,8 @@ public async Task AuthenticateAsync(IConnection connection, ConnectionDescriptio
try
{
var protocol = CreateCommandProtocol(command);
result = await protocol.ExecuteAsync(connection, cancellationToken).ConfigureAwait(false);
// TODO: CSOT: implement operationContext support for Auth.
result = await protocol.ExecuteAsync(new OperationContext(Timeout.InfiniteTimeSpan, cancellationToken), connection).ConfigureAwait(false);
conversationId ??= result?.GetValue("conversationId").AsInt32;
}
catch (MongoException ex)
Expand Down
9 changes: 4 additions & 5 deletions src/MongoDB.Driver/Core/Bindings/IChannel.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using MongoDB.Bson;
using MongoDB.Bson.IO;
Expand All @@ -33,6 +32,7 @@ internal interface IChannel : IDisposable
ConnectionDescription ConnectionDescription { get; }

TResult Command<TResult>(
OperationContext operationContext,
ICoreSession session,
ReadPreference readPreference,
DatabaseNamespace databaseNamespace,
Expand All @@ -43,10 +43,10 @@ TResult Command<TResult>(
Action<IMessageEncoderPostProcessor> postWriteAction,
CommandResponseHandling responseHandling,
IBsonSerializer<TResult> resultSerializer,
MessageEncoderSettings messageEncoderSettings,
CancellationToken cancellationToken);
MessageEncoderSettings messageEncoderSettings);

Task<TResult> CommandAsync<TResult>(
OperationContext operationContext,
ICoreSession session,
ReadPreference readPreference,
DatabaseNamespace databaseNamespace,
Expand All @@ -57,8 +57,7 @@ Task<TResult> CommandAsync<TResult>(
Action<IMessageEncoderPostProcessor> postWriteAction,
CommandResponseHandling responseHandling,
IBsonSerializer<TResult> resultSerializer,
MessageEncoderSettings messageEncoderSettings,
CancellationToken cancellationToken);
MessageEncoderSettings messageEncoderSettings);
}

internal interface IChannelHandle : IChannel
Expand Down
15 changes: 3 additions & 12 deletions src/MongoDB.Driver/Core/Bindings/ServerChannelSource.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,20 +35,11 @@ public ServerChannelSource(IServer server, ICoreSessionHandle session)
}

// properties
public IServer Server
{
get { return _server; }
}
public IServer Server => _server;
Copy link
Contributor

Choose a reason for hiding this comment

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

This file now only has cosmetic changes.

Up to you whether they belong in this PR or not.

Copy link
Member Author

Choose a reason for hiding this comment

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

Will keep


public ServerDescription ServerDescription
{
get { return _server.Description; }
}
public ServerDescription ServerDescription => _server.Description;

public ICoreSessionHandle Session
{
get { return _session; }
}
public ICoreSessionHandle Session => _session;

// methods
public void Dispose()
Expand Down
32 changes: 16 additions & 16 deletions src/MongoDB.Driver/Core/Clusters/Cluster.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,7 @@ public IServer SelectServer(OperationContext operationContext, IServerSelector s
Ensure.IsNotNull(operationContext, nameof(operationContext));
ThrowIfDisposedOrNotOpen();

operationContext = operationContext.WithTimeout(Settings.ServerSelectionTimeout);
using var serverSelectionOperationContext = operationContext.WithTimeout(Settings.ServerSelectionTimeout);
var expirableClusterDescription = _expirableClusterDescription;
IDisposable serverSelectionWaitQueueDisposer = null;
(selector, var operationCountSelector, var stopwatch) = BeginServerSelection(expirableClusterDescription.ClusterDescription, selector);
Expand All @@ -168,16 +168,16 @@ public IServer SelectServer(OperationContext operationContext, IServerSelector s
{
while (true)
{
var result = SelectServer(expirableClusterDescription, selector, operationCountSelector);
if (result != default)
var server = SelectServer(expirableClusterDescription, selector, operationCountSelector);
if (server != null)
{
EndServerSelection(expirableClusterDescription.ClusterDescription, selector, result.ServerDescription, stopwatch);
return result.Server;
EndServerSelection(expirableClusterDescription.ClusterDescription, selector, server.Description, stopwatch);
return server;
}

serverSelectionWaitQueueDisposer ??= _serverSelectionWaitQueue.Enter(operationContext, selector, expirableClusterDescription.ClusterDescription, EventContext.OperationId);
serverSelectionWaitQueueDisposer ??= _serverSelectionWaitQueue.Enter(serverSelectionOperationContext, selector, expirableClusterDescription.ClusterDescription, EventContext.OperationId);

operationContext.WaitTask(expirableClusterDescription.Expired);
serverSelectionOperationContext.WaitTask(expirableClusterDescription.Expired);
expirableClusterDescription = _expirableClusterDescription;
}
}
Expand All @@ -197,7 +197,7 @@ public async Task<IServer> SelectServerAsync(OperationContext operationContext,
Ensure.IsNotNull(operationContext, nameof(operationContext));
ThrowIfDisposedOrNotOpen();

operationContext = operationContext.WithTimeout(Settings.ServerSelectionTimeout);
using var serverSelectionOperationContext = operationContext.WithTimeout(Settings.ServerSelectionTimeout);
var expirableClusterDescription = _expirableClusterDescription;
IDisposable serverSelectionWaitQueueDisposer = null;
(selector, var operationCountSelector, var stopwatch) = BeginServerSelection(expirableClusterDescription.ClusterDescription, selector);
Expand All @@ -206,16 +206,16 @@ public async Task<IServer> SelectServerAsync(OperationContext operationContext,
{
while (true)
{
var result = SelectServer(expirableClusterDescription, selector, operationCountSelector);
if (result != default)
var server = SelectServer(expirableClusterDescription, selector, operationCountSelector);
if (server != null)
{
EndServerSelection(expirableClusterDescription.ClusterDescription, selector, result.ServerDescription, stopwatch);
return result.Server;
EndServerSelection(expirableClusterDescription.ClusterDescription, selector, server.Description, stopwatch);
return server;
}

serverSelectionWaitQueueDisposer ??= _serverSelectionWaitQueue.Enter(operationContext, selector, expirableClusterDescription.ClusterDescription, EventContext.OperationId);
serverSelectionWaitQueueDisposer ??= _serverSelectionWaitQueue.Enter(serverSelectionOperationContext, selector, expirableClusterDescription.ClusterDescription, EventContext.OperationId);

await operationContext.WaitTaskAsync(expirableClusterDescription.Expired).ConfigureAwait(false);
await serverSelectionOperationContext.WaitTaskAsync(expirableClusterDescription.Expired).ConfigureAwait(false);
expirableClusterDescription = _expirableClusterDescription;
}
}
Expand Down Expand Up @@ -306,7 +306,7 @@ private Exception HandleServerSelectionException(ClusterDescription clusterDescr
return exception;
}

private (IClusterableServer Server, ServerDescription ServerDescription) SelectServer(ExpirableClusterDescription clusterDescriptionChangeSource, IServerSelector selector, OperationsCountServerSelector operationCountSelector)
private SelectedServer SelectServer(ExpirableClusterDescription clusterDescriptionChangeSource, IServerSelector selector, OperationsCountServerSelector operationCountSelector)
{
MongoIncompatibleDriverException.ThrowIfNotSupported(clusterDescriptionChangeSource.ClusterDescription);

Expand All @@ -320,7 +320,7 @@ private Exception HandleServerSelectionException(ClusterDescription clusterDescr
var selectedServer = clusterDescriptionChangeSource.ConnectedServers.FirstOrDefault(s => EndPointHelper.Equals(s.EndPoint, selectedServerDescription.EndPoint));
if (selectedServer != null)
{
return (selectedServer, selectedServerDescription);
return new(selectedServer, selectedServerDescription);
}
}

Expand Down
14 changes: 8 additions & 6 deletions src/MongoDB.Driver/Core/Clusters/LoadBalancedCluster.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,7 @@ public IServer SelectServer(OperationContext operationContext, IServerSelector s
Ensure.IsNotNull(operationContext, nameof(operationContext));
ThrowIfDisposed();

var serverSelectionOperationContext = operationContext.WithTimeout(_settings.ServerSelectionTimeout);
using var serverSelectionOperationContext = operationContext.WithTimeout(_settings.ServerSelectionTimeout);

_serverSelectionEventLogger.LogAndPublish(new ClusterSelectingServerEvent(
_description,
Expand Down Expand Up @@ -205,10 +205,11 @@ public IServer SelectServer(OperationContext operationContext, IServerSelector s
stopwatch.Elapsed,
null,
EventContext.OperationName));

return new SelectedServer(_server, _server.Description);
}

return _server ??
throw new InvalidOperationException("The server must be created before usage."); // should not be reached
throw new InvalidOperationException("The server must be created before usage."); // should not be reached
}

public async Task<IServer> SelectServerAsync(OperationContext operationContext, IServerSelector selector)
Expand All @@ -217,7 +218,7 @@ public async Task<IServer> SelectServerAsync(OperationContext operationContext,
Ensure.IsNotNull(operationContext, nameof(operationContext));
ThrowIfDisposed();

var serverSelectionOperationContext = operationContext.WithTimeout(_settings.ServerSelectionTimeout);
using var serverSelectionOperationContext = operationContext.WithTimeout(_settings.ServerSelectionTimeout);

_serverSelectionEventLogger.LogAndPublish(new ClusterSelectingServerEvent(
_description,
Expand Down Expand Up @@ -245,10 +246,11 @@ public async Task<IServer> SelectServerAsync(OperationContext operationContext,
stopwatch.Elapsed,
null,
EventContext.OperationName));

return new SelectedServer(_server, _server.Description);
}

return _server ??
throw new InvalidOperationException("The server must be created before usage."); // should not be reached
throw new InvalidOperationException("The server must be created before usage."); // should not be reached
}

public ICoreSessionHandle StartSession(CoreSessionOptions options = null)
Expand Down
5 changes: 2 additions & 3 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 @@ -15,7 +15,6 @@

using Snappier;
using System.IO;
using System.Threading;
using MongoDB.Driver.Core.Misc;

namespace MongoDB.Driver.Core.Compression
Expand All @@ -34,7 +33,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, Timeout.InfiniteTimeSpan, CancellationToken.None);
input.ReadBytes(OperationContext.NoTimeout, uncompressedBytes, offset: 0, count: uncompressedSize);
var maxCompressedSize = Snappy.GetMaxCompressedLength(uncompressedSize);
var compressedBytes = new byte[maxCompressedSize];
var compressedSize = Snappy.Compress(uncompressedBytes, compressedBytes);
Expand All @@ -50,7 +49,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, Timeout.InfiniteTimeSpan, CancellationToken.None);
input.ReadBytes(OperationContext.NoTimeout, compressedBytes, offset: 0, count: compressedSize);
var uncompressedSize = Snappy.GetUncompressedLength(compressedBytes);
var decompressedBytes = new byte[uncompressedSize];
var decompressedSize = Snappy.Decompress(compressedBytes, decompressedBytes);
Expand Down
Loading