Repository: hbase Updated Branches: refs/heads/branch-1.0 f92c2fbc0 -> ca8846eee
HBASE-12339 WALPerformanceEvaluation should have a log roller. Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/ca8846ee Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/ca8846ee Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/ca8846ee Branch: refs/heads/branch-1.0 Commit: ca8846eeef1cc980194924038f831817eca67a0e Parents: f92c2fb Author: Sean Busbey <bus...@apache.org> Authored: Sun Oct 26 23:53:45 2014 -0500 Committer: Enis Soztutar <e...@apache.org> Committed: Tue Jan 6 11:24:38 2015 -0800 ---------------------------------------------------------------------- .../hadoop/hbase/regionserver/LogRoller.java | 5 +++- .../hadoop/hbase/MockRegionServerServices.java | 28 +++++++++++++++----- .../hbase/wal/WALPerformanceEvaluation.java | 20 ++++++++++---- 3 files changed, 40 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/ca8846ee/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java index aa5998b..3dcc69e 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LogRoller.java @@ -40,6 +40,8 @@ import java.io.IOException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantLock; +import com.google.common.annotations.VisibleForTesting; + /** * Runs periodically to determine if the WAL should be rolled. * @@ -50,7 +52,8 @@ import java.util.concurrent.locks.ReentrantLock; * TODO: change to a pool of threads */ @InterfaceAudience.Private -class LogRoller extends HasThread { +@VisibleForTesting +public class LogRoller extends HasThread { static final Log LOG = LogFactory.getLog(LogRoller.class); private final ReentrantLock rollLock = new ReentrantLock(); private final AtomicBoolean rollLog = new AtomicBoolean(false); http://git-wip-us.apache.org/repos/asf/hbase/blob/ca8846ee/hbase-server/src/test/java/org/apache/hadoop/hbase/MockRegionServerServices.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/MockRegionServerServices.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/MockRegionServerServices.java index 013f0ef..347279e 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/MockRegionServerServices.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/MockRegionServerServices.java @@ -23,7 +23,10 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.hbase.client.ClusterConnection; @@ -51,28 +54,36 @@ import com.google.protobuf.Service; /** * Basic mock region server services. Should only be instantiated by HBaseTestingUtility.b */ -class MockRegionServerServices implements RegionServerServices { +public class MockRegionServerServices implements RegionServerServices { + protected static final Log LOG = LogFactory.getLog(MockRegionServerServices.class); private final Map<String, HRegion> regions = new HashMap<String, HRegion>(); - private boolean stopping = false; private final ConcurrentSkipListMap<byte[], Boolean> rit = new ConcurrentSkipListMap<byte[], Boolean>(Bytes.BYTES_COMPARATOR); private HFileSystem hfs = null; + private final Configuration conf; private ZooKeeperWatcher zkw = null; private ServerName serverName = null; private RpcServerInterface rpcServer = null; private volatile boolean abortRequested; + private volatile boolean stopping = false; + private final AtomicBoolean running = new AtomicBoolean(true); MockRegionServerServices(ZooKeeperWatcher zkw) { - this.zkw = zkw; + this(zkw, null); } MockRegionServerServices(ZooKeeperWatcher zkw, ServerName serverName) { this.zkw = zkw; this.serverName = serverName; + this.conf = (zkw == null ? new Configuration() : zkw.getConfiguration()); } MockRegionServerServices(){ - this(null); + this(null, null); + } + + public MockRegionServerServices(Configuration conf) { + this.conf = conf; } @Override @@ -165,7 +176,7 @@ class MockRegionServerServices implements RegionServerServices { @Override public Configuration getConfiguration() { - return zkw == null ? null : zkw.getConfiguration(); + return conf; } @Override @@ -176,12 +187,15 @@ class MockRegionServerServices implements RegionServerServices { @Override public void stop(String why) { - //no-op + this.stopping = true; + if (running.compareAndSet(true, false)) { + LOG.info("Shutting down due to request '" + why + "'"); + } } @Override public boolean isStopped() { - return false; + return !(running.get()); } @Override http://git-wip-us.apache.org/repos/asf/hbase/blob/ca8846ee/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/WALPerformanceEvaluation.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/WALPerformanceEvaluation.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/WALPerformanceEvaluation.java index bfc7147..bf8fba0 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/WALPerformanceEvaluation.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/WALPerformanceEvaluation.java @@ -44,15 +44,18 @@ import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.MockRegionServerServices; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.io.crypto.KeyProviderForTesting; import org.apache.hadoop.hbase.regionserver.HRegion; +import org.apache.hadoop.hbase.regionserver.LogRoller; import org.apache.hadoop.hbase.trace.SpanReceiverHost; import org.apache.hadoop.hbase.wal.WALProvider.Writer; import org.apache.hadoop.hbase.wal.WAL; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.hadoop.hbase.util.Threads; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.htrace.Sampler; @@ -313,13 +316,16 @@ public final class WALPerformanceEvaluation extends Configured implements Tool { final WALFactory wals = new WALFactory(getConf(), null, "wals"); final HRegion[] regions = new HRegion[numRegions]; final Runnable[] benchmarks = new Runnable[numRegions]; + final MockRegionServerServices mockServices = new MockRegionServerServices(getConf()); + final LogRoller roller = new LogRoller(mockServices, mockServices); + Threads.setDaemonThreadRunning(roller.getThread(), "WALPerfEval.logRoller"); try { for(int i = 0; i < numRegions; i++) { // Initialize Table Descriptor // a table per desired region means we can avoid carving up the key space final HTableDescriptor htd = createHTableDescriptor(i, numFamilies); - regions[i] = openRegion(fs, rootRegionDir, htd, wals, roll); + regions[i] = openRegion(fs, rootRegionDir, htd, wals, roll, roller); benchmarks[i] = Trace.wrap(new WALPutBenchmark(regions[i], htd, numIterations, noSync, syncInterval, traceFreq)); } @@ -335,6 +341,7 @@ public final class WALPerformanceEvaluation extends Configured implements Tool { } } if (verify) { + LOG.info("verifying written log entries."); Path dir = new Path(FSUtils.getRootDir(getConf()), DefaultWALProvider.getWALDirectoryName("wals")); long editCount = 0; @@ -351,11 +358,16 @@ public final class WALPerformanceEvaluation extends Configured implements Tool { } } } finally { + mockServices.stop("test clean up."); for (int i = 0; i < numRegions; i++) { if (regions[i] != null) { closeRegion(regions[i]); } } + if (null != roller) { + LOG.info("shutting down log roller."); + Threads.shutdown(roller.getThread()); + } wals.shutdown(); // Remove the root dir for this test region if (cleanup) cleanRegionRootDir(fs, rootRegionDir); @@ -465,13 +477,14 @@ public final class WALPerformanceEvaluation extends Configured implements Tool { private final Set<WAL> walsListenedTo = new HashSet<WAL>(); private HRegion openRegion(final FileSystem fs, final Path dir, final HTableDescriptor htd, - final WALFactory wals, final long whenToRoll) throws IOException { + final WALFactory wals, final long whenToRoll, final LogRoller roller) throws IOException { // Initialize HRegion HRegionInfo regionInfo = new HRegionInfo(htd.getTableName()); // Initialize WAL final WAL wal = wals.getWAL(regionInfo.getEncodedNameAsBytes()); // If we haven't already, attach a listener to this wal to handle rolls and metrics. if (walsListenedTo.add(wal)) { + roller.addWAL(wal); wal.registerWALActionsListener(new WALActionsListener.Base() { private int appends = 0; @@ -484,8 +497,6 @@ public final class WALPerformanceEvaluation extends Configured implements Tool { // We used to do explicit call to rollWriter but changed it to a request // to avoid dead lock (there are less threads going on in this class than // in the regionserver -- regionserver does not have the issue). - // TODO I think this means no rolling actually happens; the request relies on there - // being a LogRoller. DefaultWALProvider.requestLogRoll(wal); } } @@ -502,7 +513,6 @@ public final class WALPerformanceEvaluation extends Configured implements Tool { appendMeter.mark(size); } }); - wal.rollWriter(); } return HRegion.createHRegion(regionInfo, dir, getConf(), htd, wal);
