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