common/util/src/main/java/org/opendaylight/yangtools/util/concurrent/CachedThreadPoolExecutor.java

   1 /*
   2  * Copyright (c) 2014 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.yangtools.util.concurrent;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.base.MoreObjects;
  13 import com.google.common.base.MoreObjects.ToStringHelper;
  14 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  15 import java.util.concurrent.BlockingQueue;
  16 import java.util.concurrent.LinkedBlockingQueue;
  17 import java.util.concurrent.RejectedExecutionException;
  18 import java.util.concurrent.RejectedExecutionHandler;
  19 import java.util.concurrent.SynchronousQueue;
  20 import java.util.concurrent.ThreadPoolExecutor;
  21 import java.util.concurrent.TimeUnit;
  22 import org.slf4j.LoggerFactory;
  23
  24 /**
  25  * A ThreadPoolExecutor with a specified bounded queue capacity that favors reusing previously
  26  * constructed threads, when they are available, over creating new threads.
  27  *
  28  * <p>See {@link SpecialExecutors#newBoundedCachedThreadPool} for more details.
  29  *
  30  * @author Thomas Pantelis
  31  */
  32 public class CachedThreadPoolExecutor extends ThreadPoolExecutor {
  33
  34     private static final long IDLE_TIMEOUT_IN_SEC = 60L;
  35
  36     private final ExecutorQueue executorQueue;
  37
  38     private final String threadPrefix;
  39
  40     private final int maximumQueueSize;
  41
  42     private final RejectedTaskHandler rejectedTaskHandler;
  43
  44     /**
  45      * Constructs an instance.
  46      *
  47      * @param maximumPoolSize
  48      *            the maximum number of threads to allow in the pool. Threads will terminate after
  49      *            being idle for 60 seconds.
  50      * @param maximumQueueSize
  51      *            the capacity of the queue.
  52      * @param threadPrefix
  53      *            the name prefix for threads created by this executor.
  54      * @param loggerIdentity
  55      *               the class to use as logger name for logging uncaught exceptions from the threads.
  56      */
  57     // due to loggerIdentity argument usage
  58     @SuppressWarnings("checkstyle:LoggerFactoryClassParameter")
  59     public CachedThreadPoolExecutor(final int maximumPoolSize, final int maximumQueueSize, final String threadPrefix,
  60             final Class<?> loggerIdentity) {
  61         // We're using a custom SynchronousQueue that has a backing bounded LinkedBlockingQueue.
  62         // We don't specify any core threads (first parameter) so, when a task is submitted,
  63         // the base class will always try to offer to the queue. If there is an existing waiting
  64         // thread, the offer will succeed and the task will be handed to the thread to execute. If
  65         // there's no waiting thread, either because there are no threads in the pool or all threads
  66         // are busy, the base class will try to create a new thread. If the maximum thread limit has
  67         // been reached, the task will be rejected. We specify a RejectedTaskHandler that tries
  68         // to offer to the backing queue. If that succeeds, the task will execute as soon as a
  69         // thread becomes available. If the offer fails to the backing queue, the task is rejected.
  70         super(0, maximumPoolSize, IDLE_TIMEOUT_IN_SEC, TimeUnit.SECONDS,
  71                new ExecutorQueue(maximumQueueSize));
  72
  73         this.threadPrefix = requireNonNull(threadPrefix);
  74         this.maximumQueueSize = maximumQueueSize;
  75
  76         setThreadFactory(ThreadFactoryProvider.builder().namePrefix(threadPrefix)
  77                 .logger(LoggerFactory.getLogger(loggerIdentity)).build().get());
  78
  79         executorQueue = (ExecutorQueue)super.getQueue();
  80
  81         rejectedTaskHandler = new RejectedTaskHandler(
  82                 executorQueue.getBackingQueue(), CountingRejectedExecutionHandler.newAbortPolicy());
  83         super.setRejectedExecutionHandler(rejectedTaskHandler);
  84     }
  85
  86     /**
  87      * Constructor.
  88      * @deprecated Please use {@link #CachedThreadPoolExecutor(int, int, String, Class)} instead.
  89      */
  90     @Deprecated
  91     public CachedThreadPoolExecutor(final int maximumPoolSize, final int maximumQueueSize, final String threadPrefix) {
  92         this(maximumPoolSize, maximumQueueSize, threadPrefix, CachedThreadPoolExecutor.class);
  93     }
  94
  95     @Override
  96     public void setRejectedExecutionHandler(final RejectedExecutionHandler handler) {
  97         rejectedTaskHandler.setDelegateRejectedExecutionHandler(requireNonNull(handler));
  98     }
  99
 100     @Override
 101     public RejectedExecutionHandler getRejectedExecutionHandler() {
 102         return rejectedTaskHandler.getDelegateRejectedExecutionHandler();
 103     }
 104
 105     @Override
 106     public BlockingQueue<Runnable> getQueue() {
 107         return executorQueue.getBackingQueue();
 108     }
 109
 110     public long getLargestQueueSize() {
 111         return executorQueue.getBackingQueue().getLargestQueueSize();
 112     }
 113
 114     protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
 115         return toStringHelper;
 116     }
 117
 118     @Override
 119     public final String toString() {
 120         return addToStringAttributes(MoreObjects.toStringHelper(this)
 121                 .add("Thread Prefix", threadPrefix)
 122                 .add("Current Thread Pool Size", getPoolSize())
 123                 .add("Largest Thread Pool Size", getLargestPoolSize())
 124                 .add("Max Thread Pool Size", getMaximumPoolSize())
 125                 .add("Current Queue Size", executorQueue.getBackingQueue().size())
 126                 .add("Largest Queue Size", getLargestQueueSize())
 127                 .add("Max Queue Size", maximumQueueSize)
 128                 .add("Active Thread Count", getActiveCount())
 129                 .add("Completed Task Count", getCompletedTaskCount())
 130                 .add("Total Task Count", getTaskCount())).toString();
 131     }
 132
 133     /**
 134      * A customized SynchronousQueue that has a backing bounded LinkedBlockingQueue. This class
 135      * overrides the #poll methods to first try to poll the backing queue for a task. If the backing
 136      * queue is empty, it calls the base SynchronousQueue#poll method. In this manner, we get the
 137      * thread reuse behavior of the SynchronousQueue with the added ability to queue tasks when all
 138      * threads are busy.
 139      */
 140     private static class ExecutorQueue extends SynchronousQueue<Runnable> {
 141
 142         private static final long serialVersionUID = 1L;
 143
 144         private static final long POLL_WAIT_TIME_IN_MS = 300;
 145
 146         @SuppressFBWarnings("SE_BAD_FIELD")
 147         // Runnable is not Serializable
 148         private final TrackingLinkedBlockingQueue<Runnable> backingQueue;
 149
 150         ExecutorQueue(final int maxBackingQueueSize) {
 151             backingQueue = new TrackingLinkedBlockingQueue<>(maxBackingQueueSize);
 152         }
 153
 154         TrackingLinkedBlockingQueue<Runnable> getBackingQueue() {
 155             return backingQueue;
 156         }
 157
 158         @Override
 159         public Runnable poll(final long timeout, final TimeUnit unit) throws InterruptedException {
 160             long totalWaitTime = unit.toMillis(timeout);
 161             long waitTime = Math.min(totalWaitTime, POLL_WAIT_TIME_IN_MS);
 162             Runnable task = null;
 163
 164             // We loop here, each time polling the backingQueue first then our queue, instead of
 165             // polling each just once. This is to handle the following timing edge case:
 166             //
 167             //   We poll the backingQueue and it's empty but, before the call to super.poll,
 168             //   a task is offered but no thread is immediately available and the task is put on the
 169             //   backingQueue. There is a slight chance that all the other threads could be at the
 170             //   same point, in which case they would all call super.poll and wait. If we only
 171             //   called poll once, no thread would execute the task (unless/until another task was
 172             //   later submitted). But by looping and breaking the specified timeout into small
 173             //   periods, one thread will eventually wake up and get the task from the backingQueue
 174             //   and execute it, although slightly delayed.
 175
 176             while (task == null) {
 177                 // First try to get a task from the backing queue.
 178                 task = backingQueue.poll();
 179                 if (task == null) {
 180                     // No task in backing - call the base class to wait for one to be offered.
 181                     task = super.poll(waitTime, TimeUnit.MILLISECONDS);
 182
 183                     totalWaitTime -= POLL_WAIT_TIME_IN_MS;
 184                     if (totalWaitTime <= 0) {
 185                         break;
 186                     }
 187
 188                     waitTime = Math.min(totalWaitTime, POLL_WAIT_TIME_IN_MS);
 189                 }
 190             }
 191
 192             return task;
 193         }
 194
 195         @Override
 196         public Runnable poll() {
 197             Runnable task = backingQueue.poll();
 198             return task != null ? task : super.poll();
 199         }
 200     }
 201
 202     /**
 203      * Internal RejectedExecutionHandler that tries to offer rejected tasks to the backing queue.
 204      * If the queue is full, we throw a RejectedExecutionException by default. The client can
 205      * override this behavior be specifying their own RejectedExecutionHandler, in which case we
 206      * delegate to that handler.
 207      */
 208     private static class RejectedTaskHandler implements RejectedExecutionHandler {
 209
 210         private final LinkedBlockingQueue<Runnable> backingQueue;
 211         private volatile RejectedExecutionHandler delegateRejectedExecutionHandler;
 212
 213         RejectedTaskHandler(final LinkedBlockingQueue<Runnable> backingQueue,
 214                              final RejectedExecutionHandler delegateRejectedExecutionHandler) {
 215             this.backingQueue = backingQueue;
 216             this.delegateRejectedExecutionHandler = delegateRejectedExecutionHandler;
 217         }
 218
 219         void setDelegateRejectedExecutionHandler(
 220                 final RejectedExecutionHandler delegateRejectedExecutionHandler) {
 221             this.delegateRejectedExecutionHandler = delegateRejectedExecutionHandler;
 222         }
 223
 224         RejectedExecutionHandler getDelegateRejectedExecutionHandler() {
 225             return delegateRejectedExecutionHandler;
 226         }
 227
 228         @Override
 229         public void rejectedExecution(final Runnable task, final ThreadPoolExecutor executor) {
 230             if (executor.isShutdown()) {
 231                 throw new RejectedExecutionException("Executor has been shutdown.");
 232             }
 233
 234             if (!backingQueue.offer(task)) {
 235                 delegateRejectedExecutionHandler.rejectedExecution(task, executor);
 236             }
 237         }
 238     }
 239 }