diff --git a/CHANGES.txt b/CHANGES.txt index 3e7cebdfaa..29ad1dfb02 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -6,6 +6,7 @@ * Avoid getting hanging repairs due to repair message timeouts (CASSANDRA-17613) * Prevent infinite loop in repair coordinator on FailSession (CASSANDRA-17834) Merged from 3.11: + * 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) Merged from 3.0: diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java index 531b492377..23ca516444 100644 --- a/src/java/org/apache/cassandra/utils/memory/BufferPool.java +++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java @@ -286,12 +286,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 @@ -299,6 +314,7 @@ public class BufferPool }; void registerNormal(Chunk chunk); + void acquire(Chunk chunk); void recycleNormal(Chunk oldVersion, Chunk newVersion); void recyclePartial(Chunk chunk); } @@ -362,6 +378,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) diff --git a/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java index fc603c9d76..117ee7c064 100644 --- a/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java +++ b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java @@ -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; @@ -75,7 +75,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) @@ -83,7 +84,7 @@ public class LongBufferPoolTest return (DebugChunk) chunk.debugAttachment; } } - long recycleRound = 1; + AtomicLong recycleRound = new AtomicLong(1); final List normalChunks = new ArrayList<>(); public synchronized void registerNormal(BufferPool.Chunk chunk) @@ -91,20 +92,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(); } } @@ -122,18 +142,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)); @@ -174,9 +194,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; @@ -189,17 +213,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, 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 @@ -217,17 +242,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() { @@ -257,9 +271,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()); @@ -273,21 +338,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; + } } } @@ -319,23 +395,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 { - 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) { @@ -381,9 +462,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) @@ -416,6 +494,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) @@ -487,6 +588,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 burn = new SPSCQueue<>(); final CountDownLatch doneAdd = new CountDownLatch(1); testEnv.addCheckedFuture(testEnv.executorService.submit(new TestUntil(bufferPool, testEnv.until) @@ -497,10 +600,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(); @@ -522,8 +625,10 @@ public class LongBufferPoolTest if (rand.nextBoolean()) bufferPool.put(buffer); else + { + pendingBuffersCount.incrementAndGet(); burn.add(buffer); - + } count++; } void cleanup() @@ -542,6 +647,7 @@ public class LongBufferPoolTest return; } bufferPool.put(buffer); + pendingBuffersCount.decrementAndGet(); } void cleanup() { diff --git a/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java b/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java index eb3cc1b47a..67e1b9ff64 100644 --- a/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java +++ b/test/unit/org/apache/cassandra/utils/memory/BufferPoolTest.java @@ -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 buffers0 = new ArrayList<>(); + BufferPool.Chunk chunk0 = allocate(allocationPerChunk, size, buffers0); + List buffers1 = new ArrayList<>(); + BufferPool.Chunk chunk1 = allocate(allocationPerChunk, size, buffers1); + List 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 buffers) { for (int i = 0; i < num; i++)