* It keeps a local scheduler that periodically sends Gossip ticks to
* itself to send bucket store's bucket versions to a randomly selected remote
* gossiper.
- *
+ *
+ *
* When bucket versions are received from a remote gossiper, it is compared
* with bucket store's bucket versions. Which ever buckets are newer
* locally, are sent to remote gossiper. If any bucket is older in bucket store,
* a gossip status is sent to remote gossiper so that it can send the newer buckets.
- *
+ *
+ *
* When a bucket is received from a remote gossiper, its sent to the bucket store
* for update.
- *
*/
+public class Gossiper extends AbstractUntypedActorWithMetering {
-public class Gossiper extends UntypedActor {
+ private final Logger log = LoggerFactory.getLogger(getClass());
- final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
-
- Cluster cluster = Cluster.get(getContext().system());
+ private Cluster cluster;
/**
* ActorSystem's address for the current cluster node.
*/
- private Address selfAddress = cluster.selfAddress();
+ private Address selfAddress;
/**
- * All known cluster members
+ * All known cluster members.
*/
private List
clusterMembers = new ArrayList<>();
@@ -76,29 +80,41 @@ public class Gossiper extends UntypedActor {
private Boolean autoStartGossipTicks = true;
- public Gossiper(){}
+ private final RemoteRpcProviderConfig config;
+
+ public Gossiper(RemoteRpcProviderConfig config) {
+ this.config = Preconditions.checkNotNull(config);
+ }
/**
- * Helpful for testing
+ * Constructor for testing.
+ *
* @param autoStartGossipTicks used for turning off gossip ticks during testing.
* Gossip tick can be manually sent.
*/
- public Gossiper(Boolean autoStartGossipTicks){
+ @VisibleForTesting
+ public Gossiper(Boolean autoStartGossipTicks, RemoteRpcProviderConfig config) {
+ this(config);
this.autoStartGossipTicks = autoStartGossipTicks;
}
@Override
- public void preStart(){
-
- cluster.subscribe(getSelf(),
- ClusterEvent.initialStateAsEvents(),
- ClusterEvent.MemberEvent.class,
- ClusterEvent.UnreachableMember.class);
+ public void preStart() {
+ ActorRefProvider provider = getContext().provider();
+ selfAddress = provider.getDefaultAddress();
+
+ if ( provider instanceof ClusterActorRefProvider ) {
+ cluster = Cluster.get(getContext().system());
+ cluster.subscribe(getSelf(),
+ ClusterEvent.initialStateAsEvents(),
+ ClusterEvent.MemberEvent.class,
+ ClusterEvent.UnreachableMember.class);
+ }
if (autoStartGossipTicks) {
gossipTask = getContext().system().scheduler().schedule(
new FiniteDuration(1, TimeUnit.SECONDS), //initial delay
- new FiniteDuration(500, TimeUnit.MILLISECONDS), //interval
+ config.getGossipTickInterval(), //interval
getSelf(), //target
new Messages.GossiperMessages.GossipTick(), //message
getContext().dispatcher(), //execution context
@@ -108,44 +124,42 @@ public class Gossiper extends UntypedActor {
}
@Override
- public void postStop(){
- if (cluster != null)
+ public void postStop() {
+ if (cluster != null) {
cluster.unsubscribe(getSelf());
- if (gossipTask != null)
+ }
+ if (gossipTask != null) {
gossipTask.cancel();
+ }
}
+ @SuppressWarnings({ "rawtypes", "unchecked" })
@Override
- public void onReceive(Object message) throws Exception {
-
- log.debug("Received message: node[{}], message[{}]", selfAddress, message);
-
+ protected void handleReceive(Object message) throws Exception {
//Usually sent by self via gossip task defined above. But its not enforced.
//These ticks can be sent by another actor as well which is esp. useful while testing
- if (message instanceof GossipTick)
+ if (message instanceof GossipTick) {
receiveGossipTick();
-
- //Message from remote gossiper with its bucket versions
- else if (message instanceof GossipStatus)
+ } else if (message instanceof GossipStatus) {
+ // Message from remote gossiper with its bucket versions
receiveGossipStatus((GossipStatus) message);
-
- //Message from remote gossiper with buckets. This is usually in response to GossipStatus message
- //The contained buckets are newer as determined by the remote gossiper by comparing the GossipStatus
- //message with its local versions
- else if (message instanceof GossipEnvelope)
+ } else if (message instanceof GossipEnvelope) {
+ // Message from remote gossiper with buckets. This is usually in response to GossipStatus
+ // message. The contained buckets are newer as determined by the remote gossiper by
+ // comparing the GossipStatus message with its local versions.
receiveGossip((GossipEnvelope) message);
-
- else if (message instanceof ClusterEvent.MemberUp) {
+ } else if (message instanceof ClusterEvent.MemberUp) {
receiveMemberUp(((ClusterEvent.MemberUp) message).member());
} else if (message instanceof ClusterEvent.MemberRemoved) {
receiveMemberRemoveOrUnreachable(((ClusterEvent.MemberRemoved) message).member());
- } else if ( message instanceof ClusterEvent.UnreachableMember){
+ } else if ( message instanceof ClusterEvent.UnreachableMember) {
receiveMemberRemoveOrUnreachable(((ClusterEvent.UnreachableMember) message).member());
- } else
+ } else {
unhandled(message);
+ }
}
/**
@@ -155,7 +169,7 @@ public class Gossiper extends UntypedActor {
*/
void receiveMemberRemoveOrUnreachable(Member member) {
//if its self, then stop itself
- if (selfAddress.equals(member.address())){
+ if (selfAddress.equals(member.address())) {
getContext().stop(getSelf());
return;
}
@@ -165,46 +179,52 @@ public class Gossiper extends UntypedActor {
}
/**
- * Add member to the local copy of member list if it doesnt already
- * @param member
+ * Add member to the local copy of member list if it doesn't already.
+ *
+ * @param member the member to add
*/
void receiveMemberUp(Member member) {
- if (selfAddress.equals(member.address()))
+ if (selfAddress.equals(member.address())) {
return; //ignore up notification for self
+ }
- if (!clusterMembers.contains(member.address()))
+ if (!clusterMembers.contains(member.address())) {
clusterMembers.add(member.address());
+ }
log.debug("Added member [{}], Active member list [{}]", member.address(), clusterMembers);
}
/**
- * Sends Gossip status to other members in the cluster.
- * 1. If there are no member, ignore the tick.
- * 2. If there's only 1 member, send gossip status (bucket versions) to it.
+ * Sends Gossip status to other members in the cluster.
+ *
+ * 1. If there are no member, ignore the tick.
+ * 2. If there's only 1 member, send gossip status (bucket versions) to it.
* 3. If there are more than one member, randomly pick one and send gossip status (bucket versions) to it.
*/
- void receiveGossipTick(){
- if (clusterMembers.size() == 0) return; //no members to send gossip status to
+ void receiveGossipTick() {
+ if (clusterMembers.size() == 0) {
+ return; //no members to send gossip status to
+ }
- Address remoteMemberToGossipTo = null;
+ Address remoteMemberToGossipTo;
- if (clusterMembers.size() == 1)
+ if (clusterMembers.size() == 1) {
remoteMemberToGossipTo = clusterMembers.get(0);
- else {
+ } else {
Integer randomIndex = ThreadLocalRandom.current().nextInt(0, clusterMembers.size());
remoteMemberToGossipTo = clusterMembers.get(randomIndex);
}
- log.debug("Gossiping to [{}]", remoteMemberToGossipTo);
+ log.trace("Gossiping to [{}]", remoteMemberToGossipTo);
getLocalStatusAndSendTo(remoteMemberToGossipTo);
}
/**
* Process gossip status received from a remote gossiper. Remote versions are compared with
- * the local copy.
- *
+ * the local copy.
+ *
* For each bucket
*
*
If local copy is newer, the newer buckets are sent in GossipEnvelope to remote
@@ -214,13 +234,16 @@ public class Gossiper extends UntypedActor {
*
* @param status bucket versions from a remote member
*/
- void receiveGossipStatus(GossipStatus status){
+ void receiveGossipStatus(GossipStatus status) {
//Don't accept messages from non-members
- if (!clusterMembers.contains(status.from()))
+ if (!clusterMembers.contains(status.from())) {
return;
+ }
final ActorRef sender = getSender();
- Future