2 * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.config.threadpool.util;
11 import com.google.common.base.Optional;
12 import java.io.Closeable;
13 import java.util.concurrent.BlockingQueue;
14 import java.util.concurrent.ExecutorService;
15 import java.util.concurrent.Executors;
16 import java.util.concurrent.LinkedBlockingQueue;
17 import java.util.concurrent.RejectedExecutionException;
18 import java.util.concurrent.RejectedExecutionHandler;
19 import java.util.concurrent.ThreadFactory;
20 import java.util.concurrent.ThreadPoolExecutor;
21 import java.util.concurrent.TimeUnit;
22 import org.opendaylight.controller.config.threadpool.ThreadPool;
25 * Implementation of {@link ThreadPool} using flexible number of threads wraps
26 * {@link ExecutorService}.
28 public class FlexibleThreadPoolWrapper implements ThreadPool, Closeable {
29 private final ThreadPoolExecutor executor;
31 public FlexibleThreadPoolWrapper(int minThreadCount, int maxThreadCount, long keepAlive, TimeUnit timeUnit,
32 ThreadFactory threadFactory) {
33 this(minThreadCount, maxThreadCount, keepAlive, timeUnit, threadFactory, getQueue(Optional.<Integer>absent()));
36 public FlexibleThreadPoolWrapper(int minThreadCount, int maxThreadCount, long keepAlive, TimeUnit timeUnit,
37 ThreadFactory threadFactory, Optional<Integer> queueCapacity) {
38 this(minThreadCount, maxThreadCount, keepAlive, timeUnit, threadFactory, getQueue(queueCapacity));
41 private FlexibleThreadPoolWrapper(int minThreadCount, int maxThreadCount, long keepAlive, TimeUnit timeUnit,
42 ThreadFactory threadFactory, BlockingQueue<Runnable> queue) {
44 executor = new ThreadPoolExecutor(minThreadCount, maxThreadCount, keepAlive, timeUnit,
45 queue, threadFactory, new FlexibleRejectionHandler());
46 executor.prestartAllCoreThreads();
50 * Overriding the queue:
51 * ThreadPoolExecutor would not create new threads if the queue is not full, thus adding
52 * occurs in RejectedExecutionHandler.
53 * This impl saturates threadpool first, then queue. When both are full caller will get blocked.
55 private static ForwardingBlockingQueue getQueue(Optional<Integer> capacity) {
56 final BlockingQueue<Runnable> delegate = capacity.isPresent() ? new LinkedBlockingQueue<>(capacity.get()) : new LinkedBlockingQueue<>();
57 return new ForwardingBlockingQueue(delegate);
61 public ExecutorService getExecutor() {
62 return Executors.unconfigurableExecutorService(executor);
65 public int getMinThreadCount() {
66 return executor.getCorePoolSize();
69 public void setMinThreadCount(int minThreadCount) {
70 executor.setCorePoolSize(minThreadCount);
74 public int getMaxThreadCount() {
75 return executor.getMaximumPoolSize();
78 public void setMaxThreadCount(int maxThreadCount) {
79 executor.setMaximumPoolSize(maxThreadCount);
82 public long getKeepAliveMillis() {
83 return executor.getKeepAliveTime(TimeUnit.MILLISECONDS);
86 public void setKeepAliveMillis(long keepAliveMillis) {
87 executor.setKeepAliveTime(keepAliveMillis, TimeUnit.MILLISECONDS);
90 public void setThreadFactory(ThreadFactory threadFactory) {
91 executor.setThreadFactory(threadFactory);
94 public void prestartAllCoreThreads() {
95 executor.prestartAllCoreThreads();
104 * if the max threads are met, then it will raise a rejectedExecution. We then push to the queue.
106 private static class FlexibleRejectionHandler implements RejectedExecutionHandler {
108 public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {
110 executor.getQueue().put(r);
111 } catch (InterruptedException e) {
112 throw new RejectedExecutionException("Interrupted while waiting on the queue", e);
117 private static class ForwardingBlockingQueue extends com.google.common.util.concurrent.ForwardingBlockingQueue<Runnable> {
118 private final BlockingQueue<Runnable> delegate;
120 public ForwardingBlockingQueue(BlockingQueue<Runnable> delegate) {
121 this.delegate = delegate;
125 protected BlockingQueue<Runnable> delegate() {
130 public boolean offer(final Runnable r) {
131 // ThreadPoolExecutor will spawn a new thread after core size is reached only
132 // if the queue.offer returns false.