opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/gossip/BucketStore.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 akka.actor.ActorRef;
  12 import akka.actor.ActorRefProvider;
  13 import akka.actor.Address;
  14 import akka.actor.Props;
  15 import akka.cluster.ClusterActorRefProvider;
  16 import com.google.common.base.Preconditions;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import java.util.Set;
  20 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActorWithMetering;
  21 import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
  22 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBuckets;
  23 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBucketsReply;
  24 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions;
  25 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply;
  26 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembers;
  27 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketsByMembersReply;
  28 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.UpdateRemoteBuckets;
  29 import org.opendaylight.controller.utils.ConditionalProbe;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 /**
  34  * A store that syncs its data across nodes in the cluster.
  35  * It maintains a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Bucket} per node. Buckets are versioned.
  36  * A node can write ONLY to its bucket. This way, write conflicts are avoided.
  37  * <p>
  38  * Buckets are sync'ed across nodes using Gossip protocol (http://en.wikipedia.org/wiki/Gossip_protocol)<p>
  39  * This store uses a {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}.
  40  *
  41  */
  42 public class BucketStore<T extends Copier<T>> extends AbstractUntypedActorWithMetering {
  43
  44     private static final Long NO_VERSION = -1L;
  45
  46     protected final Logger log = LoggerFactory.getLogger(getClass());
  47
  48     /**
  49      * Bucket owned by the node
  50      */
  51     private final BucketImpl<T> localBucket = new BucketImpl<>();
  52
  53     /**
  54      * Buckets ownded by other known nodes in the cluster
  55      */
  56     private final Map<Address, Bucket<T>> remoteBuckets = new HashMap<>();
  57
  58     /**
  59      * Bucket version for every known node in the cluster including this node
  60      */
  61     private final Map<Address, Long> versions = new HashMap<>();
  62
  63     /**
  64      * Cluster address for this node
  65      */
  66     private Address selfAddress;
  67
  68     private ConditionalProbe probe;
  69
  70     private final RemoteRpcProviderConfig config;
  71
  72     public BucketStore(RemoteRpcProviderConfig config){
  73         this.config = Preconditions.checkNotNull(config);
  74     }
  75
  76     @Override
  77     public void preStart(){
  78         ActorRefProvider provider = getContext().provider();
  79         selfAddress = provider.getDefaultAddress();
  80
  81         if ( provider instanceof ClusterActorRefProvider) {
  82             getContext().actorOf(Props.create(Gossiper.class, config).withMailbox(config.getMailBoxName()), "gossiper");
  83         }
  84     }
  85
  86     @Override
  87     protected void handleReceive(Object message) throws Exception {
  88         if (probe != null) {
  89             probe.tell(message, getSelf());
  90         }
  91
  92         if (message instanceof ConditionalProbe) {
  93             // The ConditionalProbe is only used for unit tests.
  94             log.info("Received probe {} {}", getSelf(), message);
  95             probe = (ConditionalProbe) message;
  96             // Send back any message to tell the caller we got the probe.
  97             getSender().tell("Got it", getSelf());
  98         } else if (message instanceof GetAllBuckets) {
  99             receiveGetAllBuckets();
 100         } else if (message instanceof GetBucketsByMembers) {
 101             receiveGetBucketsByMembers(((GetBucketsByMembers) message).getMembers());
 102         } else if (message instanceof GetBucketVersions) {
 103             receiveGetBucketVersions();
 104         } else if (message instanceof UpdateRemoteBuckets) {
 105             receiveUpdateRemoteBuckets(((UpdateRemoteBuckets<T>) message).getBuckets());
 106         } else {
 107             if(log.isDebugEnabled()) {
 108                 log.debug("Unhandled message [{}]", message);
 109             }
 110             unhandled(message);
 111         }
 112     }
 113
 114     /**
 115      * Returns all the buckets the this node knows about, self owned + remote
 116      */
 117     void receiveGetAllBuckets(){
 118         final ActorRef sender = getSender();
 119         sender.tell(new GetAllBucketsReply<T>(getAllBuckets()), getSelf());
 120     }
 121
 122     /**
 123      * Helper to collect all known buckets
 124      *
 125      * @return self owned + remote buckets
 126      */
 127     Map<Address, Bucket<T>> getAllBuckets(){
 128         Map<Address, Bucket<T>> all = new HashMap<>(remoteBuckets.size() + 1);
 129
 130         //first add the local bucket
 131         all.put(selfAddress, new BucketImpl<>(localBucket));
 132
 133         //then get all remote buckets
 134         all.putAll(remoteBuckets);
 135
 136         return all;
 137     }
 138
 139     /**
 140      * Returns buckets for requested members that this node knows about
 141      *
 142      * @param members requested members
 143      */
 144     void receiveGetBucketsByMembers(Set<Address> members){
 145         final ActorRef sender = getSender();
 146         Map<Address, Bucket<T>> buckets = getBucketsByMembers(members);
 147         sender.tell(new GetBucketsByMembersReply<T>(buckets), getSelf());
 148     }
 149
 150     /**
 151      * Helper to collect buckets for requested memebers
 152      *
 153      * @param members requested members
 154      * @return buckets for requested memebers
 155      */
 156     Map<Address, Bucket<T>> getBucketsByMembers(Set<Address> members) {
 157         Map<Address, Bucket<T>> buckets = new HashMap<>();
 158
 159         //first add the local bucket if asked
 160         if (members.contains(selfAddress)) {
 161             buckets.put(selfAddress, new BucketImpl<>(localBucket));
 162         }
 163
 164         //then get buckets for requested remote nodes
 165         for (Address address : members){
 166             if (remoteBuckets.containsKey(address)) {
 167                 buckets.put(address, remoteBuckets.get(address));
 168             }
 169         }
 170
 171         return buckets;
 172     }
 173
 174     /**
 175      * Returns versions for all buckets known
 176      */
 177     void receiveGetBucketVersions(){
 178         final ActorRef sender = getSender();
 179         GetBucketVersionsReply reply = new GetBucketVersionsReply(versions);
 180         sender.tell(reply, getSelf());
 181     }
 182
 183     /**
 184      * Update local copy of remote buckets where local copy's version is older
 185      *
 186      * @param receivedBuckets buckets sent by remote
 187      *                        {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}
 188      */
 189     void receiveUpdateRemoteBuckets(Map<Address, Bucket<T>> receivedBuckets){
 190         log.debug("{}: receiveUpdateRemoteBuckets: {}", selfAddress, receivedBuckets);
 191         if (receivedBuckets == null || receivedBuckets.isEmpty())
 192          {
 193             return; //nothing to do
 194         }
 195
 196         //Remote cant update self's bucket
 197         receivedBuckets.remove(selfAddress);
 198
 199         for (Map.Entry<Address, Bucket<T>> entry : receivedBuckets.entrySet()){
 200
 201             Long localVersion = versions.get(entry.getKey());
 202             if (localVersion == null) {
 203                 localVersion = NO_VERSION;
 204             }
 205
 206             Bucket<T> receivedBucket = entry.getValue();
 207
 208             if (receivedBucket == null) {
 209                 continue;
 210             }
 211
 212             Long remoteVersion = receivedBucket.getVersion();
 213             if (remoteVersion == null) {
 214                 remoteVersion = NO_VERSION;
 215             }
 216
 217             //update only if remote version is newer
 218             if ( remoteVersion.longValue() > localVersion.longValue() ) {
 219                 remoteBuckets.put(entry.getKey(), receivedBucket);
 220                 versions.put(entry.getKey(), remoteVersion);
 221             }
 222         }
 223
 224         if(log.isDebugEnabled()) {
 225             log.debug("State after update - Local Bucket [{}], Remote Buckets [{}]", localBucket, remoteBuckets);
 226         }
 227     }
 228
 229     public BucketImpl<T> getLocalBucket() {
 230         return localBucket;
 231     }
 232
 233     protected void updateLocalBucket(T data) {
 234         localBucket.setData(data);
 235         versions.put(selfAddress, localBucket.getVersion());
 236     }
 237
 238     public Map<Address, Bucket<T>> getRemoteBuckets() {
 239         return remoteBuckets;
 240     }
 241
 242     public Map<Address, Long> getVersions() {
 243         return versions;
 244     }
 245 }