Add ReplicableEntry.raiseChangedFor(), raiseChangedForAllExcept(), im…

…prove MapRemoteOperations

src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java

@@ -47,6 +47,16 @@ default void raiseChanged() {
         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();

src/main/java/net/openhft/chronicle/hash/replication/RemoteOperationContext.java

@@ -25,13 +25,26 @@
  * @param <K> the key type of accessed {@link ChronicleHash}
  */
 public interface RemoteOperationContext<K> extends HashContext<K> {
-
+
+    /**
+     * {@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();
 }

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();
 

src/main/java/net/openhft/chronicle/map/Replica.java

@@ -30,9 +30,25 @@ public interface Replica extends Closeable {
 
     interface QueryContext<K, V> {
 
-        void remotePut(Data<V> 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<V> 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);
     }
 
     /**

src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java

@@ -334,19 +334,30 @@ public ModificationIterator acquireModificationIterator(byte remoteIdentifier) {
     }
 
     public void raiseChange(long tierIndex, long pos) {
+        // -1 is invalid remoteIdentifier => 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 void readExternalEntry(@NotNull Bytes source, byte remoteNodeIdentifier)
         } else {
             assert hunk == ENTRY_HUNK;
             try (CompiledReplicatedMapQueryContext<K, V, R> remoteOpContext = mapContext()) {
-                remoteOpContext.processReplicatedEvent(source);
+                remoteOpContext.processReplicatedEvent(remoteNodeIdentifier, source);
             }
         }
     }
@@ -858,7 +869,7 @@ public void dirtyEntries(long fromTimeStamp) {
                         if (bootstrapOnlyLocalEntries) {
                             if (e.originIdentifier() == localIdentifier &&
                                     e.originTimestamp() >= fromTimeStamp) {
-                                raiseChangeDuringBootstrap(c);
+                                raiseChange0(c.tierIndex(), c.pos());
                             }
                         } else {
                             // TODO currently, all entries, originating not from the current node,
@@ -867,26 +878,25 @@ public void dirtyEntries(long fromTimeStamp) {
                             // reconnecting vs. different Map start-up
                             if (e.originIdentifier() != localIdentifier ||
                                     e.originTimestamp() >= fromTimeStamp) {
-                                raiseChangeDuringBootstrap(c);
+                                raiseChange0(c.tierIndex(), c.pos());
                             }
                         }
                     });
                 }
             }
         }
 
-        private void raiseChangeDuringBootstrap(CompiledReplicatedMapIterationContext<K, V, R> 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);
             }
         }
     }

src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java

@@ -119,6 +119,20 @@ public void raiseChanged() {
         ru.raiseChange();
     }
 
+    @Override
+    public void raiseChangedFor(byte remoteIdentifier) {
+        checkOnEachPublicOperation.checkOnEachPublicOperation();
+        s.innerUpdateLock.lock();
+        ru.raiseChangeFor(remoteIdentifier);
+    }
+
+    @Override
+    public void raiseChangedForAllExcept(byte remoteIdentifier) {
+        checkOnEachPublicOperation.checkOnEachPublicOperation();
+        s.innerUpdateLock.lock();
+        ru.raiseChangeForAllExcept(remoteIdentifier);
+    }
+
     @Override
     public boolean isChanged() {
         checkOnEachPublicOperation.checkOnEachPublicOperation();

src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java

@@ -48,10 +48,10 @@ public Data<V> dummyZeroValue() {
         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 void processReplicatedEvent(Bytes replicatedInputBytes) {
     }
 
     @Override
-    public void remotePut(Data<V> newValue, byte remoteIdentifier, long timestamp) {
-        ru.initReplicationUpdate(remoteIdentifier, timestamp);
+    public void remotePut(
+            Data<V> 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);
     }

src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java

@@ -34,14 +34,18 @@ public abstract class ReplicationUpdate<K> implements RemoteOperationContext<K>
 
     @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 void raiseChange() {
         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 byte remoteIdentifier() {
         return innerRemoteIdentifier;
     }
 
+    @Override
+    public byte remoteNodeIdentifier() {
+        checkOnEachPublicOperation.checkOnEachPublicOperation();
+        return innerRemoteNodeIdentifier;
+    }
+
     @Override
     public byte currentNodeIdentifier() {
         return mh.m().identifier();