This is an automated email from the ASF dual-hosted git repository. liyu pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/hbase.git
commit c6298c709a10e581f20a5038f2cb2463f01b7195 Author: LiangJun He <[email protected]> AuthorDate: Sat Jun 11 08:48:23 2022 +0800 HBASE-27091 Speed up the loading of table descriptor from filesystem (#4493) Signed-off-by: Huaxiang Sun <[email protected]> Signed-off-by: Yu Li <[email protected]> --- .../org/apache/hadoop/hbase/HBaseServerBase.java | 15 +++- .../org/apache/hadoop/hbase/TableDescriptors.java | 8 ++- .../org/apache/hadoop/hbase/master/HMaster.java | 1 + .../hadoop/hbase/regionserver/HRegionServer.java | 1 + .../hadoop/hbase/util/FSTableDescriptors.java | 80 +++++++++++++++++++--- .../hadoop/hbase/util/TestFSTableDescriptors.java | 31 +++++++++ 6 files changed, 124 insertions(+), 12 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/HBaseServerBase.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/HBaseServerBase.java index f39e260914..e148c7da84 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/HBaseServerBase.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/HBaseServerBase.java @@ -231,8 +231,10 @@ public abstract class HBaseServerBase<R extends HBaseRpcServicesBase<?>> extends // init the filesystem this.dataFs = new HFileSystem(this.conf, useHBaseChecksum); this.dataRootDir = CommonFSUtils.getRootDir(this.conf); + int tableDescriptorParallelLoadThreads = + conf.getInt("hbase.tabledescriptor.parallel.load.threads", 0); this.tableDescriptors = new FSTableDescriptors(this.dataFs, this.dataRootDir, - !canUpdateTableDescriptor(), cacheTableDescriptor()); + !canUpdateTableDescriptor(), cacheTableDescriptor(), tableDescriptorParallelLoadThreads); } public HBaseServerBase(Configuration conf, String name) throws IOException { @@ -466,6 +468,17 @@ public abstract class HBaseServerBase<R extends HBaseRpcServicesBase<?>> extends } } + protected final void closeTableDescriptors() { + if (this.tableDescriptors != null) { + LOG.info("Close table descriptors"); + try { + this.tableDescriptors.close(); + } catch (IOException e) { + LOG.debug("Failed to close table descriptors gracefully", e); + } + } + } + /** * In order to register ShutdownHook, this method is called when HMaster and HRegionServer are * started. For details, please refer to HBASE-26951 diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/TableDescriptors.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/TableDescriptors.java index 1dc17eff0d..9ecdf39679 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/TableDescriptors.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/TableDescriptors.java @@ -17,6 +17,7 @@ */ package org.apache.hadoop.hbase; +import java.io.Closeable; import java.io.IOException; import java.util.Map; import org.apache.hadoop.hbase.client.TableDescriptor; @@ -26,7 +27,7 @@ import org.apache.yetus.audience.InterfaceAudience; * Get, remove and modify table descriptors. */ @InterfaceAudience.Private -public interface TableDescriptors { +public interface TableDescriptors extends Closeable { /** * Test whether a given table exists, i.e, has a table descriptor. @@ -35,6 +36,11 @@ public interface TableDescriptors { return get(tableName) != null; } + @Override + default void close() throws IOException { + // do nothing by default + } + /** * @return TableDescriptor for tablename */ diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java index c9556a80ed..2b818d9cc2 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java @@ -601,6 +601,7 @@ public class HMaster extends HBaseServerBase<MasterRpcServices> implements Maste this.rpcServices.stop(); } closeZooKeeper(); + closeTableDescriptors(); span.setStatus(StatusCode.OK); } finally { span.end(); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java index 0ee4c5d746..1865929dc5 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java @@ -977,6 +977,7 @@ public class HRegionServer extends HBaseServerBase<RSRpcServices> ZNodeClearer.deleteMyEphemeralNodeOnDisk(); closeZooKeeper(); + closeTableDescriptors(); LOG.info("Exiting; stopping=" + this.serverName + "; zookeeper connection closed."); span.setStatus(StatusCode.OK); } finally { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSTableDescriptors.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSTableDescriptors.java index 8f9dd4426d..27d66d14ca 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSTableDescriptors.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/FSTableDescriptors.java @@ -21,6 +21,7 @@ import com.google.errorprone.annotations.RestrictedApi; import edu.umd.cs.findbugs.annotations.Nullable; import java.io.EOFException; import java.io.IOException; +import java.io.InterruptedIOException; import java.util.Arrays; import java.util.Comparator; import java.util.List; @@ -28,6 +29,12 @@ import java.util.Map; import java.util.Optional; import java.util.TreeMap; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.lang3.NotImplementedException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; @@ -55,6 +62,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hbase.thirdparty.com.google.common.primitives.Ints; +import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder; /** * Implementation of {@link TableDescriptors} that reads descriptors from the passed filesystem. It @@ -79,6 +87,8 @@ public class FSTableDescriptors implements TableDescriptors { private final boolean fsreadonly; private final boolean usecache; private volatile boolean fsvisited; + private boolean tableDescriptorParallelLoadEnable = false; + private ThreadPoolExecutor executor; long cachehits = 0; long invocations = 0; @@ -108,10 +118,23 @@ public class FSTableDescriptors implements TableDescriptors { public FSTableDescriptors(final FileSystem fs, final Path rootdir, final boolean fsreadonly, final boolean usecache) { + this(fs, rootdir, fsreadonly, usecache, 0); + } + + public FSTableDescriptors(final FileSystem fs, final Path rootdir, final boolean fsreadonly, + final boolean usecache, final int tableDescriptorParallelLoadThreads) { this.fs = fs; this.rootdir = rootdir; this.fsreadonly = fsreadonly; this.usecache = usecache; + if (tableDescriptorParallelLoadThreads > 0) { + tableDescriptorParallelLoadEnable = true; + executor = new ThreadPoolExecutor(tableDescriptorParallelLoadThreads, + tableDescriptorParallelLoadThreads, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), + new ThreadFactoryBuilder().setNameFormat("FSTableDescriptorLoad-pool-%d").setDaemon(true) + .setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build()); + executor.allowCoreThreadTimeOut(true); + } } public static void tryUpdateMetaTableDescriptor(Configuration conf) throws IOException { @@ -235,27 +258,56 @@ public class FSTableDescriptors implements TableDescriptors { */ @Override public Map<String, TableDescriptor> getAll() throws IOException { - Map<String, TableDescriptor> tds = new TreeMap<>(); + Map<String, TableDescriptor> tds = new ConcurrentSkipListMap<>(); if (fsvisited) { for (Map.Entry<TableName, TableDescriptor> entry : this.cache.entrySet()) { tds.put(entry.getKey().getNameWithNamespaceInclAsString(), entry.getValue()); } } else { - LOG.trace("Fetching table descriptors from the filesystem."); - boolean allvisited = usecache; - for (Path d : FSUtils.getTableDirs(fs, rootdir)) { - TableDescriptor htd = get(CommonFSUtils.getTableName(d)); - if (htd == null) { - allvisited = false; - } else { - tds.put(htd.getTableName().getNameWithNamespaceInclAsString(), htd); + LOG.info("Fetching table descriptors from the filesystem."); + final long startTime = EnvironmentEdgeManager.currentTime(); + AtomicBoolean allvisited = new AtomicBoolean(usecache); + List<Path> tableDirs = FSUtils.getTableDirs(fs, rootdir); + if (!tableDescriptorParallelLoadEnable) { + for (Path dir : tableDirs) { + internalGet(dir, tds, allvisited); + } + } else { + CountDownLatch latch = new CountDownLatch(tableDirs.size()); + for (Path dir : tableDirs) { + executor.submit(new Runnable() { + @Override + public void run() { + try { + internalGet(dir, tds, allvisited); + } finally { + latch.countDown(); + } + } + }); + } + try { + latch.await(); + } catch (InterruptedException ie) { + throw (InterruptedIOException) new InterruptedIOException().initCause(ie); } } - fsvisited = allvisited; + fsvisited = allvisited.get(); + LOG.info("Fetched table descriptors(size=" + tds.size() + ") cost " + + (EnvironmentEdgeManager.currentTime() - startTime) + "ms."); } return tds; } + private void internalGet(Path dir, Map<String, TableDescriptor> tds, AtomicBoolean allvisited) { + TableDescriptor htd = get(CommonFSUtils.getTableName(dir)); + if (htd == null) { + allvisited.set(false); + } else { + tds.put(htd.getTableName().getNameWithNamespaceInclAsString(), htd); + } + } + /** * Find descriptors by namespace. * @see #get(org.apache.hadoop.hbase.TableName) @@ -379,6 +431,14 @@ public class FSTableDescriptors implements TableDescriptors { return Bytes.toString(b); } + @Override + public void close() throws IOException { + // Close the executor when parallel loading enabled. + if (tableDescriptorParallelLoadEnable) { + this.executor.shutdown(); + } + } + static final class SequenceIdAndFileLength { final int sequenceId; diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestFSTableDescriptors.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestFSTableDescriptors.java index 24ca058dc2..5e2b4b5295 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestFSTableDescriptors.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestFSTableDescriptors.java @@ -285,6 +285,32 @@ public class TestFSTableDescriptors { + htds.getAll().size(), count + 1, htds.getAll().size()); } + @Test + public void testParallelGetAll() throws IOException, InterruptedException { + final String name = "testParallelGetAll"; + FileSystem fs = FileSystem.get(UTIL.getConfiguration()); + // Enable parallel load table descriptor. + FSTableDescriptors htds = new FSTableDescriptorsTest(fs, testDir, true, 20); + final int count = 100; + // Write out table infos. + for (int i = 0; i < count; i++) { + htds.createTableDescriptor( + TableDescriptorBuilder.newBuilder(TableName.valueOf(name + i)).build()); + } + // add hbase:meta + htds + .createTableDescriptor(TableDescriptorBuilder.newBuilder(TableName.META_TABLE_NAME).build()); + + int getTableDescriptorSize = htds.getAll().size(); + assertEquals("getAll() didn't return all TableDescriptors, expected: " + (count + 1) + " got: " + + getTableDescriptorSize, count + 1, getTableDescriptorSize); + + // get again to check whether the cache works well + getTableDescriptorSize = htds.getAll().size(); + assertEquals("getAll() didn't return all TableDescriptors with cache, expected: " + (count + 1) + + " got: " + getTableDescriptorSize, count + 1, getTableDescriptorSize); + } + @Test public void testGetAllOrdering() throws Exception { FileSystem fs = FileSystem.get(UTIL.getConfiguration()); @@ -467,6 +493,11 @@ public class TestFSTableDescriptors { super(fs, rootdir, false, usecache); } + public FSTableDescriptorsTest(FileSystem fs, Path rootdir, boolean usecache, + int tableDescriptorParallelLoadThreads) { + super(fs, rootdir, false, usecache, tableDescriptorParallelLoadThreads); + } + @Override public TableDescriptor get(TableName tablename) { LOG.info((super.isUsecache() ? "Cached" : "Non-Cached") + " TableDescriptor.get() on "
