Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-134 ( release draft cut of the code so that I can test it across my network )

Browse Source
This commit is contained in:
Rob Austin
2014-10-06 16:01:41 +01:00
parent 085d15f1d9
commit 9205498530
2 changed files with 22 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+20 -11
View File
@@ -42,7 +42,7 @@ public class NodeDiscovery {
final KnownNodes knownNodes = new KnownNodes();
final Set<AddressAndPort> knownHostPorts = new ConcurrentSkipListSet<AddressAndPort>();
final DirectBitSet knownIdentifiers = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate
final DirectBitSet knownAndProposedIdentifiers = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate
(128 / 8)));
final AtomicReference<CountDownLatch> countDownLatch = new AtomicReference<CountDownLatch>(new
@@ -51,21 +51,26 @@ public class NodeDiscovery {
final NodeDiscoveryEventListener nodeDiscoveryEventListener = new NodeDiscoveryEventListener() {
@Override
public void onRemoteNodeEvent(KnownNodes remoteNode, ConcurrentExpiryMap<AddressAndPort, DiscoveryNodeBytesMarshallable.ProposedNodes> proposedIdentifiersWithHost) {
public void onRemoteNodeEvent(KnownNodes remoteNodes,
ConcurrentExpiryMap<AddressAndPort,
DiscoveryNodeBytesMarshallable.ProposedNodes> proposedIdentifiersWithHost) {
knownHostPorts.addAll(remoteNode.addressAndPorts());
knownHostPorts.addAll(remoteNodes.addressAndPorts());
orBitSets(remoteNode.activeIdentifierBitSet(), knownIdentifiers);
orBitSets(remoteNodes.identifers(), knownAndProposedIdentifiers);
for (DiscoveryNodeBytesMarshallable.ProposedNodes proposedIdentifierWithHost :
proposedIdentifiersWithHost.values()) {
if (!proposedIdentifierWithHost.addressAndPort().equals(ourAddressAndPort)) {
int remoteIdentifier = proposedIdentifierWithHost.identifier();
knownIdentifiers.set(remoteIdentifier, true);
int proposedIdentifer = proposedIdentifierWithHost.identifier();
if (proposedIdentifer!=-1) {
knownAndProposedIdentifiers.set(proposedIdentifer, true);
}
knownHostPorts.add(proposedIdentifierWithHost.addressAndPort());
if (remoteIdentifier == proposedIdentifier.get())
if (proposedIdentifer == proposedIdentifier.get())
useAnotherIdentifier.set(true);
countDownLatch.get().countDown();
@@ -109,7 +114,7 @@ public class NodeDiscovery {
OUTER:
for (; ; ) {
useAnotherIdentifier.set(false);
identifier = proposeRandomUnusedIdentifier(knownIdentifiers, isFistTime);
identifier = proposeRandomUnusedIdentifier(knownAndProposedIdentifiers, isFistTime);
proposedIdentifier.set(identifier);
LOG.info("proposing to use identifier=" + identifier);
@@ -156,7 +161,7 @@ public class NodeDiscovery {
@Override
public boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress) {
final SocketAddress socketAddress = identifiers.putIfAbsent(remoteIdentifier, remoteAddress);
knownNodes.activeIdentifierBitSet().set(remoteIdentifier);
knownNodes.identifers().set(remoteIdentifier);
return socketAddress == null;
}
};
@@ -468,11 +473,15 @@ class KnownNodes implements BytesMarshallable {
}
public void add(AddressAndPort inetSocketAddress, byte identifier) {
activeIdentifierBitSet().set(identifier);
identifers().set(identifier);
addressAndPorts.add(inetSocketAddress);
}
public DirectBitSet activeIdentifierBitSet() {
/**
* all the known identifiers
* @return
*/
public DirectBitSet identifers() {
return atsDirectBitSet;
}