git.opendaylight Code Review - controller.git/blob

   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     protected RemoteRpcProviderConfig getConfig() {
 115         return config;
 116     }
 117
 118     /**
 119      * Returns all the buckets the this node knows about, self owned + remote
 120      */
 121     void receiveGetAllBuckets(){
 122         final ActorRef sender = getSender();
 123         sender.tell(new GetAllBucketsReply<>(getAllBuckets()), getSelf());
 124     }
 125
 126     /**
 127      * Helper to collect all known buckets
 128      *
 129      * @return self owned + remote buckets
 130      */
 131     Map<Address, Bucket<T>> getAllBuckets(){
 132         Map<Address, Bucket<T>> all = new HashMap<>(remoteBuckets.size() + 1);
 133
 134         //first add the local bucket
 135         all.put(selfAddress, new BucketImpl<>(localBucket));
 136
 137         //then get all remote buckets
 138         all.putAll(remoteBuckets);
 139
 140         return all;
 141     }
 142
 143     /**
 144      * Returns buckets for requested members that this node knows about
 145      *
 146      * @param members requested members
 147      */
 148     void receiveGetBucketsByMembers(Set<Address> members){
 149         final ActorRef sender = getSender();
 150         Map<Address, Bucket<T>> buckets = getBucketsByMembers(members);
 151         sender.tell(new GetBucketsByMembersReply<>(buckets), getSelf());
 152     }
 153
 154     /**
 155      * Helper to collect buckets for requested memebers
 156      *
 157      * @param members requested members
 158      * @return buckets for requested memebers
 159      */
 160     Map<Address, Bucket<T>> getBucketsByMembers(Set<Address> members) {
 161         Map<Address, Bucket<T>> buckets = new HashMap<>();
 162
 163         //first add the local bucket if asked
 164         if (members.contains(selfAddress)) {
 165             buckets.put(selfAddress, new BucketImpl<>(localBucket));
 166         }
 167
 168         //then get buckets for requested remote nodes
 169         for (Address address : members){
 170             if (remoteBuckets.containsKey(address)) {
 171                 buckets.put(address, remoteBuckets.get(address));
 172             }
 173         }
 174
 175         return buckets;
 176     }
 177
 178     /**
 179      * Returns versions for all buckets known
 180      */
 181     void receiveGetBucketVersions(){
 182         final ActorRef sender = getSender();
 183         GetBucketVersionsReply reply = new GetBucketVersionsReply(versions);
 184         sender.tell(reply, getSelf());
 185     }
 186
 187     /**
 188      * Update local copy of remote buckets where local copy's version is older
 189      *
 190      * @param receivedBuckets buckets sent by remote
 191      *                        {@link org.opendaylight.controller.remote.rpc.registry.gossip.Gossiper}
 192      */
 193     void receiveUpdateRemoteBuckets(Map<Address, Bucket<T>> receivedBuckets){
 194         log.debug("{}: receiveUpdateRemoteBuckets: {}", selfAddress, receivedBuckets);
 195         if (receivedBuckets == null || receivedBuckets.isEmpty())
 196          {
 197             return; //nothing to do
 198         }
 199
 200         //Remote cant update self's bucket
 201         receivedBuckets.remove(selfAddress);
 202
 203         for (Map.Entry<Address, Bucket<T>> entry : receivedBuckets.entrySet()){
 204
 205             Long localVersion = versions.get(entry.getKey());
 206             if (localVersion == null) {
 207                 localVersion = NO_VERSION;
 208             }
 209
 210             Bucket<T> receivedBucket = entry.getValue();
 211
 212             if (receivedBucket == null) {
 213                 continue;
 214             }
 215
 216             Long remoteVersion = receivedBucket.getVersion();
 217             if (remoteVersion == null) {
 218                 remoteVersion = NO_VERSION;
 219             }
 220
 221             //update only if remote version is newer
 222             if ( remoteVersion.longValue() > localVersion.longValue() ) {
 223                 remoteBuckets.put(entry.getKey(), receivedBucket);
 224                 versions.put(entry.getKey(), remoteVersion);
 225             }
 226         }
 227
 228         if(log.isDebugEnabled()) {
 229             log.debug("State after update - Local Bucket [{}], Remote Buckets [{}]", localBucket, remoteBuckets);
 230         }
 231
 232         onBucketsUpdated();
 233     }
 234
 235     protected void onBucketsUpdated() {
 236     }
 237
 238     public BucketImpl<T> getLocalBucket() {
 239         return localBucket;
 240     }
 241
 242     protected void updateLocalBucket(T data) {
 243         localBucket.setData(data);
 244         versions.put(selfAddress, localBucket.getVersion());
 245     }
 246
 247     public Map<Address, Bucket<T>> getRemoteBuckets() {
 248         return remoteBuckets;
 249     }
 250
 251     public Map<Address, Long> getVersions() {
 252         return versions;
 253     }
 254 }