mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
Local updates couldn't be late
This commit is contained in:
leventov
@@ -27,6 +27,7 @@ import java.io.IOException;
|
||||
import java.io.RandomAccessFile;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.Arrays;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
@@ -60,6 +61,18 @@ public class Builder {
|
||||
return file;
|
||||
}
|
||||
|
||||
public static void waitTillEqual(Map map1, Map map2, int timeOut) {
|
||||
for (int t = 0; t < timeOut; t++) {
|
||||
if (map1.equals(map2))
|
||||
break;
|
||||
try {
|
||||
Thread.sleep(1);
|
||||
} catch (InterruptedException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static <T extends ChronicleMap<Integer, DummyValue>> T newMapDummyValue(
|
||||
final byte identifier,
|
||||
final int serverPort,
|
||||
|
||||
@@ -1,255 +0,0 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.chronicle.hash.replication.TimeProvider;
|
||||
import net.openhft.lang.io.ByteBufferBytes;
|
||||
import net.openhft.lang.model.Byteable;
|
||||
import net.openhft.lang.model.DataValueClasses;
|
||||
import net.openhft.lang.values.IntValue;
|
||||
import org.junit.After;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.Random;
|
||||
import java.util.Set;
|
||||
import java.util.TreeMap;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public class NoTcpReplicationSoakTest {
|
||||
|
||||
private ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?> map1;
|
||||
private ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?> map2;
|
||||
private IntValue value;
|
||||
static int s_port = 8010;
|
||||
private long time;
|
||||
|
||||
TimeProvider timeProvider = new TimeProvider() {
|
||||
@Override
|
||||
public long currentTime() {
|
||||
return time;
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@Before
|
||||
public void setup() throws IOException {
|
||||
value = DataValueClasses.newDirectReference(IntValue.class);
|
||||
((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0);
|
||||
|
||||
|
||||
{
|
||||
|
||||
|
||||
map1 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class)
|
||||
.replication((byte) 1)
|
||||
.timeProvider(timeProvider)
|
||||
.instance()
|
||||
.name("map1")
|
||||
.create();
|
||||
}
|
||||
{
|
||||
|
||||
map2 = (ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class)
|
||||
.replication((byte) 2)
|
||||
.timeProvider(timeProvider)
|
||||
.instance()
|
||||
.name("map2")
|
||||
.create();
|
||||
|
||||
}
|
||||
s_port += 2;
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
@After
|
||||
public void tearDown() throws InterruptedException {
|
||||
|
||||
for (final Closeable closeable : new Closeable[]{map1, map2}) {
|
||||
try {
|
||||
closeable.close();
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
System.gc();
|
||||
}
|
||||
|
||||
Set<Thread> threads;
|
||||
|
||||
@Before
|
||||
public void sampleThreads() {
|
||||
threads = Thread.getAllStackTraces().keySet();
|
||||
}
|
||||
|
||||
@After
|
||||
public void checkThreadsShutdown() {
|
||||
StatelessClientTest.checkThreadsShutdown(threads);
|
||||
}
|
||||
|
||||
AtomicInteger task = new AtomicInteger();
|
||||
|
||||
@Test
|
||||
public void testSoakTestWithRandomData() throws IOException, InterruptedException {
|
||||
ExecutorService executorService = Executors.newSingleThreadExecutor();
|
||||
System.out.print("SoakTesting ");
|
||||
for (int j = 1; j < 10000; j++) {
|
||||
if (j % 100 == 0)
|
||||
System.out.print(".");
|
||||
|
||||
Random rnd = new Random(j);
|
||||
|
||||
|
||||
if (rnd.nextInt(10) < 2)
|
||||
time++;
|
||||
|
||||
final long t = time;
|
||||
|
||||
final int key = rnd.nextInt(1000);
|
||||
|
||||
// add random delay
|
||||
addRandomDelay(executorService, rnd);
|
||||
|
||||
if (rnd.nextBoolean()) {
|
||||
final CharSequence value = "test" + j;
|
||||
if (rnd.nextBoolean()) {
|
||||
try (MapKeyContext<?, CharSequence> c = map1.context(key)) {
|
||||
ReplicatedChronicleMap.ReplicatedContext rc =
|
||||
(ReplicatedChronicleMap.ReplicatedContext) c;
|
||||
rc.newTimestamp = t;
|
||||
rc.newIdentifier = (byte) 1;
|
||||
c.put(value);
|
||||
}
|
||||
|
||||
task.incrementAndGet();
|
||||
executorService.submit(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
try (MapKeyContext<?, CharSequence> c = map2.context(key)) {
|
||||
ReplicatedChronicleMap.ReplicatedContext rc =
|
||||
(ReplicatedChronicleMap.ReplicatedContext) c;
|
||||
rc.newTimestamp = t;
|
||||
rc.newIdentifier = (byte) 1;
|
||||
c.put(value);
|
||||
}
|
||||
task.decrementAndGet();
|
||||
}
|
||||
});
|
||||
|
||||
} else {
|
||||
try (MapKeyContext<?, CharSequence> c = map2.context(key)) {
|
||||
ReplicatedChronicleMap.ReplicatedContext rc =
|
||||
(ReplicatedChronicleMap.ReplicatedContext) c;
|
||||
rc.newTimestamp = t;
|
||||
rc.newIdentifier = (byte) 2;
|
||||
c.put(value);
|
||||
}
|
||||
|
||||
task.incrementAndGet();
|
||||
executorService.submit(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
try (MapKeyContext<?, CharSequence> c = map1.context(key)) {
|
||||
ReplicatedChronicleMap.ReplicatedContext rc =
|
||||
(ReplicatedChronicleMap.ReplicatedContext) c;
|
||||
rc.newTimestamp = t;
|
||||
rc.newIdentifier = (byte) 2;
|
||||
c.put(value);
|
||||
}
|
||||
task.decrementAndGet();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
} else {
|
||||
|
||||
if (rnd.nextBoolean()) {
|
||||
try (MapKeyContext<?, CharSequence> c = map1.context(key)) {
|
||||
ReplicatedChronicleMap.ReplicatedContext rc =
|
||||
(ReplicatedChronicleMap.ReplicatedContext) c;
|
||||
rc.newTimestamp = t;
|
||||
rc.newIdentifier = (byte) 1;
|
||||
c.remove();
|
||||
}
|
||||
|
||||
task.incrementAndGet();
|
||||
executorService.submit(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
try (MapKeyContext<?, CharSequence> c = map2.context(key)) {
|
||||
ReplicatedChronicleMap.ReplicatedContext rc =
|
||||
(ReplicatedChronicleMap.ReplicatedContext) c;
|
||||
rc.newTimestamp = t;
|
||||
rc.newIdentifier = (byte) 1;
|
||||
c.remove();
|
||||
}
|
||||
task.decrementAndGet();
|
||||
}
|
||||
});
|
||||
|
||||
} else {
|
||||
try (MapKeyContext<?, CharSequence> c = map2.context(key)) {
|
||||
ReplicatedChronicleMap.ReplicatedContext rc =
|
||||
(ReplicatedChronicleMap.ReplicatedContext) c;
|
||||
rc.newTimestamp = t;
|
||||
rc.newIdentifier = (byte) 2;
|
||||
c.remove();
|
||||
}
|
||||
|
||||
task.incrementAndGet();
|
||||
executorService.submit(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
try (MapKeyContext<?, CharSequence> c = map1.context(key)) {
|
||||
ReplicatedChronicleMap.ReplicatedContext rc =
|
||||
(ReplicatedChronicleMap.ReplicatedContext) c;
|
||||
rc.newTimestamp = t;
|
||||
rc.newIdentifier = (byte) 2;
|
||||
c.remove();
|
||||
}
|
||||
task.decrementAndGet();
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
while (task.get() > 0) {
|
||||
|
||||
}
|
||||
Assert.assertEquals(new TreeMap(map1), new TreeMap(map2));
|
||||
}
|
||||
|
||||
private void addRandomDelay(ExecutorService executorService, Random rnd) {
|
||||
if (rnd.nextInt(30) < 2) {
|
||||
|
||||
task.incrementAndGet();
|
||||
executorService.submit(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
task.decrementAndGet();
|
||||
try {
|
||||
Thread.sleep(1);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import net.openhft.chronicle.hash.replication.LateUpdateException;
|
||||
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
|
||||
import net.openhft.chronicle.hash.replication.TimeProvider;
|
||||
import net.openhft.chronicle.map.jsr166.JSR166TestCase;
|
||||
import org.junit.After;
|
||||
@@ -28,10 +28,13 @@ import org.mockito.Mockito;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.BiConsumer;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
import static net.openhft.chronicle.map.Builder.waitTillEqual;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
@@ -59,39 +62,55 @@ public class TimeBasedReplicationTest extends JSR166TestCase {
|
||||
StatelessClientTest.checkThreadsShutdown(threads);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIgnoreALatePut() throws IOException {
|
||||
public void testIgnoreLateAction(BiConsumer<ChronicleMap<CharSequence, CharSequence>,
|
||||
ChronicleMap<CharSequence, CharSequence>> action)
|
||||
throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
try (ChronicleMap<CharSequence, CharSequence> map =
|
||||
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1, TcpTransportAndNetworkConfig.of(8086))
|
||||
.create()) {
|
||||
try (ChronicleMap<CharSequence, CharSequence> map2 =
|
||||
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 2, TcpTransportAndNetworkConfig.of(8087,
|
||||
new InetSocketAddress("localhost", 8086)))
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
assertEquals(map.size(), 1);
|
||||
assertEquals(map.get("key-1"), "value-1");
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
assertEquals(map.size(), 1);
|
||||
assertEquals(map.get("key-1"), "value-1");
|
||||
|
||||
// now test assume that we receive a late update to the map, the following update should be ignored
|
||||
late(timeProvider);
|
||||
// now test assume that we receive a late update to the map,
|
||||
// the following update should be ignored
|
||||
late(timeProvider);
|
||||
|
||||
assertThrows(LateUpdateException.class, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
map.put("key-1", "value-2");
|
||||
}
|
||||
});
|
||||
action.accept(map, map2);
|
||||
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
assertEquals(map.size(), 1);
|
||||
assertEquals(map.get("key-1"), "value-1");
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
waitTillEqual(map, map2, 5000);
|
||||
assertEquals(map.size(), 1);
|
||||
// prove that late update to map2 ignored
|
||||
assertEquals(map.get("key-1"), "value-1");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIgnoreALatePut() throws IOException {
|
||||
testIgnoreLateAction((m1, m2) -> {
|
||||
m2.put("key-1", "value-2");
|
||||
assertEquals("value-2", m2.get("key-1"));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIgnoreALatePutIfAbsent() throws IOException {
|
||||
|
||||
@@ -125,268 +144,83 @@ public class TimeBasedReplicationTest extends JSR166TestCase {
|
||||
|
||||
@Test
|
||||
public void testIgnoreALateReplace() throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
|
||||
// now test assume that we receive a late update to the map,
|
||||
// the following update should be ignored
|
||||
late(timeProvider);
|
||||
|
||||
assertThrows(LateUpdateException.class, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
map.replace("key-1", "value-2");
|
||||
}
|
||||
});
|
||||
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
assertEquals(map.size(), 1);
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
}
|
||||
testIgnoreLateAction((m1, m2) -> {
|
||||
// in order to be able to call replace() "locally"
|
||||
m2.put("key-1", "dummy");
|
||||
m2.replace("key-1", "value-2");
|
||||
assertEquals("value-2", m2.get("key-1"));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIgnoreALateReplaceWithValue() throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
|
||||
// now test assume that we receive a late update to the map, the following update
|
||||
// should be ignored
|
||||
late(timeProvider);
|
||||
|
||||
assertThrows(LateUpdateException.class, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
map.replace("key-1", "value-1", "value-2");
|
||||
}
|
||||
});
|
||||
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
}
|
||||
testIgnoreLateAction((m1, m2) -> {
|
||||
// in order to be able to call replace() "locally"
|
||||
m2.put("key-1", "dummy");
|
||||
m2.replace("key-1", "dummy", "value-2");
|
||||
assertEquals("value-2", m2.get("key-1"));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIgnoreALateRemoveWithValue() throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
|
||||
// now test assume that we receive a late update to the map, the following update
|
||||
// should be ignored
|
||||
late(timeProvider);
|
||||
|
||||
assertThrows(LateUpdateException.class, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
map.remove("key-1", "value-1");
|
||||
}
|
||||
});
|
||||
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
}
|
||||
testIgnoreLateAction((m1, m2) -> {
|
||||
// in order to be able to call remove() "locally"
|
||||
m2.put("key-1", "dummy");
|
||||
m2.remove("key-1", "dummy");
|
||||
assertEquals(null, m2.get("key-1"));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIgnoreALateRemove() throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
|
||||
// now test assume that we receive a late update to the map, the following update
|
||||
// should be ignored
|
||||
late(timeProvider);
|
||||
|
||||
assertThrows(LateUpdateException.class, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
map.remove("key-1");
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIgnoreWithRemoteRemove() throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of
|
||||
(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
String key1 = "key-1";
|
||||
map.put(key1, "value-1");
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get(key1));
|
||||
|
||||
// now test assume that we receive a late update to the map,
|
||||
// the following update should be ignored
|
||||
late(timeProvider);
|
||||
assertFalse(map.remove(key1, "value-2"));
|
||||
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get(key1));
|
||||
assertTrue(map.containsValue("value-1"));
|
||||
assertFalse(map.containsValue("value-2"));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testIgnoreWithRemotePut() throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of
|
||||
(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
assertEquals(1, map.size());
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
|
||||
// now test assume that we receive a late update to the map, the following update
|
||||
// should be ignored
|
||||
final long late = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(5);
|
||||
try (MapKeyContext<?, CharSequence> c = map.context("key-1")) {
|
||||
c.writeLock().lock();
|
||||
((ReplicatedChronicleMap.ReplicatedContext) c).newIdentifier = (byte) 2;
|
||||
((ReplicatedChronicleMap.ReplicatedContext) c).newTimestamp = late;
|
||||
assertFalse(((ReplicatedChronicleMap.ReplicatedContext) c).put("value-2"));
|
||||
}
|
||||
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
|
||||
assertEquals("value-1", map.get("key-1"));
|
||||
assertEquals(1, map.size(), 0);
|
||||
}
|
||||
testIgnoreLateAction((m1, m2) -> {
|
||||
// in order to be able to call remove() "locally"
|
||||
m2.put("key-1", "dummy");
|
||||
m2.remove("key-1");
|
||||
assertEquals(null, m2.get("key-1"));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRemoveFollowedByLatePut() throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of
|
||||
(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
try (ChronicleMap<CharSequence, CharSequence> map =
|
||||
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1, TcpTransportAndNetworkConfig.of(8086))
|
||||
.create()) {
|
||||
try (ChronicleMap<CharSequence, CharSequence> map2 =
|
||||
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 2, TcpTransportAndNetworkConfig.of(8087,
|
||||
new InetSocketAddress("localhost", 8086)))
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
map.remove("key-1", "value-1");
|
||||
assertEquals(0, map.size());
|
||||
assertEquals(null, map.get("key-1"));
|
||||
assertEquals(false, map.containsKey("key-1"));
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
map.remove("key-1");
|
||||
assertEquals(map.size(), 0);
|
||||
assertEquals(null, map.get("key-1"));
|
||||
|
||||
// test assume that we receive a late update to the map, the following update should be ignored
|
||||
final long late = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(50);
|
||||
try (MapKeyContext<?, CharSequence> c = map.context("key-1")) {
|
||||
c.writeLock().lock();
|
||||
((ReplicatedChronicleMap.ReplicatedContext) c).newIdentifier = IDENTIFIER;
|
||||
((ReplicatedChronicleMap.ReplicatedContext) c).newTimestamp = late;
|
||||
assertThrows(LateUpdateException.class, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
((ReplicatedChronicleMap.ReplicatedContext) c).put("value-2");
|
||||
}
|
||||
});
|
||||
// now test assume that we receive a late update to the map,
|
||||
// the following update should be ignored
|
||||
late(timeProvider);
|
||||
|
||||
map2.put("key-1", "value-2");
|
||||
assertEquals("value-2", map2.get("key-1"));
|
||||
|
||||
// we'll now flip the time back to the current in order to do the read the result
|
||||
current(timeProvider);
|
||||
waitTillEqual(map, map2, 5000);
|
||||
assertEquals(map.size(), 0);
|
||||
// prove that late put was ignored
|
||||
assertEquals(null, map.get("key-1"));
|
||||
}
|
||||
|
||||
assertEquals(null, map.get("key-1"));
|
||||
assertEquals(false, map.containsKey("key-1"));
|
||||
assertEquals(0, map.size(), 0);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPutRemovePut() throws IOException {
|
||||
|
||||
final TimeProvider timeProvider = Mockito.mock(TimeProvider.class);
|
||||
try (ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
|
||||
.entries(10)
|
||||
.timeProvider(timeProvider)
|
||||
.replication((byte) 1)
|
||||
.create()) {
|
||||
current(timeProvider);
|
||||
|
||||
// we do a put at the current time
|
||||
map.put("key-1", "value-1");
|
||||
map.remove("key-1");
|
||||
assertEquals(0, map.size());
|
||||
assertEquals(null, map.put("key-1", "new-value-2"));
|
||||
assertEquals(true, map.containsKey("key-1"));
|
||||
assertEquals("new-value-2", map.get("key-1"));
|
||||
assertEquals(1, map.size(), 0);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -395,6 +229,7 @@ public class TimeBasedReplicationTest extends JSR166TestCase {
|
||||
}
|
||||
|
||||
private void late(TimeProvider timeProvider) {
|
||||
Mockito.when(timeProvider.currentTime()).thenReturn(System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(5));
|
||||
Mockito.when(timeProvider.currentTime())
|
||||
.thenReturn(System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(5));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user