package org.apache.catalina.core;
-import java.util.Collection;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.catalina.Executor;
import org.apache.catalina.LifecycleException;
import org.apache.catalina.LifecycleListener;
import org.apache.catalina.util.LifecycleSupport;
-import java.util.concurrent.RejectedExecutionException;
+import org.apache.tomcat.util.threads.TaskQueue;
+import org.apache.tomcat.util.threads.TaskThreadFactory;
+import org.apache.tomcat.util.threads.ThreadPoolExecutor;
public class StandardThreadExecutor implements Executor {
public void start() throws LifecycleException {
lifecycle.fireLifecycleEvent(BEFORE_START_EVENT, null);
TaskQueue taskqueue = new TaskQueue(maxQueueSize);
- TaskThreadFactory tf = new TaskThreadFactory(namePrefix);
+ TaskThreadFactory tf = new TaskThreadFactory(namePrefix,daemon,getThreadPriority());
lifecycle.fireLifecycleEvent(START_EVENT, null);
executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), maxIdleTime, TimeUnit.MILLISECONDS,taskqueue, tf);
taskqueue.setParent( (ThreadPoolExecutor) executor);
public void execute(Runnable command, long timeout, TimeUnit unit) {
if ( executor != null ) {
- try {
- executor.execute(command);
- } catch (RejectedExecutionException rx) {
- //there could have been contention around the queue
- try {
- if ( !( (TaskQueue) executor.getQueue()).force(command,timeout,unit) ) throw new RejectedExecutionException("Work queue full.");
- }catch (InterruptedException x) {
- throw new RejectedExecutionException("Interrupted.",x);
- }
- }
- } else throw new IllegalStateException("StandardThreadPool not started.");
+ executor.execute(command,timeout,unit);
+ } else {
+ throw new IllegalStateException("StandardThreadExecutor not started.");
+ }
}
public int getQueueSize() {
return (executor != null) ? executor.getQueue().size() : -1;
}
-
- // ---------------------------------------------- TaskQueue Inner Class
- class TaskQueue extends LinkedBlockingQueue<Runnable> {
- ThreadPoolExecutor parent = null;
-
- public TaskQueue() {
- super();
- }
-
- public TaskQueue(int capacity) {
- super(capacity);
- }
-
- public TaskQueue(Collection<? extends Runnable> c) {
- super(c);
- }
-
- public void setParent(ThreadPoolExecutor tp) {
- parent = tp;
- }
-
- public boolean force(Runnable o) {
- if ( parent.isShutdown() ) throw new RejectedExecutionException("Executor not running, can't force a command into the queue");
- return super.offer(o); //forces the item onto the queue, to be used if the task is rejected
- }
-
- public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
- if ( parent.isShutdown() ) throw new RejectedExecutionException("Executor not running, can't force a command into the queue");
- return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected
- }
-
- public boolean offer(Runnable o) {
- //we can't do any checks
- if (parent==null) return super.offer(o);
- //we are maxed out on threads, simply queue the object
- if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
- //we have idle threads, just add it to the queue
- //this is an approximation, so it could use some tuning
- if (parent.getActiveCount()<(parent.getPoolSize())) return super.offer(o);
- //if we have less threads than maximum force creation of a new thread
- if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
- //if we reached here, we need to add it to the queue
- return super.offer(o);
- }
- }
-
- // ---------------------------------------------- ThreadFactory Inner Class
- class TaskThreadFactory implements ThreadFactory {
- final ThreadGroup group;
- final AtomicInteger threadNumber = new AtomicInteger(1);
- final String namePrefix;
-
- TaskThreadFactory(String namePrefix) {
- SecurityManager s = System.getSecurityManager();
- group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
- this.namePrefix = namePrefix;
- }
-
- public Thread newThread(Runnable r) {
- Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement());
- t.setDaemon(daemon);
- t.setPriority(getThreadPriority());
- return t;
- }
- }
-
-
}
import org.apache.tomcat.util.net.SecureNioChannel.ApplicationBufferHandler;
import org.apache.tomcat.util.net.jsse.NioX509KeyManager;
import org.apache.tomcat.util.res.StringManager;
+import org.apache.tomcat.util.threads.TaskQueue;
+import org.apache.tomcat.util.threads.TaskThreadFactory;
/**
* NIO tailored thread pool, providing the following services:
* External Executor based thread pool.
*/
protected Executor executor = null;
- public void setExecutor(Executor executor) { this.executor = executor; }
+ public void setExecutor(Executor executor) {
+ this.executor = executor;
+ this.internalExecutor = (executor==null);
+ }
public Executor getExecutor() { return executor; }
+ /**
+ * Are we using an internal executor
+ */
+ protected boolean internalExecutor = true;
protected boolean useExecutor = true;
/**
protected int maxThreads = 200;
public void setMaxThreads(int maxThreads) {
this.maxThreads = maxThreads;
- if (running) {
- if (executor!=null) {
- if (executor instanceof ThreadPoolExecutor) {
- ((ThreadPoolExecutor)executor).setMaximumPoolSize(maxThreads);
- }
- }
+ if (running && executor!=null && executor instanceof ThreadPoolExecutor) {
+ ((ThreadPoolExecutor)executor).setMaximumPoolSize(maxThreads);
}
}
public int getMaxThreads() { return maxThreads; }
// Create worker collection
if ( executor == null ) {
+ internalExecutor = true;
TaskQueue taskqueue = new TaskQueue();
- TaskThreadFactory tf = new TaskThreadFactory(getName() + "-exec-");
+ TaskThreadFactory tf = new TaskThreadFactory(getName() + "-exec-", daemon, getThreadPriority());
executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), 60, TimeUnit.SECONDS,taskqueue, tf);
- taskqueue.setParent( (ThreadPoolExecutor) executor, this);
+ taskqueue.setParent( (ThreadPoolExecutor) executor);
}
// Start poller threads
keyCache.clear();
nioChannels.clear();
processorCache.clear();
- if ( executor!=null ) {
+ if ( executor!=null && internalExecutor ) {
if ( executor instanceof ThreadPoolExecutor ) {
//this is our internal one, so we need to shut it down
ThreadPoolExecutor tpe = (ThreadPoolExecutor) executor;
tpe.shutdown();
TaskQueue queue = (TaskQueue) tpe.getQueue();
- queue.setParent(null,null);
+ queue.setParent(null);
}
executor = null;
}
}
}
-
- // ---------------------------------------------- TaskQueue Inner Class
- public static class TaskQueue extends LinkedBlockingQueue<Runnable> {
- ThreadPoolExecutor parent = null;
- NioEndpoint endpoint = null;
-
- public TaskQueue() {
- super();
- }
- public TaskQueue(int initialCapacity) {
- super(initialCapacity);
- }
-
- public TaskQueue(Collection<? extends Runnable> c) {
- super(c);
- }
-
-
- public void setParent(ThreadPoolExecutor tp, NioEndpoint ep) {
- parent = tp;
- this.endpoint = ep;
- }
-
- public boolean offer(Runnable o) {
- //we can't do any checks
- if (parent==null) return super.offer(o);
- //we are maxed out on threads, simply queue the object
- if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
- //we have idle threads, just add it to the queue
- //this is an approximation, so it could use some tuning
- if (endpoint.activeSocketProcessors.get()<(parent.getPoolSize())) return super.offer(o);
- //if we have less threads than maximum force creation of a new thread
- if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
- //if we reached here, we need to add it to the queue
- return super.offer(o);
- }
- }
-
- // ---------------------------------------------- ThreadFactory Inner Class
- class TaskThreadFactory implements ThreadFactory {
- final ThreadGroup group;
- final AtomicInteger threadNumber = new AtomicInteger(1);
- final String namePrefix;
-
- TaskThreadFactory(String namePrefix) {
- SecurityManager s = System.getSecurityManager();
- group = (s != null)? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
- this.namePrefix = namePrefix;
- }
-
- public Thread newThread(Runnable r) {
- Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement());
- t.setDaemon(daemon);
- t.setPriority(getThreadPriority());
- return t;
- }
- }
-
// ----------------------------------------------- SendfileData Inner Class
-
-
/**
* SendfileData class.
*/
// KeepAlive flag
public boolean keepAlive;
}
-
}
--- /dev/null
+/*
+ * 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.tomcat.util.threads;
+
+import java.util.Collection;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+/**
+ * As task queue specifically designed to run with a thread pool executor.
+ * The task queue is optimised to properly utilize threads within
+ * a thread pool executor. If you use a normal queue, the executor will spawn threads
+ * when there are idle threads and you wont be able to force items unto the queue itself
+ * @author fhanik
+ *
+ */
+public class TaskQueue extends LinkedBlockingQueue<Runnable> {
+ ThreadPoolExecutor parent = null;
+
+ public TaskQueue() {
+ super();
+ }
+
+ public TaskQueue(int capacity) {
+ super(capacity);
+ }
+
+ public TaskQueue(Collection<? extends Runnable> c) {
+ super(c);
+ }
+
+ public void setParent(ThreadPoolExecutor tp) {
+ parent = tp;
+ }
+
+ public boolean force(Runnable o) {
+ if ( parent.isShutdown() ) throw new RejectedExecutionException("Executor not running, can't force a command into the queue");
+ return super.offer(o); //forces the item onto the queue, to be used if the task is rejected
+ }
+
+ public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
+ if ( parent.isShutdown() ) throw new RejectedExecutionException("Executor not running, can't force a command into the queue");
+ return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected
+ }
+
+ public boolean offer(Runnable o) {
+ //we can't do any checks
+ if (parent==null) return super.offer(o);
+ //we are maxed out on threads, simply queue the object
+ if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
+ //we have idle threads, just add it to the queue
+ if (parent.getActiveCount()<(parent.getPoolSize())) return super.offer(o);
+ //if we have less threads than maximum force creation of a new thread
+ if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
+ //if we reached here, we need to add it to the queue
+ return super.offer(o);
+ }
+}
--- /dev/null
+/*
+ * 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.tomcat.util.threads;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+/**
+ * Simple task thread factory to use to create threads for an executor implementation.
+ * @author fhanik
+ *
+ */
+public class TaskThreadFactory implements ThreadFactory {
+ final ThreadGroup group;
+ final AtomicInteger threadNumber = new AtomicInteger(1);
+ final String namePrefix;
+ final boolean daemon;
+ final int threadPriority;
+ public TaskThreadFactory(String namePrefix, boolean daemon, int priority) {
+ SecurityManager s = System.getSecurityManager();
+ group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
+ this.namePrefix = namePrefix;
+ this.daemon = daemon;
+ this.threadPriority = priority;
+ }
+
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement());
+ t.setDaemon(daemon);
+ t.setPriority(threadPriority);
+ return t;
+ }
+
+}
--- /dev/null
+/*
+ * 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.tomcat.util.threads;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+/**
+ * Same as a java.util.concurrent.ThreadPoolExecutor but implements a much more efficient
+ * getActiveCount method, to be used to properly handle the work queue
+ * @author fhanik
+ *
+ */
+public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {
+
+ final AtomicInteger activeCount = new AtomicInteger(0);
+
+ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
+ super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
+ }
+
+ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
+ RejectedExecutionHandler handler) {
+ super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
+ }
+
+ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
+ super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
+ }
+
+ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
+ super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
+ }
+
+ @Override
+ protected void afterExecute(Runnable r, Throwable t) {
+ activeCount.decrementAndGet();
+ }
+
+ @Override
+ protected void beforeExecute(Thread t, Runnable r) {
+ activeCount.incrementAndGet();
+ }
+
+ @Override
+ public int getActiveCount() {
+ return activeCount.get();
+ }
+
+ public void execute(Runnable command, long timeout, TimeUnit unit) {
+
+ }
+
+
+
+
+
+}
projects / tomcat7.0 / commitdiff