001 /*
002 * Copyright (C) 2007 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 * http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016
017 package com.google.common.util.concurrent;
018
019 import com.google.common.annotations.Beta;
020
021 import java.util.Collections;
022 import java.util.List;
023 import java.util.concurrent.AbstractExecutorService;
024 import java.util.concurrent.ExecutorService;
025 import java.util.concurrent.Executors;
026 import java.util.concurrent.RejectedExecutionException;
027 import java.util.concurrent.ScheduledExecutorService;
028 import java.util.concurrent.ScheduledThreadPoolExecutor;
029 import java.util.concurrent.ThreadFactory;
030 import java.util.concurrent.ThreadPoolExecutor;
031 import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
032 import java.util.concurrent.TimeUnit;
033 import java.util.concurrent.locks.Condition;
034 import java.util.concurrent.locks.Lock;
035 import java.util.concurrent.locks.ReentrantLock;
036
037 /**
038 * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link
039 * ExecutorService}, and {@link ThreadFactory}.
040 *
041 * @author Eric Fellheimer
042 * @author Kyle Littlefield
043 * @author Justin Mahoney
044 * @since 3
045 */
046 @Beta
047 public final class MoreExecutors {
048 private MoreExecutors() {}
049
050 /**
051 * Converts the given ThreadPoolExecutor into an ExecutorService that exits
052 * when the application is complete. It does so by using daemon threads and
053 * adding a shutdown hook to wait for their completion.
054 *
055 * <p>This is mainly for fixed thread pools.
056 * See {@link Executors#newFixedThreadPool(int)}.
057 *
058 * @param executor the executor to modify to make sure it exits when the
059 * application is finished
060 * @param terminationTimeout how long to wait for the executor to
061 * finish before terminating the JVM
062 * @param timeUnit unit of time for the time parameter
063 * @return an unmodifiable version of the input which will not hang the JVM
064 */
065 public static ExecutorService getExitingExecutorService(
066 ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
067 executor.setThreadFactory(new ThreadFactoryBuilder()
068 .setDaemon(true)
069 .setThreadFactory(executor.getThreadFactory())
070 .build());
071
072 ExecutorService service = Executors.unconfigurableExecutorService(executor);
073
074 addDelayedShutdownHook(service, terminationTimeout, timeUnit);
075
076 return service;
077 }
078
079 /**
080 * Converts the given ScheduledThreadPoolExecutor into a
081 * ScheduledExecutorService that exits when the application is complete. It
082 * does so by using daemon threads and adding a shutdown hook to wait for
083 * their completion.
084 *
085 * <p>This is mainly for fixed thread pools.
086 * See {@link Executors#newScheduledThreadPool(int)}.
087 *
088 * @param executor the executor to modify to make sure it exits when the
089 * application is finished
090 * @param terminationTimeout how long to wait for the executor to
091 * finish before terminating the JVM
092 * @param timeUnit unit of time for the time parameter
093 * @return an unmodifiable version of the input which will not hang the JVM
094 */
095 public static ScheduledExecutorService getExitingScheduledExecutorService(
096 ScheduledThreadPoolExecutor executor, long terminationTimeout,
097 TimeUnit timeUnit) {
098 executor.setThreadFactory(new ThreadFactoryBuilder()
099 .setDaemon(true)
100 .setThreadFactory(executor.getThreadFactory())
101 .build());
102
103 ScheduledExecutorService service =
104 Executors.unconfigurableScheduledExecutorService(executor);
105
106 addDelayedShutdownHook(service, terminationTimeout, timeUnit);
107
108 return service;
109 }
110
111 /**
112 * Add a shutdown hook to wait for thread completion in the given
113 * {@link ExecutorService service}. This is useful if the given service uses
114 * daemon threads, and we want to keep the JVM from exiting immediately on
115 * shutdown, instead giving these daemon threads a chance to terminate
116 * normally.
117 * @param service ExecutorService which uses daemon threads
118 * @param terminationTimeout how long to wait for the executor to finish
119 * before terminating the JVM
120 * @param timeUnit unit of time for the time parameter
121 */
122 public static void addDelayedShutdownHook(
123 final ExecutorService service, final long terminationTimeout,
124 final TimeUnit timeUnit) {
125 Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
126 @Override
127 public void run() {
128 try {
129 // We'd like to log progress and failures that may arise in the
130 // following code, but unfortunately the behavior of logging
131 // is undefined in shutdown hooks.
132 // This is because the logging code installs a shutdown hook of its
133 // own. See Cleaner class inside {@link LogManager}.
134 service.shutdown();
135 service.awaitTermination(terminationTimeout, timeUnit);
136 } catch (InterruptedException ignored) {
137 // We're shutting down anyway, so just ignore.
138 }
139 }
140 }));
141 }
142
143 /**
144 * Converts the given ThreadPoolExecutor into an ExecutorService that exits
145 * when the application is complete. It does so by using daemon threads and
146 * adding a shutdown hook to wait for their completion.
147 *
148 * <p>This method waits 120 seconds before continuing with JVM termination,
149 * even if the executor has not finished its work.
150 *
151 * <p>This is mainly for fixed thread pools.
152 * See {@link Executors#newFixedThreadPool(int)}.
153 *
154 * @param executor the executor to modify to make sure it exits when the
155 * application is finished
156 * @return an unmodifiable version of the input which will not hang the JVM
157 */
158 public static ExecutorService getExitingExecutorService(
159 ThreadPoolExecutor executor) {
160 return getExitingExecutorService(executor, 120, TimeUnit.SECONDS);
161 }
162
163 /**
164 * Converts the given ThreadPoolExecutor into a ScheduledExecutorService that
165 * exits when the application is complete. It does so by using daemon threads
166 * and adding a shutdown hook to wait for their completion.
167 *
168 * <p>This method waits 120 seconds before continuing with JVM termination,
169 * even if the executor has not finished its work.
170 *
171 * <p>This is mainly for fixed thread pools.
172 * See {@link Executors#newScheduledThreadPool(int)}.
173 *
174 * @param executor the executor to modify to make sure it exits when the
175 * application is finished
176 * @return an unmodifiable version of the input which will not hang the JVM
177 */
178 public static ScheduledExecutorService getExitingScheduledExecutorService(
179 ScheduledThreadPoolExecutor executor) {
180 return getExitingScheduledExecutorService(executor, 120, TimeUnit.SECONDS);
181 }
182
183 /**
184 * Creates an executor service that runs each task in the thread
185 * that invokes {@code execute/submit}, as in {@link CallerRunsPolicy} This
186 * applies both to individually submitted tasks and to collections of tasks
187 * submitted via {@code invokeAll} or {@code invokeAny}. In the latter case,
188 * tasks will run serially on the calling thread. Tasks are run to
189 * completion before a {@code Future} is returned to the caller (unless the
190 * executor has been shutdown).
191 *
192 * <p>Although all tasks are immediately executed in the thread that
193 * submitted the task, this {@code ExecutorService} imposes a small
194 * locking overhead on each task submission in order to implement shutdown
195 * and termination behavior.
196 *
197 * <p>The implementation deviates from the {@code ExecutorService}
198 * specification with regards to the {@code shutdownNow} method. First,
199 * "best-effort" with regards to canceling running tasks is implemented
200 * as "no-effort". No interrupts or other attempts are made to stop
201 * threads executing tasks. Second, the returned list will always be empty,
202 * as any submitted task is considered to have started execution.
203 * This applies also to tasks given to {@code invokeAll} or {@code invokeAny}
204 * which are pending serial execution, even the subset of the tasks that
205 * have not yet started execution. It is unclear from the
206 * {@code ExecutorService} specification if these should be included, and
207 * it's much easier to implement the interpretation that they not be.
208 * Finally, a call to {@code shutdown} or {@code shutdownNow} may result
209 * in concurrent calls to {@code invokeAll/invokeAny} throwing
210 * RejectedExecutionException, although a subset of the tasks may already
211 * have been executed.
212 */
213 public static ExecutorService sameThreadExecutor() {
214 return new SameThreadExecutorService();
215 }
216
217 /*
218 * TODO(cpovirk): make this and other classes implement
219 * ListeningExecutorService?
220 */
221 // See sameThreadExecutor javadoc for behavioral notes.
222 private static class SameThreadExecutorService
223 extends AbstractExecutorService {
224 /**
225 * Lock used whenever accessing the state variables
226 * (runningTasks, shutdown, terminationCondition) of the executor
227 */
228 private final Lock lock = new ReentrantLock();
229
230 /** Signaled after the executor is shutdown and running tasks are done */
231 private final Condition termination = lock.newCondition();
232
233 /*
234 * Conceptually, these two variables describe the executor being in
235 * one of three states:
236 * - Active: shutdown == false
237 * - Shutdown: runningTasks > 0 and shutdown == true
238 * - Terminated: runningTasks == 0 and shutdown == true
239 */
240 private int runningTasks = 0;
241 private boolean shutdown = false;
242
243 @Override
244 public void execute(Runnable command) {
245 startTask();
246 try {
247 command.run();
248 } finally {
249 endTask();
250 }
251 }
252
253 @Override
254 public boolean isShutdown() {
255 lock.lock();
256 try {
257 return shutdown;
258 } finally {
259 lock.unlock();
260 }
261 }
262
263 @Override
264 public void shutdown() {
265 lock.lock();
266 try {
267 shutdown = true;
268 } finally {
269 lock.unlock();
270 }
271 }
272
273 // See sameThreadExecutor javadoc for unusual behavior of this method.
274 @Override
275 public List<Runnable> shutdownNow() {
276 shutdown();
277 return Collections.emptyList();
278 }
279
280 @Override
281 public boolean isTerminated() {
282 lock.lock();
283 try {
284 return shutdown && runningTasks == 0;
285 } finally {
286 lock.unlock();
287 }
288 }
289
290 @Override
291 public boolean awaitTermination(long timeout, TimeUnit unit)
292 throws InterruptedException {
293 long nanos = unit.toNanos(timeout);
294 lock.lock();
295 try {
296 for (;;) {
297 if (isTerminated()) {
298 return true;
299 } else if (nanos <= 0) {
300 return false;
301 } else {
302 nanos = termination.awaitNanos(nanos);
303 }
304 }
305 } finally {
306 lock.unlock();
307 }
308 }
309
310 /**
311 * Checks if the executor has been shut down and increments the running
312 * task count.
313 *
314 * @throws RejectedExecutionException if the executor has been previously
315 * shutdown
316 */
317 private void startTask() {
318 lock.lock();
319 try {
320 if (isShutdown()) {
321 throw new RejectedExecutionException("Executor already shutdown");
322 }
323 runningTasks++;
324 } finally {
325 lock.unlock();
326 }
327 }
328
329 /**
330 * Decrements the running task count.
331 */
332 private void endTask() {
333 lock.lock();
334 try {
335 runningTasks--;
336 if (isTerminated()) {
337 termination.signalAll();
338 }
339 } finally {
340 lock.unlock();
341 }
342 }
343 }
344 }