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

HCOLL-132 changed the code to use @net.openhft.lang.model.constraints.NotNull

Browse Source
This commit is contained in:
Rob Austin
2014-09-12 16:00:10 +01:00
parent d30d85c8da
commit d0c33d0b08
5 changed files with 90 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+73 -80
View File
@@ -38,40 +38,36 @@ import static net.openhft.chronicle.map.VanillaChronicleMap.Hasher.hash;
import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
/**
* A Replicating Multi Master HashMap
* <p/>
* <p>Each remote hash map, mirrors its changes over to another remote hash map, neither hash map is considered the
* master store of data, each hash map uses timestamps to reconcile changes. We refer to an instance of a remote
* hash-map as a node. A node will be connected to any number of other nodes, for the first implementation the maximum
* number of nodes will be fixed. The data that is stored locally in each node will become eventually consistent. So
* changes made to one node, for example by calling put() will be replicated over to the other node. To achieve a high
* level of performance and throughput, the call to put() won’t block, with concurrentHashMap, It is typical to check
* the return code of some methods to obtain the old value for example remove(). Due to the loose coupling and lock free
* nature of this multi master implementation, this return value will only be the old value on the nodes local data
* store. In other words the nodes are only concurrent locally. Its worth realising that another node performing exactly
* the same operation may return a different value. However reconciliation will ensure the maps themselves become
* eventually consistent.
* <p/>
* <p>Reconciliation
* <p/>
* <p>If two ( or more nodes ) were to receive a change to their maps for the same key but different values, say by a
* user of the maps, calling the put(key, value). Then, initially each node will update its local store and each local
* store will hold a different value, but the aim of multi master replication is to provide eventual consistency across
* the nodes. So, with multi master when ever a node is changed it will notify the other nodes of its change. We will
* refer to this notification as an event. The event will hold a timestamp indicating the time the change occurred, it
* will also hold the state transition, in this case it was a put with a key and value. Eventual consistency is achieved
* by looking at the timestamp from the remote node, if for a given key, the remote nodes timestamp is newer than the
* local nodes timestamp, then the event from the remote node will be applied to the local node, otherwise the event
* will be ignored.
* <p/>
* <p>However there is an edge case that we have to concern ourselves with, If two nodes update their map at the same
* time with different values, we have to deterministically resolve which update wins, because of eventual consistency
* both nodes should end up locally holding the same data. Although it is rare two remote nodes could receive an update
* to their maps at exactly the same time for the same key, we have to handle this edge case, its therefore important
* not to rely on timestamps alone to reconcile the updates. Typically the update with the newest timestamp should win,
* but in this example both timestamps are the same, and the decision made to one node should be identical to the
* decision made to the other. We resolve this simple dilemma by using a node identifier, each node will have a unique
* identifier, the update from the node with the smallest identifier wins.
* A Replicating Multi Master HashMap <p/> <p>Each remote hash map, mirrors its changes over to another remote
* hash map, neither hash map is considered the master store of data, each hash map uses timestamps to
* reconcile changes. We refer to an instance of a remote hash-map as a node. A node will be connected to any
* number of other nodes, for the first implementation the maximum number of nodes will be fixed. The data
* that is stored locally in each node will become eventually consistent. So changes made to one node, for
* example by calling put() will be replicated over to the other node. To achieve a high level of performance
* and throughput, the call to put() won’t block, with concurrentHashMap, It is typical to check the return
* code of some methods to obtain the old value for example remove(). Due to the loose coupling and lock free
* nature of this multi master implementation, this return value will only be the old value on the nodes
* local data store. In other words the nodes are only concurrent locally. Its worth realising that another
* node performing exactly the same operation may return a different value. However reconciliation will ensure
* the maps themselves become eventually consistent. <p/> <p>Reconciliation <p/> <p>If two ( or more nodes )
* were to receive a change to their maps for the same key but different values, say by a user of the maps,
* calling the put(key, value). Then, initially each node will update its local store and each local store
* will hold a different value, but the aim of multi master replication is to provide eventual consistency
* across the nodes. So, with multi master when ever a node is changed it will notify the other nodes of its
* change. We will refer to this notification as an event. The event will hold a timestamp indicating the time
* the change occurred, it will also hold the state transition, in this case it was a put with a key and
* value. Eventual consistency is achieved by looking at the timestamp from the remote node, if for a given
* key, the remote nodes timestamp is newer than the local nodes timestamp, then the event from the remote
* node will be applied to the local node, otherwise the event will be ignored. <p/> <p>However there is an
* edge case that we have to concern ourselves with, If two nodes update their map at the same time with
* different values, we have to deterministically resolve which update wins, because of eventual consistency
* both nodes should end up locally holding the same data. Although it is rare two remote nodes could receive
* an update to their maps at exactly the same time for the same key, we have to handle this edge case, its
* therefore important not to rely on timestamps alone to reconcile the updates. Typically the update with the
* newest timestamp should win, but in this example both timestamps are the same, and the decision made to one
* node should be identical to the decision made to the other. We resolve this simple dilemma by using a node
* identifier, each node will have a unique identifier, the update from the node with the smallest identifier
* wins.
*
* @param <K> the entries key type
* @param <V> the entries value type
@@ -187,8 +183,8 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* Used in conjunction with map replication, all put events that originate from a remote node will be processed
* using this method.
* Used in conjunction with map replication, all put events that originate from a remote node will be
* processed using this method.
*
* @param key key with which the specified value is to be associated
* @param value value to be associated with the specified key
@@ -204,7 +200,7 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
@Override
public V putIfAbsent(@NotNull K key, V value) {
public V putIfAbsent(@net.openhft.lang.model.constraints.NotNull K key, V value) {
return put0(key, value, false, localIdentifier, timeProvider.currentTimeMillis());
}
@@ -253,8 +249,8 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* Used in conjunction with map replication, all remove events that originate from a remote node
* will be processed using this method.
* Used in conjunction with map replication, all remove events that originate from a remote node will be
* processed using this method.
*
* @param key key with which the specified value is associated
* @param value value expected to be associated with the specified key
@@ -307,7 +303,7 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
@Override
public boolean remove(@NotNull final Object key, final Object value) {
public boolean remove(@net.openhft.lang.model.constraints.NotNull final Object key, final Object value) {
if (value == null)
return false; // CHM compatibility; I would throw NPE
return removeIfValueIs(key, (V) value,
@@ -326,7 +322,7 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
@Override
V replaceIfValueIs(@NotNull final K key, final V existingValue, final V newValue) {
V replaceIfValueIs(@net.openhft.lang.model.constraints.NotNull final K key, final V existingValue, final V newValue) {
checkKey(key);
checkValue(newValue);
Bytes keyBytes = getKeyAsBytes(key);
@@ -338,8 +334,8 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* This method does not set a segment lock, A segment lock should be obtained before calling
* this method, especially when being used in a multi threaded context.
* This method does not set a segment lock, A segment lock should be obtained before calling this method,
* especially when being used in a multi threaded context.
*/
@Override
public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
@@ -414,8 +410,8 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* This method does not set a segment lock, A segment lock should be obtained before calling
* this method, especially when being used in a multi threaded context.
* This method does not set a segment lock, A segment lock should be obtained before calling this method,
* especially when being used in a multi threaded context.
*/
@Override
public void readExternalEntry(@NotNull Bytes source) {
@@ -821,10 +817,9 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* Used only with replication, its sometimes possible to receive an old ( or stale update ) from a remote map.
* This method is used to determine if we should ignore such updates.
* <p/>
* <p>We can reject put() and removes() when comparing times stamps with remote systems
* Used only with replication, its sometimes possible to receive an old ( or stale update ) from a
* remote map. This method is used to determine if we should ignore such updates. <p/> <p>We can
* reject put() and removes() when comparing times stamps with remote systems
*
* @param entry the maps entry
* @param timestamp the time the entry was created or updated
@@ -851,19 +846,19 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* Puts entry. If {@code value} implements {@link net.openhft.lang.model.Byteable} interface and {@code
* usingValue} is {@code true}, the value is backed with the bytes of this entry.
* Puts entry. If {@code value} implements {@link net.openhft.lang.model.Byteable} interface and
* {@code usingValue} is {@code true}, the value is backed with the bytes of this entry.
*
* @param keyBytes serialized key
* @param hash2 a hash of the {@code keyBytes}. Caller was searching for the key in the {@code
* searchedHashLookup} using this hash.
* @param hash2 a hash of the {@code keyBytes}. Caller was searching for the key in the
* {@code searchedHashLookup} using this hash.
* @param value the value to put
* @param usingValue {@code true} if the value should be backed with the bytes of the entry, if it
* implements {@link net.openhft.lang.model.Byteable} interface, {@code false} if it
* should put itself
* @param usingValue {@code true} if the value should be backed with the bytes of the entry,
* if it implements {@link net.openhft.lang.model.Byteable} interface,
* {@code false} if it should put itself
* @param identifier the identifier of the outer CHM node
* @param timestamp the timestamp when the entry was put <s>(this could be later if it was a remote
* put)</s> this method is called only from usual put or acquire
* @param timestamp the timestamp when the entry was put <s>(this could be later if it was a
* remote put)</s> this method is called only from usual put or acquire
* @param searchedHashLookup the hash lookup that used to find the entry based on the key
* @return offset of the written entry in the Segment bytes
* @see VanillaChronicleMap.Segment#putEntry(Bytes, Object, boolean)
@@ -1149,8 +1144,8 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* receive an update from the map, via the MapEventListener and delegates the changes
* to the currently active modification iterators
* receive an update from the map, via the MapEventListener and delegates the changes to the currently
* active modification iterators
*/
class ModificationDelegator extends MapEventListener<K, V, ChronicleMap<K, V>> {
private static final long serialVersionUID = 0L;
@@ -1267,16 +1262,16 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* Always throws {@code NotSerializableException} since instances of this class
* are not intended to be serializable.
* Always throws {@code NotSerializableException} since instances of this class are not intended to be
* serializable.
*/
private void writeObject(ObjectOutputStream out) throws IOException {
throw new NotSerializableException(getClass().getCanonicalName());
}
/**
* Always throws {@code NotSerializableException} since instances of this class
* are not intended to be serializable.
* Always throws {@code NotSerializableException} since instances of this class are not intended to be
* serializable.
*/
private void readObject(ObjectInputStream in)
throws IOException, ClassNotFoundException {
@@ -1287,16 +1282,14 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* Once a change occurs to a map, map replication requires that these changes are picked up by another thread, this
* class provides an iterator like interface to poll for such changes.
* <p/>
* <p>In most cases the thread that adds data to the node is unlikely to be the same thread that replicates the data
* over to the other nodes, so data will have to be marshaled between the main thread storing data to the map, and
* the thread running the replication.
* <p/>
* <p>One way to perform this marshalling, would be to pipe the data into a queue. However, This class takes another
* approach. It uses a bit set, and marks bits which correspond to the indexes of the entries that have changed. It
* then provides an iterator like interface to poll for such changes.
* Once a change occurs to a map, map replication requires that these changes are picked up by another
* thread, this class provides an iterator like interface to poll for such changes. <p/> <p>In most cases
* the thread that adds data to the node is unlikely to be the same thread that replicates the data over
* to the other nodes, so data will have to be marshaled between the main thread storing data to the map,
* and the thread running the replication. <p/> <p>One way to perform this marshalling, would be to pipe
* the data into a queue. However, This class takes another approach. It uses a bit set, and marks bits
* which correspond to the indexes of the entries that have changed. It then provides an iterator like
* interface to poll for such changes.
*
* @author Rob Austin.
*/
@@ -1375,8 +1368,8 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
/**
* you can continue to poll hasNext() until data becomes available. If are are in the middle of processing an
* entry via {@code nextEntry}, hasNext will return true until the bit is cleared
* you can continue to poll hasNext() until data becomes available. If are are in the middle of
* processing an entry via {@code nextEntry}, hasNext will return true until the bit is cleared
*
* @return true if there is an entry
*/
@@ -1444,16 +1437,16 @@ class ReplicatedChronicleMap<K, V> extends VanillaChronicleMap<K, V>
}
/**
* Always throws {@code NotSerializableException} since instances of this class
* are not intended to be serializable.
* Always throws {@code NotSerializableException} since instances of this class are not intended to be
* serializable.
*/
private void writeObject(ObjectOutputStream out) throws IOException {
throw new NotSerializableException(getClass().getCanonicalName());
}
/**
* Always throws {@code NotSerializableException} since instances of this class
* are not intended to be serializable.
* Always throws {@code NotSerializableException} since instances of this class are not intended to be
* serializable.
*/
private void readObject(ObjectInputStream in)
throws IOException, ClassNotFoundException {