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.0' into cassandra-4.1

This commit is contained in:
Aleksey Yeschenko 2022-10-06 15:29:01 +01:00
parent c4bccb000a 4e1d31e729
commit 0b083d3e73
4 changed files with 219 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -1,7 +1,8 @@
4.1-beta1
4.1-beta2
* Allow pre-V5 global limit on bytes in flight to revert to zero asynchronously in RateLimitingTest (CASSANDRA-17927)
Merged from 4.0:
Merged from 3.11:
* Make LongBufferPoolTest insensitive to timing (CASSANDRA-16681)
Merged from 3.0:
* Fix auto-completing "WITH" when creating a materialized view (CASSANDRA-17879)

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

@ -291,12 +291,27 @@ public class BufferPool
return globalPool;
}
/**
* Forces to recycle free local chunks back to the global pool.
* This is needed because if buffers were freed by a different thread than the one
* that allocated them, recycling might not have happened and the local pool may still own some
* fully empty chunks.
*/
@VisibleForTesting
public void releaseLocal()
{
localPool.get().release();
}
interface Debug
{
public static Debug NO_OP = new Debug()
{
@Override
public void registerNormal(Chunk chunk) {}
@Override
public void acquire(Chunk chunk) {}
@Override
public void recycleNormal(Chunk oldVersion, Chunk newVersion) {}
@Override
@ -304,6 +319,7 @@ public class BufferPool
};
void registerNormal(Chunk chunk);
void acquire(Chunk chunk);
void recycleNormal(Chunk oldVersion, Chunk newVersion);
void recyclePartial(Chunk chunk);
}
@ -376,6 +392,14 @@ public class BufferPool
/** Return a chunk, the caller will take owership of the parent chunk. */
public Chunk get()
{
Chunk chunk = getInternal();
if (chunk != null)
debug.acquire(chunk);
return chunk;
}
private Chunk getInternal()
{
Chunk chunk = chunks.poll();
if (chunk != null)

219
test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java
View File

@ -24,7 +24,7 @@ import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.AfterClass;
@ -76,7 +76,8 @@ public class LongBufferPoolTest
{
static class DebugChunk
{
volatile long lastRecycled;
volatile long lastAcquired = 0; // the number of last round when the chunk was acquired from the global pool
volatile long lastRecycled = 0; // the number of last round when the chunk was recycled back to the global pool
static DebugChunk get(BufferPool.Chunk chunk)
{
if (chunk.debugAttachment == null)
@ -84,7 +85,7 @@ public class LongBufferPoolTest
return (DebugChunk) chunk.debugAttachment;
}
}
long recycleRound = 1;
AtomicLong recycleRound = new AtomicLong(1);
final List<BufferPool.Chunk> normalChunks = new ArrayList<>();
public synchronized void registerNormal(BufferPool.Chunk chunk)
@ -92,21 +93,39 @@ public class LongBufferPoolTest
chunk.debugAttachment = new DebugChunk();
normalChunks.add(chunk);
}
@Override
public void acquire(BufferPool.Chunk chunk)
{
DebugChunk.get(chunk).lastAcquired = recycleRound.get();
}
@Override
public void recycleNormal(BufferPool.Chunk oldVersion, BufferPool.Chunk newVersion)
{
newVersion.debugAttachment = oldVersion.debugAttachment;
DebugChunk.get(oldVersion).lastRecycled = recycleRound;
DebugChunk.get(oldVersion).lastRecycled = recycleRound.get();
}
@Override
public void recyclePartial(BufferPool.Chunk chunk)
{
DebugChunk.get(chunk).lastRecycled = recycleRound;
DebugChunk.get(chunk).lastRecycled = recycleRound.get();
}
public synchronized void check()
{
long lastRecycledMax = 0;
long currentRound = recycleRound.get();
for (BufferPool.Chunk chunk : normalChunks)
assert DebugChunk.get(chunk).lastRecycled == recycleRound;
recycleRound++;
{
DebugChunk dc = DebugChunk.get(chunk);
assert dc.lastRecycled >= dc.lastAcquired: "Last recycled " + dc.lastRecycled + " < last acquired " + dc.lastAcquired;
lastRecycledMax = Math.max(lastRecycledMax, dc.lastRecycled);
}
assert lastRecycledMax == currentRound : "No chunk recycled in round " + currentRound + ". " +
"Last chunk recycled in round " + lastRecycledMax + '.';
recycleRound.incrementAndGet();
}
}
@ -124,18 +143,18 @@ public class LongBufferPoolTest
}
@Test
public void testPoolAllocateWithRecyclePartially() throws InterruptedException, ExecutionException
public void testPoolAllocateWithRecyclePartially() throws InterruptedException, ExecutionException, BrokenBarrierException, TimeoutException
{
testPoolAllocate(true);
}
@Test
public void testPoolAllocateWithoutRecyclePartially() throws InterruptedException, ExecutionException
public void testPoolAllocateWithoutRecyclePartially() throws InterruptedException, ExecutionException, BrokenBarrierException, TimeoutException
{
testPoolAllocate(false);
}
private void testPoolAllocate(boolean recyclePartially) throws InterruptedException, ExecutionException
private void testPoolAllocate(boolean recyclePartially) throws InterruptedException, ExecutionException, BrokenBarrierException, TimeoutException
{
BufferPool pool = new BufferPool("test_pool", 16 << 20, recyclePartially);
testAllocate(pool, Runtime.getRuntime().availableProcessors() * 2, TimeUnit.MINUTES.toNanos(2L));
@ -176,9 +195,13 @@ public class LongBufferPoolTest
final long until;
final CountDownLatch latch;
final SPSCQueue<BufferCheck>[] sharedRecycle;
final AtomicBoolean[] makingProgress;
final AtomicBoolean burnFreed;
final AtomicBoolean[] freedAllMemory;
volatile boolean shouldFreeMemoryAndSuspend = false;
final CyclicBarrier stopAllocationsBarrier;
final CyclicBarrier freedAllMemoryBarrier;
final CyclicBarrier resumeAllocationsBarrier;
final ExecutorService executorService;
final List<Future<Boolean>> threadResultFuture;
final int targetSizeQuanta;
@ -191,17 +214,18 @@ public class LongBufferPoolTest
until = nanoTime() + duration;
latch = new CountDownLatch(threadCount);
sharedRecycle = new SPSCQueue[threadCount];
makingProgress = new AtomicBoolean[threadCount];
burnFreed = new AtomicBoolean(false);
freedAllMemory = new AtomicBoolean[threadCount];
// N worker threads + burner thread + main thread:
stopAllocationsBarrier = new CyclicBarrier(threadCount + 2);
freedAllMemoryBarrier = new CyclicBarrier(threadCount + 2);
resumeAllocationsBarrier = new CyclicBarrier(threadCount + 2);
executorService = Executors.newFixedThreadPool(threadCount + 2, new NamedThreadFactory("test"));
threadResultFuture = new ArrayList<>(threadCount);
for (int i = 0; i < sharedRecycle.length; i++)
{
sharedRecycle[i] = new SPSCQueue<>();
makingProgress[i] = new AtomicBoolean(false);
freedAllMemory[i] = new AtomicBoolean(false);
}
// Divide the poolSize across our threads, deliberately over-subscribing it. Threads
@ -219,17 +243,6 @@ public class LongBufferPoolTest
threadResultFuture.add(future);
}
int countStalledThreads()
{
int stalledThreads = 0;
for (AtomicBoolean progress : makingProgress)
{
if (!progress.getAndSet(false))
stalledThreads++;
}
return stalledThreads;
}
int countDoneThreads()
{
@ -259,9 +272,60 @@ public class LongBufferPoolTest
fail("Checked thread threw exception: " + ex.toString());
}
}
/**
* Implementers must assure all buffers were returned to the buffer pool on run exit.
*/
interface MemoryFreeTask
{
void run();
}
/**
* If the main test loop requested stopping the threads by setting
* {@link TestEnvironment#shouldFreeMemoryAndSuspend},
* waits until all threads reach this call and then frees the memory by running the given memory free task.
* After the task finishes, it waits on the {@link TestEnvironment#freedAllMemoryBarrier} and
* {@link TestEnvironment#resumeAllocationsBarrier} to let the main test loop perform the post-free checks.
* The call exits after {@link TestEnvironment#resumeAllocationsBarrier} is reached by all threads.
*
* @param task the task that should return all buffers held by this thread to the buffer pool
*/
void maybeSuspendAndFreeMemory(MemoryFreeTask task) throws InterruptedException, BrokenBarrierException
{
if (shouldFreeMemoryAndSuspend)
{
try
{
// Wait until allocations stop in all threads; this guanrantees this thread won't
// receive any new buffers from other threads while freeing memory.
stopAllocationsBarrier.await();
// Free our memory
task.run();
// Now wait for the other threads to free their memory
freedAllMemoryBarrier.await();
// Now all memory is freed, but let's not resume allocations until the main test thread
// performs the required checks.
// At this point, used memory indicated by the pool
// should be == 0 and all buffers should be recycled.
resumeAllocationsBarrier.await();
}
catch (BrokenBarrierException | InterruptedException e)
{
// At the end of the test some threads may have already exited,
// so they can't arrive at one of the barriers, and we may end up here.
// This is fine if this happens after the test deadline, and we
// just allow the test worker to exit cleanly.
// It must not happen before the test deadline though, it would likely be a bug,
// so we rethrow in that case.
if (System.nanoTime() < until)
throw e;
}
}
}
}
public void testAllocate(BufferPool bufferPool, int threadCount, long duration) throws InterruptedException, ExecutionException
public void testAllocate(BufferPool bufferPool, int threadCount, long duration) throws InterruptedException, ExecutionException, BrokenBarrierException, TimeoutException
{
logger.info("{} - testing {} threads for {}m", DATE_FORMAT.format(new Date()), threadCount, TimeUnit.NANOSECONDS.toMinutes(duration));
logger.info("Testing BufferPool with memoryUsageThreshold={} and enabling BufferPool.DEBUG", bufferPool.memoryUsageThreshold());
@ -275,21 +339,32 @@ public class LongBufferPoolTest
for (int threadIdx = 0; threadIdx < threadCount; threadIdx++)
testEnv.addCheckedFuture(startWorkerThread(bufferPool, testEnv, threadIdx));
while (!testEnv.latch.await(10L, TimeUnit.SECONDS))
while (!testEnv.latch.await(1L, TimeUnit.SECONDS))
{
int stalledThreads = testEnv.countStalledThreads();
int doneThreads = testEnv.countDoneThreads();
if (doneThreads == 0) // If any threads have completed, they will stop making progress/recycling buffers.
{ // Assertions failures on the threads will be caught below.
assert stalledThreads == 0;
boolean allFreed = testEnv.burnFreed.getAndSet(false);
for (AtomicBoolean freedMemory : testEnv.freedAllMemory)
allFreed = allFreed && freedMemory.getAndSet(false);
if (allFreed)
debug.check();
else
logger.info("All threads did not free all memory in this time slot - skipping buffer recycle check");
try
{
// request all threads to release all buffers to the bufferPool
testEnv.shouldFreeMemoryAndSuspend = true;
testEnv.stopAllocationsBarrier.await(10, TimeUnit.SECONDS);
// wait until all memory released
testEnv.freedAllMemoryBarrier.await(10, TimeUnit.SECONDS);
// now all buffers should be back in the pool, and no more allocations happening
assert bufferPool.usedSizeInBytes() == 0 : "Some buffers haven't been freed. Memory in use = "
+ bufferPool.usedSizeInBytes() + " (expected 0)";
debug.check();
// resume threads only after debug.cycleRound has been increased
testEnv.shouldFreeMemoryAndSuspend = false;
testEnv.resumeAllocationsBarrier.await(10, TimeUnit.SECONDS);
}
catch (TimeoutException e)
{
// a thread that is done will not reach the barriers, so timeout is unexpected only if
// all threads are still running
if (testEnv.countDoneThreads() == 0)
{
logger.error("Some threads have stalled and didn't reach the barrier", e);
return;
}
}
}
@ -321,23 +396,28 @@ public class LongBufferPoolTest
final SPSCQueue<BufferCheck> shareFrom = testEnv.sharedRecycle[threadIdx];
final DynamicList<BufferCheck> checks = new DynamicList<>((int) Math.max(1, targetSize / (1 << 10)));
final SPSCQueue<BufferCheck> shareTo = testEnv.sharedRecycle[(threadIdx + 1) % testEnv.threadCount];
final Future<Boolean> neighbourResultFuture = testEnv.threadResultFuture.get((threadIdx + 1) % testEnv.threadCount);
final ThreadLocalRandom rand = ThreadLocalRandom.current();
int totalSize = 0;
int freeingSize = 0;
int size = 0;
void checkpoint()
{
if (!testEnv.makingProgress[threadIdx].get())
testEnv.makingProgress[threadIdx].set(true);
}
void testOne() throws Exception
{
long currentTargetSize = (rand.nextInt(testEnv.poolSize / 1024) == 0 || !testEnv.freedAllMemory[threadIdx].get()) ? 0 : targetSize;
testEnv.maybeSuspendAndFreeMemory(this::freeAll);
long currentTargetSize = rand.nextInt(testEnv.poolSize / 1024) == 0 ? 0 : targetSize;
int spinCount = 0;
while (totalSize > currentTargetSize - freeingSize)
{
// Don't get stuck in this loop if other threads might be suspended:
if (testEnv.shouldFreeMemoryAndSuspend)
return;
// Don't get stuck in this loop if the neighbour thread exited:
if (neighbourResultFuture.isDone())
return;
// free buffers until we're below our target size
if (checks.size() == 0)
{
@ -383,9 +463,6 @@ public class LongBufferPoolTest
}
}
if (currentTargetSize == 0)
testEnv.freedAllMemory[threadIdx].compareAndSet(false, true);
// allocate a new buffer
size = (int) Math.max(1, AVG_BUFFER_SIZE + (STDEV_BUFFER_SIZE * rand.nextGaussian()));
if (size <= BufferPool.NORMAL_CHUNK_SIZE)
@ -418,6 +495,29 @@ public class LongBufferPoolTest
while (recycleFromNeighbour());
}
/**
* Returns all allocated buffers back to the buffer pool.
*/
void freeAll()
{
while (checks.size() > 0)
{
BufferCheck check = sample();
checks.remove(check.listnode);
check.validate();
bufferPool.put(check.buffer);
}
BufferCheck check;
while ((check = shareFrom.poll()) != null)
{
check.validate();
bufferPool.put(check.buffer);
}
bufferPool.releaseLocal();
}
void cleanup()
{
while (checks.size() > 0)
@ -489,6 +589,8 @@ public class LongBufferPoolTest
private void startBurnerThreads(BufferPool bufferPool, TestEnvironment testEnv)
{
// setup some high churn allocate/deallocate, without any checking
final AtomicLong pendingBuffersCount = new AtomicLong(0);
final SPSCQueue<ByteBuffer> burn = new SPSCQueue<>();
final CountDownLatch doneAdd = new CountDownLatch(1);
testEnv.addCheckedFuture(testEnv.executorService.submit(new TestUntil(bufferPool, testEnv.until)
@ -499,10 +601,10 @@ public class LongBufferPoolTest
{
if (count * BufferPool.NORMAL_CHUNK_SIZE >= testEnv.poolSize / 10)
{
if (burn.exhausted)
if (pendingBuffersCount.get() == 0)
{
count = 0;
testEnv.burnFreed.compareAndSet(false, true);
testEnv.maybeSuspendAndFreeMemory(bufferPool::releaseLocal);
} else
{
Thread.yield();
@ -524,8 +626,10 @@ public class LongBufferPoolTest
if (rand.nextBoolean())
bufferPool.put(buffer);
else
{
pendingBuffersCount.incrementAndGet();
burn.add(buffer);
}
count++;
}
void cleanup()
@ -544,6 +648,7 @@ public class LongBufferPoolTest
return;
}
bufferPool.put(buffer);
pendingBuffersCount.decrementAndGet();
}
void cleanup()
{

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

@ -964,6 +964,37 @@ public class BufferPoolTest
bufferPool.put(buffer);
}
@Test
public void testReleaseLocal()
{
final int size = BufferPool.TINY_ALLOCATION_UNIT;
final int allocationPerChunk = 64;
// occupy 3 tiny chunks
List<ByteBuffer> buffers0 = new ArrayList<>();
BufferPool.Chunk chunk0 = allocate(allocationPerChunk, size, buffers0);
List<ByteBuffer> buffers1 = new ArrayList<>();
BufferPool.Chunk chunk1 = allocate(allocationPerChunk, size, buffers1);
List<ByteBuffer> buffers2 = new ArrayList<>();
BufferPool.Chunk chunk2 = allocate(allocationPerChunk, size, buffers2);
// release them from the pool
bufferPool.releaseLocal();
assertNull(chunk0.owner());
assertNull(chunk1.owner());
assertNull(chunk2.owner());
assertEquals(BufferPool.Chunk.Status.EVICTED, chunk0.status());
assertEquals(BufferPool.Chunk.Status.EVICTED, chunk1.status());
assertEquals(BufferPool.Chunk.Status.EVICTED, chunk2.status());
// cleanup allocated buffers, that should still work fine even though we released them from the localPool
for (ByteBuffer buffer : Iterables.concat(buffers0, buffers1, buffers2))
bufferPool.put(buffer);
assertEquals(0, bufferPool.usedSizeInBytes());
}
private BufferPool.Chunk allocate(int num, int bufferSize, List<ByteBuffer> buffers)
{
for (int i = 0; i < num; i++)