From a5e8452cb4ea19adb2bb74d10192523b49d2ca05 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 3 Oct 2014 20:26:32 +0100 Subject: [PATCH] HCOLL-134 improved NodeDiscovery, but its still not fully correctly working, I still have to do more work on it. --- .../openhft/chronicle/map/NodeDiscovery.java | 72 ++++++++++++------- .../chronicle/map/UdpChannelReplicator.java | 2 +- .../chronicle/map/NodeDiscoveryTest.java | 4 +- 3 files changed, 47 insertions(+), 31 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 3136552e2..fc73d4a4b 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -14,12 +14,10 @@ import java.net.*; import java.nio.ByteBuffer; import java.nio.channels.DatagramChannel; import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ConcurrentSkipListSet; +import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import static java.nio.ByteBuffer.allocateDirect; import static java.nio.ByteBuffer.wrap; @@ -42,16 +40,19 @@ public class NodeDiscovery { final UdpReplicationConfig udpConfig = UdpReplicationConfig .simple(Inet4Address.getByName("255.255.255.255"), udpBroadcastPort); + final KnownNodes knownNodes = new KnownNodes(); + final Set knownHostPorts = new ConcurrentSkipListSet(); final DirectBitSet knownIdentifiers = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate (128 / 8))); + final AtomicReference countDownLatch = new AtomicReference(new + CountDownLatch(1)); + final AddressAndPort ourAddressAndPort = new AddressAndPort(InetAddress.getLocalHost() .getAddress(), (short) tcpPort); - final Set knownHostPorts = new ConcurrentSkipListSet(); - final UDPEventListener udpEventListener = new UDPEventListener() { @Override @@ -71,68 +72,84 @@ public class NodeDiscovery { if (remoteIdentifier == proposedIdentifier.get()) useAnotherIdentifier.set(true); + + countDownLatch.get().countDown(); } } } }; - final DiscoveryNodeBytesMarshallable externalizable = new DiscoveryNodeBytesMarshallable(knownNodes, udpEventListener); + final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster + = new NodeDiscoveryBroadcaster(udpConfig, 1024, externalizable); + + externalizable.setModificationNotifier(nodeDiscoveryBroadcaster); + final DiscoveryNodeBytesMarshallable.ProposedNodes ourHostPort = new DiscoveryNodeBytesMarshallable.ProposedNodes(ourAddressAndPort, (byte) -1); // to start with we will send a bootstrap that just contains our hostname without and identifier - externalizable.sendBootStrap(ourHostPort); - Thread.sleep(10); + for (int i = 0; i < 20; i++) { + externalizable.sendBootStrap(ourHostPort); - // we should not get back some identifiers + // once the count down latch is trigger we know we go something back from one of the nodes + if (countDownLatch.get().await(50, TimeUnit.MILLISECONDS)) + break; + } + + // we should now get back some identifiers // the identifiers will come back to the callback on the nio thread, the update arrives at the // onRemoteNodeEvent - externalizable.sendBootStrap(ourHostPort); - Thread.sleep(10); + byte identifier; - + // now we are going to propose an identifier boolean isFistTime = true; + OUTER: for (; ; ) { - + useAnotherIdentifier.set(false); identifier = proposeRandomUnusedIdentifier(knownIdentifiers, isFistTime); proposedIdentifier.set(identifier); + LOG.info("proposing to use identifier=" + identifier); isFistTime = false; final DiscoveryNodeBytesMarshallable.ProposedNodes proposedNodes = new DiscoveryNodeBytesMarshallable.ProposedNodes(ourAddressAndPort, identifier); - externalizable.sendBootStrap(proposedNodes); + Thread.sleep(500); - Thread.sleep(10); - - for (int j = 0; j < 3; j++) { + countDownLatch.set(new CountDownLatch(1)); + for (int i = 0; i < 20; i++) { externalizable.sendBootStrap(proposedNodes); - Thread.sleep(10); - if (useAnotherIdentifier.get()) { - // given that another node host proposed the same identifier, we will choose a different one. - continue; + // once the count down latch is trigger we know we go something back from one of the nodes + if (countDownLatch.get().await(50, TimeUnit.MILLISECONDS)) { + if (useAnotherIdentifier.get()) { + // given that another node host proposed the same identifier, we will choose a different one. + LOG.info("Another node is using identifier=" + identifier+", " + + "going to have to select another one."); + continue OUTER; + } else { + break OUTER; + } } + + } + LOG.info("looks like we are the only node in the grid, so going to use identifier="+identifier); + break; } - final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster - = new NodeDiscoveryBroadcaster(udpConfig, 1024, externalizable); - - externalizable.setModificationNotifier(nodeDiscoveryBroadcaster); - // we should make a local copy as this may change final IdentifierListener identifierListener = new IdentifierListener() { @@ -839,6 +856,7 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { public void sendBootStrap(ProposedNodes proposedNodes) { setOurProposedIdentifier(proposedNodes); bootstrapRequired.set(true); + onChange(); } diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java index 2ab0ba098..26c20d123 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -197,7 +197,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. public void onChange() { // the write have to be enabled on the same thread as the selector shouldEnableOpWrite = true; - selector.wakeup(); + } private void enableWrites() { diff --git a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java index 1349dd667..60a9b52fd 100644 --- a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java +++ b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java @@ -7,8 +7,6 @@ public class NodeDiscoveryTest { @Test public void testDiscoverMap() throws Exception { final NodeDiscovery nodeDiscovery = new NodeDiscovery(); - final ChronicleMap map = nodeDiscovery.discoverMap(1235, 1236); - - // Thread.sleep(10000); + final ChronicleMap map = nodeDiscovery.discoverMap(8123, 1237); } } \ No newline at end of file