diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java index c855c05e5..8d791bddc 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java @@ -47,6 +47,16 @@ public interface ReplicableEntryDelegating extends ReplicableEntry { d().raiseChanged(); } + @Override + default void raiseChangedFor(byte remoteIdentifier) { + d().raiseChangedFor(remoteIdentifier); + } + + @Override + default void raiseChangedForAllExcept(byte remoteIdentifier) { + d().raiseChangedForAllExcept(remoteIdentifier); + } + @Override default boolean isChanged() { return d().isChanged(); diff --git a/src/main/java/net/openhft/chronicle/hash/replication/RemoteOperationContext.java b/src/main/java/net/openhft/chronicle/hash/replication/RemoteOperationContext.java index 41d5c4a00..947dc2bbe 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/RemoteOperationContext.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/RemoteOperationContext.java @@ -25,13 +25,26 @@ import net.openhft.chronicle.hash.HashContext; * @param the key type of accessed {@link ChronicleHash} */ public interface RemoteOperationContext extends HashContext { - + + /** + * {@link ReplicableEntry#originIdentifier()} of the replicated entry. + */ byte remoteIdentifier(); /** - * @see ReplicableEntry#originTimestamp() + * {@link ReplicableEntry#originTimestamp()} of the replicated entry. */ long remoteTimestamp(); + /** + * Returns the identifier of the current Chronicle Node (this context object is obtained on). + */ byte currentNodeIdentifier(); + + /** + * Returns the identifier of the node, from which current replication event came from, or -1 if + * unknown or not applicable (the current replication event came not from another node, but, + * e. g., applied manually). + */ + byte remoteNodeIdentifier(); } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java index e28734f3c..26ff8d685 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java @@ -72,21 +72,37 @@ public interface ReplicableEntry { void updateOrigin(byte newIdentifier, long newTimestamp); /** - * Suppress the entry, if it was scheduled to be replicated over to remote Chronicle nodes. + * Suppress the entry, if it was scheduled to be replicated over to all remote Chronicle nodes. */ void dropChanged(); /** - * Propagate the entry, schedule it to be replicated over to remote Chronicle nodes. + * Propagate the entry, schedule it to be replicated over to all remote Chronicle nodes. */ void raiseChanged(); + /** + * Propagate the entry, schedule it to be replicated over to the remote Chronicle nodes with + * the specified identifier. + * + * @param remoteIdentifier the identifier of the node to replicate this entry to + */ + void raiseChangedFor(byte remoteIdentifier); + + /** + * Propagate the entry, schedule it to be replicated over to all remote Chronicle nodes, except + * the node with the specified identifier. + * + * @param remoteIdentifier the identifier of the node not to replicate this entry to + */ + void raiseChangedForAllExcept(byte remoteIdentifier); + /** * Check is the entry is scheduled to be replicated to the remote Chronicle nodes, to which * the connection is currently established. * - * @return {@code true} is the entry is "dirty" locally, i. e. should be replicated to remote - * Chronicle nodes, {@code false} otherwise + * @return {@code true} is the entry is "dirty" locally, i. e. should be replicated to any of + * remote Chronicle nodes, {@code false} otherwise */ boolean isChanged(); diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 51274a4e4..57acf47b8 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -30,9 +30,25 @@ public interface Replica extends Closeable { interface QueryContext { - void remotePut(Data newValue, byte remoteIdentifier, long timestamp); + /** + * @param newValue the value of the replicated entry + * @param remoteEntryIdentifier origin identifier of the replicated entry + * @param remoteEntryTimestamp timestamp of the replicated entry + * @param remoteNodeIdentifier identifier of the remote node this replication came from, + * or -1 if unknown/not applicable + */ + void remotePut( + Data newValue, + byte remoteEntryIdentifier, long remoteEntryTimestamp, byte remoteNodeIdentifier); - void remoteRemove(byte remoteIdentifier, long timestamp); + /** + * @param remoteEntryIdentifier origin identifier of the replicated entry + * @param remoteEntryTimestamp timestamp of the replicated entry + * @param remoteNodeIdentifier identifier of the remote node this replication came from, + * or -1 if unknown/not applicable + */ + void remoteRemove( + byte remoteEntryIdentifier, long remoteEntryTimestamp, byte remoteNodeIdentifier); } /** diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 968b9fe44..fa37b7603 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -334,19 +334,30 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap raise change for all + raiseChangeForAllExcept(tierIndex, pos, (byte) -1); + } + + public void raiseChangeFor(long tierIndex, long pos, byte remoteIdentifier) { + acquireModificationIterator(remoteIdentifier).raiseChange0(tierIndex, pos); + } + + public void raiseChangeForAllExcept(long tierIndex, long pos, byte remoteIdentifier) { if (tierIndex <= actualSegments) { long segmentIndex = tierIndex - 1; long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize; - for (ModificationIterator modificationIterator : assignedModificationIterators) { - modificationIterator.raiseChangeInSegment(offsetToTierBitSet, pos); + for (ModificationIterator it : assignedModificationIterators) { + if (it.remoteIdentifier != remoteIdentifier) + it.raiseChangeInSegment(offsetToTierBitSet, pos); } } else { long extraTierIndex = tierIndex - 1 - actualSegments; int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); long offsetToTierBitSet = (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; - for (ModificationIterator modificationIterator : assignedModificationIterators) { - modificationIterator.raiseChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos); + for (ModificationIterator it : assignedModificationIterators) { + if (it.remoteIdentifier != remoteIdentifier) + it.raiseChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos); } } } @@ -545,7 +556,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap remoteOpContext = mapContext()) { - remoteOpContext.processReplicatedEvent(source); + remoteOpContext.processReplicatedEvent(remoteNodeIdentifier, source); } } } @@ -858,7 +869,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap= fromTimeStamp) { - raiseChangeDuringBootstrap(c); + raiseChange0(c.tierIndex(), c.pos()); } } else { // TODO currently, all entries, originating not from the current node, @@ -867,7 +878,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap= fromTimeStamp) { - raiseChangeDuringBootstrap(c); + raiseChange0(c.tierIndex(), c.pos()); } } }); @@ -875,18 +886,17 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap c) { - long tierIndex = c.tierIndex(); + void raiseChange0(long tierIndex, long pos) { if (tierIndex <= actualSegments) { long segmentIndex = tierIndex - 1; long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize; - raiseChangeInSegment(offsetToTierBitSet, c.pos()); + raiseChangeInSegment(offsetToTierBitSet, pos); } else { long extraTierIndex = tierIndex - 1 - actualSegments; int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); long offsetToTierBitSet = (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; - raiseChangeInTierBulk(bulkIndex, offsetToTierBitSet, c.pos()); + raiseChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos); } } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java index 456b0f31b..a62950705 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java @@ -119,6 +119,20 @@ public abstract class ReplicatedMapEntryStages extends MapEntryStages implements RemoteOperationContext return dummyValue; } - public void processReplicatedEvent(Bytes replicatedInputBytes) { + public void processReplicatedEvent(byte remoteNodeIdentifier, Bytes replicatedInputBytes) { long timestamp = replicatedInputBytes.readStopBit(); byte identifier = replicatedInputBytes.readByte(); - ru.initReplicationUpdate(identifier, timestamp); + ru.initReplicationUpdate(identifier, timestamp, remoteNodeIdentifier); boolean isDeleted = replicatedInputBytes.readBoolean(); long keySize = mh.m().keySizeMarshaller.readSize(replicatedInputBytes); @@ -73,15 +73,18 @@ public abstract class ReplicatedInput implements RemoteOperationContext } @Override - public void remotePut(Data newValue, byte remoteIdentifier, long timestamp) { - ru.initReplicationUpdate(remoteIdentifier, timestamp); + public void remotePut( + Data newValue, + byte remoteEntryIdentifier, long remoteEntryTimestamp, byte remoteNodeIdentifier) { + ru.initReplicationUpdate(remoteEntryIdentifier, remoteEntryTimestamp, remoteNodeIdentifier); s.innerUpdateLock.lock(); mh.m().remoteOperations.put(this, newValue); } @Override - public void remoteRemove(byte remoteIdentifier, long timestamp) { - ru.initReplicationUpdate(remoteIdentifier, timestamp); + public void remoteRemove( + byte remoteEntryIdentifier, long remoteEntryTimestamp, byte remoteNodeIdentifier) { + ru.initReplicationUpdate(remoteEntryIdentifier, remoteEntryTimestamp, remoteNodeIdentifier); s.innerWriteLock.lock(); mh.m().remoteOperations.remove(this); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java index 5778e0c37..e3434ed8d 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java @@ -34,14 +34,18 @@ public abstract class ReplicationUpdate implements RemoteOperationContext @Stage("ReplicationUpdate") public byte innerRemoteIdentifier = (byte) 0; @Stage("ReplicationUpdate") public long innerRemoteTimestamp; + @Stage("ReplicationUpdate") public byte innerRemoteNodeIdentifier; public abstract boolean replicationUpdateInit(); - public void initReplicationUpdate(byte identifier, long timestamp) { + public void initReplicationUpdate(byte identifier, long timestamp, byte remoteNodeIdentifier) { innerRemoteTimestamp = timestamp; if (identifier == 0) throw new IllegalStateException("identifier can't be 0"); innerRemoteIdentifier = identifier; + if (remoteNodeIdentifier == 0) + throw new IllegalStateException("remote node identifier can't be 0"); + innerRemoteNodeIdentifier = remoteNodeIdentifier; } public void dropChange() { @@ -62,6 +66,14 @@ public abstract class ReplicationUpdate implements RemoteOperationContext mh.m().raiseChange(s.tierIndex, e.pos); } + public void raiseChangeFor(byte remoteIdentifier) { + mh.m().raiseChangeFor(s.tierIndex, e.pos, remoteIdentifier); + } + + public void raiseChangeForAllExcept(byte remoteIdentifier) { + mh.m().raiseChangeForAllExcept(s.tierIndex, e.pos, remoteIdentifier); + } + public boolean changed() { return mh.m().isChanged(s.tierIndex, e.pos); } @@ -78,6 +90,12 @@ public abstract class ReplicationUpdate implements RemoteOperationContext return innerRemoteIdentifier; } + @Override + public byte remoteNodeIdentifier() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return innerRemoteNodeIdentifier; + } + @Override public byte currentNodeIdentifier() { return mh.m().identifier(); 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 6cfa2fcf1..fd8460d04 100644 --- a/src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java +++ b/src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java @@ -72,54 +72,6 @@ public interface MapRemoteOperations { * Handle remote {@code remove} call and {@code remove} replication event, i. e. when the entry * with the query key ({@code q.queriedKey()}) was removed on some {@code ChronicleMap} node. * - * @implNote the default implementation applies the remove event using {@link - * DefaultEventualConsistencyStrategy} "latest wins" strategy: 
{@code
-     * MapReplicableEntry entry = q.entry();
-     * if (entry != null) {
-     *     if (DefaultEventualConsistencyStrategy.decideOnRemoteModification(entry, q) == ACCEPT) {
-     *         q.remove(entry);
-     *         ReplicableEntry replicableAbsentEntry = (ReplicableEntry) q.absentEntry();
-     *         replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
-     *         // (*)
-     *         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();
-     *     ReplicableEntry replicableAbsentEntry;
-     *     if (!(absentEntry instanceof ReplicableEntry)) {
-     *         // Note in the two following lines dummy value is inserted and removed using direct
-     *         // entry.doXxx calls, not q.xxx(entry). The intention is to avoid calling possibly
-     *         // 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();
-     *         replicableAbsentEntry = (ReplicableEntry) q.absentEntry();
-     *     } else {
-     *         replicableAbsentEntry = (ReplicableEntry) absentEntry;
-     *         if (DefaultEventualConsistencyStrategy.decideOnRemoteModification(
-     *                 replicableAbsentEntry, q) == DISCARD) {
-     *             return;
-     *         }
-     *     }
-     *     replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
-     *     // For explanation see similar block above (*)
-     *     if (q.remoteIdentifier() == q.currentNodeIdentifier()) {
-     *         replicableAbsentEntry.raiseChanged();
-     *     } else {
-     *         replicableAbsentEntry.dropChanged();
-     *     }
-     * }}
- * * @param q the remote operation context */ default void remove(MapRemoteQueryContext q) { @@ -137,6 +89,10 @@ public interface MapRemoteOperations { // 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 + + // There was an attempt to use raiseChangedForAllExcept(q.remoteNodeIdentifier), + // But it fails in the case connected to what explained in the end of put() + // method. TODO it's not comprehensible WHY is this so replicableAbsentEntry.raiseChanged(); } else { // We accepted a replication event, suppress further event propagation @@ -178,37 +134,6 @@ public interface MapRemoteOperations { * key ({@code q.queriedKey()}) was changed on some remote {@code ChronicleMap} node, with the * given {@code newValue}. * - * @implNote the default implementation applies the put event using {@link - * DefaultEventualConsistencyStrategy} "latest wins" strategy: 
{@code
-     * MapReplicableEntry entry = q.entry();
-     * if (entry != null) {
-     *     if (DefaultEventualConsistencyStrategy.decideOnRemoteModification(entry, q) == ACCEPT) {
-     *         q.replaceValue(entry, newValue);
-     *         entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
-     *         // For explanation see similar block in remove() method documentation (*)
-     *         if (q.remoteIdentifier() == q.currentNodeIdentifier()) {
-     *             entry.raiseChanged();
-     *         } else {
-     *             entry.dropChanged();
-     *         }
-     *     }
-     * } else {
-     *     MapAbsentEntry absentEntry = q.absentEntry();
-     *     if (!(absentEntry instanceof ReplicableEntry) ||
-     *             DefaultEventualConsistencyStrategy.decideOnRemoteModification(
-     *                     (ReplicableEntry) absentEntry, q) == ACCEPT) {
-     *         q.insert(absentEntry, newValue);
-     *         entry = q.entry();
-     *         entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
-     *         // For explanation see similar block in remove() method documentation (*)
-     *         if (q.remoteIdentifier() == q.currentNodeIdentifier()) {
-     *             entry.raiseChanged();
-     *         } else {
-     *             entry.dropChanged();
-     *         }
-     *     }
-     * }}
- * * @param q the remote operation context * @param newValue the new value to put */ @@ -218,7 +143,7 @@ public interface MapRemoteOperations { if (decideOnRemoteModification(entry, q) == ACCEPT) { q.replaceValue(entry, newValue); entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - // For explanation see similar block in remove() method documentation (*) + // For explanation see similar block in remove() method (*) if (q.remoteIdentifier() == q.currentNodeIdentifier()) { entry.raiseChanged(); } else { @@ -234,12 +159,44 @@ public interface MapRemoteOperations { entry = q.entry(); assert entry != null; entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); - // For explanation see similar block in remove() method documentation (*) + // For explanation see similar block in remove() method (*) if (q.remoteIdentifier() == q.currentNodeIdentifier()) { entry.raiseChanged(); } else { entry.dropChanged(); } + } else { + // In case of old deleted entries cleanup + network disconnections and + // bootstrapping, it is possible that the entry is ends being removed on remote node + // and present on origin node: + // + // 1. Entry is added on node 1 + // 2. This change is replicated and arrived to node 2 + // 3. Entry is removed on node 1, it is scheduled for replication again + // 4. Network issues, disconnection, bootstrapping + // 5. Node 2 schedules the subject entry (currently present on node 2) + // for bootstrapping to the node 1 + // 6. Node 2 sends the replication event to node 1 (in-flight) + // 7. The entry removal is replicated to node 2, it is accepted, finally the entry + // is removed on node 2 + // 8. The entry is cleaned up (erased completely) on node 2, because it is already + // replicated to everywhere it should be, and timeout (if set very short) + // is expired + // 9. Replication event from node 2 (with present entry) arrives to node 1, as the + // entry is already erased from node 1, this change is accepted, and finally + // the entry is present on node 1. The is bootstrapped again, by (*) blocks + // 10. The "present" entry from node 1 is discarded on node 2, because it sees that + // this entry was removed later. + // + // The following block captures the condition from item 10, and sends the event + // again back, to give the chance to be finally removed on node 1 (in the above + // example). + if (((ReplicableEntry) absentEntry).originIdentifier() == q.remoteIdentifier() && + q.remoteIdentifier() != q.currentNodeIdentifier()) { + // If this change will arrive to the origin node and accepted, it will be + // propagated to all other nodes, by (*) blocks + ((ReplicableEntry) absentEntry).raiseChangedFor(q.remoteIdentifier()); + } } } }