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%2FBucketStore.java;h=b4af0adfe5f4160868752aab7e0210a8607c3910;hb=e9fce74e37472296faa2faf1acbd110b74196032;hp=6ffe147e71384b66ece31c4e701294c8630456c8;hpb=b52893a2ff963a4a02a96190d78f62e9d261ffce;p=controller.git
diff --git a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.java b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.java
index 6ffe147e71..b4af0adfe5 100644
--- a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.java
+++ b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.java
@@ -11,15 +11,16 @@ package org.opendaylight.controller.remote.rpc.registry.gossip;
import akka.actor.ActorRef;
import akka.actor.ActorRefProvider;
import akka.actor.Address;
-import akka.actor.Props;
+import akka.actor.Terminated;
import akka.cluster.ClusterActorRefProvider;
-import akka.event.Logging;
-import akka.event.LoggingAdapter;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.SetMultimap;
import java.util.HashMap;
import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
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.GetAllBuckets;
@@ -28,9 +29,7 @@ import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketSto
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.GetLocalBucket;
-import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetLocalBucketReply;
-import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateBucket;
+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.utils.ConditionalProbe;
@@ -38,126 +37,113 @@ import org.opendaylight.controller.utils.ConditionalProbe;
* A store that syncs its data across nodes in the cluster.
* It maintains a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Bucket} per node. Buckets are versioned.
* A node can write ONLY to its bucket. This way, write conflicts are avoided.
+ *
*
- * Buckets are sync'ed across nodes using Gossip protocol (http://en.wikipedia.org/wiki/Gossip_protocol)
+ * Buckets are sync'ed across nodes using Gossip protocol (http://en.wikipedia.org/wiki/Gossip_protocol).
* This store uses a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}.
*
*/
-public class BucketStore extends AbstractUntypedActorWithMetering {
-
- final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
+public class BucketStore> extends AbstractUntypedActorWithMetering {
+ /**
+ * Bucket owned by the node.
+ */
+ private final BucketImpl localBucket;
/**
- * Bucket owned by the node
+ * Buckets owned by other known nodes in the cluster.
*/
- private BucketImpl localBucket = new BucketImpl();
+ private final Map> remoteBuckets = new HashMap<>();
/**
- * Buckets ownded by other known nodes in the cluster
+ * Bucket version for every known node in the cluster including this node.
*/
- private ConcurrentMap remoteBuckets = new ConcurrentHashMap<>();
+ private final Map versions = new HashMap<>();
/**
- * Bucket version for every known node in the cluster including this node
+ * {@link ActorRef}s being watched for liveness due to being referenced in bucket data. Each actor is monitored
+ * once, possibly being tied to multiple addresses (and by extension, buckets).
*/
- private ConcurrentMap versions = new ConcurrentHashMap<>();
+ private final SetMultimap watchedActors = HashMultimap.create(1, 1);
/**
- * Cluster address for this node
+ * Cluster address for this node.
*/
private Address selfAddress;
+ // FIXME: should be part of test-specific subclass
private ConditionalProbe probe;
private final RemoteRpcProviderConfig config;
- public BucketStore(){
- config = new RemoteRpcProviderConfig(getContext().system().settings().config());
+ public BucketStore(final RemoteRpcProviderConfig config, final T initialData) {
+ this.config = Preconditions.checkNotNull(config);
+ this.localBucket = new BucketImpl<>(initialData);
}
@Override
- public void preStart(){
+ public void preStart() {
ActorRefProvider provider = getContext().provider();
selfAddress = provider.getDefaultAddress();
- if ( provider instanceof ClusterActorRefProvider) {
- getContext().actorOf(Props.create(Gossiper.class).withMailbox(config.getMailBoxName()), "gossiper");
+ if (provider instanceof ClusterActorRefProvider) {
+ getContext().actorOf(Gossiper.props(config).withMailbox(config.getMailBoxName()), "gossiper");
}
}
-
+ @SuppressWarnings("unchecked")
@Override
- protected void handleReceive(Object message) throws Exception {
+ protected void handleReceive(final Object message) throws Exception {
if (probe != null) {
probe.tell(message, getSelf());
}
- if (message instanceof ConditionalProbe) {
+ if (message instanceof GetBucketsByMembers) {
+ receiveGetBucketsByMembers(((GetBucketsByMembers) message).getMembers());
+ } else if (message instanceof GetBucketVersions) {
+ receiveGetBucketVersions();
+ } else if (message instanceof UpdateRemoteBuckets) {
+ receiveUpdateRemoteBuckets(((UpdateRemoteBuckets) message).getBuckets());
+ } else if (message instanceof RemoveRemoteBucket) {
+ removeBucket(((RemoveRemoteBucket) message).getAddress());
+ } else if (message instanceof Terminated) {
+ actorTerminated((Terminated) message);
+ } else if (message instanceof GetAllBuckets) {
+ // GetAllBuckets is used only for unit tests.
+ receiveGetAllBuckets();
+ } else if (message instanceof ConditionalProbe) {
// The ConditionalProbe is only used for unit tests.
- log.info("Received probe {} {}", getSelf(), message);
+ LOG.info("Received probe {} {}", getSelf(), message);
probe = (ConditionalProbe) message;
// Send back any message to tell the caller we got the probe.
getSender().tell("Got it", getSelf());
- } else if (message instanceof UpdateBucket) {
- receiveUpdateBucket(((UpdateBucket) message).getBucket());
- } else if (message instanceof GetAllBuckets) {
- receiveGetAllBucket();
- } else if (message instanceof GetLocalBucket) {
- receiveGetLocalBucket();
- } else if (message instanceof GetBucketsByMembers) {
- receiveGetBucketsByMembers(
- ((GetBucketsByMembers) message).getMembers());
- } else if (message instanceof GetBucketVersions) {
- receiveGetBucketVersions();
- } else if (message instanceof UpdateRemoteBuckets) {
- receiveUpdateRemoteBuckets(
- ((UpdateRemoteBuckets) message).getBuckets());
} else {
- if(log.isDebugEnabled()) {
- log.debug("Unhandled message [{}]", message);
- }
+ LOG.debug("Unhandled message [{}]", message);
unhandled(message);
}
}
- /**
- * Returns a copy of bucket owned by this node
- */
- private void receiveGetLocalBucket() {
- final ActorRef sender = getSender();
- GetLocalBucketReply reply = new GetLocalBucketReply(localBucket);
- sender.tell(reply, getSelf());
- }
-
- /**
- * Updates the bucket owned by this node
- *
- * @param updatedBucket
- */
- void receiveUpdateBucket(Bucket updatedBucket){
-
- localBucket = (BucketImpl) updatedBucket;
- versions.put(selfAddress, localBucket.getVersion());
+ protected RemoteRpcProviderConfig getConfig() {
+ return config;
}
/**
- * Returns all the buckets the this node knows about, self owned + remote
+ * Returns all the buckets the this node knows about, self owned + remote.
*/
- void receiveGetAllBucket(){
+ void receiveGetAllBuckets() {
final ActorRef sender = getSender();
- sender.tell(new GetAllBucketsReply(getAllBuckets()), getSelf());
+ sender.tell(new GetAllBucketsReply<>(getAllBuckets()), getSelf());
}
/**
- * Helper to collect all known buckets
+ * Helper to collect all known buckets.
*
* @return self owned + remote buckets
*/
- Map getAllBuckets(){
- Map all = new HashMap<>(remoteBuckets.size() + 1);
+ Map> getAllBuckets() {
+ Map> all = new HashMap<>(remoteBuckets.size() + 1);
//first add the local bucket
- all.put(selfAddress, localBucket);
+ all.put(selfAddress, new BucketImpl<>(localBucket));
//then get all remote buckets
all.putAll(remoteBuckets);
@@ -166,32 +152,32 @@ public class BucketStore extends AbstractUntypedActorWithMetering {
}
/**
- * Returns buckets for requested members that this node knows about
+ * Returns buckets for requested members that this node knows about.
*
* @param members requested members
*/
- void receiveGetBucketsByMembers(Set members){
+ void receiveGetBucketsByMembers(final Set members) {
final ActorRef sender = getSender();
- Map buckets = getBucketsByMembers(members);
- sender.tell(new GetBucketsByMembersReply(buckets), getSelf());
+ Map> buckets = getBucketsByMembers(members);
+ sender.tell(new GetBucketsByMembersReply<>(buckets), getSelf());
}
/**
- * Helper to collect buckets for requested memebers
+ * Helper to collect buckets for requested members.
*
* @param members requested members
- * @return buckets for requested memebers
+ * @return buckets for requested members
*/
- Map getBucketsByMembers(Set members) {
- Map buckets = new HashMap<>();
+ Map> getBucketsByMembers(final Set members) {
+ Map> buckets = new HashMap<>();
//first add the local bucket if asked
if (members.contains(selfAddress)) {
- buckets.put(selfAddress, localBucket);
+ buckets.put(selfAddress, new BucketImpl<>(localBucket));
}
//then get buckets for requested remote nodes
- for (Address address : members){
+ for (Address address : members) {
if (remoteBuckets.containsKey(address)) {
buckets.put(address, remoteBuckets.get(address));
}
@@ -201,88 +187,140 @@ public class BucketStore extends AbstractUntypedActorWithMetering {
}
/**
- * Returns versions for all buckets known
+ * Returns versions for all buckets known.
*/
- void receiveGetBucketVersions(){
+ void receiveGetBucketVersions() {
final ActorRef sender = getSender();
GetBucketVersionsReply reply = new GetBucketVersionsReply(versions);
sender.tell(reply, getSelf());
}
/**
- * Update local copy of remote buckets where local copy's version is older
+ * Update local copy of remote buckets where local copy's version is older.
*
* @param receivedBuckets buckets sent by remote
* {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}
*/
- void receiveUpdateRemoteBuckets(Map receivedBuckets){
-
- if (receivedBuckets == null || receivedBuckets.isEmpty())
- {
- return; //nothing to do
+ void receiveUpdateRemoteBuckets(final Map> receivedBuckets) {
+ LOG.debug("{}: receiveUpdateRemoteBuckets: {}", selfAddress, receivedBuckets);
+ if (receivedBuckets == null || receivedBuckets.isEmpty()) {
+ //nothing to do
+ return;
}
- //Remote cant update self's bucket
- receivedBuckets.remove(selfAddress);
+ final Map> newBuckets = new HashMap<>(receivedBuckets.size());
+ for (Entry> entry : receivedBuckets.entrySet()) {
+ final Address addr = entry.getKey();
- for (Map.Entry entry : receivedBuckets.entrySet()){
-
- Long localVersion = versions.get(entry.getKey());
- if (localVersion == null) {
- localVersion = -1L;
+ if (selfAddress.equals(addr)) {
+ // Remote cannot update our bucket
+ continue;
}
- Bucket receivedBucket = entry.getValue();
-
+ final Bucket receivedBucket = entry.getValue();
if (receivedBucket == null) {
+ LOG.debug("Ignoring null bucket from {}", addr);
continue;
}
- Long remoteVersion = receivedBucket.getVersion();
- if (remoteVersion == null) {
- remoteVersion = -1L;
+ // update only if remote version is newer
+ final long remoteVersion = receivedBucket.getVersion();
+ final Long localVersion = versions.get(addr);
+ if (localVersion != null && remoteVersion <= localVersion.longValue()) {
+ LOG.debug("Ignoring down-versioned bucket from {} ({} local {} remote)", addr, localVersion,
+ remoteVersion);
+ continue;
}
-
- //update only if remote version is newer
- if ( remoteVersion.longValue() > localVersion.longValue() ) {
- remoteBuckets.put(entry.getKey(), receivedBucket);
- versions.put(entry.getKey(), remoteVersion);
+ newBuckets.put(addr, receivedBucket);
+ versions.put(addr, remoteVersion);
+ final Bucket prevBucket = remoteBuckets.put(addr, receivedBucket);
+
+ // Deal with DeathWatch subscriptions
+ final Optional prevRef = prevBucket != null ? prevBucket.getWatchActor() : Optional.empty();
+ final Optional curRef = receivedBucket.getWatchActor();
+ if (!curRef.equals(prevRef)) {
+ prevRef.ifPresent(ref -> removeWatch(addr, ref));
+ curRef.ifPresent(ref -> addWatch(addr, ref));
}
+
+ LOG.debug("Updating bucket from {} to version {}", entry.getKey(), remoteVersion);
+ }
+
+ LOG.debug("State after update - Local Bucket [{}], Remote Buckets [{}]", localBucket, remoteBuckets);
+
+ onBucketsUpdated(newBuckets);
+ }
+
+ private void addWatch(final Address addr, final ActorRef ref) {
+ if (!watchedActors.containsKey(ref)) {
+ getContext().watch(ref);
+ LOG.debug("Watching {}", ref);
}
- if(log.isDebugEnabled()) {
- log.debug("State after update - Local Bucket [{}], Remote Buckets [{}]", localBucket, remoteBuckets);
+ watchedActors.put(ref, addr);
+ }
+
+ private void removeWatch(final Address addr, final ActorRef ref) {
+ watchedActors.remove(ref, addr);
+ if (!watchedActors.containsKey(ref)) {
+ getContext().unwatch(ref);
+ LOG.debug("No longer watching {}", ref);
}
}
- ///
- ///Getter Setters
- ///
+ private void removeBucket(final Address addr) {
+ final Bucket bucket = remoteBuckets.remove(addr);
+ if (bucket != null) {
+ bucket.getWatchActor().ifPresent(ref -> removeWatch(addr, ref));
+ onBucketRemoved(addr, bucket);
+ }
+ }
- BucketImpl getLocalBucket() {
- return localBucket;
+ private void actorTerminated(final Terminated message) {
+ LOG.info("Actor termination {} received", message);
+
+ for (Address addr : watchedActors.removeAll(message.getActor())) {
+ versions.remove(addr);
+ final Bucket bucket = remoteBuckets.remove(addr);
+ if (bucket != null) {
+ LOG.debug("Source actor dead, removing bucket {} from ", bucket, addr);
+ onBucketRemoved(addr, bucket);
+ }
+ }
}
- void setLocalBucket(BucketImpl localBucket) {
- this.localBucket = localBucket;
+ /**
+ * Callback to subclasses invoked when a bucket is removed.
+ *
+ * @param address Remote address
+ * @param bucket Bucket removed
+ */
+ protected void onBucketRemoved(final Address address, final Bucket bucket) {
+ // Default noop
}
- ConcurrentMap getRemoteBuckets() {
- return remoteBuckets;
+ /**
+ * Callback to subclasses invoked when the set of remote buckets is updated.
+ *
+ * @param newBuckets Map of address to new bucket. Never null, but can be empty.
+ */
+ protected void onBucketsUpdated(final Map> newBuckets) {
+ // Default noop
}
- void setRemoteBuckets(ConcurrentMap remoteBuckets) {
- this.remoteBuckets = remoteBuckets;
+ public BucketImpl getLocalBucket() {
+ return localBucket;
}
- ConcurrentMap getVersions() {
- return versions;
+ protected void updateLocalBucket(final T data) {
+ localBucket.setData(data);
+ versions.put(selfAddress, localBucket.getVersion());
}
- void setVersions(ConcurrentMap versions) {
- this.versions = versions;
+ public Map> getRemoteBuckets() {
+ return remoteBuckets;
}
- Address getSelfAddress() {
- return selfAddress;
+ public Map getVersions() {
+ return versions;
}
}