diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index d86a193eb..d4d54fc67 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -45,39 +45,36 @@ import static net.openhft.lang.MemoryUnit.*; import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND; /** - *

A Replicating Multi Master HashMap

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.

- *

Reconciliation

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.

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

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.

Reconciliation

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.

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 the entries key type * @param the entries value type @@ -197,14 +194,26 @@ class ReplicatedChronicleMap, 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> 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, 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 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, } - - - @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, } /** - * 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, } /** - * 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, source.position(valuePosition); LOG.debug(message + "value=" + source.toString().trim() + ")"); } + + } @NotNull @@ -720,6 +728,32 @@ class ReplicatedChronicleMap, } + 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, 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, 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, keySizeMarshaller.writeSize(entry, keySize); // write the key + long kPosition = inBytes.position(); entry.write(inBytes); entry.writeLong(timestamp); @@ -805,6 +847,9 @@ class ReplicatedChronicleMap, 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, } /** - * 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. - *

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.

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, } /** - * 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 (this could be later if - * it was a remote put) this method is called only from usual - * put or acquire + * @param timestamp the timestamp when the entry was put (this could be later if it was a + * remote put) 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, } /** - * 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> { private static final long serialVersionUID = 0L; @@ -1329,16 +1373,16 @@ class ReplicatedChronicleMap, } /** - * 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, } /** - *

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. - *

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.

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.

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.

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. */ @@ -1394,8 +1437,7 @@ class ReplicatedChronicleMap, * * @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, /** - * 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, } /** - * 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 { diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 087e44180..1613ba88b 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -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 diff --git a/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java new file mode 100644 index 000000000..6ae09b72a --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/EventListenerTestWithTCPSocketReplication.java @@ -0,0 +1,98 @@ +/* + * Copyright 2014 Higher Frequency Trading + *

+ * http://www.higherfrequencytrading.com + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.TcpReplicationConfig; +import net.openhft.lang.io.Bytes; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.concurrent.atomic.AtomicBoolean; +/** + * Test ReplicatedChronicleMap where the Replicated is over a TCP Socket + * + * @author Rob Austin. + */ + +public class EventListenerTestWithTCPSocketReplication { + + + + @Test + public void test() throws IOException, InterruptedException { + + ChronicleMap map1; + ChronicleMap map2; + final AtomicBoolean putWasCalled = new AtomicBoolean(false); + + int udpPort = 1234; + MapEventListener> eventListener = new MapEventListener>() { + private static final long serialVersionUID = 1L; + + @Override + public void onPut(ChronicleMap map, Bytes entry, int metaDataBytes, boolean added, Integer key, + CharSequence value) { + putWasCalled.getAndSet(true); + } + }; + + // ---------- SERVER1 1 ---------- + { + + // we connect the maps via a TCP socket connection on port 8077 + + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8076, new InetSocketAddress("localhost", 8077)); + ChronicleMapBuilder map1Builder = + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replicators((byte) 1, tcpConfig); + + map1 = map1Builder.create(); + } + // ---------- SERVER2 2 on the same server as ---------- + + { + TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8077); + ChronicleMapBuilder map2Builder = + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replicators((byte) 2, tcpConfig) + .eventListener(eventListener); + + map2 = map2Builder.create(); + } + + // we will stores some data into one map here + map1.put(5, "EXAMPLE"); + int t = 0; + for (; t < 5000; t++) { + if (map1.equals(map2)) { + break; + } + Thread.sleep(1); + } + + Assert.assertEquals("Maps should be equal", map1, map2); + Assert.assertTrue("Map 1 should not be empty", !map1.isEmpty()); + Assert.assertTrue("Map 2 should not be empty", !map2.isEmpty()); + Assert.assertTrue("The eventListener.onPut should have been called", putWasCalled.get()); + } + } \ No newline at end of file