Skip to content

Ignore maxWaitTime when CSOT is enabled. #1744

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Ignore maxWaitTime when CSOT is enabled. #1744

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
6b3ac5f
Ignore maxWaitTime when CSOT is enabled.
vbabanin Jun 19, 2025
28b72c6
Remove commented-out test.
vbabanin Jun 19, 2025
2cf076b
Update maxWaitTimeout handling in getAsync method
vbabanin Jun 19, 2025
440891e
Fix test race condition.
vbabanin Jun 19, 2025
1fc6b7d
Use maxWaitTimeout consistently.
vbabanin Jun 27, 2025
820b131
Adjust linebreaks.
vbabanin Jun 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion driver-core/src/main/com/mongodb/internal/TimeoutContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -446,7 +446,10 @@ public TimeoutContext withComputedServerSelectionTimeoutContext() {
return this;
}

public Timeout startWaitQueueTimeout(final StartTime checkoutStart) {
public Timeout startMaxWaitTimeout(final StartTime checkoutStart) {
if (hasTimeoutMS()) {
return assertNotNull(timeout);
}
final long ms = getTimeoutSettings().getMaxWaitTimeMS();
return checkoutStart.timeoutAfterOrInfiniteIfNegative(ms, MILLISECONDS);
}
Expand Down
76 changes: 46 additions & 30 deletions driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import com.mongodb.event.ConnectionPoolListener;
import com.mongodb.event.ConnectionPoolReadyEvent;
import com.mongodb.event.ConnectionReadyEvent;
import com.mongodb.internal.TimeoutContext;
import com.mongodb.internal.VisibleForTesting;
import com.mongodb.internal.async.SingleResultCallback;
import com.mongodb.internal.connection.SdamServerDescriptionManager.SdamIssue;
Expand Down Expand Up @@ -98,6 +99,7 @@
import static com.mongodb.event.ConnectionClosedEvent.Reason.ERROR;
import static com.mongodb.internal.Locks.lockInterruptibly;
import static com.mongodb.internal.Locks.withLock;
import static com.mongodb.internal.TimeoutContext.createMongoTimeoutException;
import static com.mongodb.internal.VisibleForTesting.AccessModifier.PRIVATE;
import static com.mongodb.internal.async.ErrorHandlingResultCallback.errorHandlingCallback;
import static com.mongodb.internal.connection.ConcurrentPool.INFINITE_SIZE;
Expand All @@ -110,12 +112,12 @@
import static com.mongodb.internal.logging.LogMessage.Entry.Name.MAX_CONNECTING;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.MAX_IDLE_TIME_MS;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.MAX_POOL_SIZE;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.MAX_WAIT_TIMEOUT_MS;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.MIN_POOL_SIZE;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.REASON_DESCRIPTION;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_HOST;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVER_PORT;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.SERVICE_ID;
import static com.mongodb.internal.logging.LogMessage.Entry.Name.WAIT_QUEUE_TIMEOUT_MS;
import static com.mongodb.internal.logging.LogMessage.Level.DEBUG;
import static java.lang.String.format;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
Expand Down Expand Up @@ -190,12 +192,12 @@ public int getGeneration(@NonNull final ObjectId serviceId) {
@Override
public InternalConnection get(final OperationContext operationContext) {
StartTime checkoutStart = connectionCheckoutStarted(operationContext);
Timeout waitQueueTimeout = operationContext.getTimeoutContext().startWaitQueueTimeout(checkoutStart);
Timeout maxWaitTimeout = operationContext.getTimeoutContext().startMaxWaitTimeout(checkoutStart);
try {
stateAndGeneration.throwIfClosedOrPaused();
PooledConnection connection = getPooledConnection(waitQueueTimeout, checkoutStart);
PooledConnection connection = getPooledConnection(maxWaitTimeout, checkoutStart, operationContext.getTimeoutContext());
if (!connection.opened()) {
connection = openConcurrencyLimiter.openOrGetAvailable(operationContext, connection, waitQueueTimeout, checkoutStart);
connection = openConcurrencyLimiter.openOrGetAvailable(operationContext, connection, maxWaitTimeout, checkoutStart);
}
connection.checkedOutForOperation(operationContext);
connectionCheckedOut(operationContext, connection, checkoutStart);
Expand All @@ -208,7 +210,7 @@ public InternalConnection get(final OperationContext operationContext) {
@Override
public void getAsync(final OperationContext operationContext, final SingleResultCallback<InternalConnection> callback) {
StartTime checkoutStart = connectionCheckoutStarted(operationContext);
Timeout maxWaitTimeout = checkoutStart.timeoutAfterOrInfiniteIfNegative(settings.getMaxWaitTime(NANOSECONDS), NANOSECONDS);
Timeout maxWaitTimeout = operationContext.getTimeoutContext().startMaxWaitTimeout(checkoutStart);
SingleResultCallback<PooledConnection> eventSendingCallback = (connection, failure) -> {
SingleResultCallback<InternalConnection> errHandlingCallback = errorHandlingCallback(callback, LOGGER);
if (failure == null) {
Expand All @@ -225,13 +227,13 @@ public void getAsync(final OperationContext operationContext, final SingleResult
eventSendingCallback.onResult(null, e);
return;
}
asyncWorkManager.enqueue(new Task(maxWaitTimeout, checkoutStart, t -> {
asyncWorkManager.enqueue(new Task(maxWaitTimeout, checkoutStart, operationContext.getTimeoutContext(), t -> {
if (t != null) {
eventSendingCallback.onResult(null, t);
} else {
PooledConnection connection;
try {
connection = getPooledConnection(maxWaitTimeout, checkoutStart);
connection = getPooledConnection(maxWaitTimeout, checkoutStart, operationContext.getTimeoutContext());
} catch (Exception e) {
eventSendingCallback.onResult(null, e);
return;
Expand Down Expand Up @@ -330,22 +332,24 @@ public int getGeneration() {
return stateAndGeneration.generation();
}

private PooledConnection getPooledConnection(final Timeout waitQueueTimeout, final StartTime startTime) throws MongoTimeoutException {
private PooledConnection getPooledConnection(final Timeout maxWaitTimeout,
final StartTime startTime,
final TimeoutContext timeoutContext) throws MongoTimeoutException {
try {
UsageTrackingInternalConnection internalConnection = waitQueueTimeout.call(NANOSECONDS,
UsageTrackingInternalConnection internalConnection = maxWaitTimeout.call(NANOSECONDS,
() -> pool.get(-1L, NANOSECONDS),
(ns) -> pool.get(ns, NANOSECONDS),
() -> pool.get(0L, NANOSECONDS));
while (shouldPrune(internalConnection)) {
pool.release(internalConnection, true);
internalConnection = waitQueueTimeout.call(NANOSECONDS,
internalConnection = maxWaitTimeout.call(NANOSECONDS,
() -> pool.get(-1L, NANOSECONDS),
(ns) -> pool.get(ns, NANOSECONDS),
() -> pool.get(0L, NANOSECONDS));
}
return new PooledConnection(internalConnection);
} catch (MongoTimeoutException e) {
throw createTimeoutException(startTime);
throw createTimeoutException(startTime, timeoutContext);
}
}

Expand All @@ -359,13 +363,15 @@ private PooledConnection getPooledConnectionImmediate() {
return internalConnection == null ? null : new PooledConnection(internalConnection);
}

private MongoTimeoutException createTimeoutException(final StartTime startTime) {
private MongoTimeoutException createTimeoutException(final StartTime startTime, final TimeoutContext timeoutContext) {
long elapsedMs = startTime.elapsed().toMillis();
int numPinnedToCursor = pinnedStatsManager.getNumPinnedToCursor();
int numPinnedToTransaction = pinnedStatsManager.getNumPinnedToTransaction();
String errorMessage;

if (numPinnedToCursor == 0 && numPinnedToTransaction == 0) {
return new MongoTimeoutException(format("Timed out after %d ms while waiting for a connection to server %s.",
elapsedMs, serverId.getAddress()));
errorMessage = format("Timed out after %d ms while waiting for a connection to server %s.",
elapsedMs, serverId.getAddress());
} else {
int maxSize = pool.getMaxSize();
int numInUse = pool.getInUseCount();
Expand Down Expand Up @@ -394,13 +400,15 @@ private MongoTimeoutException createTimeoutException(final StartTime startTime)
int numOtherInUse = numInUse - numPinnedToCursor - numPinnedToTransaction;
assertTrue(numOtherInUse >= 0);
assertTrue(numPinnedToCursor + numPinnedToTransaction + numOtherInUse <= maxSize);
return new MongoTimeoutException(format("Timed out after %d ms while waiting for a connection to server %s. Details: "
errorMessage = format("Timed out after %d ms while waiting for a connection to server %s. Details: "
+ "maxPoolSize: %s, connections in use by cursors: %d, connections in use by transactions: %d, "
+ "connections in use by other operations: %d",
elapsedMs, serverId.getAddress(),
sizeToString(maxSize), numPinnedToCursor, numPinnedToTransaction,
numOtherInUse));
numOtherInUse);
}

return timeoutContext.hasTimeoutMS() ? createMongoTimeoutException(errorMessage) : new MongoTimeoutException(errorMessage);
}

@VisibleForTesting(otherwise = PRIVATE)
Expand Down Expand Up @@ -497,7 +505,7 @@ private void connectionPoolCreated(final ConnectionPoolListener connectionPoolLi
entries.add(new LogMessage.Entry(MIN_POOL_SIZE, settings.getMinSize()));
entries.add(new LogMessage.Entry(MAX_POOL_SIZE, settings.getMaxSize()));
entries.add(new LogMessage.Entry(MAX_CONNECTING, settings.getMaxConnecting()));
entries.add(new LogMessage.Entry(WAIT_QUEUE_TIMEOUT_MS, settings.getMaxWaitTime(MILLISECONDS)));
entries.add(new LogMessage.Entry(MAX_WAIT_TIMEOUT_MS, settings.getMaxWaitTime(MILLISECONDS)));

logMessage("Connection pool created", clusterId, message, entries);
}
Expand Down Expand Up @@ -903,11 +911,11 @@ private final class OpenConcurrencyLimiter {
}

PooledConnection openOrGetAvailable(final OperationContext operationContext, final PooledConnection connection,
final Timeout waitQueueTimeout, final StartTime startTime)
final Timeout maxWaitTimeout, final StartTime startTime)
throws MongoTimeoutException {
PooledConnection result = openWithConcurrencyLimit(
operationContext, connection, OpenWithConcurrencyLimitMode.TRY_GET_AVAILABLE,
waitQueueTimeout, startTime);
maxWaitTimeout, startTime);
return assertNotNull(result);
}

Expand Down Expand Up @@ -950,7 +958,7 @@ void openImmediatelyAndTryHandOverOrRelease(final OperationContext operationCont
* </ol>
*
* @param operationContext the operation context
* @param waitQueueTimeout Applies only to the first phase.
* @param maxWaitTimeout Applies only to the first phase.
* @return An {@linkplain PooledConnection#opened() opened} connection which is either the specified
* {@code connection}, or potentially a different one if {@code mode} is
* {@link OpenWithConcurrencyLimitMode#TRY_GET_AVAILABLE}, or {@code null} if {@code mode} is
Expand All @@ -959,13 +967,14 @@ void openImmediatelyAndTryHandOverOrRelease(final OperationContext operationCont
*/
@Nullable
private PooledConnection openWithConcurrencyLimit(final OperationContext operationContext,
final PooledConnection connection, final OpenWithConcurrencyLimitMode mode,
final Timeout waitQueueTimeout, final StartTime startTime)
final PooledConnection connection, final OpenWithConcurrencyLimitMode mode,
final Timeout maxWaitTimeout, final StartTime startTime)
throws MongoTimeoutException {
PooledConnection availableConnection;
try {//phase one
availableConnection = acquirePermitOrGetAvailableOpenedConnection(
mode == OpenWithConcurrencyLimitMode.TRY_GET_AVAILABLE, waitQueueTimeout, startTime);
mode == OpenWithConcurrencyLimitMode.TRY_GET_AVAILABLE, maxWaitTimeout, startTime,
operationContext.getTimeoutContext());
} catch (Exception e) {
connection.closeSilently();
throw e;
Expand Down Expand Up @@ -1007,7 +1016,8 @@ void openWithConcurrencyLimitAsync(
final SingleResultCallback<PooledConnection> callback) {
PooledConnection availableConnection;
try {//phase one
availableConnection = acquirePermitOrGetAvailableOpenedConnection(true, maxWaitTimeout, startTime);
availableConnection =
acquirePermitOrGetAvailableOpenedConnection(true, maxWaitTimeout, startTime, operationContext.getTimeoutContext());
} catch (Exception e) {
connection.closeSilently();
callback.onResult(null, e);
Expand Down Expand Up @@ -1038,7 +1048,8 @@ void openWithConcurrencyLimitAsync(
*/
@Nullable
private PooledConnection acquirePermitOrGetAvailableOpenedConnection(final boolean tryGetAvailable,
final Timeout waitQueueTimeout, final StartTime startTime)
final Timeout maxWaitTimeout, final StartTime startTime,
final TimeoutContext timeoutContext)
throws MongoTimeoutException, MongoInterruptedException {
PooledConnection availableConnection = null;
boolean expressedDesireToGetAvailableConnection = false;
Expand Down Expand Up @@ -1066,10 +1077,10 @@ private PooledConnection acquirePermitOrGetAvailableOpenedConnection(final boole
& !stateAndGeneration.throwIfClosedOrPaused()
& (availableConnection = tryGetAvailable ? tryGetAvailableConnection() : null) == null) {

Timeout.onExistsAndExpired(waitQueueTimeout, () -> {
throw createTimeoutException(startTime);
Timeout.onExistsAndExpired(maxWaitTimeout, () -> {
throw createTimeoutException(startTime, timeoutContext);
});
waitQueueTimeout.awaitOn(permitAvailableOrHandedOverOrClosedOrPausedCondition,
maxWaitTimeout.awaitOn(permitAvailableOrHandedOverOrClosedOrPausedCondition,
() -> "acquiring permit or getting available opened connection");
}
if (availableConnection == null) {
Expand Down Expand Up @@ -1389,10 +1400,15 @@ final class Task {
private final Timeout timeout;
private final StartTime startTime;
private final Consumer<RuntimeException> action;
private final TimeoutContext timeoutContext;
private boolean completed;

Task(final Timeout timeout, final StartTime startTime, final Consumer<RuntimeException> action) {
Task(final Timeout timeout,
final StartTime startTime,
final TimeoutContext timeoutContext,
final Consumer<RuntimeException> action) {
this.timeout = timeout;
this.timeoutContext = timeoutContext;
this.startTime = startTime;
this.action = action;
}
Expand All @@ -1406,7 +1422,7 @@ void failAsClosed() {
}

void failAsTimedOut() {
doComplete(() -> createTimeoutException(startTime));
doComplete(() -> createTimeoutException(startTime, timeoutContext));
}

private void doComplete(final Supplier<RuntimeException> failureSupplier) {
Expand Down
2 changes: 1 addition & 1 deletion driver-core/src/main/com/mongodb/internal/logging/LogMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ public enum Name {
MIN_POOL_SIZE("minPoolSize"),
MAX_POOL_SIZE("maxPoolSize"),
MAX_CONNECTING("maxConnecting"),
WAIT_QUEUE_TIMEOUT_MS("waitQueueTimeoutMS"),
MAX_WAIT_TIMEOUT_MS("waitQueueTimeoutMS"),
SELECTOR("selector"),
TOPOLOGY_DESCRIPTION("topologyDescription"),
REMAINING_TIME_MS("remainingTimeMS"),
Expand Down
2 changes: 2 additions & 0 deletions driver-core/src/main/com/mongodb/internal/time/StartTime.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
* A point in time used to track how much time has elapsed. In contrast to a
* Timeout, it is guaranteed to not be in the future, and is never infinite.
*
* Implementations of this interface must be immutable.
*
* @see TimePoint
*/
public interface StartTime {
Expand Down
4 changes: 3 additions & 1 deletion driver-core/src/main/com/mongodb/internal/time/Timeout.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@
import com.mongodb.internal.function.CheckedFunction;
import com.mongodb.internal.function.CheckedRunnable;
import com.mongodb.internal.function.CheckedSupplier;
import com.mongodb.lang.Nullable;
import com.mongodb.lang.NonNull;
import com.mongodb.lang.Nullable;

import java.util.Arrays;
import java.util.Collections;
Expand All @@ -40,6 +40,8 @@
/**
* A Timeout is a "deadline", point in time by which something must happen.
*
* Implementations of this interface must be immutable.
*
* @see TimePoint
*/
public interface Timeout {
Comment on lines +43 to 47
Copy link
Member

Choose a reason for hiding this comment

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

  • Is this change needed because of a change done in this PR, or is it simply something that you noticed and improved?
  • Shouldn't the same be documented for StartTime?

Copy link
Member Author

@vbabanin vbabanin Jun 26, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

This wasn’t required by changes in this PR, just a small improvement I noticed while reviewing the Timeout code. I thought documenting immutability in the Javadoc would be useful for anyone double-checking this detail, without needing to examine the implementations.

I’ve also added the note for StartTime as you suggested.

Expand Down
Loading