Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-19677 Miscellaneous HFileCleaner Improvements (BELUGA BEHR)

This commit is contained in:
tedyu 2017-12-31 09:47:53 -08:00
parent 5b3513a5ee
commit 2f5ca9010b
1 changed files with 22 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/HFileCleaner.java
View File

@ -168,17 +168,13 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
if (task.fileLength >= this.throttlePoint) { if (task.fileLength >= this.throttlePoint) {
if (!this.largeFileQueue.offer(task)) { if (!this.largeFileQueue.offer(task)) {
// should never arrive here as long as we use PriorityQueue // should never arrive here as long as we use PriorityQueue
if (LOG.isTraceEnabled()) {
LOG.trace("Large file deletion queue is full"); LOG.trace("Large file deletion queue is full");
}
return false; return false;
} }
} else { } else {
if (!this.smallFileQueue.offer(task)) { if (!this.smallFileQueue.offer(task)) {
// should never arrive here as long as we use PriorityQueue // should never arrive here as long as we use PriorityQueue
if (LOG.isTraceEnabled()) {
LOG.trace("Small file deletion queue is full"); LOG.trace("Small file deletion queue is full");
}
return false; return false;
} }
} }
@ -208,7 +204,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
large.setDaemon(true); large.setDaemon(true);
large.setName(n + "-HFileCleaner.large." + i + "-" + System.currentTimeMillis()); large.setName(n + "-HFileCleaner.large." + i + "-" + System.currentTimeMillis());
large.start(); large.start();
LOG.debug("Starting hfile cleaner for large files: " + large.getName()); LOG.debug("Starting hfile cleaner for large files: {}", large);
threads.add(large); threads.add(large);
} }
@ -223,7 +219,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
small.setDaemon(true); small.setDaemon(true);
small.setName(n + "-HFileCleaner.small." + i + "-" + System.currentTimeMillis()); small.setName(n + "-HFileCleaner.small." + i + "-" + System.currentTimeMillis());
small.start(); small.start();
LOG.debug("Starting hfile cleaner for small files: " + small.getName()); LOG.debug("Starting hfile cleaner for small files: {}", small);
threads.add(small); threads.add(small);
} }
} }
@ -235,20 +231,16 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
try { try {
task = queue.take(); task = queue.take();
} catch (InterruptedException e) { } catch (InterruptedException e) {
if (LOG.isDebugEnabled()) {
LOG.debug("Interrupted while trying to take a task from queue", e); LOG.debug("Interrupted while trying to take a task from queue", e);
}
break; break;
} }
if (task != null) { if (task != null) {
if (LOG.isDebugEnabled()) { LOG.debug("Removing: {} from archive", task.filePath);
LOG.debug("Removing: " + task.filePath + " from archive");
}
boolean succeed; boolean succeed;
try { try {
succeed = this.fs.delete(task.filePath, false); succeed = this.fs.delete(task.filePath, false);
} catch (IOException e) { } catch (IOException e) {
LOG.warn("Failed to delete file " + task.filePath, e); LOG.warn("Failed to delete file {}", task.filePath, e);
succeed = false; succeed = false;
} }
task.setResult(succeed); task.setResult(succeed);
@ -258,9 +250,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
} }
} }
} finally { } finally {
if (LOG.isDebugEnabled()) { LOG.debug("Exit thread: {}", Thread.currentThread());
LOG.debug("Exit thread: " + Thread.currentThread());
}
} }
} }
@ -275,9 +265,9 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
} else { } else {
if (deletedSmallFiles.get() == Long.MAX_VALUE) { if (deletedSmallFiles.get() == Long.MAX_VALUE) {
LOG.info("Deleted more than Long.MAX_VALUE small files, reset counter to 0"); LOG.info("Deleted more than Long.MAX_VALUE small files, reset counter to 0");
deletedSmallFiles.set(0L);; deletedSmallFiles.set(0L);
} }
if (fromLargeQueue && LOG.isTraceEnabled()) { if (fromLargeQueue) {
LOG.trace("Stolen a small file deletion task in large file thread"); LOG.trace("Stolen a small file deletion task in large file thread");
} }
deletedSmallFiles.incrementAndGet(); deletedSmallFiles.incrementAndGet();
@ -289,9 +279,7 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
*/ */
private void stopHFileDeleteThreads() { private void stopHFileDeleteThreads() {
running = false; running = false;
if (LOG.isDebugEnabled()) {
LOG.debug("Stopping file delete threads"); LOG.debug("Stopping file delete threads");
}
for(Thread thread: threads){ for(Thread thread: threads){
thread.interrupt(); thread.interrupt();
} }
@ -397,12 +385,8 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
// record the left over tasks // record the left over tasks
List<HFileDeleteTask> leftOverTasks = List<HFileDeleteTask> leftOverTasks =
new ArrayList<>(largeFileQueue.size() + smallFileQueue.size()); new ArrayList<>(largeFileQueue.size() + smallFileQueue.size());
for (HFileDeleteTask task : largeFileQueue) { leftOverTasks.addAll(largeFileQueue);
leftOverTasks.add(task); leftOverTasks.addAll(smallFileQueue);
}
for (HFileDeleteTask task : smallFileQueue) {
leftOverTasks.add(task);
}
largeFileQueue = new StealJobQueue<>(largeQueueInitSize, smallQueueInitSize, COMPARATOR); largeFileQueue = new StealJobQueue<>(largeQueueInitSize, smallQueueInitSize, COMPARATOR);
smallFileQueue = largeFileQueue.getStealFromQueue(); smallFileQueue = largeFileQueue.getStealFromQueue();
threads.clear(); threads.clear();
@ -423,39 +407,39 @@ public class HFileCleaner extends CleanerChore<BaseHFileCleanerDelegate> {
int throttlePoint = int throttlePoint =
conf.getInt(HFILE_DELETE_THROTTLE_THRESHOLD, DEFAULT_HFILE_DELETE_THROTTLE_THRESHOLD); conf.getInt(HFILE_DELETE_THROTTLE_THRESHOLD, DEFAULT_HFILE_DELETE_THROTTLE_THRESHOLD);
if (throttlePoint != this.throttlePoint) { if (throttlePoint != this.throttlePoint) {
LOG.debug("Updating throttle point, from " + this.throttlePoint + " to " + throttlePoint); LOG.debug("Updating throttle point, from {} to {}", this.throttlePoint, throttlePoint);
this.throttlePoint = throttlePoint; this.throttlePoint = throttlePoint;
updated = true; updated = true;
} }
int largeQueueInitSize = int largeQueueInitSize =
conf.getInt(LARGE_HFILE_QUEUE_INIT_SIZE, DEFAULT_LARGE_HFILE_QUEUE_INIT_SIZE); conf.getInt(LARGE_HFILE_QUEUE_INIT_SIZE, DEFAULT_LARGE_HFILE_QUEUE_INIT_SIZE);
if (largeQueueInitSize != this.largeQueueInitSize) { if (largeQueueInitSize != this.largeQueueInitSize) {
LOG.debug("Updating largeQueueInitSize, from " + this.largeQueueInitSize + " to " LOG.debug("Updating largeQueueInitSize, from {} to {}", this.largeQueueInitSize,
+ largeQueueInitSize); largeQueueInitSize);
this.largeQueueInitSize = largeQueueInitSize; this.largeQueueInitSize = largeQueueInitSize;
updated = true; updated = true;
} }
int smallQueueInitSize = int smallQueueInitSize =
conf.getInt(SMALL_HFILE_QUEUE_INIT_SIZE, DEFAULT_SMALL_HFILE_QUEUE_INIT_SIZE); conf.getInt(SMALL_HFILE_QUEUE_INIT_SIZE, DEFAULT_SMALL_HFILE_QUEUE_INIT_SIZE);
if (smallQueueInitSize != this.smallQueueInitSize) { if (smallQueueInitSize != this.smallQueueInitSize) {
LOG.debug("Updating smallQueueInitSize, from " + this.smallQueueInitSize + " to " LOG.debug("Updating smallQueueInitSize, from {} to {}", this.smallQueueInitSize,
+ smallQueueInitSize); smallQueueInitSize);
this.smallQueueInitSize = smallQueueInitSize; this.smallQueueInitSize = smallQueueInitSize;
updated = true; updated = true;
} }
int largeFileDeleteThreadNumber = int largeFileDeleteThreadNumber =
conf.getInt(LARGE_HFILE_DELETE_THREAD_NUMBER, DEFAULT_LARGE_HFILE_DELETE_THREAD_NUMBER); conf.getInt(LARGE_HFILE_DELETE_THREAD_NUMBER, DEFAULT_LARGE_HFILE_DELETE_THREAD_NUMBER);
if (largeFileDeleteThreadNumber != this.largeFileDeleteThreadNumber) { if (largeFileDeleteThreadNumber != this.largeFileDeleteThreadNumber) {
LOG.debug("Updating largeFileDeleteThreadNumber, from " + this.largeFileDeleteThreadNumber LOG.debug("Updating largeFileDeleteThreadNumber, from {} to {}",
+ " to " + largeFileDeleteThreadNumber); this.largeFileDeleteThreadNumber, largeFileDeleteThreadNumber);
this.largeFileDeleteThreadNumber = largeFileDeleteThreadNumber; this.largeFileDeleteThreadNumber = largeFileDeleteThreadNumber;
updated = true; updated = true;
} }
int smallFileDeleteThreadNumber = int smallFileDeleteThreadNumber =
conf.getInt(SMALL_HFILE_DELETE_THREAD_NUMBER, DEFAULT_SMALL_HFILE_DELETE_THREAD_NUMBER); conf.getInt(SMALL_HFILE_DELETE_THREAD_NUMBER, DEFAULT_SMALL_HFILE_DELETE_THREAD_NUMBER);
if (smallFileDeleteThreadNumber != this.smallFileDeleteThreadNumber) { if (smallFileDeleteThreadNumber != this.smallFileDeleteThreadNumber) {
LOG.debug("Updating smallFileDeleteThreadNumber, from " + this.smallFileDeleteThreadNumber LOG.debug("Updating smallFileDeleteThreadNumber, from {} to {}",
+ " to " + smallFileDeleteThreadNumber); this.smallFileDeleteThreadNumber, smallFileDeleteThreadNumber);
this.smallFileDeleteThreadNumber = smallFileDeleteThreadNumber; this.smallFileDeleteThreadNumber = smallFileDeleteThreadNumber;
updated = true; updated = true;
} }