From f53a93275cc875143bdeed774d2c184dc51cc2fc Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 19 Jan 2015 22:10:29 +0700 Subject: [PATCH] Late updates to ReplicatedChronicleMap throw LateUpdateException --- .../hash/replication/LateUpdateException.java | 61 ++++++++++++ .../chronicle/map/AbstractChronicleMap.java | 3 +- .../chronicle/map/ReplicatedChronicleMap.java | 19 +++- .../map/TimeBasedReplicationTest.java | 92 +++++++++++++------ 4 files changed, 140 insertions(+), 35 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/hash/replication/LateUpdateException.java diff --git a/src/main/java/net/openhft/chronicle/hash/replication/LateUpdateException.java b/src/main/java/net/openhft/chronicle/hash/replication/LateUpdateException.java new file mode 100644 index 000000000..cd8f99da5 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/replication/LateUpdateException.java @@ -0,0 +1,61 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash.replication; + +/** + * Thrown if update to replicated {@link net.openhft.chronicle.hash.ChronicleHash} is rejected + * due to local time is lesser than the time of the latest update to the entry. This most likely + * indicates bugs or largely inconsistent timing across replicating nodes. If this is tolerated, + * you should catch {@code LateUpdateException} and ignore it. + */ +public class LateUpdateException extends RuntimeException { + private static final long serialVersionUID = 0L; + + private final byte originIdentifier; + private final long lastUpdateTimestamp; + private final byte thisNodeIdentifier; + private final long thisUpdateTimestamp; + + public LateUpdateException(byte originIdentifier, long lastUpdateTimestamp, + byte thisNodeIdentifier, long thisUpdateTimestamp) { + super("Late update : " + + "last update time: " + lastUpdateTimestamp + + ", last update origin node identifier: " + originIdentifier + + ", time of this update: " + thisUpdateTimestamp + + ", identifier of this node: " + thisNodeIdentifier); + this.originIdentifier = originIdentifier; + this.lastUpdateTimestamp = lastUpdateTimestamp; + this.thisNodeIdentifier = thisNodeIdentifier; + this.thisUpdateTimestamp = thisUpdateTimestamp; + } + + public long thisUpdateTimestamp() { + return thisUpdateTimestamp; + } + + public byte thisNodeIdentifier() { + return thisNodeIdentifier; + } + + public long lastUpdateTimestamp() { + return lastUpdateTimestamp; + } + + public byte lastUpdateOriginNodeIdentifier() { + return originIdentifier; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java index 30bcde53a..fdfc336c9 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java @@ -222,9 +222,8 @@ abstract class AbstractChronicleMap extends AbstractMap // 2. Always "try to read" the previous value (not via if (c.containsKey()) {... // because BytesChronicleMap relies on it to write null to the output stream V prevValue = c.getUsing(null); - if (prevValue != null) { + if (prevValue != null) c.put(value); - } return prevValue; } } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 7c6d575fa..db864a043 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -19,6 +19,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.AbstractReplication; +import net.openhft.chronicle.hash.replication.LateUpdateException; import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.chronicle.hash.serialization.internal.BytesBytesInterop; @@ -515,11 +516,19 @@ final class ReplicatedChronicleMap newTimestamp) - return true; - return identifier > newIdentifier; + boolean shouldIgnore; + if (timestamp < newTimestamp) { + shouldIgnore = false; + } else if (timestamp > newTimestamp) { + shouldIgnore = true; + } else { + shouldIgnore = identifier > newIdentifier; + } + if (shouldIgnore && !remoteUpdate()) { + assert newIdentifier == m().identifier(); + throw new LateUpdateException(identifier, timestamp, newIdentifier, newTimestamp); + } + return shouldIgnore; } public void dirtyEntries(final long dirtyFromTimeStamp, diff --git a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java index a2eae5a64..64d0e936b 100644 --- a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.hash.replication.LateUpdateException; import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.map.jsr166.JSR166TestCase; import org.junit.After; @@ -77,7 +78,12 @@ public class TimeBasedReplicationTest extends JSR166TestCase { // now test assume that we receive a late update to the map, the following update should be ignored late(timeProvider); - map.put("key-1", "value-2"); + assertThrows(LateUpdateException.class, new Runnable() { + @Override + public void run() { + map.put("key-1", "value-2"); + } + }); // we'll now flip the time back to the current in order to do the read the result current(timeProvider); @@ -102,11 +108,13 @@ public class TimeBasedReplicationTest extends JSR166TestCase { 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 + // now test assume that we receive a late update to the map, the following update + // should be ignored late(timeProvider); - final Object o = map.putIfAbsent("key-1", "value-2"); - assertEquals(o, null); + // should NOT throw LateUpdateException because the value is present and it is just + // returned. + assertEquals(map.putIfAbsent("key-1", "value-2"), null); // we'll now flip the time back to the current in order to do the read the result current(timeProvider); @@ -135,8 +143,12 @@ public class TimeBasedReplicationTest extends JSR166TestCase { // the following update should be ignored late(timeProvider); - final Object o = map.replace("key-1", "value-2"); - assertEquals(o, null); + 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); @@ -161,10 +173,16 @@ public class TimeBasedReplicationTest extends JSR166TestCase { 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 + // now test assume that we receive a late update to the map, the following update + // should be ignored late(timeProvider); - assertEquals(null, map.replace("key-1", "value-2")); + 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); @@ -189,10 +207,16 @@ public class TimeBasedReplicationTest extends JSR166TestCase { 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 + // now test assume that we receive a late update to the map, the following update + // should be ignored late(timeProvider); - assertEquals(false, map.remove("key-1", "value-1")); + 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); @@ -211,26 +235,33 @@ public class TimeBasedReplicationTest extends JSR166TestCase { .replication((byte) 1) .create()) { - current(timeProvider); + 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")); + // 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); + // now test assume that we receive a late update to the map, the following update + // should be ignored + late(timeProvider); - map.remove("key-1"); + 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")); + } - // 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 { @@ -280,12 +311,12 @@ public class TimeBasedReplicationTest extends JSR166TestCase { 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 - // now test assume that we receive a late update to the map, the following update should be ignored + // 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 = IDENTIFIER; + ((ReplicatedChronicleMap.ReplicatedContext) c).newIdentifier = (byte) 2; ((ReplicatedChronicleMap.ReplicatedContext) c).newTimestamp = late; assertFalse(((ReplicatedChronicleMap.ReplicatedContext) c).put("value-2")); } @@ -323,7 +354,12 @@ public class TimeBasedReplicationTest extends JSR166TestCase { c.writeLock().lock(); ((ReplicatedChronicleMap.ReplicatedContext) c).newIdentifier = IDENTIFIER; ((ReplicatedChronicleMap.ReplicatedContext) c).newTimestamp = late; - assertFalse(((ReplicatedChronicleMap.ReplicatedContext) c).put("value-2")); + assertThrows(LateUpdateException.class, new Runnable() { + @Override + public void run() { + ((ReplicatedChronicleMap.ReplicatedContext) c).put("value-2"); + } + }); } assertEquals(null, map.get("key-1"));