Skip to content

HBASE-22422 Retain an ByteBuff with refCnt=0 when getBlock from LRUCache#242

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
openinx merged 1 commit into from
May 28, 2019
Merged

HBASE-22422 Retain an ByteBuff with refCnt=0 when getBlock from LRUCache #242

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
3 changes: 3 additions & 0 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/BlockCacheUtil.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -230,6 +230,9 @@ public static boolean shouldReplaceExistingCacheBlock(BlockCache blockCache,
BlockCacheKey cacheKey, Cacheable newBlock) {
// NOTICE: The getBlock has retained the existingBlock inside.
Cacheable existingBlock = blockCache.getBlock(cacheKey, false, false, false);
if (existingBlock == null) {

@Apache9Apache9May 27, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, this means we may get NPE in the past?

@openinxopeninxMay 28, 2019

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, see the LruBlockCache#cacheBlock:

 LruCachedBlock cb = map.get(cacheKey);
if (cb != null && !BlockCacheUtil.shouldReplaceExistingCacheBlock(this, cacheKey, buf)) {
return;
}

The existence pre-check and accessing block in shouldReplaceExistingCacheBlock is not atomic op. so if any eviction happen between them, the NPE will happen. I added a UT testMultiThreadGetAndEvictBlock to address this.

return true;
}
try {
int comparison = BlockCacheUtil.validateBlockAddition(existingBlock, newBlock, cacheKey);
if (comparison < 0) {
Expand Down
40 changes: 24 additions & 16 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlock.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,8 @@
*/
package org.apache.hadoop.hbase.io.hfile;

import static org.apache.hadoop.hbase.io.ByteBuffAllocator.HEAP;

import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
Expand DownExpand Up@@ -669,18 +671,24 @@ HFileBlock unpack(HFileContext fileContext, FSReader reader) throws IOException

HFileBlock unpacked = new HFileBlock(this);
unpacked.allocateBuffer(); // allocates space for the decompressed block

HFileBlockDecodingContext ctx = blockType == BlockType.ENCODED_DATA
? reader.getBlockDecodingContext() : reader.getDefaultBlockDecodingContext();

ByteBuff dup = this.buf.duplicate();
dup.position(this.headerSize());
dup = dup.slice();

ctx.prepareDecoding(unpacked.getOnDiskSizeWithoutHeader(),
unpacked.getUncompressedSizeWithoutHeader(), unpacked.getBufferWithoutHeader(true), dup);

return unpacked;
boolean succ = false;
try {
HFileBlockDecodingContext ctx = blockType == BlockType.ENCODED_DATA
? reader.getBlockDecodingContext() : reader.getDefaultBlockDecodingContext();
// Create a duplicated buffer without the header part.
ByteBuff dup = this.buf.duplicate();
dup.position(this.headerSize());
dup = dup.slice();
// Decode the dup into unpacked#buf
ctx.prepareDecoding(unpacked.getOnDiskSizeWithoutHeader(),
unpacked.getUncompressedSizeWithoutHeader(), unpacked.getBufferWithoutHeader(true), dup);
succ = true;
return unpacked;
} finally {
if (!succ) {
unpacked.release();
}
}
}

/**
Expand All@@ -701,7 +709,7 @@ private void allocateBuffer() {

buf = newBuf;
// set limit to exclude next block's header
buf.limit(headerSize + uncompressedSizeWithoutHeader + cksumBytes);
buf.limit(capacityNeeded);
}

/**
Expand DownExpand Up@@ -1689,7 +1697,7 @@ private int getNextBlockOnDiskSize(boolean readNextHeader, ByteBuff onDiskBlock,
}

private ByteBuff allocate(int size, boolean intoHeap) {
return intoHeap ? ByteBuffAllocator.HEAP.allocate(size) : allocator.allocate(size);
return intoHeap ? HEAP.allocate(size) : allocator.allocate(size);
}

/**
Expand DownExpand Up@@ -1739,7 +1747,7 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
if (LOG.isTraceEnabled()) {
LOG.trace("Extra see to get block size!", new RuntimeException());
}
headerBuf = new SingleByteBuff(ByteBuffer.allocate(hdrSize));
headerBuf = HEAP.allocate(hdrSize);
readAtOffset(is, headerBuf, hdrSize, false, offset, pread);
headerBuf.rewind();
}
Expand DownExpand Up@@ -1782,7 +1790,7 @@ protected HFileBlock readBlockDataInternal(FSDataInputStream is, long offset,
// If nextBlockOnDiskSizeWithHeader is not zero, the onDiskBlock already
// contains the header of next block, so no need to set next block's header in it.
HFileBlock hFileBlock = new HFileBlock(curBlock, checksumSupport, MemoryType.EXCLUSIVE,
offset, nextBlockOnDiskSize, fileContext, allocator);
offset, nextBlockOnDiskSize, fileContext, intoHeap ? HEAP: allocator);
// Run check on uncompressed sizings.
if (!fileContext.isCompressedOrEncrypted()) {
hFileBlock.sanityCheckUncompressed();
Expand Down
26 changes: 12 additions & 14 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileBlockIndex.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -313,10 +313,13 @@ public BlockWithScanInfo loadDataBlockWithScanInfo(Cell key, HFileBlock currentB
int index = -1;

HFileBlock block = null;
boolean dataBlock = false;
KeyOnlyKeyValue tmpNextIndexKV = new KeyValue.KeyOnlyKeyValue();
while (true) {
try {
// Must initialize it with null here, because if don't and once an exception happen in
// readBlock, then we'll release the previous assigned block twice in the finally block.
// (See HBASE-22422)
block = null;

@ramkrish86ramkrish86May 27, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch.

if (currentBlock != null && currentBlock.getOffset() == currentOffset) {
// Avoid reading the same block again, even with caching turned off.
// This is crucial for compaction-type workload which might have
Expand All@@ -336,9 +339,8 @@ public BlockWithScanInfo loadDataBlockWithScanInfo(Cell key, HFileBlock currentB
// this also accounts for ENCODED_DATA
expectedBlockType = BlockType.DATA;
}
block =
cachingBlockReader.readBlock(currentOffset, currentOnDiskSize, shouldCache, pread,
isCompaction, true, expectedBlockType, expectedDataBlockEncoding);
block = cachingBlockReader.readBlock(currentOffset, currentOnDiskSize, shouldCache,
pread, isCompaction, true, expectedBlockType, expectedDataBlockEncoding);
}

if (block == null) {
Expand All@@ -348,7 +350,6 @@ public BlockWithScanInfo loadDataBlockWithScanInfo(Cell key, HFileBlock currentB

// Found a data block, break the loop and check our level in the tree.
if (block.getBlockType().isData()) {
dataBlock = true;
break;
}

Expand DownExpand Up@@ -381,15 +382,15 @@ public BlockWithScanInfo loadDataBlockWithScanInfo(Cell key, HFileBlock currentB
nextIndexedKey = tmpNextIndexKV;
}
} finally {
if (!dataBlock && block != null) {
if (block != null && !block.getBlockType().isData()) {
// Release the block immediately if it is not the data block
block.release();
}
}
}

if (lookupLevel != searchTreeLevel) {
assert dataBlock == true;
assert block.getBlockType().isData();
// Though we have retrieved a data block we have found an issue
// in the retrieved data block. Hence returned the block so that
// the ref count can be decremented
Expand All@@ -401,8 +402,7 @@ public BlockWithScanInfo loadDataBlockWithScanInfo(Cell key, HFileBlock currentB
}

// set the next indexed key for the current block.
BlockWithScanInfo blockWithScanInfo = new BlockWithScanInfo(block, nextIndexedKey);
return blockWithScanInfo;
return new BlockWithScanInfo(block, nextIndexedKey);
}

@Override
Expand DownExpand Up@@ -576,8 +576,7 @@ public HFileBlock seekToDataBlock(final Cell key, HFileBlock currentBlock, boole
boolean pread, boolean isCompaction, DataBlockEncoding expectedDataBlockEncoding)
throws IOException {
BlockWithScanInfo blockWithScanInfo = loadDataBlockWithScanInfo(key, currentBlock,
cacheBlocks,
pread, isCompaction, expectedDataBlockEncoding);
cacheBlocks, pread, isCompaction, expectedDataBlockEncoding);
if (blockWithScanInfo == null) {
return null;
} else {
Expand All@@ -600,9 +599,8 @@ public HFileBlock seekToDataBlock(final Cell key, HFileBlock currentBlock, boole
* @throws IOException
*/
public abstract BlockWithScanInfo loadDataBlockWithScanInfo(Cell key, HFileBlock currentBlock,
boolean cacheBlocks,
boolean pread, boolean isCompaction, DataBlockEncoding expectedDataBlockEncoding)
throws IOException;
boolean cacheBlocks, boolean pread, boolean isCompaction,
DataBlockEncoding expectedDataBlockEncoding) throws IOException;

/**
* An approximation to the {@link HFile}'s mid-key. Operates on block
Expand Down
21 changes: 10 additions & 11 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileReaderImpl.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -1128,15 +1128,13 @@ protected void readAndUpdateNewBlock(long firstDataBlockOffset) throws IOExcepti
updateCurrentBlock(newBlock);
}

protected int loadBlockAndSeekToKey(HFileBlock seekToBlock, Cell nextIndexedKey,
boolean rewind, Cell key, boolean seekBefore) throws IOException {
if (this.curBlock == null
|| this.curBlock.getOffset() != seekToBlock.getOffset()) {
protected int loadBlockAndSeekToKey(HFileBlock seekToBlock, Cell nextIndexedKey, boolean rewind,
Cell key, boolean seekBefore) throws IOException {
if (this.curBlock == null || this.curBlock.getOffset() != seekToBlock.getOffset()) {
updateCurrentBlock(seekToBlock);
} else if (rewind) {
blockBuffer.rewind();
}

// Update the nextIndexedKey
this.nextIndexedKey = nextIndexedKey;
return blockSeek(key, seekBefore);
Expand DownExpand Up@@ -1473,9 +1471,11 @@ public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize,
// Validate encoding type for data blocks. We include encoding
// type in the cache key, and we expect it to match on a cache hit.
if (cachedBlock.getDataBlockEncoding() != dataBlockEncoder.getDataBlockEncoding()) {
// Remember to release the block when in exceptional path.
cachedBlock.release();
throw new IOException("Cached block under key " + cacheKey + " "
+ "has wrong encoding: " + cachedBlock.getDataBlockEncoding() + " (expected: "
+ dataBlockEncoder.getDataBlockEncoding() + ")");
+ "has wrong encoding: " + cachedBlock.getDataBlockEncoding() + " (expected: "
+ dataBlockEncoder.getDataBlockEncoding() + ")");
}
}
// Cache-hit. Return!
Expand All@@ -1499,15 +1499,14 @@ public HFileBlock readBlock(long dataBlockOffset, long onDiskBlockSize,
BlockType.BlockCategory category = hfileBlock.getBlockType().getCategory();

// Cache the block if necessary
AtomicBoolean cachedRaw = new AtomicBoolean(false);
cacheConf.getBlockCache().ifPresent(cache -> {
if (cacheBlock && cacheConf.shouldCacheBlockOnRead(category)) {
cachedRaw.set(cacheConf.shouldCacheCompressed(category));
cache.cacheBlock(cacheKey, cachedRaw.get() ? hfileBlock : unpacked,
cache.cacheBlock(cacheKey,
cacheConf.shouldCacheCompressed(category) ? hfileBlock : unpacked,
cacheConf.isInMemory());
}
});
if (unpacked != hfileBlock && !cachedRaw.get()) {
if (unpacked != hfileBlock) {
Comment thread
Apache9 marked this conversation as resolved.
// End of life here if hfileBlock is an independent block.
hfileBlock.release();
}
Expand Down
23 changes: 16 additions & 7 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/LruBlockCache.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -504,7 +504,14 @@ private long updateSizeMetrics(LruCachedBlock cb, boolean evict) {
@Override
public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching, boolean repeat,
boolean updateCacheMetrics) {
LruCachedBlock cb = map.get(cacheKey);
LruCachedBlock cb = map.computeIfPresent(cacheKey, (key, val) -> {
// It will be referenced by RPC path, so increase here. NOTICE: Must do the retain inside
// this block. because if retain outside the map#computeIfPresent, the evictBlock may remove
// the block and release, then we're retaining a block with refCnt=0 which is disallowed.
// see HBASE-22422.
val.getBuffer().retain();
return val;
});
if (cb == null) {
if (!repeat && updateCacheMetrics) {
stats.miss(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
Expand DownExpand Up@@ -532,10 +539,10 @@ public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching, boolean repea
}
return null;
}
if (updateCacheMetrics) stats.hit(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
if (updateCacheMetrics) {
stats.hit(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
}
cb.access(count.incrementAndGet());
// It will be referenced by RPC path, so increase here.
cb.getBuffer().retain();
return cb.getBuffer();
}

Expand DownExpand Up@@ -592,16 +599,14 @@ protected long evictBlock(LruCachedBlock block, boolean evictedByEvictionProcess
if (previous == null) {
return 0;
}
// Decrease the block's reference count, and if refCount is 0, then it'll auto-deallocate.
previous.getBuffer().release();
updateSizeMetrics(block, true);
long val = elements.decrementAndGet();
if (LOG.isTraceEnabled()) {
long size = map.size();
assertCounterSanity(size, val);
}
if (block.getBuffer().getBlockType().isData()) {
dataBlockElements.decrement();
dataBlockElements.decrement();
}
if (evictedByEvictionProcess) {
// When the eviction of the block happened because of invalidation of HFiles, no need to
Expand All@@ -611,6 +616,10 @@ protected long evictBlock(LruCachedBlock block, boolean evictedByEvictionProcess
victimHandler.cacheBlock(block.getCacheKey(), block.getBuffer());
}
}
// Decrease the block's reference count, and if refCount is 0, then it'll auto-deallocate. DO
// NOT move this up because if do that then the victimHandler may access the buffer with
// refCnt = 0 which is disallowed.
previous.getBuffer().release();

@Apache9Apache9May 16, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So this is the problem. Mind explaining more? Why in victimHandler we will access the previous? And is it possible to add a UT?

@openinxopeninxMay 17, 2019•
edited
Loading

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why in victimHandler we will access the previous? And is it possible to add a UT?

For InclusiveCombinedBlockCache , we will move the evicted block from LRUCache to an larger L2 cache (such as MemcachedBlockCache for longer caching I think), So if release in line#596, then victimHandler will cache an block which point to an unknown area because its memory has been free.
Yeah, will provide a UT.

return block.heapSize();
}

Expand Down
29 changes: 19 additions & 10 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -42,6 +42,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.Lock;
Expand DownExpand Up@@ -1533,21 +1534,28 @@ public boolean containsKey(BlockCacheKey key) {
}

public RAMQueueEntry get(BlockCacheKey key) {
RAMQueueEntry re = delegate.get(key);
if (re != null) {
// It'll be referenced by RPC, so retain here.
return delegate.computeIfPresent(key, (k, re) -> {
// It'll be referenced by RPC, so retain atomically here. if the get and retain is not
// atomic, another thread may remove and release the block, when retaining in this thread we
// may retain a block with refCnt=0 which is disallowed. (see HBASE-22422)
re.getData().retain();
}
return re;
return re;
});
}

/**
* Return the previous associated value, or null if absent. It has the same meaning as
* {@link ConcurrentMap#putIfAbsent(Object, Object)}
*/
public RAMQueueEntry putIfAbsent(BlockCacheKey key, RAMQueueEntry entry) {
RAMQueueEntry previous = delegate.putIfAbsent(key, entry);
if (previous == null) {
AtomicBoolean absent = new AtomicBoolean(false);
RAMQueueEntry re = delegate.computeIfAbsent(key, k -> {
// The RAMCache reference to this entry, so reference count should be increment.
entry.getData().retain();
}
return previous;
absent.set(true);
return entry;
});
return absent.get() ? null : re;

@ramkrish86ramkrish86May 29, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think I got it now. You have changed the get() to computeIfPresent(). So if the remove has already removed the entry then the get() cannot do a retain(). So a similar change is also needed for putIfAbsent() also? Rest looks good to me.

@openinxopeninxMay 29, 2019

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, if don't make the put and retain in atomic , then if remove & release happen between the put and retain, finally we 're retain a block with refCnt=0 which is also disallowed. Thanks.

}

public boolean remove(BlockCacheKey key) {

@ramkrish86ramkrish86May 29, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While removing the atomicity is not needed? Because we do computeIfAbsent and that is already now atomically guarded?

Expand DownExpand Up@@ -1576,8 +1584,9 @@ public boolean isEmpty() {
public void clear() {
Iterator<Map.Entry<BlockCacheKey, RAMQueueEntry>> it = delegate.entrySet().iterator();
while (it.hasNext()) {
it.next().getValue().getData().release();
RAMQueueEntry re = it.next().getValue();
it.remove();
re.getData().release();
}
}
}
Expand Down
17 changes: 17 additions & 0 deletions hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCombinedBlockCache.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,11 +17,16 @@
*/
package org.apache.hadoop.hbase.io.hfile;

import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_IOENGINE_KEY;
import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_SIZE_KEY;
import static org.junit.Assert.assertEquals;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.io.hfile.CombinedBlockCache.CombinedCacheStats;
import org.apache.hadoop.hbase.testclassification.SmallTests;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
Expand All@@ -33,6 +38,8 @@ public class TestCombinedBlockCache {
public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestCombinedBlockCache.class);

private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();

@Test
public void testCombinedCacheStats() {
CacheStats lruCacheStats = new CacheStats("lruCacheStats", 2);
Expand DownExpand Up@@ -102,4 +109,14 @@ public void testCombinedCacheStats() {
assertEquals(0.75, stats.getHitRatioPastNPeriods(), delta);
assertEquals(0.8, stats.getHitCachingRatioPastNPeriods(), delta);
}

@Test
public void testMultiThreadGetAndEvictBlock() throws Exception {
Configuration conf = UTIL.getConfiguration();
conf.set(BUCKET_CACHE_IOENGINE_KEY, "offheap");
conf.setInt(BUCKET_CACHE_SIZE_KEY, 32);
BlockCache blockCache = BlockCacheFactory.createBlockCache(conf);
Assert.assertTrue(blockCache instanceof CombinedBlockCache);
TestLruBlockCache.testMultiThreadGetAndEvictBlockInternal(blockCache);
}
}
Loading