diff --git a/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java b/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java index c7a984793..c286883c2 100644 --- a/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java +++ b/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java @@ -35,7 +35,7 @@ public class DiscoveryCluster { final UdpReplicationConfig udpConfig = UdpReplicationConfig .simple(Inet4Address.getByName("255.255.255.255"), udpBroadcastPort); - final RemoteNodes remoteNodes = new RemoteNodes(); + final KnownNodes knownNodes = new KnownNodes(); final DirectBitSet knownIdentifiers = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate (128 / 8))); @@ -48,13 +48,13 @@ public class DiscoveryCluster { final UDPEventListener udpEventListener = new UDPEventListener() { @Override - public void onRemoteNodeEvent(RemoteNodes remoteNode, ConcurrentExpiryMap proposedIdentifiersWithHost) { + public void onRemoteNodeEvent(KnownNodes remoteNode, ConcurrentExpiryMap proposedIdentifiersWithHost) { knownHostPorts.addAll(remoteNode.addressAndPorts()); orBitSets(remoteNode.activeIdentifierBitSet(), knownIdentifiers); - for (BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost : + for (BytesExternalizableImpl.ProposedNodes proposedIdentifierWithHost : proposedIdentifiersWithHost.values()) { if (!proposedIdentifierWithHost.addressAndPort().equals(ourAddressAndPort)) { @@ -71,10 +71,10 @@ public class DiscoveryCluster { }; - final BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener); + final BytesExternalizableImpl externalizable = new BytesExternalizableImpl(knownNodes, udpEventListener); - final BytesExternalizableImpl.ProposedIdentifierWithHost ourHostPort = new - BytesExternalizableImpl.ProposedIdentifierWithHost(ourAddressAndPort, (byte) -1); + final BytesExternalizableImpl.ProposedNodes ourHostPort = new + BytesExternalizableImpl.ProposedNodes(ourAddressAndPort, (byte) -1); // to start with we will send a bootstrap that just contains our hostname without and identifier @@ -99,16 +99,16 @@ public class DiscoveryCluster { isFistTime = false; - final BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost = new - BytesExternalizableImpl.ProposedIdentifierWithHost(ourAddressAndPort, identifier); + final BytesExternalizableImpl.ProposedNodes proposedNodes = new + BytesExternalizableImpl.ProposedNodes(ourAddressAndPort, identifier); - externalizable.sendBootStrap(proposedIdentifierWithHost); + externalizable.sendBootStrap(proposedNodes); Thread.sleep(10); for (int j = 0; j < 3; j++) { - externalizable.sendBootStrap(proposedIdentifierWithHost); + externalizable.sendBootStrap(proposedNodes); Thread.sleep(10); if (useAnotherIdentifier.get()) { @@ -136,13 +136,13 @@ public class DiscoveryCluster { @Override public boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress) { final SocketAddress socketAddress = identifiers.putIfAbsent(remoteIdentifier, remoteAddress); - remoteNodes.activeIdentifierBitSet().set(remoteIdentifier); + knownNodes.activeIdentifierBitSet().set(remoteIdentifier); return socketAddress == null; } }; // add our identifier and host:port to the list of known identifiers - remoteNodes.add(ourAddressAndPort, identifier); + knownNodes.add(ourAddressAndPort, identifier); final TcpReplicationConfig tcpConfig = TcpReplicationConfig .of(tcpPort, toInetSocketArray(knownHostPorts)) @@ -201,7 +201,8 @@ public class DiscoveryCluster { } - private byte proposeRandomUnusedIdentifier(final DirectBitSet knownIdentifiers, boolean isFirstTime) throws UnknownHostException { + public static byte proposeRandomUnusedIdentifier(final DirectBitSet knownIdentifiers, + boolean isFirstTime) throws UnknownHostException { byte possible; diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java index aca15b397..327736be7 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java @@ -19,7 +19,7 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.atomic.AtomicBoolean; -import static net.openhft.chronicle.map.BytesExternalizableImpl.ProposedIdentifierWithHost; +import static java.util.concurrent.TimeUnit.SECONDS; import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES; import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG; @@ -227,13 +227,13 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { } -class RemoteNodes implements BytesMarshallable { +class KnownNodes implements BytesMarshallable { private Bytes activeIdentifiersBitSetBytes; private ConcurrentSkipListSet addressAndPorts = new ConcurrentSkipListSet(); private ATSDirectBitSet atsDirectBitSet; - RemoteNodes() { + KnownNodes() { this.activeIdentifiersBitSetBytes = new ByteBufferBytes(ByteBuffer.allocate(128 / 8)); this.addressAndPorts = new ConcurrentSkipListSet(); @@ -407,22 +407,22 @@ class AddressAndPort implements Comparable, BytesMarshallable { class BytesExternalizableImpl implements BytesMarshallable { - private final RemoteNodes remoteNode; + private final KnownNodes remoteNode; private final AtomicBoolean bootstrapRequired = new AtomicBoolean(); private final UDPEventListener udpEventListener; - private ProposedIdentifierWithHost ourProposedIdentifier; + private ProposedNodes ourProposedIdentifier; - final ConcurrentExpiryMap proposedIdentifiersWithHost = new - ConcurrentExpiryMap(AddressAndPort.class, - ProposedIdentifierWithHost.class); + final ConcurrentExpiryMap proposedIdentifiersWithHost = new + ConcurrentExpiryMap(AddressAndPort.class, + ProposedNodes.class); - public ProposedIdentifierWithHost getOurProposedIdentifier() { + public ProposedNodes getOurProposedIdentifier() { return ourProposedIdentifier; } - private void setOurProposedIdentifier(ProposedIdentifierWithHost ourProposedIdentifier) { + private void setOurProposedIdentifier(ProposedNodes ourProposedIdentifier) { this.ourProposedIdentifier = ourProposedIdentifier; } @@ -433,12 +433,12 @@ class BytesExternalizableImpl implements BytesMarshallable { Replica.ModificationNotifier modificationNotifier; - public BytesExternalizableImpl(final RemoteNodes remoteNode, UDPEventListener udpEventListener) { + public BytesExternalizableImpl(final KnownNodes remoteNode, UDPEventListener udpEventListener) { this.remoteNode = remoteNode; this.udpEventListener = udpEventListener; } - public RemoteNodes getRemoteNodes() { + public KnownNodes getRemoteNodes() { return remoteNode; } @@ -456,8 +456,8 @@ class BytesExternalizableImpl implements BytesMarshallable { remoteNode.writeMarshallable(out); - // we are no going to broadcast all the nodes that have been bootstaping in the last 200 - proposedIdentifiersWithHost.expireEntries(System.currentTimeMillis() - 200); + // we are no going to broadcast all the nodes that have been bootstaping in the last second + proposedIdentifiersWithHost.expireEntries(System.currentTimeMillis() - SECONDS.toMillis(1)); proposedIdentifiersWithHost.writeMarshallable(out); @@ -465,7 +465,7 @@ class BytesExternalizableImpl implements BytesMarshallable { } private boolean writeBootstrap(Bytes out) { - final ProposedIdentifierWithHost ourProposedIdentifier = getOurProposedIdentifier(); + final ProposedNodes ourProposedIdentifier = getOurProposedIdentifier(); if (ourProposedIdentifier == null || ourProposedIdentifier.addressAndPort == null) return false; @@ -480,7 +480,7 @@ class BytesExternalizableImpl implements BytesMarshallable { * @param in * @return returns true if the UDP message contains the text 'BOOTSTRAP' */ - private ProposedIdentifierWithHost readBootstrapMessage(Bytes in) { + private ProposedNodes readBootstrapMessage(Bytes in) { final long start = in.position(); @@ -501,7 +501,7 @@ class BytesExternalizableImpl implements BytesMarshallable { return null; } - final ProposedIdentifierWithHost result = new ProposedIdentifierWithHost(); + final ProposedNodes result = new ProposedNodes(); result.readMarshallable(in); return result; @@ -515,7 +515,7 @@ class BytesExternalizableImpl implements BytesMarshallable { @Override public void readMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes in) throws IllegalStateException { - final ProposedIdentifierWithHost bootstrap = readBootstrapMessage(in); + final ProposedNodes bootstrap = readBootstrapMessage(in); if (bootstrap != null) { LOG.debug("received Bootstrap"); @@ -543,8 +543,6 @@ class BytesExternalizableImpl implements BytesMarshallable { this.remoteNode.readMarshallable(in); - // we are no going to broadcast all the nodes that have been bootstaping in the last 200 - proposedIdentifiersWithHost.expireEntries(System.currentTimeMillis() - 200); proposedIdentifiersWithHost.readMarshallable(in); if (udpEventListener != null) @@ -559,13 +557,13 @@ class BytesExternalizableImpl implements BytesMarshallable { modificationNotifier.onChange(); } - public static class ProposedIdentifierWithHost implements BytesMarshallable { + public static class ProposedNodes implements BytesMarshallable { private byte identifier; private long timestamp; private AddressAndPort addressAndPort; - public ProposedIdentifierWithHost() { + public ProposedNodes() { } @@ -573,14 +571,14 @@ class BytesExternalizableImpl implements BytesMarshallable { return identifier; } - public ProposedIdentifierWithHost(@NotNull final AddressAndPort addressAndPort, - byte identifier) { + public ProposedNodes(@NotNull final AddressAndPort addressAndPort, + byte identifier) { this.addressAndPort = addressAndPort; this.identifier = identifier; this.timestamp = System.currentTimeMillis(); } - public ProposedIdentifierWithHost(byte[] address, short port, byte identifier) { + public ProposedNodes(byte[] address, short port, byte identifier) { this(new AddressAndPort(address, port), identifier); } @@ -619,7 +617,7 @@ class BytesExternalizableImpl implements BytesMarshallable { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; - ProposedIdentifierWithHost that = (ProposedIdentifierWithHost) o; + ProposedNodes that = (ProposedNodes) o; if (identifier != that.identifier) return false; if (!addressAndPort.equals(that.addressAndPort)) return false; @@ -640,10 +638,10 @@ class BytesExternalizableImpl implements BytesMarshallable { * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the host:port * and perhaps even proposed identifier of the node that sent it. * - * @param proposedIdentifierWithHost + * @param proposedNodes */ - public void sendBootStrap(ProposedIdentifierWithHost proposedIdentifierWithHost) { - setOurProposedIdentifier(proposedIdentifierWithHost); + public void sendBootStrap(ProposedNodes proposedNodes) { + setOurProposedIdentifier(proposedNodes); bootstrapRequired.set(true); } @@ -656,7 +654,7 @@ interface UDPEventListener { * called when we have received a UDP message, this is called after the message has been parsed */ - void onRemoteNodeEvent(RemoteNodes remoteNode, ConcurrentExpiryMap proposedIdentifiersWithHost); + void onRemoteNodeEvent(KnownNodes remoteNode, ConcurrentExpiryMap proposedIdentifiersWithHost); } class ConcurrentExpiryMap implements BytesMarshallable { @@ -680,7 +678,7 @@ class ConcurrentExpiryMap map1; - private ChronicleMap map2; - private volatile byte proposedIdentifier = 1; - private volatile boolean useAnotherIdentifier = false; - - - @Test - @Ignore - public void test2() throws IOException, InterruptedException { - - DiscoveryCluster discoveryCluster = new DiscoveryCluster(); - - ChronicleMap map = discoveryCluster.discoverMap(1025, 8087); - - Thread.sleep(10000); - } + public static final int PROPOSED_IDENTIFIER = 5; @Test @@ -52,42 +31,41 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { AddressAndPort server2AddressAndPort = new AddressAndPort(server1Address, (short) 1234); - RemoteNodes server1RemoteNodes = new RemoteNodes(); - BytesExternalizableImpl server1BytesExternalizable = new BytesExternalizableImpl(server1RemoteNodes, null); + KnownNodes server1KnownNodes = new KnownNodes(); + BytesExternalizableImpl server1BytesExternalizable = new BytesExternalizableImpl(server1KnownNodes, null); // for the unit test we are using a ByteBufferBytes, but iltimately this data would have been send // and received via UDP ByteBufferBytes udpData = new ByteBufferBytes(allocate(1024)); - // SERER1 + // SERER1 - is the new node join the cluster // we will first send the boostrap along with our host and port { - BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost = new BytesExternalizableImpl - .ProposedIdentifierWithHost(server1AddressAndPort, (byte) -1); + BytesExternalizableImpl.ProposedNodes proposedNodes = new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) -1); - server1BytesExternalizable.sendBootStrap(proposedIdentifierWithHost); + server1BytesExternalizable.sendBootStrap(proposedNodes); server1BytesExternalizable.writeMarshallable(udpData); } - RemoteNodes server2RemoteNodes = new RemoteNodes(); - server2RemoteNodes.add(server2AddressAndPort, (byte) SERVER2_IDENTIFER); - BytesExternalizableImpl server2BytesExternalizable = new BytesExternalizableImpl(server2RemoteNodes, null); + KnownNodes server2KnownNodes = new KnownNodes(); + server2KnownNodes.add(server2AddressAndPort, (byte) SERVER2_IDENTIFER); + BytesExternalizableImpl server2BytesExternalizable = new BytesExternalizableImpl(server2KnownNodes, null); // add our identifier and host:port to the list of known identifiers - // SERVER 2 + // SERVER 2 - is the node already in the cluster // read the bootstrap along with the host and port it came from { udpData.flip(); server2BytesExternalizable.readMarshallable(udpData); Assert.assertTrue(server2BytesExternalizable.proposedIdentifiersWithHost.values().contains - (new BytesExternalizableImpl.ProposedIdentifierWithHost(server1AddressAndPort, (byte) -1))); + (new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) -1))); } // SERVER 2 @@ -126,43 +104,76 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { } + // SERVER 1 + // send/broadcast - propose an identifier to use, by choosing one that is not used already + // for the purpose of this test we wont choose one at random ( as the code does ) but just pick + // identifier=5 { + udpData.clear(); + KnownNodes knownNodes = new KnownNodes(); - RemoteNodes remoteNodes = new RemoteNodes(); + BytesExternalizableImpl bytesExternalizable = new BytesExternalizableImpl(knownNodes, null); - BytesExternalizableImpl bytesExternalizable = new BytesExternalizableImpl(remoteNodes, null); + BytesExternalizableImpl.ProposedNodes proposedNodes = new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) PROPOSED_IDENTIFIER); - - BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost = new BytesExternalizableImpl - .ProposedIdentifierWithHost(server1AddressAndPort, (byte) -1); - - bytesExternalizable.sendBootStrap(proposedIdentifierWithHost); + // send the bootstrap along with this newly proposed identifier + bytesExternalizable.sendBootStrap(proposedNodes); bytesExternalizable.writeMarshallable(udpData); } - + // SERVER 2 - is the node already in the cluster + // read the bootstrap along with the proposed identifier and host:port it came from { - - RemoteNodes remoteNodes = new RemoteNodes(); - - - UDPEventListener udpEventListener = new UDPEventListener() { - - @Override - public void onRemoteNodeEvent(RemoteNodes remoteNode, ConcurrentExpiryMap proposedIdentifiersWithHost) { - int i = 1; - } - }; - BytesMarshallable bytesExternalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener); - - udpData.flip(); - bytesExternalizable.readMarshallable(udpData); + server2BytesExternalizable.readMarshallable(udpData); + + Assert.assertTrue(server2BytesExternalizable.proposedIdentifiersWithHost.values().contains + (new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) PROPOSED_IDENTIFIER))); } + + + // SERVER 2 + // write/broadcast the the response to receiving a bootstrap message + { + //ByteBufferBytes bitSetBytes = new ByteBufferBytes(allocate(128 / 8)); + // final DirectBitSet knownIdentifiers = new ATSDirectBitSet(bitSetBytes); + + // RemoteNodes remoteNodes = new RemoteNodes(bitSetBytes); + + + udpData.clear(); + + // broadcasts in response to the bootstrap all the host:ports and identifiers that it knows about + // in this case given that there is only this node in the grid and one node joining, + // it will only be the node that has the bootstrap and this node, who's data will be sent + server2BytesExternalizable.writeMarshallable(udpData); + + } + + + // SERVER 1 + // read the bootstrap along with all the proposed identifiers ( this will include our + // PROPOSED_IDENTIFIER and any that come from other server + { + udpData.flip(); + server1BytesExternalizable.readMarshallable(udpData); + + Assert.assertTrue(server1BytesExternalizable.getRemoteNodes().addressAndPorts().contains + (server2AddressAndPort)); + + Assert.assertTrue(server1BytesExternalizable.getRemoteNodes().activeIdentifierBitSet().get + (SERVER2_IDENTIFER)); + + Assert.assertTrue(server1BytesExternalizable.proposedIdentifiersWithHost.values().contains + (new BytesExternalizableImpl.ProposedNodes(server1AddressAndPort, (byte) PROPOSED_IDENTIFIER))); + + } + + } diff --git a/src/test/java/net/openhft/chronicle/map/example/WordCount.java b/src/test/java/net/openhft/chronicle/map/example/WordCount.java deleted file mode 100644 index af8a83afb..000000000 --- a/src/test/java/net/openhft/chronicle/map/example/WordCount.java +++ /dev/null @@ -1,35 +0,0 @@ -package net.openhft.chronicle.map.example; - -public class WordCount { - - static String[] words; - static int expectedSize; - static { - try { - // english version of war and peace -> ascii - String fullText = - new String(Files.readAllBytes(Paths.get("war_and_peace.txt")), UTF_8); - words = fullText.split("\\s+"); - expectedSize = (int) Arrays.stream(words).distinct().count(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - static interface ByteableIntValue extends IntValue, Byteable {} - - @Benchmark - public int chronicleMap() throws IOException { - ChronicleMap map = - ChronicleMapBuilder.of(String.class, ByteableIntValue.class) - .entrySize(16) - .actualSegments(1) - .entries(expectedSize) - .create(); - ByteableIntValue v = map.acquireUsing("a", null); - for (String word : words) { - map.acquireUsing(word, v).addValue(1); - } - return map.acquireUsing("a", v).getValue(); - } -} \ No newline at end of file