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