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