Package de.aristaflow.adept2.util.threading

Class AFThreadPoolExecutor

java.lang.Object

java.util.concurrent.AbstractExecutorService

java.util.concurrent.ThreadPoolExecutor

de.aristaflow.adept2.util.threading.AFThreadPoolExecutor

All Implemented Interfaces:

Executor, ExecutorService

Direct Known Subclasses:

AbstractPipeliningExecutor, CachedThreadPoolExecutor

public class AFThreadPoolExecutor
extends ThreadPoolExecutor

This thread pool executor enhances a normal thread pool by:

• prepending the state of a thread to the thread name,
• logging exceptions (and results) of futures where no one has been interested in and
• logging statistics.

Thread names will be prefixed as following:

• NEW: Used for newly created threads.
• ACT: Used for active threads that are currently executing tasks.
• IDL: Used for idle threads, i.e. threads that may be reused.

Every task is wrapped by a LoggingFuture. A LoggingFuture logs an occurred exception if no one retrieved the exception. This way the exception will not be lost. Additionally, it will log if the result is relevant but no one retrieved it.
Statistics logging will take place if the size of the queue reaches the number of allowed threads. Logs will be repeated when reaching a size that is multiple of the limit. For single thread pools the statistics will be logged, if the size of the queue is larger than 100. This allows to recognise thread pools, whose configuration may be optimised.

Nested Class Summary

Nested classes/interfaces inherited from class java.util.concurrent.ThreadPoolExecutor

ThreadPoolExecutor.AbortPolicy, ThreadPoolExecutor.CallerRunsPolicy, ThreadPoolExecutor.DiscardOldestPolicy, ThreadPoolExecutor.DiscardPolicy

Field Summary

Fields

Modifier and Type	Field	Description
protected AtomicLong	activeThreads	The amount of active threads.
protected AtomicLong	completedTasks	The amount of all tasks that have been completed.
protected AtomicLong	submittedTasks	The amount of all tasks that have been submitted (known via execute(Runnable).

Constructor Summary

Constructors

Constructor	Description
AFThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Adept2ThreadFactory threadFactory, int queueLimit)	Creates a new AFThreadPoolExecutor with the given initial parameters and default rejected execution handler.
AFThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Adept2ThreadFactory threadFactory, int queueLimit, Level minLogLevelPoolStatistics)	Creates a new AFThreadPoolExecutor with the given initial parameters and default rejected execution handler.
AFThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Adept2ThreadFactory threadFactory, RejectedExecutionHandler handler, int queueLimit)	Creates a new ThreadPoolExecutor with the given initial parameters.
AFThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Adept2ThreadFactory threadFactory, RejectedExecutionHandler handler, int queueLimit, Level minLogLevelPoolStatistics)	Creates a new ThreadPoolExecutor with the given initial parameters.

Method Summary

Modifier and Type	Method	Description
protected void	afterExecute(Runnable r, Throwable t)
protected void	beforeExecute(Thread t, Runnable r)
void	execute(Runnable command)
protected <T> RunnableFuture<T>	newTaskFor(Runnable runnable, T value)
protected <T> RunnableFuture<T>	newTaskFor(Callable<T> callable)
boolean	setLogExecuteCallStack(boolean logExecuteCallStack)	Sets whether to log the stack of the caller of execute(Runnable).
void	updateQueueLimit(int limit)	Update the queue limit to inform the thread pool about a changed situation, i.e. the initial worklist update is completed.

Methods inherited from class java.util.concurrent.ThreadPoolExecutor

allowCoreThreadTimeOut, allowsCoreThreadTimeOut, awaitTermination, finalize, getActiveCount, getCompletedTaskCount, getCorePoolSize, getKeepAliveTime, getLargestPoolSize, getMaximumPoolSize, getPoolSize, getQueue, getRejectedExecutionHandler, getTaskCount, getThreadFactory, isShutdown, isTerminated, isTerminating, prestartAllCoreThreads, prestartCoreThread, purge, remove, setCorePoolSize, setKeepAliveTime, setMaximumPoolSize, setRejectedExecutionHandler, setThreadFactory, shutdown, shutdownNow, terminated, toString

Methods inherited from class java.util.concurrent.AbstractExecutorService

invokeAll, invokeAll, invokeAny, invokeAny, submit, submit, submit

Methods inherited from class java.lang.Object

clone, equals, getClass, hashCode, notify, notifyAll, wait, wait, wait

Field Detail

submittedTasks

protected final AtomicLong submittedTasks

The amount of all tasks that have been submitted (known via execute(Runnable).

completedTasks

protected final AtomicLong completedTasks

The amount of all tasks that have been completed.

activeThreads

protected final AtomicLong activeThreads

The amount of active threads. A thread counts as active from right before executing a task until right after executing a tasks. The amount is the same as the amount of threads of which the name starts with "[ACT]".

Constructor Detail

AFThreadPoolExecutor

public AFThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            Adept2ThreadFactory threadFactory,
                            int queueLimit)

Creates a new AFThreadPoolExecutor with the given initial parameters and default rejected execution handler. Name and logger of the Adept2ThreadFactory will be used to log statistics.

Parameters:

corePoolSize - the number of threads to keep in the pool, even if they are idle.

maximumPoolSize - the maximum number of threads to allow in the pool.

keepAliveTime - when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.

unit - the time unit for the keepAliveTime argument.

workQueue - the queue to use for holding tasks before they are executed. This queue will hold only the Runnable tasks submitted by the execute method.

threadFactory - the factory to use when the executor creates a new thread.

queueLimit - The amount of tasks in the queue before warnings will be logged. Use -1 to use the default limits.

Throws:

IllegalArgumentException - if corePoolSize or keepAliveTime less than zero, or if maximumPoolSize less than or equal to zero, or if corePoolSize greater than maximumPoolSize.

NullPointerException - if workQueue or threadFactory are null.

AFThreadPoolExecutor

public AFThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            Adept2ThreadFactory threadFactory,
                            int queueLimit,
                            Level minLogLevelPoolStatistics)

Creates a new AFThreadPoolExecutor with the given initial parameters and default rejected execution handler. Name and logger of the Adept2ThreadFactory will be used to log statistics.

Parameters:

corePoolSize - the number of threads to keep in the pool, even if they are idle.

maximumPoolSize - the maximum number of threads to allow in the pool.

keepAliveTime - when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.

unit - the time unit for the keepAliveTime argument.

workQueue - the queue to use for holding tasks before they are executed. This queue will hold only the Runnable tasks submitted by the execute method.

threadFactory - the factory to use when the executor creates a new thread.

queueLimit - The amount of tasks in the queue before warnings will be logged. Use -1 to use the default limits.

minLogLevelPoolStatistics - The log level which need to be reached for logging pool statistics. Pool statistics logs with a lower log level will be ignored.

Throws:

IllegalArgumentException - if corePoolSize or keepAliveTime less than zero, or if maximumPoolSize less than or equal to zero, or if corePoolSize greater than maximumPoolSize.

NullPointerException - if workQueue or threadFactory are null.

AFThreadPoolExecutor

public AFThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            Adept2ThreadFactory threadFactory,
                            RejectedExecutionHandler handler,
                            int queueLimit)

Creates a new ThreadPoolExecutor with the given initial parameters. Name and logger of the Adept2ThreadFactory will be used to log statistics.

Parameters:

corePoolSize - the number of threads to keep in the pool, even if they are idle.

maximumPoolSize - the maximum number of threads to allow in the pool.

keepAliveTime - when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.

unit - the time unit for the keepAliveTime argument.

workQueue - the queue to use for holding tasks before they are executed. This queue will hold only the Runnable tasks submitted by the execute method.

threadFactory - the factory to use when the executor creates a new thread.

handler - the handler to use when execution is blocked because the thread bounds and queue capacities are reached.

queueLimit - The amount of tasks in the queue before warnings will be logged. Use -1 to use the default limits.

Throws:

IllegalArgumentException - if corePoolSize or keepAliveTime less than zero, or if maximumPoolSize less than or equal to zero, or if corePoolSize greater than maximumPoolSize.

NullPointerException - if workQueue or threadFactory or handler are null.

AFThreadPoolExecutor

public AFThreadPoolExecutor(int corePoolSize,
                            int maximumPoolSize,
                            long keepAliveTime,
                            TimeUnit unit,
                            BlockingQueue<Runnable> workQueue,
                            Adept2ThreadFactory threadFactory,
                            RejectedExecutionHandler handler,
                            int queueLimit,
                            Level minLogLevelPoolStatistics)

Creates a new ThreadPoolExecutor with the given initial parameters. Name and logger of the Adept2ThreadFactory will be used to log statistics.

Parameters:

corePoolSize - the number of threads to keep in the pool, even if they are idle.

maximumPoolSize - the maximum number of threads to allow in the pool.

keepAliveTime - when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating.

unit - the time unit for the keepAliveTime argument.

workQueue - the queue to use for holding tasks before they are executed. This queue will hold only the Runnable tasks submitted by the execute method.

threadFactory - the factory to use when the executor creates a new thread.

handler - the handler to use when execution is blocked because the thread bounds and queue capacities are reached.

queueLimit - The amount of tasks in the queue before warnings will be logged. Use -1 to use the default limits.

minLogLevelPoolStatistics - The log level which need to be reached for logging pool statistics. Pool statistics logs with a lower log level will be ignored.

Throws:

IllegalArgumentException - if corePoolSize or keepAliveTime less than zero, or if maximumPoolSize less than or equal to zero, or if corePoolSize greater than maximumPoolSize.

NullPointerException - if workQueue or threadFactory or handler are null.

Method Detail

beforeExecute

protected void beforeExecute(Thread t,
                             Runnable r)

Overrides:

beforeExecute in class ThreadPoolExecutor

execute

public void execute(Runnable command)

Specified by:

execute in interface Executor

Overrides:

execute in class ThreadPoolExecutor

newTaskFor

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable,
                                           T value)

Overrides:

newTaskFor in class AbstractExecutorService

newTaskFor

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable)

Overrides:

newTaskFor in class AbstractExecutorService

afterExecute

protected void afterExecute(Runnable r,
                            Throwable t)

Overrides:

afterExecute in class ThreadPoolExecutor

updateQueueLimit

public void updateQueueLimit(int limit)

Update the queue limit to inform the thread pool about a changed situation, i.e. the initial worklist update is completed.

Parameters:

limit - The new queue limit.

setLogExecuteCallStack

public boolean setLogExecuteCallStack(boolean logExecuteCallStack)

Sets whether to log the stack of the caller of execute(Runnable). This allows to identify which code parts have enqueued some asynchronous work. The logger requires Level.FINE for this.

Parameters:

logExecuteCallStack - Whether to log the stack of the caller of execute(Runnable).

Returns:

The previous value of logExecuteCallStack.