Skip to content

Stop BufferPoolPruner thread when last MongoClient closes #2032

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
arimu1 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Stop BufferPoolPruner thread when last MongoClient closes #2032

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 70 additions & 5 deletions driver-core/src/main/com/mongodb/internal/connection/PowerOfTwoBufferPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import com.mongodb.internal.diagnostics.logging.Logger;
import com.mongodb.internal.diagnostics.logging.Loggers;
import com.mongodb.internal.thread.DaemonThreadFactory;
import com.mongodb.lang.Nullable;
import org.bson.ByteBuf;
import org.bson.ByteBufNIO;

Expand All @@ -31,6 +32,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
* <p>This class is not part of the public API and may be removed or changed at any time</p>
Expand Down Expand Up @@ -63,7 +65,10 @@ public ByteBuffer getBuffer() {

private final Map<Integer, BufferPool> powerOfTwoToPoolMap = new HashMap<>();
private final long maxIdleTimeNanos;
private final ScheduledExecutorService pruner;
private final Object prunerLock = new Object();
private final AtomicInteger pruneRetainCount = new AtomicInteger();
@Nullable
private ScheduledExecutorService pruner;

/**
* Construct an instance with a highest power of two of 24.
Expand Down Expand Up @@ -96,19 +101,79 @@ public ByteBuffer getBuffer() {
powerOfTwo = powerOfTwo << 1;
}
maxIdleTimeNanos = timeUnit.toNanos(maxIdleTime);
pruner = Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory("BufferPoolPruner"));
}

/**
* Call this method at most once to enable a background thread that prunes idle buffers from the pool
* Enable a background thread that prunes idle buffers from the pool. Idempotent; safe to call again after
* {@link #disablePruning()}.
*/
PowerOfTwoBufferPool enablePruning() {
pruner.scheduleAtFixedRate(this::prune, maxIdleTimeNanos, maxIdleTimeNanos / 2, TimeUnit.NANOSECONDS);
synchronized (prunerLock) {
enablePruningLocked();
}
return this;
}

/**
* Stop the pruning thread if it is running. Idempotent.
*/
void disablePruning() {
pruner.shutdownNow();
synchronized (prunerLock) {
disablePruningLocked();
}
}

/**
* Record that a MongoClient using this pool has been opened. Ensures pruning is running.
* <p>
* Used for {@link #DEFAULT} so the shared pruner stays alive while any client is open and is stopped when the
* last client is closed (avoids Tomcat webapp classloader leaks from an orphaned BufferPoolPruner thread).
*/
public void retainPruning() {
synchronized (prunerLock) {
pruneRetainCount.incrementAndGet();
enablePruningLocked();
}
}

/**
* Record that a MongoClient using this pool has been closed. Stops pruning when no retainers remain.
*/
public void releasePruning() {
synchronized (prunerLock) {
int remaining = pruneRetainCount.decrementAndGet();
if (remaining <= 0) {
pruneRetainCount.set(0);
disablePruningLocked();
}
}
}

boolean isPruningEnabled() {
synchronized (prunerLock) {
return pruner != null && !pruner.isShutdown();
}
}

private void enablePruningLocked() {
if (pruner == null || pruner.isShutdown()) {
pruner = Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory("BufferPoolPruner"));
pruner.scheduleAtFixedRate(this::prune, maxIdleTimeNanos, maxIdleTimeNanos / 2, TimeUnit.NANOSECONDS);
}
}

private void disablePruningLocked() {
if (pruner != null) {
ScheduledExecutorService toShutdown = pruner;
pruner = null;
toShutdown.shutdownNow();
try {
// Ensure the BufferPoolPruner thread has exited so containers (e.g. Tomcat) can reclaim the classloader
toShutdown.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

@Override
Expand Down
69 changes: 69 additions & 0 deletions driver-core/src/test/unit/com/mongodb/internal/connection/PowerOfTwoBufferPoolTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,10 @@
import java.util.concurrent.TimeUnit;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;

public class PowerOfTwoBufferPoolTest {
private PowerOfTwoBufferPool pool;
Expand Down Expand Up @@ -91,4 +93,71 @@ public void testPruning() throws InterruptedException {
pool.disablePruning();
}
}

@Test
public void testDisablePruningStopsPruner() {
PowerOfTwoBufferPool pool = new PowerOfTwoBufferPool(10, 1, TimeUnit.MINUTES).enablePruning();
assertTrue(pool.isPruningEnabled());
pool.disablePruning();
assertFalse(pool.isPruningEnabled());
// Idempotent
pool.disablePruning();
assertFalse(pool.isPruningEnabled());
}

@Test
public void testEnablePruningAfterDisableRestartsPruner() {
PowerOfTwoBufferPool pool = new PowerOfTwoBufferPool(10, 1, TimeUnit.MINUTES).enablePruning();
pool.disablePruning();
assertFalse(pool.isPruningEnabled());
pool.enablePruning();
assertTrue(pool.isPruningEnabled());
pool.disablePruning();
}

@Test
public void testRetainReleasePruningStopsWhenLastReleased() {
PowerOfTwoBufferPool pool = new PowerOfTwoBufferPool(10, 1, TimeUnit.MINUTES);
assertFalse(pool.isPruningEnabled());

pool.retainPruning();
assertTrue(pool.isPruningEnabled());
pool.retainPruning();
assertTrue(pool.isPruningEnabled());

pool.releasePruning();
assertTrue(pool.isPruningEnabled());
pool.releasePruning();
assertFalse(pool.isPruningEnabled());
}

@Test
public void testDisablePruningTerminatesPrunerThread() throws InterruptedException {
// Account for any pre-existing BufferPoolPruner threads (e.g. PowerOfTwoBufferPool.DEFAULT)
int threadsBefore = countBufferPoolPrunerThreads();
PowerOfTwoBufferPool pool = new PowerOfTwoBufferPool(10, 1, TimeUnit.MINUTES).enablePruning();
// Force the scheduled thread to start
pool.getBuffer(64).release();
Thread.sleep(50);
assertTrue(countBufferPoolPrunerThreads() > threadsBefore);

pool.disablePruning();
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
while (countBufferPoolPrunerThreads() > threadsBefore && System.nanoTime() < deadline) {
Thread.sleep(10);
}
assertEquals(threadsBefore, countBufferPoolPrunerThreads());
assertFalse(pool.isPruningEnabled());
}

private static int countBufferPoolPrunerThreads() {
int count = 0;
for (Thread thread : Thread.getAllStackTraces().keySet()) {
String name = thread.getName();
if (name != null && name.startsWith("BufferPoolPruner-") && thread.isAlive()) {
count++;
}
}
return count;
}
}
34 changes: 23 additions & 11 deletions ...eactive-streams/src/main/com/mongodb/reactivestreams/client/internal/MongoClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import com.mongodb.internal.TimeoutSettings;
import com.mongodb.internal.connection.ClientMetadata;
import com.mongodb.internal.connection.Cluster;
import com.mongodb.internal.connection.PowerOfTwoBufferPool;
import com.mongodb.internal.diagnostics.logging.Logger;
import com.mongodb.internal.diagnostics.logging.Loggers;
import com.mongodb.internal.observability.micrometer.TracingManager;
Expand Down Expand Up @@ -123,6 +124,11 @@ private MongoClientImpl(final MongoClientSettings settings, final MongoDriverInf

BsonDocument clientMetadataDocument = delegate.getCluster().getClientMetadata().getBsonDocument();
LOGGER.info(format("MongoClient with metadata %s created with settings %s", clientMetadataDocument.toJson(), settings));
// Real clients always receive a StreamFactoryFactory as externalResourceCloser. Keep the shared
// buffer-pool pruner alive while any such client is open; stop it when the last one closes.
if (externalResourceCloser != null) {
PowerOfTwoBufferPool.DEFAULT.retainPruning();
}
}

Cluster getCluster() {
Expand All @@ -149,17 +155,23 @@ public MongoClientSettings getSettings() {
@Override
public void close() {
if (!closed.getAndSet(true)) {
Crypt crypt = getCrypt();
if (crypt != null) {
crypt.close();
}
getServerSessionPool().close();
getCluster().close();
if (externalResourceCloser != null) {
try {
externalResourceCloser.close();
} catch (Exception e) {
LOGGER.warn("Exception closing resource", e);
try {
Crypt crypt = getCrypt();
if (crypt != null) {
crypt.close();
}
getServerSessionPool().close();
getCluster().close();
if (externalResourceCloser != null) {
try {
externalResourceCloser.close();
} catch (Exception e) {
LOGGER.warn("Exception closing resource", e);
}
}
} finally {
if (externalResourceCloser != null) {
PowerOfTwoBufferPool.DEFAULT.releasePruning();
}
}
}
Expand Down
34 changes: 23 additions & 11 deletions driver-sync/src/main/com/mongodb/client/internal/MongoClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import com.mongodb.internal.connection.Cluster;
import com.mongodb.internal.connection.DefaultClusterFactory;
import com.mongodb.internal.connection.InternalConnectionPoolSettings;
import com.mongodb.internal.connection.PowerOfTwoBufferPool;
import com.mongodb.internal.connection.StreamFactory;
import com.mongodb.internal.connection.StreamFactoryFactory;
import com.mongodb.internal.diagnostics.logging.Logger;
Expand Down Expand Up @@ -113,22 +114,33 @@ public MongoClientImpl(final Cluster cluster,

BsonDocument clientMetadataDocument = delegate.getCluster().getClientMetadata().getBsonDocument();
LOGGER.info(format("MongoClient with metadata %s created with settings %s", clientMetadataDocument.toJson(), settings));
// Real clients always receive a StreamFactoryFactory as externalResourceCloser. Keep the shared
// buffer-pool pruner alive while any such client is open; stop it when the last one closes.
if (externalResourceCloser != null) {
PowerOfTwoBufferPool.DEFAULT.retainPruning();
}
}

@Override
public void close() {
if (!closed.getAndSet(true)) {
Crypt crypt = delegate.getCrypt();
if (crypt != null) {
crypt.close();
}
delegate.getServerSessionPool().close();
delegate.getCluster().close();
if (externalResourceCloser != null) {
try {
externalResourceCloser.close();
} catch (Exception e) {
LOGGER.warn("Exception closing resource", e);
try {
Crypt crypt = delegate.getCrypt();
if (crypt != null) {
crypt.close();
}
delegate.getServerSessionPool().close();
delegate.getCluster().close();
if (externalResourceCloser != null) {
try {
externalResourceCloser.close();
} catch (Exception e) {
LOGGER.warn("Exception closing resource", e);
}
}
} finally {
if (externalResourceCloser != null) {
PowerOfTwoBufferPool.DEFAULT.releasePruning();
}
}
}
Expand Down