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:

java.util.concurrent.Executor, java.util.concurrent.ExecutorService

Direct Known Subclasses:

OrderedMemoryAwareThreadPoolExecutor

public class MemoryAwareThreadPoolExecutor
extends java.util.concurrent.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.

Version:

$Rev: 2351 $, $Date: 2010-08-26 11:55:10 +0900 (Thu, 26 Aug 2010) $

Author:

The Netty Project, Trustin Lee

Nested Class Summary

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

java.util.concurrent.ThreadPoolExecutor.AbortPolicy, java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy, java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy, java.util.concurrent.ThreadPoolExecutor.DiscardPolicy

Constructor Summary

MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize)
Creates a new instance.

MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit)
Creates a new instance.

MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit, ObjectSizeEstimator objectSizeEstimator, java.util.concurrent.ThreadFactory threadFactory)
Creates a new instance.

MemoryAwareThreadPoolExecutor(int corePoolSize, long maxChannelMemorySize, long maxTotalMemorySize, long keepAliveTime, java.util.concurrent.TimeUnit unit, java.util.concurrent.ThreadFactory threadFactory)
Creates a new instance.

Method Summary

protected void	beforeExecute(java.lang.Thread t, java.lang.Runnable r)
protected void	decreaseCounter(java.lang.Runnable task)
protected void	doExecute(java.lang.Runnable task) Put the actual execution logic here.
protected void	doUnorderedExecute(java.lang.Runnable task) Executes the specified task without maintaining the event order.
void	execute(java.lang.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.
ObjectSizeEstimator	getObjectSizeEstimator() Returns the ObjectSizeEstimator of this pool.
protected boolean	increaseCounter(java.lang.Runnable task)
boolean	remove(java.lang.Runnable task)
void	setMaxChannelMemorySize(long maxChannelMemorySize) Sets the maximum total size of the queued events per channel.
void	setMaxTotalMemorySize(long maxTotalMemorySize) Sets the maximum total size of the queued events for this pool.
void	setObjectSizeEstimator(ObjectSizeEstimator objectSizeEstimator) Sets the ObjectSizeEstimator of this pool.
protected boolean	shouldCount(java.lang.Runnable task) Returns true if and only if the specified task should be counted to limit the global and per-channel memory consumption.
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, shutdownNow

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, toString, 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,
                                     java.util.concurrent.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,
                                     java.util.concurrent.TimeUnit unit,
                                     java.util.concurrent.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,
                                     java.util.concurrent.TimeUnit unit,
                                     ObjectSizeEstimator objectSizeEstimator,
                                     java.util.concurrent.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 java.util.concurrent.ThreadPoolExecutor

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

public void setMaxTotalMemorySize(long maxTotalMemorySize)

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

execute

public void execute(java.lang.Runnable command)

Specified by:

execute in interface java.util.concurrent.Executor

Overrides:

execute in class java.util.concurrent.ThreadPoolExecutor

doExecute

protected void doExecute(java.lang.Runnable task)

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

doUnorderedExecute

protected final void doUnorderedExecute(java.lang.Runnable task)

Executes the specified task without maintaining the event order.

remove

public boolean remove(java.lang.Runnable task)

Overrides:

remove in class java.util.concurrent.ThreadPoolExecutor

beforeExecute

protected void beforeExecute(java.lang.Thread t,
                             java.lang.Runnable r)

Overrides:

beforeExecute in class java.util.concurrent.ThreadPoolExecutor

increaseCounter

protected boolean increaseCounter(java.lang.Runnable task)

decreaseCounter

protected void decreaseCounter(java.lang.Runnable task)

shouldCount

protected boolean shouldCount(java.lang.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.