001/* 002 * Copyright (C) 2011 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.Futures.immediateCancelledFuture; 020import static com.google.common.util.concurrent.MoreExecutors.directExecutor; 021import static com.google.common.util.concurrent.Platform.restoreInterruptIfIsInterruptedException; 022import static java.util.Objects.requireNonNull; 023 024import com.google.common.annotations.GwtIncompatible; 025import com.google.common.base.Supplier; 026import com.google.errorprone.annotations.CanIgnoreReturnValue; 027import com.google.errorprone.annotations.concurrent.GuardedBy; 028import com.google.j2objc.annotations.WeakOuter; 029import java.util.concurrent.Callable; 030import java.util.concurrent.Executor; 031import java.util.concurrent.Executors; 032import java.util.concurrent.Future; 033import java.util.concurrent.ScheduledExecutorService; 034import java.util.concurrent.ScheduledFuture; 035import java.util.concurrent.ThreadFactory; 036import java.util.concurrent.TimeUnit; 037import java.util.concurrent.TimeoutException; 038import java.util.concurrent.locks.ReentrantLock; 039import java.util.logging.Level; 040import java.util.logging.Logger; 041import javax.annotation.CheckForNull; 042import org.checkerframework.checker.nullness.qual.Nullable; 043 044/** 045 * Base class for services that can implement {@link #startUp} and {@link #shutDown} but while in 046 * the "running" state need to perform a periodic task. Subclasses can implement {@link #startUp}, 047 * {@link #shutDown} and also a {@link #runOneIteration} method that will be executed periodically. 048 * 049 * <p>This class uses the {@link ScheduledExecutorService} returned from {@link #executor} to run 050 * the {@link #startUp} and {@link #shutDown} methods and also uses that service to schedule the 051 * {@link #runOneIteration} that will be executed periodically as specified by its {@link 052 * Scheduler}. When this service is asked to stop via {@link #stopAsync} it will cancel the periodic 053 * task (but not interrupt it) and wait for it to stop before running the {@link #shutDown} method. 054 * 055 * <p>Subclasses are guaranteed that the life cycle methods ({@link #runOneIteration}, {@link 056 * #startUp} and {@link #shutDown}) will never run concurrently. Notably, if any execution of {@link 057 * #runOneIteration} takes longer than its schedule defines, then subsequent executions may start 058 * late. Also, all life cycle methods are executed with a lock held, so subclasses can safely modify 059 * shared state without additional synchronization necessary for visibility to later executions of 060 * the life cycle methods. 061 * 062 * <h3>Usage Example</h3> 063 * 064 * <p>Here is a sketch of a service which crawls a website and uses the scheduling capabilities to 065 * rate limit itself. 066 * 067 * <pre>{@code 068 * class CrawlingService extends AbstractScheduledService { 069 * private Set<Uri> visited; 070 * private Queue<Uri> toCrawl; 071 * protected void startUp() throws Exception { 072 * toCrawl = readStartingUris(); 073 * } 074 * 075 * protected void runOneIteration() throws Exception { 076 * Uri uri = toCrawl.remove(); 077 * Collection<Uri> newUris = crawl(uri); 078 * visited.add(uri); 079 * for (Uri newUri : newUris) { 080 * if (!visited.contains(newUri)) { toCrawl.add(newUri); } 081 * } 082 * } 083 * 084 * protected void shutDown() throws Exception { 085 * saveUris(toCrawl); 086 * } 087 * 088 * protected Scheduler scheduler() { 089 * return Scheduler.newFixedRateSchedule(0, 1, TimeUnit.SECONDS); 090 * } 091 * } 092 * }</pre> 093 * 094 * <p>This class uses the life cycle methods to read in a list of starting URIs and save the set of 095 * outstanding URIs when shutting down. Also, it takes advantage of the scheduling functionality to 096 * rate limit the number of queries we perform. 097 * 098 * @author Luke Sandberg 099 * @since 11.0 100 */ 101@GwtIncompatible 102@ElementTypesAreNonnullByDefault 103public abstract class AbstractScheduledService implements Service { 104 private static final Logger logger = Logger.getLogger(AbstractScheduledService.class.getName()); 105 106 /** 107 * A scheduler defines the policy for how the {@link AbstractScheduledService} should run its 108 * task. 109 * 110 * <p>Consider using the {@link #newFixedDelaySchedule} and {@link #newFixedRateSchedule} factory 111 * methods, these provide {@link Scheduler} instances for the common use case of running the 112 * service with a fixed schedule. If more flexibility is needed then consider subclassing {@link 113 * CustomScheduler}. 114 * 115 * @author Luke Sandberg 116 * @since 11.0 117 */ 118 public abstract static class Scheduler { 119 /** 120 * Returns a {@link Scheduler} that schedules the task using the {@link 121 * ScheduledExecutorService#scheduleWithFixedDelay} method. 122 * 123 * @param initialDelay the time to delay first execution 124 * @param delay the delay between the termination of one execution and the commencement of the 125 * next 126 * @param unit the time unit of the initialDelay and delay parameters 127 */ 128 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 129 public static Scheduler newFixedDelaySchedule( 130 final long initialDelay, final long delay, final TimeUnit unit) { 131 checkNotNull(unit); 132 checkArgument(delay > 0, "delay must be > 0, found %s", delay); 133 return new Scheduler() { 134 @Override 135 public Cancellable schedule( 136 AbstractService service, ScheduledExecutorService executor, Runnable task) { 137 return new FutureAsCancellable( 138 executor.scheduleWithFixedDelay(task, initialDelay, delay, unit)); 139 } 140 }; 141 } 142 143 /** 144 * Returns a {@link Scheduler} that schedules the task using the {@link 145 * ScheduledExecutorService#scheduleAtFixedRate} method. 146 * 147 * @param initialDelay the time to delay first execution 148 * @param period the period between successive executions of the task 149 * @param unit the time unit of the initialDelay and period parameters 150 */ 151 @SuppressWarnings("GoodTime") // should accept a java.time.Duration 152 public static Scheduler newFixedRateSchedule( 153 final long initialDelay, final long period, final TimeUnit unit) { 154 checkNotNull(unit); 155 checkArgument(period > 0, "period must be > 0, found %s", period); 156 return new Scheduler() { 157 @Override 158 public Cancellable schedule( 159 AbstractService service, ScheduledExecutorService executor, Runnable task) { 160 return new FutureAsCancellable( 161 executor.scheduleAtFixedRate(task, initialDelay, period, unit)); 162 } 163 }; 164 } 165 166 /** Schedules the task to run on the provided executor on behalf of the service. */ 167 abstract Cancellable schedule( 168 AbstractService service, ScheduledExecutorService executor, Runnable runnable); 169 170 private Scheduler() {} 171 } 172 173 /* use AbstractService for state management */ 174 private final AbstractService delegate = new ServiceDelegate(); 175 176 @WeakOuter 177 private final class ServiceDelegate extends AbstractService { 178 179 // A handle to the running task so that we can stop it when a shutdown has been requested. 180 // These two fields are volatile because their values will be accessed from multiple threads. 181 @CheckForNull private volatile Cancellable runningTask; 182 @CheckForNull private volatile ScheduledExecutorService executorService; 183 184 // This lock protects the task so we can ensure that none of the template methods (startUp, 185 // shutDown or runOneIteration) run concurrently with one another. 186 // TODO(lukes): why don't we use ListenableFuture to sequence things? Then we could drop the 187 // lock. 188 private final ReentrantLock lock = new ReentrantLock(); 189 190 @WeakOuter 191 class Task implements Runnable { 192 @Override 193 public void run() { 194 lock.lock(); 195 try { 196 /* 197 * requireNonNull is safe because Task isn't run (or at least it doesn't succeed in taking 198 * the lock) until after it's scheduled and the runningTask field is set. 199 */ 200 if (requireNonNull(runningTask).isCancelled()) { 201 // task may have been cancelled while blocked on the lock. 202 return; 203 } 204 AbstractScheduledService.this.runOneIteration(); 205 } catch (Throwable t) { 206 restoreInterruptIfIsInterruptedException(t); 207 try { 208 shutDown(); 209 } catch (Exception ignored) { 210 restoreInterruptIfIsInterruptedException(ignored); 211 logger.log( 212 Level.WARNING, 213 "Error while attempting to shut down the service after failure.", 214 ignored); 215 } 216 notifyFailed(t); 217 // requireNonNull is safe now, just as it was above. 218 requireNonNull(runningTask).cancel(false); // prevent future invocations. 219 } finally { 220 lock.unlock(); 221 } 222 } 223 } 224 225 private final Runnable task = new Task(); 226 227 @Override 228 protected final void doStart() { 229 executorService = 230 MoreExecutors.renamingDecorator( 231 executor(), 232 new Supplier<String>() { 233 @Override 234 public String get() { 235 return serviceName() + " " + state(); 236 } 237 }); 238 executorService.execute( 239 new Runnable() { 240 @Override 241 public void run() { 242 lock.lock(); 243 try { 244 startUp(); 245 runningTask = scheduler().schedule(delegate, executorService, task); 246 notifyStarted(); 247 } catch (Throwable t) { 248 restoreInterruptIfIsInterruptedException(t); 249 notifyFailed(t); 250 if (runningTask != null) { 251 // prevent the task from running if possible 252 runningTask.cancel(false); 253 } 254 } finally { 255 lock.unlock(); 256 } 257 } 258 }); 259 } 260 261 @Override 262 protected final void doStop() { 263 // Both requireNonNull calls are safe because doStop can run only after a successful doStart. 264 requireNonNull(runningTask); 265 requireNonNull(executorService); 266 runningTask.cancel(false); 267 executorService.execute( 268 new Runnable() { 269 @Override 270 public void run() { 271 try { 272 lock.lock(); 273 try { 274 if (state() != State.STOPPING) { 275 // This means that the state has changed since we were scheduled. This implies 276 // that an execution of runOneIteration has thrown an exception and we have 277 // transitioned to a failed state, also this means that shutDown has already 278 // been called, so we do not want to call it again. 279 return; 280 } 281 shutDown(); 282 } finally { 283 lock.unlock(); 284 } 285 notifyStopped(); 286 } catch (Throwable t) { 287 restoreInterruptIfIsInterruptedException(t); 288 notifyFailed(t); 289 } 290 } 291 }); 292 } 293 294 @Override 295 public String toString() { 296 return AbstractScheduledService.this.toString(); 297 } 298 } 299 300 /** Constructor for use by subclasses. */ 301 protected AbstractScheduledService() {} 302 303 /** 304 * Run one iteration of the scheduled task. If any invocation of this method throws an exception, 305 * the service will transition to the {@link Service.State#FAILED} state and this method will no 306 * longer be called. 307 */ 308 protected abstract void runOneIteration() throws Exception; 309 310 /** 311 * Start the service. 312 * 313 * <p>By default this method does nothing. 314 */ 315 protected void startUp() throws Exception {} 316 317 /** 318 * Stop the service. This is guaranteed not to run concurrently with {@link #runOneIteration}. 319 * 320 * <p>By default this method does nothing. 321 */ 322 protected void shutDown() throws Exception {} 323 324 /** 325 * Returns the {@link Scheduler} object used to configure this service. This method will only be 326 * called once. 327 */ 328 // TODO(cpovirk): @ForOverride 329 protected abstract Scheduler scheduler(); 330 331 /** 332 * Returns the {@link ScheduledExecutorService} that will be used to execute the {@link #startUp}, 333 * {@link #runOneIteration} and {@link #shutDown} methods. If this method is overridden the 334 * executor will not be {@linkplain ScheduledExecutorService#shutdown shutdown} when this service 335 * {@linkplain Service.State#TERMINATED terminates} or {@linkplain Service.State#TERMINATED 336 * fails}. Subclasses may override this method to supply a custom {@link ScheduledExecutorService} 337 * instance. This method is guaranteed to only be called once. 338 * 339 * <p>By default this returns a new {@link ScheduledExecutorService} with a single thread pool 340 * that sets the name of the thread to the {@linkplain #serviceName() service name}. Also, the 341 * pool will be {@linkplain ScheduledExecutorService#shutdown() shut down} when the service 342 * {@linkplain Service.State#TERMINATED terminates} or {@linkplain Service.State#TERMINATED 343 * fails}. 344 */ 345 protected ScheduledExecutorService executor() { 346 @WeakOuter 347 class ThreadFactoryImpl implements ThreadFactory { 348 @Override 349 public Thread newThread(Runnable runnable) { 350 return MoreExecutors.newThread(serviceName(), runnable); 351 } 352 } 353 final ScheduledExecutorService executor = 354 Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl()); 355 // Add a listener to shut down the executor after the service is stopped. This ensures that the 356 // JVM shutdown will not be prevented from exiting after this service has stopped or failed. 357 // Technically this listener is added after start() was called so it is a little gross, but it 358 // is called within doStart() so we know that the service cannot terminate or fail concurrently 359 // with adding this listener so it is impossible to miss an event that we are interested in. 360 addListener( 361 new Listener() { 362 @Override 363 public void terminated(State from) { 364 executor.shutdown(); 365 } 366 367 @Override 368 public void failed(State from, Throwable failure) { 369 executor.shutdown(); 370 } 371 }, 372 directExecutor()); 373 return executor; 374 } 375 376 /** 377 * Returns the name of this service. {@link AbstractScheduledService} may include the name in 378 * debugging output. 379 * 380 * @since 14.0 381 */ 382 protected String serviceName() { 383 return getClass().getSimpleName(); 384 } 385 386 @Override 387 public String toString() { 388 return serviceName() + " [" + state() + "]"; 389 } 390 391 @Override 392 public final boolean isRunning() { 393 return delegate.isRunning(); 394 } 395 396 @Override 397 public final State state() { 398 return delegate.state(); 399 } 400 401 /** @since 13.0 */ 402 @Override 403 public final void addListener(Listener listener, Executor executor) { 404 delegate.addListener(listener, executor); 405 } 406 407 /** @since 14.0 */ 408 @Override 409 public final Throwable failureCause() { 410 return delegate.failureCause(); 411 } 412 413 /** @since 15.0 */ 414 @CanIgnoreReturnValue 415 @Override 416 public final Service startAsync() { 417 delegate.startAsync(); 418 return this; 419 } 420 421 /** @since 15.0 */ 422 @CanIgnoreReturnValue 423 @Override 424 public final Service stopAsync() { 425 delegate.stopAsync(); 426 return this; 427 } 428 429 /** @since 15.0 */ 430 @Override 431 public final void awaitRunning() { 432 delegate.awaitRunning(); 433 } 434 435 /** @since 15.0 */ 436 @Override 437 public final void awaitRunning(long timeout, TimeUnit unit) throws TimeoutException { 438 delegate.awaitRunning(timeout, unit); 439 } 440 441 /** @since 15.0 */ 442 @Override 443 public final void awaitTerminated() { 444 delegate.awaitTerminated(); 445 } 446 447 /** @since 15.0 */ 448 @Override 449 public final void awaitTerminated(long timeout, TimeUnit unit) throws TimeoutException { 450 delegate.awaitTerminated(timeout, unit); 451 } 452 453 interface Cancellable { 454 void cancel(boolean mayInterruptIfRunning); 455 456 boolean isCancelled(); 457 } 458 459 private static final class FutureAsCancellable implements Cancellable { 460 private final Future<?> delegate; 461 462 FutureAsCancellable(Future<?> delegate) { 463 this.delegate = delegate; 464 } 465 466 @Override 467 public void cancel(boolean mayInterruptIfRunning) { 468 delegate.cancel(mayInterruptIfRunning); 469 } 470 471 @Override 472 public boolean isCancelled() { 473 return delegate.isCancelled(); 474 } 475 } 476 477 /** 478 * A {@link Scheduler} that provides a convenient way for the {@link AbstractScheduledService} to 479 * use a dynamically changing schedule. After every execution of the task, assuming it hasn't been 480 * cancelled, the {@link #getNextSchedule} method will be called. 481 * 482 * @author Luke Sandberg 483 * @since 11.0 484 */ 485 public abstract static class CustomScheduler extends Scheduler { 486 487 /** A callable class that can reschedule itself using a {@link CustomScheduler}. */ 488 private final class ReschedulableCallable implements Callable<@Nullable Void> { 489 490 /** The underlying task. */ 491 private final Runnable wrappedRunnable; 492 493 /** The executor on which this Callable will be scheduled. */ 494 private final ScheduledExecutorService executor; 495 496 /** 497 * The service that is managing this callable. This is used so that failure can be reported 498 * properly. 499 */ 500 /* 501 * This reference is part of a reference cycle, which is typically something we want to avoid 502 * under j2objc -- but it is not detected by our j2objc cycle test. The cycle: 503 * 504 * - CustomScheduler.service contains an instance of ServiceDelegate. (It needs it so that it 505 * can call notifyFailed.) 506 * 507 * - ServiceDelegate.runningTask contains an instance of ReschedulableCallable (at least in 508 * the case that the service is using CustomScheduler). (It needs it so that it can cancel 509 * the task and detect whether it has been cancelled.) 510 * 511 * - ReschedulableCallable has a reference back to its enclosing CustomScheduler. (It needs it 512 * so that it can call getNextSchedule). 513 * 514 * Maybe there is a way to avoid this cycle. But we think the cycle is safe enough to ignore: 515 * Each task is retained for only as long as it is running -- so it's retained only as long as 516 * it would already be retained by the underlying executor. 517 * 518 * If the cycle test starts reporting this cycle in the future, we should add an entry to 519 * cycle_suppress_list.txt. 520 */ 521 private final AbstractService service; 522 523 /** 524 * This lock is used to ensure safe and correct cancellation, it ensures that a new task is 525 * not scheduled while a cancel is ongoing. Also it protects the currentFuture variable to 526 * ensure that it is assigned atomically with being scheduled. 527 */ 528 private final ReentrantLock lock = new ReentrantLock(); 529 530 /** The future that represents the next execution of this task. */ 531 @GuardedBy("lock") 532 @CheckForNull 533 private SupplantableFuture cancellationDelegate; 534 535 ReschedulableCallable( 536 AbstractService service, ScheduledExecutorService executor, Runnable runnable) { 537 this.wrappedRunnable = runnable; 538 this.executor = executor; 539 this.service = service; 540 } 541 542 @Override 543 @CheckForNull 544 public Void call() throws Exception { 545 wrappedRunnable.run(); 546 reschedule(); 547 return null; 548 } 549 550 /** 551 * Atomically reschedules this task and assigns the new future to {@link 552 * #cancellationDelegate}. 553 */ 554 @CanIgnoreReturnValue 555 public Cancellable reschedule() { 556 // invoke the callback outside the lock, prevents some shenanigans. 557 Schedule schedule; 558 try { 559 schedule = CustomScheduler.this.getNextSchedule(); 560 } catch (Throwable t) { 561 restoreInterruptIfIsInterruptedException(t); 562 service.notifyFailed(t); 563 return new FutureAsCancellable(immediateCancelledFuture()); 564 } 565 // We reschedule ourselves with a lock held for two reasons. 1. we want to make sure that 566 // cancel calls cancel on the correct future. 2. we want to make sure that the assignment 567 // to currentFuture doesn't race with itself so that currentFuture is assigned in the 568 // correct order. 569 Throwable scheduleFailure = null; 570 Cancellable toReturn; 571 lock.lock(); 572 try { 573 toReturn = initializeOrUpdateCancellationDelegate(schedule); 574 } catch (RuntimeException | Error e) { 575 // If an exception is thrown by the subclass then we need to make sure that the service 576 // notices and transitions to the FAILED state. We do it by calling notifyFailed directly 577 // because the service does not monitor the state of the future so if the exception is not 578 // caught and forwarded to the service the task would stop executing but the service would 579 // have no idea. 580 // TODO(lukes): consider building everything in terms of ListenableScheduledFuture then 581 // the AbstractService could monitor the future directly. Rescheduling is still hard... 582 // but it would help with some of these lock ordering issues. 583 scheduleFailure = e; 584 toReturn = new FutureAsCancellable(immediateCancelledFuture()); 585 } finally { 586 lock.unlock(); 587 } 588 // Call notifyFailed outside the lock to avoid lock ordering issues. 589 if (scheduleFailure != null) { 590 service.notifyFailed(scheduleFailure); 591 } 592 return toReturn; 593 } 594 595 @GuardedBy("lock") 596 /* 597 * The GuardedBy checker warns us that we're not holding cancellationDelegate.lock. But in 598 * fact we are holding it because it is the same as this.lock, which we know we are holding, 599 * thanks to @GuardedBy above. (cancellationDelegate.lock is initialized to this.lock in the 600 * call to `new SupplantableFuture` below.) 601 */ 602 @SuppressWarnings("GuardedBy") 603 private Cancellable initializeOrUpdateCancellationDelegate(Schedule schedule) { 604 if (cancellationDelegate == null) { 605 return cancellationDelegate = new SupplantableFuture(lock, submitToExecutor(schedule)); 606 } 607 if (!cancellationDelegate.currentFuture.isCancelled()) { 608 cancellationDelegate.currentFuture = submitToExecutor(schedule); 609 } 610 return cancellationDelegate; 611 } 612 613 private ScheduledFuture<@Nullable Void> submitToExecutor(Schedule schedule) { 614 return executor.schedule(this, schedule.delay, schedule.unit); 615 } 616 } 617 618 /** 619 * Contains the most recently submitted {@code Future}, which may be cancelled or updated, 620 * always under a lock. 621 */ 622 private static final class SupplantableFuture implements Cancellable { 623 private final ReentrantLock lock; 624 625 @GuardedBy("lock") 626 private Future<@Nullable Void> currentFuture; 627 628 SupplantableFuture(ReentrantLock lock, Future<@Nullable Void> currentFuture) { 629 this.lock = lock; 630 this.currentFuture = currentFuture; 631 } 632 633 @Override 634 public void cancel(boolean mayInterruptIfRunning) { 635 /* 636 * Lock to ensure that a task cannot be rescheduled while a cancel is ongoing. 637 * 638 * In theory, cancel() could execute arbitrary listeners -- bad to do while holding a lock. 639 * However, we don't expose currentFuture to users, so they can't attach listeners. And the 640 * Future might not even be a ListenableFuture, just a plain Future. That said, similar 641 * problems can exist with methods like FutureTask.done(), not to mention slow calls to 642 * Thread.interrupt() (as discussed in InterruptibleTask). At the end of the day, it's 643 * unlikely that cancel() will be slow, so we can probably get away with calling it while 644 * holding a lock. Still, it would be nice to avoid somehow. 645 */ 646 lock.lock(); 647 try { 648 currentFuture.cancel(mayInterruptIfRunning); 649 } finally { 650 lock.unlock(); 651 } 652 } 653 654 @Override 655 public boolean isCancelled() { 656 lock.lock(); 657 try { 658 return currentFuture.isCancelled(); 659 } finally { 660 lock.unlock(); 661 } 662 } 663 } 664 665 @Override 666 final Cancellable schedule( 667 AbstractService service, ScheduledExecutorService executor, Runnable runnable) { 668 return new ReschedulableCallable(service, executor, runnable).reschedule(); 669 } 670 671 /** 672 * A value object that represents an absolute delay until a task should be invoked. 673 * 674 * @author Luke Sandberg 675 * @since 11.0 676 */ 677 protected static final class Schedule { 678 679 private final long delay; 680 private final TimeUnit unit; 681 682 /** 683 * @param delay the time from now to delay execution 684 * @param unit the time unit of the delay parameter 685 */ 686 public Schedule(long delay, TimeUnit unit) { 687 this.delay = delay; 688 this.unit = checkNotNull(unit); 689 } 690 } 691 692 /** 693 * Calculates the time at which to next invoke the task. 694 * 695 * <p>This is guaranteed to be called immediately after the task has completed an iteration and 696 * on the same thread as the previous execution of {@link 697 * AbstractScheduledService#runOneIteration}. 698 * 699 * @return a schedule that defines the delay before the next execution. 700 */ 701 // TODO(cpovirk): @ForOverride 702 protected abstract Schedule getNextSchedule() throws Exception; 703 } 704}