Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Add ReplicableEntry.raiseChangedFor(), raiseChangedForAllExcept(), improve MapRemoteOperations

Browse Source
This commit is contained in:
leventov
2015-12-30 22:33:42 +03:00
parent 2c1721e2e3
commit 04adb466a2
9 changed files with 164 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java
+10
View File
@@ -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();
src/main/java/net/openhft/chronicle/hash/replication/RemoteOperationContext.java
+15 -2
View File
@@ -25,13 +25,26 @@ import net.openhft.chronicle.hash.HashContext;
* @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
+20 -4
View File
@@ -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
+18 -2
View File
@@ -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
+21 -11
View File
@@ -334,19 +334,30 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
}
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 class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
} else {
assert hunk == ENTRY_HUNK;
try (CompiledReplicatedMapQueryContext<K, V, R> remoteOpContext = mapContext()) {
remoteOpContext.processReplicatedEvent(source);
remoteOpContext.processReplicatedEvent(remoteNodeIdentifier, source);
}
}
}
@@ -858,7 +869,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
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,7 +878,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
// reconnecting vs. different Map start-up
if (e.originIdentifier() != localIdentifier ||
e.originTimestamp() >= fromTimeStamp) {
raiseChangeDuringBootstrap(c);
raiseChange0(c.tierIndex(), c.pos());
}
}
});
@@ -875,18 +886,17 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
}
}
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
+14
View File
@@ -119,6 +119,20 @@ public abstract class ReplicatedMapEntryStages<K, V> extends MapEntryStages<K, V
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
+9 -6
View File
@@ -48,10 +48,10 @@ public abstract class ReplicatedInput<K, V, R> 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<K, V, R> implements RemoteOperationContext
}
@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
+19 -1
View File
@@ -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 abstract class ReplicationUpdate<K> implements RemoteOperationContext<K>
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<K> implements RemoteOperationContext<K>
return innerRemoteIdentifier;
}
@Override
public byte remoteNodeIdentifier() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return innerRemoteNodeIdentifier;
}
@Override
public byte currentNodeIdentifier() {
return mh.m().identifier();
src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java
+38 -81
View File
@@ -72,54 +72,6 @@ public interface MapRemoteOperations<K, V, R> {
* 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: <pre>{@code
* MapReplicableEntry<K, V> 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<K, V> 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();
* }
* }}</pre>
*
* @param q the remote operation context
*/
default void remove(MapRemoteQueryContext<K, V, R> q) {
@@ -137,6 +89,10 @@ public interface MapRemoteOperations<K, V, R> {
// 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<K, V, R> {
* 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: <pre>{@code
* MapReplicableEntry<K, V> 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<K, V> 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();
* }
* }
* }}</pre>
*
* @param q the remote operation context
* @param newValue the new value to put
*/
@@ -218,7 +143,7 @@ public interface MapRemoteOperations<K, V, R> {
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<K, V, R> {
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());
}
}
}
}