1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.master.cleaner;
19
20 import com.google.common.annotations.VisibleForTesting;
21 import java.io.IOException;
22 import java.util.ArrayList;
23 import java.util.List;
24 import java.util.Map;
25 import java.util.concurrent.BlockingQueue;
26 import java.util.concurrent.TimeUnit;
27 import java.util.concurrent.atomic.AtomicLong;
28 import org.apache.commons.logging.Log;
29 import org.apache.commons.logging.LogFactory;
30 import org.apache.hadoop.conf.Configuration;
31 import org.apache.hadoop.fs.FileStatus;
32 import org.apache.hadoop.fs.FileSystem;
33 import org.apache.hadoop.fs.Path;
34 import org.apache.hadoop.hbase.Stoppable;
35 import org.apache.hadoop.hbase.classification.InterfaceAudience;
36 import org.apache.hadoop.hbase.conf.ConfigurationObserver;
37 import org.apache.hadoop.hbase.io.HFileLink;
38 import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
39 import org.apache.hadoop.hbase.util.StealJobQueue;
40
41
42
43
44
45 @InterfaceAudience.Private
46 public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> implements
47 ConfigurationObserver {
48
49 public static final String MASTER_HFILE_CLEANER_PLUGINS = "hbase.master.hfilecleaner.plugins";
50
51 public HFileCleaner(final int period, final Stoppable stopper, Configuration conf, FileSystem fs,
52 Path directory) {
53 this(period, stopper, conf, fs, directory, null);
54 }
55
56
57 public final static String HFILE_DELETE_THROTTLE_THRESHOLD =
58 "hbase.regionserver.thread.hfilecleaner.throttle";
59 public final static int DEFAULT_HFILE_DELETE_THROTTLE_THRESHOLD = 64 * 1024 * 1024;
60
61
62 public final static String LARGE_HFILE_QUEUE_INIT_SIZE =
63 "hbase.regionserver.hfilecleaner.large.queue.size";
64 public final static int DEFAULT_LARGE_HFILE_QUEUE_INIT_SIZE = 10240;
65
66
67 public final static String SMALL_HFILE_QUEUE_INIT_SIZE =
68 "hbase.regionserver.hfilecleaner.small.queue.size";
69 public final static int DEFAULT_SMALL_HFILE_QUEUE_INIT_SIZE = 10240;
70
71
72 public final static String LARGE_HFILE_DELETE_THREAD_NUMBER =
73 "hbase.regionserver.hfilecleaner.large.thread.count";
74 public final static int DEFAULT_LARGE_HFILE_DELETE_THREAD_NUMBER = 1;
75
76
77 public final static String SMALL_HFILE_DELETE_THREAD_NUMBER =
78 "hbase.regionserver.hfilecleaner.small.thread.count";
79 public final static int DEFAULT_SMALL_HFILE_DELETE_THREAD_NUMBER = 1;
80
81 public static final String HFILE_DELETE_THREAD_TIMEOUT_MSEC =
82 "hbase.regionserver.hfilecleaner.thread.timeout.msec";
83 @VisibleForTesting
84 static final long DEFAULT_HFILE_DELETE_THREAD_TIMEOUT_MSEC = 60 * 1000L;
85
86 public static final String HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC =
87 "hbase.regionserver.hfilecleaner.thread.check.interval.msec";
88 @VisibleForTesting
89 static final long DEFAULT_HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC = 1000L;
90
91 private static final Log LOG = LogFactory.getLog(HFileCleaner.class);
92
93 StealJobQueue<HFileDeleteTask> largeFileQueue;
94 BlockingQueue<HFileDeleteTask> smallFileQueue;
95 private int throttlePoint;
96 private int largeQueueInitSize;
97 private int smallQueueInitSize;
98 private int largeFileDeleteThreadNumber;
99 private int smallFileDeleteThreadNumber;
100 private long cleanerThreadTimeoutMsec;
101 private long cleanerThreadCheckIntervalMsec;
102 private List<Thread> threads = new ArrayList<Thread>();
103 private boolean running;
104
105 private AtomicLong deletedLargeFiles = new AtomicLong();
106 private AtomicLong deletedSmallFiles = new AtomicLong();
107
108
109
110
111
112
113
114
115
116 public HFileCleaner(final int period, final Stoppable stopper, Configuration conf, FileSystem fs,
117 Path directory, Map<String, Object> params) {
118 super("HFileCleaner", period, stopper, conf, fs,
119 directory, MASTER_HFILE_CLEANER_PLUGINS, params);
120 throttlePoint =
121 conf.getInt(HFILE_DELETE_THROTTLE_THRESHOLD, DEFAULT_HFILE_DELETE_THROTTLE_THRESHOLD);
122 largeQueueInitSize =
123 conf.getInt(LARGE_HFILE_QUEUE_INIT_SIZE, DEFAULT_LARGE_HFILE_QUEUE_INIT_SIZE);
124 smallQueueInitSize =
125 conf.getInt(SMALL_HFILE_QUEUE_INIT_SIZE, DEFAULT_SMALL_HFILE_QUEUE_INIT_SIZE);
126 largeFileQueue = new StealJobQueue<>(largeQueueInitSize, smallQueueInitSize);
127 smallFileQueue = largeFileQueue.getStealFromQueue();
128 largeFileDeleteThreadNumber =
129 conf.getInt(LARGE_HFILE_DELETE_THREAD_NUMBER, DEFAULT_LARGE_HFILE_DELETE_THREAD_NUMBER);
130 smallFileDeleteThreadNumber =
131 conf.getInt(SMALL_HFILE_DELETE_THREAD_NUMBER, DEFAULT_SMALL_HFILE_DELETE_THREAD_NUMBER);
132 cleanerThreadTimeoutMsec =
133 conf.getLong(HFILE_DELETE_THREAD_TIMEOUT_MSEC, DEFAULT_HFILE_DELETE_THREAD_TIMEOUT_MSEC);
134 cleanerThreadCheckIntervalMsec =
135 conf.getLong(HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC,
136 DEFAULT_HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC);
137 startHFileDeleteThreads();
138 }
139
140 @Override
141 protected boolean validate(Path file) {
142 if (HFileLink.isBackReferencesDir(file) || HFileLink.isBackReferencesDir(file.getParent())) {
143 return true;
144 }
145 return StoreFileInfo.validateStoreFileName(file.getName());
146 }
147
148
149
150
151 public List<BaseHFileCleanerDelegate> getDelegatesForTesting() {
152 return this.cleanersChain;
153 }
154
155 @Override
156 public int deleteFiles(Iterable<FileStatus> filesToDelete) {
157 int deletedFiles = 0;
158 List<HFileDeleteTask> tasks = new ArrayList<HFileDeleteTask>();
159
160 for (FileStatus file : filesToDelete) {
161 HFileDeleteTask task = deleteFile(file);
162 if (task != null) {
163 tasks.add(task);
164 }
165 }
166
167 for (HFileDeleteTask task : tasks) {
168 if (task.getResult(cleanerThreadCheckIntervalMsec)) {
169 deletedFiles++;
170 }
171 }
172 return deletedFiles;
173 }
174
175
176
177
178
179
180 private HFileDeleteTask deleteFile(FileStatus file) {
181 HFileDeleteTask task = new HFileDeleteTask(file, cleanerThreadTimeoutMsec);
182 boolean enqueued = dispatch(task);
183 return enqueued ? task : null;
184 }
185
186 private boolean dispatch(HFileDeleteTask task) {
187 if (task.fileLength >= this.throttlePoint) {
188 if (!this.largeFileQueue.offer(task)) {
189
190 if (LOG.isTraceEnabled()) {
191 LOG.trace("Large file deletion queue is full");
192 }
193 return false;
194 }
195 } else {
196 if (!this.smallFileQueue.offer(task)) {
197
198 if (LOG.isTraceEnabled()) {
199 LOG.trace("Small file deletion queue is full");
200 }
201 return false;
202 }
203 }
204 return true;
205 }
206
207 @Override
208 public void cleanup() {
209 super.cleanup();
210 stopHFileDeleteThreads();
211 }
212
213
214
215
216 private void startHFileDeleteThreads() {
217 final String n = Thread.currentThread().getName();
218 running = true;
219
220 for (int i = 0; i < largeFileDeleteThreadNumber; i++) {
221 Thread large = new Thread() {
222 @Override
223 public void run() {
224 consumerLoop(largeFileQueue);
225 }
226 };
227 large.setDaemon(true);
228 large.setName(n + "-HFileCleaner.large." + i + "-" + System.currentTimeMillis());
229 large.start();
230 LOG.debug("Starting hfile cleaner for large files: " + large.getName());
231 threads.add(large);
232 }
233
234
235 for (int i = 0; i < smallFileDeleteThreadNumber; i++) {
236 Thread small = new Thread() {
237 @Override
238 public void run() {
239 consumerLoop(smallFileQueue);
240 }
241 };
242 small.setDaemon(true);
243 small.setName(n + "-HFileCleaner.small." + i + "-" + System.currentTimeMillis());
244 small.start();
245 LOG.debug("Starting hfile cleaner for small files: " + small.getName());
246 threads.add(small);
247 }
248 }
249
250 protected void consumerLoop(BlockingQueue<HFileDeleteTask> queue) {
251 try {
252 while (running) {
253 HFileDeleteTask task = null;
254 try {
255 task = queue.take();
256 } catch (InterruptedException e) {
257 if (LOG.isDebugEnabled()) {
258 LOG.debug("Interrupted while trying to take a task from queue", e);
259 }
260 break;
261 }
262 if (task != null) {
263 if (LOG.isDebugEnabled()) {
264 LOG.debug("Removing: " + task.filePath + " from archive");
265 }
266 boolean succeed;
267 try {
268 succeed = this.fs.delete(task.filePath, false);
269 } catch (IOException e) {
270 LOG.warn("Failed to delete file " + task.filePath, e);
271 succeed = false;
272 }
273 task.setResult(succeed);
274 if (succeed) {
275 countDeletedFiles(task.fileLength >= throttlePoint, queue == largeFileQueue);
276 }
277 }
278 }
279 } finally {
280 if (LOG.isDebugEnabled()) {
281 LOG.debug("Exit thread: " + Thread.currentThread());
282 }
283 }
284 }
285
286
287 private void countDeletedFiles(boolean isLargeFile, boolean fromLargeQueue) {
288 if (isLargeFile) {
289 if (deletedLargeFiles.get() == Long.MAX_VALUE) {
290 LOG.info("Deleted more than Long.MAX_VALUE large files, reset counter to 0");
291 deletedLargeFiles.set(0L);
292 }
293 deletedLargeFiles.incrementAndGet();
294 } else {
295 if (deletedSmallFiles.get() == Long.MAX_VALUE) {
296 LOG.info("Deleted more than Long.MAX_VALUE small files, reset counter to 0");
297 deletedSmallFiles.set(0L);
298 }
299 if (fromLargeQueue && LOG.isTraceEnabled()) {
300 LOG.trace("Stolen a small file deletion task in large file thread");
301 }
302 deletedSmallFiles.incrementAndGet();
303 }
304 }
305
306
307
308
309 private void stopHFileDeleteThreads() {
310 running = false;
311 if (LOG.isDebugEnabled()) {
312 LOG.debug("Stopping file delete threads");
313 }
314 for(Thread thread: threads){
315 thread.interrupt();
316 }
317 }
318
319 static class HFileDeleteTask implements Comparable<HFileDeleteTask> {
320
321 boolean done = false;
322 boolean result;
323 final Path filePath;
324 final long fileLength;
325 final long timeoutMsec;
326
327 public HFileDeleteTask(FileStatus file, long timeoutMsec) {
328 this.filePath = file.getPath();
329 this.fileLength = file.getLen();
330 this.timeoutMsec = timeoutMsec;
331 }
332
333 public synchronized void setResult(boolean result) {
334 this.done = true;
335 this.result = result;
336 notify();
337 }
338
339 public synchronized boolean getResult(long waitIfNotFinished) {
340 long waitTimeMsec = 0;
341 try {
342 while (!done) {
343 long startTimeNanos = System.nanoTime();
344 wait(waitIfNotFinished);
345 waitTimeMsec += TimeUnit.MILLISECONDS.convert(System.nanoTime() - startTimeNanos,
346 TimeUnit.NANOSECONDS);
347 if (done) {
348 return this.result;
349 }
350 if (waitTimeMsec > timeoutMsec) {
351 LOG.warn("Wait more than " + timeoutMsec + " ms for deleting " + this.filePath
352 + ", exit...");
353 return false;
354 }
355 }
356 } catch (InterruptedException e) {
357 LOG.warn("Interrupted while waiting for result of deleting " + filePath
358 + ", will return false", e);
359 return false;
360 }
361 return this.result;
362 }
363
364 @Override
365 public int compareTo(HFileDeleteTask o) {
366 long sub = this.fileLength - o.fileLength;
367
368
369 return (sub > 0) ? -1 : (sub < 0 ? 1 : 0);
370 }
371
372 @Override
373 public boolean equals(Object o) {
374 if (this == o) {
375 return true;
376 }
377 if (o == null || !(o instanceof HFileDeleteTask)) {
378 return false;
379 }
380 HFileDeleteTask otherTask = (HFileDeleteTask) o;
381 return this.filePath.equals(otherTask.filePath) && (this.fileLength == otherTask.fileLength);
382 }
383
384 @Override
385 public int hashCode() {
386 return filePath.hashCode();
387 }
388 }
389
390 @VisibleForTesting
391 public List<Thread> getCleanerThreads() {
392 return threads;
393 }
394
395 @VisibleForTesting
396 public long getNumOfDeletedLargeFiles() {
397 return deletedLargeFiles.get();
398 }
399
400 @VisibleForTesting
401 public long getNumOfDeletedSmallFiles() {
402 return deletedSmallFiles.get();
403 }
404
405 @VisibleForTesting
406 public long getLargeQueueInitSize() {
407 return largeQueueInitSize;
408 }
409
410 @VisibleForTesting
411 public long getSmallQueueInitSize() {
412 return smallQueueInitSize;
413 }
414
415 @VisibleForTesting
416 public long getThrottlePoint() {
417 return throttlePoint;
418 }
419
420 @VisibleForTesting
421 long getCleanerThreadTimeoutMsec() {
422 return cleanerThreadTimeoutMsec;
423 }
424
425 @VisibleForTesting
426 long getCleanerThreadCheckIntervalMsec() {
427 return cleanerThreadCheckIntervalMsec;
428 }
429
430 @Override
431 public void onConfigurationChange(Configuration conf) {
432 if (!checkAndUpdateConfigurations(conf)) {
433 LOG.debug("Update configuration triggered but nothing changed for this cleaner");
434 return;
435 }
436 stopHFileDeleteThreads();
437
438 List<HFileDeleteTask> leftOverTasks =
439 new ArrayList<>(largeFileQueue.size() + smallFileQueue.size());
440 for (HFileDeleteTask task : largeFileQueue) {
441 leftOverTasks.add(task);
442 }
443 for (HFileDeleteTask task : smallFileQueue) {
444 leftOverTasks.add(task);
445 }
446 largeFileQueue = new StealJobQueue<>(largeQueueInitSize, smallQueueInitSize);
447 smallFileQueue = largeFileQueue.getStealFromQueue();
448 threads.clear();
449 startHFileDeleteThreads();
450
451 for (HFileDeleteTask task : leftOverTasks) {
452 dispatch(task);
453 }
454 }
455
456
457
458
459
460
461 private boolean checkAndUpdateConfigurations(Configuration conf) {
462 boolean updated = false;
463 int throttlePoint =
464 conf.getInt(HFILE_DELETE_THROTTLE_THRESHOLD, DEFAULT_HFILE_DELETE_THROTTLE_THRESHOLD);
465 if (throttlePoint != this.throttlePoint) {
466 LOG.debug("Updating throttle point, from " + this.throttlePoint + " to " + throttlePoint);
467 this.throttlePoint = throttlePoint;
468 updated = true;
469 }
470 int largeQueueInitSize =
471 conf.getInt(LARGE_HFILE_QUEUE_INIT_SIZE, DEFAULT_LARGE_HFILE_QUEUE_INIT_SIZE);
472 if (largeQueueInitSize != this.largeQueueInitSize) {
473 LOG.debug("Updating largeQueueInitSize, from " + this.largeQueueInitSize + " to "
474 + largeQueueInitSize);
475 this.largeQueueInitSize = largeQueueInitSize;
476 updated = true;
477 }
478 int smallQueueInitSize =
479 conf.getInt(SMALL_HFILE_QUEUE_INIT_SIZE, DEFAULT_SMALL_HFILE_QUEUE_INIT_SIZE);
480 if (smallQueueInitSize != this.smallQueueInitSize) {
481 LOG.debug("Updating smallQueueInitSize, from " + this.smallQueueInitSize + " to "
482 + smallQueueInitSize);
483 this.smallQueueInitSize = smallQueueInitSize;
484 updated = true;
485 }
486 int largeFileDeleteThreadNumber =
487 conf.getInt(LARGE_HFILE_DELETE_THREAD_NUMBER, DEFAULT_LARGE_HFILE_DELETE_THREAD_NUMBER);
488 if (largeFileDeleteThreadNumber != this.largeFileDeleteThreadNumber) {
489 LOG.debug("Updating largeFileDeleteThreadNumber, from " + this.largeFileDeleteThreadNumber
490 + " to " + largeFileDeleteThreadNumber);
491 this.largeFileDeleteThreadNumber = largeFileDeleteThreadNumber;
492 updated = true;
493 }
494 int smallFileDeleteThreadNumber =
495 conf.getInt(SMALL_HFILE_DELETE_THREAD_NUMBER, DEFAULT_SMALL_HFILE_DELETE_THREAD_NUMBER);
496 if (smallFileDeleteThreadNumber != this.smallFileDeleteThreadNumber) {
497 LOG.debug("Updating smallFileDeleteThreadNumber, from " + this.smallFileDeleteThreadNumber
498 + " to " + smallFileDeleteThreadNumber);
499 this.smallFileDeleteThreadNumber = smallFileDeleteThreadNumber;
500 updated = true;
501 }
502 long cleanerThreadTimeoutMsec =
503 conf.getLong(HFILE_DELETE_THREAD_TIMEOUT_MSEC, DEFAULT_HFILE_DELETE_THREAD_TIMEOUT_MSEC);
504 if (cleanerThreadTimeoutMsec != this.cleanerThreadTimeoutMsec) {
505 this.cleanerThreadTimeoutMsec = cleanerThreadTimeoutMsec;
506 updated = true;
507 }
508 long cleanerThreadCheckIntervalMsec =
509 conf.getLong(HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC,
510 DEFAULT_HFILE_DELETE_THREAD_CHECK_INTERVAL_MSEC);
511 if (cleanerThreadCheckIntervalMsec != this.cleanerThreadCheckIntervalMsec) {
512 this.cleanerThreadCheckIntervalMsec = cleanerThreadCheckIntervalMsec;
513 updated = true;
514 }
515 return updated;
516 }
517 }