hbase-server/src/main/java/org/apache/hadoop/hbase/master/region/MasterRegionFlusherAndCompactor.java - hbase - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.hadoop.hbase.master.region;

 import java.io.Closeable;
 import java.io.IOException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Abortable;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.log.HBaseMarkers;
 import org.apache.hadoop.hbase.procedure2.store.region.RegionProcedureStore;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.regionserver.HStore;
 import org.apache.hadoop.hbase.regionserver.Store;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.HFileArchiveUtil;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;

 /**
  * As long as there is no RegionServerServices for a master local region, we need implement the
  * flush and compaction logic by our own.
  * <p/>
  * The flush logic is very simple, every time after calling a modification method in
  * {@link RegionProcedureStore}, we will call the {@link #onUpdate()} method below, and in this
  * method, we will check the memstore size and if it is above the flush size, we will call
  * {@link HRegion#flush(boolean)} to force flush all stores.
  * <p/>
  * And for compaction, the logic is also very simple. After flush, we will check the store file
  * count, if it is above the compactMin, we will do a major compaction.
  */
 @InterfaceAudience.Private
 class MasterRegionFlusherAndCompactor implements Closeable {

   private static final Logger LOG = LoggerFactory.getLogger(MasterRegionFlusherAndCompactor.class);

   private final Configuration conf;

   private final Abortable abortable;

   private final HRegion region;

   // as we can only count this outside the region's write/flush process so it is not accurate, but
   // it is enough.
   private final AtomicLong changesAfterLastFlush = new AtomicLong(0);

   private final long flushSize;

   private final long flushPerChanges;

   private final long flushIntervalMs;

   private final int compactMin;

   private final Path globalArchivePath;

   private final String archivedHFileSuffix;

   private final Thread flushThread;

   private final Lock flushLock = new ReentrantLock();

   private final Condition flushCond = flushLock.newCondition();

   private boolean flushRequest = false;

   private long lastFlushTime;

   private final ExecutorService compactExecutor;

   private final Lock compactLock = new ReentrantLock();

   private boolean compactRequest = false;

   private volatile boolean closed = false;

   MasterRegionFlusherAndCompactor(Configuration conf, Abortable abortable, HRegion region,
     long flushSize, long flushPerChanges, long flushIntervalMs, int compactMin,
     Path globalArchivePath, String archivedHFileSuffix) {
     this.conf = conf;
     this.abortable = abortable;
     this.region = region;
     this.flushSize = flushSize;
     this.flushPerChanges = flushPerChanges;
     this.flushIntervalMs = flushIntervalMs;
     this.compactMin = compactMin;
     this.globalArchivePath = globalArchivePath;
     this.archivedHFileSuffix = archivedHFileSuffix;
     flushThread = new Thread(this::flushLoop, region.getRegionInfo().getTable() + "-Flusher");
     flushThread.setDaemon(true);
     flushThread.start();
     compactExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
       .setNameFormat(region.getRegionInfo().getTable() + "-Store-Compactor").setDaemon(true)
       .build());
     LOG.info("Constructor flushSize={}, flushPerChanges={}, flushIntervalMs={}, compactMin={}",
       flushSize, flushPerChanges, flushIntervalMs, compactMin);
   }

   // inject our flush related configurations
   static void setupConf(Configuration conf, long flushSize, long flushPerChanges,
     long flushIntervalMs) {
     conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, flushSize);
     conf.setLong(HRegion.MEMSTORE_FLUSH_PER_CHANGES, flushPerChanges);
     conf.setLong(HRegion.MEMSTORE_PERIODIC_FLUSH_INTERVAL, flushIntervalMs);
     LOG.info("Injected flushSize={}, flushPerChanges={}, flushIntervalMs={}", flushSize,
       flushPerChanges, flushIntervalMs);
   }

   private void moveHFileToGlobalArchiveDir() throws IOException {
     FileSystem fs = region.getRegionFileSystem().getFileSystem();
     for (HStore store : region.getStores()) {
       store.closeAndArchiveCompactedFiles();
       Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, region.getRegionInfo(),
         store.getColumnFamilyDescriptor().getName());
       Path globalStoreArchiveDir = HFileArchiveUtil.getStoreArchivePathForArchivePath(
         globalArchivePath, region.getRegionInfo(), store.getColumnFamilyDescriptor().getName());
       try {
         if (fs.exists(storeArchiveDir)) {
           MasterRegionUtils.moveFilesUnderDir(fs, storeArchiveDir, globalStoreArchiveDir,
             archivedHFileSuffix);
         } else {
           LOG.warn(
             "Archived dir {} does not exist, there is no need to move archived hfiles from {} "
               + "to global dir {} .",
             storeArchiveDir, storeArchiveDir, globalStoreArchiveDir);
         }
       } catch (IOException e) {
         LOG.warn("Failed to move archived hfiles from {} to global dir {}", storeArchiveDir,
           globalStoreArchiveDir, e);
       }
     }
   }

   private void compact() {
     try {
       region.compact(true);
       moveHFileToGlobalArchiveDir();
     } catch (IOException e) {
       LOG.error("Failed to compact master local region", e);
     }
     compactLock.lock();
     try {
       if (needCompaction()) {
         compactExecutor.execute(this::compact);
       } else {
         compactRequest = false;
       }
     } finally {
       compactLock.unlock();
     }
   }

   private boolean needCompaction() {
     for (Store store : region.getStores()) {
       if (store.getStorefilesCount() >= compactMin) {
         return true;
       }
     }
     return false;
   }

   private void flushLoop() {
     recordLastFlushTime();
     while (!closed) {
       flushLock.lock();
       try {
         while (!flushRequest) {
           long waitTimeMs = lastFlushTime + flushIntervalMs - EnvironmentEdgeManager.currentTime();
           if (waitTimeMs <= 0) {
             flushRequest = true;
             break;
           }
           flushCond.await(waitTimeMs, TimeUnit.MILLISECONDS);
           if (closed) {
             return;
           }
         }
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         continue;
       } finally {
         flushLock.unlock();
       }
       assert flushRequest;
       resetChangesAfterLastFlush();
       try {
         region.flush(true);
         recordLastFlushTime();
       } catch (IOException e) {
         LOG.error(HBaseMarkers.FATAL, "Failed to flush master local region, aborting...", e);
         abortable.abort("Failed to flush master local region", e);
         return;
       }
       compactLock.lock();
       try {
         if (!compactRequest && needCompaction()) {
           compactRequest = true;
           compactExecutor.execute(this::compact);
         }
       } finally {
         compactLock.unlock();
       }
       flushLock.lock();
       try {
         // reset the flushRequest flag
         if (!shouldFlush(changesAfterLastFlush.get())) {
           flushRequest = false;
         }
       } finally {
         flushLock.unlock();
       }
     }
   }

   private boolean shouldFlush(long changes) {
     long heapSize = region.getMemStoreHeapSize();
     long offHeapSize = region.getMemStoreOffHeapSize();
     boolean flush = heapSize + offHeapSize >= flushSize || changes > flushPerChanges;
     if (flush && LOG.isTraceEnabled()) {
       LOG.trace("shouldFlush totalMemStoreSize={}, flushSize={}, changes={}, flushPerChanges={}",
         heapSize + offHeapSize, flushSize, changes, flushPerChanges);
     }
     return flush;
   }

   void onUpdate() {
     long changes = changesAfterLastFlush.incrementAndGet();
     if (shouldFlush(changes)) {
       requestFlush();
     }
   }

   void requestFlush() {
     flushLock.lock();
     try {
       if (flushRequest) {
         return;
       }
       flushRequest = true;
       flushCond.signalAll();
     } finally {
       flushLock.unlock();
     }
   }

   void resetChangesAfterLastFlush() {
     changesAfterLastFlush.set(0);
   }

   void recordLastFlushTime() {
     lastFlushTime = EnvironmentEdgeManager.currentTime();
   }

   @Override
   public void close() {
     closed = true;
     flushThread.interrupt();
     compactExecutor.shutdown();
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.hadoop.hbase.master.region;

	import java.io.Closeable;
	import java.io.IOException;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Executors;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.AtomicLong;
	import java.util.concurrent.locks.Condition;
	import java.util.concurrent.locks.Lock;
	import java.util.concurrent.locks.ReentrantLock;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.hbase.Abortable;
	import org.apache.hadoop.hbase.HConstants;
	import org.apache.hadoop.hbase.log.HBaseMarkers;
	import org.apache.hadoop.hbase.procedure2.store.region.RegionProcedureStore;
	import org.apache.hadoop.hbase.regionserver.HRegion;
	import org.apache.hadoop.hbase.regionserver.HStore;
	import org.apache.hadoop.hbase.regionserver.Store;
	import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
	import org.apache.hadoop.hbase.util.HFileArchiveUtil;
	import org.apache.yetus.audience.InterfaceAudience;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;

	/**
	* As long as there is no RegionServerServices for a master local region, we need implement the
	* flush and compaction logic by our own.
	* <p/>
	* The flush logic is very simple, every time after calling a modification method in
	* {@link RegionProcedureStore}, we will call the {@link #onUpdate()} method below, and in this
	* method, we will check the memstore size and if it is above the flush size, we will call
	* {@link HRegion#flush(boolean)} to force flush all stores.
	* <p/>
	* And for compaction, the logic is also very simple. After flush, we will check the store file
	* count, if it is above the compactMin, we will do a major compaction.
	*/
	@InterfaceAudience.Private
	class MasterRegionFlusherAndCompactor implements Closeable {

	private static final Logger LOG = LoggerFactory.getLogger(MasterRegionFlusherAndCompactor.class);

	private final Configuration conf;

	private final Abortable abortable;

	private final HRegion region;

	// as we can only count this outside the region's write/flush process so it is not accurate, but
	// it is enough.
	private final AtomicLong changesAfterLastFlush = new AtomicLong(0);

	private final long flushSize;

	private final long flushPerChanges;

	private final long flushIntervalMs;

	private final int compactMin;

	private final Path globalArchivePath;

	private final String archivedHFileSuffix;

	private final Thread flushThread;

	private final Lock flushLock = new ReentrantLock();

	private final Condition flushCond = flushLock.newCondition();

	private boolean flushRequest = false;

	private long lastFlushTime;

	private final ExecutorService compactExecutor;

	private final Lock compactLock = new ReentrantLock();

	private boolean compactRequest = false;

	private volatile boolean closed = false;

	MasterRegionFlusherAndCompactor(Configuration conf, Abortable abortable, HRegion region,
	long flushSize, long flushPerChanges, long flushIntervalMs, int compactMin,
	Path globalArchivePath, String archivedHFileSuffix) {
	this.conf = conf;
	this.abortable = abortable;
	this.region = region;
	this.flushSize = flushSize;
	this.flushPerChanges = flushPerChanges;
	this.flushIntervalMs = flushIntervalMs;
	this.compactMin = compactMin;
	this.globalArchivePath = globalArchivePath;
	this.archivedHFileSuffix = archivedHFileSuffix;
	flushThread = new Thread(this::flushLoop, region.getRegionInfo().getTable() + "-Flusher");
	flushThread.setDaemon(true);
	flushThread.start();
	compactExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
	.setNameFormat(region.getRegionInfo().getTable() + "-Store-Compactor").setDaemon(true)
	.build());
	LOG.info("Constructor flushSize={}, flushPerChanges={}, flushIntervalMs={}, compactMin={}",
	flushSize, flushPerChanges, flushIntervalMs, compactMin);
	}

	// inject our flush related configurations
	static void setupConf(Configuration conf, long flushSize, long flushPerChanges,
	long flushIntervalMs) {
	conf.setLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, flushSize);
	conf.setLong(HRegion.MEMSTORE_FLUSH_PER_CHANGES, flushPerChanges);
	conf.setLong(HRegion.MEMSTORE_PERIODIC_FLUSH_INTERVAL, flushIntervalMs);
	LOG.info("Injected flushSize={}, flushPerChanges={}, flushIntervalMs={}", flushSize,
	flushPerChanges, flushIntervalMs);
	}

	private void moveHFileToGlobalArchiveDir() throws IOException {
	FileSystem fs = region.getRegionFileSystem().getFileSystem();
	for (HStore store : region.getStores()) {
	store.closeAndArchiveCompactedFiles();
	Path storeArchiveDir = HFileArchiveUtil.getStoreArchivePath(conf, region.getRegionInfo(),
	store.getColumnFamilyDescriptor().getName());
	Path globalStoreArchiveDir = HFileArchiveUtil.getStoreArchivePathForArchivePath(
	globalArchivePath, region.getRegionInfo(), store.getColumnFamilyDescriptor().getName());
	try {
	if (fs.exists(storeArchiveDir)) {
	MasterRegionUtils.moveFilesUnderDir(fs, storeArchiveDir, globalStoreArchiveDir,
	archivedHFileSuffix);
	} else {
	LOG.warn(
	"Archived dir {} does not exist, there is no need to move archived hfiles from {} "
	+ "to global dir {} .",
	storeArchiveDir, storeArchiveDir, globalStoreArchiveDir);
	}
	} catch (IOException e) {
	LOG.warn("Failed to move archived hfiles from {} to global dir {}", storeArchiveDir,
	globalStoreArchiveDir, e);
	}
	}
	}

	private void compact() {
	try {
	region.compact(true);
	moveHFileToGlobalArchiveDir();
	} catch (IOException e) {
	LOG.error("Failed to compact master local region", e);
	}
	compactLock.lock();
	try {
	if (needCompaction()) {
	compactExecutor.execute(this::compact);
	} else {
	compactRequest = false;
	}
	} finally {
	compactLock.unlock();
	}
	}

	private boolean needCompaction() {
	for (Store store : region.getStores()) {
	if (store.getStorefilesCount() >= compactMin) {
	return true;
	}
	}
	return false;
	}

	private void flushLoop() {
	recordLastFlushTime();
	while (!closed) {
	flushLock.lock();
	try {
	while (!flushRequest) {
	long waitTimeMs = lastFlushTime + flushIntervalMs - EnvironmentEdgeManager.currentTime();
	if (waitTimeMs <= 0) {
	flushRequest = true;
	break;
	}
	flushCond.await(waitTimeMs, TimeUnit.MILLISECONDS);
	if (closed) {
	return;
	}
	}
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	continue;
	} finally {
	flushLock.unlock();
	}
	assert flushRequest;
	resetChangesAfterLastFlush();
	try {
	region.flush(true);
	recordLastFlushTime();
	} catch (IOException e) {
	LOG.error(HBaseMarkers.FATAL, "Failed to flush master local region, aborting...", e);
	abortable.abort("Failed to flush master local region", e);
	return;
	}
	compactLock.lock();
	try {
	if (!compactRequest && needCompaction()) {
	compactRequest = true;
	compactExecutor.execute(this::compact);
	}
	} finally {
	compactLock.unlock();
	}
	flushLock.lock();
	try {
	// reset the flushRequest flag
	if (!shouldFlush(changesAfterLastFlush.get())) {
	flushRequest = false;
	}
	} finally {
	flushLock.unlock();
	}
	}
	}

	private boolean shouldFlush(long changes) {
	long heapSize = region.getMemStoreHeapSize();
	long offHeapSize = region.getMemStoreOffHeapSize();
	boolean flush = heapSize + offHeapSize >= flushSize \|\| changes > flushPerChanges;
	if (flush && LOG.isTraceEnabled()) {
	LOG.trace("shouldFlush totalMemStoreSize={}, flushSize={}, changes={}, flushPerChanges={}",
	heapSize + offHeapSize, flushSize, changes, flushPerChanges);
	}
	return flush;
	}

	void onUpdate() {
	long changes = changesAfterLastFlush.incrementAndGet();
	if (shouldFlush(changes)) {
	requestFlush();
	}
	}

	void requestFlush() {
	flushLock.lock();
	try {
	if (flushRequest) {
	return;
	}
	flushRequest = true;
	flushCond.signalAll();
	} finally {
	flushLock.unlock();
	}
	}

	void resetChangesAfterLastFlush() {
	changesAfterLastFlush.set(0);
	}

	void recordLastFlushTime() {
	lastFlushTime = EnvironmentEdgeManager.currentTime();
	}

	@Override
	public void close() {
	closed = true;
	flushThread.interrupt();
	compactExecutor.shutdown();
	}
	}