diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index ce4d94227..ecbbdad9e 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -120,11 +120,6 @@ abstract class AbstractChannelReplicator implements Closeable { closeables.closeQuietly(channel); } - /** - * forces the TCP and UDP replicators to re-bootstrap - * This is called whenever a new CHM is added to a custer - */ - public abstract void forceBootstrap(); void checkThrottleInterval() throws ClosedChannelException { if (throttler != null) diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 294c054e9..9645445bf 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -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 - *

- *

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 @@ -187,8 +183,8 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } /** - * 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 extends VanillaChronicleMap } @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 extends VanillaChronicleMap } /** - * 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 extends VanillaChronicleMap } @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 extends VanillaChronicleMap } @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 extends VanillaChronicleMap } /** - * 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 extends VanillaChronicleMap } /** - * 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 extends VanillaChronicleMap } /** - * 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 @@ -851,19 +846,19 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } /** - * 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 (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(Bytes, Object, boolean) @@ -1149,8 +1144,8 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } /** - * 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; @@ -1267,16 +1262,16 @@ class ReplicatedChronicleMap extends VanillaChronicleMap } /** - * 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 extends VanillaChronicleMap } /** - * 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. */ @@ -1375,8 +1368,8 @@ class ReplicatedChronicleMap extends VanillaChronicleMap /** - * 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 extends VanillaChronicleMap } /** - * 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 0dda8b663..f9da96941 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -23,21 +23,22 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; import java.io.IOException; -import java.net.*; +import java.net.ConnectException; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.SocketException; import java.nio.ByteBuffer; import java.nio.channels.*; import java.util.BitSet; -import java.util.Map; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import static java.nio.channels.SelectionKey.*; import static java.util.concurrent.TimeUnit.MILLISECONDS; /** - * Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a - * socket connection

{@see net.openhft.map.OutSocketReplicator} + * Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a socket + * connection

{@see net.openhft.map.OutSocketReplicator} * * @author Rob Austin. */ @@ -64,9 +65,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { /** * @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. + * 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. * @throws IOException */ TcpReplicator(@NotNull final Replica replica, @@ -295,10 +296,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { closeEarlyAndQuietly(key.channel()); } - @Override - public void forceBootstrap() { - forceBootstrap.lazySet(true); - } /** * called when the selector receives a OP_CONNECT message @@ -522,9 +519,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { /** * 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 + * 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 { diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 3d91081ee..6c4faef0a 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -81,6 +81,9 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific serverConnector = new ServerConnector(); + modificationIterator = replica.acquireModificationIterator( + ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this); + start(); } @@ -90,13 +93,6 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific super.close(); } - - @Override - public void forceBootstrap() { - modificationIterator = replica.acquireModificationIterator( - ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this); - } - /** * binds to the server socket and process data This method will block until interrupted */ diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index c4a24be82..59ce65c1e 100644 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -478,7 +478,8 @@ class VanillaChronicleMap extends AbstractMap * @param newValue the new value you wish to store in the map * @return the value that was replaced */ - 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);