1 /*
2 * Copyright (C) 2007 The Guava Authors
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
5 * in compliance with the License. You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software distributed under the License
10 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
11 * or implied. See the License for the specific language governing permissions and limitations under
12 * the License.
13 */
14
15 package com.google.common.util.concurrent;
16
17 import static com.google.common.base.Preconditions.checkArgument;
18 import static com.google.common.base.Preconditions.checkNotNull;
19 import static com.google.common.util.concurrent.Internal.toNanosSaturated;
20
21 import com.google.common.annotations.Beta;
22 import com.google.common.annotations.GwtCompatible;
23 import com.google.common.annotations.GwtIncompatible;
24 import com.google.common.annotations.VisibleForTesting;
25 import com.google.common.base.Supplier;
26 import com.google.common.base.Throwables;
27 import com.google.common.collect.Lists;
28 import com.google.common.collect.Queues;
29 import com.google.common.util.concurrent.ForwardingListenableFuture.SimpleForwardingListenableFuture;
30 import com.google.errorprone.annotations.CanIgnoreReturnValue;
31 import com.google.errorprone.annotations.concurrent.GuardedBy;
32 import java.lang.reflect.InvocationTargetException;
33 import java.time.Duration;
34 import java.util.Collection;
35 import java.util.Collections;
36 import java.util.Iterator;
37 import java.util.List;
38 import java.util.concurrent.BlockingQueue;
39 import java.util.concurrent.Callable;
40 import java.util.concurrent.Delayed;
41 import java.util.concurrent.ExecutionException;
42 import java.util.concurrent.Executor;
43 import java.util.concurrent.ExecutorService;
44 import java.util.concurrent.Executors;
45 import java.util.concurrent.Future;
46 import java.util.concurrent.RejectedExecutionException;
47 import java.util.concurrent.ScheduledExecutorService;
48 import java.util.concurrent.ScheduledFuture;
49 import java.util.concurrent.ScheduledThreadPoolExecutor;
50 import java.util.concurrent.ThreadFactory;
51 import java.util.concurrent.ThreadPoolExecutor;
52 import java.util.concurrent.TimeUnit;
53 import java.util.concurrent.TimeoutException;
54 import org.checkerframework.checker.nullness.qual.Nullable;
55
56 /**
57 * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link ExecutorService},
58 * and {@link java.util.concurrent.ThreadFactory}.
59 *
60 * @author Eric Fellheimer
61 * @author Kyle Littlefield
62 * @author Justin Mahoney
63 * @since 3.0
64 */
65 @GwtCompatible(emulated = true)
66 @ElementTypesAreNonnullByDefault
67 public final class MoreExecutors {
68 private MoreExecutors() {}
69
70 /**
71 * Converts the given ThreadPoolExecutor into an ExecutorService that exits when the application
72 * is complete. It does so by using daemon threads and adding a shutdown hook to wait for their
73 * completion.
74 *
75 * <p>This is mainly for fixed thread pools. See {@link Executors#newFixedThreadPool(int)}.
76 *
77 * @param executor the executor to modify to make sure it exits when the application is finished
78 * @param terminationTimeout how long to wait for the executor to finish before terminating the
79 * JVM
80 * @return an unmodifiable version of the input which will not hang the JVM
81 * @since 28.0
82 */
83 @Beta
84 @GwtIncompatible // TODO
85 public static ExecutorService getExitingExecutorService(
86 ThreadPoolExecutor executor, Duration terminationTimeout) {
87 return getExitingExecutorService(
88 executor, toNanosSaturated(terminationTimeout), TimeUnit.NANOSECONDS);
89 }
90
91 /**
92 * Converts the given ThreadPoolExecutor into an ExecutorService that exits when the application
93 * is complete. It does so by using daemon threads and adding a shutdown hook to wait for their
94 * completion.
95 *
96 * <p>This is mainly for fixed thread pools. See {@link Executors#newFixedThreadPool(int)}.
97 *
98 * @param executor the executor to modify to make sure it exits when the application is finished
99 * @param terminationTimeout how long to wait for the executor to finish before terminating the
100 * JVM
101 * @param timeUnit unit of time for the time parameter
102 * @return an unmodifiable version of the input which will not hang the JVM
103 */
104 @Beta
105 @GwtIncompatible // TODO
106 @SuppressWarnings("GoodTime") // should accept a java.time.Duration
107 public static ExecutorService getExitingExecutorService(
108 ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
109 return new Application().getExitingExecutorService(executor, terminationTimeout, timeUnit);
110 }
111
112 /**
113 * Converts the given ThreadPoolExecutor into an ExecutorService that exits when the application
114 * is complete. It does so by using daemon threads and adding a shutdown hook to wait for their
115 * completion.
116 *
117 * <p>This method waits 120 seconds before continuing with JVM termination, even if the executor
118 * has not finished its work.
119 *
120 * <p>This is mainly for fixed thread pools. See {@link Executors#newFixedThreadPool(int)}.
121 *
122 * @param executor the executor to modify to make sure it exits when the application is finished
123 * @return an unmodifiable version of the input which will not hang the JVM
124 */
125 @Beta
126 @GwtIncompatible // concurrency
127 public static ExecutorService getExitingExecutorService(ThreadPoolExecutor executor) {
128 return new Application().getExitingExecutorService(executor);
129 }
130
131 /**
132 * Converts the given ScheduledThreadPoolExecutor into a ScheduledExecutorService that exits when
133 * the application is complete. It does so by using daemon threads and adding a shutdown hook to
134 * wait for their completion.
135 *
136 * <p>This is mainly for fixed thread pools. See {@link Executors#newScheduledThreadPool(int)}.
137 *
138 * @param executor the executor to modify to make sure it exits when the application is finished
139 * @param terminationTimeout how long to wait for the executor to finish before terminating the
140 * JVM
141 * @return an unmodifiable version of the input which will not hang the JVM
142 * @since 28.0
143 */
144 @Beta
145 @GwtIncompatible // java.time.Duration
146 public static ScheduledExecutorService getExitingScheduledExecutorService(
147 ScheduledThreadPoolExecutor executor, Duration terminationTimeout) {
148 return getExitingScheduledExecutorService(
149 executor, toNanosSaturated(terminationTimeout), TimeUnit.NANOSECONDS);
150 }
151
152 /**
153 * Converts the given ScheduledThreadPoolExecutor into a ScheduledExecutorService that exits when
154 * the application is complete. It does so by using daemon threads and adding a shutdown hook to
155 * wait for their completion.
156 *
157 * <p>This is mainly for fixed thread pools. See {@link Executors#newScheduledThreadPool(int)}.
158 *
159 * @param executor the executor to modify to make sure it exits when the application is finished
160 * @param terminationTimeout how long to wait for the executor to finish before terminating the
161 * JVM
162 * @param timeUnit unit of time for the time parameter
163 * @return an unmodifiable version of the input which will not hang the JVM
164 */
165 @Beta
166 @GwtIncompatible // TODO
167 @SuppressWarnings("GoodTime") // should accept a java.time.Duration
168 public static ScheduledExecutorService getExitingScheduledExecutorService(
169 ScheduledThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
170 return new Application()
171 .getExitingScheduledExecutorService(executor, terminationTimeout, timeUnit);
172 }
173
174 /**
175 * Converts the given ScheduledThreadPoolExecutor into a ScheduledExecutorService that exits when
176 * the application is complete. It does so by using daemon threads and adding a shutdown hook to
177 * wait for their completion.
178 *
179 * <p>This method waits 120 seconds before continuing with JVM termination, even if the executor
180 * has not finished its work.
181 *
182 * <p>This is mainly for fixed thread pools. See {@link Executors#newScheduledThreadPool(int)}.
183 *
184 * @param executor the executor to modify to make sure it exits when the application is finished
185 * @return an unmodifiable version of the input which will not hang the JVM
186 */
187 @Beta
188 @GwtIncompatible // TODO
189 public static ScheduledExecutorService getExitingScheduledExecutorService(
190 ScheduledThreadPoolExecutor executor) {
191 return new Application().getExitingScheduledExecutorService(executor);
192 }
193
194 /**
195 * Add a shutdown hook to wait for thread completion in the given {@link ExecutorService service}.
196 * This is useful if the given service uses daemon threads, and we want to keep the JVM from
197 * exiting immediately on shutdown, instead giving these daemon threads a chance to terminate
198 * normally.
199 *
200 * @param service ExecutorService which uses daemon threads
201 * @param terminationTimeout how long to wait for the executor to finish before terminating the
202 * JVM
203 * @since 28.0
204 */
205 @Beta
206 @GwtIncompatible // java.time.Duration
207 public static void addDelayedShutdownHook(ExecutorService service, Duration terminationTimeout) {
208 addDelayedShutdownHook(service, toNanosSaturated(terminationTimeout), TimeUnit.NANOSECONDS);
209 }
210
211 /**
212 * Add a shutdown hook to wait for thread completion in the given {@link ExecutorService service}.
213 * This is useful if the given service uses daemon threads, and we want to keep the JVM from
214 * exiting immediately on shutdown, instead giving these daemon threads a chance to terminate
215 * normally.
216 *
217 * @param service ExecutorService which uses daemon threads
218 * @param terminationTimeout how long to wait for the executor to finish before terminating the
219 * JVM
220 * @param timeUnit unit of time for the time parameter
221 */
222 @Beta
223 @GwtIncompatible // TODO
224 @SuppressWarnings("GoodTime") // should accept a java.time.Duration
225 public static void addDelayedShutdownHook(
226 ExecutorService service, long terminationTimeout, TimeUnit timeUnit) {
227 new Application().addDelayedShutdownHook(service, terminationTimeout, timeUnit);
228 }
229
230 /** Represents the current application to register shutdown hooks. */
231 @GwtIncompatible // TODO
232 @VisibleForTesting
233 static class Application {
234
235 final ExecutorService getExitingExecutorService(
236 ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
237 useDaemonThreadFactory(executor);
238 ExecutorService service = Executors.unconfigurableExecutorService(executor);
239 addDelayedShutdownHook(executor, terminationTimeout, timeUnit);
240 return service;
241 }
242
243 final ExecutorService getExitingExecutorService(ThreadPoolExecutor executor) {
244 return getExitingExecutorService(executor, 120, TimeUnit.SECONDS);
245 }
246
247 final ScheduledExecutorService getExitingScheduledExecutorService(
248 ScheduledThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
249 useDaemonThreadFactory(executor);
250 ScheduledExecutorService service = Executors.unconfigurableScheduledExecutorService(executor);
251 addDelayedShutdownHook(executor, terminationTimeout, timeUnit);
252 return service;
253 }
254
255 final ScheduledExecutorService getExitingScheduledExecutorService(
256 ScheduledThreadPoolExecutor executor) {
257 return getExitingScheduledExecutorService(executor, 120, TimeUnit.SECONDS);
258 }
259
260 final void addDelayedShutdownHook(
261 final ExecutorService service, final long terminationTimeout, final TimeUnit timeUnit) {
262 checkNotNull(service);
263 checkNotNull(timeUnit);
264 addShutdownHook(
265 MoreExecutors.newThread(
266 "DelayedShutdownHook-for-" + service,
267 new Runnable() {
268 @Override
269 public void run() {
270 try {
271 // We'd like to log progress and failures that may arise in the
272 // following code, but unfortunately the behavior of logging
273 // is undefined in shutdown hooks.
274 // This is because the logging code installs a shutdown hook of its
275 // own. See Cleaner class inside {@link LogManager}.
276 service.shutdown();
277 service.awaitTermination(terminationTimeout, timeUnit);
278 } catch (InterruptedException ignored) {
279 // We're shutting down anyway, so just ignore.
280 }
281 }
282 }));
283 }
284
285 @VisibleForTesting
286 void addShutdownHook(Thread hook) {
287 Runtime.getRuntime().addShutdownHook(hook);
288 }
289 }
290
291 @GwtIncompatible // TODO
292 private static void useDaemonThreadFactory(ThreadPoolExecutor executor) {
293 executor.setThreadFactory(
294 new ThreadFactoryBuilder()
295 .setDaemon(true)
296 .setThreadFactory(executor.getThreadFactory())
297 .build());
298 }
299
300 // See newDirectExecutorService javadoc for behavioral notes.
301 @GwtIncompatible // TODO
302 private static final class DirectExecutorService extends AbstractListeningExecutorService {
303 /** Lock used whenever accessing the state variables (runningTasks, shutdown) of the executor */
304 private final Object lock = new Object();
305
306 /*
307 * Conceptually, these two variables describe the executor being in
308 * one of three states:
309 * - Active: shutdown == false
310 * - Shutdown: runningTasks > 0 and shutdown == true
311 * - Terminated: runningTasks == 0 and shutdown == true
312 */
313 @GuardedBy("lock")
314 private int runningTasks = 0;
315
316 @GuardedBy("lock")
317 private boolean shutdown = false;
318
319 @Override
320 public void execute(Runnable command) {
321 startTask();
322 try {
323 command.run();
324 } finally {
325 endTask();
326 }
327 }
328
329 @Override
330 public boolean isShutdown() {
331 synchronized (lock) {
332 return shutdown;
333 }
334 }
335
336 @Override
337 public void shutdown() {
338 synchronized (lock) {
339 shutdown = true;
340 if (runningTasks == 0) {
341 lock.notifyAll();
342 }
343 }
344 }
345
346 // See newDirectExecutorService javadoc for unusual behavior of this method.
347 @Override
348 public List<Runnable> shutdownNow() {
349 shutdown();
350 return Collections.emptyList();
351 }
352
353 @Override
354 public boolean isTerminated() {
355 synchronized (lock) {
356 return shutdown && runningTasks == 0;
357 }
358 }
359
360 @Override
361 public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
362 long nanos = unit.toNanos(timeout);
363 synchronized (lock) {
364 while (true) {
365 if (shutdown && runningTasks == 0) {
366 return true;
367 } else if (nanos <= 0) {
368 return false;
369 } else {
370 long now = System.nanoTime();
371 TimeUnit.NANOSECONDS.timedWait(lock, nanos);
372 nanos -= System.nanoTime() - now; // subtract the actual time we waited
373 }
374 }
375 }
376 }
377
378 /**
379 * Checks if the executor has been shut down and increments the running task count.
380 *
381 * @throws RejectedExecutionException if the executor has been previously shutdown
382 */
383 private void startTask() {
384 synchronized (lock) {
385 if (shutdown) {
386 throw new RejectedExecutionException("Executor already shutdown");
387 }
388 runningTasks++;
389 }
390 }
391
392 /** Decrements the running task count. */
393 private void endTask() {
394 synchronized (lock) {
395 int numRunning = --runningTasks;
396 if (numRunning == 0) {
397 lock.notifyAll();
398 }
399 }
400 }
401 }
402
403 /**
404 * Creates an executor service that runs each task in the thread that invokes {@code
405 * execute/submit}, as in {@code ThreadPoolExecutor.CallerRunsPolicy}. This applies both to
406 * individually submitted tasks and to collections of tasks submitted via {@code invokeAll} or
407 * {@code invokeAny}. In the latter case, tasks will run serially on the calling thread. Tasks are
408 * run to completion before a {@code Future} is returned to the caller (unless the executor has
409 * been shutdown).
410 *
411 * <p>Although all tasks are immediately executed in the thread that submitted the task, this
412 * {@code ExecutorService} imposes a small locking overhead on each task submission in order to
413 * implement shutdown and termination behavior.
414 *
415 * <p>The implementation deviates from the {@code ExecutorService} specification with regards to
416 * the {@code shutdownNow} method. First, "best-effort" with regards to canceling running tasks is
417 * implemented as "no-effort". No interrupts or other attempts are made to stop threads executing
418 * tasks. Second, the returned list will always be empty, as any submitted task is considered to
419 * have started execution. This applies also to tasks given to {@code invokeAll} or {@code
420 * invokeAny} which are pending serial execution, even the subset of the tasks that have not yet
421 * started execution. It is unclear from the {@code ExecutorService} specification if these should
422 * be included, and it's much easier to implement the interpretation that they not be. Finally, a
423 * call to {@code shutdown} or {@code shutdownNow} may result in concurrent calls to {@code
424 * invokeAll/invokeAny} throwing RejectedExecutionException, although a subset of the tasks may
425 * already have been executed.
426 *
427 * @since 18.0 (present as MoreExecutors.sameThreadExecutor() since 10.0)
428 */
429 @GwtIncompatible // TODO
430 public static ListeningExecutorService newDirectExecutorService() {
431 return new DirectExecutorService();
432 }
433
434 /**
435 * Returns an {@link Executor} that runs each task in the thread that invokes {@link
436 * Executor#execute execute}, as in {@code ThreadPoolExecutor.CallerRunsPolicy}.
437 *
438 * <p>This executor is appropriate for tasks that are lightweight and not deeply chained.
439 * Inappropriate {@code directExecutor} usage can cause problems, and these problems can be
440 * difficult to reproduce because they depend on timing. For example:
441 *
442 * <ul>
443 * <li>A call like {@code future.transform(function, directExecutor())} may execute the function
444 * immediately in the thread that is calling {@code transform}. (This specific case happens
445 * if the future is already completed.) If {@code transform} call was made from a UI thread
446 * or other latency-sensitive thread, a heavyweight function can harm responsiveness.
447 * <li>If the task will be executed later, consider which thread will trigger the execution --
448 * since that thread will execute the task inline. If the thread is a shared system thread
449 * like an RPC network thread, a heavyweight task can stall progress of the whole system or
450 * even deadlock it.
451 * <li>If many tasks will be triggered by the same event, one heavyweight task may delay other
452 * tasks -- even tasks that are not themselves {@code directExecutor} tasks.
453 * <li>If many such tasks are chained together (such as with {@code
454 * future.transform(...).transform(...).transform(...)....}), they may overflow the stack.
455 * (In simple cases, callers can avoid this by registering all tasks with the same {@link
456 * MoreExecutors#newSequentialExecutor} wrapper around {@code directExecutor()}. More
457 * complex cases may require using thread pools or making deeper changes.)
458 * </ul>
459 *
460 * Additionally, beware of executing tasks with {@code directExecutor} while holding a lock. Since
461 * the task you submit to the executor (or any other arbitrary work the executor does) may do slow
462 * work or acquire other locks, you risk deadlocks.
463 *
464 * <p>This instance is equivalent to:
465 *
466 * <pre>{@code
467 * final class DirectExecutor implements Executor {
468 * public void execute(Runnable r) {
469 * r.run();
470 * }
471 * }
472 * }</pre>
473 *
474 * <p>This should be preferred to {@link #newDirectExecutorService()} because implementing the
475 * {@link ExecutorService} subinterface necessitates significant performance overhead.
476 *
477 * @since 18.0
478 */
479 public static Executor directExecutor() {
480 return DirectExecutor.INSTANCE;
481 }
482
483 /**
484 * Returns an {@link Executor} that runs each task executed sequentially, such that no two tasks
485 * are running concurrently. Submitted tasks have a happens-before order as defined in the Java
486 * Language Specification.
487 *
488 * <p>The executor uses {@code delegate} in order to {@link Executor#execute execute} each task in
489 * turn, and does not create any threads of its own.
490 *
491 * <p>After execution begins on a thread from the {@code delegate} {@link Executor}, tasks are
492 * polled and executed from a task queue until there are no more tasks. The thread will not be
493 * released until there are no more tasks to run.
494 *
495 * <p>If a task is submitted while a thread is executing tasks from the task queue, the thread
496 * will not be released until that submitted task is also complete.
497 *
498 * <p>If a task is {@linkplain Thread#interrupt interrupted} while a task is running:
499 *
500 * <ol>
501 * <li>execution will not stop until the task queue is empty.
502 * <li>tasks will begin execution with the thread marked as not interrupted - any interruption
503 * applies only to the task that was running at the point of interruption.
504 * <li>if the thread was interrupted before the SequentialExecutor's worker begins execution,
505 * the interrupt will be restored to the thread after it completes so that its {@code
506 * delegate} Executor may process the interrupt.
507 * <li>subtasks are run with the thread uninterrupted and interrupts received during execution
508 * of a task are ignored.
509 * </ol>
510 *
511 * <p>{@code RuntimeException}s thrown by tasks are simply logged and the executor keeps trucking.
512 * If an {@code Error} is thrown, the error will propagate and execution will stop until the next
513 * time a task is submitted.
514 *
515 * <p>When an {@code Error} is thrown by an executed task, previously submitted tasks may never
516 * run. An attempt will be made to restart execution on the next call to {@code execute}. If the
517 * {@code delegate} has begun to reject execution, the previously submitted tasks may never run,
518 * despite not throwing a RejectedExecutionException synchronously with the call to {@code
519 * execute}. If this behaviour is problematic, use an Executor with a single thread (e.g. {@link
520 * Executors#newSingleThreadExecutor}).
521 *
522 * @since 23.3 (since 23.1 as {@code sequentialExecutor})
523 */
524 @Beta
525 @GwtIncompatible
526 public static Executor newSequentialExecutor(Executor delegate) {
527 return new SequentialExecutor(delegate);
528 }
529
530 /**
531 * Creates an {@link ExecutorService} whose {@code submit} and {@code invokeAll} methods submit
532 * {@link ListenableFutureTask} instances to the given delegate executor. Those methods, as well
533 * as {@code execute} and {@code invokeAny}, are implemented in terms of calls to {@code
534 * delegate.execute}. All other methods are forwarded unchanged to the delegate. This implies that
535 * the returned {@code ListeningExecutorService} never calls the delegate's {@code submit}, {@code
536 * invokeAll}, and {@code invokeAny} methods, so any special handling of tasks must be implemented
537 * in the delegate's {@code execute} method or by wrapping the returned {@code
538 * ListeningExecutorService}.
539 *
540 * <p>If the delegate executor was already an instance of {@code ListeningExecutorService}, it is
541 * returned untouched, and the rest of this documentation does not apply.
542 *
543 * @since 10.0
544 */
545 @GwtIncompatible // TODO
546 public static ListeningExecutorService listeningDecorator(ExecutorService delegate) {
547 return (delegate instanceof ListeningExecutorService)
548 ? (ListeningExecutorService) delegate
549 : (delegate instanceof ScheduledExecutorService)
550 ? new ScheduledListeningDecorator((ScheduledExecutorService) delegate)
551 : new ListeningDecorator(delegate);
552 }
553
554 /**
555 * Creates a {@link ScheduledExecutorService} whose {@code submit} and {@code invokeAll} methods
556 * submit {@link ListenableFutureTask} instances to the given delegate executor. Those methods, as
557 * well as {@code execute} and {@code invokeAny}, are implemented in terms of calls to {@code
558 * delegate.execute}. All other methods are forwarded unchanged to the delegate. This implies that
559 * the returned {@code ListeningScheduledExecutorService} never calls the delegate's {@code
560 * submit}, {@code invokeAll}, and {@code invokeAny} methods, so any special handling of tasks
561 * must be implemented in the delegate's {@code execute} method or by wrapping the returned {@code
562 * ListeningScheduledExecutorService}.
563 *
564 * <p>If the delegate executor was already an instance of {@code
565 * ListeningScheduledExecutorService}, it is returned untouched, and the rest of this
566 * documentation does not apply.
567 *
568 * @since 10.0
569 */
570 @GwtIncompatible // TODO
571 public static ListeningScheduledExecutorService listeningDecorator(
572 ScheduledExecutorService delegate) {
573 return (delegate instanceof ListeningScheduledExecutorService)
574 ? (ListeningScheduledExecutorService) delegate
575 : new ScheduledListeningDecorator(delegate);
576 }
577
578 @GwtIncompatible // TODO
579 private static class ListeningDecorator extends AbstractListeningExecutorService {
580 private final ExecutorService delegate;
581
582 ListeningDecorator(ExecutorService delegate) {
583 this.delegate = checkNotNull(delegate);
584 }
585
586 @Override
587 public final boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
588 return delegate.awaitTermination(timeout, unit);
589 }
590
591 @Override
592 public final boolean isShutdown() {
593 return delegate.isShutdown();
594 }
595
596 @Override
597 public final boolean isTerminated() {
598 return delegate.isTerminated();
599 }
600
601 @Override
602 public final void shutdown() {
603 delegate.shutdown();
604 }
605
606 @Override
607 public final List<Runnable> shutdownNow() {
608 return delegate.shutdownNow();
609 }
610
611 @Override
612 public final void execute(Runnable command) {
613 delegate.execute(command);
614 }
615
616 @Override
617 public final String toString() {
618 return super.toString() + "[" + delegate + "]";
619 }
620 }
621
622 @GwtIncompatible // TODO
623 private static final class ScheduledListeningDecorator extends ListeningDecorator
624 implements ListeningScheduledExecutorService {
625 @SuppressWarnings("hiding")
626 final ScheduledExecutorService delegate;
627
628 ScheduledListeningDecorator(ScheduledExecutorService delegate) {
629 super(delegate);
630 this.delegate = checkNotNull(delegate);
631 }
632
633 @Override
634 public ListenableScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
635 TrustedListenableFutureTask<@Nullable Void> task =
636 TrustedListenableFutureTask.create(command, null);
637 ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit);
638 return new ListenableScheduledTask<@Nullable Void>(task, scheduled);
639 }
640
641 @Override
642 public <V extends @Nullable Object> ListenableScheduledFuture<V> schedule(
643 Callable<V> callable, long delay, TimeUnit unit) {
644 TrustedListenableFutureTask<V> task = TrustedListenableFutureTask.create(callable);
645 ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit);
646 return new ListenableScheduledTask<V>(task, scheduled);
647 }
648
649 @Override
650 public ListenableScheduledFuture<?> scheduleAtFixedRate(
651 Runnable command, long initialDelay, long period, TimeUnit unit) {
652 NeverSuccessfulListenableFutureTask task = new NeverSuccessfulListenableFutureTask(command);
653 ScheduledFuture<?> scheduled = delegate.scheduleAtFixedRate(task, initialDelay, period, unit);
654 return new ListenableScheduledTask<@Nullable Void>(task, scheduled);
655 }
656
657 @Override
658 public ListenableScheduledFuture<?> scheduleWithFixedDelay(
659 Runnable command, long initialDelay, long delay, TimeUnit unit) {
660 NeverSuccessfulListenableFutureTask task = new NeverSuccessfulListenableFutureTask(command);
661 ScheduledFuture<?> scheduled =
662 delegate.scheduleWithFixedDelay(task, initialDelay, delay, unit);
663 return new ListenableScheduledTask<@Nullable Void>(task, scheduled);
664 }
665
666 private static final class ListenableScheduledTask<V extends @Nullable Object>
667 extends SimpleForwardingListenableFuture<V> implements ListenableScheduledFuture<V> {
668
669 private final ScheduledFuture<?> scheduledDelegate;
670
671 public ListenableScheduledTask(
672 ListenableFuture<V> listenableDelegate, ScheduledFuture<?> scheduledDelegate) {
673 super(listenableDelegate);
674 this.scheduledDelegate = scheduledDelegate;
675 }
676
677 @Override
678 public boolean cancel(boolean mayInterruptIfRunning) {
679 boolean cancelled = super.cancel(mayInterruptIfRunning);
680 if (cancelled) {
681 // Unless it is cancelled, the delegate may continue being scheduled
682 scheduledDelegate.cancel(mayInterruptIfRunning);
683
684 // TODO(user): Cancel "this" if "scheduledDelegate" is cancelled.
685 }
686 return cancelled;
687 }
688
689 @Override
690 public long getDelay(TimeUnit unit) {
691 return scheduledDelegate.getDelay(unit);
692 }
693
694 @Override
695 public int compareTo(Delayed other) {
696 return scheduledDelegate.compareTo(other);
697 }
698 }
699
700 @GwtIncompatible // TODO
701 private static final class NeverSuccessfulListenableFutureTask
702 extends AbstractFuture.TrustedFuture<@Nullable Void> implements Runnable {
703 private final Runnable delegate;
704
705 public NeverSuccessfulListenableFutureTask(Runnable delegate) {
706 this.delegate = checkNotNull(delegate);
707 }
708
709 @Override
710 public void run() {
711 try {
712 delegate.run();
713 } catch (Throwable t) {
714 setException(t);
715 throw Throwables.propagate(t);
716 }
717 }
718
719 @Override
720 protected String pendingToString() {
721 return "task=[" + delegate + "]";
722 }
723 }
724 }
725
726 /*
727 * This following method is a modified version of one found in
728 * http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/test/tck/AbstractExecutorServiceTest.java?revision=1.30
729 * which contained the following notice:
730 *
731 * Written by Doug Lea with assistance from members of JCP JSR-166 Expert Group and released to
732 * the public domain, as explained at http://creativecommons.org/publicdomain/zero/1.0/
733 *
734 * Other contributors include Andrew Wright, Jeffrey Hayes, Pat Fisher, Mike Judd.
735 */
736
737 /**
738 * An implementation of {@link ExecutorService#invokeAny} for {@link ListeningExecutorService}
739 * implementations.
740 */
741 @GwtIncompatible
742 @ParametricNullness
743 static <T extends @Nullable Object> T invokeAnyImpl(
744 ListeningExecutorService executorService,
745 Collection<? extends Callable<T>> tasks,
746 boolean timed,
747 Duration timeout)
748 throws InterruptedException, ExecutionException, TimeoutException {
749 return invokeAnyImpl(
750 executorService, tasks, timed, toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
751 }
752
753 /**
754 * An implementation of {@link ExecutorService#invokeAny} for {@link ListeningExecutorService}
755 * implementations.
756 */
757 @SuppressWarnings("GoodTime") // should accept a java.time.Duration
758 @GwtIncompatible
759 @ParametricNullness
760 static <T extends @Nullable Object> T invokeAnyImpl(
761 ListeningExecutorService executorService,
762 Collection<? extends Callable<T>> tasks,
763 boolean timed,
764 long timeout,
765 TimeUnit unit)
766 throws InterruptedException, ExecutionException, TimeoutException {
767 checkNotNull(executorService);
768 checkNotNull(unit);
769 int ntasks = tasks.size();
770 checkArgument(ntasks > 0);
771 List<Future<T>> futures = Lists.newArrayListWithCapacity(ntasks);
772 BlockingQueue<Future<T>> futureQueue = Queues.newLinkedBlockingQueue();
773 long timeoutNanos = unit.toNanos(timeout);
774
775 // For efficiency, especially in executors with limited
776 // parallelism, check to see if previously submitted tasks are
777 // done before submitting more of them. This interleaving
778 // plus the exception mechanics account for messiness of main
779 // loop.
780
781 try {
782 // Record exceptions so that if we fail to obtain any
783 // result, we can throw the last exception we got.
784 ExecutionException ee = null;
785 long lastTime = timed ? System.nanoTime() : 0;
786 Iterator<? extends Callable<T>> it = tasks.iterator();
787
788 futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
789 --ntasks;
790 int active = 1;
791
792 while (true) {
793 Future<T> f = futureQueue.poll();
794 if (f == null) {
795 if (ntasks > 0) {
796 --ntasks;
797 futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
798 ++active;
799 } else if (active == 0) {
800 break;
801 } else if (timed) {
802 f = futureQueue.poll(timeoutNanos, TimeUnit.NANOSECONDS);
803 if (f == null) {
804 throw new TimeoutException();
805 }
806 long now = System.nanoTime();
807 timeoutNanos -= now - lastTime;
808 lastTime = now;
809 } else {
810 f = futureQueue.take();
811 }
812 }
813 if (f != null) {
814 --active;
815 try {
816 return f.get();
817 } catch (ExecutionException eex) {
818 ee = eex;
819 } catch (RuntimeException rex) {
820 ee = new ExecutionException(rex);
821 }
822 }
823 }
824
825 if (ee == null) {
826 ee = new ExecutionException(null);
827 }
828 throw ee;
829 } finally {
830 for (Future<T> f : futures) {
831 f.cancel(true);
832 }
833 }
834 }
835
836 /**
837 * Submits the task and adds a listener that adds the future to {@code queue} when it completes.
838 */
839 @GwtIncompatible // TODO
840 private static <T extends @Nullable Object> ListenableFuture<T> submitAndAddQueueListener(
841 ListeningExecutorService executorService,
842 Callable<T> task,
843 final BlockingQueue<Future<T>> queue) {
844 final ListenableFuture<T> future = executorService.submit(task);
845 future.addListener(
846 new Runnable() {
847 @Override
848 public void run() {
849 queue.add(future);
850 }
851 },
852 directExecutor());
853 return future;
854 }
855
856 /**
857 * Returns a default thread factory used to create new threads.
858 *
859 * <p>When running on AppEngine with access to <a
860 * href="https://cloud.google.com/appengine/docs/standard/java/javadoc/">AppEngine legacy
861 * APIs</a>, this method returns {@code ThreadManager.currentRequestThreadFactory()}. Otherwise,
862 * it returns {@link Executors#defaultThreadFactory()}.
863 *
864 * @since 14.0
865 */
866 @Beta
867 @GwtIncompatible // concurrency
868 public static ThreadFactory platformThreadFactory() {
869 if (!isAppEngineWithApiClasses()) {
870 return Executors.defaultThreadFactory();
871 }
872 try {
873 return (ThreadFactory)
874 Class.forName("com.google.appengine.api.ThreadManager")
875 .getMethod("currentRequestThreadFactory")
876 .invoke(null);
877 /*
878 * Do not merge the 3 catch blocks below. javac would infer a type of
879 * ReflectiveOperationException, which Animal Sniffer would reject. (Old versions of Android
880 * don't *seem* to mind, but there might be edge cases of which we're unaware.)
881 */
882 } catch (IllegalAccessException e) {
883 throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
884 } catch (ClassNotFoundException e) {
885 throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
886 } catch (NoSuchMethodException e) {
887 throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
888 } catch (InvocationTargetException e) {
889 throw Throwables.propagate(e.getCause());
890 }
891 }
892
893 @GwtIncompatible // TODO
894 private static boolean isAppEngineWithApiClasses() {
895 if (System.getProperty("com.google.appengine.runtime.environment") == null) {
896 return false;
897 }
898 try {
899 Class.forName("com.google.appengine.api.utils.SystemProperty");
900 } catch (ClassNotFoundException e) {
901 return false;
902 }
903 try {
904 // If the current environment is null, we're not inside AppEngine.
905 return Class.forName("com.google.apphosting.api.ApiProxy")
906 .getMethod("getCurrentEnvironment")
907 .invoke(null)
908 != null;
909 } catch (ClassNotFoundException e) {
910 // If ApiProxy doesn't exist, we're not on AppEngine at all.
911 return false;
912 } catch (InvocationTargetException e) {
913 // If ApiProxy throws an exception, we're not in a proper AppEngine environment.
914 return false;
915 } catch (IllegalAccessException e) {
916 // If the method isn't accessible, we're not on a supported version of AppEngine;
917 return false;
918 } catch (NoSuchMethodException e) {
919 // If the method doesn't exist, we're not on a supported version of AppEngine;
920 return false;
921 }
922 }
923
924 /**
925 * Creates a thread using {@link #platformThreadFactory}, and sets its name to {@code name} unless
926 * changing the name is forbidden by the security manager.
927 */
928 @GwtIncompatible // concurrency
929 static Thread newThread(String name, Runnable runnable) {
930 checkNotNull(name);
931 checkNotNull(runnable);
932 Thread result = platformThreadFactory().newThread(runnable);
933 try {
934 result.setName(name);
935 } catch (SecurityException e) {
936 // OK if we can't set the name in this environment.
937 }
938 return result;
939 }
940
941 // TODO(lukes): provide overloads for ListeningExecutorService? ListeningScheduledExecutorService?
942 // TODO(lukes): provide overloads that take constant strings? Function<Runnable, String>s to
943 // calculate names?
944
945 /**
946 * Creates an {@link Executor} that renames the {@link Thread threads} that its tasks run in.
947 *
948 * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
949 * right before each task is run. The renaming is best effort, if a {@link SecurityManager}
950 * prevents the renaming then it will be skipped but the tasks will still execute.
951 *
952 * @param executor The executor to decorate
953 * @param nameSupplier The source of names for each task
954 */
955 @GwtIncompatible // concurrency
956 static Executor renamingDecorator(final Executor executor, final Supplier<String> nameSupplier) {
957 checkNotNull(executor);
958 checkNotNull(nameSupplier);
959 return new Executor() {
960 @Override
961 public void execute(Runnable command) {
962 executor.execute(Callables.threadRenaming(command, nameSupplier));
963 }
964 };
965 }
966
967 /**
968 * Creates an {@link ExecutorService} that renames the {@link Thread threads} that its tasks run
969 * in.
970 *
971 * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
972 * right before each task is run. The renaming is best effort, if a {@link SecurityManager}
973 * prevents the renaming then it will be skipped but the tasks will still execute.
974 *
975 * @param service The executor to decorate
976 * @param nameSupplier The source of names for each task
977 */
978 @GwtIncompatible // concurrency
979 static ExecutorService renamingDecorator(
980 final ExecutorService service, final Supplier<String> nameSupplier) {
981 checkNotNull(service);
982 checkNotNull(nameSupplier);
983 return new WrappingExecutorService(service) {
984 @Override
985 protected <T extends @Nullable Object> Callable<T> wrapTask(Callable<T> callable) {
986 return Callables.threadRenaming(callable, nameSupplier);
987 }
988
989 @Override
990 protected Runnable wrapTask(Runnable command) {
991 return Callables.threadRenaming(command, nameSupplier);
992 }
993 };
994 }
995
996 /**
997 * Creates a {@link ScheduledExecutorService} that renames the {@link Thread threads} that its
998 * tasks run in.
999 *
1000 * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
1001 * right before each task is run. The renaming is best effort, if a {@link SecurityManager}
1002 * prevents the renaming then it will be skipped but the tasks will still execute.
1003 *
1004 * @param service The executor to decorate
1005 * @param nameSupplier The source of names for each task
1006 */
1007 @GwtIncompatible // concurrency
1008 static ScheduledExecutorService renamingDecorator(
1009 final ScheduledExecutorService service, final Supplier<String> nameSupplier) {
1010 checkNotNull(service);
1011 checkNotNull(nameSupplier);
1012 return new WrappingScheduledExecutorService(service) {
1013 @Override
1014 protected <T extends @Nullable Object> Callable<T> wrapTask(Callable<T> callable) {
1015 return Callables.threadRenaming(callable, nameSupplier);
1016 }
1017
1018 @Override
1019 protected Runnable wrapTask(Runnable command) {
1020 return Callables.threadRenaming(command, nameSupplier);
1021 }
1022 };
1023 }
1024
1025 /**
1026 * Shuts down the given executor service gradually, first disabling new submissions and later, if
1027 * necessary, cancelling remaining tasks.
1028 *
1029 * <p>The method takes the following steps:
1030 *
1031 * <ol>
1032 * <li>calls {@link ExecutorService#shutdown()}, disabling acceptance of new submitted tasks.
1033 * <li>awaits executor service termination for half of the specified timeout.
1034 * <li>if the timeout expires, it calls {@link ExecutorService#shutdownNow()}, cancelling
1035 * pending tasks and interrupting running tasks.
1036 * <li>awaits executor service termination for the other half of the specified timeout.
1037 * </ol>
1038 *
1039 * <p>If, at any step of the process, the calling thread is interrupted, the method calls {@link
1040 * ExecutorService#shutdownNow()} and returns.
1041 *
1042 * @param service the {@code ExecutorService} to shut down
1043 * @param timeout the maximum time to wait for the {@code ExecutorService} to terminate
1044 * @return {@code true} if the {@code ExecutorService} was terminated successfully, {@code false}
1045 * if the call timed out or was interrupted
1046 * @since 28.0
1047 */
1048 @Beta
1049 @CanIgnoreReturnValue
1050 @GwtIncompatible // java.time.Duration
1051 public static boolean shutdownAndAwaitTermination(ExecutorService service, Duration timeout) {
1052 return shutdownAndAwaitTermination(service, toNanosSaturated(timeout), TimeUnit.NANOSECONDS);
1053 }
1054
1055 /**
1056 * Shuts down the given executor service gradually, first disabling new submissions and later, if
1057 * necessary, cancelling remaining tasks.
1058 *
1059 * <p>The method takes the following steps:
1060 *
1061 * <ol>
1062 * <li>calls {@link ExecutorService#shutdown()}, disabling acceptance of new submitted tasks.
1063 * <li>awaits executor service termination for half of the specified timeout.
1064 * <li>if the timeout expires, it calls {@link ExecutorService#shutdownNow()}, cancelling
1065 * pending tasks and interrupting running tasks.
1066 * <li>awaits executor service termination for the other half of the specified timeout.
1067 * </ol>
1068 *
1069 * <p>If, at any step of the process, the calling thread is interrupted, the method calls {@link
1070 * ExecutorService#shutdownNow()} and returns.
1071 *
1072 * @param service the {@code ExecutorService} to shut down
1073 * @param timeout the maximum time to wait for the {@code ExecutorService} to terminate
1074 * @param unit the time unit of the timeout argument
1075 * @return {@code true} if the {@code ExecutorService} was terminated successfully, {@code false}
1076 * if the call timed out or was interrupted
1077 * @since 17.0
1078 */
1079 @Beta
1080 @CanIgnoreReturnValue
1081 @GwtIncompatible // concurrency
1082 @SuppressWarnings("GoodTime") // should accept a java.time.Duration
1083 public static boolean shutdownAndAwaitTermination(
1084 ExecutorService service, long timeout, TimeUnit unit) {
1085 long halfTimeoutNanos = unit.toNanos(timeout) / 2;
1086 // Disable new tasks from being submitted
1087 service.shutdown();
1088 try {
1089 // Wait for half the duration of the timeout for existing tasks to terminate
1090 if (!service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS)) {
1091 // Cancel currently executing tasks
1092 service.shutdownNow();
1093 // Wait the other half of the timeout for tasks to respond to being cancelled
1094 service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS);
1095 }
1096 } catch (InterruptedException ie) {
1097 // Preserve interrupt status
1098 Thread.currentThread().interrupt();
1099 // (Re-)Cancel if current thread also interrupted
1100 service.shutdownNow();
1101 }
1102 return service.isTerminated();
1103 }
1104
1105 /**
1106 * Returns an Executor that will propagate {@link RejectedExecutionException} from the delegate
1107 * executor to the given {@code future}.
1108 *
1109 * <p>Note, the returned executor can only be used once.
1110 */
1111 static Executor rejectionPropagatingExecutor(
1112 final Executor delegate, final AbstractFuture<?> future) {
1113 checkNotNull(delegate);
1114 checkNotNull(future);
1115 if (delegate == directExecutor()) {
1116 // directExecutor() cannot throw RejectedExecutionException
1117 return delegate;
1118 }
1119 return new Executor() {
1120 @Override
1121 public void execute(Runnable command) {
1122 try {
1123 delegate.execute(command);
1124 } catch (RejectedExecutionException e) {
1125 future.setException(e);
1126 }
1127 }
1128 };
1129 }
1130 }