comnetwork commented on a change in pull request #3680:
URL: https://github.com/apache/hbase/pull/3680#discussion_r708214160
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
##########
@@ -906,134 +968,135 @@ public void run() {
}
LOG.info(this.getName() + " exiting, cacheEnabled=" + cacheEnabled);
}
+ }
- /**
- * Put the new bucket entry into backingMap. Notice that we are allowed to
replace the existing
- * cache with a new block for the same cache key. there's a corner case:
one thread cache a
- * block in ramCache, copy to io-engine and add a bucket entry to
backingMap. Caching another
- * new block with the same cache key do the same thing for the same cache
key, so if not evict
- * the previous bucket entry, then memory leak happen because the previous
bucketEntry is gone
- * but the bucketAllocator do not free its memory.
- * @see BlockCacheUtil#shouldReplaceExistingCacheBlock(BlockCache
blockCache,BlockCacheKey
- * cacheKey, Cacheable newBlock)
- * @param key Block cache key
- * @param bucketEntry Bucket entry to put into backingMap.
- */
- private void putIntoBackingMap(BlockCacheKey key, BucketEntry bucketEntry)
{
- BucketEntry previousEntry = backingMap.put(key, bucketEntry);
- if (previousEntry != null && previousEntry != bucketEntry) {
- previousEntry.withWriteLock(offsetLock, () -> {
- blockEvicted(key, previousEntry, false);
- return null;
- });
- }
+ /**
+ * Put the new bucket entry into backingMap. Notice that we are allowed to
replace the existing
+ * cache with a new block for the same cache key. there's a corner case: one
thread cache a block
+ * in ramCache, copy to io-engine and add a bucket entry to backingMap.
Caching another new block
+ * with the same cache key do the same thing for the same cache key, so if
not evict the previous
+ * bucket entry, then memory leak happen because the previous bucketEntry is
gone but the
+ * bucketAllocator do not free its memory.
+ * @see BlockCacheUtil#shouldReplaceExistingCacheBlock(BlockCache
blockCache,BlockCacheKey
+ * cacheKey, Cacheable newBlock)
+ * @param key Block cache key
+ * @param bucketEntry Bucket entry to put into backingMap.
+ */
+ protected void putIntoBackingMap(BlockCacheKey key, BucketEntry bucketEntry)
{
+ BucketEntry previousEntry = backingMap.put(key, bucketEntry);
+ if (previousEntry != null && previousEntry != bucketEntry) {
+ previousEntry.withWriteLock(offsetLock, () -> {
+ blockEvicted(key, previousEntry, false);
+ return null;
+ });
}
+ }
- /**
- * Flush the entries in ramCache to IOEngine and add bucket entry to
backingMap.
- * Process all that are passed in even if failure being sure to remove
from ramCache else we'll
- * never undo the references and we'll OOME.
- * @param entries Presumes list passed in here will be processed by this
invocation only. No
- * interference expected.
- * @throws InterruptedException
- */
- void doDrain(final List<RAMQueueEntry> entries) throws
InterruptedException {
- if (entries.isEmpty()) {
- return;
- }
- // This method is a little hard to follow. We run through the passed in
entries and for each
- // successful add, we add a non-null BucketEntry to the below
bucketEntries. Later we must
- // do cleanup making sure we've cleared ramCache of all entries
regardless of whether we
- // successfully added the item to the bucketcache; if we don't do the
cleanup, we'll OOME by
- // filling ramCache. We do the clean up by again running through the
passed in entries
- // doing extra work when we find a non-null bucketEntries corresponding
entry.
- final int size = entries.size();
- BucketEntry[] bucketEntries = new BucketEntry[size];
- // Index updated inside loop if success or if we can't succeed. We retry
if cache is full
- // when we go to add an entry by going around the loop again without
upping the index.
- int index = 0;
- while (cacheEnabled && index < size) {
- RAMQueueEntry re = null;
- try {
- re = entries.get(index);
- if (re == null) {
- LOG.warn("Couldn't get entry or changed on us; who else is messing
with it?");
- index++;
- continue;
- }
- BucketEntry bucketEntry = re.writeToCache(ioEngine, bucketAllocator,
realCacheSize);
- // Successfully added. Up index and add bucketEntry. Clear io
exceptions.
- bucketEntries[index] = bucketEntry;
- if (ioErrorStartTime > 0) {
- ioErrorStartTime = -1;
- }
- index++;
- } catch (BucketAllocatorException fle) {
- LOG.warn("Failed allocation for " + (re == null ? "" : re.getKey())
+ "; " + fle);
- // Presume can't add. Too big? Move index on. Entry will be cleared
from ramCache below.
- bucketEntries[index] = null;
+ /**
Review comment:
Here I move `doDrain`from `WriterThread `to `BucketCache `to facilitate
test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
