diff --git a/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java b/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java index 92641abba..767601861 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java @@ -18,7 +18,6 @@ package net.openhft.chronicle.hash.replication; import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.ChronicleHashBuilder; -import net.openhft.chronicle.map.Replica; import net.openhft.chronicle.map.replication.MapRemoteOperations; import net.openhft.chronicle.set.replication.SetRemoteOperations; @@ -73,6 +72,11 @@ public final class DefaultEventualConsistencyStrategy { // "back" from other nodes in the system, are discarded on this new node (this is the // of the condition originIdentifier == currentNodeIdentifier). But those new updates // should win on other nodes. + // + // Another case, in which we could have remoteTimestamp == originTimestamp && + // remoteIdentifier == originIdentifier, is replication of barely the same entry, if an + // entry is bootstrapped "back" from remote node to it's origin node. In this case the + // following condition also plays right (the update is discarded, due to it's redundancy). return originIdentifier == context.currentNodeIdentifier() ? DISCARD : ACCEPT; } diff --git a/src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java b/src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java index 79c22c3a8..6cfa2fcf1 100644 --- a/src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java +++ b/src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java @@ -19,10 +19,7 @@ package net.openhft.chronicle.map.replication; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy; import net.openhft.chronicle.hash.replication.ReplicableEntry; -import net.openhft.chronicle.map.ChronicleMap; -import net.openhft.chronicle.map.ChronicleMapBuilder; -import net.openhft.chronicle.map.MapAbsentEntry; -import net.openhft.chronicle.map.MapEntryOperations; +import net.openhft.chronicle.map.*; import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy.AcceptanceDecision.ACCEPT; import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy.AcceptanceDecision.DISCARD; @@ -83,7 +80,18 @@ public interface MapRemoteOperations { * q.remove(entry); * ReplicableEntry replicableAbsentEntry = (ReplicableEntry) q.absentEntry(); * replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - * replicableAbsentEntry.dropChanged(); + * // (*) + * if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + * // The entry with origin (replicatedIdentifier() is remote entry's origin), + * // equal to the current node, should be lost on on this node (or this node is + * // a different Chronicle Map instance, because the previous was lost, just using + * // the same identifier), if it happens we should propagate this recovered + * // event again, to other nodes + * replicableAbsentEntry.raiseChanged(); + * } else { + * // We accepted a replication event, suppress further even propagation + * replicableAbsentEntry.dropChanged(); + * } * } * } else { * MapAbsentEntry absentEntry = q.absentEntry(); @@ -104,7 +112,12 @@ public interface MapRemoteOperations { * } * } * replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - * replicableAbsentEntry.dropChanged(); + * // For explanation see similar block above (*) + * if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + * replicableAbsentEntry.raiseChanged(); + * } else { + * replicableAbsentEntry.dropChanged(); + * } * }} * * @param q the remote operation context @@ -115,11 +128,24 @@ public interface MapRemoteOperations { if (decideOnRemoteModification(entry, q) == ACCEPT) { q.remove(entry); ReplicableEntry replicableAbsentEntry = (ReplicableEntry) q.absentEntry(); + assert replicableAbsentEntry != null; replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - replicableAbsentEntry.dropChanged(); + // (*) + if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + // The entry with origin (replicatedIdentifier() is remote entry's origin), + // equal to the current node, should be lost on on this node (or this node is + // a different Chronicle Map instance, because the previous was lost, just using + // the same identifier), if it happens we should propagate this recovered + // event again, to other nodes + replicableAbsentEntry.raiseChanged(); + } else { + // We accepted a replication event, suppress further event propagation + replicableAbsentEntry.dropChanged(); + } } } else { MapAbsentEntry absentEntry = q.absentEntry(); + assert absentEntry != null; ReplicableEntry replicableAbsentEntry; if (!(absentEntry instanceof ReplicableEntry)) { // Note in the two following lines dummy value is inserted and removed using direct @@ -127,15 +153,23 @@ public interface MapRemoteOperations { // overridden MapEntryOperations, because this is technical procedure of making // "truly absent" entry "deleted", not actual insertion and removal. absentEntry.doInsert(q.dummyZeroValue()); - q.entry().doRemove(); + entry = q.entry(); + assert entry != null; + entry.doRemove(); replicableAbsentEntry = (ReplicableEntry) q.absentEntry(); + assert replicableAbsentEntry != null; } else { replicableAbsentEntry = (ReplicableEntry) absentEntry; if (decideOnRemoteModification(replicableAbsentEntry, q) == DISCARD) return; } replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - replicableAbsentEntry.dropChanged(); + // For explanation see similar block above (*) + if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + replicableAbsentEntry.raiseChanged(); + } else { + replicableAbsentEntry.dropChanged(); + } } } @@ -151,7 +185,12 @@ public interface MapRemoteOperations { * if (DefaultEventualConsistencyStrategy.decideOnRemoteModification(entry, q) == ACCEPT) { * q.replaceValue(entry, newValue); * entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - * entry.dropChanged(); + * // For explanation see similar block in remove() method documentation (*) + * if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + * entry.raiseChanged(); + * } else { + * entry.dropChanged(); + * } * } * } else { * MapAbsentEntry absentEntry = q.absentEntry(); @@ -161,7 +200,12 @@ public interface MapRemoteOperations { * q.insert(absentEntry, newValue); * entry = q.entry(); * entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - * entry.dropChanged(); + * // For explanation see similar block in remove() method documentation (*) + * if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + * entry.raiseChanged(); + * } else { + * entry.dropChanged(); + * } * } * }} * @@ -174,16 +218,28 @@ public interface MapRemoteOperations { if (decideOnRemoteModification(entry, q) == ACCEPT) { q.replaceValue(entry, newValue); entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - entry.dropChanged(); + // For explanation see similar block in remove() method documentation (*) + if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + entry.raiseChanged(); + } else { + entry.dropChanged(); + } } } else { MapAbsentEntry absentEntry = q.absentEntry(); + assert absentEntry != null; if (!(absentEntry instanceof ReplicableEntry) || decideOnRemoteModification((ReplicableEntry) absentEntry, q) == ACCEPT) { q.insert(absentEntry, newValue); entry = q.entry(); + assert entry != null; entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - entry.dropChanged(); + // For explanation see similar block in remove() method documentation (*) + if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + entry.raiseChanged(); + } else { + entry.dropChanged(); + } } } } diff --git a/src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java b/src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java index 9bea427a6..00a680c8f 100644 --- a/src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java +++ b/src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java @@ -31,11 +31,18 @@ public interface SetRemoteOperations { if (decideOnRemoteModification(entry, q) == ACCEPT) { q.remove(entry); ReplicableEntry replicableAbsentEntry = (ReplicableEntry) q.absentEntry(); + assert replicableAbsentEntry != null; replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - replicableAbsentEntry.dropChanged(); + // See MapRemoteOperations + if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + replicableAbsentEntry.raiseChanged(); + } else { + replicableAbsentEntry.dropChanged(); + } } } else { SetAbsentEntry absentEntry = q.absentEntry(); + assert absentEntry != null; ReplicableEntry replicableAbsentEntry; if (!(absentEntry instanceof ReplicableEntry)) { // Note in the two following lines dummy entry is inserted and removed using direct @@ -43,15 +50,23 @@ public interface SetRemoteOperations { // overridden SetEntryOperations, because this is technical procedure of making // "truly absent" entry "deleted", not actual insertion and removal. absentEntry.doInsert(); - q.entry().doRemove(); + entry = q.entry(); + assert entry != null; + entry.doRemove(); replicableAbsentEntry = (ReplicableEntry) q.absentEntry(); + assert replicableAbsentEntry != null; } else { replicableAbsentEntry = (ReplicableEntry) absentEntry; if (decideOnRemoteModification(replicableAbsentEntry, q) == DISCARD) return; } replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - replicableAbsentEntry.dropChanged(); + // See MapRemoteOperations + if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + replicableAbsentEntry.raiseChanged(); + } else { + replicableAbsentEntry.dropChanged(); + } } } @@ -60,7 +75,12 @@ public interface SetRemoteOperations { if (entry != null) { if (decideOnRemoteModification(entry, q) == ACCEPT) { entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - entry.dropChanged(); + // See MapRemoteOperations + if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + entry.raiseChanged(); + } else { + entry.dropChanged(); + } } } else { SetAbsentEntry absentEntry = q.absentEntry(); @@ -71,7 +91,12 @@ public interface SetRemoteOperations { entry = q.entry(); // q.entry() is not null after insert assert entry != null; entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - entry.dropChanged(); + // See MapRemoteOperations + if (q.remoteIdentifier() == q.currentNodeIdentifier()) { + entry.raiseChanged(); + } else { + entry.dropChanged(); + } } } }