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

Do propagate accepted replication events, which originated from the node we are currently on, because it might be recovery

Browse Source
This commit is contained in:
leventov
2015-12-30 21:15:59 +03:00
parent b2cd5af700
commit 2c1721e2e3
3 changed files with 104 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java
+5 -1
View File
@@ -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;
}
src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java
+69 -13
View File
@@ -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<K, V, R> {
* 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<K, V> absentEntry = q.absentEntry();
@@ -104,7 +112,12 @@ public interface MapRemoteOperations<K, V, R> {
* }
* }
* replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
* replicableAbsentEntry.dropChanged();
* // For explanation see similar block above (*)
* if (q.remoteIdentifier() == q.currentNodeIdentifier()) {
* replicableAbsentEntry.raiseChanged();
* } else {
* replicableAbsentEntry.dropChanged();
* }
* }}</pre>
*
* @param q the remote operation context
@@ -115,11 +128,24 @@ public interface MapRemoteOperations<K, V, R> {
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<K, V> 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<K, V, R> {
// 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<K, V, R> {
* 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<K, V> absentEntry = q.absentEntry();
@@ -161,7 +200,12 @@ public interface MapRemoteOperations<K, V, R> {
* 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();
* }
* }
* }}</pre>
*
@@ -174,16 +218,28 @@ public interface MapRemoteOperations<K, V, R> {
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<K, V> 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();
}
}
}
}
src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java
+30 -5
View File
@@ -31,11 +31,18 @@ public interface SetRemoteOperations<K, R> {
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<K> 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<K, R> {
// 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<K, R> {
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<K> absentEntry = q.absentEntry();
@@ -71,7 +91,12 @@ public interface SetRemoteOperations<K, R> {
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();
}
}
}
}