From 01faf866d96d2af09009cb8ff880878b4a7e11cc Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 18 Nov 2014 15:30:52 +0000 Subject: [PATCH 1/3] HCOLL-134 more work on node discovery --- .../java/net/openhft/chronicle/map/ReplicationHubFindByName.java | 1 - 1 file changed, 1 deletion(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java b/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java index 24b4344c2..8d16cc7fa 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java @@ -64,7 +64,6 @@ class ReplicationHubFindByName implements FindByName { this.replicationHub = replicationHub; ReplicationChannel channel = replicationHub.createChannel((short) MAP_BY_NAME_CHANNEL); - MapEventListener listener = new MapEventListener() { @Override From 266da0be31319a7038cb6243b786c50d72bdc5d3 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 18 Nov 2014 18:40:01 +0000 Subject: [PATCH 2/3] HCOLL-134 added the following method - bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries) --- .../map/AbstractChronicleMapBuilder.java | 8 ++++++ .../chronicle/map/ChannelProvider.java | 3 -- .../chronicle/map/ChronicleMapBuilder.java | 16 ++++++++++- .../chronicle/map/ChronicleMapBuilderI.java | 11 ++++++++ .../openhft/chronicle/map/NodeDiscovery.java | 2 ++ .../OffHeapUpdatableChronicleMapBuilder.java | 2 ++ .../chronicle/map/ReplicatedChronicleMap.java | 28 +++++++++++++------ .../map/ReplicationHubFindByName.java | 4 +-- .../openhft/chronicle/map/MapByNameTest.java | 26 ++++++++--------- 9 files changed, 71 insertions(+), 29 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index 1eec457db..4935d72fd 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -142,6 +142,7 @@ abstract class AbstractChronicleMapBuilder prepareValueBytes = null; private SingleChronicleHashReplication singleHashReplication = null; + public boolean bootstapOnlyLocalEntries = false; AbstractChronicleMapBuilder(Class keyClass, Class valueClass) { keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); @@ -962,6 +963,13 @@ abstract class AbstractChronicleMapBuilder bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries) { + this.bootstapOnlyLocalEntries = bootstapOnlyLocalEntries; + return this; + } + public String name() { return this.name; } diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index dabc9d800..5d5779a1c 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -402,9 +402,6 @@ final class ChannelProvider implements Closeable { final PayloadProvider iterator = new PayloadProvider() { - - // todo it maybe possibel to replace this with a single atomic refernce, - // todo this way no aditional objects will be created final Queue payloads = new LinkedTransferQueue(); @Override diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index fe2e7d240..1226d21f5 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -492,7 +492,21 @@ public final class ChronicleMapBuilder implements ChronicleMapBuilderI createPersistedTo(File file) throws IOException { - return delegate.createPersistedTo(file); + return delegate.createPersistedTo(file); + } + + /** + * @param bootstapOnlyLocalEntries if set to true - when a new node joins a TCP replication grid, the new + * node will be populated with data, only for the nodes that created that + * data. Otherwise, all the nodes will publish all the data they have ( + * potentially swamping the new node with duplicates ) however this does + * guarantee that all the data is replicated over to the new node, and is + * useful especially in the case that the originating node is not + * currently running. + */ + public ChronicleMapBuilder bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries) { + delegate.bootstapOnlyLocalEntries(bootstapOnlyLocalEntries); + return this; } } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java index 4dddab982..bb765910e 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilderI.java @@ -296,6 +296,17 @@ interface ChronicleMapBuilderI extends Serializable { ChronicleMapBuilderI name(String name); + /** + * @param bootstapOnlyLocalEntries if set to true - when a new node joins a TCP replication grid, the new + * node will be populated with data, only for the nodes that created that + * data. Otherwise, all the nodes will publish all the data they have ( + * potentially swamping the new node with duplicates ) however this does + * guarantee that all the data is replicated over to the new node, and is + * useful especially in the case that the originating node is not + * currently running. + */ + ChronicleMapBuilderI bootstapOnlyLocalEntries(boolean bootstapOnlyLocalEntries); + String name(); diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 89d7c7c13..44d547f64 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -248,6 +248,7 @@ public class NodeDiscovery { final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(ourAddressAndPort.getPort(), toInetSocketCollection(knownHostPorts0)) + .heartBeatInterval(1, SECONDS); @@ -257,6 +258,7 @@ public class NodeDiscovery { final ReplicationHub replicationHub = ReplicationHub.builder() .maxEntrySize(10 * 1024) .tcpTransportAndNetwork(tcpConfig) + .remoteNodeValidator(remoteNodeValidator) .createWithId(identifier); return new ReplicationHubFindByName(replicationHub); diff --git a/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java index 68b4391fa..6ff8cec11 100644 --- a/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java @@ -163,6 +163,8 @@ final class OffHeapUpdatableChronicleMapBuilder throw new UnsupportedOperationException("not supported for this combination of key/value type"); } + + @Override public OffHeapUpdatableChronicleMapBuilder constantValueSizeBySample(V sampleValue) { throw new UnsupportedOperationException("not supported for this combination of key/value type"); diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index c68a813f3..49ee7fbe2 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -97,6 +97,7 @@ final class ReplicatedChronicleMap, private transient ModificationDelegator modificationDelegator; private transient long startOfModificationIterators; + private boolean bootstapOnlyLocalEntries; public ReplicatedChronicleMap(@NotNull AbstractChronicleMapBuilder builder, byte identifier) @@ -105,6 +106,7 @@ final class ReplicatedChronicleMap, this.timeProvider = builder.timeProvider(); this.localIdentifier = identifier; + this.bootstapOnlyLocalEntries = builder.bootstapOnlyLocalEntries; if (localIdentifier == -1) { throw new IllegalStateException("localIdentifier should not be -1"); @@ -821,8 +823,8 @@ final class ReplicatedChronicleMap, * called from a remote node when it wishes to propagate a remove event */ void remotePut(@NotNull final Bytes inBytes, long hash2, - final byte identifier, final long timestamp, - long valuePos, long valueLimit) { + final byte identifier, final long timestamp, + long valuePos, long valueLimit) { BytesReader valueReader = null; @@ -955,10 +957,10 @@ final class ReplicatedChronicleMap, return keySize == keySizeMarshaller.readSize(entry) && entry.startsWith(keyBytes); } - @Override + @Override V putWithoutLock(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V value, long hash2, boolean replaceIfPresent) { - return putWithoutLock(copies, metaKeyInterop, keyInterop, key, value, hash2, replaceIfPresent, - localIdentifier, timeProvider.currentTimeMillis()); + return putWithoutLock(copies, metaKeyInterop, keyInterop, key, value, hash2, replaceIfPresent, + localIdentifier, timeProvider.currentTimeMillis()); } V put(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, @@ -1231,8 +1233,16 @@ final class ReplicatedChronicleMap, anotherLookup.replace(hash, prevPos, pos); } + + /** + * @param timeStamp + * @param callback + * @param bootstapOnlyLocalEntries 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) { + final ModificationIterator.EntryModifiableCallback callback, + final boolean bootstapOnlyLocalEntries) { this.readLock(); try { @@ -1247,8 +1257,8 @@ final class ReplicatedChronicleMap, final long entryTimestamp = entry.readLong(); - if (entryTimestamp >= timeStamp && - entry.readByte() == ReplicatedChronicleMap.this.identifier()) + if (entryTimestamp >= timeStamp && (!bootstapOnlyLocalEntries || + entry.readByte() == ReplicatedChronicleMap.this.identifier())) callback.set(index, pos); } }); @@ -1655,7 +1665,7 @@ final class ReplicatedChronicleMap, // iterate over all the segments and mark bit in the modification iterator // that correspond to entries with an older timestamp for (final Segment segment : (Segment[]) segments) { - segment.dirtyEntries(fromTimeStamp, entryModifiableCallback); + segment.dirtyEntries(fromTimeStamp, entryModifiableCallback, bootstapOnlyLocalEntries); } } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java b/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java index 8d16cc7fa..25b6ee89a 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java @@ -43,7 +43,7 @@ class ReplicationHubFindByName implements FindByName { public static final int MAP_BY_NAME_CHANNEL = 1; - private final AtomicInteger nextFreeChannel = new AtomicInteger(1); + private final AtomicInteger nextFreeChannel = new AtomicInteger(2); private final Map map; private final ReplicationHub replicationHub; @@ -73,7 +73,7 @@ class ReplicationHubFindByName implements FindByName { }; this.map = (Map) of(CharSequence.class, ChronicleMapBuilderWithChannelId.class) - .entrySize(2500) + .entrySize(3000) .entries(128) .eventListener(listener) .instance() diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java index 07d11ab63..45c5d0c1d 100644 --- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java @@ -16,7 +16,6 @@ package net.openhft.chronicle.map; -import junit.framework.Assert; import net.openhft.chronicle.hash.replication.ReplicationHub; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import org.junit.Before; @@ -26,6 +25,8 @@ import org.junit.Test; import java.io.IOException; import java.util.concurrent.TimeoutException; +import static junit.framework.Assert.assertEquals; + /** * @author Rob Austin. */ @@ -45,22 +46,19 @@ public class MapByNameTest { } @Test - @Ignore public void testSerializingBuilder() throws TimeoutException, InterruptedException, IOException { - { - final ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .minSegments(2) - .name("map1") - .replication((byte) 1, TcpTransportAndNetworkConfig.of(8244)) - .removeReturnsNull(true); - // findMapByName.add(builder); - } + final ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(2) + .minSegments(2) + .name("map1") + .replication((byte) 1, TcpTransportAndNetworkConfig.of(8244)) + .removeReturnsNull(true); - final ChronicleMap map = findMapByName.from("map1"); + final ChronicleMap map = findMapByName.create(builder); map.put("hello", "world"); - Assert.assertEquals(map.get("hello"), "world"); + assertEquals(map.get("hello"), "world"); } // @Test(expected = IllegalArgumentException.class) @@ -102,8 +100,8 @@ public class MapByNameTest { .name("myMap4") .removeReturnsNull(true); - // ChronicleMap map = mapByName.create(builder); - // map.put("hello", "world"); + // ChronicleMap map = mapByName.create(builder); + // map.put("hello", "world"); final ChronicleMap map = mapByName.from("myMap4"); From c9cf9626172cf452c03ce98ca61a0732c22ea6c1 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 18 Nov 2014 19:54:54 +0000 Subject: [PATCH 3/3] HCOLL-134 added some logging --- .../java/net/openhft/chronicle/map/ChannelProvider.java | 7 +++++-- .../java/net/openhft/chronicle/map/MapByNameTest.java | 8 ++++---- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index 5d5779a1c..9a3571b0e 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -318,7 +318,10 @@ final class ChannelProvider implements Closeable { } private void add(int chronicleChannel, - Replica replica, EntryExternalizable entryExternalizable) { + Replica replica, + @NotNull EntryExternalizable entryExternalizable) { + + LOG.info("adding chronicleChannel=" + chronicleChannel + ",entryExternalizable=" + entryExternalizable); channelDataLock.writeLock().lock(); try { if (chronicleChannels[chronicleChannel] != null) { @@ -327,7 +330,7 @@ final class ChannelProvider implements Closeable { } chronicleChannels[chronicleChannel] = replica; chronicleChannelList.add(replica); - chronicleChannelIds.add((int) chronicleChannel); + chronicleChannelIds.add(chronicleChannel); channelEntryExternalizables[chronicleChannel] = entryExternalizable; if (chronicleChannel == 0) diff --git a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java index 45c5d0c1d..313b8d1ab 100644 --- a/src/test/java/net/openhft/chronicle/map/MapByNameTest.java +++ b/src/test/java/net/openhft/chronicle/map/MapByNameTest.java @@ -97,13 +97,13 @@ public class MapByNameTest { ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence .class, CharSequence.class) .minSegments(2) - .name("myMap4") + .name("myMap5") .removeReturnsNull(true); - // ChronicleMap map = mapByName.create(builder); - // map.put("hello", "world"); + ChronicleMap map = mapByName.create(builder); + map.put("hello", "world"); - final ChronicleMap map = mapByName.from("myMap4"); + // final ChronicleMap map = mapByName.from("myMap4"); // allow time for replication