org.jboss.netty.handler.execution

Class MemoryAwareThreadPoolExecutor

java.lang.Object

java.util.concurrent.AbstractExecutorService

java.util.concurrent.ThreadPoolExecutor

org.jboss.netty.handler.execution.MemoryAwareThreadPoolExecutor

All Implemented Interfaces:

Executor, ExecutorService

Direct Known Subclasses:

OrderedMemoryAwareThreadPoolExecutor

public class MemoryAwareThreadPoolExecutor
extends ThreadPoolExecutor

A ThreadPoolExecutor which blocks the task submission when there's too many tasks in the queue. Both per-Channel and per-Executor limitation can be applied.

When a task (i.e. Runnable) is submitted, MemoryAwareThreadPoolExecutor calls ObjectSizeEstimator.estimateSize(Object) to get the estimated size of the task in bytes to calculate the amount of memory occupied by the unprocessed tasks.

If the total size of the unprocessed tasks exceeds either per-Channel or per-Executor threshold, any further execute(Runnable) call will block until the tasks in the queue are processed so that the total size goes under the threshold.

Using an alternative task size estimation strategy

Although the default implementation does its best to guess the size of an object of unknown type, it is always good idea to to use an alternative ObjectSizeEstimator implementation instead of the DefaultObjectSizeEstimator to avoid incorrect task size calculation, especially when:

• you are using MemoryAwareThreadPoolExecutor independently from ExecutionHandler,
• you are submitting a task whose type is not ChannelEventRunnable, or
• the message type of the MessageEvent in the ChannelEventRunnable is not ChannelBuffer.

Here is an example that demonstrates how to implement an ObjectSizeEstimator which understands a user-defined object:

 public class MyRunnable implements Runnable {

     private final byte[] data;

     public MyRunnable(byte[] data) {
         this.data = data;
     }

     public void run() {
         // Process 'data' ..
     }
 }

 public class MyObjectSizeEstimator extends DefaultObjectSizeEstimator {

     @Override
     public int estimateSize(Object o) {
         if (o instanceof MyRunnable) {
             return ((MyRunnable) o).data.length + 8;
         }
         return super.estimateSize(o);
     }
 }

 ThreadPoolExecutor pool = new MemoryAwareThreadPoolExecutor(
         16, 65536, 1048576, 30, TimeUnit.SECONDS,
         new MyObjectSizeEstimator(),
         Executors.defaultThreadFactory());

 pool.execute(new MyRunnable(data));
 

Event execution order

Please note that this executor does not maintain the order of the ChannelEvents for the same Channel. For example, you can even receive a "channelClosed" event before a "messageReceived" event, as depicted by the following diagram. For example, the events can be processed as depicted below:

           --------------------------------> Timeline -------------------------------->

 Thread X: --- Channel A (Event 2) --- Channel A (Event 1) --------------------------->

 Thread Y: --- Channel A (Event 3) --- Channel B (Event 2) --- Channel B (Event 3) --->

 Thread Z: --- Channel B (Event 1) --- Channel B (Event 4) --- Channel A (Event 4) --->
 

To maintain the event order, you must use OrderedMemoryAwareThreadPoolExecutor.

Nested Class Summary

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

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

Constructor Summary

Constructors

Constructor and Description
MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize) Creates a new instance.
MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, TimeUnit unit) Creates a new instance.
MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, TimeUnit unit, ObjectSizeEstimator objectSizeEstimator, ThreadFactory threadFactory) Creates a new instance.
MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) Creates a new instance.

Method Summary

Methods

Modifier and Type	Method and Description
protected void	beforeExecute(Thread t, Runnable r)
protected void	decreaseCounter(Runnable task)
protected void	doExecute(Runnable task) Put the actual execution logic here.
protected void	doUnorderedExecute(Runnable task) Executes the specified task without maintaining the event order.
void	execute(Runnable command)
long	getMaxChannelMemorySize() Returns the maximum total size of the queued events per channel.
long	getMaxTotalMemorySize() Returns the maximum total size of the queued events for this pool.
boolean	getNotifyChannelFuturesOnShutdown() Returns if the ChannelFuture's of the ChannelEventRunnable's should be notified about the shutdown of this MemoryAwareThreadPoolExecutor.
ObjectSizeEstimator	getObjectSizeEstimator() Returns the ObjectSizeEstimator of this pool.
protected void	increaseCounter(Runnable task)
boolean	remove(Runnable task)
void	setMaxChannelMemorySize(long maxChannelMemorySize) Sets the maximum total size of the queued events per channel.
void	setMaxTotalMemorySize(long maxTotalMemorySize) Deprecated. maxTotalMemorySize is not modifiable anymore.
void	setNotifyChannelFuturesOnShutdown(boolean notifyOnShutdown) If set to false no queued ChannelEventRunnable's ChannelFuture will get notified once shutdownNow() is called.
void	setObjectSizeEstimator(ObjectSizeEstimator objectSizeEstimator) Sets the ObjectSizeEstimator of this pool.
protected boolean	shouldCount(Runnable task) Returns true if and only if the specified task should be counted to limit the global and per-channel memory consumption.
List<Runnable>	shutdownNow() This will call shutdownNow(boolean) with the value of getNotifyChannelFuturesOnShutdown().
List<Runnable>	shutdownNow(boolean notify) See ThreadPoolExecutor.shutdownNow() for how it handles the shutdown.
protected void	terminated()

Methods inherited from class java.util.concurrent.ThreadPoolExecutor

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

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, getClass, hashCode, notify, notifyAll, wait, wait, wait

Constructor Detail

MemoryAwareThreadPoolExecutor

public MemoryAwareThreadPoolExecutor(int corePoolSize,
                             long maxChannelMemorySize,
                             long maxTotalMemorySize)

Creates a new instance.

Parameters:

corePoolSize - the maximum number of active threads

maxChannelMemorySize - the maximum total size of the queued events per channel. Specify 0 to disable.

maxTotalMemorySize - the maximum total size of the queued events for this pool Specify 0 to disable.

MemoryAwareThreadPoolExecutor

public MemoryAwareThreadPoolExecutor(int corePoolSize,
                             long maxChannelMemorySize,
                             long maxTotalMemorySize,
                             long keepAliveTime,
                             TimeUnit unit)

Creates a new instance.

Parameters:

corePoolSize - the maximum number of active threads

maxChannelMemorySize - the maximum total size of the queued events per channel. Specify 0 to disable.

maxTotalMemorySize - the maximum total size of the queued events for this pool Specify 0 to disable.

keepAliveTime - the amount of time for an inactive thread to shut itself down

unit - the TimeUnit of keepAliveTime

MemoryAwareThreadPoolExecutor

public MemoryAwareThreadPoolExecutor(int corePoolSize,
                             long maxChannelMemorySize,
                             long maxTotalMemorySize,
                             long keepAliveTime,
                             TimeUnit unit,
                             ThreadFactory threadFactory)

Creates a new instance.

Parameters:

corePoolSize - the maximum number of active threads

maxChannelMemorySize - the maximum total size of the queued events per channel. Specify 0 to disable.

maxTotalMemorySize - the maximum total size of the queued events for this pool Specify 0 to disable.

keepAliveTime - the amount of time for an inactive thread to shut itself down

unit - the TimeUnit of keepAliveTime

threadFactory - the ThreadFactory of this pool

MemoryAwareThreadPoolExecutor

public MemoryAwareThreadPoolExecutor(int corePoolSize,
                             long maxChannelMemorySize,
                             long maxTotalMemorySize,
                             long keepAliveTime,
                             TimeUnit unit,
                             ObjectSizeEstimator objectSizeEstimator,
                             ThreadFactory threadFactory)

Creates a new instance.

Parameters:

corePoolSize - the maximum number of active threads

maxChannelMemorySize - the maximum total size of the queued events per channel. Specify 0 to disable.

maxTotalMemorySize - the maximum total size of the queued events for this pool Specify 0 to disable.

keepAliveTime - the amount of time for an inactive thread to shut itself down

unit - the TimeUnit of keepAliveTime

threadFactory - the ThreadFactory of this pool

objectSizeEstimator - the ObjectSizeEstimator of this pool

Method Detail

terminated

protected void terminated()

Overrides:

terminated in class ThreadPoolExecutor

shutdownNow

public List<Runnable> shutdownNow()

This will call shutdownNow(boolean) with the value of getNotifyChannelFuturesOnShutdown().

Specified by:

shutdownNow in interface ExecutorService

Overrides:

shutdownNow in class ThreadPoolExecutor

shutdownNow

public List<Runnable> shutdownNow(boolean notify)

See ThreadPoolExecutor.shutdownNow() for how it handles the shutdown. If true is given to this method it also notifies all ChannelFuture's of the not executed ChannelEventRunnable's.

Be aware that if you call this with false you will need to handle the notification of the ChannelFuture's by your self. So only use this if you really have a use-case for it.

getObjectSizeEstimator

public ObjectSizeEstimator getObjectSizeEstimator()

Returns the ObjectSizeEstimator of this pool.

setObjectSizeEstimator

public void setObjectSizeEstimator(ObjectSizeEstimator objectSizeEstimator)

Sets the ObjectSizeEstimator of this pool.

getMaxChannelMemorySize

public long getMaxChannelMemorySize()

Returns the maximum total size of the queued events per channel.

setMaxChannelMemorySize

public void setMaxChannelMemorySize(long maxChannelMemorySize)

Sets the maximum total size of the queued events per channel. Specify 0 to disable.

getMaxTotalMemorySize

public long getMaxTotalMemorySize()

Returns the maximum total size of the queued events for this pool.

setMaxTotalMemorySize

@Deprecated
public void setMaxTotalMemorySize(long maxTotalMemorySize)

Deprecated. maxTotalMemorySize is not modifiable anymore.

setNotifyChannelFuturesOnShutdown

public void setNotifyChannelFuturesOnShutdown(boolean notifyOnShutdown)

If set to false no queued ChannelEventRunnable's ChannelFuture will get notified once shutdownNow() is called. If set to true every queued ChannelEventRunnable will get marked as failed via ChannelFuture.setFailure(Throwable).

Please only set this to false if you want to handle the notification by yourself and know what you are doing. Default is true.

getNotifyChannelFuturesOnShutdown

public boolean getNotifyChannelFuturesOnShutdown()

Returns if the ChannelFuture's of the ChannelEventRunnable's should be notified about the shutdown of this MemoryAwareThreadPoolExecutor.

execute

public void execute(Runnable command)

Specified by:

execute in interface Executor

Overrides:

execute in class ThreadPoolExecutor

doExecute

protected void doExecute(Runnable task)

Put the actual execution logic here. The default implementation simply calls doUnorderedExecute(Runnable).

doUnorderedExecute

protected final void doUnorderedExecute(Runnable task)

Executes the specified task without maintaining the event order.

remove

public boolean remove(Runnable task)

Overrides:

remove in class ThreadPoolExecutor

beforeExecute

protected void beforeExecute(Thread t,
                 Runnable r)

Overrides:

beforeExecute in class ThreadPoolExecutor

increaseCounter

protected void increaseCounter(Runnable task)

decreaseCounter

protected void decreaseCounter(Runnable task)

shouldCount

protected boolean shouldCount(Runnable task)

Returns true if and only if the specified task should be counted to limit the global and per-channel memory consumption. To override this method, you must call super.shouldCount() to make sure important tasks are not counted.