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 trunk

This commit is contained in:
Bereng 2021-12-03 11:35:56 +01:00
parent 8e28dc0eba 15eaa95f09
commit 7fd20bff2b
1 changed files with 25 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
test/unit/org/apache/cassandra/utils/RecomputingSupplierTest.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.utils; package org.apache.cassandra.utils;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
@ -28,23 +27,24 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport; import java.util.concurrent.locks.LockSupport;
import java.util.function.LongBinaryOperator;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import org.apache.cassandra.Util;
public class RecomputingSupplierTest public class RecomputingSupplierTest
{ {
// This test case verifies that recomputing supplier never returns out of order values during concurrent updates and
// eventually returns the most recent value.
@Test @Test
public void recomputingSupplierTest() throws Throwable public void recomputingSupplierTest() throws Throwable
{ {
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10); ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);
ExecutorService testExecutor = Executors.newFixedThreadPool(10); ExecutorService testExecutor = Executors.newFixedThreadPool(10);
AtomicReference<Throwable> thrown = new AtomicReference<>(); AtomicReference<Throwable> thrown = new AtomicReference<>();
CountDownLatch latch = new CountDownLatch(1);
final AtomicLong counter = new AtomicLong(0); final AtomicLong counter = new AtomicLong(0);
final RecomputingSupplier<Long> supplier = new RecomputingSupplier<>(() -> { final RecomputingSupplier<Long> supplier = new RecomputingSupplier<>(() -> {
try try
{ {
@ -57,7 +57,6 @@ public class RecomputingSupplierTest
catch (Throwable e) catch (Throwable e)
{ {
thrown.set(e); thrown.set(e);
latch.countDown();
throw new RuntimeException(e); throw new RuntimeException(e);
} }
}, executor); }, executor);
@ -73,7 +72,6 @@ public class RecomputingSupplierTest
catch (Throwable e) catch (Throwable e)
{ {
thrown.set(e); thrown.set(e);
latch.countDown();
} }
}); });
} }
@ -87,37 +85,46 @@ public class RecomputingSupplierTest
try try
{ {
long seenBeforeGet = lastSeen.get(); long seenBeforeGet = lastSeen.get();
Long v = supplier.get(1, TimeUnit.SECONDS); Long v = supplier.get(5, TimeUnit.SECONDS);
if (v != null) if (v != null)
{ {
lastSeen.accumulateAndGet(v, Math::max); lastSeen.accumulateAndGet(v, Math::max);
Assert.assertTrue(String.format("Seen %d out of order. Last seen value %d", v, seenBeforeGet), Assert.assertTrue(String.format("Seen %d out of order. Last seen value %d", v, seenBeforeGet),
v >= seenBeforeGet); v >= seenBeforeGet);
} }
} }
catch (Throwable e) catch (Throwable e)
{ {
thrown.set(e); thrown.set(e);
latch.countDown();
} }
} }
}); });
} }
latch.await(10, TimeUnit.SECONDS); Util.spinAssertEquals(true, () -> counter.get() > 1000, 30);
testExecutor.shutdown(); testExecutor.shutdown();
testExecutor.awaitTermination(10, TimeUnit.SECONDS); Assert.assertTrue(testExecutor.awaitTermination(30, TimeUnit.SECONDS));
executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);
if (thrown.get() != null) if (thrown.get() != null)
throw new AssertionError(supplier.toString(), thrown.get()); throw new AssertionError(supplier.toString(), thrown.get());
Assert.assertTrue(counter.get() > 1); // We actually did some work Util.spinAssertEquals(true, () -> {
Assert.assertEquals(supplier.get(1, TimeUnit.SECONDS).longValue(), counter.get()); try
{
return supplier.get(1, TimeUnit.SECONDS) == counter.get();
}
catch (InterruptedException | ExecutionException | TimeoutException e)
{
e.printStackTrace();
return false;
}
}, 10);
executor.shutdown();
Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
if (thrown.get() != null)
throw new AssertionError(supplier.toString(), thrown.get());
} }
@Test @Test