From 0a2726ac4ad0808264bc35d40ce9f81cb2470a4b Mon Sep 17 00:00:00 2001 From: leventov Date: Thu, 19 Feb 2015 19:42:25 +0700 Subject: [PATCH] Local updates couldn't be late --- .../hash/replication/TimeProvider.java | 8 +- .../chronicle/map/ReplicatedChronicleMap.java | 30 +- .../net/openhft/chronicle/map/Builder.java | 13 + .../map/NoTcpReplicationSoakTest.java | 255 ------------ .../map/TimeBasedReplicationTest.java | 371 +++++------------- 5 files changed, 139 insertions(+), 538 deletions(-) delete mode 100644 src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java b/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java index 3703a6147..4aae3a9ce 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java @@ -37,19 +37,21 @@ import java.io.Serializable; * @see ChronicleHashBuilder#timeProvider(TimeProvider) */ public abstract class TimeProvider implements Serializable { + private static final long serialVersionUID = 0L; + /** * Delegates {@link #currentTime()} to {@link System#currentTime()}. */ public static final TimeProvider SYSTEM = new System(); - private static final long serialVersionUID = 0L; public abstract long currentTime(); private static class System extends TimeProvider { - private static final long serialVersionUID = 0L; + private static final long serialVersionUID = 1L; + private static final long SCALE = 1000; public long currentTime() { - return java.lang.System.currentTimeMillis(); + return java.lang.System.currentTimeMillis() * SCALE; } @Override diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 9b13bb51a..0aeed758c 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -541,22 +541,28 @@ final class ReplicatedChronicleMap newTimestamp) { - shouldIgnore = true; - } else { - shouldIgnore = identifier > newIdentifier; + assert testTimeStampInSensibleRange(); + if (!remoteUpdate()) { + newTimestamp = timestamp + 1L; + return false; } - if (shouldIgnore && !remoteUpdate()) { - assert newIdentifier == rm().identifier(); - throw new LateUpdateException(identifier, timestamp, newIdentifier, newTimestamp); - } - return shouldIgnore; + return timestamp > newTimestamp || + (timestamp == newTimestamp && identifier > newIdentifier); } public void dirtyEntries(final long dirtyFromTimeStamp, diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index aee3866a3..4e6f62a51 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -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 newMapDummyValue( final byte identifier, final int serverPort, diff --git a/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java deleted file mode 100644 index 00cce8dd4..000000000 --- a/src/test/java/net/openhft/chronicle/map/NoTcpReplicationSoakTest.java +++ /dev/null @@ -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 map1; - private ReplicatedChronicleMap 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 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 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 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 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 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 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 c = map2.context(key)) { - ReplicatedChronicleMap.ReplicatedContext rc = - (ReplicatedChronicleMap.ReplicatedContext) c; - rc.newTimestamp = t; - rc.newIdentifier = (byte) 1; - c.remove(); - } - task.decrementAndGet(); - } - }); - - } else { - try (MapKeyContext 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 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(); - } - } - }); - } - } - - - - -} - diff --git a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java index 7aba61fa5..236e28a82 100644 --- a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java @@ -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> 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 map = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10) + .timeProvider(timeProvider) + .replication((byte) 1, TcpTransportAndNetworkConfig.of(8086)) + .create()) { + try (ChronicleMap 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 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 map = + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10) + .timeProvider(timeProvider) + .replication((byte) 1, TcpTransportAndNetworkConfig.of(8086)) + .create()) { + try (ChronicleMap 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 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)); } }