X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-remoterpc-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fremote%2Frpc%2Fregistry%2Fgossip%2FGossiper.java;h=f597c316e5cdfda042fef8f6ea12741a905b20cf;hb=refs%2Fchanges%2F62%2F27562%2F9;hp=0b64136c49f9bda7bb331c9f5111d2146b91da9d;hpb=08d30cebbbec8da5596840c6ef830687a76d9d33;p=controller.git diff --git a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/Gossiper.java b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/Gossiper.java index 0b64136c49..f597c316e5 100644 --- a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/Gossiper.java +++ b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/Gossiper.java @@ -8,20 +8,17 @@ package org.opendaylight.controller.remote.rpc.registry.gossip; import akka.actor.ActorRef; +import akka.actor.ActorRefProvider; import akka.actor.ActorSelection; import akka.actor.Address; import akka.actor.Cancellable; -import akka.actor.UntypedActor; import akka.cluster.Cluster; +import akka.cluster.ClusterActorRefProvider; import akka.cluster.ClusterEvent; import akka.cluster.Member; import akka.dispatch.Mapper; -import akka.event.Logging; -import akka.event.LoggingAdapter; import akka.pattern.Patterns; -import scala.concurrent.Future; -import scala.concurrent.duration.FiniteDuration; - +import com.google.common.base.Preconditions; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -29,40 +26,48 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; - -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembers; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembersReply; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateRemoteBuckets; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipEnvelope; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipStatus; -import static org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipTick; +import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActorWithMetering; +import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembers; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembersReply; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateRemoteBuckets; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipEnvelope; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipStatus; +import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.GossiperMessages.GossipTick; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import scala.concurrent.Future; +import scala.concurrent.duration.FiniteDuration; /** * Gossiper that syncs bucket store across nodes in the cluster. - *
- * 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. + *
+ * 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 UntypedActor { +public class Gossiper extends AbstractUntypedActorWithMetering { - final LoggingAdapter log = Logging.getLogger(getContext().system(), this); + private final Logger log = LoggerFactory.getLogger(getClass()); - 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 @@ -73,28 +78,39 @@ 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 - * @param autoStartGossipTicks used for turning off gossip ticks during testing. Gossip tick can be manually sent. + * @param autoStartGossipTicks used for turning off gossip ticks during testing. + * Gossip tick can be manually sent. */ - public Gossiper(Boolean autoStartGossipTicks){ + 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); + 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 @@ -105,33 +121,29 @@ public class Gossiper extends UntypedActor { @Override public void postStop(){ - if (cluster != null) + if (cluster != null) { cluster.unsubscribe(getSelf()); - if (gossipTask != null) + } + if (gossipTask != null) { gossipTask.cancel(); + } } @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) { @@ -140,8 +152,9 @@ public class Gossiper extends UntypedActor { } else if ( message instanceof ClusterEvent.UnreachableMember){ receiveMemberRemoveOrUnreachable(((ClusterEvent.UnreachableMember) message).member()); - } else + } else { unhandled(message); + } } /** @@ -157,7 +170,9 @@ public class Gossiper extends UntypedActor { } clusterMembers.remove(member.address()); - log.debug("Removed member [{}], Active member list [{}]", member.address(), clusterMembers); + if(log.isDebugEnabled()) { + log.debug("Removed member [{}], Active member list [{}]", member.address(), clusterMembers); + } } /** @@ -166,13 +181,16 @@ public class Gossiper extends UntypedActor { */ 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); + } + if(log.isDebugEnabled()) { + log.debug("Added member [{}], Active member list [{}]", member.address(), clusterMembers); + } } /** @@ -182,18 +200,21 @@ public class Gossiper extends UntypedActor { * 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 + 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); + if(log.isTraceEnabled()) { + log.trace("Gossiping to [{}]", remoteMemberToGossipTo); + } getLocalStatusAndSendTo(remoteMemberToGossipTo); } @@ -211,13 +232,14 @@ public class Gossiper extends UntypedActor { * @param status bucket versions from a remote member */ void receiveGossipStatus(GossipStatus status){ - //Dont want to accept messages from non-members - if (!clusterMembers.contains(status.from())) + //Don't accept messages from non-members + if (!clusterMembers.contains(status.from())) { return; + } final ActorRef sender = getSender(); - - Future