From dc2acba043c6978b32a9556e0d610251d5535ce6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ko=C5=82aczkowski?= Date: Wed, 11 May 2022 11:18:34 +0200 Subject: [PATCH] Make LongBufferPoolTest insensitive to timing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit changes the way how LongBufferPoolTest verifies that all buffers were freed and recycled. Now, the delay between rounds can be arbitrarily short and timing does not affect the validity of checks. patch by Piotr Kołaczkowski; reviewed by Aleksey Yeschenko for CASSANDRA-16681 --- CHANGES.txt | 1 + .../cassandra/utils/memory/BufferPool.java | 46 ++++- .../utils/memory/LongBufferPoolTest.java | 192 +++++++++++++----- 3 files changed, 186 insertions(+), 53 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 66731f2d20..91b7d4c4fe 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.11.14 + * Make LongBufferPoolTest insensitive to timing (CASSANDRA-16681) * Suppress CVE-2022-25857 and other snakeyaml CVEs (CASSANDRA-17907) * Fix potential IndexOutOfBoundsException in PagingState in mixed mode clusters (CASSANDRA-17840) * Document usage of closed token intervals in manual compaction (CASSANDRA-17575) diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java index e91c9e2a52..557b2756e9 100644 --- a/src/java/org/apache/cassandra/utils/memory/BufferPool.java +++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java @@ -185,6 +185,18 @@ public class BufferPool globalPool.debug.check(); } + /** + * 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 + static void releaseLocal() + { + localPool.get().release(); + } + public static long sizeInBytes() { return globalPool.sizeInBytes(); @@ -192,12 +204,16 @@ public class BufferPool static final class Debug { - long recycleRound = 1; + volatile long recycleRound = 0; final Queue allChunks = new ConcurrentLinkedQueue<>(); void register(Chunk chunk) { allChunks.add(chunk); } + void acquire(Chunk chunk) + { + chunk.lastAcquired = recycleRound; + } void recycle(Chunk chunk) { chunk.lastRecycled = recycleRound; @@ -205,7 +221,8 @@ public class BufferPool void check() { for (Chunk chunk : allChunks) - assert chunk.lastRecycled == recycleRound; + assert chunk.lastRecycled >= chunk.lastAcquired; + // check is called by a single test thread, so no need for atomic here; recycleRound++; } } @@ -244,6 +261,14 @@ public class BufferPool /** Return a chunk, the caller will take owership of the parent chunk. */ public Chunk get() + { + Chunk chunk = getInternal(); + if (DEBUG && chunk != null) + debug.acquire(chunk); + return chunk; + } + + private Chunk getInternal() { while (true) { @@ -470,6 +495,20 @@ public class BufferPool } } } + + @VisibleForTesting + void release() + { + chunkCount = 0; + for (int i = 0; i < chunks.length; i++) + { + if (chunks[i] != null) + { + chunks[i].release(); + chunks[i] = null; + } + } + } } private static final class LocalPoolRef extends PhantomReference @@ -562,7 +601,8 @@ public class BufferPool // if this is set, it means the chunk may not be recycled because we may still allocate from it; // if it has been unset the local pool has finished with it, and it may be recycled private volatile LocalPool owner; - private long lastRecycled; + private volatile long lastAcquired; + private volatile long lastRecycled; private final Chunk original; Chunk(Chunk recycle) diff --git a/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java index 66abe5a92c..7bc97269e7 100644 --- a/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java +++ b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java @@ -24,14 +24,16 @@ 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.BeforeClass; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.utils.DynamicList; import static org.junit.Assert.*; @@ -60,6 +62,12 @@ import static org.junit.Assert.*; */ public class LongBufferPoolTest { + @BeforeClass + public static void setup() throws Exception + { + DatabaseDescriptor.daemonInitialization(); + } + private static final Logger logger = LoggerFactory.getLogger(LongBufferPoolTest.class); private static final int AVG_BUFFER_SIZE = 16 << 10; @@ -67,7 +75,7 @@ public class LongBufferPoolTest private static final DateFormat DATE_FORMAT = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss"); @Test - public void testAllocate() throws InterruptedException, ExecutionException + public void testAllocate() throws InterruptedException, ExecutionException, BrokenBarrierException, TimeoutException { testAllocate(Runtime.getRuntime().availableProcessors() * 2, TimeUnit.MINUTES.toNanos(2L), 16 << 20); } @@ -107,9 +115,13 @@ public class LongBufferPoolTest final long until; final CountDownLatch latch; final SPSCQueue[] 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> threadResultFuture; final int targetSizeQuanta; @@ -122,17 +134,18 @@ public class LongBufferPoolTest until = System.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); 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 @@ -150,17 +163,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() { @@ -190,9 +192,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(int threadCount, long duration, int poolSize) throws InterruptedException, ExecutionException + public void testAllocate(int threadCount, long duration, int poolSize) throws InterruptedException, ExecutionException, BrokenBarrierException, TimeoutException { System.out.println(String.format("%s - testing %d threads for %dm", DATE_FORMAT.format(new Date()), @@ -210,21 +263,30 @@ public class LongBufferPoolTest for (int threadIdx = 0; threadIdx < threadCount; threadIdx++) testEnv.addCheckedFuture(startWorkerThread(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) - BufferPool.assertAllRecycled(); - 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 + BufferPool.assertAllRecycled(); + // 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; + } } } @@ -258,24 +320,28 @@ public class LongBufferPoolTest final SPSCQueue shareFrom = testEnv.sharedRecycle[threadIdx]; final DynamicList checks = new DynamicList<>((int) Math.max(1, targetSize / (1 << 10))); final SPSCQueue shareTo = testEnv.sharedRecycle[(threadIdx + 1) % testEnv.threadCount]; + final Future 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 { + testEnv.maybeSuspendAndFreeMemory(this::freeAll); - long currentTargetSize = (rand.nextInt(testEnv.poolSize / 1024) == 0 || !testEnv.freedAllMemory[threadIdx].get()) ? 0 : targetSize; + 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) { @@ -321,9 +387,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.CHUNK_SIZE) @@ -356,6 +419,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) @@ -427,6 +513,8 @@ public class LongBufferPoolTest private void startBurnerThreads(TestEnvironment testEnv) { // setup some high churn allocate/deallocate, without any checking + + final AtomicLong pendingBuffersCount = new AtomicLong(0); final SPSCQueue burn = new SPSCQueue<>(); final CountDownLatch doneAdd = new CountDownLatch(1); testEnv.addCheckedFuture(testEnv.executorService.submit(new TestUntil(testEnv.until) @@ -437,11 +525,12 @@ public class LongBufferPoolTest { if (count * BufferPool.CHUNK_SIZE >= testEnv.poolSize / 10) { - if (burn.exhausted) + if (pendingBuffersCount.get() == 0) { count = 0; - testEnv.burnFreed.compareAndSet(false, true); - } else + testEnv.maybeSuspendAndFreeMemory(BufferPool::releaseLocal); + } + else { Thread.yield(); } @@ -460,8 +549,10 @@ public class LongBufferPoolTest if (rand.nextBoolean()) BufferPool.put(buffer); else + { + pendingBuffersCount.incrementAndGet(); burn.add(buffer); - + } count++; } void cleanup() @@ -480,6 +571,7 @@ public class LongBufferPoolTest return; } BufferPool.put(buffer); + pendingBuffersCount.decrementAndGet(); } void cleanup() {