opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStoreActor.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.remote.rpc.registry.gossip;
  10
  11 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreAccess.Singletons.GET_ALL_BUCKETS;
  12 import static org.opendaylight.controller.remote.rpc.registry.gossip.BucketStoreAccess.Singletons.GET_BUCKET_VERSIONS;
  13
  14 import akka.actor.ActorRef;
  15 import akka.actor.ActorRefProvider;
  16 import akka.actor.Address;
  17 import akka.actor.PoisonPill;
  18 import akka.actor.Terminated;
  19 import akka.cluster.ClusterActorRefProvider;
  20 import akka.persistence.DeleteSnapshotsFailure;
  21 import akka.persistence.DeleteSnapshotsSuccess;
  22 import akka.persistence.RecoveryCompleted;
  23 import akka.persistence.SaveSnapshotFailure;
  24 import akka.persistence.SaveSnapshotSuccess;
  25 import akka.persistence.SnapshotOffer;
  26 import akka.persistence.SnapshotSelectionCriteria;
  27 import com.google.common.annotations.VisibleForTesting;
  28 import com.google.common.base.Preconditions;
  29 import com.google.common.base.Verify;
  30 import com.google.common.collect.HashMultimap;
  31 import com.google.common.collect.ImmutableMap;
  32 import com.google.common.collect.SetMultimap;
  33 import java.util.Collection;
  34 import java.util.HashMap;
  35 import java.util.Map;
  36 import java.util.Map.Entry;
  37 import java.util.Optional;
  38 import java.util.function.Consumer;
  39 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
  40 import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
  41
  42 /**
  43  * A store that syncs its data across nodes in the cluster.
  44  * It maintains a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Bucket} per node. Buckets are versioned.
  45  * A node can write ONLY to its bucket. This way, write conflicts are avoided.
  46  *
  47  * <p>
  48  * Buckets are sync'ed across nodes using Gossip protocol (http://en.wikipedia.org/wiki/Gossip_protocol).
  49  * This store uses a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}.
  50  */
  51 public abstract class BucketStoreActor<T extends BucketData<T>> extends
  52         AbstractUntypedPersistentActorWithMetering {
  53     // Internal marker interface for messages which are just bridges to execute a method
  54     @FunctionalInterface
  55     private interface ExecuteInActor extends Consumer<BucketStoreActor<?>> {
  56
  57     }
  58
  59     /**
  60      * Buckets owned by other known nodes in the cluster.
  61      */
  62     private final Map<Address, Bucket<T>> remoteBuckets = new HashMap<>();
  63
  64     /**
  65      * Bucket version for every known node in the cluster including this node.
  66      */
  67     private final Map<Address, Long> versions = new HashMap<>();
  68
  69     /**
  70      * {@link ActorRef}s being watched for liveness due to being referenced in bucket data. Each actor is monitored
  71      * once, possibly being tied to multiple addresses (and by extension, buckets).
  72      */
  73     private final SetMultimap<ActorRef, Address> watchedActors = HashMultimap.create(1, 1);
  74
  75     private final RemoteRpcProviderConfig config;
  76     private final String persistenceId;
  77
  78     /**
  79      * Cluster address for this node.
  80      */
  81     private Address selfAddress;
  82
  83     /**
  84      * Bucket owned by the node. Initialized during recovery (due to incarnation number).
  85      */
  86     private LocalBucket<T> localBucket;
  87     private T initialData;
  88     private Integer incarnation;
  89     private boolean persisting;
  90
  91     protected BucketStoreActor(final RemoteRpcProviderConfig config, final String persistenceId, final T initialData) {
  92         this.config = Preconditions.checkNotNull(config);
  93         this.initialData = Preconditions.checkNotNull(initialData);
  94         this.persistenceId = Preconditions.checkNotNull(persistenceId);
  95     }
  96
  97     static ExecuteInActor getBucketsByMembersMessage(final Collection<Address> members) {
  98         return actor -> actor.getBucketsByMembers(members);
  99     }
 100
 101     static ExecuteInActor removeBucketMessage(final Address addr) {
 102         return actor -> actor.removeBucket(addr);
 103     }
 104
 105     static ExecuteInActor updateRemoteBucketsMessage(final Map<Address, Bucket<?>> buckets) {
 106         return actor -> actor.updateRemoteBuckets(buckets);
 107     }
 108
 109     public final T getLocalData() {
 110         return getLocalBucket().getData();
 111     }
 112
 113     public final Map<Address, Bucket<T>> getRemoteBuckets() {
 114         return remoteBuckets;
 115     }
 116
 117     public final Map<Address, Long> getVersions() {
 118         return versions;
 119     }
 120
 121     @Override
 122     public final String persistenceId() {
 123         return persistenceId;
 124     }
 125
 126     @Override
 127     public void preStart() {
 128         ActorRefProvider provider = getContext().provider();
 129         selfAddress = provider.getDefaultAddress();
 130
 131         if (provider instanceof ClusterActorRefProvider) {
 132             getContext().actorOf(Gossiper.props(config).withMailbox(config.getMailBoxName()), "gossiper");
 133         }
 134     }
 135
 136     @Override
 137     protected void handleCommand(final Object message) throws Exception {
 138         if (GET_ALL_BUCKETS == message) {
 139             // GetAllBuckets is used only in testing
 140             getSender().tell(getAllBuckets(), self());
 141             return;
 142         }
 143
 144         if (persisting) {
 145             handleSnapshotMessage(message);
 146             return;
 147         }
 148
 149         if (message instanceof ExecuteInActor) {
 150             ((ExecuteInActor) message).accept(this);
 151         } else if (GET_BUCKET_VERSIONS == message) {
 152             // FIXME: do we need to send ourselves?
 153             getSender().tell(ImmutableMap.copyOf(versions), getSelf());
 154         } else if (message instanceof Terminated) {
 155             actorTerminated((Terminated) message);
 156         } else if (message instanceof DeleteSnapshotsSuccess) {
 157             LOG.debug("{}: got command: {}", persistenceId(), message);
 158         } else if (message instanceof DeleteSnapshotsFailure) {
 159             LOG.warn("{}: failed to delete prior snapshots", persistenceId(),
 160                 ((DeleteSnapshotsFailure) message).cause());
 161         } else {
 162             LOG.debug("Unhandled message [{}]", message);
 163             unhandled(message);
 164         }
 165     }
 166
 167     private void handleSnapshotMessage(final Object message) {
 168         if (message instanceof SaveSnapshotFailure) {
 169             LOG.error("{}: failed to persist state", persistenceId(), ((SaveSnapshotFailure) message).cause());
 170             persisting = false;
 171             self().tell(PoisonPill.getInstance(), ActorRef.noSender());
 172         } else if (message instanceof SaveSnapshotSuccess) {
 173             LOG.debug("{}: got command: {}", persistenceId(), message);
 174             SaveSnapshotSuccess saved = (SaveSnapshotSuccess)message;
 175             deleteSnapshots(new SnapshotSelectionCriteria(saved.metadata().sequenceNr(),
 176                     saved.metadata().timestamp() - 1, 0L, 0L));
 177             persisting = false;
 178             unstash();
 179         } else {
 180             LOG.debug("{}: stashing command {}", persistenceId(), message);
 181             stash();
 182         }
 183     }
 184
 185     @Override
 186     protected final void handleRecover(final Object message) throws Exception {
 187         if (message instanceof RecoveryCompleted) {
 188             if (incarnation != null) {
 189                 incarnation = incarnation + 1;
 190             } else {
 191                 incarnation = 0;
 192             }
 193
 194             this.localBucket = new LocalBucket<>(incarnation.intValue(), initialData);
 195             initialData = null;
 196             LOG.debug("{}: persisting new incarnation {}", persistenceId(), incarnation);
 197             persisting = true;
 198             saveSnapshot(incarnation);
 199         } else if (message instanceof SnapshotOffer) {
 200             incarnation = (Integer) ((SnapshotOffer)message).snapshot();
 201             LOG.debug("{}: recovered incarnation {}", persistenceId(), incarnation);
 202         } else {
 203             LOG.warn("{}: ignoring recovery message {}", persistenceId(), message);
 204         }
 205     }
 206
 207     protected final RemoteRpcProviderConfig getConfig() {
 208         return config;
 209     }
 210
 211     protected final void updateLocalBucket(final T data) {
 212         final LocalBucket<T> local = getLocalBucket();
 213         final boolean bumpIncarnation = local.setData(data);
 214         versions.put(selfAddress, local.getVersion());
 215
 216         if (bumpIncarnation) {
 217             LOG.debug("Version wrapped. incrementing incarnation");
 218
 219             Verify.verify(incarnation < Integer.MAX_VALUE, "Ran out of incarnations, cannot continue");
 220             incarnation = incarnation + 1;
 221
 222             persisting = true;
 223             saveSnapshot(incarnation);
 224         }
 225     }
 226
 227     /**
 228      * Callback to subclasses invoked when a bucket is removed.
 229      *
 230      * @param address Remote address
 231      * @param bucket Bucket removed
 232      */
 233     protected abstract void onBucketRemoved(final Address address, final Bucket<T> bucket);
 234
 235     /**
 236      * Callback to subclasses invoked when the set of remote buckets is updated.
 237      *
 238      * @param newBuckets Map of address to new bucket. Never null, but can be empty.
 239      */
 240     protected abstract void onBucketsUpdated(final Map<Address, Bucket<T>> newBuckets);
 241
 242     /**
 243      * Helper to collect all known buckets.
 244      *
 245      * @return self owned + remote buckets
 246      */
 247     private Map<Address, Bucket<T>> getAllBuckets() {
 248         Map<Address, Bucket<T>> all = new HashMap<>(remoteBuckets.size() + 1);
 249
 250         //first add the local bucket
 251         all.put(selfAddress, getLocalBucket().snapshot());
 252
 253         //then get all remote buckets
 254         all.putAll(remoteBuckets);
 255
 256         return all;
 257     }
 258
 259     /**
 260      * Helper to collect buckets for requested members.
 261      *
 262      * @param members requested members
 263      */
 264     private void getBucketsByMembers(final Collection<Address> members) {
 265         Map<Address, Bucket<T>> buckets = new HashMap<>();
 266
 267         //first add the local bucket if asked
 268         if (members.contains(selfAddress)) {
 269             buckets.put(selfAddress, getLocalBucket().snapshot());
 270         }
 271
 272         //then get buckets for requested remote nodes
 273         for (Address address : members) {
 274             if (remoteBuckets.containsKey(address)) {
 275                 buckets.put(address, remoteBuckets.get(address));
 276             }
 277         }
 278
 279         getSender().tell(buckets, getSelf());
 280     }
 281
 282     private void removeBucket(final Address addr) {
 283         final Bucket<T> bucket = remoteBuckets.remove(addr);
 284         if (bucket != null) {
 285             bucket.getWatchActor().ifPresent(ref -> removeWatch(addr, ref));
 286             onBucketRemoved(addr, bucket);
 287         }
 288     }
 289
 290     /**
 291      * Update local copy of remote buckets where local copy's version is older.
 292      *
 293      * @param receivedBuckets buckets sent by remote
 294      *                        {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}
 295      */
 296     @VisibleForTesting
 297     void updateRemoteBuckets(final Map<Address, Bucket<?>> receivedBuckets) {
 298         LOG.debug("{}: receiveUpdateRemoteBuckets: {}", selfAddress, receivedBuckets);
 299         if (receivedBuckets == null || receivedBuckets.isEmpty()) {
 300             //nothing to do
 301             return;
 302         }
 303
 304         final Map<Address, Bucket<T>> newBuckets = new HashMap<>(receivedBuckets.size());
 305         for (Entry<Address, Bucket<?>> entry : receivedBuckets.entrySet()) {
 306             final Address addr = entry.getKey();
 307
 308             if (selfAddress.equals(addr)) {
 309                 // Remote cannot update our bucket
 310                 continue;
 311             }
 312
 313             @SuppressWarnings("unchecked")
 314             final Bucket<T> receivedBucket = (Bucket<T>) entry.getValue();
 315             if (receivedBucket == null) {
 316                 LOG.debug("Ignoring null bucket from {}", addr);
 317                 continue;
 318             }
 319
 320             // update only if remote version is newer
 321             final long remoteVersion = receivedBucket.getVersion();
 322             final Long localVersion = versions.get(addr);
 323             if (localVersion != null && remoteVersion <= localVersion.longValue()) {
 324                 LOG.debug("Ignoring down-versioned bucket from {} ({} local {} remote)", addr, localVersion,
 325                     remoteVersion);
 326                 continue;
 327             }
 328             newBuckets.put(addr, receivedBucket);
 329             versions.put(addr, remoteVersion);
 330             final Bucket<T> prevBucket = remoteBuckets.put(addr, receivedBucket);
 331
 332             // Deal with DeathWatch subscriptions
 333             final Optional<ActorRef> prevRef = prevBucket != null ? prevBucket.getWatchActor() : Optional.empty();
 334             final Optional<ActorRef> curRef = receivedBucket.getWatchActor();
 335             if (!curRef.equals(prevRef)) {
 336                 prevRef.ifPresent(ref -> removeWatch(addr, ref));
 337                 curRef.ifPresent(ref -> addWatch(addr, ref));
 338             }
 339
 340             LOG.debug("Updating bucket from {} to version {}", entry.getKey(), remoteVersion);
 341         }
 342
 343         LOG.debug("State after update - Local Bucket [{}], Remote Buckets [{}]", localBucket, remoteBuckets);
 344
 345         onBucketsUpdated(newBuckets);
 346     }
 347
 348     private void addWatch(final Address addr, final ActorRef ref) {
 349         if (!watchedActors.containsKey(ref)) {
 350             getContext().watch(ref);
 351             LOG.debug("Watching {}", ref);
 352         }
 353         watchedActors.put(ref, addr);
 354     }
 355
 356     private void removeWatch(final Address addr, final ActorRef ref) {
 357         watchedActors.remove(ref, addr);
 358         if (!watchedActors.containsKey(ref)) {
 359             getContext().unwatch(ref);
 360             LOG.debug("No longer watching {}", ref);
 361         }
 362     }
 363
 364     private void actorTerminated(final Terminated message) {
 365         LOG.info("Actor termination {} received", message);
 366
 367         for (Address addr : watchedActors.removeAll(message.getActor())) {
 368             versions.remove(addr);
 369             final Bucket<T> bucket = remoteBuckets.remove(addr);
 370             if (bucket != null) {
 371                 LOG.debug("Source actor dead, removing bucket {} from ", bucket, addr);
 372                 onBucketRemoved(addr, bucket);
 373             }
 374         }
 375     }
 376
 377     @VisibleForTesting
 378     protected boolean isPersisting() {
 379         return persisting;
 380     }
 381
 382     private LocalBucket<T> getLocalBucket() {
 383         Preconditions.checkState(localBucket != null, "Attempted to access local bucket before recovery completed");
 384         return localBucket;
 385     }
 386 }