mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
HCOLL-134 ( release draft cut of the code so that I can test it across my network )
This commit is contained in:
Rob Austin
@@ -4,58 +4,241 @@ 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;
|
||||
import java.net.Inet4Address;
|
||||
import java.net.*;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.ConcurrentSkipListSet;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static java.nio.ByteBuffer.wrap;
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
|
||||
import static net.openhft.chronicle.map.Replicators.tcp;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public class DiscoveryCluster implements UDPEventListener {
|
||||
public class DiscoveryCluster {
|
||||
|
||||
public static final int PORT = 1235;
|
||||
|
||||
DirectBitSet bitSet = newBitSet(128);
|
||||
public ChronicleMap<Integer, CharSequence> discoverMap(int udpBroadcastPort, final int tcpPort) throws IOException, InterruptedException {
|
||||
|
||||
public void discover() throws IOException, InterruptedException {
|
||||
final AtomicInteger proposedIdentifier = new AtomicInteger();
|
||||
final AtomicBoolean useAnotherIdentifier = new AtomicBoolean();
|
||||
|
||||
final UdpReplicationConfig udpConfig = UdpReplicationConfig
|
||||
.simple(Inet4Address.getByName("255.255.255.255"), PORT);
|
||||
.simple(Inet4Address.getByName("255.255.255.255"), udpBroadcastPort);
|
||||
|
||||
final Bytes identifierBitSetBits = new ByteBufferBytes(ByteBuffer.allocate(128 / 8));
|
||||
|
||||
final RemoteNodes remoteNodes = new RemoteNodes(identifierBitSetBits);
|
||||
final DirectBitSet knownIdentifiers = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate
|
||||
(128 / 8)));
|
||||
|
||||
final AddressAndPort ourAddressAndPort = new AddressAndPort(InetAddress.getLocalHost()
|
||||
.getAddress(),
|
||||
(short) tcpPort);
|
||||
|
||||
final Set<AddressAndPort> knownHostPorts = new ConcurrentSkipListSet<AddressAndPort>();
|
||||
|
||||
final UDPEventListener udpEventListener = new UDPEventListener() {
|
||||
|
||||
@Override
|
||||
public void onRemoteNodeEvent(@NotNull final Set<AddressAndPort> usedHostPorts,
|
||||
@NotNull final ATSDirectBitSet usedIdentifiers,
|
||||
@NotNull final Set<BytesExternalizableImpl.ProposedIdentifierWithHost> proposedHostPortIdentifiers) {
|
||||
|
||||
knownHostPorts.addAll(usedHostPorts);
|
||||
|
||||
orBitSets(usedIdentifiers, knownIdentifiers);
|
||||
|
||||
for (BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost : proposedHostPortIdentifiers) {
|
||||
if (!proposedIdentifierWithHost.addressAndPort().equals(ourAddressAndPort)) {
|
||||
|
||||
int remoteIdentifier = proposedIdentifierWithHost.identifier();
|
||||
knownIdentifiers.set(remoteIdentifier, true);
|
||||
knownHostPorts.add(proposedIdentifierWithHost.addressAndPort());
|
||||
|
||||
if (remoteIdentifier == proposedIdentifier.get())
|
||||
useAnotherIdentifier.set(true);
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
final BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener);
|
||||
|
||||
final BytesExternalizableImpl.ProposedIdentifierWithHost ourHostPort = new
|
||||
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();
|
||||
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();
|
||||
Thread.sleep(10);
|
||||
byte identifier;
|
||||
|
||||
|
||||
boolean isFistTime = true;
|
||||
|
||||
for (; ; ) {
|
||||
|
||||
identifier = proposeRandomUnusedIdentifier(knownIdentifiers, isFistTime);
|
||||
proposedIdentifier.set(identifier);
|
||||
|
||||
isFistTime = false;
|
||||
|
||||
final BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost = new
|
||||
BytesExternalizableImpl.ProposedIdentifierWithHost(ourAddressAndPort, identifier);
|
||||
|
||||
externalizable.setOurProposedIdentifier(proposedIdentifierWithHost);
|
||||
externalizable.sendBootStrap();
|
||||
|
||||
Thread.sleep(10);
|
||||
|
||||
for (int j = 0; j < 3; j++) {
|
||||
|
||||
externalizable.sendBootStrap();
|
||||
Thread.sleep(10);
|
||||
|
||||
if (useAnotherIdentifier.get()) {
|
||||
// given that another node host proposed the same identifier, we will choose a different one.
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, this);
|
||||
|
||||
final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster
|
||||
= new NodeDiscoveryBroadcaster(udpConfig, 1024, externalizable);
|
||||
|
||||
externalizable.setModificationNotifier(nodeDiscoveryBroadcaster);
|
||||
externalizable.sendBootStrap();
|
||||
Thread.sleep(100);
|
||||
externalizable.sendBootStrap();
|
||||
// we have to wait a while, we have not way of knowing how long it will take for a UDP message to
|
||||
// arrive, but the more UDP messages we receive the better, is it will make our selection of a
|
||||
// unique identifier more reliable.
|
||||
Thread.sleep(100);
|
||||
|
||||
// we should make a local copy as this may change
|
||||
|
||||
final IdentifierListener identifierListener = new IdentifierListener() {
|
||||
|
||||
final ConcurrentMap<Byte, SocketAddress> identifiers = new ConcurrentHashMap<Byte,
|
||||
SocketAddress>();
|
||||
|
||||
@Override
|
||||
public boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress) {
|
||||
final SocketAddress socketAddress = identifiers.putIfAbsent(remoteIdentifier, remoteAddress);
|
||||
return socketAddress == null;
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
final TcpReplicationConfig tcpConfig = TcpReplicationConfig
|
||||
.of(tcpPort, toInetSocketArray(knownHostPorts))
|
||||
.heartBeatInterval(1, SECONDS).nonUniqueIdentifierListener(identifierListener);
|
||||
|
||||
return ChronicleMapBuilder.of(Integer.class,
|
||||
CharSequence.class)
|
||||
.entries(20000L)
|
||||
.addReplicator(tcp(identifier, tcpConfig)).create();
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId) {
|
||||
private InetSocketAddress[] toInetSocketArray(Set<AddressAndPort> source) throws
|
||||
UnknownHostException {
|
||||
|
||||
if (remoteProposedId == -1)
|
||||
bitSet.set(remoteProposedId);
|
||||
// make a safe copy
|
||||
final HashSet<AddressAndPort> addressAndPorts = new HashSet<AddressAndPort>(source);
|
||||
|
||||
// propose and identifier
|
||||
if (addressAndPorts.isEmpty())
|
||||
return new InetSocketAddress[0];
|
||||
|
||||
final InetSocketAddress[] addresses = new InetSocketAddress[addressAndPorts.size()];
|
||||
|
||||
int i = 0;
|
||||
|
||||
for (final AddressAndPort addressAndPort : addressAndPorts) {
|
||||
addresses[i++] = new InetSocketAddress(InetAddress.getByAddress(addressAndPort.address())
|
||||
.getHostAddress(), addressAndPort.port());
|
||||
}
|
||||
return addresses;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset
|
||||
* and returns the destination
|
||||
*
|
||||
* @param source
|
||||
* @param destination
|
||||
* @return
|
||||
*/
|
||||
private DirectBitSet orBitSets(@NotNull final DirectBitSet source,
|
||||
@NotNull final DirectBitSet destination) {
|
||||
|
||||
// merges the two bit-sets together
|
||||
for (int i = (int) source.nextSetBit(0); i > 0;
|
||||
i = (int) source.nextSetBit(i + 1)) {
|
||||
try {
|
||||
destination.set(i, true);
|
||||
} catch (IndexOutOfBoundsException e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
}
|
||||
|
||||
return destination;
|
||||
}
|
||||
|
||||
|
||||
private byte proposeRandomUnusedIdentifier(final DirectBitSet knownIdentifiers, boolean isFirstTime) throws UnknownHostException {
|
||||
byte possible;
|
||||
|
||||
|
||||
// the first time, rather than choosing a random number, we will choose the last value of the IP
|
||||
// address as our random number, ( or at least something that is based upon it)
|
||||
if (isFirstTime) {
|
||||
byte[] address = InetAddress.getLocalHost().getAddress();
|
||||
int lastAddress = address[address.length - 1];
|
||||
if (lastAddress > 127)
|
||||
lastAddress = lastAddress - 127;
|
||||
if (lastAddress > 127)
|
||||
lastAddress = lastAddress - 127;
|
||||
|
||||
possible = (byte) lastAddress;
|
||||
} else
|
||||
possible = (byte) (Math.random() * 128);
|
||||
|
||||
int count = 0;
|
||||
for (; ; ) {
|
||||
|
||||
if (knownIdentifiers.setIfClear(possible)) {
|
||||
return possible;
|
||||
}
|
||||
|
||||
count++;
|
||||
|
||||
if (count == 128) {
|
||||
throw new IllegalStateException("The grid is full, its not possible for any more nodes to " +
|
||||
"going the grid.");
|
||||
}
|
||||
|
||||
if (possible == 128)
|
||||
possible = 0;
|
||||
else
|
||||
possible++;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* creates a bit set based on a number of bits
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import java.net.SocketAddress;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public interface IdentifierListener {
|
||||
|
||||
/**
|
||||
* checks the identifier that is unique and we haven't see it before, unless it comes from the same port
|
||||
* and host.
|
||||
*
|
||||
* @param remoteIdentifier
|
||||
* @param remoteAddress
|
||||
* @return
|
||||
*/
|
||||
boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress);
|
||||
|
||||
}
|
||||
@@ -9,14 +9,19 @@ import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.DatagramChannel;
|
||||
import java.util.Comparator;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashSet;
|
||||
import java.util.Queue;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
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 net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES;
|
||||
import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
|
||||
|
||||
@@ -42,7 +47,6 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
|
||||
BOOTSTRAP_BYTES.append(BOOTSTRAP);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @param replicationConfig
|
||||
* @param externalizable
|
||||
@@ -56,15 +60,10 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
|
||||
|
||||
super(replicationConfig, serializedEntrySize, UNUSED);
|
||||
|
||||
|
||||
final UdpSocketChannelEntryWriter writer = new UdpSocketChannelEntryWriter(1024, externalizable,
|
||||
this);
|
||||
final UdpSocketChannelEntryReader reader = new UdpSocketChannelEntryReader(1024, externalizable
|
||||
);
|
||||
|
||||
final UdpSocketChannelEntryWriter writer = new UdpSocketChannelEntryWriter(1024, externalizable, this);
|
||||
final UdpSocketChannelEntryReader reader = new UdpSocketChannelEntryReader(1024, externalizable);
|
||||
|
||||
setReader(reader);
|
||||
|
||||
setWriter(writer);
|
||||
|
||||
start();
|
||||
@@ -72,12 +71,10 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
|
||||
|
||||
private static class UdpSocketChannelEntryReader implements EntryReader {
|
||||
|
||||
|
||||
private final ByteBuffer in;
|
||||
private final ByteBufferBytes out;
|
||||
private final BytesExternalizable externalizable;
|
||||
|
||||
|
||||
/**
|
||||
* @param serializedEntrySize the maximum size of an entry include the meta data
|
||||
* @param externalizable supports reading and writing serialize entries
|
||||
@@ -135,11 +132,9 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
|
||||
|
||||
public static class UdpSocketChannelEntryWriter implements EntryWriter {
|
||||
|
||||
|
||||
private final ByteBuffer out;
|
||||
private final ByteBufferBytes in;
|
||||
|
||||
|
||||
@NotNull
|
||||
private final BytesExternalizable externalizable;
|
||||
private UdpChannelReplicator udpReplicator;
|
||||
@@ -154,8 +149,6 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
|
||||
// we make the buffer twice as large just to give ourselves headroom
|
||||
out = ByteBuffer.allocateDirect(serializedEntrySize * 2);
|
||||
in = new ByteBufferBytes(out);
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -194,19 +187,15 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
|
||||
|
||||
long size = in.position() - start;
|
||||
|
||||
|
||||
// we'll write the size inverted at the start
|
||||
in.writeShort(0, ~((int) size));
|
||||
in.writeUnsignedShort(SIZE_OF_SHORT, (int) size);
|
||||
|
||||
|
||||
out.limit((int) in.position());
|
||||
|
||||
udpReplicator.disableWrites();
|
||||
|
||||
return socketChannel.write(out);
|
||||
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@@ -230,48 +219,31 @@ class RemoteNodes {
|
||||
|
||||
|
||||
private final Bytes activeIdentifiersBitSetBytes;
|
||||
private ConcurrentSkipListSet inetSocketAddresses;
|
||||
private final ConcurrentSkipListSet<AddressAndPort> addressAndPorts;
|
||||
private final ATSDirectBitSet atsDirectBitSet;
|
||||
private byte ourProposedID = -1;
|
||||
private final ATSDirectBitSet allProposedID;
|
||||
|
||||
|
||||
/**
|
||||
* @param activeIdentifiersBitSetBytes byte sof a bitset containing the known identifiers
|
||||
*/
|
||||
RemoteNodes(final Bytes activeIdentifiersBitSetBytes) {
|
||||
|
||||
this.inetSocketAddresses = new ConcurrentSkipListSet<InetSocketAddress>(new Comparator<InetSocketAddress>() {
|
||||
|
||||
@Override
|
||||
public int compare(InetSocketAddress o1, InetSocketAddress o2) {
|
||||
|
||||
int result = Integer.compare(o1.getPort(), o2.getPort());
|
||||
if (result != 0)
|
||||
return result;
|
||||
|
||||
return o1.getHostName().compareTo(o2.getHostName());
|
||||
|
||||
}
|
||||
});
|
||||
this.addressAndPorts = new ConcurrentSkipListSet<AddressAndPort>();
|
||||
|
||||
this.activeIdentifiersBitSetBytes = activeIdentifiersBitSetBytes;
|
||||
this.atsDirectBitSet = new ATSDirectBitSet(this.activeIdentifiersBitSetBytes);
|
||||
|
||||
// this will only hold bits in it while we in the phase of searching for an id
|
||||
this.allProposedID = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate(128 / 8)));
|
||||
}
|
||||
|
||||
|
||||
public Set<InetSocketAddress> inetSocketAddresses() {
|
||||
return inetSocketAddresses;
|
||||
public Set<AddressAndPort> addressAndPorts() {
|
||||
return addressAndPorts;
|
||||
}
|
||||
|
||||
public Bytes activeIdentifierBytes() {
|
||||
return activeIdentifiersBitSetBytes;
|
||||
}
|
||||
|
||||
public void add(InetSocketAddress inetSocketAddress) {
|
||||
inetSocketAddresses.add(inetSocketAddress);
|
||||
public void add(AddressAndPort inetSocketAddress) {
|
||||
addressAndPorts.add(inetSocketAddress);
|
||||
}
|
||||
|
||||
public DirectBitSet activeIdentifierBitSet() {
|
||||
@@ -279,15 +251,88 @@ class RemoteNodes {
|
||||
}
|
||||
|
||||
|
||||
public byte ourProposedID() {
|
||||
return ourProposedID;
|
||||
}
|
||||
|
||||
|
||||
class AddressAndPort implements Comparable<AddressAndPort> {
|
||||
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 DirectBitSet allProposedID() {
|
||||
return allProposedID;
|
||||
public byte[] getAddress() {
|
||||
return address;
|
||||
}
|
||||
|
||||
public short getPort() {
|
||||
return port;
|
||||
}
|
||||
|
||||
public byte[] address() {
|
||||
|
||||
return address;
|
||||
}
|
||||
|
||||
public short port() {
|
||||
return port;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) return true;
|
||||
if (o == null || getClass() != o.getClass()) return false;
|
||||
|
||||
AddressAndPort that = (AddressAndPort) o;
|
||||
|
||||
if (port != that.port) return false;
|
||||
if (!Arrays.equals(address, that.address)) return false;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int result = Arrays.hashCode(address);
|
||||
result = 31 * result + (int) port;
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public int compareTo(AddressAndPort o) {
|
||||
int i = 0;
|
||||
for (byte b : address) {
|
||||
int compare = Byte.compare(b, o.address[i++]);
|
||||
if (compare != 0)
|
||||
return compare;
|
||||
|
||||
}
|
||||
return Short.compare(port, o.port);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "AddressAndPort{" +
|
||||
"address=" + numericToTextFormat(address) +
|
||||
", port=" + port +
|
||||
'}';
|
||||
}
|
||||
|
||||
static String numericToTextFormat(byte[] src) {
|
||||
return (src[0] & 0xff) + "." + (src[1] & 0xff) + "." + (src[2] & 0xff) + "." + (src[3] & 0xff);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
class BytesExternalizableImpl implements BytesExternalizable {
|
||||
|
||||
private final RemoteNodes allNodes;
|
||||
@@ -295,6 +340,16 @@ class BytesExternalizableImpl implements BytesExternalizable {
|
||||
private final AtomicBoolean bootstrapRequired = new AtomicBoolean();
|
||||
private final UDPEventListener udpEventListener;
|
||||
|
||||
private ProposedIdentifierWithHost ourProposedIdentifier;
|
||||
|
||||
public ProposedIdentifierWithHost getOurProposedIdentifier() {
|
||||
return ourProposedIdentifier;
|
||||
}
|
||||
|
||||
public void setOurProposedIdentifier(ProposedIdentifierWithHost ourProposedIdentifier) {
|
||||
this.ourProposedIdentifier = ourProposedIdentifier;
|
||||
}
|
||||
|
||||
public void setModificationNotifier(Replica.ModificationNotifier modificationNotifier) {
|
||||
this.modificationNotifier = modificationNotifier;
|
||||
|
||||
@@ -319,24 +374,40 @@ class BytesExternalizableImpl implements BytesExternalizable {
|
||||
|
||||
|
||||
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());
|
||||
return;
|
||||
}
|
||||
|
||||
final Set<InetSocketAddress> inetSocketAddresses = allNodes.inetSocketAddresses();
|
||||
final Set<AddressAndPort> addressAndPorts = allNodes.addressAndPorts();
|
||||
|
||||
// write the number of hosts and ports
|
||||
short count = (short) inetSocketAddresses.size();
|
||||
short count = (short) addressAndPorts.size();
|
||||
destination.writeUnsignedShort(count);
|
||||
|
||||
// write all the host and ports
|
||||
for (InetSocketAddress inetSocketAddress : inetSocketAddresses) {
|
||||
destination.writeUTF(inetSocketAddress.getHostName());
|
||||
destination.writeInt(inetSocketAddress.getPort());
|
||||
for (AddressAndPort inetSocketAddress : addressAndPorts) {
|
||||
for (byte b : inetSocketAddress.address()) {
|
||||
destination.writeByte(b);
|
||||
}
|
||||
|
||||
destination.writeShort(inetSocketAddress.port());
|
||||
}
|
||||
|
||||
destination.write(allNodes.ourProposedID());
|
||||
// destination.write(allNodes.ourProposedID());
|
||||
|
||||
Bytes bytes = allNodes.activeIdentifierBytes();
|
||||
//bytes.position(0);
|
||||
@@ -346,36 +417,123 @@ class BytesExternalizableImpl implements BytesExternalizable {
|
||||
// we will store the size of the bitset
|
||||
destination.writeUnsignedShort((int) bytes.remaining());
|
||||
|
||||
// then write it out
|
||||
destination.write(bytes);
|
||||
// we are no going to broadcast all the nodes that have been bootstaping in the last 200
|
||||
expireOldBootstrapStats(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;
|
||||
|
||||
// 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<AddressAndPort, ProposedIdentifierWithHost> bootstrapStatsMap = new
|
||||
ConcurrentHashMap<AddressAndPort, ProposedIdentifierWithHost>();
|
||||
|
||||
final Queue<ProposedIdentifierWithHost> proposedIdentifierWithHostExpiryQueue = new ConcurrentLinkedQueue<ProposedIdentifierWithHost>();
|
||||
|
||||
|
||||
private void expireOldBootstrapStats(long timeOlderThan) {
|
||||
for (; ; ) {
|
||||
|
||||
final ProposedIdentifierWithHost proposedIdentifierWithHost = proposedIdentifierWithHostExpiryQueue.peek();
|
||||
|
||||
if (proposedIdentifierWithHost == null)
|
||||
break;
|
||||
|
||||
if (proposedIdentifierWithHost.timestamp < timeOlderThan) {
|
||||
// only remote it if it has not changed
|
||||
bootstrapStatsMap.remove(proposedIdentifierWithHost.addressAndPort, proposedIdentifierWithHost);
|
||||
}
|
||||
|
||||
proposedIdentifierWithHostExpiryQueue.poll();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void readExternalBytes(@NotNull Bytes source) {
|
||||
|
||||
if (isBootstrap(source)) {
|
||||
final ProposedIdentifierWithHost bootstrap = readBootstrapMessage(source);
|
||||
if (bootstrap != null) {
|
||||
LOG.debug("received Bootstrap");
|
||||
|
||||
// we've received a bootstrap message so will will now rebroadcast what we know
|
||||
onChange();
|
||||
bootstrapStatsMap.put(bootstrap.addressAndPort, bootstrap);
|
||||
proposedIdentifierWithHostExpiryQueue.add(bootstrap);
|
||||
|
||||
|
||||
try {
|
||||
|
||||
// we've received a bootstrap message so will will now rebroadcast what we know,
|
||||
// after a random delay
|
||||
|
||||
Thread.sleep((int) (Math.random() * 9.0));
|
||||
|
||||
onChange();
|
||||
|
||||
} catch (InterruptedException e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
int count = source.readUnsignedShort();
|
||||
|
||||
|
||||
final Set<AddressAndPort> allNodeInTheGrid = new HashSet<AddressAndPort>();
|
||||
|
||||
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();
|
||||
|
||||
final String host = source.readUTF();
|
||||
final int port = source.readInt();
|
||||
final short port = source.readShort();
|
||||
|
||||
LOG.debug("received InetSocketAddress(" + host + "," + port + ")");
|
||||
allNodes.add(new InetSocketAddress(host, port));
|
||||
LOG.debug("received InetSocketAddress(" + address + "," + port + ")");
|
||||
allNodeInTheGrid.add(new AddressAndPort(address, port));
|
||||
}
|
||||
|
||||
|
||||
byte remoteProposedId = source.readByte();
|
||||
int countRebroadcast = source.readShort();
|
||||
|
||||
final Set<ProposedIdentifierWithHost> proposedIdentifierWithHosts = new
|
||||
HashSet<ProposedIdentifierWithHost>();
|
||||
|
||||
// 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();
|
||||
@@ -385,14 +543,14 @@ class BytesExternalizableImpl implements BytesExternalizable {
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
//todo we should not have to do this ! - maybe a bug in BitSet of ByteBuffer
|
||||
Bytes sourceBytes = toNewBuffer(source);
|
||||
Bytes usedIdentifiers = toNewBuffer(source);
|
||||
|
||||
// merges the source into the destination and returns the destination
|
||||
orBitSets(new ATSDirectBitSet(sourceBytes), allNodes.activeIdentifierBitSet());
|
||||
udpEventListener.onRemoteNodeEvent(allNodeInTheGrid,
|
||||
new ATSDirectBitSet(usedIdentifiers),
|
||||
proposedIdentifierWithHosts);
|
||||
|
||||
|
||||
udpEventListener.onRemoteNodeEvent(allNodes, remoteProposedId);
|
||||
}
|
||||
|
||||
|
||||
@@ -426,11 +584,62 @@ class BytesExternalizableImpl implements BytesExternalizable {
|
||||
}
|
||||
|
||||
|
||||
public static class ProposedIdentifierWithHost {
|
||||
|
||||
private final int identifier;
|
||||
private final long timestamp;
|
||||
private final AddressAndPort addressAndPort;
|
||||
|
||||
public ProposedIdentifierWithHost(@NotNull final AddressAndPort addressAndPort,
|
||||
byte identifier) {
|
||||
this.addressAndPort = addressAndPort;
|
||||
this.identifier = identifier;
|
||||
this.timestamp = System.currentTimeMillis();
|
||||
}
|
||||
|
||||
public ProposedIdentifierWithHost(byte[] address, short port, byte identifier) {
|
||||
this(new AddressAndPort(address, port), identifier);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) return true;
|
||||
if (o == null || getClass() != o.getClass()) return false;
|
||||
|
||||
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;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int result = identifier;
|
||||
result = 31 * result + (int) (timestamp ^ (timestamp >>> 32));
|
||||
result = 31 * result + (addressAndPort != null ? addressAndPort.hashCode() : 0);
|
||||
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'
|
||||
*/
|
||||
private boolean isBootstrap(Bytes source) {
|
||||
private ProposedIdentifierWithHost readBootstrapMessage(Bytes source) {
|
||||
|
||||
final long start = source.position();
|
||||
|
||||
@@ -439,58 +648,40 @@ class BytesExternalizableImpl implements BytesExternalizable {
|
||||
long size = BOOTSTRAP_BYTES.limit();
|
||||
|
||||
if (size < source.remaining())
|
||||
return false;
|
||||
|
||||
if (source.remaining() < size)
|
||||
return false;
|
||||
return null;
|
||||
|
||||
for (int i = 0; i < size; i++) {
|
||||
|
||||
final byte actualByte = source.readByte(start + i);
|
||||
final byte byteRead = source.readByte(start + i);
|
||||
final byte expectedByte = BOOTSTRAP_BYTES.readByte(i);
|
||||
|
||||
if (!(expectedByte == actualByte))
|
||||
return false;
|
||||
if (!(expectedByte == byteRead))
|
||||
return null;
|
||||
}
|
||||
|
||||
return true;
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset
|
||||
* and returns the destination
|
||||
*
|
||||
* @param source
|
||||
* @param destination
|
||||
* @return
|
||||
*/
|
||||
private DirectBitSet orBitSets(@NotNull final DirectBitSet source,
|
||||
@NotNull final DirectBitSet destination) {
|
||||
|
||||
|
||||
// merges the two bit-sets together
|
||||
for (int i = (int) source.nextSetBit(0); i > 0;
|
||||
i = (int) source.nextSetBit(i + 1)) {
|
||||
try {
|
||||
destination.set(i, true);
|
||||
} catch (IndexOutOfBoundsException e) {
|
||||
LOG.error("", e);
|
||||
}
|
||||
}
|
||||
|
||||
return destination;
|
||||
}
|
||||
|
||||
|
||||
public void sendBootStrap() {
|
||||
bootstrapRequired.set(true);
|
||||
}
|
||||
|
||||
public void add(InetSocketAddress interfaceAddress) {
|
||||
allNodes.add(interfaceAddress);
|
||||
public void add(AddressAndPort addressAndPort) {
|
||||
allNodes.add(addressAndPort);
|
||||
}
|
||||
|
||||
|
||||
@@ -504,10 +695,9 @@ interface UDPEventListener {
|
||||
|
||||
/**
|
||||
* called when we have received a UDP message, this is called after the message has been parsed
|
||||
*
|
||||
* @param allNodes
|
||||
* @param remoteProposedId a remote node has proposed using this as an id
|
||||
*/
|
||||
|
||||
void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId);
|
||||
void onRemoteNodeEvent(Set<AddressAndPort> usedHostPorts,
|
||||
ATSDirectBitSet usedIdentifers,
|
||||
Set<ProposedIdentifierWithHost> proposedHostPortIdentifiers);
|
||||
}
|
||||
|
||||
@@ -1,9 +0,0 @@
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public interface NonUniqueIdentifierListener {
|
||||
|
||||
void onNonUniqueIdentifier(byte remoteIdentifier);
|
||||
}
|
||||
@@ -16,6 +16,8 @@
|
||||
|
||||
package net.openhft.chronicle.map;
|
||||
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
@@ -41,34 +43,38 @@ public class TcpReplicationConfig {
|
||||
private ThrottlingConfig throttlingConfig = ThrottlingConfig.noThrottling();
|
||||
private long heartBeatInterval = DEFAULT_HEART_BEAT_INTERVAL;
|
||||
private TimeUnit heartBeatIntervalUnit = DEFAULT_HEART_BEAT_INTERVAL_UNIT;
|
||||
private NonUniqueIdentifierListener nonUniqueIdentifierListener;
|
||||
public IdentifierListener identifierListener;
|
||||
|
||||
TcpReplicationConfig() {
|
||||
|
||||
}
|
||||
|
||||
public NonUniqueIdentifierListener nonUniqueIdentifierListener() {
|
||||
return nonUniqueIdentifierListener;
|
||||
public IdentifierListener identifierListener() {
|
||||
return identifierListener;
|
||||
}
|
||||
|
||||
public void nonUniqueIdentifierListener(NonUniqueIdentifierListener nonUniqueIdentifierListener) {
|
||||
this.nonUniqueIdentifierListener = nonUniqueIdentifierListener;
|
||||
public TcpReplicationConfig nonUniqueIdentifierListener(@NotNull final IdentifierListener
|
||||
identifierListener) {
|
||||
this.identifierListener = identifierListener;
|
||||
return this;
|
||||
}
|
||||
|
||||
public boolean autoReconnectedUponDroppedConnection() {
|
||||
return autoReconnectedUponDroppedConnection;
|
||||
}
|
||||
|
||||
public void autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) {
|
||||
public TcpReplicationConfig autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) {
|
||||
this.autoReconnectedUponDroppedConnection = autoReconnectedUponDroppedConnection;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ThrottlingConfig throttlingConfig() {
|
||||
return throttlingConfig;
|
||||
}
|
||||
|
||||
public void throttlingConfig(ThrottlingConfig throttlingConfig) {
|
||||
public TcpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
|
||||
this.throttlingConfig = throttlingConfig;
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
@@ -84,15 +90,16 @@ public class TcpReplicationConfig {
|
||||
return serverPort;
|
||||
}
|
||||
|
||||
public void serverPort(int serverPort) {
|
||||
public TcpReplicationConfig serverPort(int serverPort) {
|
||||
this.serverPort = serverPort;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Set<InetSocketAddress> endpoints() {
|
||||
return endpoints;
|
||||
}
|
||||
|
||||
public void endpoints(Set<InetSocketAddress> endpoints) {
|
||||
public TcpReplicationConfig endpoints(Set<InetSocketAddress> endpoints) {
|
||||
|
||||
for (final InetSocketAddress endpoint : endpoints) {
|
||||
if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName()))
|
||||
@@ -101,6 +108,7 @@ public class TcpReplicationConfig {
|
||||
}
|
||||
|
||||
this.endpoints = endpoints;
|
||||
return this;
|
||||
}
|
||||
|
||||
public int packetSize() {
|
||||
@@ -133,15 +141,17 @@ public class TcpReplicationConfig {
|
||||
}
|
||||
|
||||
|
||||
public void packetSize(int packetSize) {
|
||||
public TcpReplicationConfig packetSize(int packetSize) {
|
||||
if (packetSize <= 0)
|
||||
throw new IllegalArgumentException();
|
||||
this.packetSize = packetSize;
|
||||
return this;
|
||||
}
|
||||
|
||||
public void throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) {
|
||||
public TcpReplicationConfig throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) {
|
||||
this.heartBeatInterval = heartBeatInterval;
|
||||
this.heartBeatIntervalUnit = heartBeatIntervalUnit;
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
@@ -194,4 +204,6 @@ public class TcpReplicationConfig {
|
||||
this.heartBeatIntervalUnit = unit;
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
@@ -23,10 +23,7 @@ import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.ConnectException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.SocketException;
|
||||
import java.net.*;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.*;
|
||||
import java.util.BitSet;
|
||||
@@ -86,7 +83,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
|
||||
selectorTimeout = Math.min(heartBeatIntervalMillis, throttleBucketInterval);
|
||||
|
||||
|
||||
this.replica = replica;
|
||||
this.localIdentifier = replica.identifier();
|
||||
this.maxEntrySizeBytes = maxEntrySizeBytes;
|
||||
@@ -424,19 +420,27 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
|
||||
// this can occur sometimes when if 2 or more remote node attempt to use node discovery at the
|
||||
// same time
|
||||
if (remoteIdentifier == localIdentifier) {
|
||||
|
||||
final NonUniqueIdentifierListener listener = replicationConfig.nonUniqueIdentifierListener();
|
||||
final IdentifierListener identifierListener = replicationConfig.identifierListener;
|
||||
final SocketAddress remoteAddress = socketChannel.getRemoteAddress();
|
||||
|
||||
boolean identifierIsUnique;
|
||||
|
||||
if (identifierListener != null) {
|
||||
identifierIsUnique = identifierListener.isIdentifierUnique(remoteIdentifier, remoteAddress);
|
||||
} else {
|
||||
identifierIsUnique = remoteIdentifier != localIdentifier;
|
||||
}
|
||||
|
||||
if (identifierIsUnique) {
|
||||
|
||||
if (listener != null)
|
||||
listener.onNonUniqueIdentifier(remoteIdentifier);
|
||||
|
||||
// throwing this exception will cause us to disconnect, both the client and server
|
||||
// will be able to detect the the remote and local identifiers are the same,
|
||||
// as the identifier is send early on in the hand shaking via the connect(() and accept()
|
||||
// methods
|
||||
throw new IllegalStateException("dropping connection, as the remote and local-identifier " +
|
||||
"are the same, identifier=" + localIdentifier);
|
||||
throw new IllegalStateException("dropping connection, " +
|
||||
"as the remote-identifier is already being used, identifier=" + remoteIdentifier);
|
||||
}
|
||||
|
||||
attached.remoteModificationIterator = replica.acquireModificationIterator(remoteIdentifier,
|
||||
|
||||
Reference in New Issue
Block a user