001/*
002 * Copyright (C) 2011 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 * http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016
017package com.google.common.util.concurrent;
018
019import com.google.common.annotations.Beta;
020import com.google.common.base.Preconditions;
021import com.google.common.base.Throwables;
022
023import java.util.concurrent.Callable;
024import java.util.concurrent.Executor;
025import java.util.concurrent.Executors;
026import java.util.concurrent.Future;
027import java.util.concurrent.ScheduledExecutorService;
028import java.util.concurrent.TimeUnit;
029import java.util.concurrent.locks.ReentrantLock;
030import java.util.logging.Level;
031import java.util.logging.Logger;
032
033import javax.annotation.concurrent.GuardedBy;
034
035/**
036 * Base class for services that can implement {@link #startUp} and {@link #shutDown} but while in
037 * the "running" state need to perform a periodic task.  Subclasses can implement {@link #startUp},
038 * {@link #shutDown} and also a {@link #runOneIteration} method that will be executed periodically.
039 *
040 * <p>This class uses the {@link ScheduledExecutorService} returned from {@link #executor} to run
041 * the {@link #startUp} and {@link #shutDown} methods and also uses that service to schedule the
042 * {@link #runOneIteration} that will be executed periodically as specified by its
043 * {@link Scheduler}. When this service is asked to stop via {@link #stop} or {@link #stopAndWait},
044 * it will cancel the periodic task (but not interrupt it) and wait for it to stop before running
045 * the {@link #shutDown} method.
046 *
047 * <p>Subclasses are guaranteed that the life cycle methods ({@link #runOneIteration}, {@link
048 * #startUp} and {@link #shutDown}) will never run concurrently. Notably, if any execution of {@link
049 * #runOneIteration} takes longer than its schedule defines, then subsequent executions may start
050 * late.  Also, all life cycle methods are executed with a lock held, so subclasses can safely
051 * modify shared state without additional synchronization necessary for visibility to later
052 * executions of the life cycle methods.
053 *
054 * <h3>Usage Example</h3>
055 *
056 * Here is a sketch of a service which crawls a website and uses the scheduling capabilities to
057 * rate limit itself. <pre> {@code
058 * class CrawlingService extends AbstractScheduledService {
059 *   private Set<Uri> visited;
060 *   private Queue<Uri> toCrawl;
061 *   protected void startUp() throws Exception {
062 *     toCrawl = readStartingUris();
063 *   }
064 *
065 *   protected void runOneIteration() throws Exception {
066 *     Uri uri = toCrawl.remove();
067 *     Collection<Uri> newUris = crawl(uri);
068 *     visited.add(uri);
069 *     for (Uri newUri : newUris) {
070 *       if (!visited.contains(newUri)) { toCrawl.add(newUri); }
071 *     }
072 *   }
073 *
074 *   protected void shutDown() throws Exception {
075 *     saveUris(toCrawl);
076 *   }
077 *
078 *   protected Scheduler scheduler() {
079 *     return Scheduler.newFixedRateSchedule(0, 1, TimeUnit.SECONDS);
080 *   }
081 * }}</pre>
082 *
083 * This class uses the life cycle methods to read in a list of starting URIs and save the set of
084 * outstanding URIs when shutting down.  Also, it takes advantage of the scheduling functionality to
085 * rate limit the number of queries we perform.
086 *
087 * @author Luke Sandberg
088 * @since 11.0
089 */
090@Beta
091public abstract class AbstractScheduledService implements Service {
092  private static final Logger logger = Logger.getLogger(AbstractScheduledService.class.getName());
093
094  /**
095   * A scheduler defines the policy for how the {@link AbstractScheduledService} should run its
096   * task.
097   *
098   * <p>Consider using the {@link #newFixedDelaySchedule} and {@link #newFixedRateSchedule} factory
099   * methods, these provide {@link Scheduler} instances for the common use case of running the
100   * service with a fixed schedule.  If more flexibility is needed then consider subclassing the
101   * {@link CustomScheduler} abstract class in preference to creating your own {@link Scheduler}
102   * implementation.
103   *
104   * @author Luke Sandberg
105   * @since 11.0
106   */
107  public abstract static class Scheduler {
108    /**
109     * Returns a {@link Scheduler} that schedules the task using the
110     * {@link ScheduledExecutorService#scheduleWithFixedDelay} method.
111     *
112     * @param initialDelay the time to delay first execution
113     * @param delay the delay between the termination of one execution and the commencement of the
114     *        next
115     * @param unit the time unit of the initialDelay and delay parameters
116     */
117    public static Scheduler newFixedDelaySchedule(final long initialDelay, final long delay,
118        final TimeUnit unit) {
119      return new Scheduler() {
120        @Override
121        public Future<?> schedule(AbstractService service, ScheduledExecutorService executor,
122            Runnable task) {
123          return executor.scheduleWithFixedDelay(task, initialDelay, delay, unit);
124        }
125      };
126    }
127
128    /**
129     * Returns a {@link Scheduler} that schedules the task using the
130     * {@link ScheduledExecutorService#scheduleAtFixedRate} method.
131     *
132     * @param initialDelay the time to delay first execution
133     * @param period the period between successive executions of the task
134     * @param unit the time unit of the initialDelay and period parameters
135     */
136    public static Scheduler newFixedRateSchedule(final long initialDelay, final long period,
137        final TimeUnit unit) {
138      return new Scheduler() {
139        @Override
140        public Future<?> schedule(AbstractService service, ScheduledExecutorService executor,
141            Runnable task) {
142          return executor.scheduleAtFixedRate(task, initialDelay, period, unit);
143        }
144      };
145    }
146
147    /** Schedules the task to run on the provided executor on behalf of the service.  */
148    abstract Future<?> schedule(AbstractService service, ScheduledExecutorService executor,
149        Runnable runnable);
150
151    private Scheduler() {}
152  }
153
154  /* use AbstractService for state management */
155  private final AbstractService delegate = new AbstractService() {
156
157    // A handle to the running task so that we can stop it when a shutdown has been requested.
158    // These two fields are volatile because their values will be accessed from multiple threads.
159    private volatile Future<?> runningTask;
160    private volatile ScheduledExecutorService executorService;
161
162    // This lock protects the task so we can ensure that none of the template methods (startUp,
163    // shutDown or runOneIteration) run concurrently with one another.
164    private final ReentrantLock lock = new ReentrantLock();
165
166    private final Runnable task = new Runnable() {
167      @Override public void run() {
168        lock.lock();
169        try {
170          AbstractScheduledService.this.runOneIteration();
171        } catch (Throwable t) {
172          try {
173            shutDown();
174          } catch (Exception ignored) {
175            logger.log(Level.WARNING,
176                "Error while attempting to shut down the service after failure.", ignored);
177          }
178          notifyFailed(t);
179          throw Throwables.propagate(t);
180        } finally {
181          lock.unlock();
182        }
183      }
184    };
185
186    @Override protected final void doStart() {
187      executorService = executor();
188      executorService.execute(new Runnable() {
189        @Override public void run() {
190          lock.lock();
191          try {
192            startUp();
193            runningTask = scheduler().schedule(delegate, executorService, task);
194            notifyStarted();
195          } catch (Throwable t) {
196            notifyFailed(t);
197            throw Throwables.propagate(t);
198          } finally {
199            lock.unlock();
200          }
201        }
202      });
203    }
204
205    @Override protected final void doStop() {
206      runningTask.cancel(false);
207      executorService.execute(new Runnable() {
208        @Override public void run() {
209          try {
210            lock.lock();
211            try {
212              if (state() != State.STOPPING) {
213                // This means that the state has changed since we were scheduled.  This implies that
214                // an execution of runOneIteration has thrown an exception and we have transitioned
215                // to a failed state, also this means that shutDown has already been called, so we
216                // do not want to call it again.
217                return;
218              }
219              shutDown();
220            } finally {
221              lock.unlock();
222            }
223            notifyStopped();
224          } catch (Throwable t) {
225            notifyFailed(t);
226            throw Throwables.propagate(t);
227          }
228        }
229      });
230    }
231  };
232
233  /**
234   * Run one iteration of the scheduled task. If any invocation of this method throws an exception,
235   * the service will transition to the {@link Service.State#FAILED} state and this method will no
236   * longer be called.
237   */
238  protected abstract void runOneIteration() throws Exception;
239
240  /**
241   * Start the service.
242   *
243   * <p>By default this method does nothing.
244   */
245  protected void startUp() throws Exception {}
246
247  /**
248   * Stop the service. This is guaranteed not to run concurrently with {@link #runOneIteration}.
249   *
250   * <p>By default this method does nothing.
251   */
252  protected void shutDown() throws Exception {}
253
254  /**
255   * Returns the {@link Scheduler} object used to configure this service.  This method will only be
256   * called once.
257   */
258  protected abstract Scheduler scheduler();
259
260  /**
261   * Returns the {@link ScheduledExecutorService} that will be used to execute the {@link #startUp},
262   * {@link #runOneIteration} and {@link #shutDown} methods.  The executor will not be
263   * {@link ScheduledExecutorService#shutdown} when this service stops. Subclasses may override this
264   * method to use a custom {@link ScheduledExecutorService} instance.
265   *
266   * <p>By default this returns a new {@link ScheduledExecutorService} with a single thread thread
267   * pool.  This method will only be called once.
268   */
269  protected ScheduledExecutorService executor() {
270    return Executors.newSingleThreadScheduledExecutor();
271  }
272
273  @Override public String toString() {
274    return getClass().getSimpleName() + " [" + state() + "]";
275  }
276
277  // We override instead of using ForwardingService so that these can be final.
278
279  @Override public final ListenableFuture<State> start() {
280    return delegate.start();
281  }
282
283  @Override public final State startAndWait() {
284    return delegate.startAndWait();
285  }
286
287  @Override public final boolean isRunning() {
288    return delegate.isRunning();
289  }
290
291  @Override public final State state() {
292    return delegate.state();
293  }
294
295  @Override public final ListenableFuture<State> stop() {
296    return delegate.stop();
297  }
298
299  @Override public final State stopAndWait() {
300    return delegate.stopAndWait();
301  }
302
303  @Override public final void addListener(Listener listener, Executor executor) {
304    delegate.addListener(listener, executor);
305  }
306
307  /**
308   * A {@link Scheduler} that provides a convenient way for the {@link AbstractScheduledService} to
309   * use a dynamically changing schedule.  After every execution of the task, assuming it hasn't
310   * been cancelled, the {@link #getNextSchedule} method will be called.
311   *
312   * @author Luke Sandberg
313   * @since 11.0
314   */
315  @Beta
316  public abstract static class CustomScheduler extends Scheduler {
317
318    /**
319     * A callable class that can reschedule itself using a {@link CustomScheduler}.
320     */
321    private class ReschedulableCallable extends ForwardingFuture<Void> implements Callable<Void> {
322
323      /** The underlying task. */
324      private final Runnable wrappedRunnable;
325
326      /** The executor on which this Callable will be scheduled. */
327      private final ScheduledExecutorService executor;
328
329      /**
330       * The service that is managing this callable.  This is used so that failure can be
331       * reported properly.
332       */
333      private final AbstractService service;
334
335      /**
336       * This lock is used to ensure safe and correct cancellation, it ensures that a new task is
337       * not scheduled while a cancel is ongoing.  Also it protects the currentFuture variable to
338       * ensure that it is assigned atomically with being scheduled.
339       */
340      private final ReentrantLock lock = new ReentrantLock();
341
342      /** The future that represents the next execution of this task.*/
343      @GuardedBy("lock")
344      private Future<Void> currentFuture;
345
346      ReschedulableCallable(AbstractService service, ScheduledExecutorService executor,
347          Runnable runnable) {
348        this.wrappedRunnable = runnable;
349        this.executor = executor;
350        this.service = service;
351      }
352
353      @Override
354      public Void call() throws Exception {
355        wrappedRunnable.run();
356        reschedule();
357        return null;
358      }
359
360      /**
361       * Atomically reschedules this task and assigns the new future to {@link #currentFuture}.
362       */
363      public void reschedule() {
364        // We reschedule ourselves with a lock held for two reasons. 1. we want to make sure that
365        // cancel calls cancel on the correct future. 2. we want to make sure that the assignment
366        // to currentFuture doesn't race with itself so that currentFuture is assigned in the
367        // correct order.
368        lock.lock();
369        try {
370          if (currentFuture == null || !currentFuture.isCancelled()) {
371            final Schedule schedule = CustomScheduler.this.getNextSchedule();
372            currentFuture = executor.schedule(this, schedule.delay, schedule.unit);
373          }
374        } catch (Throwable e) {
375          // If an exception is thrown by the subclass then we need to make sure that the service
376          // notices and transitions to the FAILED state.  We do it by calling notifyFailed directly
377          // because the service does not monitor the state of the future so if the exception is not
378          // caught and forwarded to the service the task would stop executing but the service would
379          // have no idea.
380          service.notifyFailed(e);
381        } finally {
382          lock.unlock();
383        }
384      }
385
386      // N.B. Only protect cancel and isCancelled because those are the only methods that are
387      // invoked by the AbstractScheduledService.
388      @Override
389      public boolean cancel(boolean mayInterruptIfRunning) {
390        // Ensure that a task cannot be rescheduled while a cancel is ongoing.
391        lock.lock();
392        try {
393          return currentFuture.cancel(mayInterruptIfRunning);
394        } finally {
395          lock.unlock();
396        }
397      }
398
399      @Override
400      protected Future<Void> delegate() {
401        throw new UnsupportedOperationException("Only cancel is supported by this future");
402      }
403    }
404
405    @Override
406    final Future<?> schedule(AbstractService service, ScheduledExecutorService executor,
407        Runnable runnable) {
408      ReschedulableCallable task = new ReschedulableCallable(service, executor, runnable);
409      task.reschedule();
410      return task;
411    }
412
413    /**
414     * A value object that represents an absolute delay until a task should be invoked.
415     *
416     * @author Luke Sandberg
417     * @since 11.0
418     */
419    @Beta
420    protected static final class Schedule {
421
422      private final long delay;
423      private final TimeUnit unit;
424
425      /**
426       * @param delay the time from now to delay execution
427       * @param unit the time unit of the delay parameter
428       */
429      public Schedule(long delay, TimeUnit unit) {
430        this.delay = delay;
431        this.unit = Preconditions.checkNotNull(unit);
432      }
433    }
434
435    /**
436     * Calculates the time at which to next invoke the task.
437     *
438     * <p>This is guaranteed to be called immediately after the task has completed an iteration and
439     * on the same thread as the previous execution of {@link
440     * AbstractScheduledService#runOneIteration}.
441     *
442     * @return a schedule that defines the delay before the next execution.
443     */
444    protected abstract Schedule getNextSchedule() throws Exception;
445  }
446}