View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase;
20  
21  import java.util.concurrent.ScheduledThreadPoolExecutor;
22  import java.util.concurrent.TimeUnit;
23  
24  import org.apache.commons.logging.Log;
25  import org.apache.commons.logging.LogFactory;
26  import org.apache.hadoop.hbase.classification.InterfaceAudience;
27  import org.apache.hadoop.hbase.classification.InterfaceStability;
28  
29  import com.google.common.annotations.VisibleForTesting;
30  
31  /**
32   * ScheduledChore is a task performed on a period in hbase. ScheduledChores become active once
33   * scheduled with a {@link ChoreService} via {@link ChoreService#scheduleChore(ScheduledChore)}. The
34   * chore is run in a {@link ScheduledThreadPoolExecutor} and competes with other ScheduledChores for
35   * access to the threads in the core thread pool. If an unhandled exception occurs, the chore
36   * cancellation is logged. Implementers should consider whether or not the Chore will be able to
37   * execute within the defined period. It is bad practice to define a ScheduledChore whose execution
38   * time exceeds its period since it will try to hog one of the threads in the {@link ChoreService}'s
39   * thread pool.
40   * <p>
41   * Don't subclass ScheduledChore if the task relies on being woken up for something to do, such as
42   * an entry being added to a queue, etc.
43   */
44  @InterfaceAudience.Public
45  @InterfaceStability.Stable
46  public abstract class ScheduledChore implements Runnable {
47    private static final Log LOG = LogFactory.getLog(ScheduledChore.class);
48  
49    private final String name;
50  
51    /**
52     * Default values for scheduling parameters should they be excluded during construction
53     */
54    private final static TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
55    private final static long DEFAULT_INITIAL_DELAY = 0;
56  
57    /**
58     * Scheduling parameters. Used by ChoreService when scheduling the chore to run periodically
59     */
60    private final int period; // in TimeUnit units
61    private final TimeUnit timeUnit;
62    private final long initialDelay; // in TimeUnit units
63  
64    /**
65     * Interface to the ChoreService that this ScheduledChore is scheduled with. null if the chore is
66     * not scheduled.
67     */
68    private ChoreServicer choreServicer;
69  
70    /**
71     * Variables that encapsulate the meaningful state information
72     */
73    private long timeOfLastRun = -1; // system time millis
74    private long timeOfThisRun = -1; // system time millis
75    private boolean initialChoreComplete = false;
76  
77    /**
78     * A means by which a ScheduledChore can be stopped. Once a chore recognizes that it has been
79     * stopped, it will cancel itself. This is particularly useful in the case where a single stopper
80     * instance is given to multiple chores. In such a case, a single {@link Stoppable#stop(String)}
81     * command can cause many chores to stop together.
82     */
83    private final Stoppable stopper;
84  
85    interface ChoreServicer {
86      /**
87       * Cancel any ongoing schedules that this chore has with the implementer of this interface.
88       */
89      public void cancelChore(ScheduledChore chore);
90      public void cancelChore(ScheduledChore chore, boolean mayInterruptIfRunning);
91  
92      /**
93       * @return true when the chore is scheduled with the implementer of this interface
94       */
95      public boolean isChoreScheduled(ScheduledChore chore);
96  
97      /**
98       * This method tries to execute the chore immediately. If the chore is executing at the time of
99       * this call, the chore will begin another execution as soon as the current execution finishes
100      * <p>
101      * If the chore is not scheduled with a ChoreService, this call will fail.
102      * @return false when the chore could not be triggered immediately
103      */
104     public boolean triggerNow(ScheduledChore chore);
105 
106     /**
107      * A callback that tells the implementer of this interface that one of the scheduled chores is
108      * missing its start time. The implication of a chore missing its start time is that the
109      * service's current means of scheduling may not be sufficient to handle the number of ongoing
110      * chores (the other explanation is that the chore's execution time is greater than its
111      * scheduled period). The service should try to increase its concurrency when this callback is
112      * received.
113      * @param chore The chore that missed its start time
114      */
115     public void onChoreMissedStartTime(ScheduledChore chore);
116   }
117 
118   /**
119    * This constructor is for test only. It allows us to create an object and to call chore() on it.
120    */
121   @InterfaceAudience.Private
122   @VisibleForTesting
123   protected ScheduledChore() {
124     this.name = null;
125     this.stopper = null;
126     this.period = 0;
127     this.initialDelay = DEFAULT_INITIAL_DELAY;
128     this.timeUnit = DEFAULT_TIME_UNIT;
129   }
130 
131   /**
132    * @param name Name assigned to Chore. Useful for identification amongst chores of the same type
133    * @param stopper When {@link Stoppable#isStopped()} is true, this chore will cancel and cleanup
134    * @param period Period in millis with which this Chore repeats execution when scheduled.
135    */
136   public ScheduledChore(final String name, Stoppable stopper, final int period) {
137     this(name, stopper, period, DEFAULT_INITIAL_DELAY);
138   }
139 
140   /**
141    * @param name Name assigned to Chore. Useful for identification amongst chores of the same type
142    * @param stopper When {@link Stoppable#isStopped()} is true, this chore will cancel and cleanup
143    * @param period Period in millis with which this Chore repeats execution when scheduled.
144    * @param initialDelay Delay before this Chore begins to execute once it has been scheduled. A
145    *          value of 0 means the chore will begin to execute immediately. Negative delays are
146    *          invalid and will be corrected to a value of 0.
147    */
148   public ScheduledChore(final String name, Stoppable stopper, final int period,
149       final long initialDelay) {
150     this(name, stopper, period, initialDelay, DEFAULT_TIME_UNIT);
151   }
152 
153   /**
154    * @param name Name assigned to Chore. Useful for identification amongst chores of the same type
155    * @param stopper When {@link Stoppable#isStopped()} is true, this chore will cancel and cleanup
156    * @param period Period in Timeunit unit with which this Chore repeats execution when scheduled.
157    * @param initialDelay Delay in Timeunit unit before this Chore begins to execute once it has been
158    *          scheduled. A value of 0 means the chore will begin to execute immediately. Negative
159    *          delays are invalid and will be corrected to a value of 0.
160    * @param unit The unit that is used to measure period and initialDelay
161    */
162   public ScheduledChore(final String name, Stoppable stopper, final int period,
163       final long initialDelay, final TimeUnit unit) {
164     this.name = name;
165     this.stopper = stopper;
166     this.period = period;
167     this.initialDelay = initialDelay < 0 ? 0 : initialDelay;
168     this.timeUnit = unit;
169   }
170 
171   /**
172    * @see java.lang.Runnable#run()
173    */
174   @Override
175   public void run() {
176     updateTimeTrackingBeforeRun();
177     if (missedStartTime() && isScheduled()) {
178       onChoreMissedStartTime();
179       if (LOG.isInfoEnabled()) LOG.info("Chore: " + getName() + " missed its start time");
180     } else if (stopper.isStopped() || !isScheduled()) {
181       cancel(false);
182       cleanup();
183       if (LOG.isInfoEnabled()) LOG.info("Chore: " + getName() + " was stopped");
184     } else {
185       try {
186         if (!initialChoreComplete) {
187           initialChoreComplete = initialChore();
188         } else {
189           chore();
190         }
191       } catch (Throwable t) {
192         if (LOG.isErrorEnabled()) LOG.error("Caught error", t);
193         if (this.stopper.isStopped()) {
194           cancel(false);
195           cleanup();
196         }
197       }
198     }
199   }
200 
201   /**
202    * Update our time tracking members. Called at the start of an execution of this chore's run()
203    * method so that a correct decision can be made as to whether or not we missed the start time
204    */
205   private synchronized void updateTimeTrackingBeforeRun() {
206     timeOfLastRun = timeOfThisRun;
207     timeOfThisRun = System.currentTimeMillis();
208   }
209 
210   /**
211    * Notify the ChoreService that this chore has missed its start time. Allows the ChoreService to
212    * make the decision as to whether or not it would be worthwhile to increase the number of core
213    * pool threads
214    */
215   private synchronized void onChoreMissedStartTime() {
216     if (choreServicer != null) choreServicer.onChoreMissedStartTime(this);
217   }
218 
219   /**
220    * @return How long in millis has it been since this chore last run. Useful for checking if the
221    *         chore has missed its scheduled start time by too large of a margin
222    */
223   synchronized long getTimeBetweenRuns() {
224     return timeOfThisRun - timeOfLastRun;
225   }
226 
227   /**
228    * @return true when the time between runs exceeds the acceptable threshold
229    */
230   private synchronized boolean missedStartTime() {
231     return isValidTime(timeOfLastRun) && isValidTime(timeOfThisRun)
232         && getTimeBetweenRuns() > getMaximumAllowedTimeBetweenRuns();
233   }
234 
235   /**
236    * @return max allowed time in millis between runs.
237    */
238   private double getMaximumAllowedTimeBetweenRuns() {
239     // Threshold used to determine if the Chore's current run started too late
240     return 1.5 * timeUnit.toMillis(period);
241   }
242 
243   /**
244    * @param time in system millis
245    * @return true if time is earlier or equal to current milli time
246    */
247   private synchronized boolean isValidTime(final long time) {
248     return time > 0 && time <= System.currentTimeMillis();
249   }
250 
251   /**
252    * @return false when the Chore is not currently scheduled with a ChoreService
253    */
254   public synchronized boolean triggerNow() {
255     if (choreServicer != null) {
256       return choreServicer.triggerNow(this);
257     } else {
258       return false;
259     }
260   }
261 
262   synchronized void setChoreServicer(ChoreServicer service) {
263     // Chores should only ever be scheduled with a single ChoreService. If the choreServicer
264     // is changing, cancel any existing schedules of this chore.
265     if (choreServicer != null && choreServicer != service) {
266       choreServicer.cancelChore(this, false);
267     }
268     choreServicer = service;
269     timeOfThisRun = System.currentTimeMillis();
270   }
271 
272   public synchronized void cancel() {
273     cancel(true);
274   }
275 
276   public synchronized void cancel(boolean mayInterruptIfRunning) {
277     if (isScheduled()) choreServicer.cancelChore(this, mayInterruptIfRunning);
278 
279     choreServicer = null;
280   }
281 
282   public String getName() {
283     return name;
284   }
285 
286   public Stoppable getStopper() {
287     return stopper;
288   }
289 
290   /**
291    * @return period to execute chore in getTimeUnit() units
292    */
293   public int getPeriod() {
294     return period;
295   }
296 
297   /**
298    * @return initial delay before executing chore in getTimeUnit() units
299    */
300   public long getInitialDelay() {
301     return initialDelay;
302   }
303 
304   public TimeUnit getTimeUnit() {
305     return timeUnit;
306   }
307 
308   public synchronized boolean isInitialChoreComplete() {
309     return initialChoreComplete;
310   }
311 
312   @VisibleForTesting
313   synchronized ChoreServicer getChoreServicer() {
314     return choreServicer;
315   }
316 
317   @VisibleForTesting
318   synchronized long getTimeOfLastRun() {
319     return timeOfLastRun;
320   }
321 
322   @VisibleForTesting
323   synchronized long getTimeOfThisRun() {
324     return timeOfThisRun;
325   }
326 
327   /**
328    * @return true when this Chore is scheduled with a ChoreService
329    */
330   public synchronized boolean isScheduled() {
331     return choreServicer != null && choreServicer.isChoreScheduled(this);
332   }
333 
334   @InterfaceAudience.Private
335   @VisibleForTesting
336   public synchronized void choreForTesting() {
337     chore();
338   }
339 
340   /**
341    * The task to execute on each scheduled execution of the Chore
342    */
343   protected abstract void chore();
344 
345   /**
346    * Override to run a task before we start looping.
347    * @return true if initial chore was successful
348    */
349   protected boolean initialChore() {
350     // Default does nothing
351     return true;
352   }
353 
354   /**
355    * Override to run cleanup tasks when the Chore encounters an error and must stop running
356    */
357   protected synchronized void cleanup() {
358   }
359 
360   /**
361    * A summation of this chore in human readable format. Downstream users should not presume
362    * parsing of this string can relaibly be done between versions. Instead, they should rely
363    * on the public accessor methods to get the information they desire.
364    */
365   @InterfaceAudience.Private
366   @Override
367   public String toString() {
368     return "[ScheduledChore: Name: " + getName() + " Period: " + getPeriod() + " Unit: "
369         + getTimeUnit() + "]";
370   }
371 }