natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-4.1' into trunk

This commit is contained in:
Aleksey Yeschenko 2022-10-11 15:47:43 +01:00
parent af9439bcd7 3fc9bc033a
commit 04cfb57dde
3 changed files with 165 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -91,6 +91,7 @@ Merged from 4.1:
* Revert removal of withBufferSizeInMB(int size) in CQLSSTableWriter.Builder class and deprecate it in favor of withBufferSizeInMiB(int size) (CASSANDRA-17675)
* Remove expired snapshots of dropped tables after restart (CASSANDRA-17619)
Merged from 4.0:
* Fix multiple BufferPool bugs (CASSANDRA-16681)
* Fix StorageService.getNativeaddress handling of IPv6 addresses (CASSANDRA-17945)
* Mitigate direct buffer memory OOM on replacements (CASSANDRA-17895)
* Fix repair failure on assertion if two peers have overlapping mismatching ranges (CASSANDRA-17900)

200
src/java/org/apache/cassandra/utils/memory/BufferPool.java
View File

@ -106,6 +106,18 @@ import static org.apache.cassandra.utils.memory.MemoryUtil.isExactlyDirect;
*
* Note: even though partially freed chunks improves cache utilization when chunk cache holds outstanding buffer for
* arbitrary period, there is still fragmentation in the partially freed chunk because of non-uniform allocation size.
* <p/>
*
* The lifecycle of a normal Chunk:
* <pre>
* new acquire release recycle
* ────────→ in GlobalPool ──────────────→ in LocalPool ──────────────→ EVICTED ──────────────────┐
* owner = null owner = LocalPool owner = null │
* status = IN_USE status = IN_USE status = EVICTED │
* ready serves get / free serves free only │
* ↑ │
* └────────────────────────────────────────────────────────────────────────────────┘
* </pre>
*/
public class BufferPool
{
@ -631,10 +643,11 @@ public class BufferPool
private void clearForEach(Consumer<Chunk> consumer)
{
int oldCount = count;
Chunk chunk0 = this.chunk0, chunk1 = this.chunk1, chunk2 = this.chunk2;
this.chunk0 = this.chunk1 = this.chunk2 = null;
forEach(consumer, count, chunk0, chunk1, chunk2);
count = 0;
this.chunk0 = this.chunk1 = this.chunk2 = null;
forEach(consumer, oldCount, chunk0, chunk1, chunk2);
}
private static void forEach(Consumer<Chunk> consumer, int count, Chunk chunk0, Chunk chunk1, Chunk chunk2)
@ -752,6 +765,9 @@ public class BufferPool
private final LocalPoolRef leakRef;
private final MicroQueueOfChunks chunks = new MicroQueueOfChunks();
private final Thread owningThread = Thread.currentThread();
/**
* If we are on outer LocalPool, whose chunks are == NORMAL_CHUNK_SIZE, we may service allocation requests
* for buffers much smaller than
@ -815,41 +831,39 @@ public class BufferPool
return;
}
// ask the free method to take exclusive ownership of the act of recycling if chunk is owned by ourselves
long free = chunk.free(buffer, owner == this && recycleWhenFree);
// free:
// * 0L: current pool must be the owner. we can fully recyle the chunk.
// * -1L:
// * for normal chunk:
// a) if it has owner, do nothing.
// b) if it not owner, try to recyle it either fully or partially if not already recyled.
// * for tiny chunk:
// a) if it has owner, do nothing.
// b) if it has not owner, recycle the tiny chunk back to parent chunk
// * others:
// * for normal chunk: partial recycle the chunk if it can be partially recycled but not yet recycled.
// * for tiny chunk: do nothing.
if (free == 0L)
long free = chunk.free(buffer);
if (free == -1L && owner == this && owningThread == Thread.currentThread() && recycleWhenFree)
{
assert owner == this;
// 0L => we own recycling responsibility, so must recycle;
// We must remove the Chunk from our local queue
// The chunk was fully freed, and we're the owner - let's release the chunk from this pool
// and give it back to the parent.
//
// We can remove the chunk from our local queue only if we're the owner of the chunk,
// and we're running this code on the thread that owns this local pool
// because the local queue is not thread safe.
//
// Please note that we may end up running `put` on a different thread when we're called
// from chunk.tryRecycle() on a child chunk which was previously owned by tinyPool of this pool.
// Such tiny chunk will point to this pool with its recycler reference. Thanks to the recycler, a thread
// that returns the tiny chunk can end up here in a LocalPool that's not neccessarily local to the
// calling thread, as there is no guarantee a child chunk is returned to the pool
// by the same thread that originally allocated it.
// It is ok we skip recycling in such case, and it does not cause
// a leak because those chunks are still referenced by the local pool.
remove(chunk);
chunk.recycle();
chunk.release();
}
else if (free == -1L && owner != this && chunk.owner == null && !chunk.recycler.canRecyclePartially())
else if (chunk.owner == null)
{
// although we try to take recycle ownership cheaply, it is not always possible to do so if the owner is racing to unset.
// we must also check after completely freeing if the owner has since been unset, and try to recycle
// The chunk has no owner, so we can attempt to recycle it from any thread because we don't need
// to remove it from the local pool.
// For normal chunk this would recycle the chunk fully or partially if not already recycled.
// For tiny chunk, this would recycle the tiny chunk back to the parent chunk,
// if this chunk is completely free.
chunk.tryRecycle();
}
else if (chunk.owner == null && chunk.recycler.canRecyclePartially() && chunk.setInUse(Chunk.Status.EVICTED))
{
// re-cirlate partially freed normal chunk to global list
chunk.partiallyRecycle();
}
if (owner == this)
if (owner == this && owningThread == Thread.currentThread())
{
MemoryUtil.setAttachment(buffer, null);
MemoryUtil.setDirectByteBuffer(buffer, 0, 0);
@ -886,11 +900,7 @@ public class BufferPool
{
ByteBuffer ret = tryGet(size, sizeIsLowerBound);
if (ret != null)
{
metrics.hits.mark();
memoryInUse.add(ret.capacity());
return ret;
}
if (size > NORMAL_CHUNK_SIZE)
{
@ -905,7 +915,6 @@ public class BufferPool
logger.trace("Requested buffer size {} has been allocated directly due to lack of capacity", prettyPrintMemory(size));
}
metrics.misses.mark();
return allocate(size, BufferType.OFF_HEAP);
}
@ -925,10 +934,21 @@ public class BufferPool
}
else if (size > NORMAL_CHUNK_SIZE)
{
metrics.misses.mark();
return null;
}
return pool.tryGetInternal(size, sizeIsLowerBound);
ByteBuffer ret = pool.tryGetInternal(size, sizeIsLowerBound);
if (ret != null)
{
metrics.hits.mark();
memoryInUse.add(ret.capacity());
}
else
{
metrics.misses.mark();
}
return ret;
}
@Inline
@ -961,6 +981,7 @@ public class BufferPool
{
ByteBuffer buffer = chunk.slab;
Chunk parentChunk = Chunk.getParentChunk(buffer);
assert parentChunk != null; // tiny chunk always has a parent chunk
put(buffer, parentChunk);
}
@ -1005,19 +1026,18 @@ public class BufferPool
// releasing tiny chunks may result in releasing current evicted chunk
tinyPool.chunks.removeIf((child, parent) -> Chunk.getParentChunk(child.slab) == parent, evict);
evict.release();
// Mark it as evicted and will be eligible for partial recyle if recycler allows
evict.setEvicted(Chunk.Status.IN_USE);
}
}
public void release()
{
if (tinyPool != null)
tinyPool.release();
chunks.release();
reuseObjects.clear();
localPoolReferences.remove(leakRef);
leakRef.clear();
if (tinyPool != null)
tinyPool.release();
}
@VisibleForTesting
@ -1163,8 +1183,7 @@ public class BufferPool
}
/**
* Acquire the chunk for future allocations: set the owner and prep
* the free slots mask.
* Acquire the chunk for future allocations: set the owner
*/
void acquire(LocalPool owner)
{
@ -1180,28 +1199,94 @@ public class BufferPool
void release()
{
this.owner = null;
boolean statusUpdated = setEvicted();
assert statusUpdated : "Status of chunk " + this + " was not IN_USE.";
tryRecycle();
}
/**
* If the chunk is free, changes the chunk's status to IN_USE and returns the chunk to the pool
* that it was acquired from.
*
* Can recycle the chunk partially if the recycler supports it.
* This method can be called from multiple threads safely.
*
* Calling this method on a chunk that's currently in use (either owned by a LocalPool or already recycled)
* has no effect.
*/
void tryRecycle()
{
assert owner == null;
if (isFree() && freeSlotsUpdater.compareAndSet(this, -1L, 0L))
recycle();
// Note that this may race with release(), therefore the order of those checks does matter.
// The EVICTED check may fail if the chunk was already partially recycled.
if (status != Status.EVICTED)
return;
if (owner != null)
return;
// We must use consistently either tryRecycleFully or tryRecycleFullyOrPartially,
// but we must not mix those for a single chunk, because they use a different mechanism for guarding
// that the chunk would be recycled at most once until the next acquire.
//
// If the recycler cannot recycle blocks partially, we have to make sure freeSlots was zeroed properly.
// Only one thread can transition freeSlots from -1 to 0 atomically, so this is a good way
// of ensuring only one thread recycles the block. In this case the chunk's status is
// updated only after freeSlots CAS succeeds.
//
// If the recycler can recycle blocks partially, we use the status field
// to guard at-most-once recycling. We cannot rely on atomically updating freeSlots from -1 to 0, because
// in this case we cannot expect freeSlots to be -1 (if it was, it wouldn't be partial).
if (recycler.canRecyclePartially())
tryRecycleFullyOrPartially();
else
tryRecycleFully();
}
void recycle()
/**
* Returns this chunk to the pool where it was acquired from, if it wasn't returned already.
* The chunk does not have to be totally free, but should have some free bits.
* However, if the chunk is fully free, it is released fully, not partially.
*/
private void tryRecycleFullyOrPartially()
{
assert freeSlots == 0L;
recycler.recycle(this);
assert recycler.canRecyclePartially();
if (free() > 0 && setInUse())
{
assert owner == null;
if (!tryRecycleFully()) // prefer to recycle fully, as fully free chunks are returned to a higher priority queue
recyclePartially();
}
}
public void partiallyRecycle()
private boolean tryRecycleFully()
{
if (!isFree() || !freeSlotsUpdater.compareAndSet(this, -1L, 0L))
return false;
recycleFully();
return true;
}
private void recyclePartially()
{
assert owner == null;
assert status == Status.IN_USE;
recycler.recyclePartially(this);
}
private void recycleFully()
{
assert owner == null;
assert freeSlots == 0L;
Status expectedStatus = recycler.canRecyclePartially() ? Status.IN_USE : Status.EVICTED;
boolean statusUpdated = setStatus(expectedStatus, Status.IN_USE);
// impossible: could only happen if another thread updated the status in the meantime
assert statusUpdated : "Status of chunk " + this + " was not " + expectedStatus;
recycler.recycle(this);
}
/**
* We stash the chunk in the attachment of a buffer
* that was returned by get(), this method simply
@ -1377,11 +1462,10 @@ public class BufferPool
/**
* Release a buffer. Return:
* 0L if the buffer must be recycled after the call;
* -1L if it is free (and so we should tryRecycle if owner is now null)
* some other value otherwise
**/
long free(ByteBuffer buffer, boolean tryRelease)
long free(ByteBuffer buffer)
{
if (!releaseAttachment(buffer))
return 1L;
@ -1402,8 +1486,6 @@ public class BufferPool
long cur = freeSlots;
next = cur | shiftedSlotBits;
assert next == (cur ^ shiftedSlotBits); // ensure no double free
if (tryRelease && (next == -1L))
next = 0L;
if (freeSlotsUpdater.compareAndSet(this, cur, next))
return next;
}
@ -1455,7 +1537,7 @@ public class BufferPool
{
Chunk parent = getParentChunk(slab);
if (parent != null)
parent.free(slab, false);
parent.free(slab);
else
FileUtils.clean(slab);
}
@ -1466,7 +1548,7 @@ public class BufferPool
{
chunk.owner = null;
chunk.freeSlots = 0L;
chunk.recycle();
chunk.recycleFully();
}
}
@ -1480,14 +1562,14 @@ public class BufferPool
return statusUpdater.compareAndSet(this, current, update);
}
boolean setInUse(Status prev)
private boolean setInUse()
{
return setStatus(prev, Status.IN_USE);
return setStatus(Status.EVICTED, Status.IN_USE);
}
boolean setEvicted(Status prev)
private boolean setEvicted()
{
return setStatus(prev, Status.EVICTED);
return setStatus(Status.IN_USE, Status.EVICTED);
}
}

23
test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java
View File

@ -64,6 +64,26 @@ public class BufferPoolTest
assertEquals(0, bufferPool.usedSizeInBytes());
}
@Test
public void testTryGet()
{
final int size = RandomAccessReader.DEFAULT_BUFFER_SIZE;
ByteBuffer buffer = bufferPool.tryGet(size);
assertNotNull(buffer);
assertEquals(size, buffer.capacity());
assertEquals(true, buffer.isDirect());
assertEquals(size, bufferPool.usedSizeInBytes());
BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
assertNotNull(chunk);
assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
bufferPool.put(buffer);
assertEquals(null, bufferPool.unsafeCurrentChunk());
assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
assertEquals(0, bufferPool.usedSizeInBytes());
}
@Test
public void testPageAligned()
@ -99,10 +119,12 @@ public class BufferPoolTest
ByteBuffer buffer1 = bufferPool.get(size1, BufferType.OFF_HEAP);
assertNotNull(buffer1);
assertEquals(size1, buffer1.capacity());
assertEquals(size1, bufferPool.usedSizeInBytes());
ByteBuffer buffer2 = bufferPool.get(size2, BufferType.OFF_HEAP);
assertNotNull(buffer2);
assertEquals(size2, buffer2.capacity());
assertEquals(size1 + size2, bufferPool.usedSizeInBytes());
BufferPool.Chunk chunk = bufferPool.unsafeCurrentChunk();
assertNotNull(chunk);
@ -113,6 +135,7 @@ public class BufferPoolTest
assertEquals(null, bufferPool.unsafeCurrentChunk());
assertEquals(BufferPool.GlobalPool.MACRO_CHUNK_SIZE, bufferPool.sizeInBytes());
assertEquals(0, bufferPool.usedSizeInBytes());
}
@Test