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; 054import org.checkerframework.checker.nullness.qual.Nullable; 055 056/** 057 * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link ExecutorService}, 058 * and {@link java.util.concurrent.ThreadFactory}. 059 * 060 * @author Eric Fellheimer 061 * @author Kyle Littlefield 062 * @author Justin Mahoney 063 * @since 3.0 064 */ 065@GwtCompatible(emulated = true) 066@ElementTypesAreNonnullByDefault 067public final class MoreExecutors { 068 private MoreExecutors() {} 069 070 /** 071 * Converts the given ThreadPoolExecutor into an ExecutorService that exits when the application 072 * is complete. It does so by using daemon threads and adding a shutdown hook to wait for their 073 * completion. 074 * 075 * <p>This is mainly for fixed thread pools. See {@link Executors#newFixedThreadPool(int)}. 076 * 077 * @param executor the executor to modify to make sure it exits when the application is finished 078 * @param terminationTimeout how long to wait for the executor to finish before terminating the 079 * JVM 080 * @return an unmodifiable version of the input which will not hang the JVM 081 * @since 28.0 082 */ 083 @Beta 084 @GwtIncompatible // TODO 085 public static ExecutorService getExitingExecutorService( 086 ThreadPoolExecutor executor, Duration terminationTimeout) { 087 return getExitingExecutorService( 088 executor, toNanosSaturated(terminationTimeout), TimeUnit.NANOSECONDS); 089 } 090 091 /** 092 * Converts the given ThreadPoolExecutor into an ExecutorService that exits when the application 093 * is complete. It does so by using daemon threads and adding a shutdown hook to wait for their 094 * completion. 095 * 096 * <p>This is mainly for fixed thread pools. See {@link Executors#newFixedThreadPool(int)}. 097 * 098 * @param executor the executor to modify to make sure it exits when the application is finished 099 * @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>When a {@code ListenableFuture} listener is registered to run under {@code 444 * directExecutor}, the listener can execute in any of three possible threads: 445 * <ol> 446 * <li>When a thread attaches a listener to a {@code ListenableFuture} that's already 447 * complete, the listener runs immediately in that thread. 448 * <li>When a thread attaches a listener to a {@code ListenableFuture} that's 449 * <em>in</em>complete and the {@code ListenableFuture} later completes normally, the 450 * listener runs in the the thread that completes the {@code ListenableFuture}. 451 * <li>When a listener is attached to a {@code ListenableFuture} and the {@code 452 * ListenableFuture} gets cancelled, the listener runs immediately in the the thread 453 * that cancelled the {@code Future}. 454 * </ol> 455 * Given all these possibilities, it is frequently possible for listeners to execute in UI 456 * threads, RPC network threads, or other latency-sensitive threads. In those cases, slow 457 * listeners can harm responsiveness, slow the system as a whole, or worse. (See also the 458 * note about locking below.) 459 * <li>If many tasks will be triggered by the same event, one heavyweight task may delay other 460 * tasks -- even tasks that are not themselves {@code directExecutor} tasks. 461 * <li>If many such tasks are chained together (such as with {@code 462 * future.transform(...).transform(...).transform(...)....}), they may overflow the stack. 463 * (In simple cases, callers can avoid this by registering all tasks with the same {@link 464 * MoreExecutors#newSequentialExecutor} wrapper around {@code directExecutor()}. More 465 * complex cases may require using thread pools or making deeper changes.) 466 * <li>If an exception propagates out of a {@code Runnable}, it is not necessarily seen by any 467 * {@code UncaughtExceptionHandler} for the thread. For example, if the callback passed to 468 * {@link Futures#addCallback} throws an exception, that exception will be typically be 469 * logged by the {@link ListenableFuture} implementation, even if the thread is configured 470 * to do something different. In other cases, no code will catch the exception, and it may 471 * terminate whichever thread happens to trigger the execution. 472 * </ul> 473 * 474 * A specific warning about locking: Code that executes user-supplied tasks, such as {@code 475 * ListenableFuture} listeners, should take care not to do so while holding a lock. Additionally, 476 * as a further line of defense, prefer not to perform any locking inside a task that will be run 477 * under {@code directExecutor}: Not only might the wait for a lock be long, but if the running 478 * thread was holding a lock, the listener may deadlock or break lock isolation. 479 * 480 * <p>This instance is equivalent to: 481 * 482 * <pre>{@code 483 * final class DirectExecutor implements Executor { 484 * public void execute(Runnable r) { 485 * r.run(); 486 * } 487 * } 488 * }</pre> 489 * 490 * <p>This should be preferred to {@link #newDirectExecutorService()} because implementing the 491 * {@link ExecutorService} subinterface necessitates significant performance overhead. 492 * 493 * @since 18.0 494 */ 495 public static Executor directExecutor() { 496 return DirectExecutor.INSTANCE; 497 } 498 499 /** 500 * Returns an {@link Executor} that runs each task executed sequentially, such that no two tasks 501 * are running concurrently. 502 * 503 * <p>{@linkplain Executor#execute executed} tasks have a happens-before order as defined in the 504 * Java Language Specification. Tasks execute with the same happens-before order that the function 505 * calls to {@link Executor#execute `execute()`} that submitted those tasks had. 506 * 507 * <p>The executor uses {@code delegate} in order to {@link Executor#execute execute} each task in 508 * turn, and does not create any threads of its own. 509 * 510 * <p>After execution begins on a thread from the {@code delegate} {@link Executor}, tasks are 511 * polled and executed from a task queue until there are no more tasks. The thread will not be 512 * released until there are no more tasks to run. 513 * 514 * <p>If a task is submitted while a thread is executing tasks from the task queue, the thread 515 * will not be released until that submitted task is also complete. 516 * 517 * <p>If a task is {@linkplain Thread#interrupt interrupted} while a task is running: 518 * 519 * <ol> 520 * <li>execution will not stop until the task queue is empty. 521 * <li>tasks will begin execution with the thread marked as not interrupted - any interruption 522 * applies only to the task that was running at the point of interruption. 523 * <li>if the thread was interrupted before the SequentialExecutor's worker begins execution, 524 * the interrupt will be restored to the thread after it completes so that its {@code 525 * delegate} Executor may process the interrupt. 526 * <li>subtasks are run with the thread uninterrupted and interrupts received during execution 527 * of a task are ignored. 528 * </ol> 529 * 530 * <p>{@code RuntimeException}s thrown by tasks are simply logged and the executor keeps trucking. 531 * If an {@code Error} is thrown, the error will propagate and execution will stop until the next 532 * time a task is submitted. 533 * 534 * <p>When an {@code Error} is thrown by an executed task, previously submitted tasks may never 535 * run. An attempt will be made to restart execution on the next call to {@code execute}. If the 536 * {@code delegate} has begun to reject execution, the previously submitted tasks may never run, 537 * despite not throwing a RejectedExecutionException synchronously with the call to {@code 538 * execute}. If this behaviour is problematic, use an Executor with a single thread (e.g. {@link 539 * Executors#newSingleThreadExecutor}). 540 * 541 * @since 23.3 (since 23.1 as {@code sequentialExecutor}) 542 */ 543 @GwtIncompatible 544 public static Executor newSequentialExecutor(Executor delegate) { 545 return new SequentialExecutor(delegate); 546 } 547 548 /** 549 * Creates an {@link ExecutorService} whose {@code submit} and {@code invokeAll} methods submit 550 * {@link ListenableFutureTask} instances to the given delegate executor. Those methods, as well 551 * as {@code execute} and {@code invokeAny}, are implemented in terms of calls to {@code 552 * delegate.execute}. All other methods are forwarded unchanged to the delegate. This implies that 553 * the returned {@code ListeningExecutorService} never calls the delegate's {@code submit}, {@code 554 * invokeAll}, and {@code invokeAny} methods, so any special handling of tasks must be implemented 555 * in the delegate's {@code execute} method or by wrapping the returned {@code 556 * ListeningExecutorService}. 557 * 558 * <p>If the delegate executor was already an instance of {@code ListeningExecutorService}, it is 559 * returned untouched, and the rest of this documentation does not apply. 560 * 561 * @since 10.0 562 */ 563 @GwtIncompatible // TODO 564 public static ListeningExecutorService listeningDecorator(ExecutorService delegate) { 565 return (delegate instanceof ListeningExecutorService) 566 ? (ListeningExecutorService) delegate 567 : (delegate instanceof ScheduledExecutorService) 568 ? new ScheduledListeningDecorator((ScheduledExecutorService) delegate) 569 : new ListeningDecorator(delegate); 570 } 571 572 /** 573 * Creates a {@link ScheduledExecutorService} whose {@code submit} and {@code invokeAll} methods 574 * submit {@link ListenableFutureTask} instances to the given delegate executor. Those methods, as 575 * well as {@code execute} and {@code invokeAny}, are implemented in terms of calls to {@code 576 * delegate.execute}. All other methods are forwarded unchanged to the delegate. This implies that 577 * the returned {@code ListeningScheduledExecutorService} never calls the delegate's {@code 578 * submit}, {@code invokeAll}, and {@code invokeAny} methods, so any special handling of tasks 579 * must be implemented in the delegate's {@code execute} method or by wrapping the returned {@code 580 * ListeningScheduledExecutorService}. 581 * 582 * <p>If the delegate executor was already an instance of {@code 583 * ListeningScheduledExecutorService}, it is returned untouched, and the rest of this 584 * documentation does not apply. 585 * 586 * @since 10.0 587 */ 588 @GwtIncompatible // TODO 589 public static ListeningScheduledExecutorService listeningDecorator( 590 ScheduledExecutorService delegate) { 591 return (delegate instanceof ListeningScheduledExecutorService) 592 ? (ListeningScheduledExecutorService) delegate 593 : new ScheduledListeningDecorator(delegate); 594 } 595 596 @GwtIncompatible // TODO 597 private static class ListeningDecorator extends AbstractListeningExecutorService { 598 private final ExecutorService delegate; 599 600 ListeningDecorator(ExecutorService delegate) { 601 this.delegate = checkNotNull(delegate); 602 } 603 604 @Override 605 public final boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { 606 return delegate.awaitTermination(timeout, unit); 607 } 608 609 @Override 610 public final boolean isShutdown() { 611 return delegate.isShutdown(); 612 } 613 614 @Override 615 public final boolean isTerminated() { 616 return delegate.isTerminated(); 617 } 618 619 @Override 620 public final void shutdown() { 621 delegate.shutdown(); 622 } 623 624 @Override 625 public final List<Runnable> shutdownNow() { 626 return delegate.shutdownNow(); 627 } 628 629 @Override 630 public final void execute(Runnable command) { 631 delegate.execute(command); 632 } 633 634 @Override 635 public final String toString() { 636 return super.toString() + "[" + delegate + "]"; 637 } 638 } 639 640 @GwtIncompatible // TODO 641 private static final class ScheduledListeningDecorator extends ListeningDecorator 642 implements ListeningScheduledExecutorService { 643 @SuppressWarnings("hiding") 644 final ScheduledExecutorService delegate; 645 646 ScheduledListeningDecorator(ScheduledExecutorService delegate) { 647 super(delegate); 648 this.delegate = checkNotNull(delegate); 649 } 650 651 @Override 652 public ListenableScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { 653 TrustedListenableFutureTask<@Nullable Void> task = 654 TrustedListenableFutureTask.create(command, null); 655 ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit); 656 return new ListenableScheduledTask<@Nullable Void>(task, scheduled); 657 } 658 659 @Override 660 public <V extends @Nullable Object> ListenableScheduledFuture<V> schedule( 661 Callable<V> callable, long delay, TimeUnit unit) { 662 TrustedListenableFutureTask<V> task = TrustedListenableFutureTask.create(callable); 663 ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit); 664 return new ListenableScheduledTask<V>(task, scheduled); 665 } 666 667 @Override 668 public ListenableScheduledFuture<?> scheduleAtFixedRate( 669 Runnable command, long initialDelay, long period, TimeUnit unit) { 670 NeverSuccessfulListenableFutureTask task = new NeverSuccessfulListenableFutureTask(command); 671 ScheduledFuture<?> scheduled = delegate.scheduleAtFixedRate(task, initialDelay, period, unit); 672 return new ListenableScheduledTask<@Nullable Void>(task, scheduled); 673 } 674 675 @Override 676 public ListenableScheduledFuture<?> scheduleWithFixedDelay( 677 Runnable command, long initialDelay, long delay, TimeUnit unit) { 678 NeverSuccessfulListenableFutureTask task = new NeverSuccessfulListenableFutureTask(command); 679 ScheduledFuture<?> scheduled = 680 delegate.scheduleWithFixedDelay(task, initialDelay, delay, unit); 681 return new ListenableScheduledTask<@Nullable Void>(task, scheduled); 682 } 683 684 private static final class ListenableScheduledTask<V extends @Nullable Object> 685 extends SimpleForwardingListenableFuture<V> implements ListenableScheduledFuture<V> { 686 687 private final ScheduledFuture<?> scheduledDelegate; 688 689 public ListenableScheduledTask( 690 ListenableFuture<V> listenableDelegate, ScheduledFuture<?> scheduledDelegate) { 691 super(listenableDelegate); 692 this.scheduledDelegate = scheduledDelegate; 693 } 694 695 @Override 696 public boolean cancel(boolean mayInterruptIfRunning) { 697 boolean cancelled = super.cancel(mayInterruptIfRunning); 698 if (cancelled) { 699 // Unless it is cancelled, the delegate may continue being scheduled 700 scheduledDelegate.cancel(mayInterruptIfRunning); 701 702 // TODO(user): Cancel "this" if "scheduledDelegate" is cancelled. 703 } 704 return cancelled; 705 } 706 707 @Override 708 public long getDelay(TimeUnit unit) { 709 return scheduledDelegate.getDelay(unit); 710 } 711 712 @Override 713 public int compareTo(Delayed other) { 714 return scheduledDelegate.compareTo(other); 715 } 716 } 717 718 @GwtIncompatible // TODO 719 private static final class NeverSuccessfulListenableFutureTask 720 extends AbstractFuture.TrustedFuture<@Nullable Void> implements Runnable { 721 private final Runnable delegate; 722 723 public NeverSuccessfulListenableFutureTask(Runnable delegate) { 724 this.delegate = checkNotNull(delegate); 725 } 726 727 @Override 728 public void run() { 729 try { 730 delegate.run(); 731 } catch (Throwable t) { 732 setException(t); 733 throw Throwables.propagate(t); 734 } 735 } 736 737 @Override 738 protected String pendingToString() { 739 return "task=[" + delegate + "]"; 740 } 741 } 742 } 743 744 /* 745 * This following method is a modified version of one found in 746 * http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/test/tck/AbstractExecutorServiceTest.java?revision=1.30 747 * which contained the following notice: 748 * 749 * Written by Doug Lea with assistance from members of JCP JSR-166 Expert Group and released to 750 * the public domain, as explained at http://creativecommons.org/publicdomain/zero/1.0/ 751 * 752 * Other contributors include Andrew Wright, Jeffrey Hayes, Pat Fisher, Mike Judd. 753 */ 754 755 /** 756 * An implementation of {@link ExecutorService#invokeAny} for {@link ListeningExecutorService} 757 * implementations. 758 */ 759 @GwtIncompatible 760 @ParametricNullness 761 static <T extends @Nullable Object> T invokeAnyImpl( 762 ListeningExecutorService executorService, 763 Collection<? extends Callable<T>> tasks, 764 boolean timed, 765 Duration timeout) 766 throws InterruptedException, ExecutionException, TimeoutException { 767 return invokeAnyImpl( 768 executorService, tasks, timed, toNanosSaturated(timeout), TimeUnit.NANOSECONDS); 769 } 770 771 /** 772 * An implementation of {@link ExecutorService#invokeAny} for {@link ListeningExecutorService} 773 * implementations. 774 */ 775 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 776 @GwtIncompatible 777 @ParametricNullness 778 static <T extends @Nullable Object> T invokeAnyImpl( 779 ListeningExecutorService executorService, 780 Collection<? extends Callable<T>> tasks, 781 boolean timed, 782 long timeout, 783 TimeUnit unit) 784 throws InterruptedException, ExecutionException, TimeoutException { 785 checkNotNull(executorService); 786 checkNotNull(unit); 787 int ntasks = tasks.size(); 788 checkArgument(ntasks > 0); 789 List<Future<T>> futures = Lists.newArrayListWithCapacity(ntasks); 790 BlockingQueue<Future<T>> futureQueue = Queues.newLinkedBlockingQueue(); 791 long timeoutNanos = unit.toNanos(timeout); 792 793 // For efficiency, especially in executors with limited 794 // parallelism, check to see if previously submitted tasks are 795 // done before submitting more of them. This interleaving 796 // plus the exception mechanics account for messiness of main 797 // loop. 798 799 try { 800 // Record exceptions so that if we fail to obtain any 801 // result, we can throw the last exception we got. 802 ExecutionException ee = null; 803 long lastTime = timed ? System.nanoTime() : 0; 804 Iterator<? extends Callable<T>> it = tasks.iterator(); 805 806 futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue)); 807 --ntasks; 808 int active = 1; 809 810 while (true) { 811 Future<T> f = futureQueue.poll(); 812 if (f == null) { 813 if (ntasks > 0) { 814 --ntasks; 815 futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue)); 816 ++active; 817 } else if (active == 0) { 818 break; 819 } else if (timed) { 820 f = futureQueue.poll(timeoutNanos, TimeUnit.NANOSECONDS); 821 if (f == null) { 822 throw new TimeoutException(); 823 } 824 long now = System.nanoTime(); 825 timeoutNanos -= now - lastTime; 826 lastTime = now; 827 } else { 828 f = futureQueue.take(); 829 } 830 } 831 if (f != null) { 832 --active; 833 try { 834 return f.get(); 835 } catch (ExecutionException eex) { 836 ee = eex; 837 } catch (RuntimeException rex) { 838 ee = new ExecutionException(rex); 839 } 840 } 841 } 842 843 if (ee == null) { 844 ee = new ExecutionException(null); 845 } 846 throw ee; 847 } finally { 848 for (Future<T> f : futures) { 849 f.cancel(true); 850 } 851 } 852 } 853 854 /** 855 * Submits the task and adds a listener that adds the future to {@code queue} when it completes. 856 */ 857 @GwtIncompatible // TODO 858 private static <T extends @Nullable Object> ListenableFuture<T> submitAndAddQueueListener( 859 ListeningExecutorService executorService, 860 Callable<T> task, 861 final BlockingQueue<Future<T>> queue) { 862 final ListenableFuture<T> future = executorService.submit(task); 863 future.addListener( 864 new Runnable() { 865 @Override 866 public void run() { 867 queue.add(future); 868 } 869 }, 870 directExecutor()); 871 return future; 872 } 873 874 /** 875 * Returns a default thread factory used to create new threads. 876 * 877 * <p>When running on AppEngine with access to <a 878 * href="https://cloud.google.com/appengine/docs/standard/java/javadoc/">AppEngine legacy 879 * APIs</a>, this method returns {@code ThreadManager.currentRequestThreadFactory()}. Otherwise, 880 * it returns {@link Executors#defaultThreadFactory()}. 881 * 882 * @since 14.0 883 */ 884 @Beta 885 @GwtIncompatible // concurrency 886 public static ThreadFactory platformThreadFactory() { 887 if (!isAppEngineWithApiClasses()) { 888 return Executors.defaultThreadFactory(); 889 } 890 try { 891 return (ThreadFactory) 892 Class.forName("com.google.appengine.api.ThreadManager") 893 .getMethod("currentRequestThreadFactory") 894 .invoke(null); 895 /* 896 * Do not merge the 3 catch blocks below. javac would infer a type of 897 * ReflectiveOperationException, which Animal Sniffer would reject. (Old versions of Android 898 * don't *seem* to mind, but there might be edge cases of which we're unaware.) 899 */ 900 } catch (IllegalAccessException e) { 901 throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e); 902 } catch (ClassNotFoundException e) { 903 throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e); 904 } catch (NoSuchMethodException e) { 905 throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e); 906 } catch (InvocationTargetException e) { 907 throw Throwables.propagate(e.getCause()); 908 } 909 } 910 911 @GwtIncompatible // TODO 912 private static boolean isAppEngineWithApiClasses() { 913 if (System.getProperty("com.google.appengine.runtime.environment") == null) { 914 return false; 915 } 916 try { 917 Class.forName("com.google.appengine.api.utils.SystemProperty"); 918 } catch (ClassNotFoundException e) { 919 return false; 920 } 921 try { 922 // If the current environment is null, we're not inside AppEngine. 923 return Class.forName("com.google.apphosting.api.ApiProxy") 924 .getMethod("getCurrentEnvironment") 925 .invoke(null) 926 != null; 927 } catch (ClassNotFoundException e) { 928 // If ApiProxy doesn't exist, we're not on AppEngine at all. 929 return false; 930 } catch (InvocationTargetException e) { 931 // If ApiProxy throws an exception, we're not in a proper AppEngine environment. 932 return false; 933 } catch (IllegalAccessException e) { 934 // If the method isn't accessible, we're not on a supported version of AppEngine; 935 return false; 936 } catch (NoSuchMethodException e) { 937 // If the method doesn't exist, we're not on a supported version of AppEngine; 938 return false; 939 } 940 } 941 942 /** 943 * Creates a thread using {@link #platformThreadFactory}, and sets its name to {@code name} unless 944 * changing the name is forbidden by the security manager. 945 */ 946 @GwtIncompatible // concurrency 947 static Thread newThread(String name, Runnable runnable) { 948 checkNotNull(name); 949 checkNotNull(runnable); 950 Thread result = platformThreadFactory().newThread(runnable); 951 try { 952 result.setName(name); 953 } catch (SecurityException e) { 954 // OK if we can't set the name in this environment. 955 } 956 return result; 957 } 958 959 // TODO(lukes): provide overloads for ListeningExecutorService? ListeningScheduledExecutorService? 960 // TODO(lukes): provide overloads that take constant strings? Function<Runnable, String>s to 961 // calculate names? 962 963 /** 964 * Creates an {@link Executor} that renames the {@link Thread threads} that its tasks run 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 * @param executor The executor to decorate 971 * @param nameSupplier The source of names for each task 972 */ 973 @GwtIncompatible // concurrency 974 static Executor renamingDecorator(final Executor executor, final Supplier<String> nameSupplier) { 975 checkNotNull(executor); 976 checkNotNull(nameSupplier); 977 return new Executor() { 978 @Override 979 public void execute(Runnable command) { 980 executor.execute(Callables.threadRenaming(command, nameSupplier)); 981 } 982 }; 983 } 984 985 /** 986 * Creates an {@link ExecutorService} that renames the {@link Thread threads} that its tasks run 987 * in. 988 * 989 * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed 990 * right before each task is run. The renaming is best effort, if a {@link SecurityManager} 991 * prevents the renaming then it will be skipped but the tasks will still execute. 992 * 993 * @param service The executor to decorate 994 * @param nameSupplier The source of names for each task 995 */ 996 @GwtIncompatible // concurrency 997 static ExecutorService renamingDecorator( 998 final ExecutorService service, final Supplier<String> nameSupplier) { 999 checkNotNull(service); 1000 checkNotNull(nameSupplier); 1001 return new WrappingExecutorService(service) { 1002 @Override 1003 protected <T extends @Nullable Object> Callable<T> wrapTask(Callable<T> callable) { 1004 return Callables.threadRenaming(callable, nameSupplier); 1005 } 1006 1007 @Override 1008 protected Runnable wrapTask(Runnable command) { 1009 return Callables.threadRenaming(command, nameSupplier); 1010 } 1011 }; 1012 } 1013 1014 /** 1015 * Creates a {@link ScheduledExecutorService} that renames the {@link Thread threads} that its 1016 * tasks run in. 1017 * 1018 * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed 1019 * right before each task is run. The renaming is best effort, if a {@link SecurityManager} 1020 * prevents the renaming then it will be skipped but the tasks will still execute. 1021 * 1022 * @param service The executor to decorate 1023 * @param nameSupplier The source of names for each task 1024 */ 1025 @GwtIncompatible // concurrency 1026 static ScheduledExecutorService renamingDecorator( 1027 final ScheduledExecutorService service, final Supplier<String> nameSupplier) { 1028 checkNotNull(service); 1029 checkNotNull(nameSupplier); 1030 return new WrappingScheduledExecutorService(service) { 1031 @Override 1032 protected <T extends @Nullable Object> Callable<T> wrapTask(Callable<T> callable) { 1033 return Callables.threadRenaming(callable, nameSupplier); 1034 } 1035 1036 @Override 1037 protected Runnable wrapTask(Runnable command) { 1038 return Callables.threadRenaming(command, nameSupplier); 1039 } 1040 }; 1041 } 1042 1043 /** 1044 * Shuts down the given executor service gradually, first disabling new submissions and later, if 1045 * necessary, cancelling remaining tasks. 1046 * 1047 * <p>The method takes the following steps: 1048 * 1049 * <ol> 1050 * <li>calls {@link ExecutorService#shutdown()}, disabling acceptance of new submitted tasks. 1051 * <li>awaits executor service termination for half of the specified timeout. 1052 * <li>if the timeout expires, it calls {@link ExecutorService#shutdownNow()}, cancelling 1053 * pending tasks and interrupting running tasks. 1054 * <li>awaits executor service termination for the other half of the specified timeout. 1055 * </ol> 1056 * 1057 * <p>If, at any step of the process, the calling thread is interrupted, the method calls {@link 1058 * ExecutorService#shutdownNow()} and returns. 1059 * 1060 * @param service the {@code ExecutorService} to shut down 1061 * @param timeout the maximum time to wait for the {@code ExecutorService} to terminate 1062 * @return {@code true} if the {@code ExecutorService} was terminated successfully, {@code false} 1063 * if the call timed out or was interrupted 1064 * @since 28.0 1065 */ 1066 @Beta 1067 @CanIgnoreReturnValue 1068 @GwtIncompatible // java.time.Duration 1069 public static boolean shutdownAndAwaitTermination(ExecutorService service, Duration timeout) { 1070 return shutdownAndAwaitTermination(service, toNanosSaturated(timeout), TimeUnit.NANOSECONDS); 1071 } 1072 1073 /** 1074 * Shuts down the given executor service gradually, first disabling new submissions and later, if 1075 * necessary, cancelling remaining tasks. 1076 * 1077 * <p>The method takes the following steps: 1078 * 1079 * <ol> 1080 * <li>calls {@link ExecutorService#shutdown()}, disabling acceptance of new submitted tasks. 1081 * <li>awaits executor service termination for half of the specified timeout. 1082 * <li>if the timeout expires, it calls {@link ExecutorService#shutdownNow()}, cancelling 1083 * pending tasks and interrupting running tasks. 1084 * <li>awaits executor service termination for the other half of the specified timeout. 1085 * </ol> 1086 * 1087 * <p>If, at any step of the process, the calling thread is interrupted, the method calls {@link 1088 * ExecutorService#shutdownNow()} and returns. 1089 * 1090 * @param service the {@code ExecutorService} to shut down 1091 * @param timeout the maximum time to wait for the {@code ExecutorService} to terminate 1092 * @param unit the time unit of the timeout argument 1093 * @return {@code true} if the {@code ExecutorService} was terminated successfully, {@code false} 1094 * if the call timed out or was interrupted 1095 * @since 17.0 1096 */ 1097 @Beta 1098 @CanIgnoreReturnValue 1099 @GwtIncompatible // concurrency 1100 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 1101 public static boolean shutdownAndAwaitTermination( 1102 ExecutorService service, long timeout, TimeUnit unit) { 1103 long halfTimeoutNanos = unit.toNanos(timeout) / 2; 1104 // Disable new tasks from being submitted 1105 service.shutdown(); 1106 try { 1107 // Wait for half the duration of the timeout for existing tasks to terminate 1108 if (!service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS)) { 1109 // Cancel currently executing tasks 1110 service.shutdownNow(); 1111 // Wait the other half of the timeout for tasks to respond to being cancelled 1112 service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS); 1113 } 1114 } catch (InterruptedException ie) { 1115 // Preserve interrupt status 1116 Thread.currentThread().interrupt(); 1117 // (Re-)Cancel if current thread also interrupted 1118 service.shutdownNow(); 1119 } 1120 return service.isTerminated(); 1121 } 1122 1123 /** 1124 * Returns an Executor that will propagate {@link RejectedExecutionException} from the delegate 1125 * executor to the given {@code future}. 1126 * 1127 * <p>Note, the returned executor can only be used once. 1128 */ 1129 static Executor rejectionPropagatingExecutor( 1130 final Executor delegate, final AbstractFuture<?> future) { 1131 checkNotNull(delegate); 1132 checkNotNull(future); 1133 if (delegate == directExecutor()) { 1134 // directExecutor() cannot throw RejectedExecutionException 1135 return delegate; 1136 } 1137 return new Executor() { 1138 @Override 1139 public void execute(Runnable command) { 1140 try { 1141 delegate.execute(command); 1142 } catch (RejectedExecutionException e) { 1143 future.setException(e); 1144 } 1145 } 1146 }; 1147 } 1148}