X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-remoterpc-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fremote%2Frpc%2Fregistry%2Fgossip%2FGossiper.java;h=2c47c4e2a9242a7d9c56a8dc30ade963b83450b7;hb=e9fce74e37472296faa2faf1acbd110b74196032;hp=1bbcc69f5ed4d5fa6d7d8ea773823c97c9bb6e05;hpb=7225f60c394a26143f8421b0f99f2585699fa306;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 1bbcc69f5e..2c47c4e2a9 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 @@ -12,98 +12,107 @@ import akka.actor.ActorRefProvider; import akka.actor.ActorSelection; import akka.actor.Address; import akka.actor.Cancellable; +import akka.actor.Props; 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 org.opendaylight.controller.cluster.common.actor.AbstractUntypedActorWithMetering; -import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig; -import scala.concurrent.Future; -import scala.concurrent.duration.FiniteDuration; - +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.base.Verify; import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; import java.util.List; 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.RemoveRemoteBucket; +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 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. - * */ - public class Gossiper extends AbstractUntypedActorWithMetering { + private final boolean autoStartGossipTicks; + private final RemoteRpcProviderConfig config; - final LoggingAdapter log = Logging.getLogger(getContext().system(), this); + /** + * All known cluster members. + */ + private final List
clusterMembers = new ArrayList<>(); - private Cluster cluster; + /** + * Cached ActorSelections for remote peers. + */ + private final Map peers = new HashMap<>(); /** * ActorSystem's address for the current cluster node. */ private Address selfAddress; - /** - * All known cluster members - */ - private List clusterMembers = new ArrayList<>(); + private Cluster cluster; private Cancellable gossipTask; - private Boolean autoStartGossipTicks = true; + Gossiper(final RemoteRpcProviderConfig config, final Boolean autoStartGossipTicks) { + this.config = Preconditions.checkNotNull(config); + this.autoStartGossipTicks = autoStartGossipTicks.booleanValue(); + } - private RemoteRpcProviderConfig config; + Gossiper(final RemoteRpcProviderConfig config) { + this(config, Boolean.TRUE); + } - public Gossiper(){ - config = new RemoteRpcProviderConfig(getContext().system().settings().config()); + public static Props props(final RemoteRpcProviderConfig config) { + return Props.create(Gossiper.class, config); } - /** - * Helpful for testing - * @param autoStartGossipTicks used for turning off gossip ticks during testing. - * Gossip tick can be manually sent. - */ - public Gossiper(Boolean autoStartGossipTicks){ - this.autoStartGossipTicks = autoStartGossipTicks; + static Props testProps(final RemoteRpcProviderConfig config) { + return Props.create(Gossiper.class, config, Boolean.FALSE); } @Override - public void preStart(){ + public void preStart() { ActorRefProvider provider = getContext().provider(); selfAddress = provider.getDefaultAddress(); - if ( provider instanceof ClusterActorRefProvider ) { + if (provider instanceof ClusterActorRefProvider ) { cluster = Cluster.get(getContext().system()); cluster.subscribe(getSelf(), ClusterEvent.initialStateAsEvents(), ClusterEvent.MemberEvent.class, + ClusterEvent.ReachableMember.class, ClusterEvent.UnreachableMember.class); } @@ -111,50 +120,54 @@ public class Gossiper extends AbstractUntypedActorWithMetering { gossipTask = getContext().system().scheduler().schedule( new FiniteDuration(1, TimeUnit.SECONDS), //initial delay config.getGossipTickInterval(), //interval - getSelf(), //target - new Messages.GossiperMessages.GossipTick(), //message - getContext().dispatcher(), //execution context - getSelf() //sender + getSelf(), //target + new Messages.GossiperMessages.GossipTick(), //message + getContext().dispatcher(), //execution context + getSelf() //sender ); } } @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 - protected void handleReceive(Object message) throws Exception { + protected void handleReceive(final 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) { + receiveMemberUpOrReachable(((ClusterEvent.MemberUp) message).member()); - else if (message instanceof ClusterEvent.MemberUp) { - receiveMemberUp(((ClusterEvent.MemberUp) message).member()); + } else if (message instanceof ClusterEvent.ReachableMember) { + receiveMemberUpOrReachable(((ClusterEvent.ReachableMember) 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); + } } /** @@ -162,62 +175,77 @@ public class Gossiper extends AbstractUntypedActorWithMetering { * * @param member who went down */ - void receiveMemberRemoveOrUnreachable(Member member) { + private void receiveMemberRemoveOrUnreachable(final Member member) { //if its self, then stop itself - if (selfAddress.equals(member.address())){ + if (selfAddress.equals(member.address())) { getContext().stop(getSelf()); return; } - clusterMembers.remove(member.address()); - if(log.isDebugEnabled()) { - log.debug("Removed member [{}], Active member list [{}]", member.address(), clusterMembers); + removePeer(member.address()); + LOG.debug("Removed member [{}], Active member list [{}]", member.address(), clusterMembers); + } + + private void addPeer(final Address address) { + if (!clusterMembers.contains(address)) { + clusterMembers.add(address); } + peers.computeIfAbsent(address, input -> getContext().system() + .actorSelection(input.toString() + getSelf().path().toStringWithoutAddress())); + } + + private void removePeer(final Address address) { + clusterMembers.remove(address); + peers.remove(address); + getContext().parent().tell(new RemoveRemoteBucket(address), ActorRef.noSender()); } /** - * 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())) - return; //ignore up notification for self - - if (!clusterMembers.contains(member.address())) - clusterMembers.add(member.address()); - if(log.isDebugEnabled()) { - log.debug("Added member [{}], Active member list [{}]", member.address(), clusterMembers); + private void receiveMemberUpOrReachable(final Member member) { + //ignore up notification for self + if (selfAddress.equals(member.address())) { + return; } + + addPeer(member.address()); + LOG.debug("Added member [{}], Active member list [{}]", member.address(), clusterMembers); } /** - * Sends Gossip status to other members in the cluster.- * + * the local copy. + *
* For each bucket *