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 `ChannelEvent`s 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.

Copyright © 2008-2012 The Netty Project. All Rights Reserved.