diff --git a/src/MongoDB.Driver/AbortTransactionOptions.cs b/src/MongoDB.Driver/AbortTransactionOptions.cs new file mode 100644 index 00000000000..649e7be1413 --- /dev/null +++ b/src/MongoDB.Driver/AbortTransactionOptions.cs @@ -0,0 +1,31 @@ +/* Copyright 2010-present MongoDB Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System; +using MongoDB.Driver.Core.Misc; + +namespace MongoDB.Driver +{ + // TODO: CSOT: Make it public when CSOT will be ready for GA + internal sealed class AbortTransactionOptions + { + public AbortTransactionOptions(TimeSpan? timeout) + { + Timeout = Ensure.IsNullOrValidTimeout(timeout, nameof(timeout)); + } + + public TimeSpan? Timeout { get; } + } +} diff --git a/src/MongoDB.Driver/ClientSessionHandle.cs b/src/MongoDB.Driver/ClientSessionHandle.cs index 2d606fdeb1b..144e6a94991 100644 --- a/src/MongoDB.Driver/ClientSessionHandle.cs +++ b/src/MongoDB.Driver/ClientSessionHandle.cs @@ -1,4 +1,4 @@ -/* Copyright 2017-present MongoDB Inc. +/* Copyright 2010-present MongoDB Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,7 +26,7 @@ namespace MongoDB.Driver /// A client session handle. /// /// - internal sealed class ClientSessionHandle : IClientSessionHandle + internal sealed class ClientSessionHandle : IClientSessionHandle, IClientSessionInternal { // private fields private readonly IMongoClient _client; @@ -94,16 +94,20 @@ public IServerSession ServerSession // public methods /// - public void AbortTransaction(CancellationToken cancellationToken = default(CancellationToken)) - { - _coreSession.AbortTransaction(cancellationToken); - } + public void AbortTransaction(CancellationToken cancellationToken = default) + => _coreSession.AbortTransaction(cancellationToken); + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + void IClientSessionInternal.AbortTransaction(AbortTransactionOptions options, CancellationToken cancellationToken) + => _coreSession.AbortTransaction(options, cancellationToken); /// - public Task AbortTransactionAsync(CancellationToken cancellationToken = default(CancellationToken)) - { - return _coreSession.AbortTransactionAsync(cancellationToken); - } + public Task AbortTransactionAsync(CancellationToken cancellationToken = default) + => _coreSession.AbortTransactionAsync(cancellationToken); + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + Task IClientSessionInternal.AbortTransactionAsync(AbortTransactionOptions options, CancellationToken cancellationToken) + => _coreSession.AbortTransactionAsync(options, cancellationToken); /// public void AdvanceClusterTime(BsonDocument newClusterTime) @@ -118,16 +122,20 @@ public void AdvanceOperationTime(BsonTimestamp newOperationTime) } /// - public void CommitTransaction(CancellationToken cancellationToken = default(CancellationToken)) - { - _coreSession.CommitTransaction(cancellationToken); - } + public void CommitTransaction(CancellationToken cancellationToken = default) + => _coreSession.CommitTransaction(cancellationToken); + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + void IClientSessionInternal.CommitTransaction(CommitTransactionOptions options, CancellationToken cancellationToken) + => _coreSession.CommitTransaction(options, cancellationToken); /// - public Task CommitTransactionAsync(CancellationToken cancellationToken = default(CancellationToken)) - { - return _coreSession.CommitTransactionAsync(cancellationToken); - } + public Task CommitTransactionAsync(CancellationToken cancellationToken = default) + => _coreSession.CommitTransactionAsync(cancellationToken); + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + Task IClientSessionInternal.CommitTransactionAsync(CommitTransactionOptions options, CancellationToken cancellationToken) + => _coreSession.CommitTransactionAsync(options, cancellationToken); /// public void Dispose() diff --git a/src/MongoDB.Driver/CommitTransactionOptions.cs b/src/MongoDB.Driver/CommitTransactionOptions.cs new file mode 100644 index 00000000000..008e902815e --- /dev/null +++ b/src/MongoDB.Driver/CommitTransactionOptions.cs @@ -0,0 +1,32 @@ +/* Copyright 2010-present MongoDB Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System; +using MongoDB.Driver.Core.Misc; + +namespace MongoDB.Driver +{ + // TODO: CSOT: Make it public when CSOT will be ready for GA + internal sealed class CommitTransactionOptions + { + public CommitTransactionOptions(TimeSpan? timeout) + { + Timeout = Ensure.IsNullOrValidTimeout(timeout, nameof(timeout)); + } + + public TimeSpan? Timeout { get; } + } +} + diff --git a/src/MongoDB.Driver/Core/Bindings/CoreSession.cs b/src/MongoDB.Driver/Core/Bindings/CoreSession.cs index d91eee1610e..0cc6ff66483 100644 --- a/src/MongoDB.Driver/Core/Bindings/CoreSession.cs +++ b/src/MongoDB.Driver/Core/Bindings/CoreSession.cs @@ -1,4 +1,4 @@ -/* Copyright 2018-present MongoDB Inc. +/* Copyright 2010-present MongoDB Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -29,7 +29,7 @@ namespace MongoDB.Driver.Core.Bindings /// Represents a session. /// /// - public sealed class CoreSession : ICoreSession + public sealed class CoreSession : ICoreSession, ICoreSessionInternal { // private fields #pragma warning disable CA2213 // Disposable fields should be disposed @@ -141,12 +141,15 @@ public bool IsInTransaction // public methods /// - public void AbortTransaction(CancellationToken cancellationToken = default(CancellationToken)) + public void AbortTransaction(CancellationToken cancellationToken = default) + => ((ICoreSessionInternal)this).AbortTransaction(null, cancellationToken); + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + void ICoreSessionInternal.AbortTransaction(AbortTransactionOptions options, CancellationToken cancellationToken) { EnsureAbortTransactionCanBeCalled(nameof(AbortTransaction)); - // TODO: CSOT implement proper way to obtain the operationContext - var operationContext = new OperationContext(null, cancellationToken); + using var operationContext = new OperationContext(GetTimeout(options?.Timeout), cancellationToken); try { if (_currentTransaction.IsEmpty) @@ -192,12 +195,15 @@ public bool IsInTransaction } /// - public async Task AbortTransactionAsync(CancellationToken cancellationToken = default(CancellationToken)) + public Task AbortTransactionAsync(CancellationToken cancellationToken = default) + => ((ICoreSessionInternal)this).AbortTransactionAsync(null, cancellationToken); + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + async Task ICoreSessionInternal.AbortTransactionAsync(AbortTransactionOptions options, CancellationToken cancellationToken) { EnsureAbortTransactionCanBeCalled(nameof(AbortTransaction)); - // TODO: CSOT implement proper way to obtain the operationContext - var operationContext = new OperationContext(null, cancellationToken); + using var operationContext = new OperationContext(GetTimeout(options?.Timeout), cancellationToken); try { if (_currentTransaction.IsEmpty) @@ -292,12 +298,15 @@ public long AdvanceTransactionNumber() } /// - public void CommitTransaction(CancellationToken cancellationToken = default(CancellationToken)) + public void CommitTransaction(CancellationToken cancellationToken = default) + => ((ICoreSessionInternal)this).CommitTransaction(null, cancellationToken); + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + void ICoreSessionInternal.CommitTransaction(CommitTransactionOptions options, CancellationToken cancellationToken) { EnsureCommitTransactionCanBeCalled(nameof(CommitTransaction)); - // TODO: CSOT implement proper way to obtain the operationContext - var operationContext = new OperationContext(null, cancellationToken); + using var operationContext = new OperationContext(GetTimeout(options?.Timeout), cancellationToken); try { _isCommitTransactionInProgress = true; @@ -329,12 +338,15 @@ public long AdvanceTransactionNumber() } /// - public async Task CommitTransactionAsync(CancellationToken cancellationToken = default(CancellationToken)) + public Task CommitTransactionAsync(CancellationToken cancellationToken = default) + => ((ICoreSessionInternal)this).CommitTransactionAsync(null, cancellationToken); + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + async Task ICoreSessionInternal.CommitTransactionAsync(CommitTransactionOptions options, CancellationToken cancellationToken) { EnsureCommitTransactionCanBeCalled(nameof(CommitTransaction)); - // TODO: CSOT implement proper way to obtain the operationContext - var operationContext = new OperationContext(null, cancellationToken); + using var operationContext = new OperationContext(GetTimeout(options?.Timeout), cancellationToken); try { _isCommitTransactionInProgress = true; @@ -563,6 +575,9 @@ private async Task ExecuteEndTransactionOnPrimaryAsync(Operati } } + private TimeSpan? GetTimeout(TimeSpan? timeout) + => timeout ?? _options.DefaultTransactionOptions?.Timeout; + private TransactionOptions GetEffectiveTransactionOptions(TransactionOptions transactionOptions) { var readConcern = transactionOptions?.ReadConcern ?? _options.DefaultTransactionOptions?.ReadConcern ?? ReadConcern.Default; diff --git a/src/MongoDB.Driver/Core/Bindings/CoreTransaction.cs b/src/MongoDB.Driver/Core/Bindings/CoreTransaction.cs index 53747c8530c..6ed2a4f849e 100644 --- a/src/MongoDB.Driver/Core/Bindings/CoreTransaction.cs +++ b/src/MongoDB.Driver/Core/Bindings/CoreTransaction.cs @@ -1,4 +1,4 @@ -/* Copyright 2018-present MongoDB Inc. +/* Copyright 2010-present MongoDB Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -56,6 +56,8 @@ public CoreTransaction(long transactionNumber, TransactionOptions transactionOpt /// public bool IsEmpty => _isEmpty; + internal OperationContext OperationContext { get; set; } + /// /// Gets the transaction state. /// diff --git a/src/MongoDB.Driver/Core/Bindings/ICoreSessionExtensions.cs b/src/MongoDB.Driver/Core/Bindings/ICoreSessionExtensions.cs new file mode 100644 index 00000000000..5d176c6181b --- /dev/null +++ b/src/MongoDB.Driver/Core/Bindings/ICoreSessionExtensions.cs @@ -0,0 +1,67 @@ +/* Copyright 2010-present MongoDB Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System.Threading; +using System.Threading.Tasks; + +namespace MongoDB.Driver.Core.Bindings +{ + // TODO: CSOT: Make it public when CSOT will be ready for GA + internal static class ICoreSessionExtensions + { + // TODO: Merge these extension methods in ICoreSession interface on major release + public static void AbortTransaction(this ICoreSession session, AbortTransactionOptions options, CancellationToken cancellationToken = default) + { + if (options == null || session.Options.DefaultTransactionOptions?.Timeout == options.Timeout) + { + session.AbortTransaction(cancellationToken); + return; + } + + ((ICoreSessionInternal)session).AbortTransaction(options, cancellationToken); + } + + public static Task AbortTransactionAsync(this ICoreSession session, AbortTransactionOptions options, CancellationToken cancellationToken = default) + { + if (options == null || session.Options.DefaultTransactionOptions?.Timeout == options.Timeout) + { + return session.AbortTransactionAsync(cancellationToken); + } + + return ((ICoreSessionInternal)session).AbortTransactionAsync(options, cancellationToken); + } + + public static void CommitTransaction(this ICoreSession session, CommitTransactionOptions options, CancellationToken cancellationToken = default) + { + if (options == null || session.Options.DefaultTransactionOptions?.Timeout == options.Timeout) + { + session.CommitTransaction(cancellationToken); + return; + } + + ((ICoreSessionInternal)session).CommitTransaction(options, cancellationToken); + } + + public static Task CommitTransactionAsync(this ICoreSession session, CommitTransactionOptions options, CancellationToken cancellationToken = default) + { + if (options == null || session.Options.DefaultTransactionOptions?.Timeout == options.Timeout) + { + return session.CommitTransactionAsync(cancellationToken); + } + + return ((ICoreSessionInternal)session).CommitTransactionAsync(options, cancellationToken); + } + } +} diff --git a/src/MongoDB.Driver/Core/Bindings/ICoreSessionInternal.cs b/src/MongoDB.Driver/Core/Bindings/ICoreSessionInternal.cs new file mode 100644 index 00000000000..1844ae6fa8c --- /dev/null +++ b/src/MongoDB.Driver/Core/Bindings/ICoreSessionInternal.cs @@ -0,0 +1,28 @@ +/* Copyright 2010-present MongoDB Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System.Threading; +using System.Threading.Tasks; + +namespace MongoDB.Driver.Core.Bindings; + +// TODO: Merge this interface into ICoreSession on major release +internal interface ICoreSessionInternal +{ + void AbortTransaction(AbortTransactionOptions options, CancellationToken cancellationToken = default); + Task AbortTransactionAsync(AbortTransactionOptions options, CancellationToken cancellationToken = default); + void CommitTransaction(CommitTransactionOptions options, CancellationToken cancellationToken = default); + Task CommitTransactionAsync(CommitTransactionOptions options, CancellationToken cancellationToken = default); +} diff --git a/src/MongoDB.Driver/Core/Bindings/NoCoreSession.cs b/src/MongoDB.Driver/Core/Bindings/NoCoreSession.cs index 7be93d81b43..ac7e68abf06 100644 --- a/src/MongoDB.Driver/Core/Bindings/NoCoreSession.cs +++ b/src/MongoDB.Driver/Core/Bindings/NoCoreSession.cs @@ -24,7 +24,7 @@ namespace MongoDB.Driver.Core.Bindings /// An object that represents no core session. /// /// - public sealed class NoCoreSession : ICoreSession + public sealed class NoCoreSession : ICoreSession, ICoreSessionInternal { #region static // private static fields @@ -89,13 +89,25 @@ public static ICoreSessionHandle NewHandle() // public methods /// - public void AbortTransaction(CancellationToken cancellationToken = default(CancellationToken)) + public void AbortTransaction(CancellationToken cancellationToken = default) + { + throw new NotSupportedException("NoCoreSession does not support AbortTransaction."); + } + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + void ICoreSessionInternal.AbortTransaction(AbortTransactionOptions options, CancellationToken cancellationToken ) { throw new NotSupportedException("NoCoreSession does not support AbortTransaction."); } /// - public Task AbortTransactionAsync(CancellationToken cancellationToken = default(CancellationToken)) + public Task AbortTransactionAsync(CancellationToken cancellationToken = default) + { + throw new NotSupportedException("NoCoreSession does not support AbortTransactionAsync."); + } + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + Task ICoreSessionInternal.AbortTransactionAsync(AbortTransactionOptions options, CancellationToken cancellationToken ) { throw new NotSupportedException("NoCoreSession does not support AbortTransactionAsync."); } @@ -122,13 +134,25 @@ public long AdvanceTransactionNumber() } /// - public void CommitTransaction(CancellationToken cancellationToken = default(CancellationToken)) + public void CommitTransaction(CancellationToken cancellationToken = default) + { + throw new NotSupportedException("NoCoreSession does not support CommitTransaction."); + } + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + void ICoreSessionInternal.CommitTransaction(CommitTransactionOptions options, CancellationToken cancellationToken) { throw new NotSupportedException("NoCoreSession does not support CommitTransaction."); } /// - public Task CommitTransactionAsync(CancellationToken cancellationToken = default(CancellationToken)) + public Task CommitTransactionAsync(CancellationToken cancellationToken = default) + { + throw new NotSupportedException("NoCoreSession does not support CommitTransactionAsync."); + } + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + Task ICoreSessionInternal.CommitTransactionAsync(CommitTransactionOptions options, CancellationToken cancellationToken) { throw new NotSupportedException("NoCoreSession does not support CommitTransactionAsync."); } diff --git a/src/MongoDB.Driver/Core/Bindings/WrappingCoreSession.cs b/src/MongoDB.Driver/Core/Bindings/WrappingCoreSession.cs index 991e46ab115..1d61b552d9d 100644 --- a/src/MongoDB.Driver/Core/Bindings/WrappingCoreSession.cs +++ b/src/MongoDB.Driver/Core/Bindings/WrappingCoreSession.cs @@ -25,7 +25,7 @@ namespace MongoDB.Driver.Core.Bindings /// An abstract base class for a core session that wraps another core session. /// /// - public abstract class WrappingCoreSession : ICoreSession + public abstract class WrappingCoreSession : ICoreSession, ICoreSessionInternal { // private fields private bool _disposed; @@ -182,19 +182,33 @@ public ICoreSession Wrapped // public methods /// - public virtual void AbortTransaction(CancellationToken cancellationToken = default(CancellationToken)) + public virtual void AbortTransaction(CancellationToken cancellationToken = default) { ThrowIfDisposed(); _wrapped.AbortTransaction(cancellationToken); } + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + void ICoreSessionInternal.AbortTransaction(AbortTransactionOptions options, CancellationToken cancellationToken) + { + ThrowIfDisposed(); + _wrapped.AbortTransaction(options, cancellationToken); + } + /// - public virtual Task AbortTransactionAsync(CancellationToken cancellationToken = default(CancellationToken)) + public virtual Task AbortTransactionAsync(CancellationToken cancellationToken = default) { ThrowIfDisposed(); return _wrapped.AbortTransactionAsync(cancellationToken); } + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + Task ICoreSessionInternal.AbortTransactionAsync(AbortTransactionOptions options, CancellationToken cancellationToken) + { + ThrowIfDisposed(); + return _wrapped.AbortTransactionAsync(options, cancellationToken); + } + /// public virtual void AboutToSendCommand() { @@ -223,19 +237,34 @@ public long AdvanceTransactionNumber() } /// - public virtual void CommitTransaction(CancellationToken cancellationToken = default(CancellationToken)) + public virtual void CommitTransaction(CancellationToken cancellationToken = default) { ThrowIfDisposed(); _wrapped.CommitTransaction(cancellationToken); } + + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + void ICoreSessionInternal.CommitTransaction(CommitTransactionOptions options, CancellationToken cancellationToken) + { + ThrowIfDisposed(); + _wrapped.CommitTransaction(options, cancellationToken); + } + /// - public virtual Task CommitTransactionAsync(CancellationToken cancellationToken = default(CancellationToken)) + public virtual Task CommitTransactionAsync(CancellationToken cancellationToken = default) { ThrowIfDisposed(); return _wrapped.CommitTransactionAsync(cancellationToken); } + // TODO: CSOT: Make it public when CSOT will be ready for GA and add default value to cancellationToken parameter. + Task ICoreSessionInternal.CommitTransactionAsync(CommitTransactionOptions options, CancellationToken cancellationToken) + { + ThrowIfDisposed(); + return _wrapped.CommitTransactionAsync(options, cancellationToken); + } + /// public void Dispose() { diff --git a/src/MongoDB.Driver/Core/Misc/IClock.cs b/src/MongoDB.Driver/Core/Misc/IClock.cs index d409bb604ee..804f6912d68 100644 --- a/src/MongoDB.Driver/Core/Misc/IClock.cs +++ b/src/MongoDB.Driver/Core/Misc/IClock.cs @@ -1,4 +1,4 @@ -/* Copyright 2013-present MongoDB Inc. +/* Copyright 2010-present MongoDB Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,5 +20,7 @@ namespace MongoDB.Driver.Core.Misc internal interface IClock { DateTime UtcNow { get; } + + IStopwatch StartStopwatch(); } } diff --git a/src/MongoDB.Driver/Core/Misc/IStopwatch.cs b/src/MongoDB.Driver/Core/Misc/IStopwatch.cs new file mode 100644 index 00000000000..87a8fbfd0fc --- /dev/null +++ b/src/MongoDB.Driver/Core/Misc/IStopwatch.cs @@ -0,0 +1,24 @@ +/* Copyright 2010-present MongoDB Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System; + +namespace MongoDB.Driver.Core.Misc +{ + internal interface IStopwatch + { + TimeSpan Elapsed { get; } + } +} diff --git a/src/MongoDB.Driver/Core/Misc/SystemClock.cs b/src/MongoDB.Driver/Core/Misc/SystemClock.cs index f972f9aed62..f418b635e07 100644 --- a/src/MongoDB.Driver/Core/Misc/SystemClock.cs +++ b/src/MongoDB.Driver/Core/Misc/SystemClock.cs @@ -1,4 +1,4 @@ -/* Copyright 2013-present MongoDB Inc. +/* Copyright 2010-present MongoDB Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -32,5 +32,7 @@ public DateTime UtcNow { get { return DateTime.UtcNow; } } + + public IStopwatch StartStopwatch() => new SystemStopwatch(); } } diff --git a/src/MongoDB.Driver/Core/Misc/SystemStopwatch.cs b/src/MongoDB.Driver/Core/Misc/SystemStopwatch.cs new file mode 100644 index 00000000000..4d9a19b5b47 --- /dev/null +++ b/src/MongoDB.Driver/Core/Misc/SystemStopwatch.cs @@ -0,0 +1,32 @@ +/* Copyright 2010-present MongoDB Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System; +using System.Diagnostics; + +namespace MongoDB.Driver.Core.Misc +{ + internal sealed class SystemStopwatch : IStopwatch + { + private readonly Stopwatch _wrappedStopwatch; + + public SystemStopwatch() + { + _wrappedStopwatch = Stopwatch.StartNew(); + } + + public TimeSpan Elapsed => _wrappedStopwatch.Elapsed; + } +} diff --git a/src/MongoDB.Driver/Core/Operations/EndTransactionOperation.cs b/src/MongoDB.Driver/Core/Operations/EndTransactionOperation.cs index 0544b2498c3..007721c3ceb 100644 --- a/src/MongoDB.Driver/Core/Operations/EndTransactionOperation.cs +++ b/src/MongoDB.Driver/Core/Operations/EndTransactionOperation.cs @@ -58,7 +58,7 @@ public virtual BsonDocument Execute(OperationContext operationContext, IReadBind using (var channel = channelSource.GetChannel(operationContext)) using (var channelBinding = new ChannelReadWriteBinding(channelSource.Server, channel, binding.Session.Fork())) { - var operation = CreateOperation(); + var operation = CreateOperation(operationContext); return operation.Execute(operationContext, channelBinding); } } @@ -71,12 +71,12 @@ public virtual async Task ExecuteAsync(OperationContext operationC using (var channel = await channelSource.GetChannelAsync(operationContext).ConfigureAwait(false)) using (var channelBinding = new ChannelReadWriteBinding(channelSource.Server, channel, binding.Session.Fork())) { - var operation = CreateOperation(); + var operation = CreateOperation(operationContext); return await operation.ExecuteAsync(operationContext, channelBinding).ConfigureAwait(false); } } - protected virtual BsonDocument CreateCommand() + protected virtual BsonDocument CreateCommand(OperationContext operationContext) { return new BsonDocument { @@ -86,9 +86,9 @@ protected virtual BsonDocument CreateCommand() }; } - private IReadOperation CreateOperation() + private IReadOperation CreateOperation(OperationContext operationContext) { - var command = CreateCommand(); + var command = CreateCommand(operationContext); return new ReadCommandOperation(DatabaseNamespace.Admin, command, BsonDocumentSerializer.Instance, _messageEncoderSettings) { RetryRequested = false @@ -159,10 +159,10 @@ public override async Task ExecuteAsync(OperationContext operation } } - protected override BsonDocument CreateCommand() + protected override BsonDocument CreateCommand(OperationContext operationContext) { - var command = base.CreateCommand(); - if (_maxCommitTime.HasValue) + var command = base.CreateCommand(operationContext); + if (_maxCommitTime.HasValue && !operationContext.IsRootContextTimeoutConfigured()) { command.Add("maxTimeMS", (long)_maxCommitTime.Value.TotalMilliseconds); } diff --git a/src/MongoDB.Driver/Core/TransactionOptions.cs b/src/MongoDB.Driver/Core/TransactionOptions.cs index 9003b0829d4..c4305a95ad3 100644 --- a/src/MongoDB.Driver/Core/TransactionOptions.cs +++ b/src/MongoDB.Driver/Core/TransactionOptions.cs @@ -1,4 +1,4 @@ -/* Copyright 2018-present MongoDB Inc. +/* Copyright 2010-present MongoDB Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -14,6 +14,7 @@ */ using System; +using MongoDB.Driver.Core.Misc; namespace MongoDB.Driver { @@ -26,6 +27,7 @@ public class TransactionOptions private readonly TimeSpan? _maxCommitTime; private readonly ReadConcern _readConcern; private readonly ReadPreference _readPreference; + private readonly TimeSpan? _timeout; private readonly WriteConcern _writeConcern; // public constructors @@ -41,7 +43,27 @@ public TransactionOptions( Optional readPreference = default(Optional), Optional writeConcern = default(Optional), Optional maxCommitTime = default(Optional)) + : this(null, readConcern, readPreference, writeConcern, maxCommitTime) { + } + + /// + /// Initializes a new instance of the class. + /// + /// The per operation timeout + /// The read concern. + /// The read preference. + /// The write concern. + /// The max commit time. + // TODO: CSOT: Make it public when CSOT will be ready for GA + internal TransactionOptions( + TimeSpan? timeout, + Optional readConcern = default(Optional), + Optional readPreference = default(Optional), + Optional writeConcern = default(Optional), + Optional maxCommitTime = default(Optional)) + { + _timeout = Ensure.IsNullOrValidTimeout(timeout, nameof(timeout)); _readConcern = readConcern.WithDefault(null); _readPreference = readPreference.WithDefault(null); _writeConcern = writeConcern.WithDefault(null); @@ -73,6 +95,12 @@ public TransactionOptions( /// public ReadPreference ReadPreference => _readPreference; + /// + /// Gets the per operation timeout. + /// + // TODO: CSOT: Make it public when CSOT will be ready for GA + internal TimeSpan? Timeout => _timeout; + /// /// Gets the write concern. /// diff --git a/src/MongoDB.Driver/IClientSessionExtensions.cs b/src/MongoDB.Driver/IClientSessionExtensions.cs index b1ce144636c..1c16957a8db 100644 --- a/src/MongoDB.Driver/IClientSessionExtensions.cs +++ b/src/MongoDB.Driver/IClientSessionExtensions.cs @@ -13,22 +13,69 @@ * limitations under the License. */ -namespace MongoDB.Driver; +using System.Threading; +using System.Threading.Tasks; -internal static class IClientSessionExtensions +namespace MongoDB.Driver { - public static ReadPreference GetEffectiveReadPreference(this IClientSession session, ReadPreference defaultReadPreference) + // TODO: CSOT: Make it public when CSOT will be ready for GA + internal static class IClientSessionExtensions { - if (session.IsInTransaction) + // TODO: Merge these extension methods in IClientSession interface on major release + public static void AbortTransaction(this IClientSession session, AbortTransactionOptions options, CancellationToken cancellationToken = default) { - var transactionReadPreference = session.WrappedCoreSession.CurrentTransaction.TransactionOptions?.ReadPreference; - if (transactionReadPreference != null) + if (options?.Timeout == null || session.Options.DefaultTransactionOptions?.Timeout == options.Timeout) { - return transactionReadPreference; + session.AbortTransaction(cancellationToken); + return; } + + ((IClientSessionInternal)session).AbortTransaction(options, cancellationToken); + } + + public static Task AbortTransactionAsync(this IClientSession session, AbortTransactionOptions options, CancellationToken cancellationToken = default) + { + if (options?.Timeout == null || session.Options.DefaultTransactionOptions?.Timeout == options.Timeout) + { + return session.AbortTransactionAsync(cancellationToken); + } + + return ((IClientSessionInternal)session).AbortTransactionAsync(options, cancellationToken); + } + + public static void CommitTransaction(this IClientSession session, CommitTransactionOptions options, CancellationToken cancellationToken = default) + { + if (options?.Timeout == null || session.Options.DefaultTransactionOptions?.Timeout == options.Timeout) + { + session.CommitTransaction(cancellationToken); + return; + } + + ((IClientSessionInternal)session).CommitTransaction(options, cancellationToken); } - return defaultReadPreference ?? ReadPreference.Primary; + public static Task CommitTransactionAsync(this IClientSession session, CommitTransactionOptions options, CancellationToken cancellationToken = default) + { + if (options?.Timeout == null || session.Options.DefaultTransactionOptions?.Timeout == options.Timeout) + { + return session.CommitTransactionAsync(cancellationToken); + } + + return ((IClientSessionInternal)session).CommitTransactionAsync(options, cancellationToken); + } + + internal static ReadPreference GetEffectiveReadPreference(this IClientSession session, ReadPreference defaultReadPreference) + { + if (session.IsInTransaction) + { + var transactionReadPreference = session.WrappedCoreSession.CurrentTransaction.TransactionOptions?.ReadPreference; + if (transactionReadPreference != null) + { + return transactionReadPreference; + } + } + + return defaultReadPreference ?? ReadPreference.Primary; + } } } - diff --git a/src/MongoDB.Driver/IClientSessionInternal.cs b/src/MongoDB.Driver/IClientSessionInternal.cs new file mode 100644 index 00000000000..4107b7b811c --- /dev/null +++ b/src/MongoDB.Driver/IClientSessionInternal.cs @@ -0,0 +1,28 @@ +/* Copyright 2010-present MongoDB Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System.Threading; +using System.Threading.Tasks; + +namespace MongoDB.Driver; + +// TODO: Merge this interface into ICoreSession on major release +internal interface IClientSessionInternal +{ + void AbortTransaction(AbortTransactionOptions options, CancellationToken cancellationToken = default); + Task AbortTransactionAsync(AbortTransactionOptions options, CancellationToken cancellationToken = default); + void CommitTransaction(CommitTransactionOptions options, CancellationToken cancellationToken = default); + Task CommitTransactionAsync(CommitTransactionOptions options, CancellationToken cancellationToken = default); +} diff --git a/src/MongoDB.Driver/IOperationExecutor.cs b/src/MongoDB.Driver/IOperationExecutor.cs index b64c2380776..6dfd8e741c5 100644 --- a/src/MongoDB.Driver/IOperationExecutor.cs +++ b/src/MongoDB.Driver/IOperationExecutor.cs @@ -14,7 +14,6 @@ */ using System; -using System.Threading; using System.Threading.Tasks; using MongoDB.Driver.Core.Operations; @@ -23,34 +22,30 @@ namespace MongoDB.Driver internal interface IOperationExecutor : IDisposable { TResult ExecuteReadOperation( + OperationContext operationContext, IClientSessionHandle session, IReadOperation operation, ReadPreference readPreference, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken); + bool allowChannelPinning); Task ExecuteReadOperationAsync( + OperationContext operationContext, IClientSessionHandle session, IReadOperation operation, ReadPreference readPreference, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken); + bool allowChannelPinning); TResult ExecuteWriteOperation( + OperationContext operationContext, IClientSessionHandle session, IWriteOperation operation, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken); + bool allowChannelPinning); Task ExecuteWriteOperationAsync( + OperationContext operationContext, IClientSessionHandle session, IWriteOperation operation, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken); + bool allowChannelPinning); IClientSessionHandle StartImplicitSession(); } diff --git a/src/MongoDB.Driver/MongoClient.cs b/src/MongoDB.Driver/MongoClient.cs index 4fde2d760bd..2b39518cb17 100644 --- a/src/MongoDB.Driver/MongoClient.cs +++ b/src/MongoDB.Driver/MongoClient.cs @@ -563,24 +563,42 @@ private ChangeStreamOperation CreateChangeStreamOperation( _settings.RetryReads, _settings.TranslationOptions); + private OperationContext CreateOperationContext(IClientSessionHandle session, TimeSpan? timeout, CancellationToken cancellationToken) + { + var operationContext = session.WrappedCoreSession.CurrentTransaction?.OperationContext; + if (operationContext != null && timeout != null) + { + throw new InvalidOperationException("Cannot specify per operation timeout inside transaction."); + } + + return operationContext ?? new OperationContext(timeout ?? _settings.Timeout, cancellationToken); + } + private TResult ExecuteReadOperation(IClientSessionHandle session, IReadOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) { var readPreference = session.GetEffectiveReadPreference(_settings.ReadPreference); - return _operationExecutor.ExecuteReadOperation(session, operation, readPreference, false, timeout ?? _settings.Timeout, cancellationToken); + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return _operationExecutor.ExecuteReadOperation(operationContext, session, operation, readPreference, false); } - private Task ExecuteReadOperationAsync(IClientSessionHandle session, IReadOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) + private async Task ExecuteReadOperationAsync(IClientSessionHandle session, IReadOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) { var readPreference = session.GetEffectiveReadPreference(_settings.ReadPreference); - return _operationExecutor.ExecuteReadOperationAsync(session, operation, readPreference, false, timeout ?? _settings.Timeout, cancellationToken); + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return await _operationExecutor.ExecuteReadOperationAsync(operationContext, session, operation, readPreference, false).ConfigureAwait(false); } private TResult ExecuteWriteOperation(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) - => _operationExecutor.ExecuteWriteOperation(session, operation, false, timeout ?? _settings.Timeout, cancellationToken); - - private Task ExecuteWriteOperationAsync(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) - => _operationExecutor.ExecuteWriteOperationAsync(session, operation, false, timeout ?? _settings.Timeout, cancellationToken); + { + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return _operationExecutor.ExecuteWriteOperation(operationContext, session, operation, false); + } + private async Task ExecuteWriteOperationAsync(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) + { + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return await _operationExecutor.ExecuteWriteOperationAsync(operationContext, session, operation, false).ConfigureAwait(false); + } private MessageEncoderSettings GetMessageEncoderSettings() { @@ -609,7 +627,17 @@ private IClientSessionHandle StartSession(ClientSessionOptions options) throw new NotSupportedException("Combining both causal consistency and snapshot options is not supported."); } - options = options ?? new ClientSessionOptions(); + options ??= new ClientSessionOptions(); + if (_settings.Timeout.HasValue && options.DefaultTransactionOptions?.Timeout == null) + { + options.DefaultTransactionOptions = new TransactionOptions( + _settings.Timeout, + options.DefaultTransactionOptions?.ReadConcern, + options.DefaultTransactionOptions?.ReadPreference, + options.DefaultTransactionOptions?.WriteConcern, + options.DefaultTransactionOptions?.MaxCommitTime); + } + var coreSession = _cluster.StartSession(options.ToCore()); return new ClientSessionHandle(this, options, coreSession); diff --git a/src/MongoDB.Driver/MongoCollectionImpl.cs b/src/MongoDB.Driver/MongoCollectionImpl.cs index 730f78ea165..c2800d1044f 100644 --- a/src/MongoDB.Driver/MongoCollectionImpl.cs +++ b/src/MongoDB.Driver/MongoCollectionImpl.cs @@ -1199,29 +1199,48 @@ private IAsyncCursor CreateMapReduceOutputToCollectionResultCursor(IClientSessionHandle session, IReadOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) => ExecuteReadOperation(session, operation, null, timeout, cancellationToken); private TResult ExecuteReadOperation(IClientSessionHandle session, IReadOperation operation, ReadPreference explicitReadPreference, TimeSpan? timeout, CancellationToken cancellationToken) { var readPreference = explicitReadPreference ?? session.GetEffectiveReadPreference(_settings.ReadPreference); - return _operationExecutor.ExecuteReadOperation(session, operation, readPreference, true, timeout ?? _settings.Timeout, cancellationToken); + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return _operationExecutor.ExecuteReadOperation(operationContext, session, operation, readPreference, true); } private Task ExecuteReadOperationAsync(IClientSessionHandle session, IReadOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) => ExecuteReadOperationAsync(session, operation, null, timeout, cancellationToken); - private Task ExecuteReadOperationAsync(IClientSessionHandle session, IReadOperation operation, ReadPreference explicitReadPreference, TimeSpan? timeout, CancellationToken cancellationToken) + private async Task ExecuteReadOperationAsync(IClientSessionHandle session, IReadOperation operation, ReadPreference explicitReadPreference, TimeSpan? timeout, CancellationToken cancellationToken) { var readPreference = explicitReadPreference ?? session.GetEffectiveReadPreference(_settings.ReadPreference); - return _operationExecutor.ExecuteReadOperationAsync(session, operation, readPreference, true, timeout ?? _settings.Timeout, cancellationToken); + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return await _operationExecutor.ExecuteReadOperationAsync(operationContext, session, operation, readPreference, true).ConfigureAwait(false); } private TResult ExecuteWriteOperation(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) - => _operationExecutor.ExecuteWriteOperation(session, operation, true, timeout ?? _settings.Timeout, cancellationToken); + { + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return _operationExecutor.ExecuteWriteOperation(operationContext, session, operation, true); + } - private Task ExecuteWriteOperationAsync(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) - => _operationExecutor.ExecuteWriteOperationAsync(session, operation, true, timeout ?? _settings.Timeout, cancellationToken); + private async Task ExecuteWriteOperationAsync(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) + { + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return await _operationExecutor.ExecuteWriteOperationAsync(operationContext, session, operation, true).ConfigureAwait(false); + } private MessageEncoderSettings GetMessageEncoderSettings() { diff --git a/src/MongoDB.Driver/MongoDatabase.cs b/src/MongoDB.Driver/MongoDatabase.cs index 779af6deaed..87016378309 100644 --- a/src/MongoDB.Driver/MongoDatabase.cs +++ b/src/MongoDB.Driver/MongoDatabase.cs @@ -755,29 +755,48 @@ private ChangeStreamOperation CreateChangeStreamOperation( translationOptions); } + private OperationContext CreateOperationContext(IClientSessionHandle session, TimeSpan? timeout, CancellationToken cancellationToken) + { + var operationContext = session.WrappedCoreSession.CurrentTransaction?.OperationContext; + if (operationContext != null && timeout != null) + { + throw new InvalidOperationException("Cannot specify per operation timeout inside transaction."); + } + + return operationContext ?? new OperationContext(timeout ?? _settings.Timeout, cancellationToken); + } + private TResult ExecuteReadOperation(IClientSessionHandle session, IReadOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) => ExecuteReadOperation(session, operation, null, timeout, cancellationToken); private TResult ExecuteReadOperation(IClientSessionHandle session, IReadOperation operation, ReadPreference explicitReadPreference, TimeSpan? timeout, CancellationToken cancellationToken) { var readPreference = explicitReadPreference ?? session.GetEffectiveReadPreference(_settings.ReadPreference); - return _operationExecutor.ExecuteReadOperation(session, operation, readPreference, true, timeout ?? _settings.Timeout, cancellationToken); + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return _operationExecutor.ExecuteReadOperation(operationContext, session, operation, readPreference, true); } private Task ExecuteReadOperationAsync(IClientSessionHandle session, IReadOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) => ExecuteReadOperationAsync(session, operation, null, timeout, cancellationToken); - private Task ExecuteReadOperationAsync(IClientSessionHandle session, IReadOperation operation, ReadPreference explicitReadPreference, TimeSpan? timeout, CancellationToken cancellationToken) + private async Task ExecuteReadOperationAsync(IClientSessionHandle session, IReadOperation operation, ReadPreference explicitReadPreference, TimeSpan? timeout, CancellationToken cancellationToken) { var readPreference = explicitReadPreference ?? session.GetEffectiveReadPreference(_settings.ReadPreference); - return _operationExecutor.ExecuteReadOperationAsync(session, operation, readPreference, true, timeout ?? _settings.Timeout, cancellationToken); + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return await _operationExecutor.ExecuteReadOperationAsync(operationContext, session, operation, readPreference, true).ConfigureAwait(false); } private TResult ExecuteWriteOperation(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) - => _operationExecutor.ExecuteWriteOperation(session, operation, true, timeout ?? _settings.Timeout, cancellationToken); + { + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return _operationExecutor.ExecuteWriteOperation(operationContext, session, operation, true); + } - private Task ExecuteWriteOperationAsync(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) - => _operationExecutor.ExecuteWriteOperationAsync(session, operation, true, timeout ?? _settings.Timeout, cancellationToken); + private async Task ExecuteWriteOperationAsync(IClientSessionHandle session, IWriteOperation operation, TimeSpan? timeout, CancellationToken cancellationToken) + { + using var operationContext = CreateOperationContext(session, timeout, cancellationToken); + return await _operationExecutor.ExecuteWriteOperationAsync(operationContext, session, operation, true).ConfigureAwait(false); + } private IEnumerable ExtractCollectionNames(IEnumerable collections) { diff --git a/src/MongoDB.Driver/OperationContext.cs b/src/MongoDB.Driver/OperationContext.cs index d2a18e6e729..5a00621a442 100644 --- a/src/MongoDB.Driver/OperationContext.cs +++ b/src/MongoDB.Driver/OperationContext.cs @@ -14,7 +14,6 @@ */ using System; -using System.Diagnostics; using System.Threading; using System.Threading.Tasks; using MongoDB.Driver.Core.Misc; @@ -30,13 +29,14 @@ internal sealed class OperationContext : IDisposable private CancellationTokenSource _combinedCancellationTokenSource; public OperationContext(TimeSpan? timeout, CancellationToken cancellationToken) - : this(Stopwatch.StartNew(), timeout, cancellationToken) + : this(SystemClock.Instance, timeout, cancellationToken) { } - internal OperationContext(Stopwatch stopwatch, TimeSpan? timeout, CancellationToken cancellationToken) + internal OperationContext(IClock clock, TimeSpan? timeout, CancellationToken cancellationToken) { - Stopwatch = stopwatch; + Clock = Ensure.IsNotNull(clock, nameof(clock)); + Stopwatch = clock.StartStopwatch(); Timeout = Ensure.IsNullOrInfiniteOrGreaterThanOrEqualToZero(timeout, nameof(timeout)); CancellationToken = cancellationToken; RootContext = this; @@ -85,7 +85,11 @@ public CancellationToken CombinedCancellationToken return _combinedCancellationTokenSource.Token; } } - private Stopwatch Stopwatch { get; } + private IStopwatch Stopwatch { get; } + + private IClock Clock { get; } + + public TimeSpan Elapsed => Stopwatch.Elapsed; public TimeSpan? Timeout { get; } @@ -182,11 +186,10 @@ public OperationContext WithTimeout(TimeSpan timeout) timeout = remainingTimeout; } - return new OperationContext(timeout, CancellationToken) + return new OperationContext(Clock, timeout, CancellationToken) { RootContext = RootContext }; } } } - diff --git a/src/MongoDB.Driver/OperationExecutor.cs b/src/MongoDB.Driver/OperationExecutor.cs index 06c8534b70c..929c28b6063 100644 --- a/src/MongoDB.Driver/OperationExecutor.cs +++ b/src/MongoDB.Driver/OperationExecutor.cs @@ -14,7 +14,6 @@ */ using System; -using System.Threading; using System.Threading.Tasks; using MongoDB.Driver.Core; using MongoDB.Driver.Core.Bindings; @@ -39,73 +38,65 @@ public void Dispose() } public TResult ExecuteReadOperation( + OperationContext operationContext, IClientSessionHandle session, IReadOperation operation, ReadPreference readPreference, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken) + bool allowChannelPinning) { + Ensure.IsNotNull(operationContext, nameof(operationContext)); Ensure.IsNotNull(session, nameof(session)); Ensure.IsNotNull(operation, nameof(operation)); Ensure.IsNotNull(readPreference, nameof(readPreference)); - Ensure.IsNullOrValidTimeout(timeout, nameof(timeout)); ThrowIfDisposed(); - using var operationContext = new OperationContext(timeout, cancellationToken); using var binding = CreateReadBinding(session, readPreference, allowChannelPinning); return operation.Execute(operationContext, binding); } public async Task ExecuteReadOperationAsync( + OperationContext operationContext, IClientSessionHandle session, IReadOperation operation, ReadPreference readPreference, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken) + bool allowChannelPinning) { + Ensure.IsNotNull(operationContext, nameof(operationContext)); Ensure.IsNotNull(session, nameof(session)); Ensure.IsNotNull(operation, nameof(operation)); Ensure.IsNotNull(readPreference, nameof(readPreference)); - Ensure.IsNullOrValidTimeout(timeout, nameof(timeout)); ThrowIfDisposed(); - using var operationContext = new OperationContext(timeout, cancellationToken); using var binding = CreateReadBinding(session, readPreference, allowChannelPinning); return await operation.ExecuteAsync(operationContext, binding).ConfigureAwait(false); } public TResult ExecuteWriteOperation( + OperationContext operationContext, IClientSessionHandle session, IWriteOperation operation, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken) + bool allowChannelPinning) { + Ensure.IsNotNull(operationContext, nameof(operationContext)); Ensure.IsNotNull(session, nameof(session)); Ensure.IsNotNull(operation, nameof(operation)); - Ensure.IsNullOrValidTimeout(timeout, nameof(timeout)); ThrowIfDisposed(); - using var operationContext = new OperationContext(timeout, cancellationToken); using var binding = CreateReadWriteBinding(session, allowChannelPinning); return operation.Execute(operationContext, binding); } public async Task ExecuteWriteOperationAsync( + OperationContext operationContext, IClientSessionHandle session, IWriteOperation operation, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken) + bool allowChannelPinning) { + Ensure.IsNotNull(operationContext, nameof(operationContext)); Ensure.IsNotNull(session, nameof(session)); Ensure.IsNotNull(operation, nameof(operation)); - Ensure.IsNullOrValidTimeout(timeout, nameof(timeout)); ThrowIfDisposed(); - using var operationContext = new OperationContext(timeout, cancellationToken); using var binding = CreateReadWriteBinding(session, allowChannelPinning); return await operation.ExecuteAsync(operationContext, binding).ConfigureAwait(false); } diff --git a/src/MongoDB.Driver/TransactionExecutor.cs b/src/MongoDB.Driver/TransactionExecutor.cs index cb4e5daf9a8..32f94861a2e 100644 --- a/src/MongoDB.Driver/TransactionExecutor.cs +++ b/src/MongoDB.Driver/TransactionExecutor.cs @@ -1,4 +1,4 @@ -/* Copyright 2019-present MongoDB Inc. +/* Copyright 2010-present MongoDB Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,7 +18,6 @@ using System.Threading.Tasks; using MongoDB.Driver.Core.Bindings; using MongoDB.Driver.Core.Misc; -using MongoDB.Driver.Support; namespace MongoDB.Driver { @@ -27,7 +26,6 @@ internal static class TransactionExecutor // constants private const string TransientTransactionErrorLabel = "TransientTransactionError"; private const string UnknownTransactionCommitResultLabel = "UnknownTransactionCommitResult"; - private const int MaxTimeMSExpiredErrorCode = 50; private static readonly TimeSpan __transactionTimeout = TimeSpan.FromSeconds(120); public static TResult ExecuteWithRetries( @@ -37,13 +35,15 @@ public static TResult ExecuteWithRetries( IClock clock, CancellationToken cancellationToken) { - var startTime = clock.UtcNow; + var transactionTimeout = transactionOptions?.Timeout ?? clientSession.Options.DefaultTransactionOptions?.Timeout; + using var operationContext = new OperationContext(clock, transactionTimeout, cancellationToken); while (true) { clientSession.StartTransaction(transactionOptions); + clientSession.WrappedCoreSession.CurrentTransaction.OperationContext = operationContext; - var callbackOutcome = ExecuteCallback(clientSession, callback, startTime, clock, cancellationToken); + var callbackOutcome = ExecuteCallback(operationContext, clientSession, callback, cancellationToken); if (callbackOutcome.ShouldRetryTransaction) { continue; @@ -53,7 +53,7 @@ public static TResult ExecuteWithRetries( return callbackOutcome.Result; // assume callback intentionally ended the transaction } - var transactionHasBeenCommitted = CommitWithRetries(clientSession, startTime, clock, cancellationToken); + var transactionHasBeenCommitted = CommitWithRetries(operationContext, clientSession, cancellationToken); if (transactionHasBeenCommitted) { return callbackOutcome.Result; @@ -68,12 +68,15 @@ public static async Task ExecuteWithRetriesAsync( IClock clock, CancellationToken cancellationToken) { - var startTime = clock.UtcNow; + TimeSpan? transactionTimeout = transactionOptions?.Timeout ?? clientSession.Options.DefaultTransactionOptions?.Timeout; + using var operationContext = new OperationContext(clock, transactionTimeout, cancellationToken); + while (true) { clientSession.StartTransaction(transactionOptions); + clientSession.WrappedCoreSession.CurrentTransaction.OperationContext = operationContext; - var callbackOutcome = await ExecuteCallbackAsync(clientSession, callbackAsync, startTime, clock, cancellationToken).ConfigureAwait(false); + var callbackOutcome = await ExecuteCallbackAsync(operationContext, clientSession, callbackAsync, cancellationToken).ConfigureAwait(false); if (callbackOutcome.ShouldRetryTransaction) { continue; @@ -83,7 +86,7 @@ public static async Task ExecuteWithRetriesAsync( return callbackOutcome.Result; // assume callback intentionally ended the transaction } - var transactionHasBeenCommitted = await CommitWithRetriesAsync(clientSession, startTime, clock, cancellationToken).ConfigureAwait(false); + var transactionHasBeenCommitted = await CommitWithRetriesAsync(operationContext, clientSession, cancellationToken).ConfigureAwait(false); if (transactionHasBeenCommitted) { return callbackOutcome.Result; @@ -91,12 +94,13 @@ public static async Task ExecuteWithRetriesAsync( } } - private static bool HasTimedOut(DateTime startTime, DateTime currentTime) + private static bool HasTimedOut(OperationContext operationContext) { - return (currentTime - startTime) >= __transactionTimeout; + return operationContext.IsTimedOut() || + (operationContext.RootContext.Timeout == null && operationContext.RootContext.Elapsed > __transactionTimeout); } - private static CallbackOutcome ExecuteCallback(IClientSessionHandle clientSession, Func callback, DateTime startTime, IClock clock, CancellationToken cancellationToken) + private static CallbackOutcome ExecuteCallback(OperationContext operationContext, IClientSessionHandle clientSession, Func callback, CancellationToken cancellationToken) { try { @@ -107,10 +111,16 @@ private static CallbackOutcome ExecuteCallback(IClientSessionH { if (IsTransactionInStartingOrInProgressState(clientSession)) { - clientSession.AbortTransaction(cancellationToken); + AbortTransactionOptions abortOptions = null; + if (operationContext.IsRootContextTimeoutConfigured()) + { + abortOptions = new AbortTransactionOptions(operationContext.RootContext.Timeout); + } + + clientSession.AbortTransaction(abortOptions, cancellationToken); } - if (HasErrorLabel(ex, TransientTransactionErrorLabel) && !HasTimedOut(startTime, clock.UtcNow)) + if (HasErrorLabel(ex, TransientTransactionErrorLabel) && !HasTimedOut(operationContext)) { return new CallbackOutcome.WithShouldRetryTransaction(); } @@ -119,7 +129,7 @@ private static CallbackOutcome ExecuteCallback(IClientSessionH } } - private static async Task> ExecuteCallbackAsync(IClientSessionHandle clientSession, Func> callbackAsync, DateTime startTime, IClock clock, CancellationToken cancellationToken) + private static async Task> ExecuteCallbackAsync(OperationContext operationContext, IClientSessionHandle clientSession, Func> callbackAsync, CancellationToken cancellationToken) { try { @@ -130,10 +140,16 @@ private static async Task> ExecuteCallbackAsync.WithShouldRetryTransaction(); } @@ -142,24 +158,29 @@ private static async Task> ExecuteCallbackAsync CommitWithRetriesAsync(IClientSessionHandle clientSession, DateTime startTime, IClock clock, CancellationToken cancellationToken) + private static async Task CommitWithRetriesAsync(OperationContext operationContext, IClientSessionHandle clientSession, CancellationToken cancellationToken) { while (true) { try { - await clientSession.CommitTransactionAsync(cancellationToken).ConfigureAwait(false); + CommitTransactionOptions commitOptions = null; + if (operationContext.IsRootContextTimeoutConfigured()) + { + commitOptions = new CommitTransactionOptions(operationContext.RemainingTimeout); + } + + await clientSession.CommitTransactionAsync(commitOptions, cancellationToken).ConfigureAwait(false); return true; } catch (Exception ex) { - var now = clock.UtcNow; // call UtcNow once since we need to facilitate predictable mocking - if (ShouldRetryCommit(ex, startTime, now)) + if (ShouldRetryCommit(operationContext, ex)) { continue; } - if (HasErrorLabel(ex, TransientTransactionErrorLabel) && !HasTimedOut(startTime, now)) + if (HasErrorLabel(ex, TransientTransactionErrorLabel) && !HasTimedOut(operationContext)) { return false; // the transaction will be retried } @@ -211,7 +237,7 @@ private static bool HasErrorLabel(Exception ex, string errorLabel) private static bool IsMaxTimeMSExpiredException(Exception ex) { if (ex is MongoExecutionTimeoutException timeoutException && - timeoutException.Code == MaxTimeMSExpiredErrorCode) + timeoutException.Code == (int)ServerErrorCode.MaxTimeMSExpired) { return true; } @@ -222,7 +248,7 @@ private static bool IsMaxTimeMSExpiredException(Exception ex) if (writeConcernError != null) { var code = writeConcernError.GetValue("code", -1).ToInt32(); - if (code == MaxTimeMSExpiredErrorCode) + if (code == (int)ServerErrorCode.MaxTimeMSExpired) { return true; } @@ -246,11 +272,11 @@ private static bool IsTransactionInStartingOrInProgressState(IClientSessionHandl } } - private static bool ShouldRetryCommit(Exception ex, DateTime startTime, DateTime now) + private static bool ShouldRetryCommit(OperationContext operationContext, Exception ex) { return HasErrorLabel(ex, UnknownTransactionCommitResultLabel) && - !HasTimedOut(startTime, now) && + !HasTimedOut(operationContext) && !IsMaxTimeMSExpiredException(ex); } diff --git a/tests/MongoDB.Driver.Tests/ClientSessionHandleTests.cs b/tests/MongoDB.Driver.Tests/ClientSessionHandleTests.cs index d05c34abb28..36dab703623 100644 --- a/tests/MongoDB.Driver.Tests/ClientSessionHandleTests.cs +++ b/tests/MongoDB.Driver.Tests/ClientSessionHandleTests.cs @@ -1,4 +1,4 @@ -/* Copyright 2017-present MongoDB Inc. +/* Copyright 2010-present MongoDB Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,7 +25,6 @@ using MongoDB.Driver.Core.Clusters; using MongoDB.Driver.Core.Misc; using MongoDB.Driver.Core.TestHelpers; -using MongoDB.Driver.Support; using Moq; using Xunit; @@ -318,21 +317,7 @@ public void WithTransaction_callback_should_be_processed_with_expected_result( bool async) { var mockClock = CreateClockMock(DateTime.UtcNow, isRetryAttemptsWithTimeout, true); - var mockCoreSession = CreateCoreSessionMock(); - mockCoreSession.Setup(c => c.StartTransaction(It.IsAny())); - - // CommitTransaction - if (async) - { - mockCoreSession - .Setup(c => c.CommitTransactionAsync(It.IsAny())) - .Returns(Task.FromResult(0)); - } - else - { - mockCoreSession.Setup(c => c.CommitTransaction(It.IsAny())); - } // Initialize callbacks var mockCallbackProcessing = new Mock(); @@ -438,9 +423,6 @@ public void WithTransaction_callback_should_propagate_result(object value) public void WithTransaction_callback_with_a_custom_error_should_not_be_retried() { var mockCoreSession = CreateCoreSessionMock(); - mockCoreSession.Setup(c => c.StartTransaction(It.IsAny())); - mockCoreSession.Setup(c => c.AbortTransaction(It.IsAny())); // abort ignores exceptions - mockCoreSession.Setup(c => c.CommitTransaction(It.IsAny())); var subject = CreateSubject(coreSession: mockCoreSession.Object); @@ -453,13 +435,7 @@ public void WithTransaction_callback_with_a_custom_error_should_not_be_retried() [Fact] public void WithTransaction_callback_with_a_TransientTransactionError_and_exceeded_retry_timeout_should_not_be_retried() { - var now = DateTime.UtcNow; - var mockClock = new Mock(); - mockClock - .SetupSequence(c => c.UtcNow) - .Returns(now) - .Returns(now.AddSeconds(CalculateTime(true))); // the retry timeout has been exceeded - + var mockClock = CreateClockMock(DateTime.UtcNow, TimeSpan.FromSeconds(CalculateTime(true))); var subject = CreateSubject(clock: mockClock.Object); var exResult = Assert.Throws(() => subject.WithTransaction((handle, cancellationToken) => @@ -473,13 +449,7 @@ public void WithTransaction_callback_with_a_TransientTransactionError_and_exceed [ParameterAttributeData] public void WithTransaction_callback_with_a_UnknownTransactionCommitResult_should_not_be_retried([Values(true, false)] bool hasTimedOut) { - var now = DateTime.UtcNow; - var mockClock = new Mock(); - mockClock - .SetupSequence(c => c.UtcNow) - .Returns(now) - .Returns(now.AddSeconds(CalculateTime(hasTimedOut))); - + var mockClock = CreateClockMock(DateTime.UtcNow, TimeSpan.FromSeconds(CalculateTime(hasTimedOut))); var subject = CreateSubject(clock: mockClock.Object); var exResult = Assert.Throws(() => subject.WithTransaction((handle, cancellationToken) => @@ -491,43 +461,41 @@ public void WithTransaction_callback_with_a_UnknownTransactionCommitResult_shoul [Theory] // sync - [InlineData(null, new[] { WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 1, true, false)] - [InlineData(null, new[] { WithTransactionErrorState.TransientTransactionError }, false /*Should exception be thrown*/, 1, false, false)] + [InlineData(null, new[] { WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 1, false)] + [InlineData(null, new[] { WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 2, false)] - [InlineData(null, new[] { WithTransactionErrorState.ErrorWithoutLabel }, true /*Should exception be thrown*/, 1, false, false)] + [InlineData(null, new[] { WithTransactionErrorState.ErrorWithoutLabel }, true /*Should exception be thrown*/, 1, false)] - [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.TransientTransactionError }, false /*Should exception be thrown*/, 1, false, false)] - [InlineData(new[] { true, true }, new[] { WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.TransientTransactionError }, true /*Should exception be thrown*/, 1, null, false)] + [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 3, false)] + [InlineData(new[] { true }, new[] { WithTransactionErrorState.TransientTransactionError }, true /*Should exception be thrown*/, 1, false)] - [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 3, true, false)] - [InlineData(new[] { false, true }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, true /*Should exception be thrown*/, 2, null, false)] + [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 1, false)] + [InlineData(new[] { false, true }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult }, true /*Should exception be thrown*/, 1, false)] - [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 3, true, false)] + [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 1, false)] // async - [InlineData(null, new[] { WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 1, true, true)] - [InlineData(null, new[] { WithTransactionErrorState.TransientTransactionError }, false /*Should exception be thrown*/, 1, false, true)] + [InlineData(null, new[] { WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 1, true)] + [InlineData(null, new[] { WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 2, true)] - [InlineData(null, new[] { WithTransactionErrorState.ErrorWithoutLabel }, true /*Should exception be thrown*/, 1, false, true)] + [InlineData(null, new[] { WithTransactionErrorState.ErrorWithoutLabel }, true /*Should exception be thrown*/, 1, true)] - [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.TransientTransactionError }, false /*Should exception be thrown*/, 1, false, true)] - [InlineData(new[] { true, true }, new[] { WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.TransientTransactionError }, true /*Should exception be thrown*/, 1, null, true)] + [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.TransientTransactionError, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 3, true)] + [InlineData(new[] { true }, new[] { WithTransactionErrorState.TransientTransactionError }, true /*Should exception be thrown*/, 1, true)] - [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 3, true, true)] - [InlineData(new[] { false, true }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, true /*Should exception be thrown*/, 2, null, true)] + [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 1, true)] + [InlineData(new[] { false, true }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult }, true /*Should exception be thrown*/, 1, true)] - [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 3, true, true)] + [InlineData(new[] { false, false }, new[] { WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.UnknownTransactionCommitResult, WithTransactionErrorState.NoError }, false /*Should exception be thrown*/, 1, true)] public void WithTransaction_commit_after_callback_processing_should_be_processed_with_expected_result( bool[] isRetryAttemptsWithTimeout, // the array length should be the same with a number of failed attempts from `commitTransactionErrorStates` WithTransactionErrorState[] commitTransactionErrorStates, bool shouldExceptionBeThrown, - int expectedCommitTransactionAttempts, - bool? expectedFullTransactionBeRetriedState, + int transactionCallbackAttempts, bool async) { var now = DateTime.UtcNow; var mockClock = CreateClockMock(now, isRetryAttemptsWithTimeout, false); - var mockCoreSession = CreateCoreSessionMock(); // Initialize commit result @@ -566,34 +534,39 @@ public void WithTransaction_commit_after_callback_processing_should_be_processed var subject = CreateSubject(coreSession: mockCoreSession.Object, clock: mockClock.Object); - // Commit processing if (async) { + var callbackMock = new Mock>>(); + var exception = Record.ExceptionAsync(() => subject.WithTransactionAsync(callbackMock.Object)).GetAwaiter().GetResult(); + if (shouldExceptionBeThrown) { - Assert.ThrowsAnyAsync(() => TransactionExecutorReflector.CommitWithRetriesAsync(subject, now, mockClock.Object, CancellationToken.None)).GetAwaiter().GetResult(); + exception.Should().BeOfType(); } else { - var result = TransactionExecutorReflector.CommitWithRetriesAsync(subject, now, mockClock.Object, CancellationToken.None).Result; - expectedFullTransactionBeRetriedState.Should().Be(result); + exception.Should().BeNull(); } - mockCoreSession.Verify(handle => handle.CommitTransactionAsync(It.IsAny()), Times.Exactly(expectedCommitTransactionAttempts)); + callbackMock.Verify(c => c(It.IsAny(), It.IsAny()), Times.Exactly(transactionCallbackAttempts)); + mockCoreSession.Verify(handle => handle.CommitTransactionAsync(It.IsAny()), Times.Exactly(commitTransactionErrorStates.Length)); } else { + var callbackMock = new Mock>(); + var exception = Record.Exception(() => subject.WithTransaction(callbackMock.Object)); + if (shouldExceptionBeThrown) { - Assert.ThrowsAny(() => TransactionExecutorReflector.CommitWithRetries(subject, now, mockClock.Object, CancellationToken.None)); + exception.Should().BeOfType(); } else { - var result = TransactionExecutorReflector.CommitWithRetries(subject, now, mockClock.Object, CancellationToken.None); - expectedFullTransactionBeRetriedState.Should().Be(result); + exception.Should().BeNull(); } - mockCoreSession.Verify(handle => handle.CommitTransaction(It.IsAny()), Times.Exactly(expectedCommitTransactionAttempts)); + callbackMock.Verify(c => c(It.IsAny(), It.IsAny()), Times.Exactly(transactionCallbackAttempts)); + mockCoreSession.Verify(handle => handle.CommitTransaction(It.IsAny()), Times.Exactly(commitTransactionErrorStates.Length)); } } @@ -601,8 +574,6 @@ public void WithTransaction_commit_after_callback_processing_should_be_processed public void WithTransaction_should_set_valid_session_to_callback() { var mockCoreSession = CreateCoreSessionMock(); - mockCoreSession.Setup(c => c.StartTransaction(It.IsAny())); - mockCoreSession.Setup(c => c.CommitTransaction(It.IsAny())); var subject = CreateSubject(coreSession: mockCoreSession.Object); var result = subject.WithTransaction((session, cancellationToken) => session); @@ -618,9 +589,6 @@ public void WithTransaction_should_set_valid_session_to_callback() public void WithTransaction_with_error_in_callback_should_call_AbortTransaction_according_to_transaction_state(CoreTransactionState transactionState, bool shouldAbortTransactionBeCalled) { var mockCoreSession = CreateCoreSessionMock(); - mockCoreSession.Setup(c => c.StartTransaction(It.IsAny())); - mockCoreSession.Setup(c => c.AbortTransaction(It.IsAny())); // abort ignores exceptions - mockCoreSession.Setup(c => c.CommitTransaction(It.IsAny())); var subject = CreateSubject(coreSession: mockCoreSession.Object); subject.WrappedCoreSession.CurrentTransaction.SetState(transactionState); @@ -639,7 +607,6 @@ public void WithTransaction_with_error_in_StartTransaction_should_return_control mockCoreSession .Setup(c => c.StartTransaction(It.IsAny())) .Throws(); - mockCoreSession.Setup(c => c.CommitTransaction(It.IsAny())); var subject = CreateSubject(coreSession: mockCoreSession.Object); Assert.Throws(() => subject.WithTransaction((handle, cancellationToken) => 1)); @@ -652,8 +619,6 @@ public void WithTransaction_with_error_in_StartTransaction_should_return_control public void WithTransaction_without_errors_should_call_transaction_infrastructure_once() { var mockCoreSession = CreateCoreSessionMock(); - mockCoreSession.Setup(c => c.StartTransaction(It.IsAny())); - mockCoreSession.Setup(c => c.CommitTransaction(It.IsAny())); var subject = CreateSubject(coreSession: mockCoreSession.Object); SetupTransactionState(subject, true); @@ -734,8 +699,11 @@ private Mock CreateClockMock(DateTime now, bool[] isRetryAttemptsWithTim } var mockClock = new Mock(); + var mockStopwatch = new Mock(); + mockClock.Setup(m => m.StartStopwatch()).Returns(mockStopwatch.Object); var nowSetup = mockClock.SetupSequence(c => c.UtcNow); + var elapsedSetup = mockStopwatch.SetupSequence(w => w.Elapsed); if (shouldNowBeAdded) { nowSetup.Returns(now); @@ -744,7 +712,29 @@ private Mock CreateClockMock(DateTime now, bool[] isRetryAttemptsWithTim { var passedTime = CalculateTime(isTimeoutAttempt); nowSetup.Returns(now.AddSeconds(passedTime)); + elapsedSetup.Returns(TimeSpan.FromSeconds(passedTime)); } + + return mockClock; + } + + private Mock CreateClockMock(DateTime now, params TimeSpan[] intervals) + { + var mockClock = new Mock(); + + var nowSetup = mockClock.SetupSequence(c => c.UtcNow); + nowSetup.Returns(now); + var currentTime = now; + foreach (var interval in intervals) + { + currentTime += interval; + nowSetup.Returns(currentTime); + } + + var mockStopwatch = new Mock(); + mockStopwatch.SetupGet(w => w.Elapsed).Returns(() => mockClock.Object.UtcNow - now); + mockClock.Setup(m => m.StartStopwatch()).Returns(mockStopwatch.Object); + return mockClock; } @@ -770,11 +760,5 @@ internal static class ClientSessionHandleReflector internal static class TransactionExecutorReflector { public static TimeSpan __transactionTimeout() => (TimeSpan)Reflector.GetStaticFieldValue(typeof(TransactionExecutor), nameof(__transactionTimeout)); - - public static bool CommitWithRetries(IClientSessionHandle session, DateTime startTime, IClock clock, CancellationToken cancellationToken) - => (bool)Reflector.InvokeStatic(typeof(TransactionExecutor), nameof(CommitWithRetries), session, startTime, clock, cancellationToken); - - public static Task CommitWithRetriesAsync(IClientSessionHandle session, DateTime startTime, IClock clock, CancellationToken cancellationToken) - => (Task)Reflector.InvokeStatic(typeof(TransactionExecutor), nameof(CommitWithRetriesAsync), session, startTime, clock, cancellationToken); } } diff --git a/tests/MongoDB.Driver.Tests/Core/Misc/FrozenClock.cs b/tests/MongoDB.Driver.Tests/Core/Misc/FrozenClock.cs index dfffd3a3e71..3f2875b62f4 100644 --- a/tests/MongoDB.Driver.Tests/Core/Misc/FrozenClock.cs +++ b/tests/MongoDB.Driver.Tests/Core/Misc/FrozenClock.cs @@ -14,11 +14,11 @@ */ using System; -using MongoDB.Driver.Core.Misc; +using Moq; namespace MongoDB.Driver.Core.Misc { - public class FrozenClock : IClock + internal class FrozenClock : IClock { // public static methods public static FrozenClock FreezeUtcNow() @@ -39,7 +39,19 @@ public FrozenClock(DateTime utcNow) public DateTime UtcNow { get { return _utcNow; } - set { _utcNow = value; } + } + + public void AdvanceCurrentTime(TimeSpan timeSpan) + { + _utcNow += timeSpan; + } + + public IStopwatch StartStopwatch() + { + var startTime = _utcNow; + var mockStopwatch = new Mock(); + mockStopwatch.SetupGet(w => w.Elapsed).Returns(() => _utcNow - startTime); + return mockStopwatch.Object; } } } diff --git a/tests/MongoDB.Driver.Tests/Core/Misc/MetronomeTests.cs b/tests/MongoDB.Driver.Tests/Core/Misc/MetronomeTests.cs index 1444877e130..cad299155c0 100644 --- a/tests/MongoDB.Driver.Tests/Core/Misc/MetronomeTests.cs +++ b/tests/MongoDB.Driver.Tests/Core/Misc/MetronomeTests.cs @@ -15,9 +15,7 @@ using System; using System.Threading; -using System.Threading.Tasks; using FluentAssertions; -using MongoDB.Driver.Core.Async; using MongoDB.Driver.Core.Misc; using Xunit; @@ -73,7 +71,7 @@ public void GetNextTickDelay_should_be_infinite_if_period_is_infinite() public void GetNextTickDelay_should_be_threeQuarterPeriod_when_oneQuarterPeriod_past_the_last_tick() { var now = _clock.UtcNow; - _clock.UtcNow = _clock.UtcNow.Add(_quarterPeriod); + _clock.AdvanceCurrentTime(_quarterPeriod); _subject.GetNextTickDelay().Should().Be(_threeQuarterPeriod); _subject.NextTick.Should().Be(now + _period); } @@ -89,7 +87,7 @@ public void GetNextTickDelay_should_be_zero_when_first_instantiated() public void GetNextTickDelay_should_be_zero_when_time_equals_nextTick() { var now = _clock.UtcNow; - _clock.UtcNow = _clock.UtcNow.Add(_period); + _clock.AdvanceCurrentTime(_period); _subject.GetNextTickDelay().Should().Be(TimeSpan.Zero); _subject.NextTick.Should().Be(now + _period); } @@ -98,11 +96,11 @@ public void GetNextTickDelay_should_be_zero_when_time_equals_nextTick() public void GetNextTickDelay_should_not_advance_nextTick_when_called_more_than_once_during_the_same_period() { var now = _clock.UtcNow; - _clock.UtcNow = _clock.UtcNow.Add(_quarterPeriod); + _clock.AdvanceCurrentTime(_quarterPeriod); _subject.GetNextTickDelay().Should().Be(_threeQuarterPeriod); _subject.NextTick.Should().Be(now + _period); - _clock.UtcNow = _clock.UtcNow.Add(_quarterPeriod); + _clock.AdvanceCurrentTime(_quarterPeriod); _subject.GetNextTickDelay().Should().Be(_halfPeriod); _subject.NextTick.Should().Be(now + _period); } @@ -111,7 +109,7 @@ public void GetNextTickDelay_should_not_advance_nextTick_when_called_more_than_o public void GetNextTickDelay_should_skip_one_missed_tick() { var now = _clock.UtcNow; - _clock.UtcNow = _clock.UtcNow.Add(_period + _quarterPeriod); + _clock.AdvanceCurrentTime(_period + _quarterPeriod); _subject.GetNextTickDelay().Should().Be(_threeQuarterPeriod); _subject.NextTick.Should().Be(now + _period + _period); } @@ -120,7 +118,7 @@ public void GetNextTickDelay_should_skip_one_missed_tick() public void GetNextTickDelay_should_skip_two_missed_ticks() { var now = _clock.UtcNow; - _clock.UtcNow = _clock.UtcNow.Add(_period + _period + _quarterPeriod); + _clock.AdvanceCurrentTime(_period + _period + _quarterPeriod); _subject.GetNextTickDelay().Should().Be(_threeQuarterPeriod); _subject.NextTick.Should().Be(now + _period + _period + _period); } diff --git a/tests/MongoDB.Driver.Tests/Core/Operations/EndTransactionOperationTests.cs b/tests/MongoDB.Driver.Tests/Core/Operations/EndTransactionOperationTests.cs index ba929d55b27..067c5c45588 100644 --- a/tests/MongoDB.Driver.Tests/Core/Operations/EndTransactionOperationTests.cs +++ b/tests/MongoDB.Driver.Tests/Core/Operations/EndTransactionOperationTests.cs @@ -223,7 +223,7 @@ internal static class EndTransactionOperationReflector public static BsonDocument CreateCommand(this EndTransactionOperation obj) { var methodInfo = typeof(EndTransactionOperation).GetMethod("CreateCommand", BindingFlags.NonPublic | BindingFlags.Instance); - return (BsonDocument)methodInfo.Invoke(obj, null); + return (BsonDocument)methodInfo.Invoke(obj, [OperationContext.NoTimeout]); } } } diff --git a/tests/MongoDB.Driver.Tests/MockOperationExecutor.cs b/tests/MongoDB.Driver.Tests/MockOperationExecutor.cs index b4702eaae5b..ee6e0d015f3 100644 --- a/tests/MongoDB.Driver.Tests/MockOperationExecutor.cs +++ b/tests/MongoDB.Driver.Tests/MockOperationExecutor.cs @@ -63,20 +63,19 @@ public void EnqueueException(Exception exception) } public TResult ExecuteReadOperation( + OperationContext operationContext, IClientSessionHandle session, IReadOperation operation, ReadPreference readPreference, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken) + bool allowChannelPinning) { _calls.Enqueue(new ReadCall { Operation = operation, - CancellationToken = cancellationToken, + CancellationToken = operationContext.CancellationToken, ReadPreference = readPreference, SessionId = session?.WrappedCoreSession.Id, - Timeout = timeout, + Timeout = operationContext.Timeout, UsedImplicitSession = session == null || session.IsImplicit }); @@ -97,16 +96,15 @@ public TResult ExecuteReadOperation( } public Task ExecuteReadOperationAsync( + OperationContext operationContext, IClientSessionHandle session, IReadOperation operation, ReadPreference readPreference, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken) + bool allowChannelPinning) { try { - var result = ExecuteReadOperation(session, operation, readPreference, allowChannelPinning, timeout, cancellationToken); + var result = ExecuteReadOperation(operationContext, session, operation, readPreference, allowChannelPinning); return Task.FromResult(result); } catch (Exception ex) @@ -118,18 +116,17 @@ public Task ExecuteReadOperationAsync( } public TResult ExecuteWriteOperation( + OperationContext operationContext, IClientSessionHandle session, IWriteOperation operation, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken) + bool allowChannelPinning) { _calls.Enqueue(new WriteCall { Operation = operation, - CancellationToken = cancellationToken, + CancellationToken = operationContext.CancellationToken, SessionId = session?.WrappedCoreSession.Id, - Timeout = timeout, + Timeout = operationContext.Timeout, UsedImplicitSession = session == null || session.IsImplicit }); @@ -150,15 +147,14 @@ public TResult ExecuteWriteOperation( } public Task ExecuteWriteOperationAsync( + OperationContext operationContext, IClientSessionHandle session, IWriteOperation operation, - bool allowChannelPinning, - TimeSpan? timeout, - CancellationToken cancellationToken) + bool allowChannelPinning) { try { - var result = ExecuteWriteOperation(session, operation, allowChannelPinning, timeout, cancellationToken); + var result = ExecuteWriteOperation(operationContext, session, operation, allowChannelPinning); return Task.FromResult(result); } catch (Exception ex) diff --git a/tests/MongoDB.Driver.Tests/OperationContextTests.cs b/tests/MongoDB.Driver.Tests/OperationContextTests.cs index a3d00a95528..edc05c19e2d 100644 --- a/tests/MongoDB.Driver.Tests/OperationContextTests.cs +++ b/tests/MongoDB.Driver.Tests/OperationContextTests.cs @@ -15,10 +15,10 @@ using System; using System.Collections.Generic; -using System.Diagnostics; using System.Threading; using System.Threading.Tasks; using FluentAssertions; +using MongoDB.Driver.Core.Misc; using MongoDB.TestHelpers.XunitExtensions; using Xunit; @@ -30,11 +30,11 @@ public class OperationContextTests public void Constructor_should_initialize_properties() { var timeout = TimeSpan.FromSeconds(42); - var stopwatch = new Stopwatch(); + var clock = new FrozenClock(DateTime.UtcNow); using var cancellationTokenSource = new CancellationTokenSource(); var cancellationToken = cancellationTokenSource.Token; - var operationContext = new OperationContext(stopwatch, timeout, cancellationToken); + var operationContext = new OperationContext(clock, timeout, cancellationToken); operationContext.Timeout.Should().Be(timeout); operationContext.RemainingTimeout.Should().Be(timeout); @@ -54,46 +54,35 @@ public void Constructor_throws_on_negative_timeout() public void RemainingTimeout_should_calculate() { var timeout = TimeSpan.FromMilliseconds(500); - var stopwatch = Stopwatch.StartNew(); - Thread.Sleep(10); - stopwatch.Stop(); - - var operationContext = new OperationContext(stopwatch, timeout, CancellationToken.None); + var elapsed = TimeSpan.FromMilliseconds(10); + var subject = CreateSubject(timeout, elapsed, CancellationToken.None); - operationContext.RemainingTimeout.Should().Be(timeout - stopwatch.Elapsed); + subject.RemainingTimeout.Should().Be(timeout - elapsed); } [Fact] public void RemainingTimeout_should_return_infinite_for_infinite_timeout() { - var stopwatch = Stopwatch.StartNew(); - Thread.Sleep(10); - stopwatch.Stop(); - - var operationContext = new OperationContext(stopwatch, Timeout.InfiniteTimeSpan, CancellationToken.None); + var subject = CreateSubject(timeout: Timeout.InfiniteTimeSpan, elapsed: TimeSpan.FromMilliseconds(10)); - operationContext.RemainingTimeout.Should().Be(Timeout.InfiniteTimeSpan); + subject.RemainingTimeout.Should().Be(Timeout.InfiniteTimeSpan); } [Fact] public void RemainingTimeout_should_return_zero_for_timeout_context() { - var operationContext = new OperationContext(TimeSpan.FromMilliseconds(5), CancellationToken.None); - Thread.Sleep(10); + var subject = CreateSubject(timeout: TimeSpan.FromMilliseconds(5), elapsed: TimeSpan.FromMilliseconds(10)); - operationContext.RemainingTimeout.Should().Be(TimeSpan.Zero); + subject.RemainingTimeout.Should().Be(TimeSpan.Zero); } [Theory] [MemberData(nameof(IsTimedOut_test_cases))] - public void IsTimedOut_should_return_expected_result(bool expected, TimeSpan timeout, TimeSpan waitTime) + public void IsTimedOut_should_return_expected_result(bool expected, TimeSpan timeout, TimeSpan elapsed) { - var stopwatch = Stopwatch.StartNew(); - Thread.Sleep(waitTime); - stopwatch.Stop(); + var subject = CreateSubject(timeout, elapsed); - var operationContext = new OperationContext(stopwatch, timeout, CancellationToken.None); - var result = operationContext.IsTimedOut(); + var result = subject.IsTimedOut(); result.Should().Be(expected); } @@ -108,9 +97,9 @@ public void IsTimedOut_should_return_expected_result(bool expected, TimeSpan tim [Fact] public void ThrowIfTimedOutOrCanceled_should_not_throw_if_no_timeout_and_no_cancellation() { - var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); + var subject = CreateSubject(timeout: TimeSpan.FromMilliseconds(20), elapsed: TimeSpan.FromMilliseconds(10)); - var exception = Record.Exception(() => operationContext.ThrowIfTimedOutOrCanceled()); + var exception = Record.Exception(() => subject.ThrowIfTimedOutOrCanceled()); exception.Should().BeNull(); } @@ -118,10 +107,9 @@ public void ThrowIfTimedOutOrCanceled_should_not_throw_if_no_timeout_and_no_canc [Fact] public void ThrowIfTimedOutOrCanceled_throws_on_timeout() { - var operationContext = new OperationContext(TimeSpan.FromMilliseconds(10), CancellationToken.None); - Thread.Sleep(20); + var subject = CreateSubject(timeout: TimeSpan.FromMilliseconds(10), elapsed: TimeSpan.FromMilliseconds(20)); - var exception = Record.Exception(() => operationContext.ThrowIfTimedOutOrCanceled()); + var exception = Record.Exception(() => subject.ThrowIfTimedOutOrCanceled()); exception.Should().BeOfType(); } @@ -130,10 +118,10 @@ public void ThrowIfTimedOutOrCanceled_throws_on_timeout() public void ThrowIfTimedOutOrCanceled_throws_on_cancellation() { using var cancellationSource = new CancellationTokenSource(); - var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, cancellationSource.Token); + var subject = CreateSubject(timeout: Timeout.InfiniteTimeSpan, elapsed: TimeSpan.Zero, cancellationSource.Token); cancellationSource.Cancel(); - var exception = Record.Exception(() => operationContext.ThrowIfTimedOutOrCanceled()); + var exception = Record.Exception(() => subject.ThrowIfTimedOutOrCanceled()); exception.Should().BeOfType(); } @@ -142,11 +130,10 @@ public void ThrowIfTimedOutOrCanceled_throws_on_cancellation() public void ThrowIfTimedOutOrCanceled_throws_CancelledException_when_timedout_and_cancelled() { using var cancellationSource = new CancellationTokenSource(); - var operationContext = new OperationContext(TimeSpan.FromMilliseconds(10), cancellationSource.Token); - Thread.Sleep(20); + var subject = CreateSubject(timeout: TimeSpan.FromMilliseconds(10), elapsed: TimeSpan.FromMilliseconds(20), cancellationSource.Token); cancellationSource.Cancel(); - var exception = Record.Exception(() => operationContext.ThrowIfTimedOutOrCanceled()); + var exception = Record.Exception(() => subject.ThrowIfTimedOutOrCanceled()); exception.Should().BeOfType(); } @@ -156,12 +143,11 @@ public void ThrowIfTimedOutOrCanceled_throws_CancelledException_when_timedout_an public async Task Wait_should_throw_if_context_is_timedout([Values(true, false)] bool async) { var taskCompletionSource = new TaskCompletionSource(); - var operationContext = new OperationContext(TimeSpan.FromMilliseconds(10), CancellationToken.None); - Thread.Sleep(20); + var subject = CreateSubject(timeout: TimeSpan.FromMilliseconds(10), elapsed: TimeSpan.FromMilliseconds(20)); var exception = async ? - await Record.ExceptionAsync(() => operationContext.WaitTaskAsync(taskCompletionSource.Task)) : - Record.Exception(() => operationContext.WaitTask(taskCompletionSource.Task)); + await Record.ExceptionAsync(() => subject.WaitTaskAsync(taskCompletionSource.Task)) : + Record.Exception(() => subject.WaitTask(taskCompletionSource.Task)); exception.Should().BeOfType(); } @@ -173,11 +159,11 @@ public async Task Wait_should_throw_if_context_is_cancelled([Values(true, false) var taskCompletionSource = new TaskCompletionSource(); var cancellationTokenSource = new CancellationTokenSource(); cancellationTokenSource.Cancel(); - var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, cancellationTokenSource.Token); + var subject = CreateSubject(timeout: Timeout.InfiniteTimeSpan, elapsed: TimeSpan.Zero, cancellationTokenSource.Token); var exception = async ? - await Record.ExceptionAsync(() => operationContext.WaitTaskAsync(taskCompletionSource.Task)) : - Record.Exception(() => operationContext.WaitTask(taskCompletionSource.Task)); + await Record.ExceptionAsync(() => subject.WaitTaskAsync(taskCompletionSource.Task)) : + Record.Exception(() => subject.WaitTask(taskCompletionSource.Task)); exception.Should().BeOfType(); } @@ -188,11 +174,11 @@ public async Task Wait_should_rethrow_on_failed_task([Values(true, false)] bool { var ex = new InvalidOperationException(); var task = Task.FromException(ex); - var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); + var subject = CreateSubject(timeout: Timeout.InfiniteTimeSpan, elapsed: TimeSpan.FromMilliseconds(20)); var exception = async ? - await Record.ExceptionAsync(() => operationContext.WaitTaskAsync(task)) : - Record.Exception(() => operationContext.WaitTask(task)); + await Record.ExceptionAsync(() => subject.WaitTaskAsync(task)) : + Record.Exception(() => subject.WaitTask(task)); exception.Should().Be(ex); } @@ -203,20 +189,20 @@ public async Task Wait_should_rethrow_on_failed_promise_task([Values(true, false { var ex = new InvalidOperationException("Ups!"); var taskCompletionSource = new TaskCompletionSource(); - var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); + var subject = CreateSubject(timeout: Timeout.InfiniteTimeSpan, elapsed: TimeSpan.Zero); var task = Task.Run(async () => { if (async) { - await operationContext.WaitTaskAsync(taskCompletionSource.Task); + await subject.WaitTaskAsync(taskCompletionSource.Task); } else { - operationContext.WaitTask(taskCompletionSource.Task); + subject.WaitTask(taskCompletionSource.Task); } }); - Thread.Sleep(20); + Thread.Sleep(10); taskCompletionSource.SetException(ex); var exception = await Record.ExceptionAsync(() => task); @@ -228,11 +214,11 @@ public async Task Wait_should_rethrow_on_failed_promise_task([Values(true, false public async Task Wait_should_throw_on_timeout([Values(true, false)] bool async) { var taskCompletionSource = new TaskCompletionSource(); - var operationContext = new OperationContext(TimeSpan.FromMilliseconds(20), CancellationToken.None); + var subject = CreateSubject(timeout: TimeSpan.FromMilliseconds(10), elapsed: TimeSpan.FromMilliseconds(20)); var exception = async ? - await Record.ExceptionAsync(() => operationContext.WaitTaskAsync(taskCompletionSource.Task)) : - Record.Exception(() => operationContext.WaitTask(taskCompletionSource.Task)); + await Record.ExceptionAsync(() => subject.WaitTaskAsync(taskCompletionSource.Task)) : + Record.Exception(() => subject.WaitTask(taskCompletionSource.Task)); exception.Should().BeOfType(); } @@ -242,12 +228,11 @@ await Record.ExceptionAsync(() => operationContext.WaitTaskAsync(taskCompletionS public async Task Wait_should_not_throw_on_resolved_task_with_timedout_context([Values(true, false)] bool async) { var task = Task.FromResult(42); - var operationContext = new OperationContext(TimeSpan.FromMilliseconds(10), CancellationToken.None); - Thread.Sleep(20); + var subject = CreateSubject(timeout: TimeSpan.FromMilliseconds(10), elapsed: TimeSpan.FromMilliseconds(20)); var exception = async ? - await Record.ExceptionAsync(() => operationContext.WaitTaskAsync(task)) : - Record.Exception(() => operationContext.WaitTask(task)); + await Record.ExceptionAsync(() => subject.WaitTaskAsync(task)) : + Record.Exception(() => subject.WaitTask(task)); exception.Should().BeNull(); } @@ -257,8 +242,9 @@ await Record.ExceptionAsync(() => operationContext.WaitTaskAsync(task)) : [MemberData(nameof(WithTimeout_test_cases))] public void WithTimeout_should_calculate_proper_timeout(TimeSpan expected, TimeSpan originalTimeout, TimeSpan newTimeout) { - var operationContext = new OperationContext(new Stopwatch(), originalTimeout, CancellationToken.None); - var resultContext = operationContext.WithTimeout(newTimeout); + var subject = CreateSubject(timeout: originalTimeout, elapsed: TimeSpan.Zero); + + var resultContext = subject.WithTimeout(newTimeout); resultContext.Timeout.Should().Be(expected); } @@ -275,16 +261,16 @@ public void WithTimeout_should_calculate_proper_timeout(TimeSpan expected, TimeS [Fact] public void WithTimeout_should_set_RootContext() { - var operationContext = new OperationContext(new Stopwatch(), Timeout.InfiniteTimeSpan, CancellationToken.None); - var resultContext = operationContext.WithTimeout(TimeSpan.FromSeconds(10)); + var rootContext = CreateSubject(timeout: Timeout.InfiniteTimeSpan, elapsed: TimeSpan.Zero); + var resultContext = rootContext.WithTimeout(TimeSpan.FromSeconds(10)); - resultContext.RootContext.Should().Be(operationContext); + resultContext.RootContext.Should().Be(rootContext); } [Fact] public void WithTimeout_should_preserve_RootContext() { - var rootContext = new OperationContext(new Stopwatch(), Timeout.InfiniteTimeSpan, CancellationToken.None); + var rootContext = CreateSubject(timeout: Timeout.InfiniteTimeSpan, elapsed: TimeSpan.Zero); var intermediateContext = rootContext.WithTimeout(TimeSpan.FromSeconds(200)); var resultContext = intermediateContext.WithTimeout(TimeSpan.FromSeconds(10)); @@ -295,11 +281,10 @@ public void WithTimeout_should_preserve_RootContext() [Fact] public void WithTimeout_should_create_timed_out_context_on_timed_out_context() { - var operationContext = new OperationContext(TimeSpan.FromMilliseconds(5), CancellationToken.None); - Thread.Sleep(10); - operationContext.IsTimedOut().Should().BeTrue(); + var rootContext = CreateSubject(timeout: TimeSpan.FromMilliseconds(5), elapsed: TimeSpan.FromMilliseconds(10)); + rootContext.IsTimedOut().Should().BeTrue(); - var resultContext = operationContext.WithTimeout(TimeSpan.FromSeconds(10)); + var resultContext = rootContext.WithTimeout(TimeSpan.FromSeconds(7)); resultContext.IsTimedOut().Should().BeTrue(); } @@ -307,13 +292,26 @@ public void WithTimeout_should_create_timed_out_context_on_timed_out_context() [Fact] public void WithTimeout_throws_on_negative_timeout() { - var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); + var rootContext = CreateSubject(timeout: Timeout.InfiniteTimeSpan, elapsed: TimeSpan.Zero); - var exception = Record.Exception(() => operationContext.WithTimeout(TimeSpan.FromSeconds(-5))); + var exception = Record.Exception(() => rootContext.WithTimeout(TimeSpan.FromSeconds(-5))); exception.Should().BeOfType() .Subject.ParamName.Should().Be("timeout"); } + + private static OperationContext CreateSubject(TimeSpan? timeout, TimeSpan elapsed = default, CancellationToken cancellationToken = default) + { + var clock = new FrozenClock(DateTime.UtcNow); + var result = new OperationContext(clock, timeout, cancellationToken); + + if (elapsed != TimeSpan.Zero) + { + clock.AdvanceCurrentTime(elapsed); + } + + return result; + } } } diff --git a/tests/MongoDB.Driver.Tests/OperationExecutorTests.cs b/tests/MongoDB.Driver.Tests/OperationExecutorTests.cs index c703c27acdf..3512a1652b4 100644 --- a/tests/MongoDB.Driver.Tests/OperationExecutorTests.cs +++ b/tests/MongoDB.Driver.Tests/OperationExecutorTests.cs @@ -43,11 +43,12 @@ public void StartImplicitSession_should_call_cluster_StartSession() public async Task ExecuteReadOperation_throws_on_null_operation([Values(true, false)] bool async) { var subject = CreateSubject(out _); + var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); var session = Mock.Of(); var exception = async ? - await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(session, null, ReadPreference.Primary, true, Timeout.InfiniteTimeSpan, CancellationToken.None)) : - Record.Exception(() => subject.ExecuteReadOperation(session, null, ReadPreference.Primary, true, Timeout.InfiniteTimeSpan, CancellationToken.None)); + await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(operationContext, session, null, ReadPreference.Primary, true)) : + Record.Exception(() => subject.ExecuteReadOperation(operationContext, session, null, ReadPreference.Primary, true)); exception.Should().BeOfType() .Subject.ParamName.Should().Be("operation"); @@ -58,12 +59,13 @@ await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(sess public async Task ExecuteReadOperation_throws_on_null_readPreference([Values(true, false)] bool async) { var subject = CreateSubject(out _); + var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); var operation = Mock.Of>(); var session = Mock.Of(); var exception = async ? - await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(session, operation, null, true, Timeout.InfiniteTimeSpan, CancellationToken.None)) : - Record.Exception(() => subject.ExecuteReadOperation(session, operation, null, true, Timeout.InfiniteTimeSpan, CancellationToken.None)); + await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(operationContext, session, operation, null, true)) : + Record.Exception(() => subject.ExecuteReadOperation(operationContext, session, operation, null, true)); exception.Should().BeOfType() .Subject.ParamName.Should().Be("readPreference"); @@ -74,11 +76,12 @@ await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(session, ope public async Task ExecuteReadOperation_throws_on_null_session([Values(true, false)] bool async) { var subject = CreateSubject(out _); + var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); var operation = Mock.Of>(); var exception = async ? - await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(null, operation, ReadPreference.Primary, true, Timeout.InfiniteTimeSpan, CancellationToken.None)) : - Record.Exception(() => subject.ExecuteReadOperation(null, operation, ReadPreference.Primary, true, Timeout.InfiniteTimeSpan, CancellationToken.None)); + await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(operationContext, null, operation, ReadPreference.Primary, true)) : + Record.Exception(() => subject.ExecuteReadOperation(operationContext, null, operation, ReadPreference.Primary, true)); exception.Should().BeOfType() .Subject.ParamName.Should().Be("session"); @@ -89,11 +92,12 @@ await Record.ExceptionAsync(() => subject.ExecuteReadOperationAsync(null, operat public async Task ExecuteWriteOperation_throws_on_null_operation([Values(true, false)] bool async) { var subject = CreateSubject(out _); + var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); var session = Mock.Of(); var exception = async ? - await Record.ExceptionAsync(() => subject.ExecuteWriteOperationAsync(session, null, true, Timeout.InfiniteTimeSpan, CancellationToken.None)) : - Record.Exception(() => subject.ExecuteWriteOperation(session, null, true, Timeout.InfiniteTimeSpan, CancellationToken.None)); + await Record.ExceptionAsync(() => subject.ExecuteWriteOperationAsync(operationContext, session, null, true)) : + Record.Exception(() => subject.ExecuteWriteOperation(operationContext, session, null, true)); exception.Should().BeOfType() .Subject.ParamName.Should().Be("operation"); @@ -104,11 +108,12 @@ await Record.ExceptionAsync(() => subject.ExecuteWriteOperationAsync(ses public async Task ExecuteWriteOperation_throws_on_null_session([Values(true, false)] bool async) { var subject = CreateSubject(out _); + var operationContext = new OperationContext(Timeout.InfiniteTimeSpan, CancellationToken.None); var operation = Mock.Of>(); var exception = async ? - await Record.ExceptionAsync(() => subject.ExecuteWriteOperationAsync(null, operation, true, Timeout.InfiniteTimeSpan, CancellationToken.None)) : - Record.Exception(() => subject.ExecuteWriteOperation(null, operation, true, Timeout.InfiniteTimeSpan, CancellationToken.None)); + await Record.ExceptionAsync(() => subject.ExecuteWriteOperationAsync(operationContext, null, operation, true)) : + Record.Exception(() => subject.ExecuteWriteOperation(operationContext, null, operation, true)); exception.Should().BeOfType() .Subject.ParamName.Should().Be("session");