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

adding net.openhft.chronicle.map.NoTcpReplicationSoakTest

Browse Source
This commit is contained in:
Rob Austin
2014-12-02 20:08:26 +00:00
parent 622a62f206
commit c4417ddbe4
6 changed files with 579 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+91 -59
View File
@@ -48,36 +48,39 @@ 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
@@ -207,7 +210,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
DelegatingMetaBytesInterop.<Bytes, BytesInterop<Bytes>>instance(),
BytesBytesInterop.INSTANCE, key, keySize, keyBytesToInstance,
getRemoteBytesValueInterops, value, valueBytesToInstance,
replaceIfPresent, readValue, false, localIdentifier, currentTime());;
replaceIfPresent, readValue, false, localIdentifier, currentTime());
;
}
/**
@@ -239,8 +243,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
* @param timeStamp the time that that change was made, this is used for replication
* @return the value that was replaced
*/
private V put0(K key, V value, boolean replaceIfPresent,
final byte identifier, long timeStamp) {
V put0(K key, V value, boolean replaceIfPresent,
final byte identifier, long timeStamp) {
checkKey(key);
checkValue(value);
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
@@ -254,6 +258,34 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
identifier, timeStamp);
}
Object remoteRemove(K key,
final byte remoteIdentifier, long timeStamp) {
checkKey(key);
K key1 = (K) key;
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
KI keyInterop = keyInteropProvider.get(copies, originalKeyInterop);
copies = metaKeyInteropProvider.getCopies(copies);
MKI metaKeyInterop =
metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key1);
long keySize = metaKeyInterop.size(keyInterop, key1);
long hash = metaKeyInterop.hash(keyInterop, key1);
int segmentNum = getSegment(hash);
long segmentHash = segmentHash(hash);
ReplicatedChronicleMap.Segment segment = segment(segmentNum);
segment.writeLock(null);
try {
return segment.removeWithoutLock(copies, null, metaKeyInterop, keyInterop, key1, keySize,
keyIdentity(), this, (V) null, valueIdentity(), segmentHash, this, removeReturnsNull,
timeStamp, remoteIdentifier, true, false);
} finally {
segment.writeUnlock();
}
}
@Override
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<RV, ? super VBI>>
@@ -380,8 +412,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* 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,
@@ -454,8 +486,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* 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(
@@ -471,7 +503,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
final byte remoteIdentifier;
if (id != 0) {
// isDeleted = false;
remoteIdentifier = id;
} else {
throw new IllegalStateException("identifier can't be 0");
@@ -1001,12 +1032,10 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* - if expectedValue is not null, returns Boolean.TRUE (removed) or
* Boolean.FALSE (entry not found), regardless the expectedValue object is Bytes instance
* (RPC call) or the value instance
* - if expectedValue is null:
* - if resultUnused is false, null or removed value is returned
* - if resultUnused is true, null is always returned
* - if expectedValue is not null, returns Boolean.TRUE (removed) or Boolean.FALSE (entry
* not found), regardless the expectedValue object is Bytes instance (RPC call) or the value
* instance - if expectedValue is null: - if resultUnused is false, null or removed value is
* returned - if resultUnused is true, null is always returned
*/
<KB, KBI, MKBI extends MetaBytesInterop<KB, ? super KBI>,
RV, VB extends RV, VBI, MVBI extends MetaBytesInterop<? super VB, ? super VBI>>
@@ -1133,8 +1162,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
long timestampPos = entry.position();
if (shouldIgnore(entry, timestamp, identifier)) {
LOG.error("Trying to replace a value for key={} on the node with id={} " +
"at time={} (current time), but the entry is updated by node " +
"with id={} at time={}. Time is not monotonic across nodes!?",
"at time={} (current time), but the entry is updated by node " +
"with id={} at time={}. Time is not monotonic across nodes!?",
key, identifier, timestamp, entry.readByte(entry.position() - 1),
entry.readLong(entry.position() - ADDITIONAL_ENTRY_BYTES + 1));
return readValue.readNull();
@@ -1163,8 +1192,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* @param bootstrapOnlyLocalEntries if true - only entries that have been create from
* this node will be published during a bootstrap
* @param bootstrapOnlyLocalEntries if true - only entries that have been create from this
* node will be published during a bootstrap
*/
public void dirtyEntries(final long timeStamp,
final ModificationIterator.EntryModifiableCallback callback,
@@ -1300,14 +1329,15 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* <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.
*/
@@ -1344,7 +1374,8 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
*
* @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;
@@ -1368,8 +1399,9 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
/**
* 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
*/
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+11 -8
View File
@@ -628,11 +628,14 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key);
try {
final int bytesJustWritten = attached.entryWriter.writeBufferToSocket(socketChannel,
final int len = attached.entryWriter.writeBufferToSocket(socketChannel,
approxTime);
if (bytesJustWritten > 0)
contemplateThrottleWrites(bytesJustWritten);
if (len == -1)
socketChannel.close();
if (len > 0)
contemplateThrottleWrites(len);
if (!attached.entryWriter.hasBytesToWrite() && !attached.entryWriter.isWorkIncomplete())
// TURN OP_WRITE_OFF
@@ -659,6 +662,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
int len = attached.entryReader.readSocketToBuffer(socketChannel);
if (len == -1) {
socketChannel.register(selector, 0);
if (replicationConfig.autoReconnectedUponDroppedConnection()) {
AbstractConnector connector = attached.connector;
if (connector != null)
@@ -1018,6 +1022,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
// if we still have some unwritten writer from last time
lastSentTime = approxTime;
assert in.position() <= Integer.MAX_VALUE;
int size = (int) in.position();
out.limit(size);
@@ -1027,7 +1032,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
LOG.debug("bytes-written=" + len);
if (len == size) {
// assert len == size;
out.clear();
in.clear();
} else {
@@ -1195,9 +1199,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
final long limit = out.limit();
out.limit(nextEntryPos);
boolean isStateless = (state != 1);
boolean isStatelessClient = (state != 1);
if (isStateless) {
if (isStatelessClient) {
if (statelessServerConnector == null) {
LOG.error("", new IllegalArgumentException("received an event " +
"from a stateless map, stateless maps are not " +
@@ -1247,6 +1251,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
return;
in.limit(in.position());
assert out.position() < Integer.MAX_VALUE;
in.position((int) out.position());
in.compact();
@@ -1268,8 +1273,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
return out.readLong();
else
return Long.MIN_VALUE;
// return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE;
}
public long remoteHeartbeatIntervalFromBuffer() {