From f9c40680e940863f9d1e79a030a7aef0713c243b Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 5 Feb 2016 10:25:29 +0000 Subject: [PATCH] exposing the boostrap timestamp --- .../chronicle/map/ChannelProvider.java | 22 +++++++++++++++++++ .../net/openhft/chronicle/map/Replica.java | 7 ++++++ .../chronicle/map/ReplicatedChronicleMap.java | 20 ++++------------- .../TCPSocketReplicationBootStrapTest.java | 4 ++++ 4 files changed, 37 insertions(+), 16 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 7b704655c..4139d880e 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -238,6 +238,23 @@ final class ChannelProvider implements Closeable { } notifier0 = modificationNotifier; } + + /** + * + * @return the earliest bootstrap timestamp across all the channels + */ + @Override + public long bootStrapTimeStamp() { + long boostapTimeStamp = Long.MAX_VALUE; + + for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { + Replica chronicleChannel = chronicleChannelList.get(i); + boostapTimeStamp = Math.min(boostapTimeStamp, chronicleChannel + .acquireModificationIterator(remoteIdentifier).bootStrapTimeStamp()); + } + + return boostapTimeStamp; + } }; modificationIterator.set((int) remoteIdentifier, result0); @@ -528,6 +545,11 @@ final class ChannelProvider implements Closeable { modificationNotifier0 = modificationNotifier; } + @Override + public long bootStrapTimeStamp() { + return 0; + } + @Override public void addPayload(Bytes bytes) { if (bytes.remaining() == 0) diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 3b41542a3..e11561dbf 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -125,6 +125,13 @@ interface Replica extends Closeable, EngineReplicationLangBytes { * @param modificationNotifier gets notified when a change occurs */ void setModificationNotifier(@NotNull final ModificationNotifier modificationNotifier); + + /** + * @return 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. + */ + long bootStrapTimeStamp(); } /** diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 8d854036d..c48b4d5ae 100755 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -88,10 +88,9 @@ import static net.openhft.lang.io.NativeBytes.wrap; * 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 + * @param the entries key type ยง * @param the entries value type */ -public class ReplicatedChronicleMap, +class ReplicatedChronicleMap, V, VI, MVI extends MetaBytesInterop> extends VanillaChronicleMap implements Replica, EntryExternalizable, EntryResolver, EngineReplicationLangBytes { @@ -125,13 +124,6 @@ public class ReplicatedChronicleMap 0 && source.remaining() > 0) { - destination.writeByte(source.readByte()); - } - } - private int assignedModIterBitSetSizeInBytes() { return (int) CACHE_LINES.align(BYTES.alignAndConvert(127 + RESERVED_MOD_ITER, BITS), BYTES); } @@ -189,10 +181,6 @@ public class ReplicatedChronicleMap) ChronicleMapBuilder.of(Integer.class, CharSequence.class) .replication((byte) 1, map1Config).create(); + final long bootStrapTimeStamp = map1.acquireModificationIterator((byte) 1).bootStrapTimeStamp(); + System.out.println("bootStrapTimeStamp=" + bootStrapTimeStamp); + + File persistenceFile = getPersistenceFile(); TcpTransportAndNetworkConfig map2Config = TcpTransportAndNetworkConfig.of(8067)