From dc8682e52517b61e1cbb125b8390d72cdf9a28fb Mon Sep 17 00:00:00 2001 From: zhangduo Date: Wed, 30 Mar 2016 16:31:25 +0800 Subject: [PATCH] HBASE-15537 Add multi WAL support for AsyncFSWAL --- .../hadoop/hbase/wal/AbstractFSWALProvider.java | 2 +- .../hadoop/hbase/wal/RegionGroupingProvider.java | 130 +++++++++------------ .../org/apache/hadoop/hbase/wal/WALFactory.java | 72 ++++++++---- .../wal/TestBoundedRegionGroupingStrategy.java | 35 +++--- 4 files changed, 123 insertions(+), 116 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java index 2f5c299..15b2950 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java @@ -82,7 +82,7 @@ public abstract class AbstractFSWALProvider> implemen * @param factory factory that made us, identity used for FS layout. may not be null * @param conf may not be null * @param listeners may be null - * @param providerId differentiate between providers from one facotry, used for FS layout. may be + * @param providerId differentiate between providers from one factory, used for FS layout. may be * null */ @Override diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java index 0aeaccf..273a176 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java @@ -18,30 +18,24 @@ */ package org.apache.hadoop.hbase.wal; -import static org.apache.hadoop.hbase.wal.DefaultWALProvider.META_WAL_PROVIDER_ID; -import static org.apache.hadoop.hbase.wal.DefaultWALProvider.WAL_FILE_NAME_DELIMITER; +import static org.apache.hadoop.hbase.wal.AbstractFSWALProvider.META_WAL_PROVIDER_ID; +import static org.apache.hadoop.hbase.wal.AbstractFSWALProvider.WAL_FILE_NAME_DELIMITER; import java.io.IOException; import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.Lock; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.hbase.HConstants; -import org.apache.hadoop.hbase.regionserver.wal.FSHLog; -import org.apache.hadoop.hbase.regionserver.wal.MetricsWAL; - +import org.apache.hadoop.hbase.classification.InterfaceAudience; // imports for classes still in regionserver.wal import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener; import org.apache.hadoop.hbase.util.Bytes; -import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.hadoop.hbase.util.IdReadWriteLock; /** * A WAL Provider that returns a WAL per group of regions. @@ -127,19 +121,20 @@ class RegionGroupingProvider implements WALProvider { private static final String META_WAL_GROUP_NAME = "meta"; /** A group-wal mapping, recommended to make sure one-one rather than many-one mapping */ - protected final Map cached = new HashMap(); - /** Stores unique wals generated by this RegionGroupingProvider */ - private final Set logs = Collections.synchronizedSet(new HashSet()); + private final ConcurrentMap cached = new ConcurrentHashMap<>(); - /** - * we synchronize on walCacheLock to prevent wal recreation in different threads - */ - final Object walCacheLock = new Object(); + private final IdReadWriteLock createLock = new IdReadWriteLock(); + + private final Class providerClass; - protected RegionGroupingStrategy strategy = null; + private RegionGroupingStrategy strategy = null; + private WALFactory factory = null; private List listeners = null; private String providerId = null; - private Configuration conf = null; + + public RegionGroupingProvider(Class providerClass) { + this.providerClass = providerClass; + } @Override public void init(final WALFactory factory, final Configuration conf, @@ -147,6 +142,7 @@ class RegionGroupingProvider implements WALProvider { if (null != strategy) { throw new IllegalStateException("WALProvider.init should only be called once."); } + this.factory = factory; this.listeners = null == listeners ? null : Collections.unmodifiableList(listeners); StringBuilder sb = new StringBuilder().append(factory.factoryId); if (providerId != null) { @@ -158,45 +154,32 @@ class RegionGroupingProvider implements WALProvider { } this.providerId = sb.toString(); this.strategy = getStrategy(conf, REGION_GROUPING_STRATEGY, DEFAULT_REGION_GROUPING_STRATEGY); - this.conf = conf; } - /** - * Populate the cache for this group. - */ - FSHLog populateCache(String groupName) throws IOException { - boolean isMeta = META_WAL_PROVIDER_ID.equals(providerId); - String hlogPrefix; - List listeners; - if (isMeta) { - hlogPrefix = this.providerId; - // don't watch log roll for meta - listeners = Collections. singletonList(new MetricsWAL()); + private WALProvider createProvider(String group) throws IOException { + if (META_WAL_PROVIDER_ID.equals(providerId)) { + return factory.createProvider(providerClass, listeners, META_WAL_PROVIDER_ID); } else { - hlogPrefix = groupName; - listeners = this.listeners; + return factory.createProvider(providerClass, listeners, group); } - FSHLog log = new FSHLog(FileSystem.get(conf), FSUtils.getRootDir(conf), - DefaultWALProvider.getWALDirectoryName(providerId), HConstants.HREGION_OLDLOGDIR_NAME, - conf, listeners, true, hlogPrefix, isMeta ? META_WAL_PROVIDER_ID : null); - cached.put(groupName, log); - logs.add(log); - return log; } private WAL getWAL(final String group) throws IOException { - WAL log = cached.get(group); - if (null == log) { - // only lock when need to create wal, and need to lock since - // creating hlog on fs is time consuming - synchronized (this.walCacheLock) { - log = cached.get(group);// check again - if (null == log) { - log = populateCache(group); + WALProvider provider = cached.get(group); + if (provider == null) { + Lock lock = createLock.getLock(group.hashCode()).writeLock(); + lock.lock(); + try { + provider = cached.get(group); + if (provider == null) { + provider = createProvider(group); + cached.put(group, provider); } + } finally { + lock.unlock(); } } - return log; + return provider.getWAL(null, null); } @Override @@ -214,15 +197,15 @@ class RegionGroupingProvider implements WALProvider { public void shutdown() throws IOException { // save the last exception and rethrow IOException failure = null; - synchronized (logs) { - for (FSHLog wal : logs) { - try { - wal.shutdown(); - } catch (IOException exception) { - LOG.error("Problem shutting down log '" + wal + "': " + exception.getMessage()); - LOG.debug("Details of problem shutting down log '" + wal + "'", exception); - failure = exception; + for (WALProvider provider: cached.values()) { + try { + provider.shutdown(); + } catch (IOException e) { + LOG.error("Problem shutting down wal provider '" + provider + "': " + e.getMessage()); + if (LOG.isDebugEnabled()) { + LOG.debug("Details of problem shutting down wal provider '" + provider + "'", e); } + failure = e; } } if (failure != null) { @@ -234,15 +217,15 @@ class RegionGroupingProvider implements WALProvider { public void close() throws IOException { // save the last exception and rethrow IOException failure = null; - synchronized (logs) { - for (FSHLog wal : logs) { - try { - wal.close(); - } catch (IOException exception) { - LOG.error("Problem closing log '" + wal + "': " + exception.getMessage()); - LOG.debug("Details of problem closing wal '" + wal + "'", exception); - failure = exception; + for (WALProvider provider : cached.values()) { + try { + provider.close(); + } catch (IOException e) { + LOG.error("Problem closing wal provider '" + provider + "': " + e.getMessage()); + if (LOG.isDebugEnabled()) { + LOG.debug("Details of problem closing wal provider '" + provider + "'", e); } + failure = e; } } if (failure != null) { @@ -262,10 +245,8 @@ class RegionGroupingProvider implements WALProvider { @Override public long getNumLogFiles() { long numLogFiles = 0; - synchronized (logs) { - for (FSHLog wal : logs) { - numLogFiles += wal.getNumLogFiles(); - } + for (WALProvider provider : cached.values()) { + numLogFiles += provider.getNumLogFiles(); } return numLogFiles; } @@ -273,12 +254,9 @@ class RegionGroupingProvider implements WALProvider { @Override public long getLogFileSize() { long logFileSize = 0; - synchronized (logs) { - for (FSHLog wal : logs) { - logFileSize += wal.getLogFileSize(); - } + for (WALProvider provider : cached.values()) { + logFileSize += provider.getLogFileSize(); } return logFileSize; } - } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java index a2761df..b021e63 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java @@ -40,7 +40,7 @@ import org.apache.hadoop.hbase.wal.WALProvider.Writer; import org.apache.hadoop.hbase.util.CancelableProgressable; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; import org.apache.hadoop.hbase.util.LeaseNotRecoveredException; - +import org.apache.hadoop.hbase.regionserver.wal.FSHLog; // imports for things that haven't moved from regionserver.wal yet. import org.apache.hadoop.hbase.regionserver.wal.MetricsWAL; import org.apache.hadoop.hbase.regionserver.wal.ProtobufLogReader; @@ -88,9 +88,15 @@ public class WALFactory { public static final String WAL_PROVIDER = "hbase.wal.provider"; static final String DEFAULT_WAL_PROVIDER = Providers.defaultProvider.name(); - static final String META_WAL_PROVIDER = "hbase.wal.meta_provider"; + public static final String META_WAL_PROVIDER = "hbase.wal.meta_provider"; static final String DEFAULT_META_WAL_PROVIDER = Providers.defaultProvider.name(); + public static final String MULTI_WAL_ENABLED = "hbase.wal.provider.multi.enabled"; + static final boolean DEFAULT_MULTI_WAL_ENABLED = false; + + public static final String MULTI_META_WAL_ENABLED = "hbase.wal.meta_provider.multi.enabled"; + static final boolean DEFAULT_META_MULTI_WAL_ENABLED = false; + final String factoryId; final WALProvider provider; // The meta updates are written to a different wal. If this @@ -127,34 +133,54 @@ public class WALFactory { factoryId = SINGLETON_ID; } - /** - * instantiate a provider from a config property. - * requires conf to have already been set (as well as anything the provider might need to read). - */ - WALProvider getProvider(final String key, final String defaultValue, - final List listeners, final String providerId) throws IOException { - Class clazz; + private Class getProviderClass(final String key, final String defaultValue) { try { - clazz = Providers.valueOf(conf.get(key, defaultValue)).clazz; + return Providers.valueOf(conf.get(key, defaultValue)).clazz; } catch (IllegalArgumentException exception) { // Fall back to them specifying a class name // Note that the passed default class shouldn't actually be used, since the above only fails // when there is a config value present. - clazz = conf.getClass(key, DefaultWALProvider.class, WALProvider.class); + return conf.getClass(key, DefaultWALProvider.class, WALProvider.class); } + } + + WALProvider createProvider(final Class clazz, + final List listeners, final String providerId) throws IOException { LOG.info("Instantiating WALProvider of type " + clazz); try { final WALProvider result = clazz.newInstance(); result.init(this, conf, listeners, providerId); return result; - } catch (InstantiationException exception) { - LOG.error("couldn't set up WALProvider, check config key " + key); - LOG.debug("Exception details for failure to load WALProvider.", exception); - throw new IOException("couldn't set up WALProvider", exception); - } catch (IllegalAccessException exception) { - LOG.error("couldn't set up WALProvider, check config key " + key); - LOG.debug("Exception details for failure to load WALProvider.", exception); - throw new IOException("couldn't set up WALProvider", exception); + } catch (InstantiationException | IllegalAccessException e) { + LOG.error("couldn't set up WALProvider, the configured provider is " + clazz); + if (LOG.isDebugEnabled()) { + LOG.debug("Exception details for failure to load WALProvider.", e); + } + throw new IOException("couldn't set up WALProvider", e); + } + } + + private RegionGroupingProvider createMultiWALProvider(Class providerClass, + List listeners, String providerId) throws IOException { + RegionGroupingProvider provider = new RegionGroupingProvider(providerClass); + provider.init(this, conf, listeners, providerId); + return provider; + } + /** + * instantiate a provider from a config property. + * requires conf to have already been set (as well as anything the provider might need to read). + */ + WALProvider getProvider(final String providerKey, final String defaultProviderValue, + final String enableMultiKey, final boolean defaultEnableMultiValue, + final List listeners, final String providerId) throws IOException { + Class clazz = getProviderClass(providerKey, defaultProviderValue); + if (clazz == RegionGroupingProvider.class) { + // keep compatible for old multiwal config + return createMultiWALProvider(DefaultWALProvider.class, listeners, providerId); + } else if (conf.getBoolean(enableMultiKey, defaultEnableMultiValue)) { + return createMultiWALProvider(clazz, listeners, providerId); + } else { + return createProvider(clazz, listeners, providerId); } } @@ -177,7 +203,8 @@ public class WALFactory { this.factoryId = factoryId; // end required early initialization if (conf.getBoolean("hbase.regionserver.hlog.enabled", true)) { - provider = getProvider(WAL_PROVIDER, DEFAULT_WAL_PROVIDER, listeners, null); + provider = getProvider(WAL_PROVIDER, DEFAULT_WAL_PROVIDER, MULTI_WAL_ENABLED, + DEFAULT_MULTI_WAL_ENABLED, listeners, null); } else { // special handling of existing configuration behavior. LOG.warn("Running with WAL disabled."); @@ -239,8 +266,9 @@ public class WALFactory { WALProvider metaProvider = this.metaProvider.get(); if (null == metaProvider) { final WALProvider temp = getProvider(META_WAL_PROVIDER, DEFAULT_META_WAL_PROVIDER, - Collections.singletonList(new MetricsWAL()), - DefaultWALProvider.META_WAL_PROVIDER_ID); + MULTI_META_WAL_ENABLED, DEFAULT_META_MULTI_WAL_ENABLED, + Collections. singletonList(new MetricsWAL()), + DefaultWALProvider.META_WAL_PROVIDER_ID); if (this.metaProvider.compareAndSet(null, temp)) { metaProvider = temp; } else { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java index 2044f82..6578fd5 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java @@ -49,7 +49,7 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.rules.TestName; -@Category({RegionServerTests.class, LargeTests.class}) +@Category({ RegionServerTests.class, LargeTests.class }) public class TestBoundedRegionGroupingStrategy { protected static final Log LOG = LogFactory.getLog(TestBoundedRegionGroupingStrategy.class); @@ -107,7 +107,7 @@ public class TestBoundedRegionGroupingStrategy { // Run the WPE tool with three threads writing 3000 edits each concurrently. // When done, verify that all edits were written. int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf), - new String [] {"-threads", "3", "-verify", "-noclosefs", "-iterations", "3000"}); + new String[] { "-threads", "3", "-verify", "-noclosefs", "-iterations", "3000" }); assertEquals(0, errCode); } @@ -118,8 +118,8 @@ public class TestBoundedRegionGroupingStrategy { public void testMoreRegionsThanBound() throws Exception { final String parallelism = Integer.toString(DEFAULT_NUM_REGION_GROUPS * 2); int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf), - new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000", - "-regions", parallelism}); + new String[] { "-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000", + "-regions", parallelism }); assertEquals(0, errCode); } @@ -127,11 +127,11 @@ public class TestBoundedRegionGroupingStrategy { public void testBoundsGreaterThanDefault() throws Exception { final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS); try { - conf.setInt(NUM_REGION_GROUPS, temp*4); - final String parallelism = Integer.toString(temp*4); + conf.setInt(NUM_REGION_GROUPS, temp * 4); + final String parallelism = Integer.toString(temp * 4); int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf), - new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000", - "-regions", parallelism}); + new String[] { "-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000", + "-regions", parallelism }); assertEquals(0, errCode); } finally { conf.setInt(NUM_REGION_GROUPS, temp); @@ -142,11 +142,11 @@ public class TestBoundedRegionGroupingStrategy { public void testMoreRegionsThanBoundWithBoundsGreaterThanDefault() throws Exception { final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS); try { - conf.setInt(NUM_REGION_GROUPS, temp*4); - final String parallelism = Integer.toString(temp*4*2); + conf.setInt(NUM_REGION_GROUPS, temp * 4); + final String parallelism = Integer.toString(temp * 4 * 2); int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf), - new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000", - "-regions", parallelism}); + new String[] { "-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000", + "-regions", parallelism }); assertEquals(0, errCode); } finally { conf.setInt(NUM_REGION_GROUPS, temp); @@ -161,24 +161,25 @@ public class TestBoundedRegionGroupingStrategy { final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS); WALFactory wals = null; try { - conf.setInt(NUM_REGION_GROUPS, temp*4); + conf.setInt(NUM_REGION_GROUPS, temp * 4); // Set HDFS root directory for storing WAL FSUtils.setRootDir(conf, TEST_UTIL.getDataTestDirOnTestFS()); wals = new WALFactory(conf, null, currentTest.getMethodName()); - final Set seen = new HashSet(temp*4); + final Set seen = new HashSet(temp * 4); final Random random = new Random(); int count = 0; // we know that this should see one of the wals more than once - for (int i = 0; i < temp*8; i++) { + for (int i = 0; i < temp * 8; i++) { final WAL maybeNewWAL = wals.getWAL(Bytes.toBytes(random.nextInt()), null); LOG.info("Iteration " + i + ", checking wal " + maybeNewWAL); if (seen.add(maybeNewWAL)) { count++; } } - assertEquals("received back a different number of WALs that are not equal() to each other " + - "than the bound we placed.", temp*4, count); + assertEquals("received back a different number of WALs that are not equal() to each other " + + "than the bound we placed.", + temp * 4, count); } finally { if (wals != null) { wals.close(); -- 1.9.1