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-191 -Checking in a possible fix for this issue - "EventListener is not called correctly for a replicated map", but will require more testing.

Browse Source
This commit is contained in:
Rob Austin
2014-11-06 18:28:38 +00:00
parent 6e60e8cda4
commit babd62e431
3 changed files with 236 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+121 -80
View File
@@ -45,39 +45,36 @@ import static net.openhft.lang.MemoryUnit.*;
import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
/**
* <h2>A Replicating Multi Master HashMap</h2> <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>
* <h2>Reconciliation </h2> <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. </p>
* <h2>A Replicating Multi Master HashMap</h2> <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> <h2>Reconciliation </h2> <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. </p>
*
* @param <K> the entries key type
* @param <V> the entries value type
@@ -197,14 +194,26 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
return modificationDelegator;
}
/**
* when a remote node receives and event it can't use the {@link ReplicatedChronicleMap#eventListener() }
* as this will trigger the modification iterators to send out another replication event. So instead with
* call this method iterators
*
* @return
*/
MapEventListener<K, V, ChronicleMap<K, V>> eventListenerWithoutModIterators() {
return eventListener;
}
@Override
public V put(K key, V value) {
return put0(key, value, true, localIdentifier, timeProvider.currentTimeMillis());
}
/**
* 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
@@ -267,14 +276,14 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
int segmentNum = getSegment(hash);
long segmentHash = segmentHash(hash);
boolean create =lockTypeType == LockType.WRITE_LOCK;
boolean create = lockTypeType == LockType.WRITE_LOCK;
Segment segment = segment(segmentNum);
MutableLockedEntry<K, V, MKI, KI> lock = (lockTypeType == LockType.WRITE_LOCK)
? segment.writeLock()
: segment.readLock();
V v = segment.acquireWithoutLock(copies, metaKeyWriter, keyWriter, key,
V v = segment.acquireWithoutLock(copies, metaKeyWriter, keyWriter, key,
value,
segmentHash, create, create ? timeProvider.currentTimeMillis() : 0L);
@@ -293,17 +302,14 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
@Override
public V remove(final Object key) {
return removeIfValueIs(key, null, localIdentifier, timeProvider.currentTimeMillis());
}
/**
* 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
@@ -390,8 +396,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* 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,
@@ -465,8 +471,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* 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) {
@@ -536,6 +542,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
source.position(valuePosition);
LOG.debug(message + "value=" + source.toString().trim() + ")");
}
}
@NotNull
@@ -720,6 +728,32 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
void notifyRemotePut(long offset, boolean added, final long pos, long keySize, final Bytes inBytes,
final long valueSize, final long valuePosition, final long keyPosition) {
if (eventListener() != MapEventListeners.NOP) {
final MultiStoreBytes b = acquireTmpBytes();
b.storePositionAndSize(bytes, offset, entrySize);
inBytes.limit(inBytes.capacity());
// deserialize the key
final ThreadLocalCopies copies = keyReaderProvider.getCopies(null);
inBytes.position(keyPosition);
final K key = keyReaderProvider.get(copies, originalKeyReader).read(inBytes, keySize);
// deserialize the value
final ThreadLocalCopies copiesV = valueReaderProvider.getCopies(copies);
inBytes.limit(inBytes.capacity());
inBytes.position(valuePosition);
final V value = valueReaderProvider.get(copiesV, originalValueReader).read(inBytes, valueSize);
eventListener().onPut(ReplicatedChronicleMap.this, b, metaDataBytes,
added, key, value, pos, this);
}
}
/**
* called from a remote node when it wishes to propagate a remove event
*/
@@ -736,9 +770,12 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
long offset = offsetFromPos(pos);
NativeBytes entry = entry(offset);
long keyPosition = inBytes.position();
if (!remoteKeyEquals(inBytes, keySize, entry))
continue;
// key is found
entry.skip(keySize);
final long timeStampPos = entry.positionAddr();
@@ -772,6 +809,10 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
hashLookupLiveOnly.put(hash2, pos);
incrementSize();
}
// this is just like notifyPut, but it won't trigger the modification iterators
notifyRemotePut(offset, false, posFromOffset(offset), keySize, inBytes, valueSize,
valuePos, keyPosition);
return;
}
@@ -785,6 +826,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
keySizeMarshaller.writeSize(entry, keySize);
// write the key
long kPosition = inBytes.position();
entry.write(inBytes);
entry.writeLong(timestamp);
@@ -805,6 +847,9 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
incrementSize();
// this is just like notifyPut, but it won't trigger the modification iterators
notifyRemotePut(offset, true, posFromOffset(offset), keySize, inBytes, valueSize, valuePos, kPosition);
} finally {
writeUnlock();
}
@@ -876,9 +921,9 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* 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>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>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
@@ -906,16 +951,15 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* Puts entry. If {@code value} implements {@link 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 Byteable} interface and {@code usingValue} is {@code
* true}, the value is backed with the bytes of this entry.
*
* @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 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
@@ -1216,8 +1260,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* 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;
@@ -1329,16 +1373,16 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* 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 {
@@ -1349,15 +1393,14 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* <p>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. </p>
* <p>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. </p>
*
* @author Rob Austin.
*/
@@ -1394,8 +1437,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
*
* @param segmentIndex the index of the maps segment
* @param pos the position within this {@code segmentIndex}
* @return and index the has combined the {@code segmentIndex} and {@code pos} into a
* single value
* @return and index the has combined the {@code segmentIndex} and {@code pos} into a single value
*/
private long combine(int segmentIndex, long pos) {
return (((long) segmentIndex) << segmentIndexShift) | pos;
@@ -1437,9 +1479,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
/**
* 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
*/
@@ -1507,16 +1548,16 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
/**
* 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 {
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+17 -18
View File
@@ -51,8 +51,8 @@ interface Work {
}
/**
* Used with a {@link net.openhft.chronicle.map.ReplicatedChronicleMap} to send data between the maps using a socket
* connection {@link net.openhft.chronicle.map.TcpReplicator}
* Used with a {@link net.openhft.chronicle.map.ReplicatedChronicleMap} to send data between the maps using a
* socket connection {@link net.openhft.chronicle.map.TcpReplicator}
*
* @author Rob Austin.
*/
@@ -79,13 +79,12 @@ public class TcpReplicator extends AbstractChannelReplicator implements Closeabl
private long selectorTimeout;
/**
* @param maxEntrySizeBytes used to check that the last entry will fit into the buffer,
* it can not be smaller than the size of and entry, if it is
* set smaller the buffer will over flow, it can be larger then
* the entry, but setting it too large reduces the workable
* space in the buffer.
* @param maxEntrySizeBytes used to check that the last entry will fit into the buffer, it can not
* be smaller than the size of and entry, if it is set smaller the buffer
* will over flow, it can be larger then the entry, but setting it too
* large reduces the workable space in the buffer.
* @param statelessServerConnector set to NULL if not required
* @throws IOException on an io error.
* @throws IOException on an io error.
*/
public TcpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable externalizable,
@@ -425,8 +424,8 @@ public class TcpReplicator extends AbstractChannelReplicator implements Closeabl
}
/**
* this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear all the old
* bootstrap information
* this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear all the
* old bootstrap information
*
* @param key the nio SelectionKey
*/
@@ -481,7 +480,6 @@ public class TcpReplicator extends AbstractChannelReplicator implements Closeabl
final IdentifierListener identifierListener = replicationConfig.identifierListener;
final SocketAddress remoteAddress = socketChannel.getRemoteAddress();
if ((identifierListener != null && !identifierListener.isIdentifierUnique(remoteIdentifier,
remoteAddress)) || remoteIdentifier == localIdentifier)
@@ -618,9 +616,10 @@ public class TcpReplicator extends AbstractChannelReplicator implements Closeabl
}
/**
* sets interestOps to "selector keys",The change to interestOps much be on the same thread as the selector. This
* class, allows via {@link AbstractChannelReplicator .KeyInterestUpdater#set(int)} to holds a pending change in
* interestOps ( via a bitset ), this change is processed later on the same thread as the selector
* sets interestOps to "selector keys",The change to interestOps much be on the same thread as the
* selector. This class, allows via {@link AbstractChannelReplicator .KeyInterestUpdater#set(int)} to
* holds a pending change in interestOps ( via a bitset ), this change is processed later on the same
* thread as the selector
*/
private static class KeyInterestUpdater {
@@ -651,8 +650,8 @@ public class TcpReplicator extends AbstractChannelReplicator implements Closeabl
}
/**
* @param keyIndex the index of the key that has changed, the list of keys is provided by the constructor {@link
* KeyInterestUpdater(int, SelectionKey[])}
* @param keyIndex the index of the key that has changed, the list of keys is provided by the
* constructor {@link KeyInterestUpdater(int, SelectionKey[])}
*/
public void set(int keyIndex) {
changeOfOpWriteRequired.set(keyIndex);
@@ -988,8 +987,8 @@ public class TcpReplicator extends AbstractChannelReplicator implements Closeabl
/**
* removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE will get added back
* in as soon as we have data to write
* removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE will get
* added back in as soon as we have data to write
*
* @param socketChannel the socketChannel we wish to stop writing to
* @param attached data associated with the socketChannels key