Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-19506: The CellChunkMap index chunks are usually small, so in order to prevent memory underutilization, HBASE-19506 presents small chunks preallocated in a small pool

This commit is contained in:
anastas 2018-02-21 10:19:49 +02:00
parent bc1ac49de2
commit 2b4df5e36e
11 changed files with 376 additions and 186 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CellChunkImmutableSegment.java
View File

@ -32,7 +32,6 @@ import org.apache.hadoop.hbase.util.ClassSize;
import org.apache.yetus.audience.InterfaceAudience;
/**
* CellChunkImmutableSegment extends the API supported by a {@link Segment},
* and {@link ImmutableSegment}. This immutable segment is working with CellSet with
@ -43,6 +42,7 @@ public class CellChunkImmutableSegment extends ImmutableSegment {
public static final long DEEP_OVERHEAD_CCM =
ImmutableSegment.DEEP_OVERHEAD + ClassSize.CELL_CHUNK_MAP;
public static final float INDEX_CHUNK_UNUSED_SPACE_PRECENTAGE = 0.1f;
///////////////////// CONSTRUCTORS /////////////////////
/**------------------------------------------------------------------------
@ -135,20 +135,12 @@ public class CellChunkImmutableSegment extends ImmutableSegment {
private void initializeCellSet(int numOfCells, MemStoreSegmentsIterator iterator,
MemStoreCompactionStrategy.Action action) {
// calculate how many chunks we will need for index
int chunkSize = ChunkCreator.getInstance().getChunkSize();
int numOfCellsInChunk = CellChunkMap.NUM_OF_CELL_REPS_IN_CHUNK;
int numberOfChunks = calculateNumberOfChunks(numOfCells, numOfCellsInChunk);
int numOfCellsAfterCompaction = 0;
int currentChunkIdx = 0;
int offsetInCurentChunk = ChunkCreator.SIZEOF_CHUNK_HEADER;
int numUniqueKeys=0;
Cell prev = null;
// all index Chunks are allocated from ChunkCreator
Chunk[] chunks = new Chunk[numberOfChunks];
for (int i=0; i < numberOfChunks; i++) {
chunks[i] = this.getMemStoreLAB().getNewExternalChunk();
}
Chunk[] chunks = allocIndexChunks(numOfCells);
while (iterator.hasNext()) { // the iterator hides the elimination logic for compaction
boolean alreadyCopied = false;
Cell c = iterator.next();
@ -161,7 +153,7 @@ public class CellChunkImmutableSegment extends ImmutableSegment {
c = copyCellIntoMSLAB(c);
alreadyCopied = true;
}
if (offsetInCurentChunk + ClassSize.CELL_CHUNK_MAP_ENTRY > chunkSize) {
if (offsetInCurentChunk + ClassSize.CELL_CHUNK_MAP_ENTRY > chunks[currentChunkIdx].size) {
currentChunkIdx++; // continue to the next index chunk
offsetInCurentChunk = ChunkCreator.SIZEOF_CHUNK_HEADER;
}
@ -207,15 +199,7 @@ public class CellChunkImmutableSegment extends ImmutableSegment {
int numOfCells, KeyValueScanner segmentScanner, CellSet oldCellSet,
MemStoreCompactionStrategy.Action action) {
Cell curCell;
// calculate how many chunks we will need for metadata
int chunkSize = ChunkCreator.getInstance().getChunkSize();
int numOfCellsInChunk = CellChunkMap.NUM_OF_CELL_REPS_IN_CHUNK;
int numberOfChunks = calculateNumberOfChunks(numOfCells, numOfCellsInChunk);
// all index Chunks are allocated from ChunkCreator
Chunk[] chunks = new Chunk[numberOfChunks];
for (int i=0; i < numberOfChunks; i++) {
chunks[i] = this.getMemStoreLAB().getNewExternalChunk();
}
Chunk[] chunks = allocIndexChunks(numOfCells);
int currentChunkIdx = 0;
int offsetInCurentChunk = ChunkCreator.SIZEOF_CHUNK_HEADER;
@ -231,7 +215,7 @@ public class CellChunkImmutableSegment extends ImmutableSegment {
// are copied into MSLAB here.
curCell = copyCellIntoMSLAB(curCell);
}
if (offsetInCurentChunk + ClassSize.CELL_CHUNK_MAP_ENTRY > chunkSize) {
if (offsetInCurentChunk + ClassSize.CELL_CHUNK_MAP_ENTRY > chunks[currentChunkIdx].size) {
// continue to the next metadata chunk
currentChunkIdx++;
offsetInCurentChunk = ChunkCreator.SIZEOF_CHUNK_HEADER;
@ -279,14 +263,58 @@ public class CellChunkImmutableSegment extends ImmutableSegment {
return offset;
}
private int calculateNumberOfChunks(int numOfCells, int numOfCellsInChunk) {
int numberOfChunks = numOfCells/numOfCellsInChunk;
if(numOfCells%numOfCellsInChunk!=0) { // if cells cannot be divided evenly between chunks
private int calculateNumberOfChunks(int numOfCells, int chunkSize) {
int numOfCellsInChunk = calcNumOfCellsInChunk(chunkSize);
int numberOfChunks = numOfCells / numOfCellsInChunk;
if(numOfCells % numOfCellsInChunk != 0) { // if cells cannot be divided evenly between chunks
numberOfChunks++; // add one additional chunk
}
return numberOfChunks;
}
// Assuming we are going to use regular data chunks as index chunks,
// we check here how much free space will remain in the last allocated chunk
// (the least occupied one).
// If the percentage of its remaining free space is above the INDEX_CHUNK_UNUSED_SPACE
// threshold, then we will use index chunks (which are smaller) instead.
private ChunkCreator.ChunkType useIndexChunks(int numOfCells) {
int dataChunkSize = ChunkCreator.getInstance().getChunkSize();
int numOfCellsInChunk = calcNumOfCellsInChunk(dataChunkSize);
int cellsInLastChunk = numOfCells % numOfCellsInChunk;
if (cellsInLastChunk == 0) { // There is no free space in the last chunk and thus,
return ChunkCreator.ChunkType.DATA_CHUNK; // no need to use index chunks.
} else {
int chunkSpace = dataChunkSize - ChunkCreator.SIZEOF_CHUNK_HEADER;
int freeSpaceInLastChunk = chunkSpace - cellsInLastChunk * ClassSize.CELL_CHUNK_MAP_ENTRY;
if (freeSpaceInLastChunk > INDEX_CHUNK_UNUSED_SPACE_PRECENTAGE * chunkSpace) {
return ChunkCreator.ChunkType.INDEX_CHUNK;
}
return ChunkCreator.ChunkType.DATA_CHUNK;
}
}
private int calcNumOfCellsInChunk(int chunkSize) {
int chunkSpace = chunkSize - ChunkCreator.SIZEOF_CHUNK_HEADER;
int numOfCellsInChunk = chunkSpace / ClassSize.CELL_CHUNK_MAP_ENTRY;
return numOfCellsInChunk;
}
private Chunk[] allocIndexChunks(int numOfCells) {
// Decide whether to use regular or small chunks and then
// calculate how many chunks we will need for index
ChunkCreator.ChunkType chunkType = useIndexChunks(numOfCells);
int chunkSize = ChunkCreator.getInstance().getChunkSize(chunkType);
int numberOfChunks = calculateNumberOfChunks(numOfCells, chunkSize);
// all index Chunks are allocated from ChunkCreator
Chunk[] chunks = new Chunk[numberOfChunks];
// all index Chunks are allocated from ChunkCreator
for (int i = 0; i < numberOfChunks; i++) {
chunks[i] = this.getMemStoreLAB().getNewExternalChunk(chunkType);
}
return chunks;
}
private Cell copyCellIntoMSLAB(Cell cell) {
// Take care for a special case when a cell is copied from on-heap to (probably off-heap) MSLAB.
// The cell allocated as an on-heap JVM object (byte array) occupies slightly different

17
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CellChunkMap.java
View File

@ -58,11 +58,10 @@ public class CellChunkMap extends CellFlatMap {
private final Chunk[] chunks; // the array of chunks, on which the index is based
// constant number of cell-representations in a chunk
// number of cell-representations in a chunk
// depends on the size of the chunks (may be index chunks or regular data chunks)
// each chunk starts with its own ID following the cells data
public static final int NUM_OF_CELL_REPS_IN_CHUNK =
(ChunkCreator.getInstance().getChunkSize() - ChunkCreator.SIZEOF_CHUNK_HEADER) /
ClassSize.CELL_CHUNK_MAP_ENTRY;
private final int numOfCellRepsInChunk;
/**
* C-tor for creating CellChunkMap from existing Chunk array, which must be ordered
@ -77,6 +76,12 @@ public class CellChunkMap extends CellFlatMap {
Chunk[] chunks, int min, int max, boolean descending) {
super(comparator, min, max, descending);
this.chunks = chunks;
if (chunks != null && chunks.length != 0 && chunks[0] != null) {
this.numOfCellRepsInChunk = (chunks[0].size - ChunkCreator.SIZEOF_CHUNK_HEADER) /
ClassSize.CELL_CHUNK_MAP_ENTRY;
} else { // In case the chunks array was not allocated
this.numOfCellRepsInChunk = 0;
}
}
/* To be used by base (CellFlatMap) class only to create a sub-CellFlatMap
@ -90,9 +95,9 @@ public class CellChunkMap extends CellFlatMap {
@Override
protected Cell getCell(int i) {
// get the index of the relevant chunk inside chunk array
int chunkIndex = (i / NUM_OF_CELL_REPS_IN_CHUNK);
int chunkIndex = (i / numOfCellRepsInChunk);
ByteBuffer block = chunks[chunkIndex].getData();// get the ByteBuffer of the relevant chunk
int j = i - chunkIndex * NUM_OF_CELL_REPS_IN_CHUNK; // get the index of the cell-representation
int j = i - chunkIndex * numOfCellRepsInChunk; // get the index of the cell-representation
// find inside the offset inside the chunk holding the index, skip bytes for chunk id
int offsetInBytes = ChunkCreator.SIZEOF_CHUNK_HEADER + j* ClassSize.CELL_CHUNK_MAP_ENTRY;

4
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Chunk.java
View File

@ -89,6 +89,10 @@ public abstract class Chunk {
return size > ChunkCreator.getInstance().getChunkSize();
}
boolean isIndexChunk() {
return size == ChunkCreator.getInstance().getChunkSize(ChunkCreator.ChunkType.INDEX_CHUNK);
}
/**
* Actually claim the memory for this chunk. This should only be called from the thread that
* constructed the chunk. It is thread-safe against other threads calling alloc(), who will block

324
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ChunkCreator.java
View File

@ -58,27 +58,58 @@ public class ChunkCreator {
// the header size need to be changed in case chunk id size is changed
public static final int SIZEOF_CHUNK_HEADER = Bytes.SIZEOF_INT;
/**
* Types of chunks, based on their sizes
*/
public enum ChunkType {
// An index chunk is a small chunk, allocated from the index chunks pool.
// Its size is fixed and is 10% of the size of a data chunk.
INDEX_CHUNK,
// A data chunk is a regular chunk, allocated from the data chunks pool.
// Its size is fixed and given as input to the ChunkCreator c'tor.
DATA_CHUNK,
// A jumbo chunk isn't allocated from pool. Its size is bigger than the size of a
// data chunk, and is determined per chunk (meaning, there is no fixed jumbo size).
JUMBO_CHUNK
}
// mapping from chunk IDs to chunks
private Map<Integer, Chunk> chunkIdMap = new ConcurrentHashMap<Integer, Chunk>();
private final int chunkSize;
private final boolean offheap;
@VisibleForTesting
static ChunkCreator INSTANCE;
static ChunkCreator instance;
@VisibleForTesting
static boolean chunkPoolDisabled = false;
private MemStoreChunkPool pool;
private MemStoreChunkPool dataChunksPool;
private int chunkSize;
private MemStoreChunkPool indexChunksPool;
@VisibleForTesting
ChunkCreator(int chunkSize, boolean offheap, long globalMemStoreSize, float poolSizePercentage,
float initialCountPercentage, HeapMemoryManager heapMemoryManager) {
this.chunkSize = chunkSize;
float initialCountPercentage, HeapMemoryManager heapMemoryManager,
float indexChunkSizePercentage) {
this.offheap = offheap;
this.pool = initializePool(globalMemStoreSize, poolSizePercentage, initialCountPercentage);
if (heapMemoryManager != null && this.pool != null) {
// Register with Heap Memory manager
heapMemoryManager.registerTuneObserver(this.pool);
}
this.chunkSize = chunkSize; // in case pools are not allocated
initializePools(chunkSize, globalMemStoreSize, poolSizePercentage, indexChunkSizePercentage,
initialCountPercentage, heapMemoryManager);
}
@VisibleForTesting
private void initializePools(int chunkSize, long globalMemStoreSize,
float poolSizePercentage, float indexChunkSizePercentage,
float initialCountPercentage,
HeapMemoryManager heapMemoryManager) {
this.dataChunksPool = initializePool(globalMemStoreSize,
(1 - indexChunkSizePercentage) * poolSizePercentage,
initialCountPercentage, chunkSize, heapMemoryManager);
// The index chunks pool is needed only when the index type is CCM.
// Since the pools are not created at all when the index type isn't CCM,
// we don't need to check it here.
this.indexChunksPool = initializePool(globalMemStoreSize,
indexChunkSizePercentage * poolSizePercentage,
initialCountPercentage, (int) (indexChunkSizePercentage * chunkSize),
heapMemoryManager);
}
/**
@ -92,18 +123,30 @@ public class ChunkCreator {
* @return singleton MSLABChunkCreator
*/
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "LI_LAZY_INIT_STATIC",
justification = "Method is called by single thread at the starting of RS")
justification = "Method is called by single thread at the starting of RS")
@VisibleForTesting
public static ChunkCreator initialize(int chunkSize, boolean offheap, long globalMemStoreSize,
float poolSizePercentage, float initialCountPercentage, HeapMemoryManager heapMemoryManager) {
if (INSTANCE != null) return INSTANCE;
INSTANCE = new ChunkCreator(chunkSize, offheap, globalMemStoreSize, poolSizePercentage,
initialCountPercentage, heapMemoryManager);
return INSTANCE;
float poolSizePercentage, float initialCountPercentage,
HeapMemoryManager heapMemoryManager) {
if (instance != null) {
return instance;
}
instance = new ChunkCreator(chunkSize, offheap, globalMemStoreSize, poolSizePercentage,
initialCountPercentage, heapMemoryManager,
MemStoreLABImpl.INDEX_CHUNK_PERCENTAGE_DEFAULT);
return instance;
}
static ChunkCreator getInstance() {
return INSTANCE;
return instance;
}
/**
* Creates and inits a chunk. The default implementation for a specific chunk size.
* @return the chunk that was initialized
*/
Chunk getChunk(ChunkType chunkType) {
return getChunk(CompactingMemStore.IndexType.ARRAY_MAP, chunkType);
}
/**
@ -111,15 +154,37 @@ public class ChunkCreator {
* @return the chunk that was initialized
*/
Chunk getChunk() {
return getChunk(CompactingMemStore.IndexType.ARRAY_MAP, chunkSize);
return getChunk(CompactingMemStore.IndexType.ARRAY_MAP, ChunkType.DATA_CHUNK);
}
/**
* Creates and inits a chunk. The default implementation for specific index type.
* Creates and inits a chunk. The default implementation for a specific index type.
* @return the chunk that was initialized
*/
Chunk getChunk(CompactingMemStore.IndexType chunkIndexType) {
return getChunk(chunkIndexType, chunkSize);
return getChunk(chunkIndexType, ChunkType.DATA_CHUNK);
}
/**
* Creates and inits a chunk with specific index type and type.
* @return the chunk that was initialized
*/
Chunk getChunk(CompactingMemStore.IndexType chunkIndexType, ChunkType chunkType) {
switch (chunkType) {
case INDEX_CHUNK:
if (indexChunksPool != null) {
return getChunk(chunkIndexType, indexChunksPool.getChunkSize());
}
case DATA_CHUNK:
if (dataChunksPool == null) {
return getChunk(chunkIndexType, chunkSize);
} else {
return getChunk(chunkIndexType, dataChunksPool.getChunkSize());
}
default:
throw new IllegalArgumentException(
"chunkType must either be INDEX_CHUNK or DATA_CHUNK");
}
}
/**
@ -130,18 +195,28 @@ public class ChunkCreator {
*/
Chunk getChunk(CompactingMemStore.IndexType chunkIndexType, int size) {
Chunk chunk = null;
// if we have pool and this is not jumbo chunk (when size != chunkSize this is jumbo chunk)
if ((pool != null) && (size == chunkSize)) {
MemStoreChunkPool pool = null;
// if the size is suitable for one of the pools
if (dataChunksPool != null && size == dataChunksPool.getChunkSize()) {
pool = dataChunksPool;
} else if (indexChunksPool != null && size == indexChunksPool.getChunkSize()) {
pool = indexChunksPool;
}
// if we have a pool
if (pool != null) {
// the pool creates the chunk internally. The chunk#init() call happens here
chunk = this.pool.getChunk();
chunk = pool.getChunk();
// the pool has run out of maxCount
if (chunk == null) {
if (LOG.isTraceEnabled()) {
LOG.trace("Chunk pool full (maxCount={}); creating chunk offheap.",
this.pool.getMaxCount());
LOG.trace("The chunk pool is full. Reached maxCount= " + pool.getMaxCount()
+ ". Creating chunk onheap.");
}
}
}
if (chunk == null) {
// the second parameter explains whether CellChunkMap index is requested,
// in that case, put allocated on demand chunk mapping into chunkIdMap
@ -158,18 +233,18 @@ public class ChunkCreator {
* Creates and inits a chunk of a special size, bigger than a regular chunk size.
* Such a chunk will never come from pool and will always be on demand allocated.
* @return the chunk that was initialized
* @param chunkIndexType whether the requested chunk is going to be used with CellChunkMap index
* @param jumboSize the special size to be used
*/
Chunk getJumboChunk(CompactingMemStore.IndexType chunkIndexType, int jumboSize) {
if (jumboSize <= chunkSize) {
LOG.warn("Jumbo chunk size=" + jumboSize + " must be more than regular chunk size="
+ chunkSize + "; converting to regular chunk.");
return getChunk(chunkIndexType,chunkSize);
Chunk getJumboChunk(int jumboSize) {
int allocSize = jumboSize + SIZEOF_CHUNK_HEADER;
if (allocSize <= dataChunksPool.getChunkSize()) {
LOG.warn("Jumbo chunk size " + jumboSize + " must be more than regular chunk size "
+ dataChunksPool.getChunkSize() + ". Converting to regular chunk.");
return getChunk(CompactingMemStore.IndexType.CHUNK_MAP);
}
// the size of the allocation includes
// both the size requested and a place for the Chunk's header
return getChunk(chunkIndexType, jumboSize + SIZEOF_CHUNK_HEADER);
// the new chunk is going to hold the jumbo cell data and needs to be referenced by
// a strong map. Therefore the CCM index type
return getChunk(CompactingMemStore.IndexType.CHUNK_MAP, allocSize);
}
/**
@ -198,21 +273,21 @@ public class ChunkCreator {
// Chunks from pool are created covered with strong references anyway
// TODO: change to CHUNK_MAP if it is generally defined
private Chunk createChunkForPool() {
return createChunk(true, CompactingMemStore.IndexType.ARRAY_MAP, chunkSize);
private Chunk createChunkForPool(CompactingMemStore.IndexType chunkIndexType, int chunkSize) {
if (chunkSize != dataChunksPool.getChunkSize() &&
chunkSize != indexChunksPool.getChunkSize()) {
return null;
}
return createChunk(true, chunkIndexType, chunkSize);
}
@VisibleForTesting
// Used to translate the ChunkID into a chunk ref
// Used to translate the ChunkID into a chunk ref
Chunk getChunk(int id) {
// can return null if chunk was never mapped
return chunkIdMap.get(id);
}
int getChunkSize() {
return this.chunkSize;
}
boolean isOffheap() {
return this.offheap;
}
@ -226,8 +301,8 @@ public class ChunkCreator {
}
@VisibleForTesting
// the chunks in the chunkIdMap may already be released so we shouldn't relay
// on this counting for strong correctness. This method is used only in testing.
// the chunks in the chunkIdMap may already be released so we shouldn't relay
// on this counting for strong correctness. This method is used only in testing.
int numberOfMappedChunks() {
return this.chunkIdMap.size();
}
@ -245,6 +320,7 @@ public class ChunkCreator {
* collection on JVM.
*/
private class MemStoreChunkPool implements HeapMemoryTuneObserver {
private final int chunkSize;
private int maxCount;
// A queue of reclaimed chunks
@ -258,21 +334,22 @@ public class ChunkCreator {
private final AtomicLong chunkCount = new AtomicLong();
private final LongAdder reusedChunkCount = new LongAdder();
MemStoreChunkPool(int maxCount, int initialCount, float poolSizePercentage) {
MemStoreChunkPool(int chunkSize, int maxCount, int initialCount, float poolSizePercentage) {
this.chunkSize = chunkSize;
this.maxCount = maxCount;
this.poolSizePercentage = poolSizePercentage;
this.reclaimedChunks = new LinkedBlockingQueue<>();
for (int i = 0; i < initialCount; i++) {
Chunk chunk = createChunkForPool();
Chunk chunk = createChunk(true, CompactingMemStore.IndexType.ARRAY_MAP, chunkSize);
chunk.init();
reclaimedChunks.add(chunk);
}
chunkCount.set(initialCount);
final String n = Thread.currentThread().getName();
scheduleThreadPool = Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
.setNameFormat(n + "-MemStoreChunkPool Statistics").setDaemon(true).build());
.setNameFormat(n + "-MemStoreChunkPool Statistics").setDaemon(true).build());
this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(), statThreadPeriod,
statThreadPeriod, TimeUnit.SECONDS);
statThreadPeriod, TimeUnit.SECONDS);
}
/**
@ -285,6 +362,10 @@ public class ChunkCreator {
* @see #putbackChunks(Chunk)
*/
Chunk getChunk() {
return getChunk(CompactingMemStore.IndexType.ARRAY_MAP);
}
Chunk getChunk(CompactingMemStore.IndexType chunkIndexType) {
Chunk chunk = reclaimedChunks.poll();
if (chunk != null) {
chunk.reset();
@ -295,7 +376,7 @@ public class ChunkCreator {
long created = this.chunkCount.get();
if (created < this.maxCount) {
if (this.chunkCount.compareAndSet(created, created + 1)) {
chunk = createChunkForPool();
chunk = createChunkForPool(chunkIndexType, chunkSize);
break;
}
} else {
@ -306,6 +387,10 @@ public class ChunkCreator {
return chunk;
}
int getChunkSize() {
return chunkSize;
}
/**
* Add the chunks to the pool, when the pool achieves the max size, it will skip the remaining
* chunks
@ -313,7 +398,7 @@ public class ChunkCreator {
*/
private void putbackChunks(Chunk c) {
int toAdd = this.maxCount - reclaimedChunks.size();
if (c.isFromPool() && toAdd > 0) {
if (c.isFromPool() && c.size == chunkSize && toAdd > 0) {
reclaimedChunks.add(c);
} else {
// remove the chunk (that is not going to pool)
@ -338,10 +423,11 @@ public class ChunkCreator {
long created = chunkCount.get();
long reused = reusedChunkCount.sum();
long total = created + reused;
LOG.debug("Stats: current pool size=" + reclaimedChunks.size()
+ ",created chunk count=" + created
+ ",reused chunk count=" + reused
+ ",reuseRatio=" + (total == 0 ? "0" : StringUtils.formatPercent(
LOG.debug("Stats (chunk size=" + chunkSize + "): "
+ "current pool size=" + reclaimedChunks.size()
+ ",created chunk count=" + created
+ ",reused chunk count=" + reused
+ ",reuseRatio=" + (total == 0 ? "0" : StringUtils.formatPercent(
(float) reused / (float) total, 2)));
}
}
@ -358,7 +444,7 @@ public class ChunkCreator {
return;
}
int newMaxCount =
(int) (newMemstoreSize * poolSizePercentage / getChunkSize());
(int) (newMemstoreSize * poolSizePercentage / getChunkSize());
if (newMaxCount != this.maxCount) {
// We need an adjustment in the chunks numbers
if (newMaxCount > this.maxCount) {
@ -389,7 +475,8 @@ public class ChunkCreator {
}
private MemStoreChunkPool initializePool(long globalMemStoreSize, float poolSizePercentage,
float initialCountPercentage) {
float initialCountPercentage, int chunkSize,
HeapMemoryManager heapMemoryManager) {
if (poolSizePercentage <= 0) {
LOG.info("PoolSizePercentage is less than 0. So not using pool");
return null;
@ -399,47 +486,90 @@ public class ChunkCreator {
}
if (poolSizePercentage > 1.0) {
throw new IllegalArgumentException(
MemStoreLAB.CHUNK_POOL_MAXSIZE_KEY + " must be between 0.0 and 1.0");
MemStoreLAB.CHUNK_POOL_MAXSIZE_KEY + " must be between 0.0 and 1.0");
}
int maxCount = (int) (globalMemStoreSize * poolSizePercentage / getChunkSize());
int maxCount = (int) (globalMemStoreSize * poolSizePercentage / chunkSize);
if (initialCountPercentage > 1.0 || initialCountPercentage < 0) {
throw new IllegalArgumentException(
MemStoreLAB.CHUNK_POOL_INITIALSIZE_KEY + " must be between 0.0 and 1.0");
MemStoreLAB.CHUNK_POOL_INITIALSIZE_KEY + " must be between 0.0 and 1.0");
}
int initialCount = (int) (initialCountPercentage * maxCount);
LOG.info("Allocating MemStoreChunkPool with chunk size="
+ StringUtils.byteDesc(getChunkSize()) + ", max count=" + maxCount
+ ", initial count=" + initialCount);
return new MemStoreChunkPool(maxCount, initialCount, poolSizePercentage);
LOG.info("Allocating MemStoreChunkPool with chunk size "
+ StringUtils.byteDesc(chunkSize) + ", max count " + maxCount
+ ", initial count " + initialCount);
MemStoreChunkPool memStoreChunkPool = new MemStoreChunkPool(chunkSize, maxCount,
initialCount, poolSizePercentage);
if (heapMemoryManager != null && memStoreChunkPool != null) {
// Register with Heap Memory manager
heapMemoryManager.registerTuneObserver(memStoreChunkPool);
}
return memStoreChunkPool;
}
@VisibleForTesting
int getMaxCount() {
if (pool != null) {
return pool.getMaxCount();
return getMaxCount(ChunkType.DATA_CHUNK);
}
@VisibleForTesting
int getMaxCount(ChunkType chunkType) {
switch (chunkType) {
case INDEX_CHUNK:
if (indexChunksPool != null) {
return indexChunksPool.getMaxCount();
}
break;
case DATA_CHUNK:
if (dataChunksPool != null) {
return dataChunksPool.getMaxCount();
}
break;
default:
throw new IllegalArgumentException(
"chunkType must either be INDEX_CHUNK or DATA_CHUNK");
}
return 0;
}
@VisibleForTesting
int getPoolSize() {
if (pool != null) {
return pool.reclaimedChunks.size();
return getPoolSize(ChunkType.DATA_CHUNK);
}
@VisibleForTesting
int getPoolSize(ChunkType chunkType) {
switch (chunkType) {
case INDEX_CHUNK:
if (indexChunksPool != null) {
return indexChunksPool.reclaimedChunks.size();
}
break;
case DATA_CHUNK:
if (dataChunksPool != null) {
return dataChunksPool.reclaimedChunks.size();
}
break;
default:
throw new IllegalArgumentException(
"chunkType must either be INDEX_CHUNK or DATA_CHUNK");
}
return 0;
}
@VisibleForTesting
boolean isChunkInPool(int chunkId) {
if (pool != null) {
// chunks that are from pool will return true chunk reference not null
Chunk c = getChunk(chunkId);
if (c==null) {
return false;
}
return pool.reclaimedChunks.contains(c);
Chunk c = getChunk(chunkId);
if (c==null) {
return false;
}
// chunks that are from pool will return true chunk reference not null
if (dataChunksPool != null && dataChunksPool.reclaimedChunks.contains(c)) {
return true;
} else if (indexChunksPool != null && indexChunksPool.reclaimedChunks.contains(c)) {
return true;
}
return false;
}
@ -448,31 +578,58 @@ public class ChunkCreator {
*/
@VisibleForTesting
void clearChunksInPool() {
if (pool != null) {
pool.reclaimedChunks.clear();
if (dataChunksPool != null) {
dataChunksPool.reclaimedChunks.clear();
}
if (indexChunksPool != null) {
indexChunksPool.reclaimedChunks.clear();
}
}
int getChunkSize() {
return getChunkSize(ChunkType.DATA_CHUNK);
}
int getChunkSize(ChunkType chunkType) {
switch (chunkType) {
case INDEX_CHUNK:
if (indexChunksPool != null) {
return indexChunksPool.getChunkSize();
}
case DATA_CHUNK:
if (dataChunksPool != null) {
return dataChunksPool.getChunkSize();
} else { // When pools are empty
return chunkSize;
}
default:
throw new IllegalArgumentException(
"chunkType must either be INDEX_CHUNK or DATA_CHUNK");
}
}
synchronized void putbackChunks(Set<Integer> chunks) {
// if there is no pool just try to clear the chunkIdMap in case there is something
if ( pool == null ) {
if (dataChunksPool == null && indexChunksPool == null) {
this.removeChunks(chunks);
return;
}
// if there is pool, go over all chunk IDs that came back, the chunks may be from pool or not
// if there is a pool, go over all chunk IDs that came back, the chunks may be from pool or not
for (int chunkID : chunks) {
// translate chunk ID to chunk, if chunk initially wasn't in pool
// this translation will (most likely) return null
Chunk chunk = ChunkCreator.this.getChunk(chunkID);
if (chunk != null) {
// Jumbo chunks are covered with chunkIdMap, but are not from pool, so such a chunk should
// be released here without going to pool.
// Removing them from chunkIdMap will cause their removal by the GC.
if (chunk.isJumbo()) {
this.removeChunk(chunkID);
if (chunk.isFromPool() && chunk.isIndexChunk()) {
indexChunksPool.putbackChunks(chunk);
} else if (chunk.isFromPool() && chunk.size == dataChunksPool.getChunkSize()) {
dataChunksPool.putbackChunks(chunk);
} else {
pool.putbackChunks(chunk);
// chunks which are not from one of the pools
// should be released without going to the pools.
// Removing them from chunkIdMap will cause their removal by the GC.
this.removeChunk(chunkID);
}
}
// if chunk is null, it was never covered by the chunkIdMap (and so wasn't in pool also),
@ -482,3 +639,4 @@ public class ChunkCreator {
}
}

28
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableMemStoreLAB.java
View File

@ -61,32 +61,26 @@ public class ImmutableMemStoreLAB implements MemStoreLAB {
return mslab.forceCopyOfBigCellInto(cell);
}
/* Creating chunk to be used as index chunk in CellChunkMap, part of the chunks array.
** Returning a new chunk, without replacing current chunk,
/* Returning a new pool chunk, without replacing current chunk,
** meaning MSLABImpl does not make the returned chunk as CurChunk.
** The space on this chunk will be allocated externally.
** The interface is only for external callers
** The interface is only for external callers.
*/
@Override
public Chunk getNewExternalChunk() {
public Chunk getNewExternalChunk(ChunkCreator.ChunkType chunkType) {
MemStoreLAB mslab = this.mslabs.get(0);
return mslab.getNewExternalChunk();
return mslab.getNewExternalChunk(chunkType);
}
/* Creating chunk to be used as data chunk in CellChunkMap.
** This chunk is bigger the normal constant chunk size, and thus called JumboChunk it is used for
** jumbo cells (which size is bigger than normal chunks).
** Jumbo Chunks are needed only for CCM and thus are created only in
** CompactingMemStore.IndexType.CHUNK_MAP type.
** Returning a new chunk, without replacing current chunk,
** meaning MSLABImpl does not make the returned chunk as CurChunk.
** The space on this chunk will be allocated externally.
** The interface is only for external callers
*/
/* Returning a new chunk, without replacing current chunk,
** meaning MSLABImpl does not make the returned chunk as CurChunk.
** The space on this chunk will be allocated externally.
** The interface is only for external callers.
*/
@Override
public Chunk getNewExternalJumboChunk(int size) {
public Chunk getNewExternalChunk(int size) {
MemStoreLAB mslab = this.mslabs.get(0);
return mslab.getNewExternalJumboChunk(size);
return mslab.getNewExternalChunk(size);
}
@Override

20
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLAB.java
View File

@ -52,6 +52,8 @@ public interface MemStoreLAB {
String CHUNK_SIZE_KEY = "hbase.hregion.memstore.mslab.chunksize";
int CHUNK_SIZE_DEFAULT = 2048 * 1024;
String INDEX_CHUNK_PERCENTAGE_KEY = "hbase.hregion.memstore.mslab.indexchunksize";
float INDEX_CHUNK_PERCENTAGE_DEFAULT = 0.1f;
String MAX_ALLOC_KEY = "hbase.hregion.memstore.mslab.max.allocation";
int MAX_ALLOC_DEFAULT = 256 * 1024; // allocs bigger than this don't go through
// allocator
@ -94,25 +96,19 @@ public interface MemStoreLAB {
*/
void decScannerCount();
/* Creating chunk to be used as index chunk in CellChunkMap, part of the chunks array.
** Returning a new chunk, without replacing current chunk,
/* Returning a new pool chunk, without replacing current chunk,
** meaning MSLABImpl does not make the returned chunk as CurChunk.
** The space on this chunk will be allocated externally.
** The interface is only for external callers
** The interface is only for external callers.
*/
Chunk getNewExternalChunk();
Chunk getNewExternalChunk(ChunkCreator.ChunkType chunkType);
/* Creating chunk to be used as data chunk in CellChunkMap.
** This chunk is bigger than normal constant chunk size, and thus called JumboChunk it is used for
** jumbo cells (which size is bigger than normal chunks).
** Jumbo Chunks are needed only for CCM and thus are created only in
** CompactingMemStore.IndexType.CHUNK_MAP type.
** Returning a new chunk, without replacing current chunk,
/* Returning a new chunk, without replacing current chunk,
** meaning MSLABImpl does not make the returned chunk as CurChunk.
** The space on this chunk will be allocated externally.
** The interface is only for external callers
** The interface is only for external callers.
*/
Chunk getNewExternalJumboChunk(int size);
Chunk getNewExternalChunk(int size);
static MemStoreLAB newInstance(Configuration conf) {
MemStoreLAB memStoreLAB = null;

74
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreLABImpl.java
View File

@ -67,14 +67,14 @@ public class MemStoreLABImpl implements MemStoreLAB {
static final Logger LOG = LoggerFactory.getLogger(MemStoreLABImpl.class);
private AtomicReference<Chunk> curChunk = new AtomicReference<>();
private AtomicReference<Chunk> currChunk = new AtomicReference<>();
// Lock to manage multiple handlers requesting for a chunk
private ReentrantLock lock = new ReentrantLock();
// A set of chunks contained by this memstore LAB
@VisibleForTesting
Set<Integer> chunks = new ConcurrentSkipListSet<Integer>();
private final int chunkSize;
private final int dataChunkSize;
private final int maxAlloc;
private final ChunkCreator chunkCreator;
private final CompactingMemStore.IndexType idxType; // what index is used for corresponding segment
@ -94,11 +94,11 @@ public class MemStoreLABImpl implements MemStoreLAB {
}
public MemStoreLABImpl(Configuration conf) {
chunkSize = conf.getInt(CHUNK_SIZE_KEY, CHUNK_SIZE_DEFAULT);
dataChunkSize = conf.getInt(CHUNK_SIZE_KEY, CHUNK_SIZE_DEFAULT);
maxAlloc = conf.getInt(MAX_ALLOC_KEY, MAX_ALLOC_DEFAULT);
this.chunkCreator = ChunkCreator.getInstance();
// if we don't exclude allocations >CHUNK_SIZE, we'd infiniteloop on one!
Preconditions.checkArgument(maxAlloc <= chunkSize,
Preconditions.checkArgument(maxAlloc <= dataChunkSize,
MAX_ALLOC_KEY + " must be less than " + CHUNK_SIZE_KEY);
// if user requested to work with MSLABs (whether on- or off-heap), then the
@ -120,14 +120,14 @@ public class MemStoreLABImpl implements MemStoreLAB {
*/
@Override
public Cell forceCopyOfBigCellInto(Cell cell) {
int size = KeyValueUtil.length(cell);
int size = KeyValueUtil.length(cell) + ChunkCreator.SIZEOF_CHUNK_HEADER;
Preconditions.checkArgument(size >= 0, "negative size");
if (size <= chunkSize) {
if (size <= dataChunkSize) {
// Using copyCellInto for cells which are bigger than the original maxAlloc
Cell newCell = copyCellInto(cell, chunkSize);
Cell newCell = copyCellInto(cell, dataChunkSize);
return newCell;
} else {
Chunk c = getNewExternalJumboChunk(size);
Chunk c = getNewExternalChunk(size);
int allocOffset = c.alloc(size);
return copyToChunkCell(cell, c.getData(), allocOffset, size);
}
@ -240,7 +240,7 @@ public class MemStoreLABImpl implements MemStoreLAB {
* @return true if we won the race to retire the chunk
*/
private void tryRetireChunk(Chunk c) {
curChunk.compareAndSet(c, null);
currChunk.compareAndSet(c, null);
// If the CAS succeeds, that means that we won the race
// to retire the chunk. We could use this opportunity to
// update metrics on external fragmentation.
@ -255,7 +255,8 @@ public class MemStoreLABImpl implements MemStoreLAB {
*/
private Chunk getOrMakeChunk() {
// Try to get the chunk
Chunk c = curChunk.get();
Chunk c;
c = currChunk.get();
if (c != null) {
return c;
}
@ -265,14 +266,14 @@ public class MemStoreLABImpl implements MemStoreLAB {
if (lock.tryLock()) {
try {
// once again check inside the lock
c = curChunk.get();
c = currChunk.get();
if (c != null) {
return c;
}
c = this.chunkCreator.getChunk(idxType);
if (c != null) {
// set the curChunk. No need of CAS as only one thread will be here
curChunk.set(c);
currChunk.set(c);
chunks.add(c.getId());
return c;
}
@ -283,38 +284,41 @@ public class MemStoreLABImpl implements MemStoreLAB {
return null;
}
/* Creating chunk to be used as index chunk in CellChunkMap, part of the chunks array.
** Returning a new chunk, without replacing current chunk,
/* Returning a new pool chunk, without replacing current chunk,
** meaning MSLABImpl does not make the returned chunk as CurChunk.
** The space on this chunk will be allocated externally.
** The interface is only for external callers
** The interface is only for external callers.
*/
@Override
public Chunk getNewExternalChunk() {
// the new chunk is going to be part of the chunk array and will always be referenced
Chunk c = this.chunkCreator.getChunk();
chunks.add(c.getId());
return c;
public Chunk getNewExternalChunk(ChunkCreator.ChunkType chunkType) {
switch (chunkType) {
case INDEX_CHUNK:
case DATA_CHUNK:
Chunk c = this.chunkCreator.getChunk(chunkType);
chunks.add(c.getId());
return c;
case JUMBO_CHUNK: // a jumbo chunk doesn't have a fixed size
default:
return null;
}
}
/* Creating chunk to be used as data chunk in CellChunkMap.
** This chunk is bigger than normal constant chunk size, and thus called JumboChunk.
** JumboChunk is used for jumbo cell (which size is bigger than normal chunk). It is allocated
** once per cell. So even if there is space this is not reused.
** Jumbo Chunks are used only for CCM and thus are created only in
** CompactingMemStore.IndexType.CHUNK_MAP type.
** Returning a new chunk, without replacing current chunk,
/* Returning a new chunk, without replacing current chunk,
** meaning MSLABImpl does not make the returned chunk as CurChunk.
** The space on this chunk will be allocated externally.
** The interface is only for external callers
** The interface is only for external callers.
** Chunks from pools are not allocated from here, since they have fixed sizes
*/
@Override
public Chunk getNewExternalJumboChunk(int size) {
// the new chunk is going to hold the jumbo cell data and need to be referenced by a strong map
// thus giving the CCM index type
Chunk c = this.chunkCreator.getJumboChunk(CompactingMemStore.IndexType.CHUNK_MAP, size);
chunks.add(c.getId());
return c;
public Chunk getNewExternalChunk(int size) {
int allocSize = size + ChunkCreator.getInstance().SIZEOF_CHUNK_HEADER;
if (allocSize <= ChunkCreator.getInstance().getChunkSize()) {
return getNewExternalChunk(ChunkCreator.ChunkType.DATA_CHUNK);
} else {
Chunk c = this.chunkCreator.getJumboChunk(size);
chunks.add(c.getId());
return c;
}
}
@Override
@ -329,7 +333,7 @@ public class MemStoreLABImpl implements MemStoreLAB {
@VisibleForTesting
Chunk getCurrentChunk() {
return this.curChunk.get();
return currChunk.get();
}
@VisibleForTesting

3
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java
View File

@ -118,7 +118,8 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
long globalMemStoreLimit = (long) (ManagementFactory.getMemoryMXBean().getHeapMemoryUsage()
.getMax() * MemorySizeUtil.getGlobalMemStoreHeapPercent(conf, false));
chunkCreator = ChunkCreator.initialize(MemStoreLABImpl.CHUNK_SIZE_DEFAULT, false,
globalMemStoreLimit, 0.4f, MemStoreLAB.POOL_INITIAL_SIZE_DEFAULT, null);
globalMemStoreLimit, 0.4f, MemStoreLAB.POOL_INITIAL_SIZE_DEFAULT,
null);
assertTrue(chunkCreator != null);
}

6
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreChunkPool.java
View File

@ -233,10 +233,10 @@ public class TestMemStoreChunkPool {
final int chunkSize = 40;
final int valSize = 7;
ChunkCreator oldCreator = ChunkCreator.getInstance();
ChunkCreator newCreator = new ChunkCreator(chunkSize, false, 400, 1, 0.5f, null);
ChunkCreator newCreator = new ChunkCreator(chunkSize, false, 400, 1, 0.5f, null, 0);
assertEquals(initialCount, newCreator.getPoolSize());
assertEquals(maxCount, newCreator.getMaxCount());
ChunkCreator.INSTANCE = newCreator;// Replace the global ref with the new one we created.
ChunkCreator.instance = newCreator;// Replace the global ref with the new one we created.
// Used it for the testing. Later in finally we put
// back the original
final KeyValue kv = new KeyValue(Bytes.toBytes("r"), Bytes.toBytes("f"), Bytes.toBytes("q"),
@ -265,7 +265,7 @@ public class TestMemStoreChunkPool {
t3.join();
assertTrue(newCreator.getPoolSize() <= maxCount);
} finally {
ChunkCreator.INSTANCE = oldCreator;
ChunkCreator.instance = oldCreator;
}
}
}

8
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
View File

@ -206,8 +206,8 @@ public class TestMemStoreLAB {
MemStoreLABImpl mslab = new MemStoreLABImpl();
// by default setting, there should be no chunks initialized in the pool
assertTrue(mslab.getPooledChunks().isEmpty());
oldInstance = ChunkCreator.INSTANCE;
ChunkCreator.INSTANCE = null;
oldInstance = ChunkCreator.instance;
ChunkCreator.instance = null;
// reset mslab with chunk pool
Configuration conf = HBaseConfiguration.create();
conf.setDouble(MemStoreLAB.CHUNK_POOL_MAXSIZE_KEY, 0.1);
@ -251,7 +251,7 @@ public class TestMemStoreLAB {
}
// none of the chunkIds would have been returned back
assertTrue("All the chunks must have been cleared",
ChunkCreator.INSTANCE.numberOfMappedChunks() != 0);
ChunkCreator.instance.numberOfMappedChunks() != 0);
int pooledChunksNum = mslab.getPooledChunks().size();
// close the mslab
mslab.close();
@ -261,7 +261,7 @@ public class TestMemStoreLAB {
+ " after mslab closed but actually: " + (pooledChunksNum-queueLength),
pooledChunksNum-queueLength == 0);
} finally {
ChunkCreator.INSTANCE = oldInstance;
ChunkCreator.instance = oldInstance;
}
}

2
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemstoreLABWithoutPool.java
View File

@ -148,7 +148,7 @@ public class TestMemstoreLABWithoutPool {
}
// all of the chunkIds would have been returned back
assertTrue("All the chunks must have been cleared",
ChunkCreator.INSTANCE.numberOfMappedChunks() == 0);
ChunkCreator.instance.numberOfMappedChunks() == 0);
}
private Thread getChunkQueueTestThread(final MemStoreLABImpl mslab, String threadName,