You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ra...@apache.org on 2019/05/08 10:24:52 UTC
[hbase] branch branch-2 updated: HBASE-22072 High read/write
intensive regions may cause long crash (#214)
This is an automated email from the ASF dual-hosted git repository.
ramkrishna pushed a commit to branch branch-2
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2 by this push:
new 5099ef2 HBASE-22072 High read/write intensive regions may cause long crash (#214)
5099ef2 is described below
commit 5099ef216c10f109cecb2605aa31948fdd48c6ae
Author: ramkrish86 <ra...@intel.com>
AuthorDate: Wed May 8 15:30:04 2019 +0530
HBASE-22072 High read/write intensive regions may cause long crash (#214)
* HBASE-22072 High read/write intensive regions may cause long crash
recovery
* Make the 'closing' variable as volatile and move the test case to
standlone class
---
.../hadoop/hbase/regionserver/StoreScanner.java | 78 ++++--
.../hbase/regionserver/TestStoreScanner.java | 4 +-
.../regionserver/TestStoreScannerClosure.java | 263 +++++++++++++++++++++
3 files changed, 319 insertions(+), 26 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
index e0b4eb9..53b1262 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
@@ -79,8 +79,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
private int storeOffset = 0;
// Used to indicate that the scanner has closed (see HBASE-1107)
- // Do not need to be volatile because it's always accessed via synchronized methods
- private boolean closing = false;
+ private volatile boolean closing = false;
private final boolean get;
private final boolean explicitColumnQuery;
private final boolean useRowColBloom;
@@ -157,6 +156,8 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
final List<KeyValueScanner> currentScanners = new ArrayList<>();
// flush update lock
private final ReentrantLock flushLock = new ReentrantLock();
+ // lock for closing.
+ private final ReentrantLock closeLock = new ReentrantLock();
protected final long readPt;
private boolean topChanged = false;
@@ -473,31 +474,38 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
}
private void close(boolean withDelayedScannersClose) {
- if (this.closing) {
- return;
- }
- if (withDelayedScannersClose) {
- this.closing = true;
- }
- // For mob compaction, we do not have a store.
- if (this.store != null) {
- this.store.deleteChangedReaderObserver(this);
- }
- if (withDelayedScannersClose) {
- clearAndClose(scannersForDelayedClose);
- clearAndClose(memStoreScannersAfterFlush);
- clearAndClose(flushedstoreFileScanners);
- if (this.heap != null) {
- this.heap.close();
- this.currentScanners.clear();
- this.heap = null; // CLOSED!
+ closeLock.lock();
+ // If the closeLock is acquired then any subsequent updateReaders()
+ // call is ignored.
+ try {
+ if (this.closing) {
+ return;
}
- } else {
- if (this.heap != null) {
- this.scannersForDelayedClose.add(this.heap);
- this.currentScanners.clear();
- this.heap = null;
+ if (withDelayedScannersClose) {
+ this.closing = true;
+ }
+ // For mob compaction, we do not have a store.
+ if (this.store != null) {
+ this.store.deleteChangedReaderObserver(this);
}
+ if (withDelayedScannersClose) {
+ clearAndClose(scannersForDelayedClose);
+ clearAndClose(memStoreScannersAfterFlush);
+ clearAndClose(flushedstoreFileScanners);
+ if (this.heap != null) {
+ this.heap.close();
+ this.currentScanners.clear();
+ this.heap = null; // CLOSED!
+ }
+ } else {
+ if (this.heap != null) {
+ this.scannersForDelayedClose.add(this.heap);
+ this.currentScanners.clear();
+ this.heap = null;
+ }
+ }
+ } finally {
+ closeLock.unlock();
}
}
@@ -876,8 +884,25 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
if (CollectionUtils.isEmpty(sfs) && CollectionUtils.isEmpty(memStoreScanners)) {
return;
}
+ boolean updateReaders = false;
flushLock.lock();
try {
+ if (!closeLock.tryLock()) {
+ // The reason for doing this is that when the current store scanner does not retrieve
+ // any new cells, then the scanner is considered to be done. The heap of this scanner
+ // is not closed till the shipped() call is completed. Hence in that case if at all
+ // the partial close (close (false)) has been called before updateReaders(), there is no
+ // need for the updateReaders() to happen.
+ LOG.debug("StoreScanner already has the close lock. There is no need to updateReaders");
+ // no lock acquired.
+ return;
+ }
+ // lock acquired
+ updateReaders = true;
+ if (this.closing) {
+ LOG.debug("StoreScanner already closing. There is no need to updateReaders");
+ return;
+ }
flushed = true;
final boolean isCompaction = false;
boolean usePread = get || scanUsePread;
@@ -896,6 +921,9 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
}
} finally {
flushLock.unlock();
+ if (updateReaders) {
+ closeLock.unlock();
+ }
}
// Let the next() call handle re-creating and seeking
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
index d5775da..8a3f254 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
@@ -34,12 +34,14 @@ import java.util.NavigableSet;
import java.util.OptionalInt;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeepDeletedCells;
import org.apache.hadoop.hbase.KeyValue;
@@ -74,6 +76,7 @@ public class TestStoreScanner {
private static final String CF_STR = "cf";
private static final byte[] CF = Bytes.toBytes(CF_STR);
static Configuration CONF = HBaseConfiguration.create();
+ private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
private ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, Integer.MAX_VALUE, Long.MAX_VALUE,
KeepDeletedCells.FALSE, HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.getInstance(), false);
@@ -847,7 +850,6 @@ public class TestStoreScanner {
}
}
-
@Test @Ignore("this fails, since we don't handle deletions, etc, in peek")
public void testPeek() throws Exception {
KeyValue[] kvs = new KeyValue [] {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScannerClosure.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScannerClosure.java
new file mode 100644
index 0000000..1fbef26
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScannerClosure.java
@@ -0,0 +1,263 @@
+/**
+ * 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.regionserver;
+
+import static org.apache.hadoop.hbase.KeyValueTestUtil.create;
+import static org.apache.hadoop.hbase.regionserver.KeyValueScanFixture.scanFixture;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.Random;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.CellComparator;
+import org.apache.hadoop.hbase.CellComparatorImpl;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+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.KeepDeletedCells;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.io.hfile.HFileContext;
+import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
+import org.apache.hadoop.hbase.io.hfile.RandomKeyValueUtil;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * This test tests whether parallel {@link StoreScanner#close()} and
+ * {@link StoreScanner#updateReaders(List, List)} works perfectly ensuring
+ * that there are no references on the existing Storescanner readers.
+ */
+@Category({ RegionServerTests.class, MediumTests.class })
+public class TestStoreScannerClosure {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestStoreScannerClosure.class);
+
+ private static final Logger LOG = LoggerFactory.getLogger(TestStoreScannerClosure.class);
+ private static final int NUM_VALID_KEY_TYPES = KeyValue.Type.values().length - 2;
+ @Rule
+ public TestName name = new TestName();
+ private static final String CF_STR = "cf";
+ private static HRegion region;
+ private static final byte[] CF = Bytes.toBytes(CF_STR);
+ static Configuration CONF = HBaseConfiguration.create();
+ private static CacheConfig cacheConf;
+ private static FileSystem fs;
+ private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+ private static String ROOT_DIR = TEST_UTIL.getDataTestDir("TestHFile").toString();
+ private ScanInfo scanInfo = new ScanInfo(CONF, CF, 0, Integer.MAX_VALUE, Long.MAX_VALUE,
+ KeepDeletedCells.FALSE, HConstants.DEFAULT_BLOCKSIZE, 0, CellComparator.getInstance(), false);
+ private final static byte[] fam = Bytes.toBytes("cf_1");
+ private static final KeyValue[] kvs =
+ new KeyValue[] { create("R1", "cf", "a", 11, KeyValue.Type.Put, "dont-care"),
+ create("R1", "cf", "b", 11, KeyValue.Type.Put, "dont-care"),
+ create("R1", "cf", "c", 11, KeyValue.Type.Put, "dont-care"),
+ create("R1", "cf", "d", 11, KeyValue.Type.Put, "dont-care"),
+ create("R1", "cf", "e", 11, KeyValue.Type.Put, "dont-care"),
+ create("R1", "cf", "f", 11, KeyValue.Type.Put, "dont-care"),
+ create("R1", "cf", "g", 11, KeyValue.Type.Put, "dont-care"),
+ create("R1", "cf", "h", 11, KeyValue.Type.Put, "dont-care"),
+ create("R1", "cf", "i", 11, KeyValue.Type.Put, "dont-care"),
+ create("R2", "cf", "a", 11, KeyValue.Type.Put, "dont-care"), };
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ CONF = TEST_UTIL.getConfiguration();
+ cacheConf = new CacheConfig(CONF);
+ fs = TEST_UTIL.getTestFileSystem();
+ TableName tableName = TableName.valueOf("test");
+ HTableDescriptor htd = new HTableDescriptor(tableName);
+ htd.addFamily(new HColumnDescriptor(fam));
+ HRegionInfo info = new HRegionInfo(tableName, null, null, false);
+ Path path = TEST_UTIL.getDataTestDir("test");
+ region = HBaseTestingUtility.createRegionAndWAL(info, path, TEST_UTIL.getConfiguration(), htd);
+ }
+
+ @Test
+ public void testScannerCloseAndUpdateReaders1() throws Exception {
+ testScannerCloseAndUpdateReaderInternal(true, false);
+ }
+
+ @Test
+ public void testScannerCloseAndUpdateReaders2() throws Exception {
+ testScannerCloseAndUpdateReaderInternal(false, true);
+ }
+
+ private Path writeStoreFile() throws IOException {
+ Path storeFileParentDir = new Path(TEST_UTIL.getDataTestDir(), "TestHFile");
+ HFileContext meta = new HFileContextBuilder().withBlockSize(64 * 1024).build();
+ StoreFileWriter sfw = new StoreFileWriter.Builder(CONF, fs).withOutputDir(storeFileParentDir)
+ .withComparator(CellComparatorImpl.COMPARATOR).withFileContext(meta).build();
+
+ final int rowLen = 32;
+ Random RNG = new Random();
+ for (int i = 0; i < 1000; ++i) {
+ byte[] k = RandomKeyValueUtil.randomOrderedKey(RNG, i);
+ byte[] v = RandomKeyValueUtil.randomValue(RNG);
+ int cfLen = RNG.nextInt(k.length - rowLen + 1);
+ KeyValue kv = new KeyValue(k, 0, rowLen, k, rowLen, cfLen, k, rowLen + cfLen,
+ k.length - rowLen - cfLen, RNG.nextLong(), generateKeyType(RNG), v, 0, v.length);
+ sfw.append(kv);
+ }
+
+ sfw.close();
+ return sfw.getPath();
+ }
+
+ private static KeyValue.Type generateKeyType(Random rand) {
+ if (rand.nextBoolean()) {
+ // Let's make half of KVs puts.
+ return KeyValue.Type.Put;
+ } else {
+ KeyValue.Type keyType = KeyValue.Type.values()[1 + rand.nextInt(NUM_VALID_KEY_TYPES)];
+ if (keyType == KeyValue.Type.Minimum || keyType == KeyValue.Type.Maximum) {
+ throw new RuntimeException("Generated an invalid key type: " + keyType + ". "
+ + "Probably the layout of KeyValue.Type has changed.");
+ }
+ return keyType;
+ }
+ }
+
+ private HStoreFile readStoreFile(Path storeFilePath, Configuration conf) throws Exception {
+ // Open the file reader with block cache disabled.
+ HStoreFile file = new HStoreFile(fs, storeFilePath, conf, cacheConf, BloomType.NONE, true);
+ return file;
+ }
+
+ private void testScannerCloseAndUpdateReaderInternal(boolean awaitUpdate, boolean awaitClose)
+ throws IOException, InterruptedException {
+ // start write to store file.
+ Path path = writeStoreFile();
+ HStoreFile file = null;
+ List<HStoreFile> files = new ArrayList<HStoreFile>();
+ try {
+ file = readStoreFile(path, CONF);
+ files.add(file);
+ } catch (Exception e) {
+ // fail test
+ assertTrue(false);
+ }
+ scanFixture(kvs);
+ // scanners.add(storeFileScanner);
+ try (ExtendedStoreScanner scan = new ExtendedStoreScanner(region.getStore(fam), scanInfo,
+ new Scan(), getCols("a", "d"), 100L)) {
+ Thread closeThread = new Thread() {
+ public void run() {
+ scan.close(awaitClose, true);
+ }
+ };
+ closeThread.start();
+ Thread updateThread = new Thread() {
+ public void run() {
+ try {
+ scan.updateReaders(awaitUpdate, files, Collections.emptyList());
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ };
+ updateThread.start();
+ // complete both the threads
+ closeThread.join();
+ // complete both the threads
+ updateThread.join();
+ if (file.getReader() != null) {
+ // the fileReader is not null when the updateReaders has completed first.
+ // in the other case the fileReader will be null.
+ int refCount = file.getReader().getRefCount();
+ LOG.info("the store scanner count is " + refCount);
+ assertTrue("The store scanner count should be 0", refCount == 0);
+ }
+ }
+ }
+
+ private static class ExtendedStoreScanner extends StoreScanner {
+ private CountDownLatch latch = new CountDownLatch(1);
+
+ public ExtendedStoreScanner(HStore store, ScanInfo scanInfo, Scan scan,
+ NavigableSet<byte[]> columns, long readPt) throws IOException {
+ super(store, scanInfo, scan, columns, readPt);
+ }
+
+ public void updateReaders(boolean await, List<HStoreFile> sfs,
+ List<KeyValueScanner> memStoreScanners) throws IOException {
+ if (await) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+ super.updateReaders(sfs, memStoreScanners);
+ if (!await) {
+ latch.countDown();
+ }
+ }
+
+ // creating a dummy close
+ public void close(boolean await, boolean dummy) {
+ if (await) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+ super.close();
+ if (!await) {
+ latch.countDown();
+ }
+ }
+ }
+
+ NavigableSet<byte[]> getCols(String... strCols) {
+ NavigableSet<byte[]> cols = new TreeSet<>(Bytes.BYTES_COMPARATOR);
+ for (String col : strCols) {
+ byte[] bytes = Bytes.toBytes(col);
+ cols.add(bytes);
+ }
+ return cols;
+ }
+}