diff --git a/config/spotbugs/exclude.xml b/config/spotbugs/exclude.xml index 20684680865..e2fd0ae362c 100644 --- a/config/spotbugs/exclude.xml +++ b/config/spotbugs/exclude.xml @@ -15,9 +15,13 @@ --> + + + + + + + + + + + + diff --git a/driver-core/src/main/com/mongodb/MongoInterruptedException.java b/driver-core/src/main/com/mongodb/MongoInterruptedException.java index e0adce7978c..774fb430a8c 100644 --- a/driver-core/src/main/com/mongodb/MongoInterruptedException.java +++ b/driver-core/src/main/com/mongodb/MongoInterruptedException.java @@ -29,7 +29,7 @@ /** * A driver-specific non-checked counterpart to {@link InterruptedException}. - * Before this exception is thrown, the {@linkplain Thread#isInterrupted() interrupt status} of the thread will have been set + * Before this exception is thrown, the {@linkplain Thread#isInterrupted() interrupted status} of the thread will have been set * unless the {@linkplain #getCause() cause} is {@link InterruptedIOException}, in which case the driver leaves the status as is. *

* The Java SE API uses exceptions different from {@link InterruptedException} to communicate the same information:

diff --git a/driver-core/src/main/com/mongodb/assertions/Assertions.java b/driver-core/src/main/com/mongodb/assertions/Assertions.java index bf38638dc6d..e80e8189a90 100644 --- a/driver-core/src/main/com/mongodb/assertions/Assertions.java +++ b/driver-core/src/main/com/mongodb/assertions/Assertions.java @@ -205,7 +205,7 @@ public static boolean assertFalse(final boolean value) throws AssertionError { } /** - * @throws AssertionError Always + * @throws AssertionError Always. * @return Never completes normally. The return type is {@link AssertionError} to allow writing {@code throw fail()}. * This may be helpful in non-{@code void} methods. */ @@ -215,7 +215,7 @@ public static AssertionError fail() throws AssertionError { /** * @param msg The failure message. - * @throws AssertionError Always + * @throws AssertionError Always. * @return Never completes normally. The return type is {@link AssertionError} to allow writing {@code throw fail("failure message")}. * This may be helpful in non-{@code void} methods. */ @@ -223,6 +223,16 @@ public static AssertionError fail(final String msg) throws AssertionError { throw new AssertionError(assertNotNull(msg)); } + /** + * @param cause The {@linkplain AssertionError#getCause() cause}. + * @throws AssertionError Always. + * @return Never completes normally. The return type is {@link AssertionError} to allow writing {@code throw fail(cause)}. + * This may be helpful in non-{@code void} methods. + */ + public static AssertionError fail(final Throwable cause) throws AssertionError { + throw new AssertionError(null, cause); + } + /** * @param supplier the supplier to check * @return {@code supplier.get()} diff --git a/driver-core/src/main/com/mongodb/connection/AsyncTransportSettings.java b/driver-core/src/main/com/mongodb/connection/AsyncTransportSettings.java index 8e259392313..489ada57020 100644 --- a/driver-core/src/main/com/mongodb/connection/AsyncTransportSettings.java +++ b/driver-core/src/main/com/mongodb/connection/AsyncTransportSettings.java @@ -52,13 +52,15 @@ private Builder() { } /** - * The executor service, intended to be used exclusively by the mongo - * client. Closing the mongo client will result in {@linkplain ExecutorService#shutdown() orderly shutdown} - * of the executor service. - * - *

When {@linkplain SslSettings#isEnabled() TLS is not enabled}, see + * The {@link ExecutorService}, intended to be used exclusively by the {@code MongoClient}. + *

+ * {@linkplain AutoCloseable#close() Closing} the {@code MongoClient} results in + * {@linkplain ExecutorService#shutdown() orderly shutdown} of the {@code executorService}. + * The application must not directly shut down the {@code executorService}. + *

+ * When {@linkplain SslSettings#isEnabled() TLS is not enabled}, see * {@link java.nio.channels.AsynchronousChannelGroup#withThreadPool(ExecutorService)} - * for additional requirements for the executor service. + * for additional requirements for the {@code executorService}. * * @param executorService the executor service * @return this diff --git a/driver-core/src/main/com/mongodb/connection/NettyTransportSettings.java b/driver-core/src/main/com/mongodb/connection/NettyTransportSettings.java index cb3a7c7c090..c80ed298540 100644 --- a/driver-core/src/main/com/mongodb/connection/NettyTransportSettings.java +++ b/driver-core/src/main/com/mongodb/connection/NettyTransportSettings.java @@ -85,8 +85,9 @@ public Builder socketChannelClass(final Class socketCha /** * Sets the event loop group. - * - *

The application is responsible for shutting down the provided {@code eventLoopGroup}

+ *

+ * The application is responsible for shutting down the provided {@code eventLoopGroup}. + * It must not be shut down before or concurrently with {@linkplain AutoCloseable#close() closing} the {@code MongoClient}. * * @param eventLoopGroup the event loop group that all channels created by this factory will be a part of * @return this diff --git a/driver-core/src/main/com/mongodb/internal/Locks.java b/driver-core/src/main/com/mongodb/internal/Locks.java index 042fc9fd69f..66160636e22 100644 --- a/driver-core/src/main/com/mongodb/internal/Locks.java +++ b/driver-core/src/main/com/mongodb/internal/Locks.java @@ -26,7 +26,7 @@ import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException; /** - *

This class is not part of the public API and may be removed or changed at any time

+ * This class is not part of the public API and may be removed or changed at any time. */ public final class Locks { public static void withLock(final Lock lock, final Runnable action) { @@ -41,8 +41,7 @@ public static void withInterruptibleLock(final StampedLock lock, final Runnable try { stamp = lock.writeLockInterruptibly(); } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MongoInterruptedException("Interrupted waiting for lock", e); + throw interruptAndCreateMongoInterruptedException("Interrupted waiting for lock", e); } try { runnable.run(); diff --git a/driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java b/driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java index f5fb6358b87..37b55afe6af 100644 --- a/driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java +++ b/driver-core/src/main/com/mongodb/internal/async/AsyncRunnable.java @@ -20,6 +20,7 @@ import com.mongodb.internal.async.function.LoopControl; import com.mongodb.internal.async.function.RetryControl; import com.mongodb.internal.async.function.RetryingAsyncCallbackSupplier; +import com.mongodb.internal.thread.AsyncClientExecutor; import java.util.function.BooleanSupplier; import java.util.function.Predicate; @@ -233,6 +234,8 @@ default AsyncSupplier thenSupply(final AsyncSupplier supplier) { default AsyncRunnable thenRunRetryingWhile(final AsyncRunnable runnable, final Predicate shouldRetry) { return thenRun(callback -> { new RetryingAsyncCallbackSupplier( + // `AsyncClientExecutor` is not needed, given the contract of `SimpleRetryPolicy`, `RetryingAsyncCallbackSupplier` + AsyncClientExecutor.NO_OP, new RetryControl<>(new SimpleRetryPolicy(shouldRetry)), // `finish` is required here instead of `unsafeFinish` // because only `finish` meets the contract of diff --git a/driver-core/src/main/com/mongodb/internal/async/SimpleRetryPolicy.java b/driver-core/src/main/com/mongodb/internal/async/SimpleRetryPolicy.java index ade2a70c119..3854c77a891 100644 --- a/driver-core/src/main/com/mongodb/internal/async/SimpleRetryPolicy.java +++ b/driver-core/src/main/com/mongodb/internal/async/SimpleRetryPolicy.java @@ -19,8 +19,12 @@ import com.mongodb.internal.async.function.RetryPolicy; import com.mongodb.internal.async.function.RetryPolicy.Decision.RetryAttemptInfo; +import java.time.Duration; import java.util.function.Predicate; +/** + * {@link RetryAttemptInfo#getBackoff()} is always {@link Duration#isZero() zero}. + */ final class SimpleRetryPolicy implements RetryPolicy { private final Predicate shouldRetry; @@ -30,6 +34,6 @@ final class SimpleRetryPolicy implements RetryPolicy { @Override public Decision onAttemptFailure(final RetryContext retryContext, final Throwable attemptFailedResult) { - return new Decision(attemptFailedResult, shouldRetry.test(attemptFailedResult) ? new RetryAttemptInfo() : null); + return new Decision(attemptFailedResult, shouldRetry.test(attemptFailedResult) ? new RetryAttemptInfo(Duration.ZERO) : null); } } diff --git a/driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java b/driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java index 11da1c97f75..cefab04f312 100644 --- a/driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java +++ b/driver-core/src/main/com/mongodb/internal/async/SingleResultCallback.java @@ -37,7 +37,7 @@ public interface SingleResultCallback { * @param result the result, which may be null. Always null if e is not null. * @param t the throwable, or null if the operation completed normally * @throws RuntimeException Never. - * @throws Error Never, on the best effort basis. + * @throws Error Never, on the best-effort basis. */ void onResult(@Nullable T result, @Nullable Throwable t); diff --git a/driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackFunction.java b/driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackFunction.java index a869af91a39..3efe1fa97bf 100644 --- a/driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackFunction.java +++ b/driver-core/src/main/com/mongodb/internal/async/function/AsyncCallbackFunction.java @@ -51,7 +51,7 @@ public interface AsyncCallbackFunction { * @param callback A consumer of a result, {@link SingleResultCallback#onResult(Object, Throwable) completed} after * (in the happens-before order) the asynchronous function completes. * @throws RuntimeException Never. Exceptions must be relayed to the {@code callback}. - * @throws Error Never, on the best effort basis. Errors should be relayed to the {@code callback}. + * @throws Error Never, on the best-effort basis. Errors should be relayed to the {@code callback}. */ void apply(P a, SingleResultCallback callback); } diff --git a/driver-core/src/main/com/mongodb/internal/async/function/RetryPolicy.java b/driver-core/src/main/com/mongodb/internal/async/function/RetryPolicy.java index 6cbedb1ac2c..a84145b73e0 100644 --- a/driver-core/src/main/com/mongodb/internal/async/function/RetryPolicy.java +++ b/driver-core/src/main/com/mongodb/internal/async/function/RetryPolicy.java @@ -18,9 +18,11 @@ import com.mongodb.annotations.NotThreadSafe; import com.mongodb.lang.Nullable; +import java.time.Duration; import java.util.Optional; import java.util.function.Supplier; +import static com.mongodb.assertions.Assertions.assertFalse; import static com.mongodb.assertions.Assertions.assertNotNull; /** @@ -87,12 +89,25 @@ public String toString() { * The information needed to start a retry attempt. */ public static final class RetryAttemptInfo { - public RetryAttemptInfo() { + private final Duration backoff; + + public RetryAttemptInfo(final Duration backoff) { + assertFalse(backoff.isNegative()); + this.backoff = backoff; + } + + /** + * A non-{@linkplain Duration#isNegative() negative} backoff. + */ + public Duration getBackoff() { + return backoff; } @Override public String toString() { - return "RetryAttemptInfo{}"; + return "RetryAttemptInfo{" + + "backoff=" + backoff + + '}'; } } } diff --git a/driver-core/src/main/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplier.java b/driver-core/src/main/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplier.java index abdfefe55e4..0a4c5761757 100644 --- a/driver-core/src/main/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplier.java +++ b/driver-core/src/main/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplier.java @@ -16,10 +16,14 @@ package com.mongodb.internal.async.function; import com.mongodb.annotations.NotThreadSafe; +import com.mongodb.internal.async.MutableValue; import com.mongodb.internal.async.SingleResultCallback; -import com.mongodb.lang.Nullable; +import com.mongodb.internal.async.function.RetryPolicy.Decision.RetryAttemptInfo; +import com.mongodb.internal.thread.AsyncClientExecutor; -import static com.mongodb.assertions.Assertions.assertNotNull; +import java.time.Duration; + +import static com.mongodb.internal.async.AsyncRunnable.beginAsync; /** * A decorator that implements automatic retrying of failed executions of an {@link AsyncCallbackSupplier}. @@ -34,53 +38,45 @@ */ @NotThreadSafe public final class RetryingAsyncCallbackSupplier implements AsyncCallbackSupplier { + private final AsyncClientExecutor clientExecutor; private final RetryControl control; private final AsyncCallbackSupplier asyncFunction; /** + * @param clientExecutor For {@linkplain AsyncClientExecutor#sleepAsync(Duration, SingleResultCallback) delaying} attempts + * according to {@link RetryAttemptInfo#getBackoff()}. * @param control The {@link RetryControl} to control the new {@link RetryingAsyncCallbackSupplier}. * @param asyncFunction The retryable {@link AsyncCallbackSupplier} to be decorated. */ - public RetryingAsyncCallbackSupplier(final RetryControl control, final AsyncCallbackSupplier asyncFunction) { + public RetryingAsyncCallbackSupplier( + final AsyncClientExecutor clientExecutor, + final RetryControl control, + final AsyncCallbackSupplier asyncFunction) { + this.clientExecutor = clientExecutor; this.control = control; this.asyncFunction = asyncFunction; } @Override public void get(final SingleResultCallback callback) { - // `asyncFunction` and `callback` are the only externally provided pieces of code for which we do not need to care about - // them throwing exceptions. If they do, that violates their contract and there is nothing we should do about it. - asyncFunction.get(new RetryingCallback(callback)); - } - - /** - * This callback is allowed to be completed more than once. - */ - @NotThreadSafe - private class RetryingCallback implements SingleResultCallback { - private final SingleResultCallback wrapped; - - RetryingCallback(final SingleResultCallback callback) { - wrapped = callback; - } - - @Override - public void onResult(@Nullable final R attemptSuccessfulResult, @Nullable final Throwable attemptFailedResult) { - if (attemptFailedResult != null) { + MutableValue> asyncFunctionSuccessfulResult = new MutableValue<>(); + beginAsync().thenRunWhileLoop(() -> asyncFunctionSuccessfulResult.getNullable() == null, iterationCallback -> { + beginAsync().thenSupply(asyncFunctionCallback -> { + asyncFunction.get(asyncFunctionCallback); + }).thenConsume((attemptSuccessfulResult, onAttemptSuccessCallback) -> { + // `attemptSuccessfulResult` may be `null`, so we have to wrap it in `MutableValue` for the while check to notice it + asyncFunctionSuccessfulResult.set(new MutableValue<>(attemptSuccessfulResult)); + onAttemptSuccessCallback.complete(onAttemptSuccessCallback); + }).onErrorIf(e -> true, (attemptFailedResult, onAttemptFailureCallback) -> { if (attemptFailedResult instanceof Error) { - wrapped.onResult(null, attemptFailedResult); - return; - } - try { - assertNotNull(control.advanceOrThrow(attemptFailedResult)); - } catch (Throwable retryingSupplierFailedResult) { - wrapped.onResult(null, retryingSupplierFailedResult); - return; + onAttemptFailureCallback.completeExceptionally(attemptFailedResult); + } else { + RetryAttemptInfo retryAttemptInfo = control.advanceOrThrow(attemptFailedResult); + clientExecutor.sleepAsync(retryAttemptInfo.getBackoff(), onAttemptFailureCallback); } - asyncFunction.get(this); - } else { - wrapped.onResult(attemptSuccessfulResult, null); - } - } + }).finish(iterationCallback); + }).thenSupply(c -> { + c.complete(asyncFunctionSuccessfulResult.get().getNullable()); + }).finish(callback); } } diff --git a/driver-core/src/main/com/mongodb/internal/async/function/RetryingSyncSupplier.java b/driver-core/src/main/com/mongodb/internal/async/function/RetryingSyncSupplier.java index 3af14b46786..124e6aa013e 100644 --- a/driver-core/src/main/com/mongodb/internal/async/function/RetryingSyncSupplier.java +++ b/driver-core/src/main/com/mongodb/internal/async/function/RetryingSyncSupplier.java @@ -16,10 +16,16 @@ package com.mongodb.internal.async.function; import com.mongodb.annotations.NotThreadSafe; +import com.mongodb.internal.async.MutableValue; +import com.mongodb.internal.async.function.RetryPolicy.Decision.RetryAttemptInfo; +import com.mongodb.internal.thread.AsyncClientExecutor; +import com.mongodb.lang.Nullable; +import java.time.Duration; import java.util.function.Supplier; -import static com.mongodb.assertions.Assertions.assertNotNull; +import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException; +import static java.util.concurrent.TimeUnit.NANOSECONDS; /** * A decorator that implements automatic retrying of failed executions of a {@link Supplier}. @@ -37,7 +43,7 @@ public final class RetryingSyncSupplier implements Supplier { private final Supplier syncFunction; /** - * See {@link RetryingAsyncCallbackSupplier#RetryingAsyncCallbackSupplier(RetryControl, AsyncCallbackSupplier)}. + * See {@link RetryingAsyncCallbackSupplier#RetryingAsyncCallbackSupplier(AsyncClientExecutor, RetryControl, AsyncCallbackSupplier)}. */ public RetryingSyncSupplier(final RetryControl control, final Supplier syncFunction) { this.control = control; @@ -45,15 +51,29 @@ public RetryingSyncSupplier(final RetryControl control, final Supplier syn } @Override + @Nullable public R get() { - while (true) { + MutableValue> asyncFunctionSuccessfulResult = new MutableValue<>(); + while (asyncFunctionSuccessfulResult.getNullable() == null) { try { - return syncFunction.get(); + R attemptSuccessfulResult = syncFunction.get(); + // `attemptSuccessfulResult` may be `null`, so we have to wrap it in `MutableValue` for the while check to notice it + asyncFunctionSuccessfulResult.set(new MutableValue<>(attemptSuccessfulResult)); } catch (Error attemptFailedResult) { throw attemptFailedResult; } catch (Throwable attemptFailedResult) { - assertNotNull(control.advanceOrThrow(attemptFailedResult)); + RetryAttemptInfo retryAttemptInfo = control.advanceOrThrow(attemptFailedResult); + sleep(retryAttemptInfo.getBackoff()); } } + return asyncFunctionSuccessfulResult.get().getNullable(); + } + + private static void sleep(final Duration duration) { + try { + NANOSECONDS.sleep(duration.toNanos()); + } catch (InterruptedException e) { + throw interruptAndCreateMongoInterruptedException(null, e); + } } } diff --git a/driver-core/src/main/com/mongodb/internal/connection/AsynchronousSocketChannelStreamFactoryFactory.java b/driver-core/src/main/com/mongodb/internal/connection/AsynchronousSocketChannelStreamFactoryFactory.java index 8c5a8f654c5..a79572cd8dd 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/AsynchronousSocketChannelStreamFactoryFactory.java +++ b/driver-core/src/main/com/mongodb/internal/connection/AsynchronousSocketChannelStreamFactoryFactory.java @@ -16,12 +16,20 @@ package com.mongodb.internal.connection; +import com.mongodb.connection.AsyncTransportSettings; import com.mongodb.connection.SocketSettings; import com.mongodb.connection.SslSettings; +import com.mongodb.internal.thread.AsyncClientExecutor; +import com.mongodb.internal.thread.DaemonThreadFactory; +import com.mongodb.internal.thread.MongoThreadPoolExecutor; import com.mongodb.lang.Nullable; import com.mongodb.spi.dns.InetAddressResolver; import java.nio.channels.AsynchronousChannelGroup; +import java.time.Duration; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; /** * A {@code StreamFactoryFactory} implementation for AsynchronousSocketChannel-based streams. @@ -32,6 +40,8 @@ public final class AsynchronousSocketChannelStreamFactoryFactory implements Stre private final InetAddressResolver inetAddressResolver; @Nullable private final AsynchronousChannelGroup group; + private final MongoThreadPoolExecutor ownedExecutorBackingClientExecutor; + private final AsyncClientExecutor clientExecutor; public AsynchronousSocketChannelStreamFactoryFactory(final InetAddressResolver inetAddressResolver) { this(inetAddressResolver, null); @@ -42,6 +52,12 @@ public AsynchronousSocketChannelStreamFactoryFactory(final InetAddressResolver i @Nullable final AsynchronousChannelGroup group) { this.inetAddressResolver = inetAddressResolver; this.group = group; + int availableProcessors = Runtime.getRuntime().availableProcessors(); + ownedExecutorBackingClientExecutor = new MongoThreadPoolExecutor( + availableProcessors, availableProcessors, Duration.ofMinutes(5), + new LinkedBlockingQueue<>(), new DaemonThreadFactory("ClientExecutor")); + ownedExecutorBackingClientExecutor.allowCoreThreadTimeOut(true); + clientExecutor = AsyncClientExecutor.backedBy(ownedExecutorBackingClientExecutor); } @Override @@ -50,10 +66,32 @@ public StreamFactory create(final SocketSettings socketSettings, final SslSettin inetAddressResolver, socketSettings, sslSettings, group); } + /** + * VAKOTODO create ticket, leave a TODO + * To make things right, this should be + * the {@linkplain AsyncClientExecutor} {@linkplain AsyncClientExecutor#backedBy(Executor) backed by} same {@link ExecutorService} that + * the {@link AsynchronousChannelGroup} in {@link AsynchronousSocketChannelStreamFactory} uses + * (see {@link #create(SocketSettings, SslSettings)}). + * That {@link ExecutorService} may be provided by an application via {@link AsyncTransportSettings#getExecutorService()}. + * But currently it is a separate {@link MongoThreadPoolExecutor}. + */ + @Override + public AsyncClientExecutor getClientExecutor() { + return clientExecutor; + } + @Override public void close() { - if (group != null) { - group.shutdown(); + try { + clientExecutor.close(); + } finally { + try { + if (group != null) { + group.shutdown(); + } + } finally { + ownedExecutorBackingClientExecutor.shutdown(); + } } } } diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java index ac853cb002e..668af7119c3 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java +++ b/driver-core/src/main/com/mongodb/internal/connection/DefaultClusterFactory.java @@ -35,6 +35,7 @@ import com.mongodb.internal.VisibleForTesting; import com.mongodb.internal.diagnostics.logging.Logger; import com.mongodb.internal.diagnostics.logging.Loggers; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import com.mongodb.spi.dns.DnsClient; @@ -65,6 +66,7 @@ public Cluster createCluster(final ClusterSettings originalClusterSettings, fina final StreamFactory streamFactory, final TimeoutSettings heartbeatTimeoutSettings, final StreamFactory heartbeatStreamFactory, + final AsyncClientExecutor clientExecutor, @Nullable final MongoCredential credential, final LoggerSettings loggerSettings, @Nullable final CommandListener commandListener, @@ -103,9 +105,9 @@ public Cluster createCluster(final ClusterSettings originalClusterSettings, fina DnsSrvRecordMonitorFactory dnsSrvRecordMonitorFactory = new DefaultDnsSrvRecordMonitorFactory(clusterId, serverSettings, dnsClient); InternalOperationContextFactory clusterOperationContextFactory = - new InternalOperationContextFactory(clusterTimeoutSettings, serverApi); + new InternalOperationContextFactory(clusterTimeoutSettings, serverApi, clientExecutor); InternalOperationContextFactory heartBeatOperationContextFactory = - new InternalOperationContextFactory(heartbeatTimeoutSettings, serverApi); + new InternalOperationContextFactory(heartbeatTimeoutSettings, serverApi, clientExecutor); ClientMetadata clientMetadata = new ClientMetadata( applicationName, diff --git a/driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java b/driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java index 2339cf18b86..a0a8bbe6e2d 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java +++ b/driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java @@ -894,7 +894,7 @@ public boolean shouldPrune(final UsageTrackingInternalConnection usageTrackingCo /** * Package-access methods are thread-safe, - * and only they should be called outside of the {@link OpenConcurrencyLimiter}'s code. + * and only they should be called outside the {@link OpenConcurrencyLimiter}'s code. */ @ThreadSafe private final class OpenConcurrencyLimiter { @@ -1334,7 +1334,6 @@ private boolean initUnlessClosed() { * {@linkplain Task#failAsClosed() fail} asynchronously. */ @Override - @SuppressWarnings("try") public void close() { withLock(lock, () -> { if (state != State.CLOSED) { diff --git a/driver-core/src/main/com/mongodb/internal/connection/InternalOperationContextFactory.java b/driver-core/src/main/com/mongodb/internal/connection/InternalOperationContextFactory.java index 4653c90050b..851ca5a38ca 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/InternalOperationContextFactory.java +++ b/driver-core/src/main/com/mongodb/internal/connection/InternalOperationContextFactory.java @@ -18,6 +18,7 @@ import com.mongodb.ServerApi; import com.mongodb.internal.TimeoutContext; import com.mongodb.internal.TimeoutSettings; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import static com.mongodb.internal.connection.OperationContext.simpleOperationContext; @@ -27,17 +28,22 @@ public final class InternalOperationContextFactory { private final TimeoutSettings timeoutSettings; @Nullable private final ServerApi serverApi; + private final AsyncClientExecutor clientExecutor; - public InternalOperationContextFactory(final TimeoutSettings timeoutSettings, @Nullable final ServerApi serverApi) { + public InternalOperationContextFactory( + final TimeoutSettings timeoutSettings, + @Nullable final ServerApi serverApi, + final AsyncClientExecutor clientExecutor) { this.timeoutSettings = timeoutSettings; this.serverApi = serverApi; + this.clientExecutor = clientExecutor; } /** * @return a simple operation context without timeoutMS */ OperationContext create() { - return simpleOperationContext(timeoutSettings.connectionOnly(), serverApi); + return simpleOperationContext(timeoutSettings.connectionOnly(), serverApi, clientExecutor); } /** diff --git a/driver-core/src/main/com/mongodb/internal/connection/OperationContext.java b/driver-core/src/main/com/mongodb/internal/connection/OperationContext.java index 80cc143230f..cb8d0830dd1 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/OperationContext.java +++ b/driver-core/src/main/com/mongodb/internal/connection/OperationContext.java @@ -28,10 +28,12 @@ import com.mongodb.internal.IgnorableRequestContext; import com.mongodb.internal.TimeoutContext; import com.mongodb.internal.TimeoutSettings; +import com.mongodb.internal.VisibleForTesting; import com.mongodb.internal.observability.micrometer.Span; import com.mongodb.internal.observability.micrometer.TracingManager; import com.mongodb.internal.operation.OperationHelper; import com.mongodb.internal.session.SessionContext; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import com.mongodb.selector.ServerSelector; @@ -43,6 +45,7 @@ import static com.mongodb.MongoException.SYSTEM_OVERLOADED_ERROR_LABEL; import static com.mongodb.assertions.Assertions.assertFalse; +import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE; import static java.util.stream.Collectors.toList; /** @@ -60,19 +63,23 @@ public class OperationContext { private final ServerApi serverApi; @Nullable private final String operationName; + private final AsyncClientExecutor clientExecutor; @Nullable private Span tracingSpan; + @VisibleForTesting(otherwise = PRIVATE) public OperationContext(final RequestContext requestContext, final SessionContext sessionContext, final TimeoutContext timeoutContext, @Nullable final ServerApi serverApi) { - this(requestContext, sessionContext, timeoutContext, TracingManager.NO_OP, serverApi, null); + this(requestContext, sessionContext, timeoutContext, AsyncClientExecutor.NO_OP, TracingManager.NO_OP, serverApi, null); } public OperationContext(final RequestContext requestContext, final SessionContext sessionContext, final TimeoutContext timeoutContext, + final AsyncClientExecutor clientExecutor, final TracingManager tracingManager, @Nullable final ServerApi serverApi, @Nullable final String operationName) { this(NEXT_ID.incrementAndGet(), requestContext, sessionContext, timeoutContext, new ServerDeprioritization(), + clientExecutor, tracingManager, serverApi, operationName, @@ -80,52 +87,49 @@ public OperationContext(final RequestContext requestContext, final SessionContex } public OperationContext(final RequestContext requestContext, final SessionContext sessionContext, final TimeoutContext timeoutContext, + final AsyncClientExecutor clientExecutor, final TracingManager tracingManager, @Nullable final ServerApi serverApi, @Nullable final String operationName, final ServerDeprioritization serverDeprioritization) { this(NEXT_ID.incrementAndGet(), requestContext, sessionContext, timeoutContext, serverDeprioritization, + clientExecutor, tracingManager, serverApi, operationName, null); } - static OperationContext simpleOperationContext( - final TimeoutSettings timeoutSettings, @Nullable final ServerApi serverApi) { + public static OperationContext simpleOperationContext( + final TimeoutSettings timeoutSettings, @Nullable final ServerApi serverApi, final AsyncClientExecutor clientExecutor) { return new OperationContext( IgnorableRequestContext.INSTANCE, NoOpSessionContext.INSTANCE, new TimeoutContext(timeoutSettings), + clientExecutor, TracingManager.NO_OP, serverApi, - null - ); + null); } - public static OperationContext simpleOperationContext(final TimeoutContext timeoutContext) { - return new OperationContext( - IgnorableRequestContext.INSTANCE, - NoOpSessionContext.INSTANCE, - timeoutContext, - TracingManager.NO_OP, - null, - null); + @VisibleForTesting(otherwise = PRIVATE) + static OperationContext simpleOperationContext(final TimeoutSettings timeoutSettings) { + return simpleOperationContext(timeoutSettings, null, AsyncClientExecutor.NO_OP); } public OperationContext withSessionContext(final SessionContext sessionContext) { - return new OperationContext(id, requestContext, sessionContext, timeoutContext, serverDeprioritization, tracingManager, serverApi, - operationName, tracingSpan); + return new OperationContext(id, requestContext, sessionContext, timeoutContext, serverDeprioritization, clientExecutor, + tracingManager, serverApi, operationName, tracingSpan); } public OperationContext withTimeoutContext(final TimeoutContext timeoutContext) { - return new OperationContext(id, requestContext, sessionContext, timeoutContext, serverDeprioritization, tracingManager, serverApi, - operationName, tracingSpan); + return new OperationContext(id, requestContext, sessionContext, timeoutContext, serverDeprioritization, clientExecutor, + tracingManager, serverApi, operationName, tracingSpan); } public OperationContext withOperationName(final String operationName) { - return new OperationContext(id, requestContext, sessionContext, timeoutContext, serverDeprioritization, tracingManager, serverApi, - operationName, tracingSpan); + return new OperationContext(id, requestContext, sessionContext, timeoutContext, serverDeprioritization, clientExecutor, + tracingManager, serverApi, operationName, tracingSpan); } /** @@ -134,8 +138,8 @@ public OperationContext withOperationName(final String operationName) { */ public OperationContext withNewServerDeprioritization() { return new OperationContext(id, requestContext, sessionContext, timeoutContext, - new ServerDeprioritization(serverDeprioritization.enableOverloadRetargeting), tracingManager, serverApi, - operationName, tracingSpan); + new ServerDeprioritization(serverDeprioritization.enableOverloadRetargeting), clientExecutor, + tracingManager, serverApi, operationName, tracingSpan); } public long getId() { @@ -168,6 +172,10 @@ public String getOperationName() { return operationName; } + public AsyncClientExecutor getClientExecutor() { + return clientExecutor; + } + @Nullable public Span getTracingSpan() { return tracingSpan; @@ -182,6 +190,7 @@ private OperationContext(final long id, final SessionContext sessionContext, final TimeoutContext timeoutContext, final ServerDeprioritization serverDeprioritization, + final AsyncClientExecutor clientExecutor, final TracingManager tracingManager, @Nullable final ServerApi serverApi, @Nullable final String operationName, @@ -192,6 +201,7 @@ private OperationContext(final long id, this.requestContext = requestContext; this.sessionContext = sessionContext; this.timeoutContext = timeoutContext; + this.clientExecutor = clientExecutor; this.tracingManager = tracingManager; this.serverApi = serverApi; this.operationName = operationName; diff --git a/driver-core/src/main/com/mongodb/internal/connection/StreamFactoryFactory.java b/driver-core/src/main/com/mongodb/internal/connection/StreamFactoryFactory.java index 6cbe620fd43..8f3408d7bc5 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/StreamFactoryFactory.java +++ b/driver-core/src/main/com/mongodb/internal/connection/StreamFactoryFactory.java @@ -18,6 +18,7 @@ import com.mongodb.connection.SocketSettings; import com.mongodb.connection.SslSettings; +import com.mongodb.internal.thread.AsyncClientExecutor; /** * A factory of {@code StreamFactory} instances. @@ -29,10 +30,12 @@ public interface StreamFactoryFactory extends AutoCloseable { * * @param socketSettings the socket settings * @param sslSettings the SSL settings - * @return a stream factory that will apply the given settins + * @return a stream factory that will apply the given settings */ StreamFactory create(SocketSettings socketSettings, SslSettings sslSettings); + AsyncClientExecutor getClientExecutor(); + @Override void close(); } diff --git a/driver-core/src/main/com/mongodb/internal/connection/StreamFactoryHelper.java b/driver-core/src/main/com/mongodb/internal/connection/StreamFactoryHelper.java index 7aeb65720b0..b92c2570f01 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/StreamFactoryHelper.java +++ b/driver-core/src/main/com/mongodb/internal/connection/StreamFactoryHelper.java @@ -24,6 +24,7 @@ import com.mongodb.connection.SslSettings; import com.mongodb.connection.TransportSettings; import com.mongodb.internal.connection.netty.NettyStreamFactoryFactory; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import com.mongodb.spi.dns.InetAddressResolver; @@ -47,6 +48,11 @@ public StreamFactory create(final SocketSettings socketSettings, final SslSettin return new SocketStreamFactory(inetAddressResolver, socketSettings, sslSettings); } + @Override + public AsyncClientExecutor getClientExecutor() { + return AsyncClientExecutor.NO_OP; + } + @Override public void close() { //NOP diff --git a/driver-core/src/main/com/mongodb/internal/connection/TlsChannelStreamFactoryFactory.java b/driver-core/src/main/com/mongodb/internal/connection/TlsChannelStreamFactoryFactory.java index b0fae1d044d..b3d633221f3 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/TlsChannelStreamFactoryFactory.java +++ b/driver-core/src/main/com/mongodb/internal/connection/TlsChannelStreamFactoryFactory.java @@ -20,6 +20,7 @@ import com.mongodb.MongoSocketOpenException; import com.mongodb.ServerAddress; import com.mongodb.connection.AsyncCompletionHandler; +import com.mongodb.connection.AsyncTransportSettings; import com.mongodb.connection.SocketSettings; import com.mongodb.connection.SslSettings; import com.mongodb.internal.connection.tlschannel.BufferAllocator; @@ -29,6 +30,7 @@ import com.mongodb.internal.connection.tlschannel.async.AsynchronousTlsChannelGroup; import com.mongodb.internal.diagnostics.logging.Logger; import com.mongodb.internal.diagnostics.logging.Loggers; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import com.mongodb.spi.dns.InetAddressResolver; @@ -47,6 +49,7 @@ import java.security.NoSuchAlgorithmException; import java.util.Iterator; import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -68,6 +71,7 @@ public class TlsChannelStreamFactoryFactory implements StreamFactoryFactory { private final SelectorMonitor selectorMonitor; private final AsynchronousTlsChannelGroup group; + private final AsyncClientExecutor clientExecutor; private final PowerOfTwoBufferPool bufferPool = PowerOfTwoBufferPool.DEFAULT; private final InetAddressResolver inetAddressResolver; @@ -78,6 +82,7 @@ public class TlsChannelStreamFactoryFactory implements StreamFactoryFactory { @Nullable final ExecutorService executorService) { this.inetAddressResolver = inetAddressResolver; this.group = new AsynchronousTlsChannelGroup(executorService); + clientExecutor = AsyncClientExecutor.backedBy(group); selectorMonitor = new SelectorMonitor(); selectorMonitor.start(); } @@ -93,10 +98,27 @@ public StreamFactory create(final SocketSettings socketSettings, final SslSettin selectorMonitor); } + /** + * The {@linkplain AsyncClientExecutor} {@linkplain AsyncClientExecutor#backedBy(Executor) backed by} the same {@link ExecutorService} + * used by {@link #create(SocketSettings, SslSettings)} for a {@link StreamFactory}. + * That {@link ExecutorService} may be provided by an application via {@link AsyncTransportSettings#getExecutorService()}. + */ + @Override + public AsyncClientExecutor getClientExecutor() { + return clientExecutor; + } + @Override public void close() { - selectorMonitor.close(); - group.shutdown(); + try { + clientExecutor.close(); + } finally { + try { + selectorMonitor.close(); + } finally { + group.shutdown(); + } + } } /** diff --git a/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStreamFactoryFactory.java b/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStreamFactoryFactory.java index 7fe54defaa2..6a31ef3a9a0 100644 --- a/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStreamFactoryFactory.java +++ b/driver-core/src/main/com/mongodb/internal/connection/netty/NettyStreamFactoryFactory.java @@ -22,6 +22,7 @@ import com.mongodb.internal.VisibleForTesting; import com.mongodb.internal.connection.StreamFactory; import com.mongodb.internal.connection.StreamFactoryFactory; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import com.mongodb.spi.dns.InetAddressResolver; import io.netty.buffer.ByteBufAllocator; @@ -36,6 +37,7 @@ import java.security.Security; import java.util.Objects; +import java.util.concurrent.Executor; import static com.mongodb.assertions.Assertions.isTrueArgument; import static com.mongodb.assertions.Assertions.notNull; @@ -48,6 +50,7 @@ public final class NettyStreamFactoryFactory implements StreamFactoryFactory { private final EventLoopGroup eventLoopGroup; private final boolean ownsEventLoopGroup; + private final AsyncClientExecutor clientExecutor; private final Class socketChannelClass; private final ByteBufAllocator allocator; @Nullable @@ -151,7 +154,7 @@ public Builder eventLoopGroup(final EventLoopGroup eventLoopGroup) { /** * Sets a {@linkplain SslContextBuilder#forClient() client-side} {@link SslContext io.netty.handler.ssl.SslContext}, * which overrides the standard {@link SslSettings#getContext()}. - * By default it is {@code null} and {@link SslSettings#getContext()} is at play. + * By default, it is {@code null} and {@link SslSettings#getContext()} is at play. *

* This option may be used as a convenient way to utilize * OpenSSL as an alternative to the TLS/SSL protocol implementation in a JDK. @@ -203,13 +206,27 @@ public StreamFactory create(final SocketSettings socketSettings, final SslSettin sslContext); } + /** + * The {@linkplain AsyncClientExecutor} {@linkplain AsyncClientExecutor#backedBy(Executor) backed by} the same {@link EventLoopGroup} + * used by {@link #create(SocketSettings, SslSettings)} for a {@link StreamFactory}. + * That {@link EventLoopGroup} may be provided by an application via {@link NettyTransportSettings#getEventLoopGroup()}. + */ + @Override + public AsyncClientExecutor getClientExecutor() { + return clientExecutor; + } + @Override public void close() { - if (ownsEventLoopGroup) { - // ignore the returned Future. This is in line with MongoClient behavior to not block waiting for connections to be returned - // to the pool - eventLoopGroup.shutdownGracefully(); - } + try { + clientExecutor.close(); + } finally { + if (ownsEventLoopGroup) { + // ignore the returned Future. This is in line with MongoClient behavior to not block waiting for connections to be returned + // to the pool + eventLoopGroup.shutdownGracefully(); + } + } } @Override @@ -236,6 +253,7 @@ private NettyStreamFactoryFactory(final Builder builder) { socketChannelClass = builder.socketChannelClass == null ? NioSocketChannel.class : builder.socketChannelClass; eventLoopGroup = builder.eventLoopGroup == null ? new NioEventLoopGroup() : builder.eventLoopGroup; ownsEventLoopGroup = builder.eventLoopGroup == null; + clientExecutor = AsyncClientExecutor.backedBy(eventLoopGroup); sslContext = builder.sslContext; inetAddressResolver = builder.inetAddressResolver; } diff --git a/driver-core/src/main/com/mongodb/internal/diagnostics/logging/Logger.java b/driver-core/src/main/com/mongodb/internal/diagnostics/logging/Logger.java index e9907bb3953..4c1eda53e11 100644 --- a/driver-core/src/main/com/mongodb/internal/diagnostics/logging/Logger.java +++ b/driver-core/src/main/com/mongodb/internal/diagnostics/logging/Logger.java @@ -17,8 +17,7 @@ package com.mongodb.internal.diagnostics.logging; /** - * This class is not part of the public API. It may be removed or changed at any time. - * + * This class is not part of the public API and may be removed or changed at any time. */ public interface Logger { /** diff --git a/driver-core/src/main/com/mongodb/internal/operation/AsyncOperationHelper.java b/driver-core/src/main/com/mongodb/internal/operation/AsyncOperationHelper.java index 52211611215..d327ad663c9 100644 --- a/driver-core/src/main/com/mongodb/internal/operation/AsyncOperationHelper.java +++ b/driver-core/src/main/com/mongodb/internal/operation/AsyncOperationHelper.java @@ -327,7 +327,7 @@ static AsyncCallbackSupplier decorateWithRetriesAsync( final RetryControl retryControl, final OperationContext operationContext, final AsyncCallbackSupplier supplier) { - return new RetryingAsyncCallbackSupplier<>(retryControl, callback -> { + return new RetryingAsyncCallbackSupplier<>(operationContext.getClientExecutor(), retryControl, callback -> { retryControl.getPolicy().onAttemptStart(retryControl, operationContext); supplier.get(callback); }); diff --git a/driver-core/src/main/com/mongodb/internal/operation/SpecRetryPolicy.java b/driver-core/src/main/com/mongodb/internal/operation/SpecRetryPolicy.java index c60e2939de9..b339c3b39e7 100644 --- a/driver-core/src/main/com/mongodb/internal/operation/SpecRetryPolicy.java +++ b/driver-core/src/main/com/mongodb/internal/operation/SpecRetryPolicy.java @@ -31,6 +31,7 @@ import com.mongodb.internal.connection.OperationContext.ServerDeprioritization; import com.mongodb.lang.Nullable; +import java.time.Duration; import java.util.EnumMap; import java.util.EnumSet; import java.util.Optional; @@ -138,7 +139,7 @@ public Decision onAttemptFailure(final RetryContext retryContext, final Throwabl boolean retry = retryableError && !maxAttemptsReached; Decision decision = new Decision( decideProspectiveFailedResult(retryContext.getProspectiveFailedResult().orElse(null), maybeInternalAttemptFailedResult), - retry ? new RetryAttemptInfo() : null); + retry ? new RetryAttemptInfo(Duration.ZERO) : null); policies.write().ifPresent(IndividualPolicies.State.Write::resetRequirementsInfo); return decision; } diff --git a/driver-core/src/main/com/mongodb/internal/session/ServerSessionPool.java b/driver-core/src/main/com/mongodb/internal/session/ServerSessionPool.java index 9111eaed3a9..df301021308 100644 --- a/driver-core/src/main/com/mongodb/internal/session/ServerSessionPool.java +++ b/driver-core/src/main/com/mongodb/internal/session/ServerSessionPool.java @@ -24,11 +24,14 @@ import com.mongodb.internal.IgnorableRequestContext; import com.mongodb.internal.TimeoutContext; import com.mongodb.internal.TimeoutSettings; +import com.mongodb.internal.VisibleForTesting; import com.mongodb.internal.connection.Cluster; import com.mongodb.internal.connection.Connection; import com.mongodb.internal.connection.NoOpSessionContext; import com.mongodb.internal.connection.OperationContext; +import com.mongodb.internal.observability.micrometer.TracingManager; import com.mongodb.internal.selector.ReadPreferenceServerSelector; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.internal.validator.NoOpFieldNameValidator; import com.mongodb.lang.Nullable; import com.mongodb.selector.ServerSelector; @@ -50,6 +53,7 @@ import java.util.concurrent.atomic.LongAdder; import static com.mongodb.assertions.Assertions.isTrue; +import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE; import static java.util.concurrent.TimeUnit.MINUTES; /** @@ -67,17 +71,23 @@ interface Clock { long millis(); } - public ServerSessionPool(final Cluster cluster, final TimeoutSettings timeoutSettings, @Nullable final ServerApi serverApi) { + public ServerSessionPool( + final Cluster cluster, + final AsyncClientExecutor clientExecutor, + final TimeoutSettings timeoutSettings, + @Nullable final ServerApi serverApi) { this(cluster, new OperationContext(IgnorableRequestContext.INSTANCE, NoOpSessionContext.INSTANCE, - new TimeoutContext(timeoutSettings.connectionOnly()), serverApi)); + new TimeoutContext(timeoutSettings.connectionOnly()), clientExecutor, TracingManager.NO_OP, serverApi, null)); } - public ServerSessionPool(final Cluster cluster, final OperationContext operationContext) { + @VisibleForTesting(otherwise = PRIVATE) + ServerSessionPool(final Cluster cluster, final OperationContext operationContext) { this(cluster, operationContext, System::currentTimeMillis); } - public ServerSessionPool(final Cluster cluster, final OperationContext operationContext, final Clock clock) { + @VisibleForTesting(otherwise = PRIVATE) + ServerSessionPool(final Cluster cluster, final OperationContext operationContext, final Clock clock) { this.cluster = cluster; this.operationContext = operationContext; this.clock = clock; diff --git a/driver-core/src/main/com/mongodb/internal/thread/AsyncClientExecutor.java b/driver-core/src/main/com/mongodb/internal/thread/AsyncClientExecutor.java new file mode 100644 index 00000000000..442b9af08ac --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/thread/AsyncClientExecutor.java @@ -0,0 +1,79 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import com.mongodb.annotations.ThreadSafe; +import com.mongodb.internal.async.SingleResultCallback; +import com.mongodb.internal.connection.StreamFactoryFactory; + +import java.time.Duration; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; + +/** + * An executor for a {@code MongoClient} (currently, only for an asynchronous one). + * An implementation may use resources shared by multiple clients, if appropriate. + *

+ * May be used to execute internal code that is not blocking, or application code. + * When an asynchronous client is used, the application code we may execute is supposed to not be blocking, but we cannot enforce that. + * If an application violates the contract, it bears the responsibility. + *

+ * Purposefully not {@link ExecutorService}, because it does not manage the underlying resources, if any. + * They must be managed externally to {@link AsyncClientExecutor}. + * Nonetheless, it is still {@link AutoCloseable}. See {@link #close()} for the details. + *

+ * This class is not part of the public API and may be removed or changed at any time. + * + * @see StreamFactoryFactory#getClientExecutor() + * @see CommonExecutor + */ +@ThreadSafe +public interface AsyncClientExecutor extends AutoCloseable { + AsyncClientExecutor NO_OP = UnimplementedAsyncClientExecutor.instance(); + + /** + * @param executor The executor to use for executing tasks. + * If it is a {@link ScheduledExecutorService}, then it is also used for scheduling, + * otherwise {@link CommonExecutor} is used for scheduling. + */ + static AsyncClientExecutor backedBy(final Executor executor) { + return new DefaultAsyncClientExecutor(executor); + } + + /** + * The callback-based counterpart to {@link Thread#sleep(long, int)}. + * + * @param duration A non-{@linkplain Duration#isNegative() negative} duration. + * If {@code duration} is {@linkplain Duration#isZero() zero}, + * the {@code callback} is {@linkplain SingleResultCallback#complete(SingleResultCallback) completed} + * by the {@link Thread} that invoked the method. + * Regardless of the {@link Duration}, {@link #close()} causes the {@code callback} to be completed with + * {@link RejectedExecutionException}. + */ + void sleepAsync(Duration duration, SingleResultCallback callback); + + /** + * Must be called before shutting down the {@linkplain #backedBy(Executor) backing executor}, + * to notify this {@link AsyncClientExecutor} that the backing executor may be about to shut down. + * This method guarantees exactly-once {@linkplain SingleResultCallback#onResult(Object, Throwable) completion} + * of all callbacks that may have not been completed otherwise if the backing executor shuts down. + * An example of such a callback is one scheduled via {@link #sleepAsync(Duration, SingleResultCallback)}. + */ + @Override + void close(); +} diff --git a/driver-core/src/main/com/mongodb/internal/thread/CommonExecutor.java b/driver-core/src/main/com/mongodb/internal/thread/CommonExecutor.java new file mode 100644 index 00000000000..3fb74e49cb2 --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/thread/CommonExecutor.java @@ -0,0 +1,100 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import com.mongodb.connection.AsyncTransportSettings; +import com.mongodb.internal.diagnostics.logging.Logger; +import com.mongodb.internal.diagnostics.logging.Loggers; + +import java.time.Duration; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import static com.mongodb.assertions.Assertions.fail; +import static java.lang.String.format; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +/** + * A per-{@link ClassLoader} executor. + *

+ * We do not always have access to {@link ScheduledExecutorService} in a {@code MongoClient}. + * For example, even if {@link AsyncTransportSettings#getExecutorService()} is present, it is merely an {@link ExecutorService}, + * and does not have to be a {@link ScheduledExecutorService}. + * {@link CommonExecutor} is always accessible and may be used to schedule tasks when a more suitable alternative does not exist, + * but must not be used to execute such scheduled tasks. + *

+ * This class is not part of the public API and may be removed or changed at any time. + */ +// VAKOTODO create a ticket and leave a TODO to use Cleaner when we are at Java SE 17 to shut down internal executors if the class is GCed. +public final class CommonExecutor { + private static final Logger LOGGER = Loggers.getLogger("client"); + private static final CommonExecutor INSTANCE = new CommonExecutor(); + + private final MongoScheduledThreadPoolExecutor singleThreadScheduler; + + public static CommonExecutor commonExecutor() { + return INSTANCE; + } + + private CommonExecutor() { + singleThreadScheduler = new MongoScheduledThreadPoolExecutor(1, new DaemonThreadFactory("CommonScheduler")); + } + + /** + * @param task The task to be scheduled. If it is {@link Executor#execute(Runnable) executed}, then the execution is guaranteed + * to be done via the {@code executor}. However, if the {@code executor} is shut down, then it does not execute the {@code task}, + * and there is nothing we can do about that. + * @param delay A non-{@linkplain Duration#isNegative() negative} delay. + * @param executor The {@link Executor} to use for {@code task} {@linkplain Runnable#run() execution}, + * so that the {@code task} is not executed by a thread managed by {@link CommonExecutor}. + * @return The {@link ScheduledFuture} representing only + * the {@linkplain ScheduledExecutorService#schedule(Runnable, long, TimeUnit) scheduling part}, + * and not the execution part done by the {@code executor}. + */ + ScheduledFuture schedule(final Runnable task, final Duration delay, final Executor executor) { + try { + return singleThreadScheduler.schedule( + () -> { + // Depending on the `executor` implementation, which may be provided by an application, + // invoking `executor.execute` may result in executing arbitrary code, including `task`, in the single thread + // managed by `singleThreadScheduler`. This, in turn, may affect the behavior of other `MongoClient` instances + // that use `CommonExecutor`. More specifically, this may happen if: + // - `executor.execute` executes arbitrary code in the thread invoking the method; + // - `executor.execute` executes `task` in the thread invoking the `execute` method; + // - `executor` is a `ThreadPoolExecutor` with a bounded work queue that is full, + // and with `ThreadPoolExecutor.CallerRunsPolicy`; + // - `executor` is a `ThreadPoolExecutor` with a custom `RejectedExecutionHandler` that may result in + // executing `task` in the thread invoking the `execute` method. + // + // We consider the risk of the above small enough to make the current approach favourable to the alternative + // of having to run one more thread per `MongoClient`. + try { + executor.execute(task); + } catch (Exception e) { + LOGGER.error(format("The executor %s, which was likely provided by the application, either failed to execute" + + " the scheduled task %s, or executed it in the same thread that invoked `execute`", executor, task), e); + } + }, + delay.toNanos(), NANOSECONDS); + } catch (RejectedExecutionException e) { + throw fail(e.toString()); + } + } +} diff --git a/driver-core/src/main/com/mongodb/internal/thread/DaemonThreadFactory.java b/driver-core/src/main/com/mongodb/internal/thread/DaemonThreadFactory.java index e44fd0661c4..e64890148af 100644 --- a/driver-core/src/main/com/mongodb/internal/thread/DaemonThreadFactory.java +++ b/driver-core/src/main/com/mongodb/internal/thread/DaemonThreadFactory.java @@ -22,10 +22,10 @@ /** * Custom thread factory for scheduled executor service that creates daemon threads. Otherwise, * applications that neglect to close the client will not exit. - * - *

This class is not part of the public API and may be removed or changed at any time

+ *

+ * This class is not part of the public API and may be removed or changed at any time. */ -public class DaemonThreadFactory implements ThreadFactory { +public final class DaemonThreadFactory implements ThreadFactory { private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1); private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix; diff --git a/driver-core/src/main/com/mongodb/internal/thread/DefaultAsyncClientExecutor.java b/driver-core/src/main/com/mongodb/internal/thread/DefaultAsyncClientExecutor.java new file mode 100644 index 00000000000..bf70de0bed0 --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/thread/DefaultAsyncClientExecutor.java @@ -0,0 +1,190 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import com.mongodb.annotations.NotThreadSafe; +import com.mongodb.annotations.ThreadSafe; +import com.mongodb.internal.async.SingleResultCallback; +import com.mongodb.lang.Nullable; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.locks.ReentrantLock; + +import static com.mongodb.assertions.Assertions.assertFalse; +import static com.mongodb.assertions.Assertions.assertNull; +import static com.mongodb.internal.Locks.withLock; +import static com.mongodb.internal.async.AsyncRunnable.beginAsync; +import static com.mongodb.internal.thread.CommonExecutor.commonExecutor; +import static java.lang.String.format; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +@ThreadSafe +final class DefaultAsyncClientExecutor implements AsyncClientExecutor { + private final Executor backingExecutor; + private final Set scheduledCallbackCompletions; + /** + * While holding this lock, no application code may be executed, and driver code external to {@link DefaultAsyncClientExecutor} + * should be either avoided or carefully vetted. This is to avoid unexpected delays and deadlocks. + * For example, {@link ScheduledCallbackCompletion#reject(RejectedExecutionException)}, {@link ScheduledCallbackCompletion#run()} + * must not be executed while holding the lock. + */ + private final ReentrantLock closeLock; + private volatile boolean closed; + + DefaultAsyncClientExecutor(final Executor backingExecutor) { + this.backingExecutor = backingExecutor; + scheduledCallbackCompletions = ConcurrentHashMap.newKeySet(); + closeLock = new ReentrantLock(); + closed = false; + } + + @Override + public void sleepAsync(final Duration duration, final SingleResultCallback callback) { + sleepAsync(closed, duration, callback, () -> scheduleCompletion(callback, duration)); + } + + /** + * @param closed See {@link #closed}. + * @param onPositiveDurationDelayAndCompleteCallback Runs only if + * {@code duration} is {@linkplain Duration#isNegative() positive}. + * The {@link Runnable#run()} is allowed to complete abruptly and is guaranteed to run in the same thread that invoked this method. + */ + static void sleepAsync( + final boolean closed, + final Duration duration, + final SingleResultCallback callback, + final Runnable onPositiveDurationDelayAndCompleteCallback) { + beginAsync().thenRun(c -> { + assertFalse(duration.isNegative()); + if (closed) { + throw createClosedException(); + } + if (duration.isZero()) { + c.complete(c); + } else { + onPositiveDurationDelayAndCompleteCallback.run(); + } + }).finish(callback); + } + + private void scheduleCompletion(final SingleResultCallback callback, final Duration delay) { + ScheduledCallbackCompletion scheduledCallbackCompletion = new ScheduledCallbackCompletion(callback); + RejectedExecutionException rejectionCause = withLock(closeLock, () -> { + scheduledCallbackCompletions.add(scheduledCallbackCompletion); + if (closed) { + return createClosedException(); + } + ScheduledFuture scheduledFuture; + // We handle `isShutdown`, `RejectedExecutionException` below merely + // as the best effort to improve the application experience. + // Either situation violates the contract of the `close` method. + if (backingExecutor instanceof ExecutorService && ((ExecutorService) backingExecutor).isShutdown()) { + return createBackingExecutorShutdownException(); + } + if (backingExecutor instanceof ScheduledExecutorService) { + try { + scheduledFuture = ((ScheduledExecutorService) backingExecutor).schedule(scheduledCallbackCompletion, delay.toNanos(), NANOSECONDS); + } catch (RejectedExecutionException rejected) { + return rejected; + } + } else { + scheduledFuture = commonExecutor().schedule(scheduledCallbackCompletion, delay, backingExecutor); + } + scheduledCallbackCompletion.onScheduled(scheduledFuture); + return null; + }); + if (rejectionCause != null) { + scheduledCallbackCompletion.reject(rejectionCause); + } + } + + @Override + public void close() { + Collection localScheduledCallbackCompletions = new ArrayList<>(); + withLock(closeLock, () -> { + if (closed) { + return; + } + closed = true; + // Here we do not care about any `ScheduledCallbackCompletion` added after the current critical section, + // because its `reject` is called by the method that added it. + localScheduledCallbackCompletions.addAll(scheduledCallbackCompletions); + }); + for (ScheduledCallbackCompletion scheduledCallbackCompletion : localScheduledCallbackCompletions) { + scheduledCallbackCompletion.reject(createClosedException()); + } + } + + private static RejectedExecutionException createClosedException() { + return new RejectedExecutionException("Closed"); + } + + private RejectedExecutionException createBackingExecutorShutdownException() { + return new RejectedExecutionException(format("The backing executor %s is shut down", backingExecutor)); + } + + @Override + public String toString() { + return "DefaultAsyncClientExecutor{" + + "backingExecutor=" + backingExecutor + + ", closed=" + closed + + '}'; + } + + @NotThreadSafe + private class ScheduledCallbackCompletion implements Runnable { + private final SingleResultCallback callback; + @Nullable + private ScheduledFuture scheduledFuture; + + ScheduledCallbackCompletion(final SingleResultCallback callback) { + this.callback = callback; + } + + void onScheduled(final ScheduledFuture scheduledFuture) { + assertNull(this.scheduledFuture); + this.scheduledFuture = scheduledFuture; + } + + void reject(final RejectedExecutionException cause) { + if (scheduledCallbackCompletions.remove(this)) { + try { + if (scheduledFuture != null) { + scheduledFuture.cancel(false); + } + } finally { + callback.completeExceptionally(cause); + } + } + } + + @Override + public void run() { + if (scheduledCallbackCompletions.remove(this)) { + callback.complete(callback); + } + } + } +} diff --git a/driver-core/src/main/com/mongodb/internal/thread/InterruptionUtil.java b/driver-core/src/main/com/mongodb/internal/thread/InterruptionUtil.java index 54a3ba31f24..c933e951dee 100644 --- a/driver-core/src/main/com/mongodb/internal/thread/InterruptionUtil.java +++ b/driver-core/src/main/com/mongodb/internal/thread/InterruptionUtil.java @@ -25,14 +25,14 @@ import java.util.Optional; /** - *

This class is not part of the public API and may be removed or changed at any time

+ * This class is not part of the public API and may be removed or changed at any time. */ public final class InterruptionUtil { /** * {@linkplain Thread#interrupt() Interrupts} the {@linkplain Thread#currentThread() current thread} * before creating {@linkplain MongoInterruptedException}. - * We do this because the interrupt status is cleared before throwing {@link InterruptedException}, - * we are not propagating {@link InterruptedException}, which means we must reinstate the interrupt status. + * We do this because the interrupted status is cleared before throwing {@link InterruptedException}, + * we are not propagating {@link InterruptedException}, which means we must reinstate the interrupted status. * This matches the behavior documented by {@link MongoInterruptedException}. */ public static MongoInterruptedException interruptAndCreateMongoInterruptedException( diff --git a/driver-core/src/main/com/mongodb/internal/thread/MongoScheduledThreadPoolExecutor.java b/driver-core/src/main/com/mongodb/internal/thread/MongoScheduledThreadPoolExecutor.java new file mode 100644 index 00000000000..7d8e11b6714 --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/thread/MongoScheduledThreadPoolExecutor.java @@ -0,0 +1,41 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import com.mongodb.lang.Nullable; + +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; + +import static com.mongodb.internal.thread.MongoThreadPoolExecutor.propagateTaskFailureToUncaughtExceptionHandler; + +/** + * See {@link MongoThreadPoolExecutor}. + */ +final class MongoScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor { + MongoScheduledThreadPoolExecutor( + final int corePoolSize, + final ThreadFactory threadFactory) { + super(corePoolSize, threadFactory); + setRemoveOnCancelPolicy(true); + } + + @Override + protected void afterExecute(final Runnable r, @Nullable final Throwable t) { + super.afterExecute(r, t); + propagateTaskFailureToUncaughtExceptionHandler(r, t); + } +} diff --git a/driver-core/src/main/com/mongodb/internal/thread/MongoThreadPoolExecutor.java b/driver-core/src/main/com/mongodb/internal/thread/MongoThreadPoolExecutor.java new file mode 100644 index 00000000000..0fc245a894c --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/thread/MongoThreadPoolExecutor.java @@ -0,0 +1,107 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import com.mongodb.lang.Nullable; + +import java.io.PrintStream; +import java.lang.Thread.UncaughtExceptionHandler; +import java.time.Duration; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; + +import static com.mongodb.assertions.Assertions.assertNotNull; +import static com.mongodb.assertions.Assertions.fail; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +/** + * A {@link ThreadPoolExecutor} that ensures the task failure is propagated to the {@link UncaughtExceptionHandler}, if there is any, + * even if it is wrapped into a {@link Future} that represents completion of the task. This terminates the worker thread. + * Any non-{@link Error} gets turned into {@link AssertionError}, because all {@link Exception}s must be caught and handled by tasks. + *

+ * The driver code never observes task failures through {@link Future}s that represent completion of tasks, + * which is why this class is useful. + *

+ * Handling a task failure when it is an {@link Error} this way enables applications to decide how to deal with it + * via {@link UncaughtExceptionHandler}. An {@link Error} is more likely to cause an invariant violation than an {@link Exception}, + * because it is less likely to be taken into account in code. An {@link AssertionError} outright informs about an invariant violation. + * Furthermore, a {@link VirtualMachineError} not only may happen in a peculiar situation, + * but also may be asynchronous. + * That is why it may be a good idea for an application to terminate on {@link Error}. + * We cannot make such a decision for an application, but we must do our best to give it an opportunity to react to an {@link Error}. + *

+ * If there is no {@link UncaughtExceptionHandler}, then the failure is {@linkplain Throwable#printStackTrace(PrintStream) printed} + * to {@link System#err}, see {@link ThreadGroup#uncaughtException(Thread, Throwable)}. + */ +public final class MongoThreadPoolExecutor extends ThreadPoolExecutor { + public MongoThreadPoolExecutor( + final int corePoolSize, + final int maximumPoolSize, + final Duration keepAliveTime, + final BlockingQueue workQueue, + final ThreadFactory threadFactory) { + super(corePoolSize, maximumPoolSize, keepAliveTime.toNanos(), NANOSECONDS, workQueue, threadFactory); + } + + @Override + protected void afterExecute(final Runnable r, @Nullable final Throwable t) { + super.afterExecute(r, t); + propagateTaskFailureToUncaughtExceptionHandler(r, t); + } + + static void propagateTaskFailureToUncaughtExceptionHandler(final Runnable maybeFuture, @Nullable final Throwable t) { + if (t != null) { + if (t instanceof Error) { + // nothing to do, as we know `t` is going to be thrown and caught by `UncaughtExceptionHandler`, if there is any + return; + } else { + throw fail(t); + } + } + Throwable tWrapped = unwrapThrowable(maybeFuture); + if (tWrapped != null) { + if (tWrapped instanceof Error) { + // we must throw `tWrapped` for it to be caught by `UncaughtExceptionHandler`, if there is any + throw (Error) tWrapped; + } else { + throw fail(tWrapped); + } + } + } + + @Nullable + private static Throwable unwrapThrowable(final Runnable maybeFuture) { + if (maybeFuture instanceof Future && ((Future) maybeFuture).isDone()) { + Future runnableFuture = (Future) maybeFuture; + try { + runnableFuture.get(); + } catch (CancellationException e) { + // not a task failure + return null; + } catch (ExecutionException e) { + return assertNotNull(e.getCause()); + } catch (InterruptedException e) { + // not a task failure + Thread.currentThread().interrupt(); + } + } + return null; + } +} diff --git a/driver-core/src/main/com/mongodb/internal/thread/UnimplementedAsyncClientExecutor.java b/driver-core/src/main/com/mongodb/internal/thread/UnimplementedAsyncClientExecutor.java new file mode 100644 index 00000000000..8213fe47d73 --- /dev/null +++ b/driver-core/src/main/com/mongodb/internal/thread/UnimplementedAsyncClientExecutor.java @@ -0,0 +1,52 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import com.mongodb.annotations.ThreadSafe; +import com.mongodb.internal.async.SingleResultCallback; + +import java.time.Duration; + +import static com.mongodb.assertions.Assertions.fail; + +@ThreadSafe +final class UnimplementedAsyncClientExecutor implements AsyncClientExecutor { + private static final UnimplementedAsyncClientExecutor INSTANCE = new UnimplementedAsyncClientExecutor(); + + static UnimplementedAsyncClientExecutor instance() { + return INSTANCE; + } + + private UnimplementedAsyncClientExecutor() { + } + + /** + * Must not be called with any {@link Duration} except {@linkplain Duration#isZero() zero}. + */ + @Override + public void sleepAsync(final Duration duration, final SingleResultCallback callback) { + DefaultAsyncClientExecutor.sleepAsync(false, duration, callback, () -> { + throw fail(); + }); + } + + /** + * Does nothing. + */ + @Override + public void close() { + } +} diff --git a/driver-core/src/test/functional/com/mongodb/ClusterFixture.java b/driver-core/src/test/functional/com/mongodb/ClusterFixture.java index bfe41263b3c..9dd1378d72b 100644 --- a/driver-core/src/test/functional/com/mongodb/ClusterFixture.java +++ b/driver-core/src/test/functional/com/mongodb/ClusterFixture.java @@ -66,6 +66,7 @@ import com.mongodb.internal.operation.DropDatabaseOperation; import com.mongodb.internal.operation.ReadOperation; import com.mongodb.internal.operation.WriteOperation; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; @@ -74,9 +75,7 @@ import org.bson.BsonInt32; import org.bson.BsonString; import org.bson.BsonValue; -import org.bson.Document; import org.bson.codecs.BsonDocumentCodec; -import org.bson.codecs.DocumentCodec; import javax.net.ssl.SSLException; import java.time.Duration; @@ -143,7 +142,6 @@ public final class ClusterFixture { private static Cluster asyncCluster; private static final Map BINDING_MAP = new HashMap<>(); private static final Map SESSION_CONTEXT_MAP = new HashMap<>(); - private static final Map ASYNC_SESSION_CONTEXT_MAP = new HashMap<>(); private static final Map ASYNC_BINDING_MAP = new HashMap<>(); private static ServerVersion mongoCryptVersion; @@ -205,7 +203,7 @@ public static OperationContext createOperationContext() { } public static final InternalOperationContextFactory OPERATION_CONTEXT_FACTORY = - new InternalOperationContextFactory(TIMEOUT_SETTINGS, getServerApi()); + new InternalOperationContextFactory(TIMEOUT_SETTINGS, getServerApi(), AsyncClientExecutor.NO_OP); public static OperationContext createOperationContext(final TimeoutSettings timeoutSettings) { return new OperationContext( @@ -252,19 +250,6 @@ public static boolean hasEncryptionTestsEnabled() { .count() == requiredSystemProperties.size(); } - public static Document getServerStatus() { - return new CommandReadOperation<>("admin", new BsonDocument("serverStatus", new BsonInt32(1)), - new DocumentCodec()) - .execute(getBinding(), createOperationContext()); - } - - public static boolean supportsFsync() { - Document serverStatus = getServerStatus(); - Document storageEngine = (Document) serverStatus.get("storageEngine"); - - return storageEngine != null && !storageEngine.get("name").equals("inMemory"); - } - static class ShutdownHook extends Thread { @Override public void run() { @@ -413,7 +398,6 @@ public static AsyncReadWriteBinding getAsyncBinding( if (!ASYNC_BINDING_MAP.containsKey(readPreference)) { AsyncReadWriteBinding binding = new AsyncClusterBinding(cluster, readPreference); ASYNC_BINDING_MAP.put(readPreference, binding); - ASYNC_SESSION_CONTEXT_MAP.put(readPreference, new SimpleSessionContext()); } return ASYNC_BINDING_MAP.get(readPreference); } @@ -449,7 +433,7 @@ private static Cluster createCluster(final MongoCredential credential, final Str return new DefaultClusterFactory().createCluster(ClusterSettings.builder().hosts(asList(getPrimary())).build(), ServerSettings.builder().build(), ConnectionPoolSettings.builder().maxSize(1).build(), InternalConnectionPoolSettings.builder().build(), - TIMEOUT_SETTINGS.connectionOnly(), streamFactory, TIMEOUT_SETTINGS.connectionOnly(), streamFactory, credential, + TIMEOUT_SETTINGS.connectionOnly(), streamFactory, TIMEOUT_SETTINGS.connectionOnly(), streamFactory, AsyncClientExecutor.NO_OP, credential, LoggerSettings.builder().build(), null, null, null, Collections.emptyList(), getServerApi(), null); } @@ -461,7 +445,7 @@ private static Cluster createCluster(final ConnectionString connectionString, fi InternalConnectionPoolSettings.builder().build(), TimeoutSettings.create(mongoClientSettings).connectionOnly(), streamFactory, TimeoutSettings.createHeartbeatSettings(mongoClientSettings).connectionOnly(), new SocketStreamFactory(new DefaultInetAddressResolver(), SocketSettings.builder().readTimeout(5, SECONDS).build(), - getSslSettings(connectionString)), + getSslSettings(connectionString)), AsyncClientExecutor.NO_OP, connectionString.getCredential(), LoggerSettings.builder().build(), null, null, null, connectionString.getCompressorList(), getServerApi(), null); @@ -721,7 +705,7 @@ public static T executeAsync(final ReadOperation op, final AsyncReadBi return futureResultCallback.get(TIMEOUT, SECONDS); } - public static void loopCursor(final List> batchCursors, final Block block) throws Throwable { + public static void loopCursor(final List> batchCursors, final Block block) { List> futures = new ArrayList<>(); for (AsyncBatchCursor batchCursor : batchCursors) { FutureResultCallback futureResultCallback = new FutureResultCallback<>(); @@ -763,7 +747,7 @@ public static void loopCursor(final AsyncBatchCursor batchCursor, final B }); } - public static List collectCursorResults(final AsyncBatchCursor batchCursor) throws Throwable { + public static List collectCursorResults(final AsyncBatchCursor batchCursor) { List results = new ArrayList<>(); FutureResultCallback futureResultCallback = new FutureResultCallback<>(); loopCursor(batchCursor, t -> results.add(t), futureResultCallback); diff --git a/driver-core/src/test/functional/com/mongodb/internal/connection/TlsChannelStreamFunctionalTest.java b/driver-core/src/test/functional/com/mongodb/internal/connection/TlsChannelStreamFunctionalTest.java index 46e0cc6f964..e7ae01c0879 100644 --- a/driver-core/src/test/functional/com/mongodb/internal/connection/TlsChannelStreamFunctionalTest.java +++ b/driver-core/src/test/functional/com/mongodb/internal/connection/TlsChannelStreamFunctionalTest.java @@ -21,7 +21,6 @@ import com.mongodb.ServerAddress; import com.mongodb.connection.SocketSettings; import com.mongodb.connection.SslSettings; -import com.mongodb.internal.TimeoutContext; import com.mongodb.internal.TimeoutSettings; import org.bson.ByteBuf; import org.bson.ByteBufNIO; @@ -162,7 +161,7 @@ public T answer(final InvocationOnMock invocationOnMock) throws Throwable { } private static OperationContext createOperationContext(final int connectTimeoutMs) { - return simpleOperationContext(new TimeoutContext(TimeoutSettings.DEFAULT.withConnectTimeoutMS(connectTimeoutMs))); + return simpleOperationContext(TimeoutSettings.DEFAULT.withConnectTimeoutMS(connectTimeoutMs)); } @Test diff --git a/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryControlTest.java b/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryControlTest.java index edf0f900401..71bba600689 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryControlTest.java +++ b/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryControlTest.java @@ -22,6 +22,7 @@ import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import java.time.Duration; import java.util.Optional; import static org.junit.jupiter.api.Assertions.assertAll; @@ -40,7 +41,7 @@ final class RetryControlTest { @Test void isFirstAttempt() { RetryControl retryControl = new RetryControl<>((retryContext, attemptFailedResult) -> - new Decision(attemptFailedResult, new RetryAttemptInfo())); + new Decision(attemptFailedResult, createRetryAttemptInfo())); assertTrue(retryControl.isFirstAttempt()); retryControl.advanceOrThrow(new RuntimeException()); assertFalse(retryControl.isFirstAttempt()); @@ -49,7 +50,7 @@ void isFirstAttempt() { @Test void attempt() { RetryControl retryControl = new RetryControl<>((retryContext, attemptFailedResult) -> - new Decision(attemptFailedResult, new RetryAttemptInfo())); + new Decision(attemptFailedResult, createRetryAttemptInfo())); assertEquals(0, retryControl.attempt()); retryControl.advanceOrThrow(new RuntimeException()); assertEquals(1, retryControl.attempt()); @@ -59,7 +60,7 @@ void attempt() { @Test void getPolicy() { - RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(attemptFailedResult, new RetryAttemptInfo()); + RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(attemptFailedResult, createRetryAttemptInfo()); RetryControl retryControl = new RetryControl<>(retryPolicy); assertSame(retryPolicy, retryControl.getPolicy()); } @@ -67,7 +68,7 @@ void getPolicy() { @Test void advanceOrThrowPassesCorrectArgumentsToAttemptFailure() { RetryPolicy retryPolicy = MongoMockito.mock(RetryPolicy.class, retryPolicyMock -> { - when(retryPolicyMock.onAttemptFailure(any(), any())).thenReturn(new Decision(new RuntimeException(), new RetryAttemptInfo())); + when(retryPolicyMock.onAttemptFailure(any(), any())).thenReturn(new Decision(new RuntimeException(), createRetryAttemptInfo())); }); RetryControl retryControl = new RetryControl<>(retryPolicy); RuntimeException attemptFailedResult = new RuntimeException(); @@ -84,7 +85,7 @@ void advanceOrThrowPassesCorrectArgumentsToAttemptFailure() { @Test void advanceOrThrowReturnsIfAnotherAttempt() { - RetryAttemptInfo immediateNextAttemptInfo = new RetryAttemptInfo(); + RetryAttemptInfo immediateNextAttemptInfo = createRetryAttemptInfo(); RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(attemptFailedResult, immediateNextAttemptInfo); RetryControl retryControl = new RetryControl<>(retryPolicy); RetryAttemptInfo actualImmediateNextAttemptInfo = retryControl.advanceOrThrow(new RuntimeException()); @@ -103,7 +104,7 @@ void advanceOrThrowThrowsIfNoMoreAttempts() { @Test void advanceOrThrowThrowsIfLastAttempt() { RuntimeException prospectiveFailedResult = new RuntimeException(); - RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(prospectiveFailedResult, new RetryAttemptInfo()); + RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(prospectiveFailedResult, createRetryAttemptInfo()); RetryControl retryControl = new RetryControl<>(retryPolicy); retryControl.advanceOrThrow(new RuntimeException()); assertThrows(prospectiveFailedResult.getClass(), () -> retryControl.breakAndThrowIfRetryAnd(() -> true)); @@ -114,7 +115,7 @@ void advanceOrThrowThrowsIfLastAttempt() { @Test void advanceOrThrowStoresProspectiveFailedResult() { RuntimeException prospectiveFailedResult = new RuntimeException(); - RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(prospectiveFailedResult, new RetryAttemptInfo()); + RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(prospectiveFailedResult, createRetryAttemptInfo()); RetryControl retryControl = new RetryControl<>(retryPolicy); retryControl.advanceOrThrow(new RuntimeException()); Optional actualProspectiveFailedResult = retryControl.getProspectiveFailedResult(); @@ -127,7 +128,7 @@ void advanceOrThrowStoresProspectiveFailedResult() { @Test void advanceOrThrowOverwritesProspectiveFailedResult() { - RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(attemptFailedResult, new RetryAttemptInfo()); + RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(attemptFailedResult, createRetryAttemptInfo()); RetryControl retryControl = new RetryControl<>(retryPolicy); retryControl.advanceOrThrow(new RuntimeException()); RuntimeException prospectiveFailedResult = new RuntimeException(); @@ -144,7 +145,7 @@ void advanceOrThrowOverwritesProspectiveFailedResult() { @DisplayName("breakAndThrowIfRetryAnd does nothing if first attempt") void breakAndThrowIfRetryAndDoesNothingIfFirstAttempt() { RetryControl retryControl = new RetryControl<>((retryContext, attemptFailedResult) -> - new Decision(attemptFailedResult, new RetryAttemptInfo())); + new Decision(attemptFailedResult, createRetryAttemptInfo())); assertDoesNotThrow(() -> retryControl.breakAndThrowIfRetryAnd(() -> true)); } @@ -152,7 +153,7 @@ void breakAndThrowIfRetryAndDoesNothingIfFirstAttempt() { @DisplayName("breakAndThrowIfRetryAnd throws if not first attempt") void breakAndThrowIfRetryAndThrowsIfNotFirstAttempt() { RuntimeException prospectiveFailedResult = new RuntimeException(); - RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(prospectiveFailedResult, new RetryAttemptInfo()); + RetryPolicy retryPolicy = (retryContext, attemptFailedResult) -> new Decision(prospectiveFailedResult, createRetryAttemptInfo()); RetryControl retryControl = new RetryControl<>(retryPolicy); retryControl.advanceOrThrow(new RuntimeException()); assertSame(prospectiveFailedResult, @@ -163,7 +164,7 @@ void breakAndThrowIfRetryAndThrowsIfNotFirstAttempt() { @DisplayName("breakAndThrowIfRetryAnd propagates if predicate throws") void breakAndThrowIfRetryAndPropagatesIfPredicateThrows() { RetryControl retryControl = new RetryControl<>((retryContext, attemptFailedResult) -> - new Decision(attemptFailedResult, new RetryAttemptInfo())); + new Decision(attemptFailedResult, createRetryAttemptInfo())); retryControl.advanceOrThrow(new RuntimeException()); RuntimeException predicateException = new RuntimeException(); assertSame(predicateException, @@ -178,7 +179,7 @@ void breakAndThrowIfRetryAndPropagatesIfPredicateThrows() { void breakAndThrowIfRetryAndAddsSuppressedProspectiveFailedResultIfPredicateThrows() { RuntimeException prospectiveFailedResult = new RuntimeException(); RetryControl retryControl = new RetryControl<>((retryContext, attemptFailedResult) -> - new Decision(prospectiveFailedResult, new RetryAttemptInfo())); + new Decision(prospectiveFailedResult, createRetryAttemptInfo())); retryControl.advanceOrThrow(new RuntimeException()); RuntimeException predicateException = new RuntimeException(); Throwable[] suppressed = assertThrows(predicateException.getClass(), @@ -190,4 +191,8 @@ void breakAndThrowIfRetryAndAddsSuppressedProspectiveFailedResultIfPredicateThro () -> assertSame(suppressed[0], prospectiveFailedResult) ); } + + private static RetryAttemptInfo createRetryAttemptInfo() { + return new RetryAttemptInfo(Duration.ZERO); + } } diff --git a/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplierTest.java b/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplierTest.java index 9a9ecad1bcf..d21e0cf7a97 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplierTest.java +++ b/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryingAsyncCallbackSupplierTest.java @@ -16,19 +16,47 @@ package com.mongodb.internal.async.function; import com.mongodb.internal.async.function.RetryingSyncSupplierTest.AssertingUnusedRetryPolicy; +import com.mongodb.internal.thread.AsyncClientExecutor; +import com.mongodb.internal.time.StartTime; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + import static com.mongodb.internal.async.AsyncRunnable.beginAsync; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; final class RetryingAsyncCallbackSupplierTest { + private ExecutorService executorService; + private AsyncClientExecutor clientExecutor; + + @BeforeEach + void beforeEach() { + executorService = Executors.newSingleThreadScheduledExecutor(); + clientExecutor = AsyncClientExecutor.backedBy(executorService); + } + + @AfterEach + void afterEach() { + if (executorService != null) { + executorService.shutdownNow(); + } + } + @Test void doWhileDisabledThrowsAtFirstAttempt() { RetryControl retryControl = new RetryControl<>(new AssertingUnusedRetryPolicy(false)); RuntimeException exception = new RuntimeException(); RetryingAsyncCallbackSupplier retryingSupplier = new RetryingAsyncCallbackSupplier<>( + clientExecutor, retryControl, callback -> { retryControl.doWhileDisabledAsync(actionCallback -> { @@ -44,6 +72,7 @@ void doWhileDisabledThrowsAtSecondAttempt() { RetryControl retryControl = new RetryControl<>(new AssertingUnusedRetryPolicy(true)); RuntimeException exception = new RuntimeException(); RetryingAsyncCallbackSupplier retryingSupplier = new RetryingAsyncCallbackSupplier<>( + clientExecutor, retryControl, callback -> { if (retryControl.isFirstAttempt()) { @@ -63,6 +92,7 @@ void doWhileDisabledCompletesNormally() { RetryControl retryControl = new RetryControl<>(new AssertingUnusedRetryPolicy(true)); Object result = new Object(); RetryingAsyncCallbackSupplier retryingSupplier = new RetryingAsyncCallbackSupplier<>( + clientExecutor, retryControl, callback -> { beginAsync().thenSupply(c -> { @@ -84,6 +114,7 @@ void doWhileDisabledNestedThrowsAtFirstAttempt() { RetryControl retryControl = new RetryControl<>(new AssertingUnusedRetryPolicy(false)); RuntimeException exception = new RuntimeException(); RetryingAsyncCallbackSupplier retryingSupplier = new RetryingAsyncCallbackSupplier<>( + clientExecutor, retryControl, callback -> { retryControl.doWhileDisabledAsync(actionCallback -> { @@ -99,4 +130,34 @@ void doWhileDisabledNestedThrowsAtFirstAttempt() { retryingSupplier.get((r, t) -> assertSame(exception, t)); assertTrue(retryControl.isFirstAttempt()); } + + @Test + void backoff() throws Exception { + Duration backoff = Duration.ofMillis(400); + RetryControl retryControl = new RetryControl<>((retryContext, attemptFailedResult) -> + new RetryPolicy.Decision(attemptFailedResult, new RetryPolicy.Decision.RetryAttemptInfo(backoff))); + RetryingAsyncCallbackSupplier retryingSupplier = new RetryingAsyncCallbackSupplier<>( + AsyncClientExecutor.backedBy(executorService), + retryControl, + functionCallback -> { + beginAsync().thenRun(c -> { + if (retryControl.isFirstAttempt()) { + throw new RuntimeException(); + } + c.complete(c); + }).finish(functionCallback); + }); + StartTime startTime = StartTime.now(); + CompletableFuture durationFuture = new CompletableFuture<>(); + retryingSupplier.get((result, t) -> { + if (t != null) { + durationFuture.completeExceptionally(fail(t)); + } else { + durationFuture.complete(startTime.elapsed()); + } + }); + Duration duration = durationFuture.get(backoff.toMillis() * 2, MILLISECONDS); + assertTrue(duration.compareTo(backoff) >= 0); + assertTrue(duration.compareTo(backoff.multipliedBy(2)) < 0); + } } diff --git a/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryingSyncSupplierTest.java b/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryingSyncSupplierTest.java index f1ae3452171..090d8e10280 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryingSyncSupplierTest.java +++ b/driver-core/src/test/unit/com/mongodb/internal/async/function/RetryingSyncSupplierTest.java @@ -15,9 +15,13 @@ */ package com.mongodb.internal.async.function; +import com.mongodb.internal.async.function.RetryPolicy.Decision; import com.mongodb.internal.async.function.RetryPolicy.Decision.RetryAttemptInfo; +import com.mongodb.internal.time.StartTime; import org.junit.jupiter.api.Test; +import java.time.Duration; + import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -96,6 +100,26 @@ void doWhileDisabledNestedThrowsAtFirstAttempt() { assertTrue(retryControl.isFirstAttempt()); } + @Test + void backoff() { + Duration backoff = Duration.ofMillis(50); + RetryControl retryControl = new RetryControl<>((retryContext, attemptFailedResult) -> + new Decision(attemptFailedResult, new RetryAttemptInfo(backoff))); + RetryingSyncSupplier retryingSupplier = new RetryingSyncSupplier<>( + retryControl, + () -> { + if (retryControl.isFirstAttempt()) { + throw new RuntimeException(); + } + return null; + }); + StartTime startTime = StartTime.now(); + retryingSupplier.get(); + Duration duration = startTime.elapsed(); + assertTrue(duration.compareTo(backoff) >= 0); + assertTrue(duration.compareTo(backoff.multipliedBy(2)) < 0); + } + static final class AssertingUnusedRetryPolicy implements RetryPolicy { private final boolean skipFailingOnFirstAttempt; @@ -106,7 +130,7 @@ static final class AssertingUnusedRetryPolicy implements RetryPolicy { @Override public Decision onAttemptFailure(final RetryContext retryContext, final Throwable attemptFailedResult) { if (skipFailingOnFirstAttempt && retryContext.isFirstAttempt()) { - return new Decision(attemptFailedResult, new RetryAttemptInfo()); + return new Decision(attemptFailedResult, new RetryAttemptInfo(Duration.ZERO)); } return fail(); } diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionInitializerSpecification.groovy b/driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionInitializerSpecification.groovy index a7bfaa36cce..d3f05ea8b49 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionInitializerSpecification.groovy +++ b/driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionInitializerSpecification.groovy @@ -53,7 +53,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification { def serverId = new ServerId(new ClusterId(), new ServerAddress()) def internalConnection = new TestInternalConnection(serverId, ServerType.STANDALONE) - def operationContext = simpleOperationContext(TimeoutSettings.DEFAULT, null) + def operationContext = simpleOperationContext(TimeoutSettings.DEFAULT) def 'should create correct description'() { given: diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionSpecification.groovy b/driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionSpecification.groovy index 10074ba9d3c..cb2c1cf8ac4 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionSpecification.groovy +++ b/driver-core/src/test/unit/com/mongodb/internal/connection/InternalStreamConnectionSpecification.groovy @@ -314,7 +314,7 @@ class InternalStreamConnectionSpecification extends Specification { connection.isClosed() } - def 'should throw MongoInterruptedException and leave the interrupt status set when Stream.write throws InterruptedIOException'() { + def 'should throw MongoInterruptedException and leave interrupted status set when Stream.write throws InterruptedIOException'() { given: stream.write(_, _) >> { throw new InterruptedIOException() } def connection = getOpenedConnection() @@ -329,7 +329,7 @@ class InternalStreamConnectionSpecification extends Specification { connection.isClosed() } - def 'should throw MongoInterruptedException and leave the interrupt status unset when Stream.write throws InterruptedIOException'() { + def 'should throw MongoInterruptedException and leave interrupted status unset when Stream.write throws InterruptedIOException'() { given: stream.write(_, _) >> { throw new InterruptedIOException() } def connection = getOpenedConnection() @@ -343,7 +343,7 @@ class InternalStreamConnectionSpecification extends Specification { connection.isClosed() } - def 'should throw MongoInterruptedException and leave the interrupt status set when Stream.write throws ClosedByInterruptException'() { + def 'should throw MongoInterruptedException and leave interrupted status set when Stream.write throws ClosedByInterruptException'() { given: stream.write(_, _) >> { throw new ClosedByInterruptException() } def connection = getOpenedConnection() @@ -386,7 +386,7 @@ class InternalStreamConnectionSpecification extends Specification { connection.isClosed() } - def 'should throw MongoInterruptedException and leave the interrupt status set when Stream.read throws InterruptedIOException'() { + def 'should throw MongoInterruptedException and leave interrupted status set when Stream.read throws InterruptedIOException'() { given: stream.read(_, _) >> { throw new InterruptedIOException() } def connection = getOpenedConnection() @@ -401,7 +401,7 @@ class InternalStreamConnectionSpecification extends Specification { connection.isClosed() } - def 'should throw MongoInterruptedException and leave the interrupt status unset when Stream.read throws InterruptedIOException'() { + def 'should throw MongoInterruptedException and leave interrupted status unset when Stream.read throws InterruptedIOException'() { given: stream.read(_, _) >> { throw new InterruptedIOException() } def connection = getOpenedConnection() @@ -415,7 +415,7 @@ class InternalStreamConnectionSpecification extends Specification { connection.isClosed() } - def 'should throw MongoInterruptedException and leave the interrupt status set when Stream.read throws ClosedByInterruptException'() { + def 'should throw MongoInterruptedException and leave interrupted status set when Stream.read throws ClosedByInterruptException'() { given: stream.read(_, _) >> { throw new ClosedByInterruptException() } def connection = getOpenedConnection() diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/ScramShaAuthenticatorSpecification.groovy b/driver-core/src/test/unit/com/mongodb/internal/connection/ScramShaAuthenticatorSpecification.groovy index 21f9bc28161..ed7d758c21b 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/connection/ScramShaAuthenticatorSpecification.groovy +++ b/driver-core/src/test/unit/com/mongodb/internal/connection/ScramShaAuthenticatorSpecification.groovy @@ -41,7 +41,7 @@ import static org.junit.Assert.assertEquals class ScramShaAuthenticatorSpecification extends Specification { def serverId = new ServerId(new ClusterId(), new ServerAddress('localhost', 27017)) def connectionDescription = new ConnectionDescription(serverId) - def operationContext = simpleOperationContext(TimeoutSettings.DEFAULT, null) + def operationContext = simpleOperationContext(TimeoutSettings.DEFAULT) private final static MongoCredentialWithCache SHA1_CREDENTIAL = new MongoCredentialWithCache(createScramSha1Credential('user', 'database', 'pencil' as char[])) private final static MongoCredentialWithCache SHA256_CREDENTIAL = diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/ServerSelectionSelectionTest.java b/driver-core/src/test/unit/com/mongodb/internal/connection/ServerSelectionSelectionTest.java index 5d6b5e0e1e3..5abcfbc1bb7 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/connection/ServerSelectionSelectionTest.java +++ b/driver-core/src/test/unit/com/mongodb/internal/connection/ServerSelectionSelectionTest.java @@ -41,6 +41,7 @@ import com.mongodb.internal.observability.micrometer.TracingManager; import com.mongodb.internal.selector.ReadPreferenceServerSelector; import com.mongodb.internal.selector.WritableServerSelector; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.internal.time.Timeout; import com.mongodb.lang.NonNull; import com.mongodb.lang.Nullable; @@ -303,6 +304,7 @@ private OperationContext createOperationContext() { IgnorableRequestContext.INSTANCE, NoOpSessionContext.INSTANCE, new TimeoutContext(TIMEOUT_SETTINGS.withServerSelectionTimeoutMS(0)), + AsyncClientExecutor.NO_OP, TracingManager.NO_OP, null, null, diff --git a/driver-core/src/test/unit/com/mongodb/internal/connection/SocksSocketFunctionalTest.java b/driver-core/src/test/unit/com/mongodb/internal/connection/SocksSocketFunctionalTest.java index 7bfba300a61..6d34f771e2f 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/connection/SocksSocketFunctionalTest.java +++ b/driver-core/src/test/unit/com/mongodb/internal/connection/SocksSocketFunctionalTest.java @@ -87,7 +87,7 @@ private void connectWithMiniServer(final byte[] serverBytes, final boolean withC t.join(CONNECT_TIMEOUT_MS); } catch (InterruptedException ie) { // Don't mask the primary exception (if any) with the join interruption; - // just preserve the thread's interrupt status and continue. + // just preserve the thread's interrupted status and continue. Thread.currentThread().interrupt(); } } diff --git a/driver-core/src/test/unit/com/mongodb/internal/session/ServerSessionPoolSpecification.groovy b/driver-core/src/test/unit/com/mongodb/internal/session/ServerSessionPoolSpecification.groovy index 0fc4564d322..60ec62b2faf 100644 --- a/driver-core/src/test/unit/com/mongodb/internal/session/ServerSessionPoolSpecification.groovy +++ b/driver-core/src/test/unit/com/mongodb/internal/session/ServerSessionPoolSpecification.groovy @@ -25,6 +25,7 @@ import com.mongodb.internal.connection.Cluster import com.mongodb.internal.connection.Connection import com.mongodb.internal.connection.Server import com.mongodb.internal.connection.ServerTuple +import com.mongodb.internal.thread.AsyncClientExecutor import com.mongodb.internal.validator.NoOpFieldNameValidator import org.bson.BsonArray import org.bson.BsonBinarySubType @@ -71,7 +72,7 @@ class ServerSessionPoolSpecification extends Specification { def cluster = Stub(Cluster) { getCurrentDescription() >> connectedDescription } - def pool = new ServerSessionPool(cluster, TIMEOUT_SETTINGS, getServerApi()) + def pool = createServerSessionPool(cluster) when: def session = pool.get() @@ -85,7 +86,7 @@ class ServerSessionPoolSpecification extends Specification { def cluster = Stub(Cluster) { getCurrentDescription() >> connectedDescription } - def pool = new ServerSessionPool(cluster, TIMEOUT_SETTINGS, getServerApi()) + def pool = createServerSessionPool(cluster) pool.close() when: @@ -100,7 +101,7 @@ class ServerSessionPoolSpecification extends Specification { def cluster = Stub(Cluster) { getCurrentDescription() >> connectedDescription } - def pool = new ServerSessionPool(cluster, TIMEOUT_SETTINGS, getServerApi()) + def pool = createServerSessionPool(cluster) def session = pool.get() when: @@ -207,7 +208,7 @@ class ServerSessionPoolSpecification extends Specification { def cluster = Mock(Cluster) { getCurrentDescription() >> connectedDescription } - def pool = new ServerSessionPool(cluster, TIMEOUT_SETTINGS, getServerApi()) + def pool = createServerSessionPool(cluster) def sessions = [] 10.times { sessions.add(pool.get()) } @@ -226,4 +227,8 @@ class ServerSessionPoolSpecification extends Specification { { it instanceof BsonDocumentCodec }, _) >> new BsonDocument() 1 * connection.release() } + + static createServerSessionPool(Cluster cluster) { + new ServerSessionPool(cluster, AsyncClientExecutor.NO_OP, TIMEOUT_SETTINGS, getServerApi()) + } } diff --git a/driver-core/src/test/unit/com/mongodb/internal/thread/CommonExecutorTest.java b/driver-core/src/test/unit/com/mongodb/internal/thread/CommonExecutorTest.java new file mode 100644 index 00000000000..e3690ddc077 --- /dev/null +++ b/driver-core/src/test/unit/com/mongodb/internal/thread/CommonExecutorTest.java @@ -0,0 +1,52 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; + +import static com.mongodb.internal.thread.CommonExecutor.commonExecutor; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.junit.jupiter.api.Assertions.assertNotSame; + +class CommonExecutorTest { + private static final long TIMEOUT_MILLIS = 400; + /** + * This test verifies that even in the unlikely event that the single scheduling thread is terminated, + * it is replaced with another one to execute a previously scheduled task. + */ + @Test + void singleSchedulingThreadIsReplacedIfTerminated() throws Exception { + Executor sameThreadExecutor = Runnable::run; + CompletableFuture newSchedulingThread = new CompletableFuture<>(); + commonExecutor().schedule( + () -> newSchedulingThread.complete(Thread.currentThread()), + Duration.ofMillis(TIMEOUT_MILLIS / 2), + sameThreadExecutor); + CompletableFuture terminatedSchedulingThread = new CompletableFuture<>(); + commonExecutor().schedule( + () -> { + terminatedSchedulingThread.complete(Thread.currentThread()); + throw new Error("This error is thrown in the single scheduling thread, causing its termination"); + }, + Duration.ZERO, + sameThreadExecutor); + assertNotSame(terminatedSchedulingThread.get(TIMEOUT_MILLIS, MILLISECONDS), newSchedulingThread.get(TIMEOUT_MILLIS, MILLISECONDS)); + } +} diff --git a/driver-core/src/test/unit/com/mongodb/internal/thread/DefaultAsyncClientExecutorTest.java b/driver-core/src/test/unit/com/mongodb/internal/thread/DefaultAsyncClientExecutorTest.java new file mode 100644 index 00000000000..2771fbab2ad --- /dev/null +++ b/driver-core/src/test/unit/com/mongodb/internal/thread/DefaultAsyncClientExecutorTest.java @@ -0,0 +1,237 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import com.mongodb.internal.time.StartTime; +import io.netty.channel.EventLoopGroup; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import static com.mongodb.internal.thread.InterruptionUtil.interruptAndCreateMongoInterruptedException; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; +import static org.junit.jupiter.api.Assertions.assertAll; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class DefaultAsyncClientExecutorTest { + private static final long SLEEP_DURATION_MILLIS = 200; + + private ExecutorService executorService; + private ScheduledExecutorService scheduledExecutorService; + + @BeforeEach + void beforeEach() { + executorService = Executors.newSingleThreadExecutor(); + scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); + } + + @AfterEach + void afterEach() { + if (executorService != null) { + executorService.shutdownNow(); + } + if (scheduledExecutorService != null) { + scheduledExecutorService.shutdownNow(); + } + } + + @ParameterizedTest + @ValueSource(longs = {0, SLEEP_DURATION_MILLIS}) + void sleepAsync(final long durationMs) { + Duration duration = Duration.ofMillis(durationMs); + try (DefaultAsyncClientExecutor backedByExecutorService = new DefaultAsyncClientExecutor(executorService); + DefaultAsyncClientExecutor backedByScheduledExecutorService = new DefaultAsyncClientExecutor(scheduledExecutorService)) { + assertAll( + () -> assertSleepAsync(backedByExecutorService, duration), + () -> assertSleepAsync(backedByScheduledExecutorService, duration) + ); + } + } + + private static void assertSleepAsync( + final DefaultAsyncClientExecutor clientExecutor, final Duration duration) throws Exception { + StartTime startTime = StartTime.now(); + CompletableFuture callbackDelayFuture = new CompletableFuture<>(); + CompletableFuture callbackThreadFuture = new CompletableFuture<>(); + clientExecutor.sleepAsync(duration, (result, t) -> { + if (t != null) { + callbackDelayFuture.completeExceptionally(t); + } else { + callbackDelayFuture.complete(startTime.elapsed()); + } + callbackThreadFuture.complete(Thread.currentThread()); + }); + // if `duration` is zero, the futures are expected to be completed synchronously + long timeoutMs = duration.toMillis() * 2; + Duration actualCallbackDelay = callbackDelayFuture.get(timeoutMs, MILLISECONDS); + Thread actualCallbackThread = callbackThreadFuture.get(timeoutMs, MILLISECONDS); + assertTrue(actualCallbackDelay.compareTo(duration) >= 0); + if (duration.isZero()) { + assertSame(Thread.currentThread(), actualCallbackThread); + } else { + assertTrue(actualCallbackDelay.compareTo(duration.multipliedBy(2)) < 0); + assertNotSame(Thread.currentThread(), actualCallbackThread); + } + } + + @ParameterizedTest + @ValueSource(longs = {0, SLEEP_DURATION_MILLIS}) + void closeBeforeSleepAsync(final long durationMs) { + Duration duration = Duration.ofMillis(durationMs); + try (DefaultAsyncClientExecutor backedByExecutorService = new DefaultAsyncClientExecutor(executorService); + DefaultAsyncClientExecutor backedByScheduledExecutorService = new DefaultAsyncClientExecutor(scheduledExecutorService)) { + assertAll( + () -> assertCloseOrBackingExecutorShutdownBeforeSleepAsync(backedByExecutorService, duration, backedByExecutorService::close), + () -> assertCloseOrBackingExecutorShutdownBeforeSleepAsync(backedByScheduledExecutorService, duration, backedByScheduledExecutorService::close) + ); + } + } + + /** + * {@link AsyncClientExecutor#close()} and {@link com.mongodb.connection.NettyTransportSettings.Builder#eventLoopGroup(EventLoopGroup)} + * forbit this scenario, but we still handle it. + */ + @Test + void backingExecutorShutdownBeforeSleepAsync() { + Duration duration = Duration.ofMillis(SLEEP_DURATION_MILLIS); + try (DefaultAsyncClientExecutor backedByExecutorService = new DefaultAsyncClientExecutor(executorService); + DefaultAsyncClientExecutor backedByScheduledExecutorService = new DefaultAsyncClientExecutor(scheduledExecutorService)) { + assertAll( + () -> assertCloseOrBackingExecutorShutdownBeforeSleepAsync(backedByExecutorService, duration, executorService::shutdownNow), + () -> assertCloseOrBackingExecutorShutdownBeforeSleepAsync(backedByScheduledExecutorService, duration, scheduledExecutorService::shutdownNow) + ); + } + } + + private static void assertCloseOrBackingExecutorShutdownBeforeSleepAsync( + final DefaultAsyncClientExecutor clientExecutor, final Duration duration, final Runnable doBeforeSleepAsync) throws Exception { + doBeforeSleepAsync.run(); + AtomicInteger completionCount = new AtomicInteger(); + CompletableFuture callbackFuture = new CompletableFuture<>(); + clientExecutor.sleepAsync(duration, (result, t) -> { + completionCount.incrementAndGet(); + if (t != null) { + callbackFuture.completeExceptionally(t); + } else { + callbackFuture.complete(result); + } + }); + Throwable callbackException = assertThrows(CompletionException.class, () -> callbackFuture.getNow(null)).getCause(); + assertInstanceOf(RejectedExecutionException.class, callbackException); + Thread.sleep(duration.toMillis() * 2); + assertEquals(1, completionCount.get()); + } + + @Test + void closeWhileSleepingInSleepAsync() { + Duration duration = Duration.ofMillis(SLEEP_DURATION_MILLIS); + try (DefaultAsyncClientExecutor backedByExecutorService = new DefaultAsyncClientExecutor(executorService); + DefaultAsyncClientExecutor backedByScheduledExecutorService = new DefaultAsyncClientExecutor(scheduledExecutorService)) { + assertAll( + () -> assertCloseWhileSleepingInSleepAsync(backedByExecutorService, duration), + () -> assertCloseWhileSleepingInSleepAsync(backedByScheduledExecutorService, duration) + ); + } + } + + private static void assertCloseWhileSleepingInSleepAsync( + final DefaultAsyncClientExecutor clientExecutor, final Duration duration) throws Exception { + AtomicInteger completionCount = new AtomicInteger(); + CompletableFuture callbackFuture = new CompletableFuture<>(); + clientExecutor.sleepAsync(duration, (result, t) -> { + completionCount.incrementAndGet(); + if (t != null) { + callbackFuture.completeExceptionally(t); + } else { + callbackFuture.complete(result); + } + }); + clientExecutor.close(); + long timeoutMs = duration.toMillis() * 2; + Throwable callbackException = assertThrows(ExecutionException.class, () -> callbackFuture.get(timeoutMs, MILLISECONDS)).getCause(); + assertInstanceOf(RejectedExecutionException.class, callbackException); + Thread.sleep(timeoutMs); + assertEquals(1, completionCount.get()); + } + + @Test + void closeWhileCallbackIsBeingCompletedInSleepAsync() { + Duration duration = Duration.ofMillis(SLEEP_DURATION_MILLIS); + try (DefaultAsyncClientExecutor backedByExecutorService = new DefaultAsyncClientExecutor(executorService); + DefaultAsyncClientExecutor backedByScheduledExecutorService = new DefaultAsyncClientExecutor(scheduledExecutorService)) { + assertAll( + () -> assertCloseWhileCallbackIsBeingCompletedInSleepAsync(backedByExecutorService, duration), + () -> assertCloseWhileCallbackIsBeingCompletedInSleepAsync(backedByScheduledExecutorService, duration) + ); + } + } + + private static void assertCloseWhileCallbackIsBeingCompletedInSleepAsync( + final DefaultAsyncClientExecutor clientExecutor, final Duration duration) throws Exception { + AtomicInteger completionCount = new AtomicInteger(); + CompletableFuture callbackFuture = new CompletableFuture<>(); + CompletableFuture closeFuture = new CompletableFuture<>(); + clientExecutor.sleepAsync(duration, (result, t) -> { + completionCount.incrementAndGet(); + if (t != null) { + callbackFuture.completeExceptionally(t); + } else { + callbackFuture.complete(result); + } + waitForCompletion(closeFuture, duration); + }); + waitForCompletion(callbackFuture, duration.multipliedBy(2)); + clientExecutor.close(); + closeFuture.complete(null); + long timeoutMs = duration.toMillis() * 2; + assertDoesNotThrow(() -> callbackFuture.get(timeoutMs, MILLISECONDS)); + Thread.sleep(timeoutMs); + assertEquals(1, completionCount.get()); + } + + private static void waitForCompletion(final Future future, final Duration duration) { + try { + future.get(duration.toNanos(), NANOSECONDS); + } catch (InterruptedException e) { + throw interruptAndCreateMongoInterruptedException(null, e); + } catch (TimeoutException e) { + throw new RuntimeException(e); + } catch (ExecutionException e) { + // nothing to do + } + } +} diff --git a/driver-core/src/test/unit/com/mongodb/internal/thread/MongoScheduledThreadPoolExecutorTest.java b/driver-core/src/test/unit/com/mongodb/internal/thread/MongoScheduledThreadPoolExecutorTest.java new file mode 100644 index 00000000000..14d2a7ce2ef --- /dev/null +++ b/driver-core/src/test/unit/com/mongodb/internal/thread/MongoScheduledThreadPoolExecutorTest.java @@ -0,0 +1,64 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.util.concurrent.Callable; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; + +class MongoScheduledThreadPoolExecutorTest extends MongoThreadPoolExecutorTest { + @ParameterizedTest + @CsvSource({ + "false, false", + "false, true", + "true, false", + "true, true"}) + @Override + void delegateErrorToDefaultUncaughtExceptionHandlerOrLog( + final boolean taskCompletesAbruptlyWithError, + final boolean setDefaultUncaughtExceptionHandler) throws Exception { + MongoScheduledThreadPoolExecutor executor = new MongoScheduledThreadPoolExecutor(1, new DaemonThreadFactory("test")); + try { + Error error = new Error("expected error"); + RuntimeException exception = new RuntimeException("expected exception"); + Throwable expectedThrowable = taskCompletesAbruptlyWithError ? error : exception; + Runnable runnable = () -> { + if (taskCompletesAbruptlyWithError) { + throw error; + } else { + throw exception; + } + }; + Callable callable = () -> { + runnable.run(); + return null; + }; + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.execute(runnable)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.submit(runnable)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.submit(runnable, null)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.submit(callable)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.schedule(runnable, 0, MILLISECONDS)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.schedule(callable, 0, MILLISECONDS)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.scheduleAtFixedRate(runnable, 0, 1, MILLISECONDS)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.scheduleWithFixedDelay(runnable, 0, 1, MILLISECONDS)); + } finally { + executor.shutdownNow(); + } + } +} diff --git a/driver-core/src/test/unit/com/mongodb/internal/thread/MongoThreadPoolExecutorTest.java b/driver-core/src/test/unit/com/mongodb/internal/thread/MongoThreadPoolExecutorTest.java new file mode 100644 index 00000000000..f17c87a4bce --- /dev/null +++ b/driver-core/src/test/unit/com/mongodb/internal/thread/MongoThreadPoolExecutorTest.java @@ -0,0 +1,143 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import com.mongodb.lang.Nullable; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.PrintStream; +import java.io.UnsupportedEncodingException; +import java.lang.Thread.UncaughtExceptionHandler; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.LinkedBlockingQueue; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class MongoThreadPoolExecutorTest { + private static final long TIMEOUT_MILLIS = 100; + + @Nullable + private PrintStream originalStderr; + private ByteArrayOutputStream stderrTap; + @Nullable + private UncaughtExceptionHandler originalUncaughtExceptionHandler; + + @BeforeEach + void beforeEach() throws UnsupportedEncodingException { + originalUncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler(); + originalStderr = System.err; + stderrTap = new ByteArrayOutputStream(); + System.setErr(new PrintStream(stderrTap, true, StandardCharsets.UTF_8.name())); + } + + @AfterEach + void afterEach() throws IOException { + try { + Thread.setDefaultUncaughtExceptionHandler(originalUncaughtExceptionHandler); + } finally { + try { + if (originalStderr != null) { + System.setErr(originalStderr); + } + } finally { + if (stderrTap != null) { + stderrTap.close(); + } + } + } + } + + @ParameterizedTest + @CsvSource({ + "false, false", + "false, true", + "true, false", + "true, true" + }) + void delegateErrorToDefaultUncaughtExceptionHandlerOrLog( + final boolean taskCompletesAbruptlyWithError, + final boolean setDefaultUncaughtExceptionHandler) throws Exception { + MongoThreadPoolExecutor executor = new MongoThreadPoolExecutor( + 1, 1, Duration.ofMillis(TIMEOUT_MILLIS), new LinkedBlockingQueue<>(), new DaemonThreadFactory("test")); + try { + Error error = new Error("expected error"); + RuntimeException exception = new RuntimeException("expected exception"); + Throwable expectedThrowable = taskCompletesAbruptlyWithError ? error : exception; + Runnable runnable = () -> { + if (taskCompletesAbruptlyWithError) { + throw error; + } else { + throw exception; + } + }; + Callable callable = () -> { + runnable.run(); + return null; + }; + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.execute(runnable)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.submit(runnable)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.submit(runnable, null)); + assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog(expectedThrowable, setDefaultUncaughtExceptionHandler, () -> executor.submit(callable)); + } finally { + executor.shutdownNow(); + } + } + + void assertDelegateErrorToDefaultUncaughtExceptionHandlerOrLog( + final Throwable expectedThrowable, + final boolean setDefaultUncaughtExceptionHandler, + final Runnable submitThrowingTask) throws Exception { + CompletableFuture uncaughtExceptionFuture = new CompletableFuture<>(); + if (setDefaultUncaughtExceptionHandler) { + Thread.setDefaultUncaughtExceptionHandler((t, e) -> { + uncaughtExceptionFuture.complete(e); + }); + } else { + // we remove the original `UncaughtExceptionHandler` to guarantee that uncaught exceptions are printed to `System.err` + Thread.setDefaultUncaughtExceptionHandler(null); + } + stderrTap.reset(); + submitThrowingTask.run(); + Thread.sleep(TIMEOUT_MILLIS); + if (setDefaultUncaughtExceptionHandler) { + Throwable actualUncaughtException = uncaughtExceptionFuture.get(TIMEOUT_MILLIS, MILLISECONDS); + if (expectedThrowable instanceof Error) { + assertSame(expectedThrowable, actualUncaughtException); + } else { + assertInstanceOf(Error.class, actualUncaughtException); + assertSame(expectedThrowable, actualUncaughtException.getCause()); + } + } else { + String actualLoggedMessage = stderrTap.toString(StandardCharsets.UTF_8.name()); + assertTrue(actualLoggedMessage.contains(expectedThrowable.getClass().getName()) && actualLoggedMessage.contains(expectedThrowable.getMessage()), + () -> { + return String.format("actualLoggedMessage=%s does not contain information about expectedThrowable=%s", + actualLoggedMessage, expectedThrowable); + }); + } + } +} diff --git a/driver-core/src/test/unit/com/mongodb/internal/thread/UnimplementedAsyncClientExecutorTest.java b/driver-core/src/test/unit/com/mongodb/internal/thread/UnimplementedAsyncClientExecutorTest.java new file mode 100644 index 00000000000..632e92a29d7 --- /dev/null +++ b/driver-core/src/test/unit/com/mongodb/internal/thread/UnimplementedAsyncClientExecutorTest.java @@ -0,0 +1,59 @@ +/* + * Copyright 2008-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. + */ +package com.mongodb.internal.thread; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class UnimplementedAsyncClientExecutorTest { + @ParameterizedTest + @ValueSource(longs = {0, 200}) + void sleepAsync(final long durationMs) { + Duration duration = Duration.ofMillis(durationMs); + assertSleepAsync(duration); + } + + private static void assertSleepAsync(final Duration duration) { + CompletableFuture callbackExceptionFuture = new CompletableFuture<>(); + CompletableFuture callbackThreadFuture = new CompletableFuture<>(); + try (UnimplementedAsyncClientExecutor unimplementedClientExecutor = UnimplementedAsyncClientExecutor.instance()) { + unimplementedClientExecutor.sleepAsync(duration, (result, t) -> { + if (t != null) { + callbackExceptionFuture.completeExceptionally(t); + } else { + callbackExceptionFuture.complete(result); + } + callbackThreadFuture.complete(Thread.currentThread()); + }); + } + if (duration.isZero()) { + assertDoesNotThrow(() -> callbackExceptionFuture.getNow(null)); + } else { + Throwable callbackException = assertThrows(CompletionException.class, () -> callbackExceptionFuture.getNow(null)).getCause(); + assertSame(AssertionError.class, callbackException.getClass()); + } + Thread actualCallbackThread = callbackThreadFuture.getNow(null); + assertSame(Thread.currentThread(), actualCallbackThread); + } +} diff --git a/driver-legacy/src/main/com/mongodb/MongoClient.java b/driver-legacy/src/main/com/mongodb/MongoClient.java index 06dac49c671..de1d7e641bc 100644 --- a/driver-legacy/src/main/com/mongodb/MongoClient.java +++ b/driver-legacy/src/main/com/mongodb/MongoClient.java @@ -41,6 +41,7 @@ import com.mongodb.internal.connection.StreamFactoryFactory; import com.mongodb.internal.diagnostics.logging.Logger; import com.mongodb.internal.diagnostics.logging.Loggers; +import com.mongodb.internal.observability.micrometer.TracingManager; import com.mongodb.internal.session.ServerSessionPool; import com.mongodb.internal.thread.DaemonThreadFactory; import com.mongodb.internal.validator.NoOpFieldNameValidator; @@ -860,7 +861,7 @@ private void cleanCursors() { ServerCursorAndNamespace cur; while ((cur = orphanedCursors.poll()) != null) { OperationContext operationContext = new OperationContext(IgnorableRequestContext.INSTANCE, NoOpSessionContext.INSTANCE, - new TimeoutContext(getTimeoutSettings()), options.getServerApi()); + new TimeoutContext(getTimeoutSettings()), delegate.getClientExecutor(), TracingManager.NO_OP, options.getServerApi(), null); ReadWriteBinding binding = new SingleServerBinding(delegate.getCluster(), cur.serverCursor.getAddress()); try { diff --git a/driver-legacy/src/test/unit/com/mongodb/MongoClientSpecification.groovy b/driver-legacy/src/test/unit/com/mongodb/MongoClientSpecification.groovy index 1389a41c760..f6a51e5607c 100644 --- a/driver-legacy/src/test/unit/com/mongodb/MongoClientSpecification.groovy +++ b/driver-legacy/src/test/unit/com/mongodb/MongoClientSpecification.groovy @@ -23,6 +23,8 @@ import com.mongodb.client.model.geojson.MultiPolygon import com.mongodb.connection.ClusterSettings import com.mongodb.internal.connection.ClientMetadata import com.mongodb.internal.connection.Cluster +import com.mongodb.internal.connection.StreamFactoryFactory +import com.mongodb.internal.thread.AsyncClientExecutor import org.bson.BsonDocument import org.bson.Document import org.bson.codecs.UuidCodec @@ -314,7 +316,7 @@ class MongoClientSpecification extends Specification { def clusterStub = Stub(Cluster) clusterStub.getClientMetadata() >> new ClientMetadata("test", MongoDriverInformation.builder().build()) - def client = new MongoClientImpl(clusterStub, null, MongoClientSettings.builder().build(), null, executor) + def client = new MongoClientImpl(clusterStub, null, MongoClientSettings.builder().build(), mockStreamFactoryFactory(), executor) when: client.watch((Class) null) @@ -365,4 +367,10 @@ class MongoClientSpecification extends Specification { cleanup: client?.close() } + + def mockStreamFactoryFactory() { + Mock(StreamFactoryFactory) { + getClientExecutor() >> AsyncClientExecutor.NO_OP + } + } } diff --git a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/MongoClients.java b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/MongoClients.java index 57ee076039e..da722d16ba0 100644 --- a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/MongoClients.java +++ b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/MongoClients.java @@ -27,6 +27,7 @@ import com.mongodb.internal.connection.InternalConnectionPoolSettings; import com.mongodb.internal.connection.StreamFactory; import com.mongodb.internal.connection.StreamFactoryFactory; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import com.mongodb.reactivestreams.client.internal.MongoClientImpl; import com.mongodb.spi.dns.InetAddressResolver; @@ -118,7 +119,7 @@ public static MongoClient create(final MongoClientSettings settings, @Nullable f StreamFactory streamFactory = getStreamFactory(streamFactoryFactory, settings, false); StreamFactory heartbeatStreamFactory = getStreamFactory(streamFactoryFactory, settings, true); MongoDriverInformation wrappedMongoDriverInformation = wrapMongoDriverInformation(mongoDriverInformation); - Cluster cluster = createCluster(settings, wrappedMongoDriverInformation, streamFactory, heartbeatStreamFactory); + Cluster cluster = createCluster(settings, wrappedMongoDriverInformation, streamFactory, heartbeatStreamFactory, streamFactoryFactory.getClientExecutor()); return new MongoClientImpl(settings, wrappedMongoDriverInformation, cluster, streamFactoryFactory); } @@ -135,12 +136,13 @@ public static CodecRegistry getDefaultCodecRegistry() { private static Cluster createCluster(final MongoClientSettings settings, @Nullable final MongoDriverInformation mongoDriverInformation, - final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory) { + final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory, + final AsyncClientExecutor clientExecutor) { notNull("settings", settings); return new DefaultClusterFactory().createCluster(settings.getClusterSettings(), settings.getServerSettings(), settings.getConnectionPoolSettings(), InternalConnectionPoolSettings.builder().prestartAsyncWorkManager(true).build(), TimeoutSettings.create(settings), streamFactory, TimeoutSettings.createHeartbeatSettings(settings), heartbeatStreamFactory, - settings.getCredential(), settings.getLoggerSettings(), getCommandListener(settings.getCommandListeners()), + clientExecutor, settings.getCredential(), settings.getLoggerSettings(), getCommandListener(settings.getCommandListeners()), settings.getApplicationName(), mongoDriverInformation, settings.getCompressorList(), settings.getServerApi(), settings.getDnsClient()); } diff --git a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/MongoClientImpl.java b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/MongoClientImpl.java index 8fda2e9294d..dbcc27f5022 100644 --- a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/MongoClientImpl.java +++ b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/MongoClientImpl.java @@ -29,12 +29,15 @@ import com.mongodb.client.model.bulk.ClientNamespacedWriteModel; import com.mongodb.connection.ClusterDescription; import com.mongodb.internal.TimeoutSettings; +import com.mongodb.internal.VisibleForTesting; import com.mongodb.internal.connection.ClientMetadata; import com.mongodb.internal.connection.Cluster; +import com.mongodb.internal.connection.StreamFactoryFactory; import com.mongodb.internal.diagnostics.logging.Logger; import com.mongodb.internal.diagnostics.logging.Loggers; import com.mongodb.internal.observability.micrometer.TracingManager; import com.mongodb.internal.session.ServerSessionPool; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import com.mongodb.reactivestreams.client.ChangeStreamPublisher; import com.mongodb.reactivestreams.client.ClientSession; @@ -56,6 +59,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import static com.mongodb.assertions.Assertions.notNull; +import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE; import static java.lang.String.format; import static org.bson.codecs.configuration.CodecRegistries.withUuidRepresentation; @@ -69,29 +73,30 @@ public final class MongoClientImpl implements MongoClient { private static final Logger LOGGER = Loggers.getLogger("client"); private final MongoClientSettings settings; - private final AutoCloseable externalResourceCloser; + private final StreamFactoryFactory streamFactoryFactory; private final MongoClusterImpl delegate; private final AtomicBoolean closed; public MongoClientImpl(final MongoClientSettings settings, final MongoDriverInformation mongoDriverInformation, final Cluster cluster, - @Nullable final AutoCloseable externalResourceCloser) { - this(settings, mongoDriverInformation, cluster, null, externalResourceCloser); + final StreamFactoryFactory streamFactoryFactory) { + this(settings, mongoDriverInformation, cluster, null, streamFactoryFactory); } - public MongoClientImpl(final MongoClientSettings settings, final MongoDriverInformation mongoDriverInformation, final Cluster cluster, - @Nullable final OperationExecutor executor) { - this(settings, mongoDriverInformation, cluster, executor, null); + @VisibleForTesting(otherwise = PRIVATE) + MongoClientImpl(final MongoClientSettings settings, final MongoDriverInformation mongoDriverInformation, final Cluster cluster, + final StreamFactoryFactory streamFactoryFactory, @Nullable final OperationExecutor executor) { + this(settings, mongoDriverInformation, cluster, executor, streamFactoryFactory); } private MongoClientImpl(final MongoClientSettings settings, final MongoDriverInformation mongoDriverInformation, final Cluster cluster, - @Nullable final OperationExecutor executor, @Nullable final AutoCloseable externalResourceCloser) { + @Nullable final OperationExecutor executor, final StreamFactoryFactory streamFactoryFactory) { notNull("settings", settings); notNull("cluster", cluster); TracingManager tracingManager = new TracingManager(settings.getObservabilitySettings()); TimeoutSettings timeoutSettings = TimeoutSettings.create(settings); - ServerSessionPool serverSessionPool = new ServerSessionPool(cluster, timeoutSettings, settings.getServerApi()); + ServerSessionPool serverSessionPool = new ServerSessionPool(cluster, streamFactoryFactory.getClientExecutor(), timeoutSettings, settings.getServerApi()); ClientSessionHelper clientSessionHelper = new ClientSessionHelper(this, serverSessionPool, tracingManager); AutoEncryptionSettings autoEncryptSettings = settings.getAutoEncryptionSettings(); @@ -117,7 +122,7 @@ private MongoClientImpl(final MongoClientSettings settings, final MongoDriverInf this.delegate = new MongoClusterImpl(cluster, crypt, operationExecutor, serverSessionPool, clientSessionHelper, mongoOperationPublisher); - this.externalResourceCloser = externalResourceCloser; + this.streamFactoryFactory = streamFactoryFactory; this.settings = settings; this.closed = new AtomicBoolean(); @@ -155,12 +160,10 @@ public void close() { } getServerSessionPool().close(); getCluster().close(); - if (externalResourceCloser != null) { - try { - externalResourceCloser.close(); - } catch (Exception e) { - LOGGER.warn("Exception closing resource", e); - } + try { + streamFactoryFactory.close(); + } catch (Exception e) { + LOGGER.warn("Exception closing resource", e); } } } @@ -336,4 +339,11 @@ public void appendMetadata(final MongoDriverInformation mongoDriverInformation) clientMetadata.append(mongoDriverInformation); LOGGER.info(format("MongoClient metadata has been updated to %s", clientMetadata.getBsonDocument())); } + + /** + * @see StreamFactoryFactory#getClientExecutor() + */ + public AsyncClientExecutor getClientExecutor() { + return streamFactoryFactory.getClientExecutor(); + } } diff --git a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/OperationExecutorImpl.java b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/OperationExecutorImpl.java index 5e11e0503a7..fa1956c69a1 100644 --- a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/OperationExecutorImpl.java +++ b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/OperationExecutorImpl.java @@ -243,6 +243,7 @@ private OperationContext getOperationContext(final RequestContext requestContext requestContext, new ReadConcernAwareNoOpSessionContext(readConcern), createTimeoutContext(session, timeoutSettings), + mongoClient.getClientExecutor(), tracingManager, mongoClient.getSettings().getServerApi(), commandName, diff --git a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/KeyManagementService.java b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/KeyManagementService.java index 67ebf421c9c..6dda49da685 100644 --- a/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/KeyManagementService.java +++ b/driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/crypt/KeyManagementService.java @@ -36,6 +36,7 @@ import com.mongodb.internal.crypt.capi.MongoKeyDecryptor; import com.mongodb.internal.diagnostics.logging.Logger; import com.mongodb.internal.diagnostics.logging.Loggers; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.internal.time.Timeout; import com.mongodb.lang.Nullable; import org.bson.ByteBuf; @@ -182,7 +183,7 @@ private OperationContext createOperationContext(@Nullable final Timeout operatio throw new MongoOperationTimeoutException(TIMEOUT_ERROR_MESSAGE); }); } - return OperationContext.simpleOperationContext(new TimeoutContext(timeoutSettings)); + return OperationContext.simpleOperationContext(timeoutSettings, null, AsyncClientExecutor.NO_OP); } @NonNull diff --git a/driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/MongoClientImplTest.java b/driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/MongoClientImplTest.java index 0fda131f4ff..1ba8265273f 100644 --- a/driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/MongoClientImplTest.java +++ b/driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/MongoClientImplTest.java @@ -24,9 +24,11 @@ import com.mongodb.internal.client.model.changestream.ChangeStreamLevel; import com.mongodb.internal.connection.ClientMetadata; import com.mongodb.internal.connection.Cluster; +import com.mongodb.internal.connection.StreamFactoryFactory; import com.mongodb.internal.mockito.MongoMockito; import com.mongodb.internal.observability.micrometer.TracingManager; import com.mongodb.internal.session.ServerSessionPool; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.reactivestreams.client.ChangeStreamPublisher; import com.mongodb.reactivestreams.client.ClientSession; import com.mongodb.reactivestreams.client.ListDatabasesPublisher; @@ -204,11 +206,14 @@ void testStartSession() { private MongoClientImpl createMongoClient() { MongoDriverInformation mongoDriverInformation = MongoDriverInformation.builder().driverName("reactive-streams").build(); - Cluster mock = MongoMockito.mock(Cluster.class, cluster -> { - when(cluster.getClientMetadata()) + Cluster cluster = MongoMockito.mock(Cluster.class, mock -> { + when(mock.getClientMetadata()) .thenReturn(new ClientMetadata("test", mongoDriverInformation)); }); + StreamFactoryFactory streamFactoryFactory = MongoMockito.mock(StreamFactoryFactory.class, mock -> { + when(mock.getClientExecutor()).thenReturn(AsyncClientExecutor.NO_OP); + }); return new MongoClientImpl(MongoClientSettings.builder().build(), - mongoDriverInformation, mock, OPERATION_EXECUTOR); + mongoDriverInformation, cluster, streamFactoryFactory, OPERATION_EXECUTOR); } } diff --git a/driver-sync/src/main/com/mongodb/client/internal/Clusters.java b/driver-sync/src/main/com/mongodb/client/internal/Clusters.java index 6c57505e090..ffedcfbace7 100644 --- a/driver-sync/src/main/com/mongodb/client/internal/Clusters.java +++ b/driver-sync/src/main/com/mongodb/client/internal/Clusters.java @@ -47,7 +47,7 @@ public static Cluster createCluster(final MongoClientSettings settings, return new DefaultClusterFactory().createCluster(settings.getClusterSettings(), settings.getServerSettings(), settings.getConnectionPoolSettings(), InternalConnectionPoolSettings.builder().build(), TimeoutSettings.create(settings), streamFactory, - TimeoutSettings.createHeartbeatSettings(settings), heartbeatStreamFactory, + TimeoutSettings.createHeartbeatSettings(settings), heartbeatStreamFactory, streamFactoryFactory.getClientExecutor(), settings.getCredential(), settings.getLoggerSettings(), getCommandListener(settings.getCommandListeners()), settings.getApplicationName(), mongoDriverInformation, settings.getCompressorList(), settings.getServerApi(), settings.getDnsClient()); diff --git a/driver-sync/src/main/com/mongodb/client/internal/MongoClientImpl.java b/driver-sync/src/main/com/mongodb/client/internal/MongoClientImpl.java index 9ba2139f18c..2cd7ffffa5d 100644 --- a/driver-sync/src/main/com/mongodb/client/internal/MongoClientImpl.java +++ b/driver-sync/src/main/com/mongodb/client/internal/MongoClientImpl.java @@ -36,19 +36,16 @@ import com.mongodb.client.model.bulk.ClientBulkWriteResult; import com.mongodb.client.model.bulk.ClientNamespacedWriteModel; import com.mongodb.connection.ClusterDescription; -import com.mongodb.connection.SocketSettings; import com.mongodb.internal.TimeoutSettings; import com.mongodb.internal.VisibleForTesting; import com.mongodb.internal.connection.ClientMetadata; import com.mongodb.internal.connection.Cluster; -import com.mongodb.internal.connection.DefaultClusterFactory; -import com.mongodb.internal.connection.InternalConnectionPoolSettings; -import com.mongodb.internal.connection.StreamFactory; import com.mongodb.internal.connection.StreamFactoryFactory; import com.mongodb.internal.diagnostics.logging.Logger; import com.mongodb.internal.diagnostics.logging.Loggers; import com.mongodb.internal.session.ServerSessionPool; import com.mongodb.internal.observability.micrometer.TracingManager; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import org.bson.BsonDocument; import org.bson.Document; @@ -61,7 +58,7 @@ import static com.mongodb.assertions.Assertions.notNull; import static com.mongodb.client.internal.Crypts.createCrypt; -import static com.mongodb.internal.event.EventListenerHelper.getCommandListener; +import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE; import static java.lang.String.format; import static org.bson.codecs.configuration.CodecRegistries.withUuidRepresentation; @@ -75,23 +72,23 @@ public final class MongoClientImpl implements MongoClient { private final MongoDriverInformation mongoDriverInformation; private final MongoClusterImpl delegate; private final AtomicBoolean closed; - private final AutoCloseable externalResourceCloser; + private final StreamFactoryFactory streamFactoryFactory; public MongoClientImpl(final Cluster cluster, final MongoClientSettings settings, final MongoDriverInformation mongoDriverInformation, - @Nullable final AutoCloseable externalResourceCloser) { - this(cluster, mongoDriverInformation, settings, externalResourceCloser, null); + final StreamFactoryFactory streamFactoryFactory) { + this(cluster, mongoDriverInformation, settings, streamFactoryFactory, null); } - @VisibleForTesting(otherwise = VisibleForTesting.AccessModifier.PRIVATE) + @VisibleForTesting(otherwise = PRIVATE) public MongoClientImpl(final Cluster cluster, final MongoDriverInformation mongoDriverInformation, final MongoClientSettings settings, - @Nullable final AutoCloseable externalResourceCloser, + final StreamFactoryFactory streamFactoryFactory, @Nullable final OperationExecutor operationExecutor) { - this.externalResourceCloser = externalResourceCloser; + this.streamFactoryFactory = streamFactoryFactory; this.settings = notNull("settings", settings); this.mongoDriverInformation = mongoDriverInformation; AutoEncryptionSettings autoEncryptionSettings = settings.getAutoEncryptionSettings(); @@ -100,15 +97,17 @@ public MongoClientImpl(final Cluster cluster, + SynchronousContextProvider.class.getName() + " when using the synchronous driver"); } + AsyncClientExecutor clientExecutor = streamFactoryFactory.getClientExecutor(); this.delegate = new MongoClusterImpl(autoEncryptionSettings, cluster, withUuidRepresentation(settings.getCodecRegistry(), settings.getUuidRepresentation()), (SynchronousContextProvider) settings.getContextProvider(), autoEncryptionSettings == null ? null : createCrypt(settings, autoEncryptionSettings), this, operationExecutor, settings.getReadConcern(), settings.getReadPreference(), settings.getRetryReads(), settings.getRetryWrites(), settings.getEnableOverloadRetargeting(), settings.getServerApi(), - new ServerSessionPool(cluster, TimeoutSettings.create(settings), settings.getServerApi()), + new ServerSessionPool(cluster, clientExecutor, TimeoutSettings.create(settings), settings.getServerApi()), TimeoutSettings.create(settings), settings.getUuidRepresentation(), - settings.getWriteConcern(), new TracingManager(settings.getObservabilitySettings())); + settings.getWriteConcern(), clientExecutor, + new TracingManager(settings.getObservabilitySettings())); this.closed = new AtomicBoolean(); BsonDocument clientMetadataDocument = delegate.getCluster().getClientMetadata().getBsonDocument(); @@ -124,12 +123,10 @@ public void close() { } delegate.getServerSessionPool().close(); delegate.getCluster().close(); - if (externalResourceCloser != null) { - try { - externalResourceCloser.close(); - } catch (Exception e) { - LOGGER.warn("Exception closing resource", e); - } + try { + streamFactoryFactory.close(); + } catch (Exception e) { + LOGGER.warn("Exception closing resource", e); } } } @@ -167,6 +164,7 @@ public ReadConcern getReadConcern() { } @Override + @Nullable public Long getTimeout(final TimeUnit timeUnit) { return delegate.getTimeout(timeUnit); } @@ -310,27 +308,6 @@ public ClientBulkWriteResult bulkWrite( return delegate.bulkWrite(clientSession, clientWriteModels, options); } - private static Cluster createCluster(final MongoClientSettings settings, - @Nullable final MongoDriverInformation mongoDriverInformation, - final StreamFactory streamFactory, final StreamFactory heartbeatStreamFactory) { - notNull("settings", settings); - return new DefaultClusterFactory().createCluster(settings.getClusterSettings(), settings.getServerSettings(), - settings.getConnectionPoolSettings(), InternalConnectionPoolSettings.builder().build(), - TimeoutSettings.create(settings), streamFactory, - TimeoutSettings.createHeartbeatSettings(settings), heartbeatStreamFactory, - settings.getCredential(), settings.getLoggerSettings(), getCommandListener(settings.getCommandListeners()), - settings.getApplicationName(), mongoDriverInformation, settings.getCompressorList(), settings.getServerApi(), - settings.getDnsClient()); - } - - private static StreamFactory getStreamFactory( - final StreamFactoryFactory streamFactoryFactory, - final MongoClientSettings settings, - final boolean isHeartbeat) { - SocketSettings socketSettings = isHeartbeat ? settings.getHeartbeatSocketSettings() : settings.getSocketSettings(); - return streamFactoryFactory.create(socketSettings, settings.getSslSettings()); - } - public Cluster getCluster() { return delegate.getCluster(); } @@ -354,4 +331,11 @@ public MongoClientSettings getSettings() { public MongoDriverInformation getMongoDriverInformation() { return mongoDriverInformation; } + + /** + * @see StreamFactoryFactory#getClientExecutor() + */ + public AsyncClientExecutor getClientExecutor() { + return streamFactoryFactory.getClientExecutor(); + } } diff --git a/driver-sync/src/main/com/mongodb/client/internal/MongoClusterImpl.java b/driver-sync/src/main/com/mongodb/client/internal/MongoClusterImpl.java index 5d3ac6bc678..9ed66d84f1d 100644 --- a/driver-sync/src/main/com/mongodb/client/internal/MongoClusterImpl.java +++ b/driver-sync/src/main/com/mongodb/client/internal/MongoClusterImpl.java @@ -59,6 +59,7 @@ import com.mongodb.internal.operation.ReadOperation; import com.mongodb.internal.operation.WriteOperation; import com.mongodb.internal.session.ServerSessionPool; +import com.mongodb.internal.thread.AsyncClientExecutor; import com.mongodb.lang.Nullable; import org.bson.BsonDocument; import org.bson.Document; @@ -102,6 +103,7 @@ final class MongoClusterImpl implements MongoCluster { private final UuidRepresentation uuidRepresentation; private final WriteConcern writeConcern; private final Operations operations; + private final AsyncClientExecutor clientExecutor; private final TracingManager tracingManager; MongoClusterImpl( @@ -110,7 +112,8 @@ final class MongoClusterImpl implements MongoCluster { @Nullable final OperationExecutor operationExecutor, final ReadConcern readConcern, final ReadPreference readPreference, final boolean retryReads, final boolean retryWrites, final boolean enableOverloadRetargeting, @Nullable final ServerApi serverApi, final ServerSessionPool serverSessionPool, final TimeoutSettings timeoutSettings, - final UuidRepresentation uuidRepresentation, final WriteConcern writeConcern, final TracingManager tracingManager) { + final UuidRepresentation uuidRepresentation, final WriteConcern writeConcern, + final AsyncClientExecutor clientExecutor, final TracingManager tracingManager) { this.autoEncryptionSettings = autoEncryptionSettings; this.cluster = cluster; this.codecRegistry = codecRegistry; @@ -128,6 +131,7 @@ final class MongoClusterImpl implements MongoCluster { this.timeoutSettings = timeoutSettings; this.uuidRepresentation = uuidRepresentation; this.writeConcern = writeConcern; + this.clientExecutor = clientExecutor; this.tracingManager = tracingManager; operations = new Operations<>( null, @@ -172,35 +176,35 @@ public Long getTimeout(final TimeUnit timeUnit) { public MongoCluster withCodecRegistry(final CodecRegistry codecRegistry) { return new MongoClusterImpl(autoEncryptionSettings, cluster, codecRegistry, contextProvider, crypt, originator, operationExecutor, readConcern, readPreference, retryReads, retryWrites, enableOverloadRetargeting, serverApi, serverSessionPool, timeoutSettings, - uuidRepresentation, writeConcern, tracingManager); + uuidRepresentation, writeConcern, clientExecutor, tracingManager); } @Override public MongoCluster withReadPreference(final ReadPreference readPreference) { return new MongoClusterImpl(autoEncryptionSettings, cluster, codecRegistry, contextProvider, crypt, originator, operationExecutor, readConcern, readPreference, retryReads, retryWrites, enableOverloadRetargeting, serverApi, serverSessionPool, timeoutSettings, - uuidRepresentation, writeConcern, tracingManager); + uuidRepresentation, writeConcern, clientExecutor, tracingManager); } @Override public MongoCluster withWriteConcern(final WriteConcern writeConcern) { return new MongoClusterImpl(autoEncryptionSettings, cluster, codecRegistry, contextProvider, crypt, originator, operationExecutor, readConcern, readPreference, retryReads, retryWrites, enableOverloadRetargeting, serverApi, serverSessionPool, timeoutSettings, - uuidRepresentation, writeConcern, tracingManager); + uuidRepresentation, writeConcern, clientExecutor, tracingManager); } @Override public MongoCluster withReadConcern(final ReadConcern readConcern) { return new MongoClusterImpl(autoEncryptionSettings, cluster, codecRegistry, contextProvider, crypt, originator, operationExecutor, readConcern, readPreference, retryReads, retryWrites, enableOverloadRetargeting, serverApi, serverSessionPool, timeoutSettings, - uuidRepresentation, writeConcern, tracingManager); + uuidRepresentation, writeConcern, clientExecutor, tracingManager); } @Override public MongoCluster withTimeout(final long timeout, final TimeUnit timeUnit) { return new MongoClusterImpl(autoEncryptionSettings, cluster, codecRegistry, contextProvider, crypt, originator, operationExecutor, readConcern, readPreference, retryReads, retryWrites, enableOverloadRetargeting, serverApi, serverSessionPool, - timeoutSettings.withTimeout(timeout, timeUnit), uuidRepresentation, writeConcern, tracingManager); + timeoutSettings.withTimeout(timeout, timeUnit), uuidRepresentation, writeConcern, clientExecutor, tracingManager); } @Override @@ -397,7 +401,7 @@ private ClientBulkWriteResult executeBulkWrite( return operationExecutor.execute(operations.clientBulkWriteOperation(clientWriteModels, options), readConcern, clientSession); } - final class OperationExecutorImpl implements OperationExecutor { + private final class OperationExecutorImpl implements OperationExecutor { private final TimeoutSettings executorTimeoutSettings; OperationExecutorImpl(final TimeoutSettings executorTimeoutSettings) { @@ -526,6 +530,7 @@ private OperationContext getOperationContext(final ClientSession session, final getRequestContext(), new ReadConcernAwareNoOpSessionContext(readConcern), createTimeoutContext(session, executorTimeoutSettings), + clientExecutor, tracingManager, serverApi, commandName, diff --git a/driver-sync/src/test/functional/com/mongodb/client/MongoClientTest.java b/driver-sync/src/test/functional/com/mongodb/client/MongoClientTest.java index 6d3413f032a..720ae928380 100644 --- a/driver-sync/src/test/functional/com/mongodb/client/MongoClientTest.java +++ b/driver-sync/src/test/functional/com/mongodb/client/MongoClientTest.java @@ -25,7 +25,9 @@ import com.mongodb.event.ClusterOpeningEvent; import com.mongodb.internal.connection.ClientMetadata; import com.mongodb.internal.connection.Cluster; +import com.mongodb.internal.connection.StreamFactoryFactory; import com.mongodb.internal.mockito.MongoMockito; +import com.mongodb.internal.thread.AsyncClientExecutor; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -63,7 +65,7 @@ public void clusterOpening(final ClusterOpeningEvent event) { } @Test - void shouldCloseExternalResources() throws Exception { + void shouldCloseStreamFactoryFactory() { //given MongoDriverInformation mongoDriverInformation = MongoDriverInformation.builder().build(); @@ -74,11 +76,12 @@ void shouldCloseExternalResources() throws Exception { when(mockedCluster.getClientMetadata()) .thenReturn(new ClientMetadata("test", mongoDriverInformation)); }); - AutoCloseable externalResource = MongoMockito.mock( - AutoCloseable.class, - mockedExternalResource -> { + StreamFactoryFactory streamFactoryFactory = MongoMockito.mock( + StreamFactoryFactory.class, + mock -> { + when(mock.getClientExecutor()).thenReturn(AsyncClientExecutor.NO_OP); try { - doNothing().when(mockedExternalResource).close(); + doNothing().when(mock).close(); } catch (Exception e) { throw new RuntimeException(e); } @@ -88,13 +91,13 @@ void shouldCloseExternalResources() throws Exception { cluster, MongoClientSettings.builder().build(), mongoDriverInformation, - externalResource); + streamFactoryFactory); //when mongoClient.close(); //then - Mockito.verify(externalResource).close(); + Mockito.verify(streamFactoryFactory).close(); Mockito.verify(cluster).close(); } } diff --git a/driver-sync/src/test/unit/com/mongodb/client/MongoClientSpecification.groovy b/driver-sync/src/test/unit/com/mongodb/client/MongoClientSpecification.groovy index 916d8179af5..f8a825b96ab 100644 --- a/driver-sync/src/test/unit/com/mongodb/client/MongoClientSpecification.groovy +++ b/driver-sync/src/test/unit/com/mongodb/client/MongoClientSpecification.groovy @@ -37,6 +37,8 @@ import com.mongodb.internal.TimeoutSettings import com.mongodb.internal.client.model.changestream.ChangeStreamLevel import com.mongodb.internal.connection.ClientMetadata import com.mongodb.internal.connection.Cluster +import com.mongodb.internal.connection.StreamFactoryFactory +import com.mongodb.internal.thread.AsyncClientExecutor import org.bson.BsonDocument import org.bson.Document import org.bson.codecs.UuidCodec @@ -70,7 +72,7 @@ class MongoClientSpecification extends Specification { .retryWrites(true) .codecRegistry(CODEC_REGISTRY) .build() - def client = new MongoClientImpl(Stub(Cluster), null, settings, null, new TestOperationExecutor([])) + def client = new MongoClientImpl(Stub(Cluster), null, settings, mockStreamFactoryFactory(), new TestOperationExecutor([])) when: def database = client.getDatabase('name') @@ -87,7 +89,7 @@ class MongoClientSpecification extends Specification { def 'should use ListDatabasesIterableImpl correctly'() { given: def executor = new TestOperationExecutor([null, null]) - def client = new MongoClientImpl(Stub(Cluster), null, MongoClientSettings.builder().build(), null, executor) + def client = new MongoClientImpl(Stub(Cluster), null, MongoClientSettings.builder().build(), mockStreamFactoryFactory(), executor) def listDatabasesMethod = client.&listDatabases def listDatabasesNamesMethod = client.&listDatabaseNames @@ -132,7 +134,7 @@ class MongoClientSpecification extends Specification { .build() def readPreference = settings.getReadPreference() def readConcern = settings.getReadConcern() - def client = new MongoClientImpl(Stub(Cluster), null, settings, null, executor) + def client = new MongoClientImpl(Stub(Cluster), null, settings, mockStreamFactoryFactory(), executor) def watchMethod = client.&watch when: @@ -169,7 +171,7 @@ class MongoClientSpecification extends Specification { def 'should validate the ChangeStreamIterable pipeline data correctly'() { given: def executor = new TestOperationExecutor([]) - def client = new MongoClientImpl(Stub(Cluster), null, MongoClientSettings.builder().build(), null, + def client = new MongoClientImpl(Stub(Cluster), null, MongoClientSettings.builder().build(), mockStreamFactoryFactory(), executor) when: @@ -201,7 +203,7 @@ class MongoClientSpecification extends Specification { 1 * getClientMetadata() >> new ClientMetadata("test", driverInformation) } def settings = MongoClientSettings.builder().build() - def client = new MongoClientImpl(cluster, driverInformation, settings, null, new TestOperationExecutor([])) + def client = new MongoClientImpl(cluster, driverInformation, settings, mockStreamFactoryFactory(), new TestOperationExecutor([])) expect: client.getClusterDescription() == clusterDescription @@ -216,7 +218,7 @@ class MongoClientSpecification extends Specification { .build() when: - def client = new MongoClientImpl(Stub(Cluster), null, settings, null, new TestOperationExecutor([])) + def client = new MongoClientImpl(Stub(Cluster), null, settings, mockStreamFactoryFactory(), new TestOperationExecutor([])) then: (client.getCodecRegistry().get(UUID) as UuidCodec).getUuidRepresentation() == C_SHARP_LEGACY @@ -224,4 +226,10 @@ class MongoClientSpecification extends Specification { cleanup: client?.close() } + + def mockStreamFactoryFactory() { + Mock(StreamFactoryFactory) { + getClientExecutor() >> AsyncClientExecutor.NO_OP + } + } } diff --git a/driver-sync/src/test/unit/com/mongodb/client/internal/MongoClusterSpecification.groovy b/driver-sync/src/test/unit/com/mongodb/client/internal/MongoClusterSpecification.groovy index 34f46e7b007..b6469cfb829 100644 --- a/driver-sync/src/test/unit/com/mongodb/client/internal/MongoClusterSpecification.groovy +++ b/driver-sync/src/test/unit/com/mongodb/client/internal/MongoClusterSpecification.groovy @@ -29,6 +29,7 @@ import com.mongodb.internal.client.model.changestream.ChangeStreamLevel import com.mongodb.internal.connection.Cluster import com.mongodb.internal.session.ServerSessionPool import com.mongodb.internal.observability.micrometer.TracingManager +import com.mongodb.internal.thread.AsyncClientExecutor import org.bson.BsonDocument import org.bson.Document import org.bson.codecs.UuidCodec @@ -260,6 +261,6 @@ class MongoClusterSpecification extends Specification { new MongoClusterImpl(null, cluster, settings.codecRegistry, null, null, originator, operationExecutor, settings.readConcern, settings.readPreference, settings.retryReads, settings.retryWrites, settings.enableOverloadRetargeting, null, serverSessionPool, TimeoutSettings.create(settings), settings.uuidRepresentation, - settings.writeConcern, TracingManager.NO_OP) + settings.writeConcern, AsyncClientExecutor.NO_OP, TracingManager.NO_OP) } }