From a6cd7a025163dfc9ecb5e5dc409dfc6989d86cb4 Mon Sep 17 00:00:00 2001 From: Michael Stack Date: Sat, 14 Apr 2018 10:49:40 -0700 Subject: [PATCH] HBASE-20411 Ameliorate MutableSegment synchronize Change the synchronize across three volatiles incrementing our memory sizings. Instead: * Make MemStoreSize immutable. Create a new instance on every increment. * Undo MemStoreSizing being an instance of MemStoreSize; instead it has-a. * Make two MemStoreSizing implementations; one thread-safe, the other not. * Use an AtomicReference#checkAndPut (lockless) where concurrent updates * Otherwise, use unsynchronized accounting. TODO: Conservative using thread-safe accounting when unclear. Fix. TODO: Use this technique accounting at the global level too. A hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/LocalMemStoreSizing.java Non-thread-safe memory size accounting. M hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSize.java Make MemStoreSize immutable, just a container for three longs. M hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSizing.java Make this an Interface. Implementations are a thread-safe instance and a non-thread-safe version. A hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ThreadSafeMemStoreSizing.java Thread-safe version. --- .../hbase/regionserver/AbstractMemStore.java | 8 +- .../hbase/regionserver/CompactingMemStore.java | 17 ++- .../hbase/regionserver/CompactionPipeline.java | 21 ++-- .../hadoop/hbase/regionserver/DefaultMemStore.java | 22 ++-- .../apache/hadoop/hbase/regionserver/HRegion.java | 114 ++++++++++----------- .../apache/hadoop/hbase/regionserver/HStore.java | 2 +- .../hadoop/hbase/regionserver/MemStoreFlusher.java | 13 ++- .../hadoop/hbase/regionserver/MemStoreSize.java | 95 +++++++++-------- .../hadoop/hbase/regionserver/MemStoreSizing.java | 60 ++++------- .../regionserver/NonThreadSafeMemStoreSizing.java | 64 ++++++++++++ .../hbase/regionserver/RegionServerAccounting.java | 8 +- .../apache/hadoop/hbase/regionserver/Segment.java | 29 +++--- .../regionserver/ThreadSafeMemStoreSizing.java | 62 +++++++++++ .../hbase/regionserver/TestCompactingMemStore.java | 3 +- .../TestCompactingToCellFlatMapMemStore.java | 7 +- .../hbase/regionserver/TestDefaultMemStore.java | 13 +-- .../hadoop/hbase/regionserver/TestHStore.java | 14 +-- 17 files changed, 340 insertions(+), 212 deletions(-) create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/NonThreadSafeMemStoreSizing.java create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ThreadSafeMemStoreSizing.java diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java index e6fd04d4bb..ce6d8ad78d 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AbstractMemStore.java @@ -167,13 +167,7 @@ public abstract class AbstractMemStore implements MemStore { @Override public MemStoreSize getSnapshotSize() { - return getSnapshotSizing(); - } - - MemStoreSizing getSnapshotSizing() { - return new MemStoreSizing(this.snapshot.keySize(), - this.snapshot.heapSize(), - this.snapshot.offHeapSize()); + return this.snapshot.getMemStoreSize(); } @Override diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java index ea4cc0870d..edeb17a76a 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java @@ -155,12 +155,12 @@ public class CompactingMemStore extends AbstractMemStore { */ @Override public MemStoreSize size() { - MemStoreSizing memstoreSizing = new MemStoreSizing(); + MemStoreSizing memstoreSizing = new NonThreadSafeMemStoreSizing(); memstoreSizing.incMemStoreSize(active.getMemStoreSize()); for (Segment item : pipeline.getSegments()) { memstoreSizing.incMemStoreSize(item.getMemStoreSize()); } - return memstoreSizing; + return memstoreSizing.getMemStoreSize(); } /** @@ -222,18 +222,17 @@ public class CompactingMemStore extends AbstractMemStore { */ @Override public MemStoreSize getFlushableSize() { - MemStoreSizing snapshotSizing = getSnapshotSizing(); - if (snapshotSizing.getDataSize() == 0) { + MemStoreSize mss = getSnapshotSize(); + if (mss.getDataSize() == 0) { // if snapshot is empty the tail of the pipeline (or everything in the memstore) is flushed if (compositeSnapshot) { - snapshotSizing = pipeline.getPipelineSizing(); - snapshotSizing.incMemStoreSize(active.getMemStoreSize()); + MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing(pipeline.getPipelineSize()); + mss = memStoreSizing.incMemStoreSize(this.active.getMemStoreSize()); } else { - snapshotSizing = pipeline.getTailSizing(); + mss = pipeline.getTailSize(); } } - return snapshotSizing.getDataSize() > 0 ? snapshotSizing - : new MemStoreSize(active.getMemStoreSize()); + return mss.getDataSize() > 0? mss: this.active.getMemStoreSize(); } @Override diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java index 6e53cfcd84..02ba704aa7 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java @@ -23,6 +23,7 @@ import java.util.Iterator; import java.util.LinkedList; import java.util.List; +import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; import org.apache.yetus.audience.InterfaceAudience; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -147,6 +148,8 @@ public class CompactionPipeline { } long offHeapSizeDelta = suffixOffHeapSize - newOffHeapSize; long heapSizeDelta = suffixHeapSize - newHeapSize; + // FIX THIS: Create a class just to carry the three longs. We then create new class + // internally. region.addMemStoreSize(new MemStoreSize(-dataSizeDelta, -heapSizeDelta, -offHeapSizeDelta)); LOG.debug("Suffix data size={}, new segment data size={}, " + "suffix heap size={}," + "new segment heap size={}" @@ -211,15 +214,16 @@ public class CompactionPipeline { int i = 0; for (ImmutableSegment s : pipeline) { if ( s.canBeFlattened() ) { - MemStoreSizing newMemstoreAccounting = new MemStoreSizing(); // the size to be updated + MemStoreSizing newMemstoreAccounting = new NonThreadSafeMemStoreSizing(); // size to be updated ImmutableSegment newS = SegmentFactory.instance().createImmutableSegmentByFlattening( (CSLMImmutableSegment)s,idxType,newMemstoreAccounting,action); replaceAtIndex(i,newS); if(region != null) { // update the global memstore size counter // upon flattening there is no change in the data size - region.addMemStoreSize(new MemStoreSize(0, newMemstoreAccounting.getHeapSize(), - newMemstoreAccounting.getOffHeapSize())); + MemStoreSize mss = newMemstoreAccounting.getMemStoreSize(); + Preconditions.checkArgument(mss.getDataSize() == 0, "Not zero!"); + region.addMemStoreSize(mss); } LOG.debug("Compaction pipeline segment {} flattened", s); return true; @@ -254,19 +258,18 @@ public class CompactionPipeline { return minSequenceId; } - public MemStoreSizing getTailSizing() { + public MemStoreSize getTailSize() { LinkedList localCopy = readOnlyCopy; - if (localCopy.isEmpty()) return new MemStoreSizing(); - return new MemStoreSizing(localCopy.peekLast().getMemStoreSize()); + return localCopy.isEmpty()? new MemStoreSize(): localCopy.peekLast().getMemStoreSize(); } - public MemStoreSizing getPipelineSizing() { - MemStoreSizing memStoreSizing = new MemStoreSizing(); + public MemStoreSize getPipelineSize() { + MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing(); LinkedList localCopy = readOnlyCopy; for (Segment segment : localCopy) { memStoreSizing.incMemStoreSize(segment.getMemStoreSize()); } - return memStoreSizing; + return memStoreSizing.getMemStoreSize(); } private void swapSuffix(List suffix, ImmutableSegment segment, diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java index 9ef6a6cc2b..0b89f8b263 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultMemStore.java @@ -105,8 +105,7 @@ public class DefaultMemStore extends AbstractMemStore { @Override public MemStoreSize getFlushableSize() { MemStoreSize snapshotSize = getSnapshotSize(); - return snapshotSize.getDataSize() > 0 ? snapshotSize - : new MemStoreSize(active.getMemStoreSize()); + return snapshotSize.getDataSize() > 0 ? snapshotSize: active.getMemStoreSize(); } @Override @@ -155,7 +154,7 @@ public class DefaultMemStore extends AbstractMemStore { @Override public MemStoreSize size() { - return new MemStoreSize(active.getMemStoreSize()); + return active.getMemStoreSize(); } /** @@ -194,26 +193,27 @@ public class DefaultMemStore extends AbstractMemStore { byte [] fam = Bytes.toBytes("col"); byte [] qf = Bytes.toBytes("umn"); byte [] empty = new byte[0]; - MemStoreSizing memstoreSizing = new MemStoreSizing(); + MemStoreSizing memstoreSizing = new NonThreadSafeMemStoreSizing(); for (int i = 0; i < count; i++) { // Give each its own ts memstore1.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, empty), memstoreSizing); } - LOG.info("memstore1 estimated size=" - + (memstoreSizing.getDataSize() + memstoreSizing.getHeapSize())); + LOG.info("memstore1 estimated size={}", memstoreSizing.getMemStoreSize().getDataSize() + + memstoreSizing.getMemStoreSize().getHeapSize()); for (int i = 0; i < count; i++) { memstore1.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, empty), memstoreSizing); } - LOG.info("memstore1 estimated size (2nd loading of same data)=" - + (memstoreSizing.getDataSize() + memstoreSizing.getHeapSize())); + LOG.info("memstore1 estimated size (2nd loading of same data)={}", + memstoreSizing.getMemStoreSize().getDataSize() + + memstoreSizing.getMemStoreSize().getHeapSize()); // Make a variably sized memstore. DefaultMemStore memstore2 = new DefaultMemStore(); - memstoreSizing = new MemStoreSizing(); + memstoreSizing = new NonThreadSafeMemStoreSizing(); for (int i = 0; i < count; i++) { memstore2.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, new byte[i]), memstoreSizing); } - LOG.info("memstore2 estimated size=" - + (memstoreSizing.getDataSize() + memstoreSizing.getHeapSize())); + LOG.info("memstore2 estimated size={}", memstoreSizing.getMemStoreSize().getDataSize() + + memstoreSizing.getMemStoreSize().getHeapSize()); final int seconds = 30; LOG.info("Waiting " + seconds + " seconds while heap dump is taken"); LOG.info("Exiting."); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java index f41724719e..e079959602 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java @@ -287,7 +287,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi private Map coprocessorServiceHandlers = Maps.newHashMap(); // Track data size in all memstores - private final MemStoreSizing memStoreSize = new MemStoreSizing(); + private final MemStoreSizing memStoreSizing = new ThreadSafeMemStoreSizing(); private final RegionServicesForStores regionServicesForStores = new RegionServicesForStores(this); // Debug possible data loss due to WAL off @@ -1195,32 +1195,24 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi if (this.rsAccounting != null) { rsAccounting.incGlobalMemStoreSize(memStoreSize); } - long dataSize; - synchronized (this.memStoreSize) { - this.memStoreSize.incMemStoreSize(memStoreSize); - dataSize = this.memStoreSize.getDataSize(); - } - checkNegativeMemStoreDataSize(dataSize, memStoreSize.getDataSize()); + MemStoreSize mss = this.memStoreSizing.incMemStoreSize(memStoreSize); + checkNegativeMemStoreDataSize(mss.getDataSize(), memStoreSize.getDataSize()); } public void decrMemStoreSize(MemStoreSize memStoreSize) { if (this.rsAccounting != null) { rsAccounting.decGlobalMemStoreSize(memStoreSize); } - long size; - synchronized (this.memStoreSize) { - this.memStoreSize.decMemStoreSize(memStoreSize); - size = this.memStoreSize.getDataSize(); - } - checkNegativeMemStoreDataSize(size, -memStoreSize.getDataSize()); + MemStoreSize mss = this.memStoreSizing.decMemStoreSize(memStoreSize); + checkNegativeMemStoreDataSize(mss.getDataSize(), -memStoreSize.getDataSize()); } private void checkNegativeMemStoreDataSize(long memStoreDataSize, long delta) { - // This is extremely bad if we make memStoreSize negative. Log as much info on the offending - // caller as possible. (memStoreSize might be a negative value already -- freeing memory) + // This is extremely bad if we make memStoreSizing negative. Log as much info on the offending + // caller as possible. (memStoreSizing might be a negative value already -- freeing memory) if (memStoreDataSize < 0) { LOG.error("Asked to modify this region's (" + this.toString() - + ") memStoreSize to a negative value which is incorrect. Current memStoreSize=" + + ") memStoreSizing to a negative value which is incorrect. Current memStoreSizing=" + (memStoreDataSize - delta) + ", delta=" + delta, new Exception()); } } @@ -1255,17 +1247,17 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi @Override public long getMemStoreDataSize() { - return memStoreSize.getDataSize(); + return memStoreSizing.getMemStoreSize().getDataSize(); } @Override public long getMemStoreHeapSize() { - return memStoreSize.getHeapSize(); + return memStoreSizing.getMemStoreSize().getHeapSize(); } @Override public long getMemStoreOffHeapSize() { - return memStoreSize.getOffHeapSize(); + return memStoreSizing.getMemStoreSize().getOffHeapSize(); } /** @return store services for this region, to access services required by store level needs */ @@ -1536,7 +1528,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi int failedfFlushCount = 0; int flushCount = 0; long tmp = 0; - long remainingSize = this.memStoreSize.getDataSize(); + long remainingSize = this.memStoreSizing.getMemStoreSize().getDataSize(); while (remainingSize > 0) { try { internalFlushcache(status); @@ -1545,7 +1537,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi " (carrying snapshot?) " + this); } flushCount++; - tmp = this.memStoreSize.getDataSize(); + tmp = this.memStoreSizing.getMemStoreSize().getDataSize(); if (tmp >= remainingSize) { failedfFlushCount++; } @@ -1585,7 +1577,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi getRegionServerServices().abort("Assertion failed while closing store " + getRegionInfo().getRegionNameAsString() + " " + store + ". flushableSize expected=0, actual= " + flushableSize - + ". Current memStoreSize=" + getMemStoreDataSize() + ". Maybe a coprocessor " + + ". Current memStoreSize=" + this.memStoreSizing.getMemStoreSize() + + ". Maybe a coprocessor " + "operation failed and left the memstore in a partially updated state.", null); } } @@ -1627,10 +1620,11 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi } this.closed.set(true); + MemStoreSize mss = this.memStoreSizing.getMemStoreSize(); if (!canFlush) { - this.decrMemStoreSize(new MemStoreSize(memStoreSize)); - } else if (memStoreSize.getDataSize() != 0) { - LOG.error("Memstore data size is " + memStoreSize.getDataSize()); + decrMemStoreSize(mss); + } else if (mss.getDataSize() != 0) { + LOG.error("Memstore data size is {}", mss.getDataSize()); } if (coprocessorHost != null) { status.setStatus("Running coprocessor post-close hooks"); @@ -1763,7 +1757,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi * @return True if its worth doing a flush before we put up the close flag. */ private boolean worthPreFlushing() { - return this.memStoreSize.getDataSize() > + return this.memStoreSizing.getMemStoreSize().getDataSize() > this.conf.getLong("hbase.hregion.preclose.flush.size", 1024 * 1024 * 5); } @@ -2381,12 +2375,12 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi // bulk loaded file between memory and existing hfiles. It wants a good seqeunceId that belongs // to no other that it can use to associate with the bulk load. Hence this little dance below // to go get one. - if (this.memStoreSize.getDataSize() <= 0) { + if (this.memStoreSizing.getMemStoreSize().getDataSize() <= 0) { // Take an update lock so no edits can come into memory just yet. this.updatesLock.writeLock().lock(); WriteEntry writeEntry = null; try { - if (this.memStoreSize.getDataSize() <= 0) { + if (this.memStoreSizing.getMemStoreSize().getDataSize() <= 0) { // Presume that if there are still no edits in the memstore, then there are no edits for // this region out in the WAL subsystem so no need to do any trickery clearing out // edits in the WAL sub-system. Up the sequence number so the resulting flush id is for @@ -2428,7 +2422,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi // block waiting for the lock for internal flush this.updatesLock.writeLock().lock(); status.setStatus("Preparing flush snapshotting stores in " + getRegionInfo().getEncodedName()); - MemStoreSizing totalSizeOfFlushableStores = new MemStoreSizing(); + MemStoreSizing totalSizeOfFlushableStores = new NonThreadSafeMemStoreSizing(); Map flushedFamilyNamesToSeq = new HashMap<>(); for (HStore store : storesToFlush) { @@ -2522,9 +2516,10 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi .append(StringUtils.byteDesc(store.getFlushableSize().getDataSize())); } } + MemStoreSize mss = this.memStoreSizing.getMemStoreSize(); LOG.info("Flushing " + + storesToFlush.size() + "/" + stores.size() + " column families," + - " memstore data size=" + StringUtils.byteDesc(this.memStoreSize.getDataSize()) + - " memstore heap size=" + StringUtils.byteDesc(this.memStoreSize.getHeapSize()) + + " memstore data size=" + StringUtils.byteDesc(mss.getDataSize()) + + " memstore heap size=" + StringUtils.byteDesc(mss.getHeapSize()) + ((perCfExtras != null && perCfExtras.length() > 0)? perCfExtras.toString(): "") + ((wal != null) ? "" : "; WAL is null, using passed sequenceid=" + sequenceId)); } @@ -2644,7 +2639,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi storeFlushCtxs.clear(); // Set down the memstore size by amount of flush. - this.decrMemStoreSize(prepareResult.totalFlushableSize); + this.decrMemStoreSize(prepareResult.totalFlushableSize.getMemStoreSize()); if (wal != null) { // write flush marker to WAL. If fail, we should throw DroppedSnapshotException @@ -2711,12 +2706,13 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi } long time = EnvironmentEdgeManager.currentTime() - startTime; - long flushableDataSize = prepareResult.totalFlushableSize.getDataSize(); - long flushableHeapSize = prepareResult.totalFlushableSize.getHeapSize(); - long memstoresize = this.memStoreSize.getDataSize(); + MemStoreSize mss = prepareResult.totalFlushableSize.getMemStoreSize(); + long flushableDataSize = mss.getDataSize(); + long flushableHeapSize = mss.getHeapSize(); + long memstoresize = this.memStoreSizing.getMemStoreSize().getDataSize(); String msg = "Finished memstore flush;" - + " data size ~" + StringUtils.byteDesc(flushableDataSize) + "/" + flushableDataSize - + ", heap size ~" + StringUtils.byteDesc(flushableHeapSize) + "/" + flushableHeapSize + + " data size ~" + StringUtils.byteDesc(mss.getDataSize()) + "/" + mss.getDataSize() + + ", heap size ~" + StringUtils.byteDesc(mss.getHeapSize()) + "/" + mss.getHeapSize() + ", currentsize=" + StringUtils.byteDesc(memstoresize) + "/" + memstoresize + " for " + this.getRegionInfo().getEncodedName() + " in " + time + "ms, sequenceid=" + flushOpSeqId + ", compaction requested=" + compactionRequested @@ -2726,7 +2722,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi if (rsServices != null && rsServices.getMetrics() != null) { rsServices.getMetrics().updateFlush(time - startTime, - prepareResult.totalFlushableSize.getDataSize(), flushedOutputFileSize); + mss.getDataSize(), flushedOutputFileSize); } return new FlushResultImpl(compactionRequested ? @@ -3038,7 +3034,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi protected void writeMiniBatchOperationsToMemStore( final MiniBatchOperationInProgress miniBatchOp, final long writeNumber) throws IOException { - MemStoreSizing memStoreAccounting = new MemStoreSizing(); + MemStoreSizing memStoreAccounting = new NonThreadSafeMemStoreSizing(); visitBatchOperations(true, miniBatchOp.getLastIndexExclusive(), (int index) -> { // We need to update the sequence id for following reasons. // 1) If the op is in replay mode, FSWALEntry#stampRegionSequenceId won't stamp sequence id. @@ -3051,7 +3047,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi return true; }); // update memStore size - region.incMemStoreSize(memStoreAccounting); + region.incMemStoreSize(memStoreAccounting.getMemStoreSize()); } public boolean isDone() { @@ -4178,8 +4174,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi // If catalog region, do not impose resource constraints or block updates. if (this.getRegionInfo().isMetaRegion()) return; - if (this.memStoreSize.getHeapSize() - + this.memStoreSize.getOffHeapSize() > this.blockingMemStoreSize) { + MemStoreSize mss = this.memStoreSizing.getMemStoreSize(); + if (mss.getHeapSize() + mss.getOffHeapSize() > this.blockingMemStoreSize) { blockedRequestsCount.increment(); requestFlush(); // Don't print current limit because it will vary too much. The message is used as a key @@ -4549,7 +4545,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi } boolean flush = false; - MemStoreSizing memstoreSize = new MemStoreSizing(); + MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing(); for (Cell cell: val.getCells()) { // Check this edit is for me. Also, guard against writing the special // METACOLUMN info such as HBASE::CACHEFLUSH entries @@ -4592,15 +4588,15 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi } PrivateCellUtil.setSequenceId(cell, currentReplaySeqId); - restoreEdit(store, cell, memstoreSize); + restoreEdit(store, cell, memStoreSizing); editsCount++; } + MemStoreSize mss = memStoreSizing.getMemStoreSize(); if (this.rsAccounting != null) { - rsAccounting.addRegionReplayEditsSize(getRegionInfo().getRegionName(), - memstoreSize); + rsAccounting.addRegionReplayEditsSize(getRegionInfo().getRegionName(), mss); } - incMemStoreSize(memstoreSize); - flush = isFlushSize(this.memStoreSize); + incMemStoreSize(mss); + flush = isFlushSize(this.memStoreSizing.getMemStoreSize()); if (flush) { internalFlushcache(null, currentEditSeqId, stores.values(), status, false, FlushLifeCycleTracker.DUMMY); @@ -4910,7 +4906,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi replayFlushInStores(flush, prepareFlushResult, true); // Set down the memstore size by amount of flush. - this.decrMemStoreSize(prepareFlushResult.totalFlushableSize); + this.decrMemStoreSize(prepareFlushResult.totalFlushableSize.getMemStoreSize()); this.prepareFlushResult = null; writestate.flushing = false; @@ -4943,7 +4939,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi replayFlushInStores(flush, prepareFlushResult, true); // Set down the memstore size by amount of flush. - this.decrMemStoreSize(prepareFlushResult.totalFlushableSize); + this.decrMemStoreSize(prepareFlushResult.totalFlushableSize.getMemStoreSize()); // Inspect the memstore contents to see whether the memstore contains only edits // with seqId smaller than the flush seqId. If so, we can discard those edits. @@ -5047,7 +5043,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi * @throws IOException */ private MemStoreSize dropMemStoreContentsForSeqId(long seqId, HStore store) throws IOException { - MemStoreSizing totalFreedSize = new MemStoreSizing(); + MemStoreSizing totalFreedSize = new NonThreadSafeMemStoreSizing(); this.updatesLock.writeLock().lock(); try { @@ -5074,7 +5070,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi } finally { this.updatesLock.writeLock().unlock(); } - return totalFreedSize; + return totalFreedSize.getMemStoreSize(); } private MemStoreSize doDropStoreMemStoreContentsForSeqId(HStore s, long currentSeqId) @@ -7278,8 +7274,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi return null; } ClientProtos.RegionLoadStats.Builder stats = ClientProtos.RegionLoadStats.newBuilder(); - stats.setMemStoreLoad((int) (Math.min(100, (this.memStoreSize.getHeapSize() * 100) / this - .memstoreFlushSize))); + stats.setMemStoreLoad((int) (Math.min(100, + (this.memStoreSizing.getMemStoreSize().getHeapSize() * 100) / this.memstoreFlushSize))); if (rsServices.getHeapMemoryManager() != null) { // the HeapMemoryManager uses -0.0 to signal a problem asking the JVM, // so we could just do the calculation below and we'll get a 0. @@ -7340,7 +7336,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi // This is assigned by mvcc either explicity in the below or in the guts of the WAL append // when it assigns the edit a sequencedid (A.K.A the mvcc write number). WriteEntry writeEntry = null; - MemStoreSizing memstoreAccounting = new MemStoreSizing(); + MemStoreSizing memstoreAccounting = new NonThreadSafeMemStoreSizing(); try { boolean success = false; try { @@ -7426,7 +7422,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi } finally { closeRegionOperation(); if (!mutations.isEmpty()) { - this.incMemStoreSize(memstoreAccounting); + this.incMemStoreSize(memstoreAccounting.getMemStoreSize()); requestFlushIfNeeded(); } } @@ -7530,7 +7526,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi startRegionOperation(op); List results = returnResults? new ArrayList<>(mutation.size()): null; RowLock rowLock = null; - MemStoreSizing memstoreAccounting = new MemStoreSizing(); + MemStoreSizing memstoreAccounting = new NonThreadSafeMemStoreSizing(); try { rowLock = getRowLockInternal(mutation.getRow(), false, null); lock(this.updatesLock.readLock()); @@ -7580,7 +7576,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi rowLock.release(); } // Request a cache flush if over the limit. Do it outside update lock. - incMemStoreSize(memstoreAccounting); + incMemStoreSize(memstoreAccounting.getMemStoreSize()); requestFlushIfNeeded(); closeRegionOperation(op); if (this.metricsRegion != null) { @@ -8459,7 +8455,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi } private void requestFlushIfNeeded() throws RegionTooBusyException { - if(isFlushSize(memStoreSize)) { + if(isFlushSize(this.memStoreSizing.getMemStoreSize())) { requestFlush(); } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java index 351a84704e..4232f83b1d 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java @@ -2203,7 +2203,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation, Propagat this.cacheFlushCount = snapshot.getCellsCount(); this.cacheFlushSize = snapshot.getDataSize(); committedFiles = new ArrayList<>(1); - return new MemStoreSize(snapshot.getMemStoreSize()); + return snapshot.getMemStoreSize(); } @Override diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java index 1b7e0bc512..e4cc405062 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java @@ -550,8 +550,9 @@ class MemStoreFlusher implements FlushRequester { // If this is first time we've been put off, then emit a log message. if (fqe.getRequeueCount() <= 0) { // Note: We don't impose blockingStoreFiles constraint on meta regions - LOG.warn("Region " + region.getRegionInfo().getEncodedName() + " has too many " + - "store files; delaying flush up to " + this.blockingWaitTime + "ms"); + LOG.warn("{} has too many store files({}); delaying flush up to {} ms", + region.getRegionInfo().getEncodedName(), getStoreFileCount(region), + this.blockingWaitTime); if (!this.server.compactSplitThread.requestSplit(region)) { try { this.server.compactSplitThread.requestSystemCompaction(region, @@ -672,6 +673,14 @@ class MemStoreFlusher implements FlushRequester { return false; } + private int getStoreFileCount(Region region) { + int count = 0; + for (Store store : region.getStores()) { + count += store.getStorefilesCount(); + } + return count; + } + /** * Check if the regionserver's memstore memory usage is greater than the * limit. If so, flush regions with the biggest memstores until we're down diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSize.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSize.java index 382e6e9069..e2759a2e55 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSize.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSize.java @@ -19,64 +19,81 @@ package org.apache.hadoop.hbase.regionserver; import org.apache.hadoop.hbase.HBaseInterfaceAudience; import org.apache.yetus.audience.InterfaceAudience; - /** * Reports the data size part and total heap space occupied by the MemStore. - * Read-only. + * MemStore size tracks 3 sizes: + *
    + *
  1. data size: the aggregated size of all key-value not including meta data such as + * index, time range etc. + * + *
  2. heap size: the aggregated size of all data that is allocated on-heap including all + * key-values that reside on-heap and the metadata that resides on-heap + *
  3. + *
  4. off-heap size: the aggregated size of all data that is allocated off-heap including all + * key-values that reside off-heap and the metadata that resides off-heap + * + * + * 3 examples to illustrate their usage: + *

    + * Consider a store with 100MB of key-values allocated on-heap and 20MB of metadata allocated + * on-heap. The counters are <100MB, 120MB, 0>, respectively. + *

    + *

    Consider a store with 100MB of key-values allocated off-heap and 20MB of metadata + * allocated on-heap (e.g, CAM index). The counters are <100MB, 20MB, 100MB>, respectively. + *

    + *

    + * Consider a store with 100MB of key-values from which 95MB are allocated off-heap and 5MB + * are allocated on-heap (e.g., due to upserts) and 20MB of metadata from which 15MB allocated + * off-heap (e.g, CCM index) and 5MB allocated on-heap (e.g, CSLM index in active). + * The counters are <100MB, 10MB, 110MB>, respectively. + *

    + *

    Immutable!

    * @see MemStoreSizing */ @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC) public class MemStoreSize { - // MemStore size tracks 3 sizes: - // (1) data size: the aggregated size of all key-value not including meta data such as - // index, time range etc. - // (2) heap size: the aggregated size of all data that is allocated on-heap including all - // key-values that reside on-heap and the metadata that resides on-heap - // (3) off-heap size: the aggregated size of all data that is allocated off-heap including all - // key-values that reside off-heap and the metadata that resides off-heap - // - // 3 examples to illustrate their usage: - // Consider a store with 100MB of key-values allocated on-heap and 20MB of metadata allocated - // on-heap. The counters are <100MB, 120MB, 0>, respectively. - // Consider a store with 100MB of key-values allocated off-heap and 20MB of metadata - // allocated on-heap (e.g, CAM index). The counters are <100MB, 20MB, 100MB>, respectively. - // Consider a store with 100MB of key-values from which 95MB are allocated off-heap and 5MB - // are allocated on-heap (e.g., due to upserts) and 20MB of metadata from which 15MB allocated - // off-heap (e.g, CCM index) and 5MB allocated on-heap (e.g, CSLM index in active). - // The counters are <100MB, 10MB, 110MB>, respectively. - /** *'dataSize' tracks the Cell's data bytes size alone (Key bytes, value bytes). A cell's data can * be in on heap or off heap area depending on the MSLAB and its configuration to be using on heap * or off heap LABs */ - protected volatile long dataSize; + private final long dataSize; /** 'heapSize' tracks all Cell's heap size occupancy. This will include Cell POJO heap overhead. * When Cells in on heap area, this will include the cells data size as well. */ - protected volatile long heapSize; + private final long heapSize; /** off-heap size: the aggregated size of all data that is allocated off-heap including all * key-values that reside off-heap and the metadata that resides off-heap */ - protected volatile long offHeapSize; + private final long offHeapSize; - public MemStoreSize() { + /** + * Package private constructor. + */ + MemStoreSize() { this(0L, 0L, 0L); } - public MemStoreSize(long dataSize, long heapSize, long offHeapSize) { + /** + * Package private constructor. + */ + MemStoreSize(long dataSize, long heapSize, long offHeapSize) { this.dataSize = dataSize; this.heapSize = heapSize; this.offHeapSize = offHeapSize; } - protected MemStoreSize(MemStoreSize memStoreSize) { - this.dataSize = memStoreSize.dataSize; - this.heapSize = memStoreSize.heapSize; - this.offHeapSize = memStoreSize.offHeapSize; + /** + * Package private constructor. + */ + MemStoreSize(MemStoreSize memStoreSize) { + this.dataSize = memStoreSize.getDataSize(); + this.heapSize = memStoreSize.getHeapSize(); + this.offHeapSize = memStoreSize.getOffHeapSize(); } + public boolean isEmpty() { return this.dataSize == 0 && this.heapSize == 0 && this.offHeapSize == 0; } @@ -98,24 +115,22 @@ public class MemStoreSize { if (obj == null || getClass() != obj.getClass()) { return false; } - MemStoreSize other = (MemStoreSize) obj; - return this.dataSize == other.dataSize - && this.heapSize == other.heapSize - && this.offHeapSize == other.offHeapSize; + MemStoreSize other = (MemStoreSize)obj; + return this.dataSize == other.dataSize && this.heapSize == other.heapSize && + this.offHeapSize == other.offHeapSize; } @Override public int hashCode() { - long h = 13 * this.dataSize; - h = h + 14 * this.heapSize; - h = h + 15 * this.offHeapSize; + long h = 31 * this.dataSize; + h = h + 31 * this.heapSize; + h = h + 31 * this.offHeapSize; return (int) h; } @Override public String toString() { - return "dataSize=" + this.dataSize - + " , heapSize=" + this.heapSize - + " , offHeapSize=" + this.offHeapSize; + return "dataSize=" + this.dataSize + ", heapSize=" + this.heapSize + + ", offHeapSize=" + this.offHeapSize; } -} +} \ No newline at end of file diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSizing.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSizing.java index 0b3e925668..8f93e41c6a 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSizing.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSizing.java @@ -21,61 +21,39 @@ import org.apache.yetus.audience.InterfaceAudience; /** * Accounting of current heap and data sizes. - * Allows read/write on data/heap size as opposed to {@Link MemStoreSize} which is read-only. - * For internal use. + * Like {@link TimeRangeTracker}, it has thread-safe and non-thread-safe implementations. * @see MemStoreSize */ @InterfaceAudience.Private -public class MemStoreSizing extends MemStoreSize { - public static final MemStoreSizing DUD = new MemStoreSizing() { - - @Override public void incMemStoreSize(long dataSizeDelta, long heapSizeDelta, - long offHeapSizeDelta) { - throw new RuntimeException("I'm a dud, you can't use me!"); +public interface MemStoreSizing { + static final MemStoreSizing DUD = new MemStoreSizing() { + @Override + public MemStoreSize getMemStoreSize() { + throw new RuntimeException("I'm a DUD, you can't use me!"); } - @Override public void decMemStoreSize(long dataSizeDelta, long heapSizeDelta, + @Override + public MemStoreSize incMemStoreSize(long dataSizeDelta, long heapSizeDelta, long offHeapSizeDelta) { - throw new RuntimeException("I'm a dud, you can't use me!"); + throw new RuntimeException("I'm a DUD, you can't use me!"); } }; - public MemStoreSizing() { - super(); - } - - public MemStoreSizing(long dataSize, long heapSize, long offHeapSize) { - super(dataSize, heapSize, offHeapSize); - } + MemStoreSize getMemStoreSize(); - public MemStoreSizing(MemStoreSize memStoreSize) { - super(memStoreSize); - } + MemStoreSize incMemStoreSize(long dataSizeDelta, long heapSizeDelta, + long offHeapSizeDelta); - public void incMemStoreSize(long dataSizeDelta, long heapSizeDelta, long offHeapSizeDelta) { - this.dataSize += dataSizeDelta; - this.heapSize += heapSizeDelta; - this.offHeapSize += offHeapSizeDelta; + default MemStoreSize incMemStoreSize(MemStoreSize delta) { + return incMemStoreSize(delta.getDataSize(), delta.getDataSize(), delta.getOffHeapSize()); } - public void incMemStoreSize(MemStoreSize delta) { - incMemStoreSize(delta.getDataSize(), delta.getHeapSize(), delta.getOffHeapSize()); + default MemStoreSize decMemStoreSize(long dataSizeDelta, long heapSizeDelta, + long offHeapSizeDelta) { + return incMemStoreSize(-dataSizeDelta, -heapSizeDelta, -offHeapSizeDelta); } - public void decMemStoreSize(long dataSizeDelta, long heapSizeDelta, long offHeapSizeDelta) { - this.dataSize -= dataSizeDelta; - this.heapSize -= heapSizeDelta; - this.offHeapSize -= offHeapSizeDelta; + default MemStoreSize decMemStoreSize(MemStoreSize delta) { + return decMemStoreSize(delta.getDataSize(), delta.getHeapSize(), delta.getOffHeapSize()); } - - public void decMemStoreSize(MemStoreSize delta) { - decMemStoreSize(delta.getDataSize(), delta.getHeapSize(), delta.getOffHeapSize()); - } - - public void empty() { - this.dataSize = 0L; - this.heapSize = 0L; - this.offHeapSize = 0L; - } - } \ No newline at end of file diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/NonThreadSafeMemStoreSizing.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/NonThreadSafeMemStoreSizing.java new file mode 100644 index 0000000000..285ca8a718 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/NonThreadSafeMemStoreSizing.java @@ -0,0 +1,64 @@ +/* + * 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 org.apache.yetus.audience.InterfaceAudience; + +/** + * Accounting of current heap and data sizes. + * <>. + * Use in a 'local' context only where just a single-thread is updating. No concurrency! + * Used, for example, when summing all Cells in a single batch where result is then applied to the + * Store. + * @see MemStoreSize + * @see ThreadSafeMemStoreSizing + */ +@InterfaceAudience.Private +class NonThreadSafeMemStoreSizing implements MemStoreSizing { + private MemStoreSize memStoreSize; + + NonThreadSafeMemStoreSizing() { + this(new MemStoreSize()); + } + + NonThreadSafeMemStoreSizing(long dataSize, long heapSize, long offHeapSize) { + this(new MemStoreSize(dataSize, heapSize, offHeapSize)); + } + + NonThreadSafeMemStoreSizing(MemStoreSize memStoreSize) { + this.memStoreSize = new MemStoreSize(); + } + + @Override + public MemStoreSize getMemStoreSize() { + return this.memStoreSize; + } + + @Override + public MemStoreSize incMemStoreSize(long dataSizeDelta, long heapSizeDelta, + long offHeapSizeDelta) { + // TODO: Make a version that does not create a MemStoreSize instance each time through. + // Add a incAndGetMemStoreSize that does the below and redo this method to just keep three + // longs internally and which avoids object creation. + MemStoreSize oldMss = getMemStoreSize(); + this.memStoreSize = new MemStoreSize(oldMss.getDataSize() + dataSizeDelta, + oldMss.getHeapSize() + heapSizeDelta, + oldMss.getOffHeapSize() + offHeapSizeDelta); + return this.memStoreSize; + } +} \ No newline at end of file diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerAccounting.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerAccounting.java index 1c627f74b1..0befa7e0ca 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerAccounting.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerAccounting.java @@ -231,7 +231,7 @@ public class RegionServerAccounting { // the region open operation. No need to handle multi thread issues on one region's entry in // this Map. if (replayEdistsSize == null) { - replayEdistsSize = new MemStoreSizing(); + replayEdistsSize = new ThreadSafeMemStoreSizing(); replayEditsPerRegion.put(regionName, replayEdistsSize); } replayEdistsSize.incMemStoreSize(memStoreSize); @@ -244,10 +244,10 @@ public class RegionServerAccounting { * @param regionName the region which could not open. */ public void rollbackRegionReplayEditsSize(byte[] regionName) { - MemStoreSize replayEditsSize = replayEditsPerRegion.get(regionName); - if (replayEditsSize != null) { + MemStoreSizing replayEditsSizing = replayEditsPerRegion.get(regionName); + if (replayEditsSizing != null) { clearRegionReplayEditsSize(regionName); - decGlobalMemStoreSize(replayEditsSize); + decGlobalMemStoreSize(replayEditsSizing.getMemStoreSize()); } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java index 70074bf3b4..8f5ca4259b 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Segment.java @@ -61,7 +61,7 @@ public abstract class Segment { private MemStoreLAB memStoreLAB; // Sum of sizes of all Cells added to this Segment. Cell's heapSize is considered. This is not // including the heap overhead of this class. - protected final MemStoreSizing segmentSize; + protected final MemStoreSizing segmentSizing; protected final TimeRangeTracker timeRangeTracker; protected volatile boolean tagsPresent; @@ -69,7 +69,9 @@ public abstract class Segment { // and there is no need in true Segments state protected Segment(CellComparator comparator, TimeRangeTracker trt) { this.comparator = comparator; - this.segmentSize = new MemStoreSizing(); + // Do we need to be thread safe always? What if ImmutableSegment? + // DITTO for the TimeRangeTracker. + this.segmentSizing = new ThreadSafeMemStoreSizing(); this.timeRangeTracker = trt; } @@ -85,7 +87,9 @@ public abstract class Segment { OffHeapSize += memStoreSize.getOffHeapSize(); } this.comparator = comparator; - this.segmentSize = new MemStoreSizing(dataSize, heapSize, OffHeapSize); + // Do we need to be thread safe always? What if ImmutableSegment? + // DITTO for the TimeRangeTracker. + this.segmentSizing = new ThreadSafeMemStoreSizing(dataSize, heapSize, OffHeapSize); this.timeRangeTracker = trt; } @@ -95,7 +99,9 @@ public abstract class Segment { this.comparator = comparator; this.minSequenceId = Long.MAX_VALUE; this.memStoreLAB = memStoreLAB; - this.segmentSize = new MemStoreSizing(); + // Do we need to be thread safe always? What if ImmutableSegment? + // DITTO for the TimeRangeTracker. + this.segmentSizing = new ThreadSafeMemStoreSizing(); this.tagsPresent = false; this.timeRangeTracker = trt; } @@ -105,7 +111,7 @@ public abstract class Segment { this.comparator = segment.getComparator(); this.minSequenceId = segment.getMinSequenceId(); this.memStoreLAB = segment.getMemStoreLAB(); - this.segmentSize = new MemStoreSizing(segment.getMemStoreSize()); + this.segmentSizing = new ThreadSafeMemStoreSizing(segment.segmentSizing.getMemStoreSize()); this.tagsPresent = segment.isTagsPresent(); this.timeRangeTracker = segment.getTimeRangeTracker(); } @@ -222,28 +228,29 @@ public abstract class Segment { } public MemStoreSize getMemStoreSize() { - return this.segmentSize; + return this.segmentSizing.getMemStoreSize(); } + // TODO: REMOVE THESE THREE METHODS AND USE getMemStoreSize INSTEAD. /** * @return Sum of all cell's size. */ public long keySize() { - return this.segmentSize.getDataSize(); + return this.segmentSizing.getMemStoreSize().getDataSize(); } /** * @return The heap size of this segment. */ public long heapSize() { - return this.segmentSize.getHeapSize(); + return this.segmentSizing.getMemStoreSize().getHeapSize(); } /** * @return The off-heap size of this segment. */ public long offHeapSize() { - return this.segmentSize.getOffHeapSize(); + return this.segmentSizing.getMemStoreSize().getOffHeapSize(); } /** @@ -251,9 +258,7 @@ public abstract class Segment { */ //TODO protected void incSize(long delta, long heapOverhead, long offHeapOverhead) { - synchronized (this) { - this.segmentSize.incMemStoreSize(delta, heapOverhead, offHeapOverhead); - } + this.segmentSizing.incMemStoreSize(delta, heapOverhead, offHeapOverhead); } public long getMinSequenceId() { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ThreadSafeMemStoreSizing.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ThreadSafeMemStoreSizing.java new file mode 100644 index 0000000000..cdd71b95e2 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ThreadSafeMemStoreSizing.java @@ -0,0 +1,62 @@ +/* + * 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 org.apache.yetus.audience.InterfaceAudience; + +import java.util.concurrent.atomic.AtomicReference; + +/** + * Accounting of current heap and data sizes. + * Thread-safe. Many threads can do updates against this single instance. + * @see NonThreadSafeMemStoreSizing + * @see MemStoreSize + */ +@InterfaceAudience.Private +class ThreadSafeMemStoreSizing implements MemStoreSizing { + private final AtomicReference memStoreSize; + + ThreadSafeMemStoreSizing() { + this(new MemStoreSize()); + } + + ThreadSafeMemStoreSizing(long dataSize, long heapSize, long offHeapSize) { + this(new MemStoreSize(dataSize, heapSize, offHeapSize)); + } + + ThreadSafeMemStoreSizing(MemStoreSize memStoreSize) { + this.memStoreSize = new AtomicReference(memStoreSize); + } + + public MemStoreSize getMemStoreSize() { + return this.memStoreSize.get(); + } + + public MemStoreSize incMemStoreSize(long dataSizeDelta, long heapSizeDelta, + long offHeapSizeDelta) { + while (true) { + MemStoreSize oldMss = getMemStoreSize(); + MemStoreSize newMss = new MemStoreSize(oldMss.getDataSize() + dataSizeDelta, + oldMss.getHeapSize() + heapSizeDelta, + oldMss.getOffHeapSize() + offHeapSizeDelta); + if (this.memStoreSize.compareAndSet(oldMss, newMss)) { + return newMss; + } + } + } +} \ No newline at end of file diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java index 5cbfff9644..d8a80a3d40 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java @@ -838,7 +838,8 @@ public class TestCompactingMemStore extends TestDefaultMemStore { hmc.add(kv, null); LOG.debug("added kv: " + kv.getKeyString() + ", timestamp:" + kv.getTimestamp()); } - regionServicesForStores.addMemStoreSize(new MemStoreSize(hmc.getActive().keySize() - size, + regionServicesForStores.addMemStoreSize(new MemStoreSize( + hmc.getActive().keySize() - size, hmc.getActive().heapSize() - heapOverhead, 0)); return totalLen; } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellFlatMapMemStore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellFlatMapMemStore.java index 9523091af6..16b7ca402f 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellFlatMapMemStore.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellFlatMapMemStore.java @@ -895,7 +895,7 @@ public class TestCompactingToCellFlatMapMemStore extends TestCompactingMemStore private long addRowsByKeysDataSize(final AbstractMemStore hmc, String[] keys) { byte[] fam = Bytes.toBytes("testfamily"); byte[] qf = Bytes.toBytes("testqualifier"); - MemStoreSizing memstoreSizing = new MemStoreSizing(); + MemStoreSizing memstoreSizing = new NonThreadSafeMemStoreSizing(); for (int i = 0; i < keys.length; i++) { long timestamp = System.currentTimeMillis(); Threads.sleep(1); // to make sure each kv gets a different ts @@ -905,8 +905,9 @@ public class TestCompactingToCellFlatMapMemStore extends TestCompactingMemStore hmc.add(kv, memstoreSizing); LOG.debug("added kv: " + kv.getKeyString() + ", timestamp" + kv.getTimestamp()); } - regionServicesForStores.addMemStoreSize(memstoreSizing); - return memstoreSizing.getDataSize(); + MemStoreSize mss = memstoreSizing.getMemStoreSize(); + regionServicesForStores.addMemStoreSize(mss); + return mss.getDataSize(); } private long cellBeforeFlushSize() { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultMemStore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultMemStore.java index 6da5ec0a74..77f796f8c3 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultMemStore.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultMemStore.java @@ -129,17 +129,18 @@ public class TestDefaultMemStore { public void testPutSameCell() { byte[] bytes = Bytes.toBytes(getName()); KeyValue kv = new KeyValue(bytes, bytes, bytes, bytes); - MemStoreSizing sizeChangeForFirstCell = new MemStoreSizing(); + MemStoreSizing sizeChangeForFirstCell = new NonThreadSafeMemStoreSizing(); this.memstore.add(kv, sizeChangeForFirstCell); - MemStoreSizing sizeChangeForSecondCell = new MemStoreSizing(); + MemStoreSizing sizeChangeForSecondCell = new NonThreadSafeMemStoreSizing(); this.memstore.add(kv, sizeChangeForSecondCell); // make sure memstore size increase won't double-count MSLAB chunk size - assertEquals(Segment.getCellLength(kv), sizeChangeForFirstCell.getDataSize()); + assertEquals(Segment.getCellLength(kv), sizeChangeForFirstCell.getMemStoreSize().getDataSize()); Segment segment = this.memstore.getActive(); MemStoreLAB msLab = segment.getMemStoreLAB(); if (msLab != null) { // make sure memstore size increased even when writing the same cell, if using MSLAB - assertEquals(Segment.getCellLength(kv), sizeChangeForSecondCell.getDataSize()); + assertEquals(Segment.getCellLength(kv), + sizeChangeForSecondCell.getMemStoreSize().getDataSize()); // make sure chunk size increased even when writing the same cell, if using MSLAB if (msLab instanceof MemStoreLABImpl) { // since we add the chunkID at the 0th offset of the chunk and the @@ -149,8 +150,8 @@ public class TestDefaultMemStore { } } else { // make sure no memstore size change w/o MSLAB - assertEquals(0, sizeChangeForSecondCell.getDataSize()); - assertEquals(0, sizeChangeForSecondCell.getHeapSize()); + assertEquals(0, sizeChangeForSecondCell.getMemStoreSize().getDataSize()); + assertEquals(0, sizeChangeForSecondCell.getMemStoreSize().getHeapSize()); } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java index b310f58675..fdba6e4c0a 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java @@ -262,7 +262,7 @@ public class TestHStore { MemStoreSize size = store.memstore.getFlushableSize(); assertEquals(0, size.getDataSize()); LOG.info("Adding some data"); - MemStoreSizing kvSize = new MemStoreSizing(); + MemStoreSizing kvSize = new NonThreadSafeMemStoreSizing(); store.add(new KeyValue(row, family, qf1, 1, (byte[]) null), kvSize); // add the heap size of active (mutable) segment kvSize.incMemStoreSize(0, MutableSegment.DEEP_OVERHEAD, 0); @@ -281,7 +281,7 @@ public class TestHStore { CSLMImmutableSegment.DEEP_OVERHEAD_CSLM-MutableSegment.DEEP_OVERHEAD, 0); size = store.memstore.getFlushableSize(); assertEquals(kvSize, size); - MemStoreSizing kvSize2 = new MemStoreSizing(); + MemStoreSizing kvSize2 = new NonThreadSafeMemStoreSizing(); store.add(new KeyValue(row, family, qf2, 2, (byte[])null), kvSize2); kvSize2.incMemStoreSize(0, MutableSegment.DEEP_OVERHEAD, 0); // Even though we add a new kv, we expect the flushable size to be 'same' since we have @@ -1225,7 +1225,7 @@ public class TestHStore { byte[] value0 = Bytes.toBytes("value0"); byte[] value1 = Bytes.toBytes("value1"); byte[] value2 = Bytes.toBytes("value2"); - MemStoreSizing memStoreSizing = new MemStoreSizing(); + MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing(); long ts = EnvironmentEdgeManager.currentTime(); long seqId = 100; init(name.getMethodName(), conf, TableDescriptorBuilder.newBuilder(TableName.valueOf(table)), @@ -1284,7 +1284,7 @@ public class TestHStore { init(name.getMethodName(), conf, ColumnFamilyDescriptorBuilder.newBuilder(family) .setInMemoryCompaction(MemoryCompactionPolicy.BASIC).build()); byte[] value = Bytes.toBytes("value"); - MemStoreSizing memStoreSizing = new MemStoreSizing(); + MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing(); long ts = EnvironmentEdgeManager.currentTime(); long seqId = 100; // older data whihc shouldn't be "seen" by client @@ -1362,7 +1362,7 @@ public class TestHStore { }); byte[] oldValue = Bytes.toBytes("oldValue"); byte[] currentValue = Bytes.toBytes("currentValue"); - MemStoreSizing memStoreSizing = new MemStoreSizing(); + MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing(); long ts = EnvironmentEdgeManager.currentTime(); long seqId = 100; // older data whihc shouldn't be "seen" by client @@ -1478,7 +1478,7 @@ public class TestHStore { init(name.getMethodName(), conf, ColumnFamilyDescriptorBuilder.newBuilder(family) .setInMemoryCompaction(MemoryCompactionPolicy.BASIC).build()); byte[] value = Bytes.toBytes("thisisavarylargevalue"); - MemStoreSizing memStoreSizing = new MemStoreSizing(); + MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing(); long ts = EnvironmentEdgeManager.currentTime(); long seqId = 100; // older data whihc shouldn't be "seen" by client @@ -1600,7 +1600,7 @@ public class TestHStore { conf.setLong(StoreScanner.STORESCANNER_PREAD_MAX_BYTES, 0); // Set the lower threshold to invoke the "MERGE" policy MyStore store = initMyStore(name.getMethodName(), conf, new MyStoreHook() {}); - MemStoreSizing memStoreSizing = new MemStoreSizing(); + MemStoreSizing memStoreSizing = new NonThreadSafeMemStoreSizing(); long ts = System.currentTimeMillis(); long seqID = 1L; // Add some data to the region and do some flushes -- 2.16.3