You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@commons.apache.org by gg...@apache.org on 2020/07/25 03:02:43 UTC
[commons-pool] branch master updated: Refactored EvictionTimer
usage tracking to fix POOL-386 and handle abandoned pools. (#32)
This is an automated email from the ASF dual-hosted git repository.
ggregory pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/commons-pool.git
The following commit(s) were added to refs/heads/master by this push:
new 004dd29 Refactored EvictionTimer usage tracking to fix POOL-386 and handle abandoned pools. (#32)
004dd29 is described below
commit 004dd29d1aa9916ef7559c61a049b9cf6deb9a3d
Author: Phil Steitz <ps...@apache.org>
AuthorDate: Fri Jul 24 20:02:37 2020 -0700
Refactored EvictionTimer usage tracking to fix POOL-386 and handle abandoned pools. (#32)
* Refactored EvictionTimer usage tracking to fix POOL-386 and handle abandoned pools.
* Removed unecessary clear, make WeakRunner cancel itself when ref is null.
* Ensure no executor cancel can happen durning evictor restart.
* Add class javadoc.
---
.../commons/pool2/impl/BaseGenericObjectPool.java | 24 +++-
.../apache/commons/pool2/impl/EvictionTimer.java | 105 +++++++++++++++--
.../commons/pool2/impl/AtomicIntegerFactory.java | 124 +++++++++++++++++++++
.../pool2/impl/TestBaseGenericObjectPool.java | 29 +++++
.../commons/pool2/impl/TestEvictionTimer.java | 17 ++-
.../commons/pool2/impl/TestGenericObjectPool.java | 121 +++++---------------
6 files changed, 303 insertions(+), 117 deletions(-)
diff --git a/src/main/java/org/apache/commons/pool2/impl/BaseGenericObjectPool.java b/src/main/java/org/apache/commons/pool2/impl/BaseGenericObjectPool.java
index 25a5a2b..a907014 100644
--- a/src/main/java/org/apache/commons/pool2/impl/BaseGenericObjectPool.java
+++ b/src/main/java/org/apache/commons/pool2/impl/BaseGenericObjectPool.java
@@ -783,12 +783,23 @@ public abstract class BaseGenericObjectPool<T> extends BaseObject {
*/
final void startEvictor(final long delay) {
synchronized (evictionLock) {
- EvictionTimer.cancel(evictor, evictorShutdownTimeoutMillis, TimeUnit.MILLISECONDS);
- evictor = null;
- evictionIterator = null;
- if (delay > 0) {
- evictor = new Evictor();
- EvictionTimer.schedule(evictor, delay, delay);
+ if (evictor == null) { // Starting evictor for the first time or after a cancel
+ if (delay > 0) { // Starting new evictor
+ evictor = new Evictor();
+ EvictionTimer.schedule(evictor, delay, delay);
+ }
+ } else { // Stop or restart of existing evictor
+ if (delay > 0) { // Restart
+ synchronized (EvictionTimer.class) { // Ensure no cancel can happen between cancel / schedule calls
+ EvictionTimer.cancel(evictor, evictorShutdownTimeoutMillis, TimeUnit.MILLISECONDS, true);
+ evictor = null;
+ evictionIterator = null;
+ evictor = new Evictor();
+ EvictionTimer.schedule(evictor, delay, delay);
+ }
+ } else { // Stopping evictor
+ EvictionTimer.cancel(evictor, evictorShutdownTimeoutMillis, TimeUnit.MILLISECONDS, false);
+ }
}
}
}
@@ -1183,6 +1194,7 @@ public abstract class BaseGenericObjectPool<T> extends BaseObject {
void cancel() {
scheduledFuture.cancel(false);
}
+
}
/**
diff --git a/src/main/java/org/apache/commons/pool2/impl/EvictionTimer.java b/src/main/java/org/apache/commons/pool2/impl/EvictionTimer.java
index 60928ef..eb0c0ba 100644
--- a/src/main/java/org/apache/commons/pool2/impl/EvictionTimer.java
+++ b/src/main/java/org/apache/commons/pool2/impl/EvictionTimer.java
@@ -16,13 +16,18 @@
*/
package org.apache.commons.pool2.impl;
+import java.lang.ref.WeakReference;
import java.security.AccessController;
import java.security.PrivilegedAction;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map.Entry;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+
/**
* Provides a shared idle object eviction timer for all pools.
* <p>
@@ -46,6 +51,9 @@ class EvictionTimer {
/** Executor instance */
private static ScheduledThreadPoolExecutor executor; //@GuardedBy("EvictionTimer.class")
+ /** Keys are weak references to tasks, values are runners managed by executor. */
+ private static final HashMap<WeakReference<Runnable>, WeakRunner> taskMap = new HashMap<>(); // @GuardedBy("EvictionTimer.class")
+
/** Prevents instantiation */
private EvictionTimer() {
// Hide the default constructor
@@ -62,7 +70,6 @@ class EvictionTimer {
return builder.toString();
}
-
/**
* Adds the specified eviction task to the timer. Tasks that are added with
* a call to this method *must* call {@link
@@ -79,10 +86,14 @@ class EvictionTimer {
if (null == executor) {
executor = new ScheduledThreadPoolExecutor(1, new EvictorThreadFactory());
executor.setRemoveOnCancelPolicy(true);
+ executor.scheduleAtFixedRate(new Reaper(), delay, period, TimeUnit.MILLISECONDS);
}
+ final WeakReference<Runnable> ref = new WeakReference<>(task);
+ final WeakRunner runner = new WeakRunner(ref);
final ScheduledFuture<?> scheduledFuture =
- executor.scheduleWithFixedDelay(task, delay, period, TimeUnit.MILLISECONDS);
+ executor.scheduleWithFixedDelay(runner, delay, period, TimeUnit.MILLISECONDS);
task.setScheduledFuture(scheduledFuture);
+ taskMap.put(ref, runner);
}
/**
@@ -95,22 +106,48 @@ class EvictionTimer {
* @param unit The units for the specified timeout.
*/
static synchronized void cancel(
- final BaseGenericObjectPool<?>.Evictor evictor, final long timeout, final TimeUnit unit) {
+ final BaseGenericObjectPool<?>.Evictor evictor, final long timeout, final TimeUnit unit, final boolean restarting) {
if (evictor != null) {
evictor.cancel();
+ remove(evictor);
}
- if (executor != null && executor.getQueue().isEmpty()) {
- executor.shutdown();
- try {
- executor.awaitTermination(timeout, unit);
- } catch (final InterruptedException e) {
- // Swallow
- // Significant API changes would be required to propagate this
+ if (!restarting && executor != null) {
+ if (taskMap.isEmpty()) {
+ executor.shutdown();
+ try {
+ executor.awaitTermination(timeout, unit);
+ } catch (final InterruptedException e) {
+ // Swallow
+ // Significant API changes would be required to propagate this
+ }
+ executor.setCorePoolSize(0);
+ executor = null;
+ }
+ }
+ }
+
+ /**
+ * Removes evictor from the task set and executor.
+ * Only called when holding the class lock.
+ *
+ * @param evictor Eviction task to remove
+ */
+ private static void remove(final BaseGenericObjectPool<?>.Evictor evictor) {
+ for (Entry<WeakReference<Runnable>, WeakRunner> entry : taskMap.entrySet()) {
+ if (entry.getKey().get() == evictor) {
+ executor.remove(entry.getValue());
+ taskMap.remove(entry.getKey());
+ break;
}
- executor.setCorePoolSize(0);
- executor = null;
}
}
+
+ /**
+ * @return the number of eviction tasks under management.
+ */
+ static synchronized int getNumTasks() {
+ return taskMap.size();
+ }
/**
* Thread factory that creates a daemon thread, with the context class loader from this class.
@@ -129,4 +166,48 @@ class EvictionTimer {
return thread;
}
}
+
+ /**
+ * Task that removes references to abandoned tasks and shuts
+ * down the executor if there are no live tasks left.
+ */
+ private static class Reaper implements Runnable {
+ @Override
+ public void run() {
+ synchronized (EvictionTimer.class) {
+ for (Entry<WeakReference<Runnable>, WeakRunner> entry : taskMap.entrySet()) {
+ if (entry.getKey().get() == null) {
+ executor.remove(entry.getValue());
+ taskMap.remove(entry.getKey());
+ }
+ }
+ if (taskMap.isEmpty() && executor != null) {
+ executor.shutdown();
+ executor.setCorePoolSize(0);
+ executor = null;
+ }
+ }
+ }
+ }
+
+ /**
+ * Runnable that runs the referent of a weak reference. When the referent is no
+ * no longer reachable, run is no-op.
+ */
+ private static class WeakRunner implements Runnable {
+ private final WeakReference<Runnable> ref;
+ public WeakRunner(WeakReference<Runnable> ref) {
+ this.ref = ref;
+ }
+ @Override
+ public void run() {
+ final Runnable task = ref.get();
+ if (task != null) {
+ task.run();
+ } else {
+ executor.remove(this);
+ taskMap.remove(ref);
+ }
+ }
+ }
}
diff --git a/src/test/java/org/apache/commons/pool2/impl/AtomicIntegerFactory.java b/src/test/java/org/apache/commons/pool2/impl/AtomicIntegerFactory.java
new file mode 100644
index 0000000..803d1db
--- /dev/null
+++ b/src/test/java/org/apache/commons/pool2/impl/AtomicIntegerFactory.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.commons.pool2.impl;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.commons.pool2.BasePooledObjectFactory;
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+
+/**
+ * Factory that sources PooledObjects that wrap AtomicIntegers.
+ * {@link #create()} creates an AtomicInteger with value 0, activate increments
+ * the value of the wrapped AtomicInteger and passivate decrements it. Latency
+ * of factory methods is configurable.
+ *
+ */
+public class AtomicIntegerFactory
+ extends BasePooledObjectFactory<AtomicInteger> {
+
+ private long activateLatency = 0;
+ private long passivateLatency = 0;
+ private long createLatency = 0;
+ private long destroyLatency = 0;
+ private long validateLatency = 0;
+
+ @Override
+ public void activateObject(final PooledObject<AtomicInteger> p) {
+ p.getObject().incrementAndGet();
+ try {
+ Thread.sleep(activateLatency);
+ } catch (final InterruptedException ex) {}
+ }
+
+ @Override
+ public AtomicInteger create() {
+ try {
+ Thread.sleep(createLatency);
+ } catch (final InterruptedException ex) {}
+ return new AtomicInteger(0);
+ }
+
+ @Override
+ public void destroyObject(final PooledObject<AtomicInteger> p) {
+ try {
+ Thread.sleep(destroyLatency);
+ } catch (final InterruptedException ex) {}
+ }
+
+ @Override
+ public void passivateObject(final PooledObject<AtomicInteger> p) {
+ p.getObject().decrementAndGet();
+ try {
+ Thread.sleep(passivateLatency);
+ } catch (final InterruptedException ex) {}
+ }
+
+ /**
+ * @param activateLatency the activateLatency to set
+ */
+ public void setActivateLatency(final long activateLatency) {
+ this.activateLatency = activateLatency;
+ }
+
+ /**
+ * @param createLatency the createLatency to set
+ */
+ public void setCreateLatency(final long createLatency) {
+ this.createLatency = createLatency;
+ }
+
+
+ /**
+ * @param destroyLatency the destroyLatency to set
+ */
+ public void setDestroyLatency(final long destroyLatency) {
+ this.destroyLatency = destroyLatency;
+ }
+
+
+ /**
+ * @param passivateLatency the passivateLatency to set
+ */
+ public void setPassivateLatency(final long passivateLatency) {
+ this.passivateLatency = passivateLatency;
+ }
+
+
+ /**
+ * @param validateLatency the validateLatency to set
+ */
+ public void setValidateLatency(final long validateLatency) {
+ this.validateLatency = validateLatency;
+ }
+
+
+ @Override
+ public boolean validateObject(final PooledObject<AtomicInteger> instance) {
+ try {
+ Thread.sleep(validateLatency);
+ } catch (final InterruptedException ex) {}
+ return instance.getObject().intValue() == 1;
+ }
+
+
+ @Override
+ public PooledObject<AtomicInteger> wrap(final AtomicInteger integer) {
+ return new DefaultPooledObject<>(integer);
+ }
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/commons/pool2/impl/TestBaseGenericObjectPool.java b/src/test/java/org/apache/commons/pool2/impl/TestBaseGenericObjectPool.java
index bb03c23..adb5695 100644
--- a/src/test/java/org/apache/commons/pool2/impl/TestBaseGenericObjectPool.java
+++ b/src/test/java/org/apache/commons/pool2/impl/TestBaseGenericObjectPool.java
@@ -17,6 +17,8 @@
package org.apache.commons.pool2.impl;
+import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.commons.pool2.impl.TestGenericObjectPool.SimpleFactory;
import org.junit.After;
import org.junit.Assert;
@@ -74,4 +76,31 @@ public class TestBaseGenericObjectPool {
}
Assert.assertEquals(49, pool.getMeanActiveTimeMillis(), Double.MIN_VALUE);
}
+
+ @Test
+ public void testEvictionTimerMultiplePools() throws InterruptedException {
+ final AtomicIntegerFactory factory = new AtomicIntegerFactory();
+ factory.setValidateLatency(50);
+ final GenericObjectPool<AtomicInteger> evictingPool = new GenericObjectPool<>(factory);
+ evictingPool.setTimeBetweenEvictionRunsMillis(100);
+ evictingPool.setNumTestsPerEvictionRun(5);
+ evictingPool.setTestWhileIdle(true);
+ evictingPool.setMinEvictableIdleTimeMillis(50);
+ for (int i = 0; i < 10; i++) {
+ try {
+ evictingPool.addObject();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ for (int i = 0; i < 1000; i++) {
+ GenericObjectPool<AtomicInteger> nonEvictingPool = new GenericObjectPool<>(factory);
+ nonEvictingPool.close();
+ }
+
+ Thread.sleep(1000);
+ Assert.assertEquals(0, evictingPool.getNumIdle());
+ evictingPool.close();
+ }
}
diff --git a/src/test/java/org/apache/commons/pool2/impl/TestEvictionTimer.java b/src/test/java/org/apache/commons/pool2/impl/TestEvictionTimer.java
index 97d8bdd..33b38c2 100644
--- a/src/test/java/org/apache/commons/pool2/impl/TestEvictionTimer.java
+++ b/src/test/java/org/apache/commons/pool2/impl/TestEvictionTimer.java
@@ -20,7 +20,10 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+
+import java.lang.ref.WeakReference;
import java.lang.reflect.Field;
+import java.util.HashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -65,7 +68,8 @@ public class TestEvictionTimer {
final Field evictorExecutorField = EvictionTimer.class.getDeclaredField("executor");
evictorExecutorField.setAccessible(true);
final ThreadPoolExecutor evictionExecutor = (ThreadPoolExecutor) evictorExecutorField.get(null);
- assertEquals(1, evictionExecutor.getQueue().size());
+ assertEquals(2, evictionExecutor.getQueue().size()); // Reaper plus one eviction task
+ assertEquals(1, EvictionTimer.getNumTasks());
// Start evictor #2
final BaseGenericObjectPool<String>.Evictor evictor2 = pool.new Evictor();
@@ -76,11 +80,12 @@ public class TestEvictionTimer {
sf = (ScheduledFuture<?>) evictorTaskFutureField.get(evictor2);
assertFalse(sf.isCancelled());
// 2- and, the eviction action is added to executor thread pool
- assertEquals(2, evictionExecutor.getQueue().size());
+ assertEquals(3, evictionExecutor.getQueue().size()); // Reaper plus 2 eviction tasks
+ assertEquals(2, EvictionTimer.getNumTasks());
// Stop evictor #1
EvictionTimer.cancel(evictor1, BaseObjectPoolConfig.DEFAULT_EVICTOR_SHUTDOWN_TIMEOUT_MILLIS,
- TimeUnit.MILLISECONDS);
+ TimeUnit.MILLISECONDS, false);
// Assert that eviction objects are correctly cleaned
// 1 - the evictor timer task is cancelled
@@ -88,11 +93,12 @@ public class TestEvictionTimer {
assertTrue(sf.isCancelled());
// 2- and, the eviction action is removed from executor thread pool
final ThreadPoolExecutor evictionExecutorOnStop = (ThreadPoolExecutor) evictorExecutorField.get(null);
- assertEquals(1, evictionExecutorOnStop.getQueue().size());
+ assertEquals(2, evictionExecutorOnStop.getQueue().size());
+ assertEquals(1, EvictionTimer.getNumTasks());
// Stop evictor #2
EvictionTimer.cancel(evictor2, BaseObjectPoolConfig.DEFAULT_EVICTOR_SHUTDOWN_TIMEOUT_MILLIS,
- TimeUnit.MILLISECONDS);
+ TimeUnit.MILLISECONDS, false);
// Assert that eviction objects are correctly cleaned
// 1 - the evictor timer task is cancelled
@@ -100,6 +106,7 @@ public class TestEvictionTimer {
assertTrue(sf.isCancelled());
// 2- and, the eviction thread pool executor is freed
assertNull(evictorExecutorField.get(null));
+ assertEquals(0, EvictionTimer.getNumTasks());
}
}
}
\ No newline at end of file
diff --git a/src/test/java/org/apache/commons/pool2/impl/TestGenericObjectPool.java b/src/test/java/org/apache/commons/pool2/impl/TestGenericObjectPool.java
index 67579fd..bbdbc6b 100644
--- a/src/test/java/org/apache/commons/pool2/impl/TestGenericObjectPool.java
+++ b/src/test/java/org/apache/commons/pool2/impl/TestGenericObjectPool.java
@@ -25,6 +25,8 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.lang.management.ManagementFactory;
+import java.lang.ref.WeakReference;
+import java.lang.reflect.Field;
import java.nio.charset.UnsupportedCharsetException;
import java.util.ArrayList;
import java.util.HashSet;
@@ -60,100 +62,6 @@ import org.junit.Test;
*/
public class TestGenericObjectPool extends TestBaseObjectPool {
- protected static class AtomicIntegerFactory
- extends BasePooledObjectFactory<AtomicInteger> {
-
- private long activateLatency = 0;
- private long passivateLatency = 0;
- private long createLatency = 0;
- private long destroyLatency = 0;
- private long validateLatency = 0;
-
- @Override
- public void activateObject(final PooledObject<AtomicInteger> p) {
- p.getObject().incrementAndGet();
- try {
- Thread.sleep(activateLatency);
- } catch (final InterruptedException ex) {}
- }
-
- @Override
- public AtomicInteger create() {
- try {
- Thread.sleep(createLatency);
- } catch (final InterruptedException ex) {}
- return new AtomicInteger(0);
- }
-
- @Override
- public void destroyObject(final PooledObject<AtomicInteger> p) {
- try {
- Thread.sleep(destroyLatency);
- } catch (final InterruptedException ex) {}
- }
-
- @Override
- public void passivateObject(final PooledObject<AtomicInteger> p) {
- p.getObject().decrementAndGet();
- try {
- Thread.sleep(passivateLatency);
- } catch (final InterruptedException ex) {}
- }
-
- /**
- * @param activateLatency the activateLatency to set
- */
- public void setActivateLatency(final long activateLatency) {
- this.activateLatency = activateLatency;
- }
-
- /**
- * @param createLatency the createLatency to set
- */
- public void setCreateLatency(final long createLatency) {
- this.createLatency = createLatency;
- }
-
-
- /**
- * @param destroyLatency the destroyLatency to set
- */
- public void setDestroyLatency(final long destroyLatency) {
- this.destroyLatency = destroyLatency;
- }
-
-
- /**
- * @param passivateLatency the passivateLatency to set
- */
- public void setPassivateLatency(final long passivateLatency) {
- this.passivateLatency = passivateLatency;
- }
-
-
- /**
- * @param validateLatency the validateLatency to set
- */
- public void setValidateLatency(final long validateLatency) {
- this.validateLatency = validateLatency;
- }
-
-
- @Override
- public boolean validateObject(final PooledObject<AtomicInteger> instance) {
- try {
- Thread.sleep(validateLatency);
- } catch (final InterruptedException ex) {}
- return instance.getObject().intValue() == 1;
- }
-
-
- @Override
- public PooledObject<AtomicInteger> wrap(final AtomicInteger integer) {
- return new DefaultPooledObject<>(integer);
- }
-}
-
private class ConcurrentBorrowAndEvictThread extends Thread {
private final boolean borrow;
public String obj;
@@ -1022,6 +930,11 @@ public class TestGenericObjectPool extends TestBaseObjectPool {
mbs.unregisterMBean(name);
}
Assert.assertEquals(msg.toString(), 0, registeredPoolCount);
+
+ // Make sure that EvictionTimer executor is shut down
+ final Field evictorExecutorField = EvictionTimer.class.getDeclaredField("executor");
+ evictorExecutorField.setAccessible(true);
+ assertNull(evictorExecutorField.get(null));
}
@Test(timeout = 60000)
@@ -2996,4 +2909,24 @@ public class TestGenericObjectPool extends TestBaseObjectPool {
genericObjectPool.close();
}
+ /**
+ * Check that a pool that starts an evictor, but is never closed does not
+ * leave EvictionTimer executor running. Confirmation check is in teardown.
+ */
+ @Test
+ public void testAbandonedPool() throws Exception {
+ GenericObjectPoolConfig config = new GenericObjectPoolConfig();
+ config.setJmxEnabled(false);
+ GenericObjectPool<String> abandoned = new GenericObjectPool<>(simpleFactory, config);
+ abandoned.setTimeBetweenEvictionRunsMillis(100); // Starts evictor
+
+ // This is ugly, but forces gc to hit the pool
+ WeakReference<GenericObjectPool> ref = new WeakReference<>(abandoned);
+ abandoned = null;
+ while (ref.get() != null) {
+ System.gc();
+ Thread.sleep(100);
+ }
+ }
+
}