You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tomee.apache.org by rm...@apache.org on 2015/06/15 21:04:08 UTC
tomee git commit: TOMEE-1604 control eviction thread number by
stateless container
Repository: tomee
Updated Branches:
refs/heads/master 3ac26c169 -> 2af5a3acb
TOMEE-1604 control eviction thread number by stateless container
Project: http://git-wip-us.apache.org/repos/asf/tomee/repo
Commit: http://git-wip-us.apache.org/repos/asf/tomee/commit/2af5a3ac
Tree: http://git-wip-us.apache.org/repos/asf/tomee/tree/2af5a3ac
Diff: http://git-wip-us.apache.org/repos/asf/tomee/diff/2af5a3ac
Branch: refs/heads/master
Commit: 2af5a3acbb969b80b07a8b63235356b19bc748c6
Parents: 3ac26c1
Author: Romain Manni-Bucau <rm...@apache.org>
Authored: Mon Jun 15 21:03:49 2015 +0200
Committer: Romain Manni-Bucau <rm...@apache.org>
Committed: Mon Jun 15 21:03:49 2015 +0200
----------------------------------------------------------------------
.../core/stateless/StatelessContainer.java | 14 ++++--
.../stateless/StatelessContainerFactory.java | 14 +++++-
.../stateless/StatelessInstanceManager.java | 9 +++-
.../main/java/org/apache/openejb/util/Pool.java | 47 ++++++++++++++----
.../META-INF/org.apache.openejb/service-jar.xml | 6 +++
.../SchedulerThreadStateless1ByBeanTest.java | 51 ++++++++++++++++++++
.../stateless/SchedulerThreadStatelessTest.java | 51 ++++++++++++++++++++
7 files changed, 177 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tomee/blob/2af5a3ac/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainer.java
----------------------------------------------------------------------
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainer.java b/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainer.java
index 347203f..62bee31 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainer.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainer.java
@@ -35,11 +35,11 @@ import org.apache.openejb.core.webservices.AddressingSupport;
import org.apache.openejb.core.webservices.NoAddressingSupport;
import org.apache.openejb.monitoring.StatsInterceptor;
import org.apache.openejb.spi.SecurityService;
+import org.apache.openejb.util.DaemonThreadFactory;
import org.apache.openejb.util.Duration;
import org.apache.openejb.util.Pool;
import org.apache.xbean.finder.ClassFinder;
-import javax.interceptor.AroundInvoke;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
@@ -48,6 +48,8 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import javax.interceptor.AroundInvoke;
import static org.apache.openejb.core.transaction.EjbTransactionUtil.afterInvoke;
import static org.apache.openejb.core.transaction.EjbTransactionUtil.createTransactionPolicy;
@@ -70,10 +72,16 @@ public class StatelessContainer implements org.apache.openejb.RpcContainer {
final Duration accessTimeout,
final Duration closeTimeout,
final Pool.Builder poolBuilder,
- final int callbackThreads) {
+ final int callbackThreads,
+ final boolean useOneSchedulerThreadByBean,
+ final int evictionThreads) {
this.containerID = id;
this.securityService = securityService;
- this.instanceManager = new StatelessInstanceManager(securityService, accessTimeout, closeTimeout, poolBuilder, callbackThreads);
+ this.instanceManager = new StatelessInstanceManager(
+ securityService, accessTimeout, closeTimeout, poolBuilder, callbackThreads,
+ useOneSchedulerThreadByBean ?
+ null :
+ Executors.newScheduledThreadPool(Math.max(evictionThreads, 1), new DaemonThreadFactory(id)));
}
@Override
http://git-wip-us.apache.org/repos/asf/tomee/blob/2af5a3ac/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainerFactory.java
----------------------------------------------------------------------
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainerFactory.java b/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainerFactory.java
index 0eb80f8..413a5a5 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainerFactory.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessContainerFactory.java
@@ -34,6 +34,16 @@ public class StatelessContainerFactory {
private Duration accessTimeout;
private int callbackThreads = 5;
private Duration closeTimeout;
+ private boolean useOneSchedulerThreadByBean = false;
+ private int evictionThreads = 1;
+
+ public void setEvictionThreads(final int evictionThreads) {
+ this.evictionThreads = evictionThreads;
+ }
+
+ public void setUseOneSchedulerThreadByBean(final boolean useOneSchedulerThreadByBean) {
+ this.useOneSchedulerThreadByBean = useOneSchedulerThreadByBean;
+ }
public void setCallbackThreads(final int callbackThreads) {
this.callbackThreads = callbackThreads;
@@ -125,6 +135,8 @@ public class StatelessContainerFactory {
}
public StatelessContainer create() {
- return new StatelessContainer(id, securityService, accessTimeout, closeTimeout, pool, callbackThreads);
+ return new StatelessContainer(
+ id, securityService, accessTimeout, closeTimeout, pool, callbackThreads,
+ useOneSchedulerThreadByBean, evictionThreads);
}
}
http://git-wip-us.apache.org/repos/asf/tomee/blob/2af5a3ac/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java
----------------------------------------------------------------------
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java b/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java
index 6deebdd..82cb094 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java
@@ -64,6 +64,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -88,12 +89,17 @@ public class StatelessInstanceManager {
private final SecurityService securityService;
private final Pool.Builder poolBuilder;
private final ThreadPoolExecutor executor;
+ private final ScheduledExecutorService scheduledExecutor;
- public StatelessInstanceManager(final SecurityService securityService, final Duration accessTimeout, final Duration closeTimeout, final Pool.Builder poolBuilder, final int callbackThreads) {
+ public StatelessInstanceManager(final SecurityService securityService,
+ final Duration accessTimeout, final Duration closeTimeout,
+ final Pool.Builder poolBuilder, final int callbackThreads,
+ final ScheduledExecutorService ses) {
this.securityService = securityService;
this.accessTimeout = accessTimeout;
this.closeTimeout = closeTimeout;
this.poolBuilder = poolBuilder;
+ this.scheduledExecutor = ses;
if (accessTimeout.getUnit() == null) {
accessTimeout.setUnit(TimeUnit.MILLISECONDS);
@@ -340,6 +346,7 @@ public class StatelessInstanceManager {
final StatelessSupplier supplier = new StatelessSupplier(beanContext);
builder.setSupplier(supplier);
builder.setExecutor(executor);
+ builder.setScheduledExecutor(scheduledExecutor);
final Data data = new Data(builder.build(), accessTimeout, closeTimeout);
beanContext.setContainerData(data);
http://git-wip-us.apache.org/repos/asf/tomee/blob/2af5a3ac/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java
----------------------------------------------------------------------
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java b/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java
index fb6817b..c8ed4d3 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java
@@ -31,6 +31,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
@@ -43,6 +44,7 @@ import java.util.logging.Level;
import java.util.logging.Logger;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
/**
* Any successful pop() call requires a corresponding push() or discard() call.
@@ -60,7 +62,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
@SuppressWarnings("StatementWithEmptyBody")
public class Pool<T> {
- private final LinkedList<Entry> pool = new LinkedList<Entry>();
+ private final LinkedList<Entry> pool = new LinkedList<>();
private final Semaphore instances;
private final Semaphore available;
private final Semaphore minimum;
@@ -73,7 +75,8 @@ public class Pool<T> {
private final AtomicInteger poolVersion = new AtomicInteger();
private final Supplier<T> supplier;
- private final AtomicReference<ScheduledExecutorService> scheduler = new AtomicReference<ScheduledExecutorService>();
+ private final AtomicReference<ScheduledExecutorService> scheduler = new AtomicReference<>();
+ private final AtomicReference<ScheduledFuture<?>> future = new AtomicReference<>();
private final Sweeper sweeper;
private final CountingLatch out = new CountingLatch();
@@ -127,21 +130,36 @@ public class Pool<T> {
}
public Pool start() {
- final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1, new SchedulerThreadFactory());
-
- if (this.scheduler.compareAndSet(null, scheduledExecutorService)) {
- this.scheduler.get().scheduleAtFixedRate(sweeper, 0, this.sweepInterval, MILLISECONDS);
+ ScheduledExecutorService scheduledExecutorService = this.scheduler.get();
+ boolean createdSES = scheduledExecutorService == null;
+ if (scheduledExecutorService == null) {
+ scheduledExecutorService = Executors.newScheduledThreadPool(1, new SchedulerThreadFactory());
+ if (!this.scheduler.compareAndSet(null, scheduledExecutorService)) {
+ scheduledExecutorService.shutdownNow();
+ scheduledExecutorService = this.scheduler.get();
+ createdSES = false;
+ }
}
+ if (this.future.compareAndSet(null, scheduledExecutorService.scheduleAtFixedRate(sweeper, 0, this.sweepInterval, MILLISECONDS)) && !createdSES) {
+ // we don't want to shutdown it, we'll just stop the task
+ this.scheduler.set(null);
+ }
+ System.out.println(sweepInterval + " " + createdSES);
return this;
}
public void stop() {
+ final ScheduledFuture<?> future = this.future.get();
+ if (future != null && !future.isDone() && !future.isCancelled() && !future.cancel(false)) {
+ Logger.getLogger(Pool.class.getName()).log(Level.WARNING, "Pool scheduler task termination timeout expired");
+ }
+
final ScheduledExecutorService scheduler = this.scheduler.get();
if (scheduler != null && this.scheduler.compareAndSet(scheduler, null)) {
scheduler.shutdown();
try {
- if (!scheduler.awaitTermination(10000, MILLISECONDS)) {
- Logger.getLogger(this.getClass().getName()).log(Level.WARNING, "Pool scheduler termination timeout expired");
+ if (!scheduler.awaitTermination(10, SECONDS)) { // should last something like 0s max since we killed the task
+ Logger.getLogger(Pool.class.getName()).log(Level.WARNING, "Pool scheduler termination timeout expired");
}
} catch (final InterruptedException e) {
//Ignore
@@ -150,7 +168,7 @@ public class Pool<T> {
}
public boolean running() {
- return this.scheduler.get() != null;
+ return this.future.get() != null;
}
private Executor createExecutor() {
@@ -1093,6 +1111,7 @@ public class Pool<T> {
private Duration interval = new Duration(5 * 60, TimeUnit.SECONDS);
private Supplier<T> supplier;
private Executor executor;
+ private ScheduledExecutorService scheduledExecutorService;
private boolean replaceAged;
private boolean replaceFlushed;
private boolean garbageCollection = true;
@@ -1204,9 +1223,17 @@ public class Pool<T> {
this.executor = executor;
}
+ public void setScheduledExecutor(final ScheduledExecutorService scheduledExecutorService) {
+ this.scheduledExecutorService = scheduledExecutorService;
+ }
+
public Pool<T> build() {
//noinspection unchecked
- return new Pool(max, min, strict, maxAge.getTime(MILLISECONDS), idleTimeout.getTime(MILLISECONDS), interval.getTime(MILLISECONDS), executor, supplier, replaceAged, maxAgeOffset, this.garbageCollection, replaceFlushed);
+ final Pool pool = new Pool(max, min, strict, maxAge.getTime(MILLISECONDS), idleTimeout.getTime(MILLISECONDS), interval.getTime(MILLISECONDS), executor, supplier, replaceAged, maxAgeOffset, this.garbageCollection, replaceFlushed);
+ if (scheduledExecutorService != null) {
+ pool.scheduler.set(scheduledExecutorService);
+ }
+ return pool;
}
}
http://git-wip-us.apache.org/repos/asf/tomee/blob/2af5a3ac/container/openejb-core/src/main/resources/META-INF/org.apache.openejb/service-jar.xml
----------------------------------------------------------------------
diff --git a/container/openejb-core/src/main/resources/META-INF/org.apache.openejb/service-jar.xml b/container/openejb-core/src/main/resources/META-INF/org.apache.openejb/service-jar.xml
index fd8c3ec..8b324cd 100644
--- a/container/openejb-core/src/main/resources/META-INF/org.apache.openejb/service-jar.xml
+++ b/container/openejb-core/src/main/resources/META-INF/org.apache.openejb/service-jar.xml
@@ -325,6 +325,12 @@
CloseTimeout = 5 minutes
+ # back to previous behavior (TomEE 1.x) where 1 scheduler thread was used for stateless eviction
+ # by bean (ie for 500 stateless beans you get 500 eviction threads)
+ UseOneSchedulerThreadByBean = false
+
+ # number of threads to associate to eviction threads (1 is not bad for most applications)
+ EvictionThreads = 1
</ServiceProvider>
http://git-wip-us.apache.org/repos/asf/tomee/blob/2af5a3ac/container/openejb-core/src/test/java/org/apache/openejb/core/stateless/SchedulerThreadStateless1ByBeanTest.java
----------------------------------------------------------------------
diff --git a/container/openejb-core/src/test/java/org/apache/openejb/core/stateless/SchedulerThreadStateless1ByBeanTest.java b/container/openejb-core/src/test/java/org/apache/openejb/core/stateless/SchedulerThreadStateless1ByBeanTest.java
new file mode 100644
index 0000000..d20b13f
--- /dev/null
+++ b/container/openejb-core/src/test/java/org/apache/openejb/core/stateless/SchedulerThreadStateless1ByBeanTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.openejb.core.stateless;
+
+import org.apache.openejb.junit.ApplicationComposer;
+import org.apache.openejb.testing.Classes;
+import org.apache.openejb.testing.ContainerProperties;
+import org.apache.openejb.testing.SimpleLog;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import javax.ejb.Stateless;
+
+import static org.junit.Assert.assertEquals;
+
+@SimpleLog
+@Classes(innerClassesAsBean = true)
+@RunWith(ApplicationComposer.class)
+@ContainerProperties(@ContainerProperties.Property(name = "Default Stateless Container.useOneSchedulerThreadByBean", value = "true"))
+public class SchedulerThreadStateless1ByBeanTest {
+ @Stateless public static class S1 {}
+ @Stateless public static class S2 {}
+
+ @Test
+ public void run() {
+ final int count = Thread.activeCount();
+ final Thread[] all = new Thread[count];
+ Thread.enumerate(all);
+ int schedulers = 0;
+ for (final Thread t : all) {
+ if (t.getName().startsWith("org.apache.openejb.pool.scheduler.")) {
+ schedulers++;
+ }
+ }
+ assertEquals(2, schedulers);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tomee/blob/2af5a3ac/container/openejb-core/src/test/java/org/apache/openejb/core/stateless/SchedulerThreadStatelessTest.java
----------------------------------------------------------------------
diff --git a/container/openejb-core/src/test/java/org/apache/openejb/core/stateless/SchedulerThreadStatelessTest.java b/container/openejb-core/src/test/java/org/apache/openejb/core/stateless/SchedulerThreadStatelessTest.java
new file mode 100644
index 0000000..8517a5d
--- /dev/null
+++ b/container/openejb-core/src/test/java/org/apache/openejb/core/stateless/SchedulerThreadStatelessTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.openejb.core.stateless;
+
+import org.apache.openejb.junit.ApplicationComposer;
+import org.apache.openejb.testing.Classes;
+import org.apache.openejb.testing.ContainerProperties;
+import org.apache.openejb.testing.SimpleLog;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import javax.ejb.Stateless;
+
+import static org.junit.Assert.assertEquals;
+
+@SimpleLog
+@Classes(innerClassesAsBean = true)
+@RunWith(ApplicationComposer.class)
+@ContainerProperties(@ContainerProperties.Property(name = "Default Stateless Container.useOneSchedulerThreadByBean", value = "false"))
+public class SchedulerThreadStatelessTest {
+ @Stateless public static class S1 {}
+ @Stateless public static class S2 {}
+
+ @Test
+ public void run() {
+ final int count = Thread.activeCount();
+ final Thread[] all = new Thread[count];
+ Thread.enumerate(all);
+ int schedulers = 0;
+ for (final Thread t : all) {
+ if (t.getName().startsWith("Default Stateless Container")) {
+ schedulers++;
+ }
+ }
+ assertEquals(1, schedulers);
+ }
+}