org.aredis.util.concurrent

Class LimitingTaskExecutor

java.lang.Object

java.util.concurrent.AbstractExecutorService

org.aredis.util.concurrent.LimitingTaskExecutor

All Implemented Interfaces:

Executor, ExecutorService

public class LimitingTaskExecutor
extends AbstractExecutorService

This class is a utility class implementing ExecutorService which can sit on top of any ExecutorService like ThreadPoolExecutor and limit the number of concurrent tasks submitted to it using a configured limit.

It sort of provides a fixed size view of a global ThreadPool of higher capacity instead of creating a new Fixed Size Thread Pool. It prevents eating into threads of the underlying Thread Pool when the concurrent tasks exceed the specified limit by queuing the tasks.

It will be useful for purposes like providing a small size ThreadPool for message subscription from a global Thread Pool.

Notes:

Each task executed is run by an instance of LimitingTaskExecutor.TaskWrapper. The TaskWrapper which is submitted to the underlying executorService re-uses the same run method for running any tasks in LimitingTaskExecutors queue. So the competedTasks count of the underlying executorService may not indicate the correct number of tasks processed from the LimitingTaskExecutor.

When the underlying executorService rejects a task it is queued if there is at least one outstanding task submitted successfully to de-queue it. Otherwise the task is rejected with the LimitingExecutor's configured RejectedTaskHandler. Rejections by the underlying executorSerice are detected by handling the RejectedExecutionException in case of the default configuration or by using a ThreadLocal in case of ThreadPoolExecutor.CallerRunsPolicy. Note that you should not use the ThreadPoolExecutor.DiscardPolicy handler because LimitingTaskExecutor cannot detect task rejection in that case. If you want to configure a discard policy for the underlying executorService configure your own RejectedExecutionHandler and include a call to LimitingTaskExecutor.TaskWrapper.rejectExecution() in it as below:

 
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                if (r instanceof TaskWrapper) {
                    ((TaskWrapper) r).rejectExecution();
                }
            }
 
 

TaskRejection can also happen when submitting a task when the max allowed limit of tasks are already submitted and the queue is full. The default queue is configured with a limit of 150. A task is also rejected when the LimitingTaskExecutor or the underlying executorService is shutdown. In all cases the Configured RejectedTaskHandler's rejectedExecution method is called. The default handler is ABORT_POLICY which throws a RejectedExecutionException.

The number of outstanding tasks submitted to the underlying executorService is usually within the specified limit though in some corner cases it could exceed the limit.

Author:

Suresh

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
class	LimitingTaskExecutor.TaskWrapper TaskWrapper responsible for running the tasks which is submitted to the underlying ExecutorService.

Field Summary

Fields

Modifier and Type	Field and Description
static RejectedTaskHandler	ABORT_POLICY RejectedTaskHandler which Aborts the rejected task by throwing a RejectedExecutionException.
static RejectedTaskHandler	DISCARD_POLICY RejectedTaskHandler which discards the rejected task by doing nothing.
static RejectedTaskHandler	RUN_IN_PLACE_POLICY RejectedTaskHandler which runs the rejected task in place by calling the run method of the submitted task so that the task runs in the same thread as the caller effectively slowing down the caller thread.

Constructor Summary

Constructors

Constructor and Description
LimitingTaskExecutor(ExecutorService pexecutorService, int ptaskLimit) Creates a Limiting TaskExecutor with ABORT_POLICY as the RejectedExecutionPolicy.
LimitingTaskExecutor(ExecutorService pexecutorService, int ptaskLimit, BlockingQueue<Runnable> pwaitingTasks) Creates a Limiting TaskExecutor with ABORT_POLICY as the RejectedExecutionPolicy.
LimitingTaskExecutor(ExecutorService pexecutorService, int ptaskLimit, BlockingQueue<Runnable> pwaitingTasks, RejectedTaskHandler prejectedTaskHandler) Creates a Limiting TaskExecutor.

Method Summary

Methods

Modifier and Type	Method and Description
boolean	awaitTermination(long timeout, TimeUnit unit) Calls awaitTermination on the underlying Executor Service.
void	execute(Runnable command) Executes the command by passing it on to the underlying executorService or queues it for later execution if the number of outstanding tasks submitted to the executorService is already the specified limit.
int	getNumCompetedTasks() Returns the number of tasks that have been completed so far by this LimitingTaskExecutor which is a continously increasing value.
int	getNumRunningTasks() Returns the number of outstanding tasks submitted to the underlying executorService.
int	getTaskLimit() Returns the Task Limit which specifies the max no.
BlockingQueue<Runnable>	getWaitingTaskQueue() Returns the queue used to enqueue tasks for later processing.
boolean	isShutdown() Return true if shutdown has been called on this LimitingTaskExecutor or the underlying executorService is shutdown.
boolean	isTerminated() Checks if this LimitingTaskExecutor is terminated after a shutdown call or a shutdown of the underlying executorService.
void	setTaskLimit(int ptaskLimit) Sets a new value of the task limit which should be between 1 and 1000.
void	shutdown() Shuts down this LimitingTaskExecutor.
List<Runnable>	shutdownNow() In addition to shut down the task queue is emptied and returned.
String	toString() Returns a string identifying this executor, as well as its state, including indications of run state and estimated worker and task counts.

Methods inherited from class java.util.concurrent.AbstractExecutorService

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

Methods inherited from class java.lang.Object

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

Field Detail

ABORT_POLICY

public static final RejectedTaskHandler ABORT_POLICY

RejectedTaskHandler which Aborts the rejected task by throwing a RejectedExecutionException. This is the default handler.

RUN_IN_PLACE_POLICY

public static final RejectedTaskHandler RUN_IN_PLACE_POLICY

RejectedTaskHandler which runs the rejected task in place by calling the run method of the submitted task so that the task runs in the same thread as the caller effectively slowing down the caller thread.

DISCARD_POLICY

public static final RejectedTaskHandler DISCARD_POLICY

RejectedTaskHandler which discards the rejected task by doing nothing.

Constructor Detail

LimitingTaskExecutor

public LimitingTaskExecutor(ExecutorService pexecutorService,
                    int ptaskLimit,
                    BlockingQueue<Runnable> pwaitingTasks,
                    RejectedTaskHandler prejectedTaskHandler)

Creates a Limiting TaskExecutor.

Parameters:

pexecutorService - Underlying Executor Service used to submit the tasks to

ptaskLimit - Task Limit. When this many tasks are running (i.e submitted but not completed) on the underlying Executor Service, new tasks are added to the waiting tasks Q

pwaitingTasks - The Q to use for adding waiting tasks

prejectedTaskHandler - Specifies what to do with tasks submitted when the max number of tasks are running and the waiting tasks Q is also full.

LimitingTaskExecutor

public LimitingTaskExecutor(ExecutorService pexecutorService,
                    int ptaskLimit,
                    BlockingQueue<Runnable> pwaitingTasks)

Creates a Limiting TaskExecutor with ABORT_POLICY as the RejectedExecutionPolicy.

Parameters:

pexecutorService - Underlying Executor Service used to submit the tasks to

ptaskLimit - Task Limit. When this many tasks are running (i.e submitted but not completed) on the underlying Executor Service, new tasks are added to the waiting tasks Q

pwaitingTasks - The Q to use for adding waiting tasks

LimitingTaskExecutor

public LimitingTaskExecutor(ExecutorService pexecutorService,
                    int ptaskLimit)

Creates a Limiting TaskExecutor with ABORT_POLICY as the RejectedExecutionPolicy. A LinkedBlockingQueue of capacity 150 is used as the waiting tasks Q.

Parameters:

pexecutorService - Underlying Executor Service used to submit the tasks to

ptaskLimit - Task Limit. When this many tasks are running (i.e submitted but not completed) on the underlying Executor Service, new tasks are added to the waiting tasks Q

Method Detail

shutdown

public void shutdown()

Shuts down this LimitingTaskExecutor. New tasks are rejected but already submitted and queued tasks are run normally. The underlying executorService is not shut down.

See Also:

ExecutorService.shutdown()

shutdownNow

public List<Runnable> shutdownNow()

In addition to shut down the task queue is emptied and returned. The submitted tasks however are processed normally and not interrupted as mentioned in the spec.

See Also:

ExecutorService.shutdownNow()

isShutdown

public boolean isShutdown()

Return true if shutdown has been called on this LimitingTaskExecutor or the underlying executorService is shutdown.

See Also:

ExecutorService.isShutdown()

isTerminated

public boolean isTerminated()

Checks if this LimitingTaskExecutor is terminated after a shutdown call or a shutdown of the underlying executorService.

See Also:

ExecutorService.isTerminated()

awaitTermination

public boolean awaitTermination(long timeout,
                       TimeUnit unit)
                         throws InterruptedException

Calls awaitTermination on the underlying Executor Service.

Throws:

InterruptedException

See Also:

ExecutorService.awaitTermination(long, java.util.concurrent.TimeUnit)

execute

public void execute(Runnable command)

Executes the command by passing it on to the underlying executorService or queues it for later execution if the number of outstanding tasks submitted to the executorService is already the specified limit. Read the description about this class for more details.

See Also:

Executor.execute(java.lang.Runnable)

toString

public String toString()

Returns a string identifying this executor, as well as its state, including indications of run state and estimated worker and task counts. It also includes the toString of the underlying ExecutorService.

Overrides:

toString in class Object

Returns:

a string identifying this pool, as well as its state and the toString of the underlying ExecutorService

getTaskLimit

public int getTaskLimit()

Returns the Task Limit which specifies the max no. of outstanding tasks submitted to the underlying executorService.

Returns:

Task Limit

setTaskLimit

public void setTaskLimit(int ptaskLimit)

Sets a new value of the task limit which should be between 1 and 1000.

Parameters:

ptaskLimit - New value of task limit

getWaitingTaskQueue

public BlockingQueue<Runnable> getWaitingTaskQueue()

Returns the queue used to enqueue tasks for later processing.

Returns:

Waiting Task Queue

getNumRunningTasks

public int getNumRunningTasks()

Returns the number of outstanding tasks submitted to the underlying executorService.

Returns:

Number of running tasks

getNumCompetedTasks

public int getNumCompetedTasks()

Returns the number of tasks that have been completed so far by this LimitingTaskExecutor which is a continously increasing value.

Returns:

Total Number of completed tasks