From ec09015f1dc9e2e83ac40ad8e4d60ab3bd441854 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 2 Oct 2014 19:01:39 +0100 Subject: [PATCH] HCOLL-134 ( release draft cut of the code so that I can test it across my network ) --- .../chronicle/map/ConcurrentExpiryMap.java | 14 + .../chronicle/map/DiscoveryCluster.java | 33 +- .../map/NodeDiscoveryBroadcaster.java | 623 ++++++++++-------- .../map/NodeDiscoveryBroadcasterTest.java | 139 +++- 4 files changed, 505 insertions(+), 304 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/ConcurrentExpiryMap.java diff --git a/src/main/java/net/openhft/chronicle/map/ConcurrentExpiryMap.java b/src/main/java/net/openhft/chronicle/map/ConcurrentExpiryMap.java new file mode 100644 index 000000000..426bc94ba --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ConcurrentExpiryMap.java @@ -0,0 +1,14 @@ +package net.openhft.chronicle.map; + +import java.util.Map; +import java.util.Queue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ConcurrentMap; + +/** + * wraps a concurrent map and queue so that we can easily expire old entries + * + * @author Rob Austin. + */ + diff --git a/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java b/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java index 460f33585..c7a984793 100644 --- a/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java +++ b/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java @@ -3,7 +3,6 @@ package net.openhft.chronicle.map; import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.io.ByteBufferBytes; -import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; import java.io.IOException; @@ -36,9 +35,7 @@ public class DiscoveryCluster { final UdpReplicationConfig udpConfig = UdpReplicationConfig .simple(Inet4Address.getByName("255.255.255.255"), udpBroadcastPort); - final Bytes identifierBitSetBits = new ByteBufferBytes(ByteBuffer.allocate(128 / 8)); - - final RemoteNodes remoteNodes = new RemoteNodes(identifierBitSetBits); + final RemoteNodes remoteNodes = new RemoteNodes(); final DirectBitSet knownIdentifiers = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate (128 / 8))); @@ -51,15 +48,14 @@ public class DiscoveryCluster { final UDPEventListener udpEventListener = new UDPEventListener() { @Override - public void onRemoteNodeEvent(@NotNull final Set usedHostPorts, - @NotNull final ATSDirectBitSet usedIdentifiers, - @NotNull final Set proposedHostPortIdentifiers) { + public void onRemoteNodeEvent(RemoteNodes remoteNode, ConcurrentExpiryMap proposedIdentifiersWithHost) { - knownHostPorts.addAll(usedHostPorts); + knownHostPorts.addAll(remoteNode.addressAndPorts()); - orBitSets(usedIdentifiers, knownIdentifiers); + orBitSets(remoteNode.activeIdentifierBitSet(), knownIdentifiers); - for (BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost : proposedHostPortIdentifiers) { + for (BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost : + proposedIdentifiersWithHost.values()) { if (!proposedIdentifierWithHost.addressAndPort().equals(ourAddressAndPort)) { int remoteIdentifier = proposedIdentifierWithHost.identifier(); @@ -71,6 +67,8 @@ public class DiscoveryCluster { } } } + + }; final BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener); @@ -79,15 +77,15 @@ public class DiscoveryCluster { BytesExternalizableImpl.ProposedIdentifierWithHost(ourAddressAndPort, (byte) -1); // to start with we will send a bootstrap that just contains our hostname without and identifier - externalizable.setOurProposedIdentifier(ourHostPort); - externalizable.sendBootStrap(); + + externalizable.sendBootStrap(ourHostPort); Thread.sleep(10); // we should not get back some identifiers // the identifiers will come back to the callback on the nio thread, the update arrives at the // onRemoteNodeEvent - externalizable.sendBootStrap(); + externalizable.sendBootStrap(ourHostPort); Thread.sleep(10); byte identifier; @@ -104,14 +102,13 @@ public class DiscoveryCluster { final BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost = new BytesExternalizableImpl.ProposedIdentifierWithHost(ourAddressAndPort, identifier); - externalizable.setOurProposedIdentifier(proposedIdentifierWithHost); - externalizable.sendBootStrap(); + externalizable.sendBootStrap(proposedIdentifierWithHost); Thread.sleep(10); for (int j = 0; j < 3; j++) { - externalizable.sendBootStrap(); + externalizable.sendBootStrap(proposedIdentifierWithHost); Thread.sleep(10); if (useAnotherIdentifier.get()) { @@ -144,8 +141,8 @@ public class DiscoveryCluster { } }; - - remoteNodes.activeIdentifierBitSet().set(identifier); + // add our identifier and host:port to the list of known identifiers + remoteNodes.add(ourAddressAndPort, identifier); final TcpReplicationConfig tcpConfig = TcpReplicationConfig .of(tcpPort, toInetSocketArray(knownHostPorts)) diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java index 39aca9c75..aca15b397 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java @@ -4,6 +4,7 @@ import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.BytesMarshallable; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -11,10 +12,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.DatagramChannel; -import java.util.Arrays; -import java.util.HashSet; -import java.util.Queue; -import java.util.Set; +import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; @@ -47,6 +45,33 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { BOOTSTRAP_BYTES.append(BOOTSTRAP); } + + /** + * we shoudl have to do this, but I think there is a bug in the bitset, so added it as a work around + * + * @param source + * @return + */ + static Bytes toNewBuffer(Bytes source) { + + ByteBufferBytes result = new ByteBufferBytes(ByteBuffer.allocate((int) source.remaining())); + + result.write(source); + result.clear(); + return result; + + } + + static String toString(DirectBitSet bitSet) { + final StringBuilder builder = new StringBuilder(); + for (int i = 0; i < bitSet.size(); i++) { + builder.append(bitSet.get(i) ? '1' : '0'); + } + return builder.toString(); + + // LOG.debug(type + "=" + "bitset =" + builder); + } + /** * @param replicationConfig * @param externalizable @@ -55,7 +80,7 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { NodeDiscoveryBroadcaster( @NotNull final UdpReplicationConfig replicationConfig, final int serializedEntrySize, - final BytesExternalizable externalizable) + final BytesMarshallable externalizable) throws IOException { super(replicationConfig, serializedEntrySize, UNUSED); @@ -73,14 +98,14 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { private final ByteBuffer in; private final ByteBufferBytes out; - private final BytesExternalizable externalizable; + private final BytesMarshallable externalizable; /** * @param serializedEntrySize the maximum size of an entry include the meta data * @param externalizable supports reading and writing serialize entries */ public UdpSocketChannelEntryReader(final int serializedEntrySize, - @NotNull final BytesExternalizable externalizable) { + @NotNull final BytesMarshallable externalizable) { // we make the buffer twice as large just to give ourselves headroom in = ByteBuffer.allocateDirect(serializedEntrySize * 2); @@ -124,7 +149,7 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { if (out.remaining() != size) return; - externalizable.readExternalBytes(out); + externalizable.readMarshallable(out); } @@ -136,11 +161,11 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { private final ByteBufferBytes in; @NotNull - private final BytesExternalizable externalizable; + private final BytesMarshallable externalizable; private UdpChannelReplicator udpReplicator; public UdpSocketChannelEntryWriter(final int serializedEntrySize, - @NotNull final BytesExternalizable externalizable, + @NotNull final BytesMarshallable externalizable, @NotNull final UdpChannelReplicator udpReplicator) { this.externalizable = externalizable; @@ -183,7 +208,7 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { long start = in.position(); // writes the contents of - externalizable.writeExternalBytes(in); + externalizable.writeMarshallable(in); long size = in.position() - start; @@ -202,47 +227,26 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { } -/** - * supports reading and writing serialize entries - */ -interface BytesExternalizable { +class RemoteNodes implements BytesMarshallable { + private Bytes activeIdentifiersBitSetBytes; + private ConcurrentSkipListSet addressAndPorts = new ConcurrentSkipListSet(); + private ATSDirectBitSet atsDirectBitSet; - void writeExternalBytes(@NotNull Bytes destination); - - - public void readExternalBytes(@NotNull Bytes source); - -} - -class RemoteNodes { - - - private final Bytes activeIdentifiersBitSetBytes; - private final ConcurrentSkipListSet addressAndPorts; - private final ATSDirectBitSet atsDirectBitSet; - - - /** - * @param activeIdentifiersBitSetBytes byte sof a bitset containing the known identifiers - */ - RemoteNodes(final Bytes activeIdentifiersBitSetBytes) { + RemoteNodes() { + this.activeIdentifiersBitSetBytes = new ByteBufferBytes(ByteBuffer.allocate(128 / 8)); this.addressAndPorts = new ConcurrentSkipListSet(); - - this.activeIdentifiersBitSetBytes = activeIdentifiersBitSetBytes; this.atsDirectBitSet = new ATSDirectBitSet(this.activeIdentifiersBitSetBytes); } + public Set addressAndPorts() { return addressAndPorts; } - public Bytes activeIdentifierBytes() { - return activeIdentifiersBitSetBytes; - } - - public void add(AddressAndPort inetSocketAddress) { + public void add(AddressAndPort inetSocketAddress, byte identifier) { + activeIdentifierBitSet().set(identifier); addressAndPorts.add(inetSocketAddress); } @@ -251,21 +255,66 @@ class RemoteNodes { } + @Override + public void readMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes in) throws IllegalStateException { + short size = in.readShort(); + + for (int i = 0; i < size; i++) { + final AddressAndPort addressAndPort = new AddressAndPort(); + addressAndPort.readMarshallable(in); + addressAndPorts.add(addressAndPort); + } + + ByteBufferBytes activeIdentifiersBitSetBytes = new ByteBufferBytes(ByteBuffer.allocate(128 / 8)); + activeIdentifiersBitSetBytes.readMarshallable(in); + + ATSDirectBitSet bitset = new ATSDirectBitSet(activeIdentifiersBitSetBytes); + for (long next = bitset.nextSetBit(0); next > 0; next = bitset.nextSetBit(next + 1)) { + atsDirectBitSet.set(next); + } + } + + @Override + public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { + + // make a safe copy + final Set addressAndPorts = new HashSet(this.addressAndPorts); + + // write the size + out.writeShort(addressAndPorts.size()); + + for (AddressAndPort bytesMarshallable : addressAndPorts) { + bytesMarshallable.writeMarshallable(out); + } + + activeIdentifiersBitSetBytes.clear(); + activeIdentifiersBitSetBytes.writeMarshallable(out); + + } + + @Override + public String toString() { + return "RemoteNodes{" + + " addressAndPorts=" + addressAndPorts + + ", bitSet=" + NodeDiscoveryBroadcaster.toString(atsDirectBitSet) + + '}'; + } } -class AddressAndPort implements Comparable { +class AddressAndPort implements Comparable, BytesMarshallable { private byte[] address; private short port; public AddressAndPort(byte[] address, short port) { - if (address.length != 4) - throw new IllegalStateException("address.length should equal '4' but is length=" + - address.length); this.address = address; this.port = port; } + public AddressAndPort() { + + } + public byte[] getAddress() { return address; } @@ -275,7 +324,6 @@ class AddressAndPort implements Comparable { } public byte[] address() { - return address; } @@ -326,27 +374,55 @@ class AddressAndPort implements Comparable { } static String numericToTextFormat(byte[] src) { - return (src[0] & 0xff) + "." + (src[1] & 0xff) + "." + (src[2] & 0xff) + "." + (src[3] & 0xff); + if (src.length == 4) { + return (src[0] & 0xff) + "." + (src[1] & 0xff) + "." + (src[2] & 0xff) + "." + (src[3] & 0xff); + } + throw new UnsupportedOperationException(); + } + @Override + public void readMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes in) throws IllegalStateException { + short len = in.readShort(); + address = new byte[len]; + + for (int i = 0; i < len; i++) { + address[i] = in.readByte(); + } + port = in.readShort(); + + } + + @Override + public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { + out.writeShort(getAddress().length); + for (byte address : getAddress()) { + out.write(address); + } + out.writeShort(port); + } } -class BytesExternalizableImpl implements BytesExternalizable { +class BytesExternalizableImpl implements BytesMarshallable { - private final RemoteNodes allNodes; + private final RemoteNodes remoteNode; private final AtomicBoolean bootstrapRequired = new AtomicBoolean(); private final UDPEventListener udpEventListener; private ProposedIdentifierWithHost ourProposedIdentifier; + final ConcurrentExpiryMap proposedIdentifiersWithHost = new + ConcurrentExpiryMap(AddressAndPort.class, + ProposedIdentifierWithHost.class); + public ProposedIdentifierWithHost getOurProposedIdentifier() { return ourProposedIdentifier; } - public void setOurProposedIdentifier(ProposedIdentifierWithHost ourProposedIdentifier) { + private void setOurProposedIdentifier(ProposedIdentifierWithHost ourProposedIdentifier) { this.ourProposedIdentifier = ourProposedIdentifier; } @@ -357,126 +433,93 @@ class BytesExternalizableImpl implements BytesExternalizable { Replica.ModificationNotifier modificationNotifier; - public BytesExternalizableImpl(final RemoteNodes allNodes, UDPEventListener udpEventListener) { - this.allNodes = allNodes; + public BytesExternalizableImpl(final RemoteNodes remoteNode, UDPEventListener udpEventListener) { + this.remoteNode = remoteNode; this.udpEventListener = udpEventListener; } + public RemoteNodes getRemoteNodes() { + return remoteNode; + } + /** * this is used to tell nodes that are connecting to us which host and ports are in our grid, along with * all the identifiers. - * - * @param destination a buffer the entry will be written to, the segment may reject this operation and add - * zeroBytes, if the identifier in the entry did not match the maps local */ @Override - public void writeExternalBytes(@NotNull Bytes destination) { - + public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { if (bootstrapRequired.getAndSet(false)) { - - final ProposedIdentifierWithHost ourProposedIdentifier = getOurProposedIdentifier(); - if (ourProposedIdentifier == null || ourProposedIdentifier.addressAndPort == null) - return; - - BOOTSTRAP_BYTES.clear(); - destination.write(BOOTSTRAP_BYTES); - - - for (byte b : ourProposedIdentifier.addressAndPort.address()) { - destination.writeByte(b); - } - - destination.writeShort(ourProposedIdentifier.addressAndPort.port()); - destination.writeByte(ourProposedIdentifier.identifier()); + writeBootstrap(out); return; } - final Set addressAndPorts = allNodes.addressAndPorts(); - - // write the number of hosts and ports - short count = (short) addressAndPorts.size(); - destination.writeUnsignedShort(count); - - // write all the host and ports - for (AddressAndPort inetSocketAddress : addressAndPorts) { - for (byte b : inetSocketAddress.address()) { - destination.writeByte(b); - } - - destination.writeShort(inetSocketAddress.port()); - } - - // destination.write(allNodes.ourProposedID()); - - Bytes bytes = allNodes.activeIdentifierBytes(); - //bytes.position(0); - - if (LOG.isDebugEnabled()) - toString(allNodes.activeIdentifierBitSet(), "writing"); - - // we will store the size of the bitset - destination.writeUnsignedShort((int) bytes.remaining()); + remoteNode.writeMarshallable(out); // we are no going to broadcast all the nodes that have been bootstaping in the last 200 - expireOldBootstrapStats(System.currentTimeMillis() - 200); + proposedIdentifiersWithHost.expireEntries(System.currentTimeMillis() - 200); - // this is where we will store the count of the bootstrap data which is written below - destination.skip(2); - long countLocation = destination.position(); - int countBootstrapDataRepublished = 0; + proposedIdentifiersWithHost.writeMarshallable(out); - // rebroadcast all the host:port:identifier of the nodes that have sent a bootstrap in the last 200 - // milliseconds - for (ProposedIdentifierWithHost proposedIdentifierWithHost : bootstrapStatsMap.values()) { - - for (byte b : proposedIdentifierWithHost.addressAndPort.address()) { - destination.writeByte(b); - } - - destination.writeShort(proposedIdentifierWithHost.addressAndPort.port()); - destination.writeByte(proposedIdentifierWithHost.identifier()); - countBootstrapDataRepublished++; - } - - destination.writeInt(countLocation, countBootstrapDataRepublished); } - final ConcurrentMap bootstrapStatsMap = new - ConcurrentHashMap(); + private boolean writeBootstrap(Bytes out) { + final ProposedIdentifierWithHost ourProposedIdentifier = getOurProposedIdentifier(); + if (ourProposedIdentifier == null || ourProposedIdentifier.addressAndPort == null) + return false; - final Queue proposedIdentifierWithHostExpiryQueue = new ConcurrentLinkedQueue(); + BOOTSTRAP_BYTES.clear(); + out.write(BOOTSTRAP_BYTES); + ourProposedIdentifier.writeMarshallable(out); + return true; + } - private void expireOldBootstrapStats(long timeOlderThan) { - for (; ; ) { + /** + * @param in + * @return returns true if the UDP message contains the text 'BOOTSTRAP' + */ + private ProposedIdentifierWithHost readBootstrapMessage(Bytes in) { - final ProposedIdentifierWithHost proposedIdentifierWithHost = proposedIdentifierWithHostExpiryQueue.peek(); + final long start = in.position(); - if (proposedIdentifierWithHost == null) - break; + try { - if (proposedIdentifierWithHost.timestamp < timeOlderThan) { - // only remote it if it has not changed - bootstrapStatsMap.remove(proposedIdentifierWithHost.addressAndPort, proposedIdentifierWithHost); + long size = BOOTSTRAP_BYTES.limit(); + + if (size > in.remaining()) + return null; + + // reads the text bootstrap + for (int i = 0; i < size; i++) { + + final byte byteRead = in.readByte(); + final byte expectedByte = BOOTSTRAP_BYTES.readByte(i); + + if (!(expectedByte == byteRead)) + return null; } - proposedIdentifierWithHostExpiryQueue.poll(); + final ProposedIdentifierWithHost result = new ProposedIdentifierWithHost(); + result.readMarshallable(in); + + return result; + + } finally { + in.position(start); } } @Override - public void readExternalBytes(@NotNull Bytes source) { + public void readMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes in) throws IllegalStateException { - final ProposedIdentifierWithHost bootstrap = readBootstrapMessage(source); + final ProposedIdentifierWithHost bootstrap = readBootstrapMessage(in); if (bootstrap != null) { LOG.debug("received Bootstrap"); - bootstrapStatsMap.put(bootstrap.addressAndPort, bootstrap); - proposedIdentifierWithHostExpiryQueue.add(bootstrap); - + proposedIdentifiersWithHost.put(bootstrap.addressAndPort, bootstrap); try { @@ -485,6 +528,8 @@ class BytesExternalizableImpl implements BytesExternalizable { Thread.sleep((int) (Math.random() * 9.0)); + // this is used to turn on the OP_WRITE, so that we can broadcast back the known host and + // ports in the grid onChange(); } catch (InterruptedException e) { @@ -494,102 +539,39 @@ class BytesExternalizableImpl implements BytesExternalizable { return; } - int count = source.readUnsignedShort(); + + this.remoteNode.readMarshallable(in); - final Set allNodeInTheGrid = new HashSet(); + // 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); - for (int i = 0; i < count; i++) { - byte[] address = new byte[4]; - address[0] = source.readByte(); - address[1] = source.readByte(); - address[2] = source.readByte(); - address[3] = source.readByte(); + if (udpEventListener != null) + udpEventListener.onRemoteNodeEvent(remoteNode, proposedIdentifiersWithHost); - final short port = source.readShort(); - - LOG.debug("received InetSocketAddress(" + address + "," + port + ")"); - allNodeInTheGrid.add(new AddressAndPort(address, port)); - } - - - int countRebroadcast = source.readShort(); - - final Set proposedIdentifierWithHosts = new - HashSet(); - - // rebroadcast all the host:port:identifier of the nodes that have sent a bootstrap in the last 200 - // milliseconds - for (int i = 0; i < countRebroadcast; i++) { - byte[] address = new byte[4]; - address[0] = source.readByte(); - address[1] = source.readByte(); - address[2] = source.readByte(); - address[3] = source.readByte(); - - final short port = source.readShort(); - final byte identifier = source.readByte(); - - final ProposedIdentifierWithHost proposed = new ProposedIdentifierWithHost( - address, port, identifier); - - proposedIdentifierWithHosts.add(proposed); - } - - // the number of bytes in the bitset - int sizeInBytes = source.readUnsignedShort(); - - if (sizeInBytes == 0) { - LOG.debug("received nothing.."); - return; - } - - - //todo we should not have to do this ! - maybe a bug in BitSet of ByteBuffer - Bytes usedIdentifiers = toNewBuffer(source); - - udpEventListener.onRemoteNodeEvent(allNodeInTheGrid, - new ATSDirectBitSet(usedIdentifiers), - proposedIdentifierWithHosts); } - /** - * we shoudl have to do this, but I think there is a bug in the bitset, so added it as a work around - * - * @param source - * @return - */ - private Bytes toNewBuffer(Bytes source) { - - ByteBufferBytes result = new ByteBufferBytes(ByteBuffer.allocate((int) source.remaining())); - - result.write(source); - result.clear(); - return result; - - } - - private void toString(DirectBitSet bitSet, String type) { - final StringBuilder builder = new StringBuilder(); - for (int i = 0; i < bitSet.size(); i++) { - builder.append(bitSet.get(i) ? '1' : '0'); - } - - LOG.debug(type + "=" + "bitset =" + builder); - } - public void onChange() { - modificationNotifier.onChange(); + if (modificationNotifier != null) + modificationNotifier.onChange(); } + public static class ProposedIdentifierWithHost implements BytesMarshallable { - public static class ProposedIdentifierWithHost { + private byte identifier; + private long timestamp; + private AddressAndPort addressAndPort; - private final int identifier; - private final long timestamp; - private final AddressAndPort addressAndPort; + public ProposedIdentifierWithHost() { + + } + + public byte identifier() { + return identifier; + } public ProposedIdentifierWithHost(@NotNull final AddressAndPort addressAndPort, byte identifier) { @@ -603,6 +585,35 @@ class BytesExternalizableImpl implements BytesExternalizable { } + public AddressAndPort addressAndPort() { + return addressAndPort; + } + + @Override + public void readMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes in) throws IllegalStateException { + addressAndPort = new AddressAndPort(); + addressAndPort.readMarshallable(in); + timestamp = in.readLong(); + identifier = in.readByte(); + + } + + @Override + public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { + addressAndPort.writeMarshallable(out); + out.writeLong(timestamp); + out.writeByte(identifier); + } + + @Override + public String toString() { + return "ProposedIdentifierWithHost{" + + "identifier=" + identifier + + ", timestamp=" + timestamp + + ", addressAndPort=" + addressAndPort + + '}'; + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -611,84 +622,31 @@ class BytesExternalizableImpl implements BytesExternalizable { ProposedIdentifierWithHost that = (ProposedIdentifierWithHost) o; if (identifier != that.identifier) return false; - if (timestamp != that.timestamp) return false; - if (addressAndPort != null ? !addressAndPort.equals(that.addressAndPort) : that.addressAndPort != null) - return false; + if (!addressAndPort.equals(that.addressAndPort)) return false; return true; } @Override public int hashCode() { - int result = identifier; - result = 31 * result + (int) (timestamp ^ (timestamp >>> 32)); - result = 31 * result + (addressAndPort != null ? addressAndPort.hashCode() : 0); + int result = (int) identifier; + result = 31 * result + addressAndPort.hashCode(); return result; } - - public int identifier() { - return identifier; - - } - - public AddressAndPort addressAndPort() { - return addressAndPort; - } } + /** - * @param source - * @return returns true if the UDP message contains the text 'BOOTSTRAP' + * 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 */ - private ProposedIdentifierWithHost readBootstrapMessage(Bytes source) { - - final long start = source.position(); - - try { - - long size = BOOTSTRAP_BYTES.limit(); - - if (size > source.remaining()) - return null; - - for (int i = 0; i < size; i++) { - - final byte byteRead = source.readByte(start + i); - final byte expectedByte = BOOTSTRAP_BYTES.readByte(i); - - if (!(expectedByte == byteRead)) - return null; - } - - byte[] address = new byte[4]; - address[0] = source.readByte(); - address[1] = source.readByte(); - address[2] = source.readByte(); - address[3] = source.readByte(); - - final short port = source.readShort(); - final byte identifier = source.readByte(); - - return new ProposedIdentifierWithHost(address, port, identifier); - - } finally { - source.position(start); - } - } - - - public void sendBootStrap() { + public void sendBootStrap(ProposedIdentifierWithHost proposedIdentifierWithHost) { + setOurProposedIdentifier(proposedIdentifierWithHost); bootstrapRequired.set(true); } - public void add(AddressAndPort addressAndPort) { - allNodes.add(addressAndPort); - } - - - public void add(byte identifier) { - allNodes.activeIdentifierBitSet().set(identifier); - } } @@ -698,7 +656,108 @@ interface UDPEventListener { * called when we have received a UDP message, this is called after the message has been parsed */ - void onRemoteNodeEvent(Set usedHostPorts, - ATSDirectBitSet usedIdentifers, - Set proposedHostPortIdentifiers); + void onRemoteNodeEvent(RemoteNodes remoteNode, ConcurrentExpiryMap proposedIdentifiersWithHost); } + +class ConcurrentExpiryMap implements BytesMarshallable { + + final ConcurrentMap map = new + ConcurrentHashMap(); + private final Class kClass; + private final Class vClass; + + ConcurrentExpiryMap(Class kClass, Class vClass) { + + this.kClass = kClass; + this.vClass = vClass; + } + + @Override + public void readMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes in) throws IllegalStateException { + + short size = in.readShort(); + try { + for (int i = 0; i < size; i++) { + + final K k = kClass.newInstance(); + k.writeMarshallable(in); + + final V v = vClass.newInstance(); + v.readMarshallable(in); + + map.put(k, v); + } + } catch (Exception e) { + LOG.error("", e); + } + + } + + @Override + public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { + final Map safeCopy = new HashMap(map); + out.writeShort(safeCopy.size()); + for (Map.Entry entry : safeCopy.entrySet()) { + entry.getKey().writeMarshallable(out); + entry.getValue().writeMarshallable(out); + } + } + + class W { + final long timestamp; + final V v; + + public W(V v) { + this.v = v; + this.timestamp = System.currentTimeMillis(); + } + } + + public void put(final K k, final V v) { + map.put(k, v); + final W w = new W(v); + queue.add(new Map.Entry>() { + + @Override + public K getKey() { + return k; + } + + @Override + public W getValue() { + return w; + } + + @Override + public W setValue(W value) { + throw new UnsupportedOperationException(); + } + }); + } + + + final Queue>> queue = new ConcurrentLinkedQueue>>(); + + public java.util.Collection values() { + return map.values(); + } + + + public void expireEntries(long timeOlderThan) { + for (; ; ) { + + final Map.Entry> e = this.queue.peek(); + + if (e == null) + break; + + if (e.getValue().timestamp < timeOlderThan) { + // only remote it if it has not changed + map.remove(e.getKey(), e.getValue().v); + } + + this.queue.poll(); + } + } + +} \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java index 59a2ac237..106400c08 100644 --- a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java +++ b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java @@ -1,13 +1,21 @@ package net.openhft.chronicle.map; import junit.framework.TestCase; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.serialization.BytesMarshallable; import org.junit.Ignore; import org.junit.Test; +import org.testng.Assert; import java.io.IOException; +import java.net.Inet4Address; + +import static java.nio.ByteBuffer.allocate; public class NodeDiscoveryBroadcasterTest extends TestCase { + public static final int SERVER2_IDENTIFER = 10; + public void test() { // test to stop it erroring with no test } @@ -15,10 +23,7 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { private ChronicleMap map1; private ChronicleMap map2; private volatile byte proposedIdentifier = 1; - private volatile boolean useAnotherIdentifer = false; - - - + private volatile boolean useAnotherIdentifier = false; @Test @@ -33,6 +38,132 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { } + @Test + @Ignore + public void test3() throws IOException, InterruptedException { + + // write broadcast our address on the bootstrap method + byte[] server1Address = Inet4Address.getLocalHost().getAddress(); + AddressAndPort server1AddressAndPort = new AddressAndPort(server1Address, (short) 1234); + + + // write broadcast our address on the bootstrap method + byte[] server2Address = Inet4Address.getByName("129.168.0.2").getAddress(); + AddressAndPort server2AddressAndPort = new AddressAndPort(server1Address, (short) 1234); + + + RemoteNodes server1RemoteNodes = new RemoteNodes(); + BytesExternalizableImpl server1BytesExternalizable = new BytesExternalizableImpl(server1RemoteNodes, 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 + // we will first send the boostrap along with our host and port + { + + + BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost = new BytesExternalizableImpl + .ProposedIdentifierWithHost(server1AddressAndPort, (byte) -1); + + server1BytesExternalizable.sendBootStrap(proposedIdentifierWithHost); + + server1BytesExternalizable.writeMarshallable(udpData); + + } + + RemoteNodes server2RemoteNodes = new RemoteNodes(); + server2RemoteNodes.add(server2AddressAndPort, (byte) SERVER2_IDENTIFER); + BytesExternalizableImpl server2BytesExternalizable = new BytesExternalizableImpl(server2RemoteNodes, null); + + // add our identifier and host:port to the list of known identifiers + + + // SERVER 2 + // 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))); + } + + // 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); + + } + + + // SERER1 + // receive + + // read the bootstrap along with the host and port it came from + { + udpData.flip(); + server1BytesExternalizable.readMarshallable(udpData); + + Assert.assertTrue(server1BytesExternalizable.getRemoteNodes().addressAndPorts().contains + (server2AddressAndPort)); + + Assert.assertTrue(server1BytesExternalizable.getRemoteNodes().activeIdentifierBitSet().get + (SERVER2_IDENTIFER)); + + } + + + { + + + RemoteNodes remoteNodes = new RemoteNodes(); + + BytesExternalizableImpl bytesExternalizable = new BytesExternalizableImpl(remoteNodes, null); + + + BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost = new BytesExternalizableImpl + .ProposedIdentifierWithHost(server1AddressAndPort, (byte) -1); + + bytesExternalizable.sendBootStrap(proposedIdentifierWithHost); + + bytesExternalizable.writeMarshallable(udpData); + + } + + + { + + 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); + } + } }