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 }