Skip to content

HBASE-22072 High read/write intensive regions may cause long crash #214

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ramkrish86 merged 2 commits into from
May 8, 2019
Merged

HBASE-22072 High read/write intensive regions may cause long crash #214

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5ee18cc
HBASE-22072 High read/write intensive regions may cause long crash
May 3, 2019
851cd9a
Make the 'closing' variable as volatile and move the test case to
May 6, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 52 additions & 23 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,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;
Expand Down Expand Up @@ -473,31 +475,38 @@ public void close() {
}

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();
}
}

Expand Down Expand Up @@ -876,8 +885,25 @@ public void updateReaders(List<HStoreFile> sfs, List<KeyValueScanner> memStoreSc
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) {
Comment thread
ramkrish86 marked this conversation as resolved.
LOG.debug("StoreScanner already closing. There is no need to updateReaders");
return;
}
flushed = true;
final boolean isCompaction = false;
boolean usePread = get || scanUsePread;
Expand All @@ -896,6 +922,9 @@ public void updateReaders(List<HStoreFile> sfs, List<KeyValueScanner> memStoreSc
}
} finally {
flushLock.unlock();
if (updateReaders) {
closeLock.unlock();
}
}
// Let the next() call handle re-creating and seeking
}
Expand Down
178 changes: 177 additions & 1 deletion hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,26 +32,42 @@
import java.util.List;
import java.util.NavigableSet;
import java.util.OptionalInt;
import java.util.Random;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.CellComparatorImpl;
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.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.PrivateCellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.ColumnCountGetFilter;
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.apache.hadoop.hbase.util.EnvironmentEdge;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Ignore;
import org.junit.Rule;
Expand All @@ -70,10 +86,17 @@ public class TestStoreScanner {
HBaseClassTestRule.forClass(TestStoreScanner.class);

private static final Logger LOG = LoggerFactory.getLogger(TestStoreScanner.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);

Expand All @@ -97,7 +120,20 @@ public class TestStoreScanner {
private static final int CELL_GRID_BLOCK3_BOUNDARY = 11;
private static final int CELL_GRID_BLOCK4_BOUNDARY = 15;
private static final int CELL_GRID_BLOCK5_BOUNDARY = 19;

private final static byte[] fam = Bytes.toBytes("cf_1");

@BeforeClass
public static void setUp() throws Exception {
Comment thread
ramkrish86 marked this conversation as resolved.
Outdated
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);
}
/**
* Five rows by four columns distinguished by column qualifier (column qualifier is one of the
* four rows... ONE, TWO, etc.). Exceptions are a weird row after TWO; it is TWO_POINT_TWO.
Expand Down Expand Up @@ -847,6 +883,146 @@ public void testScannerReseekDoesntNPE() throws Exception {
}
}

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();
}

public 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;
}

@Test
public void testScannerCloseAndUpdateReaders1() throws Exception {
testScannerCloseAndUpdateReaderInternal(true, false);
}

@Test
public void testScannerCloseAndUpdateReaders2() throws Exception {
testScannerCloseAndUpdateReaderInternal(false, true);
}

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();
}
}

}

@Test @Ignore("this fails, since we don't handle deletions, etc, in peek")
public void testPeek() throws Exception {
Expand Down