Author: andygumbrecht
Date: Thu Jan 17 14:42:30 2013
New Revision: 1434694
URL: http://svn.apache.org/viewvc?rev=1434694&view=rev
Log:
Fix thread pool misuse / misunderstandings.
A bounded thread pool starts with 'threadCore' core threads and should grow to
the limit defined by 'threads' during extra load.
If a pool thread is idle for more than the define timeout it will be discarded,
shrinking the pool until the core size is reached.
The pool can accept runnables up to the number defined by 'queue' +
'threadCore' before the pool will actually grow (this is where the
misunderstanding often occurs), so the 'queue' should ideally be bounded to 1
less than 'threadsCore', but certainly less than 'threads', else the pool will
never actually grow!
If the 'queue' and 'threads' are full then an attempt should be made in the
rejection handler to add the runnable to the queue for a defined period.
Failure to add to the queue in this period (ie. a pool thread has not become
available) should either result in a logged rejection or make a final attempt
to run the runnable in the current thread (the calling thread).
Ehcache > 2.6.3
Transaction logging.
Modified:
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/async/AsynchronousPool.java
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/CoreUserTransaction.java
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/resource/jdbc/pool/DefaultDataSourceCreator.java
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/ExecutorBuilder.java
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java
tomee/trunk/openejb/pom.xml
tomee/trunk/openejb/server/openejb-client/src/main/java/org/apache/openejb/client/EJBObjectHandler.java
tomee/trunk/openejb/server/openejb-ejbd/src/test/java/org/apache/openejb/server/ejbd/KeepAilveTest.java
tomee/trunk/openejb/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java
tomee/trunk/openejb/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java
Modified:
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/async/AsynchronousPool.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/async/AsynchronousPool.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/async/AsynchronousPool.java
(original)
+++
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/async/AsynchronousPool.java
Thu Jan 17 14:42:30 2013
@@ -24,16 +24,7 @@ import org.apache.openejb.util.ExecutorB
import javax.ejb.EJBException;
import javax.ejb.NoSuchEJBException;
import java.rmi.NoSuchObjectException;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -44,22 +35,22 @@ public class AsynchronousPool {
private final BlockingQueue<Runnable> blockingQueue;
private final ExecutorService executor;
- public AsynchronousPool(ThreadPoolExecutor threadPoolExecutor) {
+ public AsynchronousPool(final ThreadPoolExecutor threadPoolExecutor) {
this.blockingQueue = threadPoolExecutor.getQueue();
this.executor = threadPoolExecutor;
}
- public static AsynchronousPool create(AppContext appContext) {
+ public static AsynchronousPool create(final AppContext appContext) {
final ExecutorBuilder builder = new ExecutorBuilder()
.prefix("AsynchronousPool")
- .size(10)
+ .size(3)
.threadFactory(new DaemonThreadFactory("@Asynchronous",
appContext.getId()));
return new AsynchronousPool(builder.build(appContext.getOptions()));
}
- public Object invoke(Callable<Object> callable, boolean isVoid) throws
Throwable {
+ public Object invoke(final Callable<Object> callable, final boolean
isVoid) throws Throwable {
final AtomicBoolean asynchronousCancelled = new AtomicBoolean(false);
try {
@@ -80,7 +71,7 @@ public class AsynchronousPool {
private final AtomicBoolean asynchronousCancelled;
- private AsynchronousCall(Callable<Object> callable, AtomicBoolean
asynchronousCancelled) {
+ private AsynchronousCall(final Callable<Object> callable, final
AtomicBoolean asynchronousCancelled) {
this.callable = callable;
this.asynchronousCancelled = asynchronousCancelled;
}
@@ -116,13 +107,14 @@ public class AsynchronousPool {
private volatile boolean canceled;
- public FutureAdapter(Future<T> target, AtomicBoolean
asynchronousCancelled) {
+ public FutureAdapter(final Future<T> target, final AtomicBoolean
asynchronousCancelled) {
this.target = target;
this.asynchronousCancelled = asynchronousCancelled;
}
+ @SuppressWarnings("RedundantCast")
@Override
- public boolean cancel(boolean mayInterruptIfRunning) {
+ public boolean cancel(final boolean mayInterruptIfRunning) {
/*In EJB 3.1 spec 3.4.8.1.1
*a. If a client calls cancel on its Future object, the container
will attempt to cancel the associated asynchronous invocation only if that
invocation has not already been dispatched.
* There is no guarantee that an asynchronous invocation can be
cancelled, regardless of how quickly cancel is called after the client receives
its Future object.
@@ -171,7 +163,7 @@ public class AsynchronousPool {
}
@Override
- public T get(long timeout, TimeUnit unit) throws InterruptedException,
ExecutionException, TimeoutException {
+ public T get(final long timeout, final TimeUnit unit) throws
InterruptedException, ExecutionException, TimeoutException {
if (canceled) {
throw new CancellationException();
}
@@ -192,7 +184,7 @@ public class AsynchronousPool {
//unwarp the exception to find the root cause
while (e.getCause() != null) {
- e = (Throwable) e.getCause();
+ e = e.getCause();
}
/*
@@ -205,7 +197,7 @@ public class AsynchronousPool {
e = new NoSuchEJBException(e.getMessage(), (Exception) e);
}
- boolean isExceptionUnchecked = (e instanceof Error) || (e
instanceof RuntimeException);
+ final boolean isExceptionUnchecked = (e instanceof Error) || (e
instanceof RuntimeException);
// throw checked excpetion and EJBException directly.
if (!isExceptionUnchecked || e instanceof EJBException) {
Modified:
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/CoreUserTransaction.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/CoreUserTransaction.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/CoreUserTransaction.java
(original)
+++
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/CoreUserTransaction.java
Thu Jan 17 14:42:30 2013
@@ -19,24 +19,17 @@ package org.apache.openejb.core;
import org.apache.openejb.util.LogCategory;
import org.apache.openejb.util.Logger;
-import javax.transaction.HeuristicMixedException;
-import javax.transaction.HeuristicRollbackException;
-import javax.transaction.NotSupportedException;
-import javax.transaction.RollbackException;
-import javax.transaction.Status;
-import javax.transaction.SystemException;
-import javax.transaction.TransactionManager;
+import javax.transaction.*;
/**
* @org.apache.xbean.XBean element="userTransaction"
*/
public class CoreUserTransaction implements javax.transaction.UserTransaction,
java.io.Serializable {
private static final long serialVersionUID = 9203248912222645965L;
- private static final Logger transactionLogger =
Logger.getInstance(LogCategory.TRANSACTION,
"org.apache.openejb.util.resources");
-
+ private static transient final Logger transactionLogger =
Logger.getInstance(LogCategory.TRANSACTION,
"org.apache.openejb.util.resources");
private transient TransactionManager transactionManager;
- public CoreUserTransaction(TransactionManager transactionManager) {
+ public CoreUserTransaction(final TransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
@@ -47,48 +40,54 @@ public class CoreUserTransaction impleme
return transactionManager;
}
+ @Override
public void begin() throws NotSupportedException, SystemException {
transactionManager().begin();
- if (transactionLogger.isInfoEnabled()) {
- transactionLogger.info("Started user transaction " +
transactionManager().getTransaction());
+ if (transactionLogger.isDebugEnabled()) {
+ transactionLogger.debug("Started user transaction " +
transactionManager().getTransaction());
}
}
+ @Override
public void commit() throws RollbackException, HeuristicMixedException,
HeuristicRollbackException,
SecurityException, IllegalStateException, SystemException {
- if (transactionLogger.isInfoEnabled()) {
- transactionLogger.info("Committing user transaction " +
transactionManager().getTransaction());
+ if (transactionLogger.isDebugEnabled()) {
+ transactionLogger.debug("Committing user transaction " +
transactionManager().getTransaction());
}
transactionManager().commit();
}
- public void rollback() throws IllegalStateException, SecurityException,
SystemException {
- if (transactionLogger.isInfoEnabled()) {
- transactionLogger.info("Rolling back user transaction " +
transactionManager().getTransaction());
+ @Override
+ public int getStatus() throws SystemException {
+ final int status = transactionManager().getStatus();
+ if (transactionLogger.isDebugEnabled()) {
+ transactionLogger.debug("User transaction " +
transactionManager().getTransaction() + " has status " + getStatus(status));
}
- transactionManager().rollback();
+ return status;
}
- public int getStatus() throws SystemException {
- int status = transactionManager().getStatus();
- if (transactionLogger.isInfoEnabled()) {
- transactionLogger.info("User transaction " +
transactionManager().getTransaction() + " has status " + getStatus(status));
+ @Override
+ public void rollback() throws IllegalStateException, SecurityException,
SystemException {
+ if (transactionLogger.isDebugEnabled()) {
+ transactionLogger.debug("Rolling back user transaction " +
transactionManager().getTransaction());
}
- return status;
+ transactionManager().rollback();
}
+ @Override
public void setRollbackOnly() throws javax.transaction.SystemException {
- if (transactionLogger.isInfoEnabled()) {
- transactionLogger.info("Marking user transaction for rollback: " +
transactionManager().getTransaction());
+ if (transactionLogger.isDebugEnabled()) {
+ transactionLogger.debug("Marking user transaction for rollback: "
+ transactionManager().getTransaction());
}
transactionManager().setRollbackOnly();
}
- public void setTransactionTimeout(int seconds) throws SystemException {
+ @Override
+ public void setTransactionTimeout(final int seconds) throws
SystemException {
transactionManager().setTransactionTimeout(seconds);
}
- private static String getStatus(int status) {
+ private static String getStatus(final int status) {
final StringBuilder buffer = new StringBuilder(100);
switch (status) {
case Status.STATUS_ACTIVE:
Modified:
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java
(original)
+++
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/core/stateless/StatelessInstanceManager.java
Thu Jan 17 14:42:30 2013
@@ -33,12 +33,7 @@ import org.apache.openejb.monitoring.Man
import org.apache.openejb.monitoring.ObjectNameBuilder;
import org.apache.openejb.monitoring.StatsInterceptor;
import org.apache.openejb.spi.SecurityService;
-import org.apache.openejb.util.Duration;
-import org.apache.openejb.util.LogCategory;
-import org.apache.openejb.util.Logger;
-import org.apache.openejb.util.PassthroughFactory;
-import org.apache.openejb.util.Pool;
-import org.apache.openejb.util.SafeToolkit;
+import org.apache.openejb.util.*;
import org.apache.xbean.recipe.ObjectRecipe;
import org.apache.xbean.recipe.Option;
@@ -58,14 +53,8 @@ import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
public class StatelessInstanceManager {
private static final Logger logger =
Logger.getInstance(LogCategory.OPENEJB, "org.apache.openejb.util.resources");
@@ -90,19 +79,27 @@ public class StatelessInstanceManager {
private final Pool.Builder poolBuilder;
private final Executor executor;
- public StatelessInstanceManager(SecurityService securityService, Duration
accessTimeout, Duration closeTimeout, Pool.Builder poolBuilder, int
callbackThreads) {
+ public StatelessInstanceManager(final SecurityService securityService,
final Duration accessTimeout, final Duration closeTimeout, final Pool.Builder
poolBuilder, final int callbackThreads) {
this.securityService = securityService;
this.accessTimeout = accessTimeout;
this.closeTimeout = closeTimeout;
this.poolBuilder = poolBuilder;
- if (accessTimeout.getUnit() == null)
accessTimeout.setUnit(TimeUnit.MILLISECONDS);
+ if (accessTimeout.getUnit() == null) {
+ accessTimeout.setUnit(TimeUnit.MILLISECONDS);
+ }
+
+ final int qsize = (callbackThreads > 1 ? callbackThreads - 1 : 1);
executor = new ThreadPoolExecutor(callbackThreads, callbackThreads * 2,
- 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
- public Thread newThread(Runnable runable) {
- Thread t = new Thread(runable, "StatelessPool");
+ 1L, TimeUnit.MINUTES,
+ new LinkedBlockingQueue<Runnable>(qsize), new ThreadFactory() {
+
+ private final AtomicInteger i = new AtomicInteger(1);
+
+ @Override
+ public Thread newThread(final Runnable runable) {
+ final Thread t = new Thread(runable, "StatelessPool.worker." +
i.getAndIncrement());
t.setDaemon(true);
return t;
}
@@ -114,13 +111,14 @@ public class StatelessInstanceManager {
private class StatelessSupplier implements Pool.Supplier<Instance> {
private final BeanContext beanContext;
- private StatelessSupplier(BeanContext beanContext) {
+ private StatelessSupplier(final BeanContext beanContext) {
this.beanContext = beanContext;
}
- public void discard(Instance instance, Pool.Event reason) {
- ThreadContext ctx = new ThreadContext(beanContext, null);
- ThreadContext oldCallContext = ThreadContext.enter(ctx);
+ @Override
+ public void discard(final Instance instance, final Pool.Event reason) {
+ final ThreadContext ctx = new ThreadContext(beanContext, null);
+ final ThreadContext oldCallContext = ThreadContext.enter(ctx);
try {
freeInstance(ctx, instance);
} finally {
@@ -128,9 +126,10 @@ public class StatelessInstanceManager {
}
}
+ @Override
public Instance create() {
- ThreadContext ctx = new ThreadContext(beanContext, null);
- ThreadContext oldCallContext = ThreadContext.enter(ctx);
+ final ThreadContext ctx = new ThreadContext(beanContext, null);
+ final ThreadContext oldCallContext = ThreadContext.enter(ctx);
try {
return ceateInstance(ctx, ctx.getBeanContext());
} catch (OpenEJBException e) {
@@ -146,21 +145,21 @@ public class StatelessInstanceManager {
/**
* Removes an instance from the pool and returns it for use
* by the container in business methods.
- *
+ * <p/>
* If the pool is at it's limit the StrictPooling flag will
* cause this thread to wait.
- *
+ * <p/>
* If StrictPooling is not enabled this method will create a
* new stateless bean instance performing all required injection
* and callbacks before returning it in a method ready state.
*
- * @param callContext
- * @return
+ * @param callContext ThreadContext
+ * @return Object
* @throws OpenEJBException
*/
- public Object getInstance(ThreadContext callContext) throws
OpenEJBException {
- BeanContext beanContext = callContext.getBeanContext();
- Data data = (Data) beanContext.getContainerData();
+ public Object getInstance(final ThreadContext callContext) throws
OpenEJBException {
+ final BeanContext beanContext = callContext.getBeanContext();
+ final Data data = (Data) beanContext.getContainerData();
Instance instance = null;
try {
@@ -171,7 +170,7 @@ public class StatelessInstanceManager {
instance.setPoolEntry(entry);
}
} catch (TimeoutException e) {
- ConcurrentAccessTimeoutException timeoutException = new
ConcurrentAccessTimeoutException("No instances available in Stateless Session
Bean pool. Waited " + data.accessTimeout.toString());
+ final ConcurrentAccessTimeoutException timeoutException = new
ConcurrentAccessTimeoutException("No instances available in Stateless Session
Bean pool. Waited " + data.accessTimeout.toString());
timeoutException.fillInStackTrace();
throw new ApplicationException(timeoutException);
@@ -185,7 +184,7 @@ public class StatelessInstanceManager {
return ceateInstance(callContext, beanContext);
}
- private Instance ceateInstance(ThreadContext callContext, BeanContext
beanContext) throws org.apache.openejb.ApplicationException {
+ private Instance ceateInstance(final ThreadContext callContext, final
BeanContext beanContext) throws org.apache.openejb.ApplicationException {
try {
@@ -197,7 +196,7 @@ public class StatelessInstanceManager {
try {
callContext.setCurrentOperation(Operation.CREATE);
final Method create = beanContext.getCreateMethod();
- final InterceptorStack ejbCreate = new
InterceptorStack(context.getBean(), create, Operation.CREATE, new
ArrayList<InterceptorData>(), new HashMap());
+ final InterceptorStack ejbCreate = new
InterceptorStack(context.getBean(), create, Operation.CREATE, new
ArrayList<InterceptorData>(), new HashMap<String, Object>());
ejbCreate.invoke();
} finally {
callContext.setCurrentOperation(originalOperation);
@@ -209,7 +208,7 @@ public class StatelessInstanceManager {
if (e instanceof InvocationTargetException) {
e = ((InvocationTargetException) e).getTargetException();
}
- String t = "The bean instance " + beanContext.getDeploymentID() +
" threw a system exception:" + e;
+ final String t = "The bean instance " +
beanContext.getDeploymentID() + " threw a system exception:" + e;
logger.error(t, e);
throw new org.apache.openejb.ApplicationException(new
RemoteException("Cannot obtain a free instance.", e));
}
@@ -218,25 +217,25 @@ public class StatelessInstanceManager {
/**
* All instances are removed from the pool in getInstance(...). They are
only
* returned by the StatelessContainer via this method under two
circumstances.
- *
+ * <p/>
* 1. The business method returns normally
* 2. The business method throws an application exception
- *
+ * <p/>
* Instances are not returned to the pool if the business method threw a
system
* exception.
*
- * @param callContext
- * @param bean
+ * @param callContext ThreadContext
+ * @param bean Object
* @throws OpenEJBException
*/
- public void poolInstance(ThreadContext callContext, Object bean) throws
OpenEJBException {
+ public void poolInstance(final ThreadContext callContext, final Object
bean) throws OpenEJBException {
if (bean == null) throw new SystemException("Invalid arguments");
- Instance instance = Instance.class.cast(bean);
+ final Instance instance = Instance.class.cast(bean);
- BeanContext beanContext = callContext.getBeanContext();
- Data data = (Data) beanContext.getContainerData();
+ final BeanContext beanContext = callContext.getBeanContext();
+ final Data data = (Data) beanContext.getContainerData();
- Pool<Instance> pool = data.getPool();
+ final Pool<Instance> pool = data.getPool();
if (instance.getPoolEntry() != null) {
pool.push(instance.getPoolEntry());
@@ -249,8 +248,8 @@ public class StatelessInstanceManager {
* This method is called to release the semaphore in case of the business
method
* throwing a system exception
*
- * @param callContext
- * @param bean
+ * @param callContext ThreadContext
+ * @param bean Object
*/
public void discardInstance(final ThreadContext callContext, final Object
bean) throws SystemException {
if (bean == null) throw new SystemException("Invalid arguments");
@@ -265,15 +264,15 @@ public class StatelessInstanceManager {
}
}
- private void freeInstance(ThreadContext callContext, Instance instance) {
+ private void freeInstance(final ThreadContext callContext, final Instance
instance) {
try {
callContext.setCurrentOperation(Operation.PRE_DESTROY);
- BeanContext beanContext = callContext.getBeanContext();
+ final BeanContext beanContext = callContext.getBeanContext();
- Method remove = instance.bean instanceof SessionBean ?
removeSessionBeanMethod : null;
+ final Method remove = instance.bean instanceof SessionBean ?
removeSessionBeanMethod : null;
- List<InterceptorData> callbackInterceptors =
beanContext.getCallbackInterceptors();
- InterceptorStack interceptorStack = new
InterceptorStack(instance.bean, remove, Operation.PRE_DESTROY,
callbackInterceptors, instance.interceptors);
+ final List<InterceptorData> callbackInterceptors =
beanContext.getCallbackInterceptors();
+ final InterceptorStack interceptorStack = new
InterceptorStack(instance.bean, remove, Operation.PRE_DESTROY,
callbackInterceptors, instance.interceptors);
interceptorStack.invoke();
@@ -286,12 +285,13 @@ public class StatelessInstanceManager {
}
- public void deploy(BeanContext beanContext) throws OpenEJBException {
- Options options = new Options(beanContext.getProperties());
+ @SuppressWarnings("unchecked")
+ public void deploy(final BeanContext beanContext) throws OpenEJBException {
+ final Options options = new Options(beanContext.getProperties());
Duration accessTimeout = getDuration(options, "Timeout",
this.accessTimeout, TimeUnit.MILLISECONDS);
accessTimeout = getDuration(options, "AccessTimeout", accessTimeout,
TimeUnit.MILLISECONDS);
- Duration closeTimeout = getDuration(options, "CloseTimeout",
this.closeTimeout, TimeUnit.MINUTES);
+ final Duration closeTimeout = getDuration(options, "CloseTimeout",
this.closeTimeout, TimeUnit.MINUTES);
final ObjectRecipe recipe = PassthroughFactory.recipe(new
Pool.Builder(poolBuilder));
recipe.allow(Option.CASE_INSENSITIVE_FACTORY);
@@ -308,8 +308,7 @@ public class StatelessInstanceManager {
builder.setSupplier(supplier);
builder.setExecutor(executor);
-
- Data data = new Data(builder.build(), accessTimeout, closeTimeout);
+ final Data data = new Data(builder.build(), accessTimeout,
closeTimeout);
beanContext.setContainerData(data);
beanContext.set(EJBContext.class, data.sessionContext);
@@ -324,8 +323,8 @@ public class StatelessInstanceManager {
}
final int min = builder.getMin();
- long maxAge = builder.getMaxAge().getTime(TimeUnit.MILLISECONDS);
- double maxAgeOffset = builder.getMaxAgeOffset();
+ final long maxAge = builder.getMaxAge().getTime(TimeUnit.MILLISECONDS);
+ final double maxAgeOffset = builder.getMaxAgeOffset();
final ObjectNameBuilder jmxName = new
ObjectNameBuilder("openejb.management");
jmxName.set("J2EEServer", "openejb");
@@ -340,7 +339,7 @@ public class StatelessInstanceManager {
if (StatsInterceptor.isStatsActivated()) {
StatsInterceptor stats = null;
- for (InterceptorInstance interceptor :
beanContext.getUserAndSystemInterceptors()) {
+ for (final InterceptorInstance interceptor :
beanContext.getUserAndSystemInterceptors()) {
if (interceptor.getInterceptor() instanceof StatsInterceptor) {
stats = (StatsInterceptor) interceptor.getInterceptor();
}
@@ -352,7 +351,7 @@ public class StatelessInstanceManager {
// register the invocation stats interceptor
try {
- ObjectName objectName = jmxName.set("j2eeType",
"Invocations").build();
+ final ObjectName objectName = jmxName.set("j2eeType",
"Invocations").build();
if (server.isRegistered(objectName)) {
server.unregisterMBean(objectName);
}
@@ -365,7 +364,7 @@ public class StatelessInstanceManager {
// register the pool
try {
- ObjectName objectName = jmxName.set("j2eeType", "Pool").build();
+ final ObjectName objectName = jmxName.set("j2eeType",
"Pool").build();
if (server.isRegistered(objectName)) {
server.unregisterMBean(objectName);
}
@@ -377,7 +376,7 @@ public class StatelessInstanceManager {
// Finally, fill the pool and start it
if (!options.get("BackgroundStartup", false) && min > 0) {
- ExecutorService es = Executors.newFixedThreadPool(min);
+ final ExecutorService es = Executors.newFixedThreadPool(min);
for (int i = 0; i < min; i++) {
es.submit(new InstanceCreatorRunnable(maxAge, i, min,
maxAgeOffset, data, supplier));
}
@@ -392,23 +391,23 @@ public class StatelessInstanceManager {
data.getPool().start();
}
- private void setDefault(Duration duration, TimeUnit unit) {
+ private void setDefault(final Duration duration, final TimeUnit unit) {
if (duration.getUnit() == null) duration.setUnit(unit);
}
- private Duration getDuration(Options options, String property, Duration
defaultValue, TimeUnit defaultUnit) {
- String s = options.get(property, defaultValue.toString());
- Duration duration = new Duration(s);
+ private Duration getDuration(final Options options, final String property,
final Duration defaultValue, final TimeUnit defaultUnit) {
+ final String s = options.get(property, defaultValue.toString());
+ final Duration duration = new Duration(s);
if (duration.getUnit() == null) duration.setUnit(defaultUnit);
return duration;
}
- public void undeploy(BeanContext beanContext) {
- Data data = (Data) beanContext.getContainerData();
+ public void undeploy(final BeanContext beanContext) {
+ final Data data = (Data) beanContext.getContainerData();
if (data == null) return;
- MBeanServer server = LocalMBeanServer.get();
- for (ObjectName objectName : data.jmxNames) {
+ final MBeanServer server = LocalMBeanServer.get();
+ for (final ObjectName objectName : data.jmxNames) {
try {
server.unregisterMBean(objectName);
} catch (Exception e) {
@@ -434,11 +433,12 @@ public class StatelessInstanceManager {
private final List<ObjectName> jmxNames = new ArrayList<ObjectName>();
private final SessionContext sessionContext;
- private Data(Pool<Instance> pool, Duration accessTimeout, Duration
closeTimeout) {
+ private Data(final Pool<Instance> pool, final Duration accessTimeout,
final Duration closeTimeout) {
this.pool = pool;
this.accessTimeout = accessTimeout;
this.closeTimeout = closeTimeout;
this.sessionContext = new StatelessContext(securityService, new
Flushable() {
+ @Override
public void flush() throws IOException {
getPool().flush();
}
@@ -461,7 +461,7 @@ public class StatelessInstanceManager {
return pool.close(closeTimeout.getTime(), closeTimeout.getUnit());
}
- public ObjectName add(ObjectName name) {
+ public ObjectName add(final ObjectName name) {
jmxNames.add(name);
return name;
}
@@ -475,7 +475,7 @@ public class StatelessInstanceManager {
private Data data;
private StatelessSupplier supplier;
- private InstanceCreatorRunnable(long maxAge, long iteration, long min,
double maxAgeOffset, Data data, StatelessSupplier supplier) {
+ private InstanceCreatorRunnable(final long maxAge, final long
iteration, final long min, final double maxAgeOffset, final Data data, final
StatelessSupplier supplier) {
this.maxAge = maxAge;
this.iteration = iteration;
this.min = min;
@@ -488,7 +488,7 @@ public class StatelessInstanceManager {
public void run() {
final Instance obj = supplier.create();
if (obj != null) {
- long offset = maxAge > 0 ? (long) (maxAge / maxAgeOffset * min
* iteration) % maxAge : 0l;
+ final long offset = maxAge > 0 ? (long) (maxAge / maxAgeOffset
* min * iteration) % maxAge : 0l;
data.getPool().add(obj, offset);
}
}
Modified:
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/resource/jdbc/pool/DefaultDataSourceCreator.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/resource/jdbc/pool/DefaultDataSourceCreator.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/resource/jdbc/pool/DefaultDataSourceCreator.java
(original)
+++
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/resource/jdbc/pool/DefaultDataSourceCreator.java
Thu Jan 17 14:42:30 2013
@@ -36,7 +36,7 @@ public class DefaultDataSourceCreator ex
}
@Override
- public DataSource poolManaged(final String name, final DataSource ds,
Properties properties) {
+ public DataSource poolManaged(final String name, final DataSource ds,
final Properties properties) {
return new DbcpManagedDataSource(name, ds);
}
@@ -55,7 +55,7 @@ public class DefaultDataSourceCreator ex
}
@Override
- public DataSource pool(final String name, final DataSource ds, Properties
properties) {
+ public DataSource pool(final String name, final DataSource ds, final
Properties properties) {
return new DbcpDataSource(name, ds);
}
Modified:
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/ExecutorBuilder.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/ExecutorBuilder.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/ExecutorBuilder.java
(original)
+++
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/ExecutorBuilder.java
Thu Jan 17 14:42:30 2013
@@ -19,16 +19,7 @@ package org.apache.openejb.util;
import org.apache.openejb.loader.Options;
import org.apache.openejb.util.executor.OfferRejectedExecutionHandler;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.DelayQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.PriorityBlockingQueue;
-import java.util.concurrent.RejectedExecutionHandler;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
/**
* @version $Rev$ $Date$
@@ -43,34 +34,44 @@ public class ExecutorBuilder {
public ExecutorBuilder() {
}
- public ExecutorBuilder size(int size) {
+ public ExecutorBuilder size(final int size) {
this.size = size;
return this;
}
- public ExecutorBuilder prefix(String prefix) {
+ public ExecutorBuilder prefix(final String prefix) {
this.prefix = prefix;
return this;
}
- public ExecutorBuilder threadFactory(ThreadFactory threadFactory) {
+ public ExecutorBuilder threadFactory(final ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
return this;
}
- public ExecutorBuilder rejectedExecutionHandler(RejectedExecutionHandler
rejectedExecutionHandler) {
+ public ExecutorBuilder rejectedExecutionHandler(final
RejectedExecutionHandler rejectedExecutionHandler) {
this.rejectedExecutionHandler = rejectedExecutionHandler;
return this;
}
- public ThreadPoolExecutor build(Options options) {
- final int corePoolSize = options.get(prefix + ".CorePoolSize", size);
+ @SuppressWarnings("unchecked")
+ public ThreadPoolExecutor build(final Options options) {
+
+ int corePoolSize = options.get(prefix + ".CorePoolSize", size);
+
+ if (corePoolSize < 1) {
+ corePoolSize = 1;
+ }
// Default setting is for a fixed pool size,
MaximumPoolSize==CorePoolSize
- final int maximumPoolSize = Math.max(options.get(prefix +
".MaximumPoolSize", corePoolSize), corePoolSize);
+ int maximumPoolSize = Math.max(options.get(prefix +
".MaximumPoolSize", corePoolSize), corePoolSize);
- // Default QueueSize is bounded using the MaximumPoolSize
- final int size = options.get(prefix + ".QueueSize", maximumPoolSize);
+ if (maximumPoolSize < corePoolSize) {
+ maximumPoolSize = corePoolSize;
+ }
+
+ // Default QueueSize is bounded using the corePoolSize, else bounded
pools will never grow
+ final int qsize = options.get(prefix + ".QueueSize", corePoolSize);
// Keep Threads inactive threads alive for 60 seconds by default
final Duration keepAliveTime = options.get(prefix + ".KeepAliveTime",
new Duration(60, TimeUnit.SECONDS));
@@ -79,8 +80,9 @@ public class ExecutorBuilder {
final boolean allowCoreThreadTimeout = options.get(prefix +
".AllowCoreThreadTimeOut", true);
// If the user explicitly set the QueueSize to 0, we default QueueType
to SYNCHRONOUS
- final QueueType defaultQueueType = (size == 0) ? QueueType.SYNCHRONOUS
: QueueType.LINKED;
- final BlockingQueue queue = options.get(prefix + ".QueueType",
defaultQueueType).create(options, prefix, size);
+ final QueueType defaultQueueType = (qsize < 1) ? QueueType.SYNCHRONOUS
: QueueType.LINKED;
+
+ final BlockingQueue queue = options.get(prefix + ".QueueType",
defaultQueueType).create(options, prefix, qsize);
ThreadFactory factory = this.threadFactory;
if (factory == null) {
@@ -116,7 +118,7 @@ public class ExecutorBuilder {
PRIORITY,
SYNCHRONOUS;
- public BlockingQueue create(Options options, final String prefix,
final int queueSize) {
+ public BlockingQueue create(final Options options, final String
prefix, final int queueSize) {
switch (this) {
case ARRAY: {
return new ArrayBlockingQueue(queueSize);
Modified:
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java
(original)
+++
tomee/trunk/openejb/container/openejb-core/src/main/java/org/apache/openejb/util/Pool.java
Thu Jan 17 14:42:30 2013
@@ -140,9 +140,9 @@ public class Pool<T> {
}
private Executor createExecutor() {
- return new ThreadPoolExecutor(5, 10,
- 0L, TimeUnit.SECONDS,
- new LinkedBlockingQueue<Runnable>(), new
DaemonThreadFactory("org.apache.openejb.util.Pool", hashCode()));
+ return new ThreadPoolExecutor(3, 10,
+ 60L, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<Runnable>(2), new
DaemonThreadFactory("org.apache.openejb.util.Pool", hashCode()));
}
private void greater(final String maxName, final long max, final String
minName, final long min) {
Modified: tomee/trunk/openejb/pom.xml
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/pom.xml?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
--- tomee/trunk/openejb/pom.xml (original)
+++ tomee/trunk/openejb/pom.xml Thu Jan 17 14:42:30 2013
@@ -122,7 +122,7 @@
<!-- used mainly by jetty modules -->
<cxf.version>2.6.4</cxf.version> <!--2.6.4 requires wss4j 1.6.8-->
- <ehcache.version>2.6.0</ehcache.version> <!-- used by cxf for security
(replay attack) -->
+ <ehcache.version>2.6.3</ehcache.version> <!-- used by cxf for security
(replay attack) -->
<jetty.version>7.5.3.v20111011</jetty.version>
<pax-url.version>1.3.5</pax-url.version>
<aether.version>1.13.1</aether.version>
Modified:
tomee/trunk/openejb/server/openejb-client/src/main/java/org/apache/openejb/client/EJBObjectHandler.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/server/openejb-client/src/main/java/org/apache/openejb/client/EJBObjectHandler.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/server/openejb-client/src/main/java/org/apache/openejb/client/EJBObjectHandler.java
(original)
+++
tomee/trunk/openejb/server/openejb-client/src/main/java/org/apache/openejb/client/EJBObjectHandler.java
Thu Jan 17 14:42:30 2013
@@ -55,14 +55,14 @@ public abstract class EJBObjectHandler e
//TODO figure out how to configure and manage the thread pool on the
client side, this will do for now...
private static final int threads =
Integer.parseInt(System.getProperty("openejb.client.invoker.threads", "10"));
- private static final int queue =
Integer.parseInt(System.getProperty("openejb.client.invoker.queue", "50000"));
+ private static final int queue =
Integer.parseInt(System.getProperty("openejb.client.invoker.queue", "2"));
private static final LinkedBlockingQueue<Runnable> blockingQueue = new
LinkedBlockingQueue<Runnable>((queue < 2 ? 2 : queue));
protected static final ThreadPoolExecutor executorService;
static {
/**
- This thread pool starts with 2 core threads and can grow to the limit
defined by 'threads'.
+ This thread pool starts with 3 core threads and can grow to the limit
defined by 'threads'.
If a pool thread is idle for more than 1 minute it will be discarded,
unless the core size is reached.
It can accept upto the number of processes defined by 'queue'.
If the queue is full then an attempt is made to add the process to
the queue for 10 seconds.
@@ -70,7 +70,7 @@ public abstract class EJBObjectHandler e
is true then a final attempt is made to run the process in the
current thread (the service thread).
*/
- executorService = new ThreadPoolExecutor(2, (threads < 2 ? 2 :
threads), 1, TimeUnit.MINUTES, blockingQueue);
+ executorService = new ThreadPoolExecutor(3, (threads < 3 ? 3 :
threads), 1, TimeUnit.MINUTES, blockingQueue);
executorService.setThreadFactory(new ThreadFactory() {
private final AtomicInteger i = new AtomicInteger(0);
Modified:
tomee/trunk/openejb/server/openejb-ejbd/src/test/java/org/apache/openejb/server/ejbd/KeepAilveTest.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/server/openejb-ejbd/src/test/java/org/apache/openejb/server/ejbd/KeepAilveTest.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/server/openejb-ejbd/src/test/java/org/apache/openejb/server/ejbd/KeepAilveTest.java
(original)
+++
tomee/trunk/openejb/server/openejb-ejbd/src/test/java/org/apache/openejb/server/ejbd/KeepAilveTest.java
Thu Jan 17 14:42:30 2013
@@ -47,7 +47,7 @@ public class KeepAilveTest extends TestC
@SuppressWarnings("unchecked")
public void _testPool() throws Exception {
final int threads = 2;
- final ThreadPoolExecutor pool = new ThreadPoolExecutor(threads,
threads, 120, TimeUnit.SECONDS, new LinkedBlockingQueue());
+ final ThreadPoolExecutor pool = new ThreadPoolExecutor(threads,
threads, 120, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(threads));
final Runnable runnable = new Runnable() {
@Override
Modified:
tomee/trunk/openejb/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java
(original)
+++
tomee/trunk/openejb/server/openejb-multicast/src/main/java/org/apache/openejb/server/discovery/Tracker.java
Thu Jan 17 14:42:30 2013
@@ -187,7 +187,7 @@ public class Tracker {
return debug && log.isDebugEnabled();
}
- private final Executor executor = new ThreadPoolExecutor(1, 1, 30,
TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {
+ private final Executor executor = new ThreadPoolExecutor(1, 2, 30,
TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(1), new ThreadFactory() {
@Override
public Thread newThread(final Runnable runable) {
final Thread t = new Thread(runable, "Discovery Agent Notifier");
Modified:
tomee/trunk/openejb/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java
URL:
http://svn.apache.org/viewvc/tomee/trunk/openejb/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java?rev=1434694&r1=1434693&r2=1434694&view=diff
==============================================================================
---
tomee/trunk/openejb/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java
(original)
+++
tomee/trunk/openejb/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java
Thu Jan 17 14:42:30 2013
@@ -27,11 +27,7 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Properties;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RejectedExecutionHandler;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -39,48 +35,56 @@ import java.util.concurrent.atomic.Atomi
public class ServicePool extends ServerServiceFilter {
private static final Logger log =
Logger.getInstance(LogCategory.SERVICEPOOL,
"org.apache.openejb.util.resources");
+ private static final int KEEP_ALIVE_TIME = 1000 * 60 * 1;
private final ThreadPoolExecutor threadPool;
private final AtomicBoolean stop = new AtomicBoolean();
public ServicePool(final ServerService next, final Properties properties) {
- //Liberal defaults
- this(next, new Options(properties).get("threadsCore", 2), new
Options(properties).get("threads", 50), new Options(properties).get("queue",
50000), new Options(properties).get("block", false), new
Options(properties).get("keepAliveTime", 1000 * 60 * 5));
+ /**Defaults.
+ * This suggests that 3 core threads should cope with up to 5
runnables (threads + queue).
+ * Any more than 5 runnables will spawn a thread to cope if the thread
count is less than 50.
+ * If 50 threads are processing runnables and the queue is full then
block and wait for
+ * a slot for up to 10 seconds before rejecting the runnable.
+ * If a thread remains idle for more than 1 minute then it will be
removed.
+ */
+ this(next, new Options(properties).get("threadsCore", 3), new
Options(properties).get("threads", 50), new Options(properties).get("queue",
2), new Options(properties).get("block", false), new
Options(properties).get("keepAliveTime", KEEP_ALIVE_TIME));
}
public ServicePool(final ServerService next, final int threads) {
- this(next, threads, threads, 50000, true, 1000 * 60 * 5);
+ this(next, threads, threads, threads, true, KEEP_ALIVE_TIME);
}
- public ServicePool(final ServerService next, int threads, int queue, final
boolean block) {
- this(next, threads, threads, queue, block, 1000 * 60 * 5);
+ public ServicePool(final ServerService next, final int threads, final int
queue, final boolean block) {
+ this(next, threads, threads, queue, block, KEEP_ALIVE_TIME);
}
public ServicePool(final ServerService next, int threadCore, int threads,
int queue, final boolean block, long keepAliveTime) {
super(next);
if (keepAliveTime <= 0) {
- keepAliveTime = 1000 * 60 * 5;
+ keepAliveTime = KEEP_ALIVE_TIME;
}
if (threadCore <= 0) {
- threadCore = 100;
+ threadCore = 3;
}
if (threads < threadCore) {
- threads = threadCore;
+ threads = threadCore + 1;
}
- if (queue < 1) {
- queue = 1;
+ if (queue >= threadCore) {
+ queue = threadCore - 1;
}
/**
- This thread pool starts with 2 core threads and can grow to the limit
defined by 'threads'.
- If a pool thread is idle for more than 1 minute it will be discarded,
unless the core size is reached.
- It can accept upto the number of processes defined by 'queue'.
- If the queue is full then an attempt is made to add the process to
the queue for 10 seconds.
+ This thread pool starts with 'threadCore' core threads and can grow
to the limit defined by 'threads'.
+ If a pool thread is idle for more than 1 minute it will be discarded,
until the core size is reached.
+ It can accept upto the number of runnables defined by 'queue' +
'threadCore' before the pool will grow,
+ so the 'queue' should ideally be less than 'threadsCore', but
certainly less than 'threads'.
+ If the 'queue' and 'threads' are full then an attempt is made to add
the runnable to the queue for 10 seconds.
Failure to add to the queue in this time will either result in a
logged rejection, or if 'block'
- is true then a final attempt is made to run the process in the
current thread (the service thread).
+ is true then a final attempt is made to run the runnable in the
current thread (the service thread).
*/
threadPool = new ThreadPoolExecutor(threadCore, threads,
keepAliveTime, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(queue));
