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}