001/**
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements.  See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership.  The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License.  You may obtain a copy of the License at
010 *
011 *     http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing, software
014 * distributed under the License is distributed on an "AS IS" BASIS,
015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
016 * See the License for the specific language governing permissions and
017 * limitations under the License.
018 */
019package org.apache.hadoop.hbase;
020
021import java.util.ArrayList;
022import java.util.HashMap;
023import java.util.LinkedHashMap;
024import java.util.Map.Entry;
025import java.util.concurrent.ScheduledFuture;
026import java.util.concurrent.ScheduledThreadPoolExecutor;
027import java.util.concurrent.ThreadFactory;
028import java.util.concurrent.atomic.AtomicInteger;
029
030import org.apache.hadoop.hbase.ScheduledChore.ChoreServicer;
031import org.apache.yetus.audience.InterfaceAudience;
032import org.slf4j.Logger;
033import org.slf4j.LoggerFactory;
034
035import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
036
037/**
038 * ChoreService is a service that can be used to schedule instances of {@link ScheduledChore} to run
039 * periodically while sharing threads. The ChoreService is backed by a
040 * {@link ScheduledThreadPoolExecutor} whose core pool size changes dynamically depending on the
041 * number of {@link ScheduledChore} scheduled. All of the threads in the core thread pool of the
042 * underlying {@link ScheduledThreadPoolExecutor} are set to be daemon threads.
043 * <p>
044 * The ChoreService provides the ability to schedule, cancel, and trigger instances of
045 * {@link ScheduledChore}. The ChoreService also provides the ability to check on the status of
046 * scheduled chores. The number of threads used by the ChoreService changes based on the scheduling
047 * load and whether or not the scheduled chores are executing on time. As more chores are scheduled,
048 * there may be a need to increase the number of threads if it is noticed that chores are no longer
049 * meeting their scheduled start times. On the other hand, as chores are cancelled, an attempt is
050 * made to reduce the number of running threads to see if chores can still meet their start times
051 * with a smaller thread pool.
052 * <p>
053 * When finished with a ChoreService it is good practice to call {@link ChoreService#shutdown()}.
054 * Calling this method ensures that all scheduled chores are cancelled and cleaned up properly.
055 */
056@InterfaceAudience.Public
057public class ChoreService implements ChoreServicer {
058  private static final Logger LOG = LoggerFactory.getLogger(ChoreService.class);
059
060  /**
061   * The minimum number of threads in the core pool of the underlying ScheduledThreadPoolExecutor
062   */
063  @InterfaceAudience.Private
064  public final static int MIN_CORE_POOL_SIZE = 1;
065
066  /**
067   * This thread pool is used to schedule all of the Chores
068   */
069  private final ScheduledThreadPoolExecutor scheduler;
070
071  /**
072   * Maps chores to their futures. Futures are used to control a chore's schedule
073   */
074  private final HashMap<ScheduledChore, ScheduledFuture<?>> scheduledChores;
075
076  /**
077   * Maps chores to Booleans which indicate whether or not a chore has caused an increase in the
078   * core pool size of the ScheduledThreadPoolExecutor. Each chore should only be allowed to
079   * increase the core pool size by 1 (otherwise a single long running chore whose execution is
080   * longer than its period would be able to spawn too many threads).
081   */
082  private final HashMap<ScheduledChore, Boolean> choresMissingStartTime;
083
084  /**
085   * The coreThreadPoolPrefix is the prefix that will be applied to all threads within the
086   * ScheduledThreadPoolExecutor. The prefix is typically related to the Server that the service is
087   * running on. The prefix is useful because it allows us to monitor how the thread pool of a
088   * particular service changes over time VIA thread dumps.
089   */
090  private final String coreThreadPoolPrefix;
091
092  /**
093   *
094   * @param coreThreadPoolPrefix Prefix that will be applied to the Thread name of all threads
095   *          spawned by this service
096   */
097  @InterfaceAudience.Private
098  @VisibleForTesting
099  public ChoreService(final String coreThreadPoolPrefix) {
100    this(coreThreadPoolPrefix, MIN_CORE_POOL_SIZE, false);
101  }
102
103  /**
104   * @param coreThreadPoolPrefix Prefix that will be applied to the Thread name of all threads
105   *          spawned by this service
106   * @param jitter Should chore service add some jitter for all of the scheduled chores. When set
107   *               to true this will add -10% to 10% jitter.
108   */
109  public ChoreService(final String coreThreadPoolPrefix, final boolean jitter) {
110    this(coreThreadPoolPrefix, MIN_CORE_POOL_SIZE, jitter);
111  }
112
113  /**
114   * @param coreThreadPoolPrefix Prefix that will be applied to the Thread name of all threads
115   *          spawned by this service
116   * @param corePoolSize The initial size to set the core pool of the ScheduledThreadPoolExecutor
117   *          to during initialization. The default size is 1, but specifying a larger size may be
118   *          beneficial if you know that 1 thread will not be enough.
119   * @param jitter Should chore service add some jitter for all of the scheduled chores. When set
120   *               to true this will add -10% to 10% jitter.
121   */
122  public ChoreService(final String coreThreadPoolPrefix, int corePoolSize, boolean jitter) {
123    this.coreThreadPoolPrefix = coreThreadPoolPrefix;
124    if (corePoolSize < MIN_CORE_POOL_SIZE)  {
125      corePoolSize = MIN_CORE_POOL_SIZE;
126    }
127
128    final ThreadFactory threadFactory = new ChoreServiceThreadFactory(coreThreadPoolPrefix);
129    if (jitter) {
130      scheduler = new JitterScheduledThreadPoolExecutorImpl(corePoolSize, threadFactory, 0.1);
131    } else {
132      scheduler = new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
133    }
134
135    scheduler.setRemoveOnCancelPolicy(true);
136    scheduledChores = new HashMap<>();
137    choresMissingStartTime = new HashMap<>();
138  }
139
140  /**
141   * @param chore Chore to be scheduled. If the chore is already scheduled with another ChoreService
142   *          instance, that schedule will be cancelled (i.e. a Chore can only ever be scheduled
143   *          with a single ChoreService instance).
144   * @return true when the chore was successfully scheduled. false when the scheduling failed
145   *         (typically occurs when a chore is scheduled during shutdown of service)
146   */
147  public synchronized boolean scheduleChore(ScheduledChore chore) {
148    if (chore == null) {
149      return false;
150    }
151
152    try {
153      chore.setChoreServicer(this);
154      ScheduledFuture<?> future =
155          scheduler.scheduleAtFixedRate(chore, chore.getInitialDelay(), chore.getPeriod(),
156            chore.getTimeUnit());
157      scheduledChores.put(chore, future);
158      return true;
159    } catch (Exception exception) {
160      if (LOG.isInfoEnabled()) {
161        LOG.info("Could not successfully schedule chore: " + chore.getName());
162      }
163      return false;
164    }
165  }
166
167  /**
168   * @param chore The Chore to be rescheduled. If the chore is not scheduled with this ChoreService
169   *          yet then this call is equivalent to a call to scheduleChore.
170   */
171  private synchronized void rescheduleChore(ScheduledChore chore) {
172    if (chore == null) return;
173
174    if (scheduledChores.containsKey(chore)) {
175      ScheduledFuture<?> future = scheduledChores.get(chore);
176      future.cancel(false);
177    }
178    scheduleChore(chore);
179  }
180
181  @InterfaceAudience.Private
182  @Override
183  public synchronized void cancelChore(ScheduledChore chore) {
184    cancelChore(chore, true);
185  }
186
187  @InterfaceAudience.Private
188  @Override
189  public synchronized void cancelChore(ScheduledChore chore, boolean mayInterruptIfRunning) {
190    if (chore != null && scheduledChores.containsKey(chore)) {
191      ScheduledFuture<?> future = scheduledChores.get(chore);
192      future.cancel(mayInterruptIfRunning);
193      scheduledChores.remove(chore);
194
195      // Removing a chore that was missing its start time means it may be possible
196      // to reduce the number of threads
197      if (choresMissingStartTime.containsKey(chore)) {
198        choresMissingStartTime.remove(chore);
199        requestCorePoolDecrease();
200      }
201    }
202  }
203
204  @InterfaceAudience.Private
205  @Override
206  public synchronized boolean isChoreScheduled(ScheduledChore chore) {
207    return chore != null && scheduledChores.containsKey(chore)
208        && !scheduledChores.get(chore).isDone();
209  }
210
211  @InterfaceAudience.Private
212  @Override
213  public synchronized boolean triggerNow(ScheduledChore chore) {
214    if (chore == null) {
215      return false;
216    } else {
217      rescheduleChore(chore);
218      return true;
219    }
220  }
221
222  /**
223   * @return number of chores that this service currently has scheduled
224   */
225  int getNumberOfScheduledChores() {
226    return scheduledChores.size();
227  }
228
229  /**
230   * @return number of chores that this service currently has scheduled that are missing their
231   *         scheduled start time
232   */
233  int getNumberOfChoresMissingStartTime() {
234    return choresMissingStartTime.size();
235  }
236
237  /**
238   * @return number of threads in the core pool of the underlying ScheduledThreadPoolExecutor
239   */
240  int getCorePoolSize() {
241    return scheduler.getCorePoolSize();
242  }
243
244  /**
245   * Custom ThreadFactory used with the ScheduledThreadPoolExecutor so that all the threads are
246   * daemon threads, and thus, don't prevent the JVM from shutting down
247   */
248  static class ChoreServiceThreadFactory implements ThreadFactory {
249    private final String threadPrefix;
250    private final static String THREAD_NAME_SUFFIX = ".Chore.";
251    private AtomicInteger threadNumber = new AtomicInteger(1);
252
253    /**
254     * @param threadPrefix The prefix given to all threads created by this factory
255     */
256    public ChoreServiceThreadFactory(final String threadPrefix) {
257      this.threadPrefix = threadPrefix;
258    }
259
260    @Override
261    public Thread newThread(Runnable r) {
262      Thread thread =
263          new Thread(r, threadPrefix + THREAD_NAME_SUFFIX + threadNumber.getAndIncrement());
264      thread.setDaemon(true);
265      return thread;
266    }
267  }
268
269  /**
270   * Represents a request to increase the number of core pool threads. Typically a request
271   * originates from the fact that the current core pool size is not sufficient to service all of
272   * the currently running Chores
273   * @return true when the request to increase the core pool size succeeds
274   */
275  private synchronized boolean requestCorePoolIncrease() {
276    // There is no point in creating more threads than scheduledChores.size since scheduled runs
277    // of the same chore cannot run concurrently (i.e. happen-before behavior is enforced
278    // amongst occurrences of the same chore).
279    if (scheduler.getCorePoolSize() < scheduledChores.size()) {
280      scheduler.setCorePoolSize(scheduler.getCorePoolSize() + 1);
281      printChoreServiceDetails("requestCorePoolIncrease");
282      return true;
283    }
284    return false;
285  }
286
287  /**
288   * Represents a request to decrease the number of core pool threads. Typically a request
289   * originates from the fact that the current core pool size is more than sufficient to service the
290   * running Chores.
291   */
292  private synchronized void requestCorePoolDecrease() {
293    if (scheduler.getCorePoolSize() > MIN_CORE_POOL_SIZE) {
294      scheduler.setCorePoolSize(scheduler.getCorePoolSize() - 1);
295      printChoreServiceDetails("requestCorePoolDecrease");
296    }
297  }
298
299  @InterfaceAudience.Private
300  @Override
301  public synchronized void onChoreMissedStartTime(ScheduledChore chore) {
302    if (chore == null || !scheduledChores.containsKey(chore)) return;
303
304    // If the chore has not caused an increase in the size of the core thread pool then request an
305    // increase. This allows each chore missing its start time to increase the core pool size by
306    // at most 1.
307    if (!choresMissingStartTime.containsKey(chore) || !choresMissingStartTime.get(chore)) {
308      choresMissingStartTime.put(chore, requestCorePoolIncrease());
309    }
310
311    // Must reschedule the chore to prevent unnecessary delays of chores in the scheduler. If
312    // the chore is NOT rescheduled, future executions of this chore will be delayed more and
313    // more on each iteration. This hurts us because the ScheduledThreadPoolExecutor allocates
314    // idle threads to chores based on how delayed they are.
315    rescheduleChore(chore);
316    printChoreDetails("onChoreMissedStartTime", chore);
317  }
318
319  /**
320   * shutdown the service. Any chores that are scheduled for execution will be cancelled. Any chores
321   * in the middle of execution will be interrupted and shutdown. This service will be unusable
322   * after this method has been called (i.e. future scheduling attempts will fail).
323   */
324  public synchronized void shutdown() {
325    scheduler.shutdownNow();
326    if (LOG.isInfoEnabled()) {
327      LOG.info("Chore service for: " + coreThreadPoolPrefix + " had " + scheduledChores.keySet()
328          + " on shutdown");
329    }
330    cancelAllChores(true);
331    scheduledChores.clear();
332    choresMissingStartTime.clear();
333  }
334
335  /**
336   * @return true when the service is shutdown and thus cannot be used anymore
337   */
338  public boolean isShutdown() {
339    return scheduler.isShutdown();
340  }
341
342  /**
343   * @return true when the service is shutdown and all threads have terminated
344   */
345  public boolean isTerminated() {
346    return scheduler.isTerminated();
347  }
348
349  private void cancelAllChores(final boolean mayInterruptIfRunning) {
350    ArrayList<ScheduledChore> choresToCancel = new ArrayList<>(scheduledChores.keySet().size());
351    // Build list of chores to cancel so we can iterate through a set that won't change
352    // as chores are cancelled. If we tried to cancel each chore while iterating through
353    // keySet the results would be undefined because the keySet would be changing
354    for (ScheduledChore chore : scheduledChores.keySet()) {
355      choresToCancel.add(chore);
356    }
357    for (ScheduledChore chore : choresToCancel) {
358      cancelChore(chore, mayInterruptIfRunning);
359    }
360    choresToCancel.clear();
361  }
362
363  /**
364   * Prints a summary of important details about the chore. Used for debugging purposes
365   */
366  private void printChoreDetails(final String header, ScheduledChore chore) {
367    LinkedHashMap<String, String> output = new LinkedHashMap<>();
368    output.put(header, "");
369    output.put("Chore name: ", chore.getName());
370    output.put("Chore period: ", Integer.toString(chore.getPeriod()));
371    output.put("Chore timeBetweenRuns: ", Long.toString(chore.getTimeBetweenRuns()));
372
373    for (Entry<String, String> entry : output.entrySet()) {
374      if (LOG.isTraceEnabled()) LOG.trace(entry.getKey() + entry.getValue());
375    }
376  }
377
378  /**
379   * Prints a summary of important details about the service. Used for debugging purposes
380   */
381  private void printChoreServiceDetails(final String header) {
382    LinkedHashMap<String, String> output = new LinkedHashMap<>();
383    output.put(header, "");
384    output.put("ChoreService corePoolSize: ", Integer.toString(getCorePoolSize()));
385    output.put("ChoreService scheduledChores: ", Integer.toString(getNumberOfScheduledChores()));
386    output.put("ChoreService missingStartTimeCount: ",
387      Integer.toString(getNumberOfChoresMissingStartTime()));
388
389    for (Entry<String, String> entry : output.entrySet()) {
390      if (LOG.isTraceEnabled()) LOG.trace(entry.getKey() + entry.getValue());
391    }
392  }
393}