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