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