diff --git a/CHANGES.txt b/CHANGES.txt
index 37ad298e45..5b101ab3de 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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)
diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java
index ac4f198c79..b82e122248 100644
--- a/src/java/org/apache/cassandra/utils/memory/BufferPool.java
+++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java
@@ -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.
+ *
+ *
+ * The lifecycle of a normal Chunk:
+ *
+ * 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 │
+ * ↑ │
+ * └────────────────────────────────────────────────────────────────────────────────┘
+ *
*/
public class BufferPool
{
@@ -631,10 +643,11 @@ public class BufferPool
private void clearForEach(Consumer 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 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);
}
}
diff --git a/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java b/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java
index 4c1afdb4d5..ecc65802fb 100644
--- a/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java
+++ b/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java
@@ -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