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