diff --git a/build.xml b/build.xml
index 28c33bf3e4..3d3014ce59 100644
--- a/build.xml
+++ b/build.xml
@@ -1451,6 +1451,15 @@
+
+
+
+
+
+
+
@@ -1858,6 +1867,31 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java
index 17ac5693e6..57aa940abd 100644
--- a/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java
+++ b/test/burn/org/apache/cassandra/utils/memory/LongBufferPoolTest.java
@@ -27,19 +27,53 @@ import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
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.*;
+/**
+ * Long BufferPool test - make sure that the BufferPool allocates and recycles
+ * ByteBuffers under heavy concurrent usage.
+ *
+ * The test creates two groups of threads
+ *
+ * - the burn producer/consumer pair that allocates 1/10 poolSize and then returns
+ * all the memory to the pool. 50% is freed by the producer, 50% passed to the consumer thread.
+ *
+ * - a ring of worker threads that allocate buffers and either immediately free them,
+ * or pass to the next worker thread for it to be freed on it's behalf. Periodically
+ * all memory is freed by the thread.
+ *
+ * While the burn/worker threads run, the original main thread checks that all of the threads are still
+ * making progress every 10s (no locking issues, or exits from assertion failures),
+ * and that every chunk has been freed at least once during the previous cycle (if that was possible).
+ *
+ * The test does not expect to survive out-of-memory errors, so needs sufficient heap memory
+ * for non-direct buffers and the debug tracking objects that check the allocate buffers.
+ * (The timing is very interesting when Xmx is lowered to increase garbage collection pauses, but do
+ * not set it too low).
+ */
public class LongBufferPoolTest
{
private static final Logger logger = LoggerFactory.getLogger(LongBufferPoolTest.class);
+ private static final int AVG_BUFFER_SIZE = 16 << 10;
+ private static final int STDEV_BUFFER_SIZE = 10 << 10; // picked to ensure exceeding buffer size is rare, but occurs
+ private static final DateFormat DATE_FORMAT = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
+
+ @BeforeClass
+ public static void setup() throws Exception
+ {
+ DatabaseDescriptor.daemonInitialization();
+ }
+
@Test
public void testAllocate() throws InterruptedException, ExecutionException
{
@@ -73,282 +107,136 @@ public class LongBufferPoolTest
}
}
- public void testAllocate(int threadCount, long duration, int poolSize) throws InterruptedException, ExecutionException
+ private static final class TestEnvironment
{
- final int avgBufferSize = 16 << 10;
- final int stdevBufferSize = 10 << 10; // picked to ensure exceeding buffer size is rare, but occurs
- final DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
+ final int threadCount;
+ final long duration;
+ final int poolSize;
+ final long until;
+ final CountDownLatch latch;
+ final SPSCQueue[] sharedRecycle;
+ final AtomicBoolean[] makingProgress;
+ final AtomicBoolean burnFreed;
+ final AtomicBoolean[] freedAllMemory;
+ final ExecutorService executorService;
+ final List> threadResultFuture;
+ final int targetSizeQuanta;
- System.out.println(String.format("%s - testing %d threads for %dm",
- dateFormat.format(new Date()),
- threadCount,
- TimeUnit.NANOSECONDS.toMinutes(duration)));
-
- final long until = System.nanoTime() + duration;
- final CountDownLatch latch = new CountDownLatch(threadCount);
- final SPSCQueue[] sharedRecycle = new SPSCQueue[threadCount];
- final AtomicBoolean[] makingProgress = new AtomicBoolean[threadCount];
- for (int i = 0 ; i < sharedRecycle.length ; i++)
+ TestEnvironment(int threadCount, long duration, int poolSize)
{
- sharedRecycle[i] = new SPSCQueue<>();
- makingProgress[i] = new AtomicBoolean(true);
+ this.threadCount = threadCount;
+ this.duration = duration;
+ this.poolSize = poolSize;
+ until = System.nanoTime() + duration;
+ latch = new CountDownLatch(threadCount);
+ sharedRecycle = new SPSCQueue[threadCount];
+ makingProgress = new AtomicBoolean[threadCount];
+ burnFreed = new AtomicBoolean(false);
+ freedAllMemory = new AtomicBoolean[threadCount];
+ 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
+ // allocate a different amount of memory each - 1*quanta, 2*quanta, ... N*quanta.
+ // Thread0 is always going to be a single CHUNK, then to allocate increasing amounts
+ // using their own algorithm the targetSize should be poolSize / targetSizeQuanta.
+ //
+ // This should divide double the poolSize across the working threads,
+ // plus CHUNK_SIZE for thread0 and 1/10 poolSize for the burn producer/consumer pair.
+ targetSizeQuanta = 2 * poolSize / sum1toN(threadCount - 1);
}
- ExecutorService executorService = Executors.newFixedThreadPool(threadCount + 2);
- List> ret = new ArrayList<>(threadCount);
- long prevPoolSize = BufferPool.MEMORY_USAGE_THRESHOLD;
- BufferPool.MEMORY_USAGE_THRESHOLD = poolSize;
- BufferPool.DEBUG = true;
- // sum(1..n) = n/2 * (n + 1); we set zero to CHUNK_SIZE, so have n=threadCount-1
- int targetSizeQuanta = ((threadCount) * (threadCount - 1)) / 2;
- // fix targetSizeQuanta at 1/64th our poolSize, so that we only consciously exceed our pool size limit
- targetSizeQuanta = (targetSizeQuanta * poolSize) / 64;
-
+ void addCheckedFuture(Future future)
{
- // setup some high churn allocate/deallocate, without any checking
- final SPSCQueue burn = new SPSCQueue<>();
- final CountDownLatch doneAdd = new CountDownLatch(1);
- executorService.submit(new TestUntil(until)
- {
- int count = 0;
- void testOne() throws Exception
- {
- if (count * BufferPool.CHUNK_SIZE >= poolSize / 10)
- {
- if (burn.exhausted)
- count = 0;
- else
- Thread.yield();
- return;
- }
-
- ByteBuffer buffer = BufferPool.tryGet(BufferPool.CHUNK_SIZE);
- if (buffer == null)
- {
- Thread.yield();
- return;
- }
-
- BufferPool.put(buffer);
- burn.add(buffer);
- count++;
- }
- void cleanup()
- {
- doneAdd.countDown();
- }
- });
- executorService.submit(new TestUntil(until)
- {
- void testOne() throws Exception
- {
- ByteBuffer buffer = burn.poll();
- if (buffer == null)
- {
- Thread.yield();
- return;
- }
- BufferPool.put(buffer);
- }
- void cleanup()
- {
- Uninterruptibles.awaitUninterruptibly(doneAdd);
- }
- });
+ threadResultFuture.add(future);
}
- for (int t = 0; t < threadCount; t++)
+ int countStalledThreads()
{
- final int threadIdx = t;
- final int targetSize = t == 0 ? BufferPool.CHUNK_SIZE : targetSizeQuanta * t;
+ int stalledThreads = 0;
- ret.add(executorService.submit(new TestUntil(until)
- {
- final SPSCQueue shareFrom = sharedRecycle[threadIdx];
- final DynamicList checks = new DynamicList<>((int) Math.max(1, targetSize / (1 << 10)));
- final SPSCQueue shareTo = sharedRecycle[(threadIdx + 1) % threadCount];
- final ThreadLocalRandom rand = ThreadLocalRandom.current();
- int totalSize = 0;
- int freeingSize = 0;
- int size = 0;
-
- void checkpoint()
- {
- if (!makingProgress[threadIdx].get())
- makingProgress[threadIdx].set(true);
- }
-
- void testOne() throws Exception
- {
-
- long currentTargetSize = rand.nextInt(poolSize / 1024) == 0 ? 0 : targetSize;
- int spinCount = 0;
- while (totalSize > currentTargetSize - freeingSize)
- {
- // free buffers until we're below our target size
- if (checks.size() == 0)
- {
- // if we're out of buffers to free, we're waiting on our neighbour to free them;
- // first check if the consuming neighbour has caught up, and if so mark that free
- if (shareTo.exhausted)
- {
- totalSize -= freeingSize;
- freeingSize = 0;
- }
- else if (!recycleFromNeighbour())
- {
- if (++spinCount > 1000 && System.nanoTime() > until)
- return;
- // otherwise, free one of our other neighbour's buffers if can; and otherwise yield
- Thread.yield();
- }
- continue;
- }
-
- // pick a random buffer, with preference going to earlier ones
- BufferCheck check = sample();
- checks.remove(check.listnode);
- check.validate();
-
- size = BufferPool.roundUpNormal(check.buffer.capacity());
- if (size > BufferPool.CHUNK_SIZE)
- size = 0;
-
- // either share to free, or free immediately
- if (rand.nextBoolean())
- {
- shareTo.add(check);
- freeingSize += size;
- // interleave this with potentially messing with the other neighbour's stuff
- recycleFromNeighbour();
- }
- else
- {
- check.validate();
- BufferPool.put(check.buffer);
- totalSize -= size;
- }
- }
-
- // allocate a new buffer
- size = (int) Math.max(1, avgBufferSize + (stdevBufferSize * rand.nextGaussian()));
- if (size <= BufferPool.CHUNK_SIZE)
- {
- totalSize += BufferPool.roundUpNormal(size);
- allocate(size);
- }
- else if (rand.nextBoolean())
- {
- allocate(size);
- }
- else
- {
- // perform a burst allocation to exhaust all available memory
- while (totalSize < poolSize)
- {
- size = (int) Math.max(1, avgBufferSize + (stdevBufferSize * rand.nextGaussian()));
- if (size <= BufferPool.CHUNK_SIZE)
- {
- allocate(size);
- totalSize += BufferPool.roundUpNormal(size);
- }
- }
- }
-
- // validate a random buffer we have stashed
- checks.get(rand.nextInt(checks.size())).validate();
-
- // free all of our neighbour's remaining shared buffers
- while (recycleFromNeighbour());
- }
-
- void cleanup()
- {
- while (checks.size() > 0)
- {
- BufferCheck check = checks.get(0);
- BufferPool.put(check.buffer);
- checks.remove(check.listnode);
- }
- latch.countDown();
- }
-
- boolean recycleFromNeighbour()
- {
- BufferCheck check = shareFrom.poll();
- if (check == null)
- return false;
- check.validate();
- BufferPool.put(check.buffer);
- return true;
- }
-
- BufferCheck allocate(int size)
- {
- ByteBuffer buffer = BufferPool.get(size);
- assertNotNull(buffer);
- BufferCheck check = new BufferCheck(buffer, rand.nextLong());
- assertEquals(size, buffer.capacity());
- assertEquals(0, buffer.position());
- check.init();
- check.listnode = checks.append(check);
- return check;
- }
-
- BufferCheck sample()
- {
- // sample with preference to first elements:
- // element at index n will be selected with likelihood (size - n) / sum1ToN(size)
- int size = checks.size();
-
- // pick a random number between 1 and sum1toN(size)
- int sampleRange = sum1toN(size);
- int sampleIndex = rand.nextInt(sampleRange);
-
- // then binary search for the N, such that [sum1ToN(N), sum1ToN(N+1)) contains this random number
- int moveBy = Math.max(size / 4, 1);
- int index = size / 2;
- while (true)
- {
- int baseSampleIndex = sum1toN(index);
- int endOfSampleIndex = sum1toN(index + 1);
- if (sampleIndex >= baseSampleIndex)
- {
- if (sampleIndex < endOfSampleIndex)
- break;
- index += moveBy;
- }
- else index -= moveBy;
- moveBy = Math.max(moveBy / 2, 1);
- }
-
- // this gives us the inverse of our desired value, so just subtract it from the last index
- index = size - (index + 1);
-
- return checks.get(index);
- }
-
- private int sum1toN(int n)
- {
- return (n * (n + 1)) / 2;
- }
- }));
- }
-
- boolean first = true;
- while (!latch.await(10L, TimeUnit.SECONDS))
- {
- if (!first)
- BufferPool.assertAllRecycled();
- first = false;
for (AtomicBoolean progress : makingProgress)
{
- assert progress.get();
- progress.set(false);
+ if (!progress.getAndSet(false))
+ stalledThreads++;
+ }
+ return stalledThreads;
+ }
+
+ int countDoneThreads()
+ {
+ int doneThreads = 0;
+ for (Future r : threadResultFuture)
+ {
+ if (r.isDone())
+ doneThreads++;
+ }
+ return doneThreads;
+ }
+
+ void assertCheckedThreadsSucceeded()
+ {
+ try
+ {
+ for (Future r : threadResultFuture)
+ assertTrue(r.get());
+ }
+ catch (InterruptedException ex)
+ {
+ // If interrupted while checking, restart and check everything.
+ assertCheckedThreadsSucceeded();
+ }
+ catch (ExecutionException ex)
+ {
+ fail("Checked thread threw exception: " + ex.toString());
+ }
+ }
+ }
+
+ public void testAllocate(int threadCount, long duration, int poolSize) throws InterruptedException, ExecutionException
+ {
+ System.out.println(String.format("%s - testing %d threads for %dm",
+ DATE_FORMAT.format(new Date()),
+ threadCount,
+ TimeUnit.NANOSECONDS.toMinutes(duration)));
+ long prevPoolSize = BufferPool.MEMORY_USAGE_THRESHOLD;
+ logger.info("Overriding configured BufferPool.MEMORY_USAGE_THRESHOLD={} and enabling BufferPool.DEBUG", poolSize);
+ BufferPool.MEMORY_USAGE_THRESHOLD = poolSize;
+ BufferPool.DEBUG = true;
+
+ TestEnvironment testEnv = new TestEnvironment(threadCount, duration, poolSize);
+
+ startBurnerThreads(testEnv);
+
+ for (int threadIdx = 0; threadIdx < threadCount; threadIdx++)
+ testEnv.addCheckedFuture(startWorkerThread(testEnv, threadIdx));
+
+ while (!testEnv.latch.await(10L, 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");
}
}
- for (SPSCQueue queue : sharedRecycle)
+ for (SPSCQueue queue : testEnv.sharedRecycle)
{
BufferCheck check;
while ( null != (check = queue.poll()) )
@@ -358,14 +246,254 @@ public class LongBufferPoolTest
}
}
- assertEquals(0, executorService.shutdownNow().size());
+ assertEquals(0, testEnv.executorService.shutdownNow().size());
+ logger.info("Reverting BufferPool.MEMORY_USAGE_THRESHOLD={}", prevPoolSize);
BufferPool.MEMORY_USAGE_THRESHOLD = prevPoolSize;
- for (Future r : ret)
- assertTrue(r.get());
+ BufferPool.DEBUG = false;
+
+ testEnv.assertCheckedThreadsSucceeded();
System.out.println(String.format("%s - finished.",
- dateFormat.format(new Date())));
+ DATE_FORMAT.format(new Date())));
+ }
+
+ private Future startWorkerThread(TestEnvironment testEnv, final int threadIdx)
+ {
+ return testEnv.executorService.submit(new TestUntil(testEnv.until)
+ {
+ final int targetSize = threadIdx == 0 ? BufferPool.CHUNK_SIZE : testEnv.targetSizeQuanta * threadIdx;
+ 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 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;
+ int spinCount = 0;
+ while (totalSize > currentTargetSize - freeingSize)
+ {
+ // free buffers until we're below our target size
+ if (checks.size() == 0)
+ {
+ // if we're out of buffers to free, we're waiting on our neighbour to free them;
+ // first check if the consuming neighbour has caught up, and if so mark that free
+ if (shareTo.exhausted)
+ {
+ totalSize -= freeingSize;
+ freeingSize = 0;
+ }
+ else if (!recycleFromNeighbour())
+ {
+ if (++spinCount > 1000 && System.nanoTime() > until)
+ return;
+ // otherwise, free one of our other neighbour's buffers if can; and otherwise yield
+ Thread.yield();
+ }
+ continue;
+ }
+
+ // pick a random buffer, with preference going to earlier ones
+ BufferCheck check = sample();
+ checks.remove(check.listnode);
+ check.validate();
+
+ size = BufferPool.roundUpNormal(check.buffer.capacity());
+ if (size > BufferPool.CHUNK_SIZE)
+ size = 0;
+
+ // either share to free, or free immediately
+ if (rand.nextBoolean())
+ {
+ shareTo.add(check);
+ freeingSize += size;
+ // interleave this with potentially messing with the other neighbour's stuff
+ recycleFromNeighbour();
+ }
+ else
+ {
+ check.validate();
+ BufferPool.put(check.buffer);
+ totalSize -= size;
+ }
+ }
+
+ 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)
+ {
+ totalSize += BufferPool.roundUpNormal(size);
+ allocate(size);
+ }
+ else if (rand.nextBoolean())
+ {
+ allocate(size);
+ }
+ else
+ {
+ // perform a burst allocation to exhaust all available memory
+ while (totalSize < testEnv.poolSize)
+ {
+ size = (int) Math.max(1, AVG_BUFFER_SIZE + (STDEV_BUFFER_SIZE * rand.nextGaussian()));
+ if (size <= BufferPool.CHUNK_SIZE)
+ {
+ allocate(size);
+ totalSize += BufferPool.roundUpNormal(size);
+ }
+ }
+ }
+
+ // validate a random buffer we have stashed
+ checks.get(rand.nextInt(checks.size())).validate();
+
+ // free all of our neighbour's remaining shared buffers
+ while (recycleFromNeighbour());
+ }
+
+ void cleanup()
+ {
+ while (checks.size() > 0)
+ {
+ BufferCheck check = checks.get(0);
+ BufferPool.put(check.buffer);
+ checks.remove(check.listnode);
+ }
+ testEnv.latch.countDown();
+ }
+
+ boolean recycleFromNeighbour()
+ {
+ BufferCheck check = shareFrom.poll();
+ if (check == null)
+ return false;
+ check.validate();
+ BufferPool.put(check.buffer);
+ return true;
+ }
+
+ BufferCheck allocate(int size)
+ {
+ ByteBuffer buffer = BufferPool.get(size);
+ assertNotNull(buffer);
+ BufferCheck check = new BufferCheck(buffer, rand.nextLong());
+ assertEquals(size, buffer.capacity());
+ assertEquals(0, buffer.position());
+ check.init();
+ check.listnode = checks.append(check);
+ return check;
+ }
+
+ BufferCheck sample()
+ {
+ // sample with preference to first elements:
+ // element at index n will be selected with likelihood (size - n) / sum1ToN(size)
+ int size = checks.size();
+
+ // pick a random number between 1 and sum1toN(size)
+ int sampleRange = sum1toN(size);
+ int sampleIndex = rand.nextInt(sampleRange);
+
+ // then binary search for the N, such that [sum1ToN(N), sum1ToN(N+1)) contains this random number
+ int moveBy = Math.max(size / 4, 1);
+ int index = size / 2;
+ while (true)
+ {
+ int baseSampleIndex = sum1toN(index);
+ int endOfSampleIndex = sum1toN(index + 1);
+ if (sampleIndex >= baseSampleIndex)
+ {
+ if (sampleIndex < endOfSampleIndex)
+ break;
+ index += moveBy;
+ }
+ else index -= moveBy;
+ moveBy = Math.max(moveBy / 2, 1);
+ }
+
+ // this gives us the inverse of our desired value, so just subtract it from the last index
+ index = size - (index + 1);
+
+ return checks.get(index);
+ }
+ });
+ }
+
+ private void startBurnerThreads(TestEnvironment testEnv)
+ {
+ // setup some high churn allocate/deallocate, without any checking
+ final SPSCQueue burn = new SPSCQueue<>();
+ final CountDownLatch doneAdd = new CountDownLatch(1);
+ testEnv.addCheckedFuture(testEnv.executorService.submit(new TestUntil(testEnv.until)
+ {
+ int count = 0;
+ final ThreadLocalRandom rand = ThreadLocalRandom.current();
+ void testOne() throws Exception
+ {
+ if (count * BufferPool.CHUNK_SIZE >= testEnv.poolSize / 10)
+ {
+ if (burn.exhausted)
+ {
+ count = 0;
+ testEnv.burnFreed.compareAndSet(false, true);
+ } else
+ {
+ Thread.yield();
+ }
+ return;
+ }
+
+ ByteBuffer buffer = BufferPool.tryGet(BufferPool.CHUNK_SIZE);
+ if (buffer == null)
+ {
+ Thread.yield();
+ return;
+ }
+
+ // 50/50 chance of returning the buffer from the producer thread, or
+ // pass it on to the consumer.
+ if (rand.nextBoolean())
+ BufferPool.put(buffer);
+ else
+ burn.add(buffer);
+
+ count++;
+ }
+ void cleanup()
+ {
+ doneAdd.countDown();
+ }
+ }));
+ testEnv.threadResultFuture.add(testEnv.executorService.submit(new TestUntil(testEnv.until)
+ {
+ void testOne() throws Exception
+ {
+ ByteBuffer buffer = burn.poll();
+ if (buffer == null)
+ {
+ Thread.yield();
+ return;
+ }
+ BufferPool.put(buffer);
+ }
+ void cleanup()
+ {
+ Uninterruptibles.awaitUninterruptibly(doneAdd);
+ }
+ }));
}
static abstract class TestUntil implements Callable
@@ -399,6 +527,14 @@ public class LongBufferPoolTest
ex.printStackTrace();
return false;
}
+ catch (Throwable tr) // for java.lang.OutOfMemoryError
+ {
+ logger.error("Got throwable {}, current chunk {}",
+ tr.getMessage(),
+ BufferPool.currentChunk());
+ tr.printStackTrace();
+ return false;
+ }
finally
{
cleanup();
@@ -407,9 +543,20 @@ public class LongBufferPoolTest
}
}
- public static void main(String[] args) throws InterruptedException, ExecutionException
+ public static void main(String[] args)
{
- new LongBufferPoolTest().testAllocate(Runtime.getRuntime().availableProcessors(), TimeUnit.HOURS.toNanos(2L), 16 << 20);
+ try
+ {
+ LongBufferPoolTest.setup();
+ new LongBufferPoolTest().testAllocate(Runtime.getRuntime().availableProcessors(),
+ TimeUnit.HOURS.toNanos(2L), 16 << 20);
+ System.exit(0);
+ }
+ catch (Throwable tr)
+ {
+ System.out.println(String.format("Test failed - %s", tr.getMessage()));
+ System.exit(1); // Force exit so that non-daemon threads like REQUEST-SCHEDULER do not hang the process on failure
+ }
}
/**
@@ -451,4 +598,8 @@ public class LongBufferPoolTest
}
}
-}
\ No newline at end of file
+ private static int sum1toN(int n)
+ {
+ return (n * (n + 1)) / 2;
+ }
+}