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

CE-57 adding support for engine to engine replication

Browse Source
This commit is contained in:
Rob Austin
2015-06-11 16:10:23 +01:00
parent e6582b0a39
commit 261d315688
4 changed files with 377 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pom.xml
+8 -13
View File
@@ -43,6 +43,13 @@
<scope>import</scope>
</dependency>
<dependency>
<groupId>net.openhft</groupId>
<artifactId>chronicle-bom</artifactId>
<type>pom</type>
<version>1.3.3-SNAPSHOT</version>
<scope>import</scope>
</dependency>
<dependency>
<groupId>net.openhft</groupId>
@@ -50,23 +57,12 @@
<version>2.2</version>
</dependency>
<dependency>
<groupId>net.openhft</groupId>
<artifactId>compiler</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>net.openhft</groupId>
<artifactId>lang</artifactId>
<version>6.6.7</version>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>net.openhft</groupId>
<artifactId>compiler</artifactId>
@@ -75,7 +71,6 @@
<dependency>
<groupId>net.openhft</groupId>
<artifactId>lang</artifactId>
</dependency>
<dependency>
src/main/java/net/openhft/chronicle/map/EngineReplicationLangBytes.java
+184
View File
@@ -0,0 +1,184 @@
/*
* 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.lang.io.Bytes;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.io.Closeable;
/**
* A simpler form of the replication interface used by Engine
*
* *** PLEASE DON'T REFACTOR THIS WITHOUT FIRST CHANGING THE ENGINE CODE ***
*
* @author Rob Austin.
*/
public interface EngineReplicationLangBytes extends Closeable {
/**
* inserts or updates the entry
*
* @param key the key of the entry
* @param value the value of the entry
* @param remoteIdentifier the identifier of the remote server
* @param timestamp the timestamp send from the remote server, this time stamp was the
* time the entry was removed
*/
void put(Bytes key, Bytes value, byte remoteIdentifier, long timestamp);
/**
* marks the entry as removed
*
* @param key the key of the entry
* @param remoteIdentifier the identifier of the remote server
* @param timestamp the timestamp send from the remote server, this time stamp was the
* time the entry was removed
*/
void remove(Bytes key, byte remoteIdentifier, long timestamp);
/**
* Provides the unique Identifier associated with this instance. <p> An identifier is used to
* determine which replicating node made the change. <p> 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.
*
* @return the unique Identifier associated with this map instance
*/
byte identifier();
/**
* Gets (if it does not exist, creates) an instance of ModificationIterator associated with a
* remote node, this weak associated is bound using the {@code identifier}.
*
* @param remoteIdentifier the identifier of the remote node
* @return the ModificationIterator dedicated for replication to the remote node with the given
* identifier
* @see #identifier()
*/
EngineModificationIterator acquireEngineModificationIterator(byte remoteIdentifier);
/**
* Returns the timestamp of the last change from the specified remote node, already replicated
* to this host. <p>Used in conjunction with replication, to back fill data from a remote node.
* This node may have missed updates while it was not been running or connected via TCP.
*
* @param remoteIdentifier the identifier of the remote node to check last replicated update
* time from
* @return a timestamp of the last modification to an entry, or 0 if there are no entries.
* @see #identifier()
*/
long lastModificationTime(byte remoteIdentifier);
void setLastModificationTime(byte identifier, long timestamp);
/**
* notifies when there is a changed to the modification iterator
*/
interface EngineReplicationModificationNotifier {
EngineReplicationModificationNotifier NOP = new EngineReplicationModificationNotifier() {
@Override
public void onChange() {
}
};
/**
* called when ever there is a change applied to the modification iterator, in otherwords
* when {@link EngineModificationIterator#hasNext()} will return true
*/
void onChange();
}
/**
* Holds a record of which entries have modification. Each remote map supported will require a
* corresponding ModificationIterator instance
*/
interface EngineModificationIterator {
/**
* @return {@code true} if the is another entry to be received via {@link
* EngineModificationIterator#nextEntry(EngineReplicationLangBytes.EngineEntryCallback)}
*/
boolean hasNext();
/**
* A non-blocking call that provides the entry that has changed to {@code
* callback.onEntry()}.
*
* @param callback a callback which will be called when a new entry becomes available.
* @return {@code true} if the entry was accepted by the {@code callback.onEntry()} method,
* {@code false} if the entry was not accepted or was not available
*/
boolean nextEntry(@NotNull final EngineEntryCallback callback) throws InterruptedException;
/**
* Dirties all entries with a modification time equal to {@code fromTimeStamp} or newer. It
* means all these entries will be considered as "new" by this ModificationIterator and
* iterated once again no matter if they have already been. <p>This functionality is used
* to publish recently modified entries to a new remote node as it connects.
*
* @param fromTimeStamp the timestamp from which all entries should be dirty
*/
void dirtyEntries(long fromTimeStamp) throws InterruptedException;
/**
* the {@code modificationNotifier} is called when ever there is a change applied to the
* modification iterator
*
* @param modificationNotifier gets notified when a change occurs
*/
void setModificationNotifier(@NotNull final EngineReplicationModificationNotifier modificationNotifier);
}
/**
* Implemented typically by a replicator, This interface provides the event, which will get
* called whenever a put() or remove() has occurred to the map
*/
interface EngineEntryCallback {
/**
* Called whenever a put() or remove() has occurred to a replicating map.
*
* @param key the key on the entry
* @param value the value of the entry, will be null if the entry {@code
* isDeleted == true}
* @param isDeleted {@code true} if remove is called otherwise {@code false}
* @param bootStrapTimeStamp sent to the client on every update this is the timestamp that
* the remote client should bootstrap from when there has been a
* disconnection, this time maybe later than the message time as
* event are not send in chronological order from the bit set.
* @return {@code false} if this entry should be ignored because the identifier of the
* source node is not from one of our changes, WARNING even though we check the identifier
* in the ModificationIterator the entry may have been updated.
*/
boolean onEntry(@NotNull Bytes key,
@Nullable Bytes value, long timestamp,
byte identifier,
boolean isDeleted,
long bootStrapTimeStamp);
}
}
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+183 -5
View File
@@ -28,10 +28,7 @@ import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
import net.openhft.lang.Maths;
import net.openhft.lang.collection.ATSDirectBitSet;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.CheckedBytes;
import net.openhft.lang.io.MultiStoreBytes;
import net.openhft.lang.io.NativeBytes;
import net.openhft.lang.io.*;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -40,6 +37,7 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
@@ -50,6 +48,7 @@ import java.util.concurrent.atomic.AtomicReferenceArray;
import static net.openhft.chronicle.hash.hashing.Hasher.hash;
import static net.openhft.lang.MemoryUnit.*;
import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
import static net.openhft.lang.io.NativeBytes.wrap;
/**
* <h2>A Replicating Multi Master HashMap</h2> <p>Each remote hash map, mirrors its changes over to
@@ -92,7 +91,7 @@ import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;
final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
V, VI, MVI extends MetaBytesInterop<V, ? super VI>>
extends VanillaChronicleMap<K, KI, MKI, V, VI, MVI>
implements Replica, Replica.EntryExternalizable, Replica.EntryResolver<K, V> {
implements Replica, Replica.EntryExternalizable, Replica.EntryResolver<K, V>, EngineReplicationLangBytes {
// for file, jdbc and UDP replication
public static final int RESERVED_MOD_ITER = 8;
public static final int ADDITIONAL_ENTRY_BYTES = 10;
@@ -348,11 +347,190 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
super.close();
}
private ThreadLocal<IByteBufferBytes> buffer = new ThreadLocal<IByteBufferBytes>() {
@Override
protected IByteBufferBytes initialValue() {
return ByteBufferBytes.wrap(ByteBuffer.allocate(1024));
}
};
private Bytes threadlocalBuffer(long requiredSize) {
final IByteBufferBytes bytes = buffer.get();
bytes.clear();
bytes.buffer().clear();
if (bytes.buffer().remaining() < requiredSize) {
if (requiredSize > Integer.MAX_VALUE)
throw new IllegalStateException("entry is too large");
IByteBufferBytes result = ByteBufferBytes.wrap(ByteBuffer.allocate((int) requiredSize));
buffer.set(result);
return result;
}
return bytes;
}
private static class EntryBuffer {
NativeBytes key;
NativeBytes value;
NativeBytes getKey() {
return key;
}
void setKey(final NativeBytes key) {
this.key = key;
}
NativeBytes getValue() {
return value;
}
void setValue(final NativeBytes value) {
this.value = value;
}
}
@Override
public void put(final Bytes key, final Bytes value, final byte id, final long timestamp) {
final ThreadLocalCopies copies = SegmentState.getCopies(null);
final SegmentState segmentState = SegmentState.get(copies);
long keySize = key.remaining();
long valueSize = key.remaining();
final long requiredBufferSize = 8 + keySize + 8 + valueSize + 12;
final Bytes source = threadlocalBuffer(requiredBufferSize);
keySizeMarshaller.writeSize(source, keySize);
valueSizeMarshaller.writeSize(source, keySize);
source.writeLong(timestamp);
source.writeByte(id);
source.writeBoolean(false);
// key.bytes().
writeTo(source, key);
writeTo(source, value);
readExternalEntry(copies, segmentState, source);
}
private static void writeTo(Bytes destination, Bytes source) {
while (destination.remaining() > 0 && source.remaining() > 0) {
source.writeByte(source.readByte());
}
}
@Override
public void remove(final Bytes key,
final byte remoteIdentifier,
final long timestamp) {
final ThreadLocalCopies copies = SegmentState.getCopies(null);
final SegmentState segmentState = SegmentState.get(copies);
long keySize = key.remaining();
long valueSize = key.remaining();
final long requiredBufferSize = 8 + keySize + 8 + valueSize + 12;
final Bytes source = threadlocalBuffer(requiredBufferSize);
keySizeMarshaller.writeSize(source, keySize);
valueSizeMarshaller.writeSize(source, keySize);
source.writeLong(timestamp);
source.writeByte(remoteIdentifier);
source.writeBoolean(true);
writeTo(source, key);
readExternalEntry(copies, segmentState, source);
}
@Override
public byte identifier() {
return localIdentifier;
}
@Override
public EngineModificationIterator acquireEngineModificationIterator(final byte remoteIdentifier) {
final ModificationIterator modificationIterator = acquireModificationIterator(remoteIdentifier);
return new EngineModificationIterator() {
@Override
public boolean hasNext() {
return modificationIterator.hasNext();
}
@Override
public boolean nextEntry(@NotNull final EngineEntryCallback callback) throws InterruptedException {
return modificationIterator.nextEntry(new EntryCallback() {
@Override
public boolean onEntry(@NotNull final Bytes entry, final int chronicleId, final long bootStrapTimeStamp) {
final long keySize = keySizeMarshaller.readSize(entry);
final long keyPosition = entry.position();
entry.skip(keySize);
final long timestamp = entry.readLong();
final byte identifier = entry.readByte();
final boolean isDeleted = entry.readBoolean();
if (isDeleted)
return callback.onEntry(wrap(keyPosition, keySize), null, timestamp, identifier,
true,
bootStrapTimeStamp);
long valueSize = valueSizeMarshaller.readSize(entry);
alignment.alignPositionAddr(entry);
final long valuePosition = entry.position();
return callback.onEntry(wrap(keyPosition, keySize), wrap(valuePosition,
valueSize),
timestamp,
identifier,
false,
bootStrapTimeStamp);
}
@Override
public boolean shouldBeIgnored(final Bytes entry, final int chronicleId) {
return false; // do nothing
}
}, 0);
}
@Override
public void dirtyEntries(final long fromTimeStamp) throws InterruptedException {
modificationIterator.dirtyEntries(fromTimeStamp);
}
@Override
public void setModificationNotifier(
@NotNull final EngineReplicationModificationNotifier modificationNotifier) {
modificationIterator.setModificationNotifier(toModificationNotifier());
}
private ModificationNotifier toModificationNotifier() {
return null;
}
};
}
@Override
public ModificationIterator acquireModificationIterator(byte remoteIdentifier) {
ModificationIterator modificationIterator = modificationIterators.get(remoteIdentifier);
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+2 -1
View File
@@ -1326,7 +1326,8 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
// do nothing
}
static final class SegmentState implements StatefulCopyable<SegmentState>, AutoCloseable {
public static final class SegmentState implements StatefulCopyable<SegmentState>,
AutoCloseable {
private static final Provider<SegmentState> segmentStateProvider =
Provider.of(SegmentState.class);
private static final SegmentState originalSegmentState = new SegmentState(0);