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
@@ -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<AddressAndPort, BytesExternalizableImpl.ProposedIdentifierWithHost> proposedIdentifiersWithHost) {
|
||||
public void onRemoteNodeEvent(KnownNodes remoteNode, ConcurrentExpiryMap<AddressAndPort, BytesExternalizableImpl.ProposedNodes> 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;
|
||||
|
||||
|
||||
|
||||
@@ -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<AddressAndPort> addressAndPorts = new ConcurrentSkipListSet<AddressAndPort>();
|
||||
private ATSDirectBitSet atsDirectBitSet;
|
||||
|
||||
RemoteNodes() {
|
||||
KnownNodes() {
|
||||
|
||||
this.activeIdentifiersBitSetBytes = new ByteBufferBytes(ByteBuffer.allocate(128 / 8));
|
||||
this.addressAndPorts = new ConcurrentSkipListSet<AddressAndPort>();
|
||||
@@ -407,22 +407,22 @@ class AddressAndPort implements Comparable<AddressAndPort>, 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<AddressAndPort, ProposedIdentifierWithHost> proposedIdentifiersWithHost = new
|
||||
ConcurrentExpiryMap<AddressAndPort, ProposedIdentifierWithHost>(AddressAndPort.class,
|
||||
ProposedIdentifierWithHost.class);
|
||||
final ConcurrentExpiryMap<AddressAndPort, ProposedNodes> proposedIdentifiersWithHost = new
|
||||
ConcurrentExpiryMap<AddressAndPort, ProposedNodes>(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<AddressAndPort, ProposedIdentifierWithHost> proposedIdentifiersWithHost);
|
||||
void onRemoteNodeEvent(KnownNodes remoteNode, ConcurrentExpiryMap<AddressAndPort, BytesExternalizableImpl.ProposedNodes> proposedIdentifiersWithHost);
|
||||
}
|
||||
|
||||
class ConcurrentExpiryMap<K extends BytesMarshallable, V extends BytesMarshallable> implements BytesMarshallable {
|
||||
@@ -680,7 +678,7 @@ class ConcurrentExpiryMap<K extends BytesMarshallable, V extends BytesMarshallab
|
||||
for (int i = 0; i < size; i++) {
|
||||
|
||||
final K k = kClass.newInstance();
|
||||
k.writeMarshallable(in);
|
||||
k.readMarshallable(in);
|
||||
|
||||
final V v = vClass.newInstance();
|
||||
v.readMarshallable(in);
|
||||
|
||||
@@ -2,7 +2,6 @@ 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;
|
||||
@@ -15,27 +14,7 @@ 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
|
||||
}
|
||||
|
||||
private ChronicleMap<Integer, CharSequence> map1;
|
||||
private ChronicleMap<Integer, CharSequence> map2;
|
||||
private volatile byte proposedIdentifier = 1;
|
||||
private volatile boolean useAnotherIdentifier = false;
|
||||
|
||||
|
||||
@Test
|
||||
@Ignore
|
||||
public void test2() throws IOException, InterruptedException {
|
||||
|
||||
DiscoveryCluster discoveryCluster = new DiscoveryCluster();
|
||||
|
||||
ChronicleMap<Integer, CharSequence> 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<AddressAndPort, BytesExternalizableImpl.ProposedIdentifierWithHost> 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)));
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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<String, ByteableIntValue> 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();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user