gerrit.opnfv Code Review - onosfw.git/blob

   1 /*
   2  * Copyright 2014-2015 Open Networking Laboratory
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package org.onosproject.store.link.impl;
  17
  18 import com.google.common.base.Function;
  19 import com.google.common.collect.FluentIterable;
  20 import com.google.common.collect.ImmutableList;
  21 import com.google.common.collect.Multimaps;
  22 import com.google.common.collect.SetMultimap;
  23 import com.google.common.collect.Sets;
  24 import org.apache.commons.lang3.RandomUtils;
  25 import org.apache.felix.scr.annotations.Activate;
  26 import org.apache.felix.scr.annotations.Component;
  27 import org.apache.felix.scr.annotations.Deactivate;
  28 import org.apache.felix.scr.annotations.Reference;
  29 import org.apache.felix.scr.annotations.ReferenceCardinality;
  30 import org.apache.felix.scr.annotations.Service;
  31 import org.onlab.util.KryoNamespace;
  32 import org.onosproject.cluster.ClusterService;
  33 import org.onosproject.cluster.ControllerNode;
  34 import org.onosproject.cluster.NodeId;
  35 import org.onosproject.mastership.MastershipService;
  36 import org.onosproject.net.AnnotationKeys;
  37 import org.onosproject.net.AnnotationsUtil;
  38 import org.onosproject.net.ConnectPoint;
  39 import org.onosproject.net.DefaultAnnotations;
  40 import org.onosproject.net.DefaultLink;
  41 import org.onosproject.net.DeviceId;
  42 import org.onosproject.net.Link;
  43 import org.onosproject.net.Link.Type;
  44 import org.onosproject.net.LinkKey;
  45 import org.onosproject.net.SparseAnnotations;
  46 import org.onosproject.net.device.DeviceClockService;
  47 import org.onosproject.net.link.DefaultLinkDescription;
  48 import org.onosproject.net.link.LinkDescription;
  49 import org.onosproject.net.link.LinkEvent;
  50 import org.onosproject.net.link.LinkStore;
  51 import org.onosproject.net.link.LinkStoreDelegate;
  52 import org.onosproject.net.provider.ProviderId;
  53 import org.onosproject.store.AbstractStore;
  54 import org.onosproject.store.Timestamp;
  55 import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
  56 import org.onosproject.store.cluster.messaging.ClusterMessage;
  57 import org.onosproject.store.cluster.messaging.ClusterMessageHandler;
  58 import org.onosproject.store.cluster.messaging.MessageSubject;
  59 import org.onosproject.store.impl.Timestamped;
  60 import org.onosproject.store.serializers.KryoSerializer;
  61 import org.onosproject.store.serializers.custom.DistributedStoreSerializers;
  62 import org.slf4j.Logger;
  63
  64 import java.io.IOException;
  65 import java.util.Collections;
  66 import java.util.HashMap;
  67 import java.util.HashSet;
  68 import java.util.Map;
  69 import java.util.Map.Entry;
  70 import java.util.Objects;
  71 import java.util.Set;
  72 import java.util.concurrent.ConcurrentHashMap;
  73 import java.util.concurrent.ConcurrentMap;
  74 import java.util.concurrent.ExecutorService;
  75 import java.util.concurrent.Executors;
  76 import java.util.concurrent.ScheduledExecutorService;
  77 import java.util.concurrent.TimeUnit;
  78
  79 import static com.google.common.base.Preconditions.checkNotNull;
  80 import static com.google.common.base.Predicates.notNull;
  81 import static com.google.common.collect.Multimaps.synchronizedSetMultimap;
  82 import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
  83 import static org.onlab.util.Tools.groupedThreads;
  84 import static org.onlab.util.Tools.minPriority;
  85 import static org.onosproject.cluster.ControllerNodeToNodeId.toNodeId;
  86 import static org.onosproject.net.DefaultAnnotations.merge;
  87 import static org.onosproject.net.DefaultAnnotations.union;
  88 import static org.onosproject.net.Link.State.ACTIVE;
  89 import static org.onosproject.net.Link.State.INACTIVE;
  90 import static org.onosproject.net.Link.Type.DIRECT;
  91 import static org.onosproject.net.Link.Type.INDIRECT;
  92 import static org.onosproject.net.LinkKey.linkKey;
  93 import static org.onosproject.net.link.LinkEvent.Type.*;
  94 import static org.onosproject.store.link.impl.GossipLinkStoreMessageSubjects.LINK_ANTI_ENTROPY_ADVERTISEMENT;
  95 import static org.slf4j.LoggerFactory.getLogger;
  96
  97 /**
  98  * Manages inventory of infrastructure links in distributed data store
  99  * that uses optimistic replication and gossip based techniques.
 100  */
 101 @Component(immediate = true, enabled = false)
 102 @Service
 103 public class GossipLinkStore
 104         extends AbstractStore<LinkEvent, LinkStoreDelegate>
 105         implements LinkStore {
 106
 107     // Timeout in milliseconds to process links on remote master node
 108     private static final int REMOTE_MASTER_TIMEOUT = 1000;
 109
 110     private final Logger log = getLogger(getClass());
 111
 112     // Link inventory
 113     private final ConcurrentMap<LinkKey, Map<ProviderId, Timestamped<LinkDescription>>> linkDescs =
 114         new ConcurrentHashMap<>();
 115
 116     // Link instance cache
 117     private final ConcurrentMap<LinkKey, Link> links = new ConcurrentHashMap<>();
 118
 119     // Egress and ingress link sets
 120     private final SetMultimap<DeviceId, LinkKey> srcLinks = createSynchronizedHashMultiMap();
 121     private final SetMultimap<DeviceId, LinkKey> dstLinks = createSynchronizedHashMultiMap();
 122
 123     // Remove links
 124     private final Map<LinkKey, Timestamp> removedLinks = new ConcurrentHashMap<>();
 125
 126     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 127     protected DeviceClockService deviceClockService;
 128
 129     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 130     protected ClusterCommunicationService clusterCommunicator;
 131
 132     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 133     protected ClusterService clusterService;
 134
 135     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
 136     protected MastershipService mastershipService;
 137
 138     protected static final KryoSerializer SERIALIZER = new KryoSerializer() {
 139         @Override
 140         protected void setupKryoPool() {
 141             serializerPool = KryoNamespace.newBuilder()
 142                     .register(DistributedStoreSerializers.STORE_COMMON)
 143                     .nextId(DistributedStoreSerializers.STORE_CUSTOM_BEGIN)
 144                     .register(InternalLinkEvent.class)
 145                     .register(InternalLinkRemovedEvent.class)
 146                     .register(LinkAntiEntropyAdvertisement.class)
 147                     .register(LinkFragmentId.class)
 148                     .register(LinkInjectedEvent.class)
 149                     .build();
 150         }
 151     };
 152
 153     private ExecutorService executor;
 154
 155     private ScheduledExecutorService backgroundExecutors;
 156
 157     @Activate
 158     public void activate() {
 159
 160         executor = Executors.newCachedThreadPool(groupedThreads("onos/link", "fg-%d"));
 161
 162         backgroundExecutors =
 163                 newSingleThreadScheduledExecutor(minPriority(groupedThreads("onos/link", "bg-%d")));
 164
 165         clusterCommunicator.addSubscriber(
 166                 GossipLinkStoreMessageSubjects.LINK_UPDATE,
 167                 new InternalLinkEventListener(), executor);
 168         clusterCommunicator.addSubscriber(
 169                 GossipLinkStoreMessageSubjects.LINK_REMOVED,
 170                 new InternalLinkRemovedEventListener(), executor);
 171         clusterCommunicator.addSubscriber(
 172                 GossipLinkStoreMessageSubjects.LINK_ANTI_ENTROPY_ADVERTISEMENT,
 173                 new InternalLinkAntiEntropyAdvertisementListener(), backgroundExecutors);
 174         clusterCommunicator.addSubscriber(
 175                 GossipLinkStoreMessageSubjects.LINK_INJECTED,
 176                 new LinkInjectedEventListener(), executor);
 177
 178         long initialDelaySec = 5;
 179         long periodSec = 5;
 180         // start anti-entropy thread
 181         backgroundExecutors.scheduleAtFixedRate(new SendAdvertisementTask(),
 182                     initialDelaySec, periodSec, TimeUnit.SECONDS);
 183
 184         log.info("Started");
 185     }
 186
 187     @Deactivate
 188     public void deactivate() {
 189
 190         executor.shutdownNow();
 191
 192         backgroundExecutors.shutdownNow();
 193         try {
 194             if (!backgroundExecutors.awaitTermination(5, TimeUnit.SECONDS)) {
 195                 log.error("Timeout during executor shutdown");
 196             }
 197         } catch (InterruptedException e) {
 198             log.error("Error during executor shutdown", e);
 199         }
 200
 201         linkDescs.clear();
 202         links.clear();
 203         srcLinks.clear();
 204         dstLinks.clear();
 205         log.info("Stopped");
 206     }
 207
 208     @Override
 209     public int getLinkCount() {
 210         return links.size();
 211     }
 212
 213     @Override
 214     public Iterable<Link> getLinks() {
 215         return Collections.unmodifiableCollection(links.values());
 216     }
 217
 218     @Override
 219     public Set<Link> getDeviceEgressLinks(DeviceId deviceId) {
 220         // lock for iteration
 221         synchronized (srcLinks) {
 222             return FluentIterable.from(srcLinks.get(deviceId))
 223             .transform(lookupLink())
 224             .filter(notNull())
 225             .toSet();
 226         }
 227     }
 228
 229     @Override
 230     public Set<Link> getDeviceIngressLinks(DeviceId deviceId) {
 231         // lock for iteration
 232         synchronized (dstLinks) {
 233             return FluentIterable.from(dstLinks.get(deviceId))
 234             .transform(lookupLink())
 235             .filter(notNull())
 236             .toSet();
 237         }
 238     }
 239
 240     @Override
 241     public Link getLink(ConnectPoint src, ConnectPoint dst) {
 242         return links.get(linkKey(src, dst));
 243     }
 244
 245     @Override
 246     public Set<Link> getEgressLinks(ConnectPoint src) {
 247         Set<Link> egress = new HashSet<>();
 248         //
 249         // Change `srcLinks` to ConcurrentMap<DeviceId, (Concurrent)Set>
 250         // to remove this synchronized block, if we hit performance issue.
 251         // SetMultiMap#get returns wrapped collection to provide modifiable-view.
 252         // And the wrapped collection is not concurrent access safe.
 253         //
 254         // Our use case here does not require returned collection to be modifiable,
 255         // so the wrapped collection forces us to lock the whole multiset,
 256         // for benefit we don't need.
 257         //
 258         // Same applies to `dstLinks`
 259         synchronized (srcLinks) {
 260             for (LinkKey linkKey : srcLinks.get(src.deviceId())) {
 261                 if (linkKey.src().equals(src)) {
 262                     Link link = links.get(linkKey);
 263                     if (link != null) {
 264                         egress.add(link);
 265                     } else {
 266                         log.debug("Egress link for {} was null, skipped", linkKey);
 267                     }
 268                 }
 269             }
 270         }
 271         return egress;
 272     }
 273
 274     @Override
 275     public Set<Link> getIngressLinks(ConnectPoint dst) {
 276         Set<Link> ingress = new HashSet<>();
 277         synchronized (dstLinks) {
 278             for (LinkKey linkKey : dstLinks.get(dst.deviceId())) {
 279                 if (linkKey.dst().equals(dst)) {
 280                     Link link = links.get(linkKey);
 281                     if (link != null) {
 282                         ingress.add(link);
 283                     } else {
 284                         log.debug("Ingress link for {} was null, skipped", linkKey);
 285                     }
 286                 }
 287             }
 288         }
 289         return ingress;
 290     }
 291
 292     @Override
 293     public LinkEvent createOrUpdateLink(ProviderId providerId,
 294                                         LinkDescription linkDescription) {
 295
 296         final DeviceId dstDeviceId = linkDescription.dst().deviceId();
 297         final NodeId localNode = clusterService.getLocalNode().id();
 298         final NodeId dstNode = mastershipService.getMasterFor(dstDeviceId);
 299
 300         // Process link update only if we're the master of the destination node,
 301         // otherwise signal the actual master.
 302         LinkEvent linkEvent = null;
 303         if (localNode.equals(dstNode)) {
 304
 305             Timestamp newTimestamp = deviceClockService.getTimestamp(dstDeviceId);
 306
 307             final Timestamped<LinkDescription> deltaDesc = new Timestamped<>(linkDescription, newTimestamp);
 308
 309             LinkKey key = linkKey(linkDescription.src(), linkDescription.dst());
 310             final Timestamped<LinkDescription> mergedDesc;
 311             Map<ProviderId, Timestamped<LinkDescription>> map = getOrCreateLinkDescriptions(key);
 312
 313             synchronized (map) {
 314                 linkEvent = createOrUpdateLinkInternal(providerId, deltaDesc);
 315                 mergedDesc = map.get(providerId);
 316             }
 317
 318             if (linkEvent != null) {
 319                 log.debug("Notifying peers of a link update topology event from providerId: "
 320                                 + "{}  between src: {} and dst: {}",
 321                         providerId, linkDescription.src(), linkDescription.dst());
 322                 notifyPeers(new InternalLinkEvent(providerId, mergedDesc));
 323             }
 324
 325         } else {
 326             // FIXME Temporary hack for NPE (ONOS-1171).
 327             // Proper fix is to implement forwarding to master on ConfigProvider
 328             // redo ONOS-490
 329             if (dstNode == null) {
 330                 // silently ignore
 331                 return null;
 332             }
 333
 334
 335             LinkInjectedEvent linkInjectedEvent = new LinkInjectedEvent(providerId, linkDescription);
 336
 337             // TODO check unicast return value
 338             clusterCommunicator.unicast(linkInjectedEvent,
 339                     GossipLinkStoreMessageSubjects.LINK_INJECTED,
 340                     SERIALIZER::encode,
 341                     dstNode);
 342         }
 343
 344         return linkEvent;
 345     }
 346
 347     @Override
 348     public LinkEvent removeOrDownLink(ConnectPoint src, ConnectPoint dst) {
 349         Link link = getLink(src, dst);
 350         if (link == null) {
 351             return null;
 352         }
 353
 354         if (link.isDurable()) {
 355             // FIXME: this is not the right thing to call for the gossip store; will not sync link state!!!
 356             return link.state() == INACTIVE ? null :
 357                     updateLink(linkKey(link.src(), link.dst()), link,
 358                                new DefaultLink(link.providerId(),
 359                                                link.src(), link.dst(),
 360                                                link.type(), INACTIVE,
 361                                                link.isDurable(),
 362                                                link.annotations()));
 363         }
 364         return removeLink(src, dst);
 365     }
 366
 367     private LinkEvent createOrUpdateLinkInternal(
 368             ProviderId providerId,
 369             Timestamped<LinkDescription> linkDescription) {
 370
 371         final LinkKey key = linkKey(linkDescription.value().src(),
 372                 linkDescription.value().dst());
 373         Map<ProviderId, Timestamped<LinkDescription>> descs = getOrCreateLinkDescriptions(key);
 374
 375         synchronized (descs) {
 376             // if the link was previously removed, we should proceed if and
 377             // only if this request is more recent.
 378             Timestamp linkRemovedTimestamp = removedLinks.get(key);
 379             if (linkRemovedTimestamp != null) {
 380                 if (linkDescription.isNewerThan(linkRemovedTimestamp)) {
 381                     removedLinks.remove(key);
 382                 } else {
 383                     log.trace("Link {} was already removed ignoring.", key);
 384                     return null;
 385                 }
 386             }
 387
 388             final Link oldLink = links.get(key);
 389             // update description
 390             createOrUpdateLinkDescription(descs, providerId, linkDescription);
 391             final Link newLink = composeLink(descs);
 392             if (oldLink == null) {
 393                 return createLink(key, newLink);
 394             }
 395             return updateLink(key, oldLink, newLink);
 396         }
 397     }
 398
 399     // Guarded by linkDescs value (=locking each Link)
 400     private Timestamped<LinkDescription> createOrUpdateLinkDescription(
 401             Map<ProviderId, Timestamped<LinkDescription>> descs,
 402             ProviderId providerId,
 403             Timestamped<LinkDescription> linkDescription) {
 404
 405         // merge existing annotations
 406         Timestamped<LinkDescription> existingLinkDescription = descs.get(providerId);
 407         if (existingLinkDescription != null && existingLinkDescription.isNewer(linkDescription)) {
 408             log.trace("local info is more up-to-date, ignoring {}.", linkDescription);
 409             return null;
 410         }
 411         Timestamped<LinkDescription> newLinkDescription = linkDescription;
 412         if (existingLinkDescription != null) {
 413             // we only allow transition from INDIRECT -> DIRECT
 414             final Type newType;
 415             if (existingLinkDescription.value().type() == DIRECT) {
 416                 newType = DIRECT;
 417             } else {
 418                 newType = linkDescription.value().type();
 419             }
 420             SparseAnnotations merged = union(existingLinkDescription.value().annotations(),
 421                     linkDescription.value().annotations());
 422             newLinkDescription = new Timestamped<>(
 423                     new DefaultLinkDescription(
 424                         linkDescription.value().src(),
 425                         linkDescription.value().dst(),
 426                         newType, merged),
 427                     linkDescription.timestamp());
 428         }
 429         return descs.put(providerId, newLinkDescription);
 430     }
 431
 432     // Creates and stores the link and returns the appropriate event.
 433     // Guarded by linkDescs value (=locking each Link)
 434     private LinkEvent createLink(LinkKey key, Link newLink) {
 435         links.put(key, newLink);
 436         srcLinks.put(newLink.src().deviceId(), key);
 437         dstLinks.put(newLink.dst().deviceId(), key);
 438         return new LinkEvent(LINK_ADDED, newLink);
 439     }
 440
 441     // Updates, if necessary the specified link and returns the appropriate event.
 442     // Guarded by linkDescs value (=locking each Link)
 443     private LinkEvent updateLink(LinkKey key, Link oldLink, Link newLink) {
 444         // Note: INDIRECT -> DIRECT transition only
 445         // so that BDDP discovered Link will not overwrite LDDP Link
 446         if (oldLink.state() != newLink.state() ||
 447             (oldLink.type() == INDIRECT && newLink.type() == DIRECT) ||
 448             !AnnotationsUtil.isEqual(oldLink.annotations(), newLink.annotations())) {
 449
 450             links.put(key, newLink);
 451             // strictly speaking following can be omitted
 452             srcLinks.put(oldLink.src().deviceId(), key);
 453             dstLinks.put(oldLink.dst().deviceId(), key);
 454             return new LinkEvent(LINK_UPDATED, newLink);
 455         }
 456         return null;
 457     }
 458
 459     @Override
 460     public LinkEvent removeLink(ConnectPoint src, ConnectPoint dst) {
 461         final LinkKey key = linkKey(src, dst);
 462
 463         DeviceId dstDeviceId = dst.deviceId();
 464         Timestamp timestamp = null;
 465         try {
 466             timestamp = deviceClockService.getTimestamp(dstDeviceId);
 467         } catch (IllegalStateException e) {
 468             log.debug("Failed to remove link {}, was not the master", key);
 469             // there are times when this is called before mastership
 470             // handoff correctly completes.
 471             return null;
 472         }
 473
 474         LinkEvent event = removeLinkInternal(key, timestamp);
 475
 476         if (event != null) {
 477             log.debug("Notifying peers of a link removed topology event for a link "
 478                     + "between src: {} and dst: {}", src, dst);
 479             notifyPeers(new InternalLinkRemovedEvent(key, timestamp));
 480         }
 481         return event;
 482     }
 483
 484     private static Timestamped<LinkDescription> getPrimaryDescription(
 485                 Map<ProviderId, Timestamped<LinkDescription>> linkDescriptions) {
 486
 487         synchronized (linkDescriptions) {
 488             for (Entry<ProviderId, Timestamped<LinkDescription>>
 489                     e : linkDescriptions.entrySet()) {
 490
 491                 if (!e.getKey().isAncillary()) {
 492                     return e.getValue();
 493                 }
 494             }
 495         }
 496         return null;
 497     }
 498
 499
 500     // TODO: consider slicing out as Timestamp utils
 501     /**
 502      * Checks is timestamp is more recent than timestamped object.
 503      *
 504      * @param timestamp to check if this is more recent then other
 505      * @param timestamped object to be tested against
 506      * @return true if {@code timestamp} is more recent than {@code timestamped}
 507      *         or {@code timestamped is null}
 508      */
 509     private static boolean isMoreRecent(Timestamp timestamp, Timestamped<?> timestamped) {
 510         checkNotNull(timestamp);
 511         if (timestamped == null) {
 512             return true;
 513         }
 514         return timestamp.compareTo(timestamped.timestamp()) > 0;
 515     }
 516
 517     private LinkEvent removeLinkInternal(LinkKey key, Timestamp timestamp) {
 518         Map<ProviderId, Timestamped<LinkDescription>> linkDescriptions
 519             = getOrCreateLinkDescriptions(key);
 520
 521         synchronized (linkDescriptions) {
 522             if (linkDescriptions.isEmpty()) {
 523                 // never seen such link before. keeping timestamp for record
 524                 removedLinks.put(key, timestamp);
 525                 return null;
 526             }
 527             // accept removal request if given timestamp is newer than
 528             // the latest Timestamp from Primary provider
 529             Timestamped<LinkDescription> prim = getPrimaryDescription(linkDescriptions);
 530             if (!isMoreRecent(timestamp, prim)) {
 531                 // outdated remove request, ignore
 532                 return null;
 533             }
 534             removedLinks.put(key, timestamp);
 535             Link link = links.remove(key);
 536             linkDescriptions.clear();
 537             if (link != null) {
 538                 srcLinks.remove(link.src().deviceId(), key);
 539                 dstLinks.remove(link.dst().deviceId(), key);
 540                 return new LinkEvent(LINK_REMOVED, link);
 541             }
 542             return null;
 543         }
 544     }
 545
 546     /**
 547      * Creates concurrent readable, synchronized HashMultimap.
 548      *
 549      * @return SetMultimap
 550      */
 551     private static <K, V> SetMultimap<K, V> createSynchronizedHashMultiMap() {
 552         return synchronizedSetMultimap(
 553                Multimaps.newSetMultimap(new ConcurrentHashMap<>(),
 554                                        () -> Sets.newConcurrentHashSet()));
 555     }
 556
 557     /**
 558      * @return primary ProviderID, or randomly chosen one if none exists
 559      */
 560     private static ProviderId pickBaseProviderId(
 561             Map<ProviderId, Timestamped<LinkDescription>> linkDescriptions) {
 562
 563         ProviderId fallBackPrimary = null;
 564         for (Entry<ProviderId, Timestamped<LinkDescription>> e : linkDescriptions.entrySet()) {
 565             if (!e.getKey().isAncillary()) {
 566                 // found primary
 567                 return e.getKey();
 568             } else if (fallBackPrimary == null) {
 569                 // pick randomly as a fallback in case there is no primary
 570                 fallBackPrimary = e.getKey();
 571             }
 572         }
 573         return fallBackPrimary;
 574     }
 575
 576     // Guarded by linkDescs value (=locking each Link)
 577     private Link composeLink(Map<ProviderId, Timestamped<LinkDescription>> descs) {
 578         ProviderId baseProviderId = pickBaseProviderId(descs);
 579         Timestamped<LinkDescription> base = descs.get(baseProviderId);
 580
 581         ConnectPoint src = base.value().src();
 582         ConnectPoint dst = base.value().dst();
 583         Type type = base.value().type();
 584         DefaultAnnotations annotations = DefaultAnnotations.builder().build();
 585         annotations = merge(annotations, base.value().annotations());
 586
 587         for (Entry<ProviderId, Timestamped<LinkDescription>> e : descs.entrySet()) {
 588             if (baseProviderId.equals(e.getKey())) {
 589                 continue;
 590             }
 591
 592             // Note: In the long run we should keep track of Description timestamp
 593             // and only merge conflicting keys when timestamp is newer
 594             // Currently assuming there will never be a key conflict between
 595             // providers
 596
 597             // annotation merging. not so efficient, should revisit later
 598             annotations = merge(annotations, e.getValue().value().annotations());
 599         }
 600
 601         boolean isDurable = Objects.equals(annotations.value(AnnotationKeys.DURABLE), "true");
 602         return new DefaultLink(baseProviderId, src, dst, type, ACTIVE, isDurable, annotations);
 603     }
 604
 605     private Map<ProviderId, Timestamped<LinkDescription>> getOrCreateLinkDescriptions(LinkKey key) {
 606         Map<ProviderId, Timestamped<LinkDescription>> r;
 607         r = linkDescs.get(key);
 608         if (r != null) {
 609             return r;
 610         }
 611         r = new HashMap<>();
 612         final Map<ProviderId, Timestamped<LinkDescription>> concurrentlyAdded;
 613         concurrentlyAdded = linkDescs.putIfAbsent(key, r);
 614         if (concurrentlyAdded != null) {
 615             return concurrentlyAdded;
 616         } else {
 617             return r;
 618         }
 619     }
 620
 621     private final Function<LinkKey, Link> lookupLink = new LookupLink();
 622
 623     /**
 624      * Returns a Function to lookup Link instance using LinkKey from cache.
 625      *
 626      * @return lookup link function
 627      */
 628     private Function<LinkKey, Link> lookupLink() {
 629         return lookupLink;
 630     }
 631
 632     private final class LookupLink implements Function<LinkKey, Link> {
 633         @Override
 634         public Link apply(LinkKey input) {
 635             if (input == null) {
 636                 return null;
 637             } else {
 638                 return links.get(input);
 639             }
 640         }
 641     }
 642
 643     private void notifyDelegateIfNotNull(LinkEvent event) {
 644         if (event != null) {
 645             notifyDelegate(event);
 646         }
 647     }
 648
 649     private void broadcastMessage(MessageSubject subject, Object event) {
 650         clusterCommunicator.broadcast(event, subject, SERIALIZER::encode);
 651     }
 652
 653     private void unicastMessage(NodeId recipient, MessageSubject subject, Object event) throws IOException {
 654         clusterCommunicator.unicast(event, subject, SERIALIZER::encode, recipient);
 655     }
 656
 657     private void notifyPeers(InternalLinkEvent event) {
 658         broadcastMessage(GossipLinkStoreMessageSubjects.LINK_UPDATE, event);
 659     }
 660
 661     private void notifyPeers(InternalLinkRemovedEvent event) {
 662         broadcastMessage(GossipLinkStoreMessageSubjects.LINK_REMOVED, event);
 663     }
 664
 665     // notify peer, silently ignoring error
 666     private void notifyPeer(NodeId peer, InternalLinkEvent event) {
 667         try {
 668             unicastMessage(peer, GossipLinkStoreMessageSubjects.LINK_UPDATE, event);
 669         } catch (IOException e) {
 670             log.debug("Failed to notify peer {} with message {}", peer, event);
 671         }
 672     }
 673
 674     // notify peer, silently ignoring error
 675     private void notifyPeer(NodeId peer, InternalLinkRemovedEvent event) {
 676         try {
 677             unicastMessage(peer, GossipLinkStoreMessageSubjects.LINK_REMOVED, event);
 678         } catch (IOException e) {
 679             log.debug("Failed to notify peer {} with message {}", peer, event);
 680         }
 681     }
 682
 683     private final class SendAdvertisementTask implements Runnable {
 684
 685         @Override
 686         public void run() {
 687             if (Thread.currentThread().isInterrupted()) {
 688                 log.debug("Interrupted, quitting");
 689                 return;
 690             }
 691
 692             try {
 693                 final NodeId self = clusterService.getLocalNode().id();
 694                 Set<ControllerNode> nodes = clusterService.getNodes();
 695
 696                 ImmutableList<NodeId> nodeIds = FluentIterable.from(nodes)
 697                         .transform(toNodeId())
 698                         .toList();
 699
 700                 if (nodeIds.size() == 1 && nodeIds.get(0).equals(self)) {
 701                     log.trace("No other peers in the cluster.");
 702                     return;
 703                 }
 704
 705                 NodeId peer;
 706                 do {
 707                     int idx = RandomUtils.nextInt(0, nodeIds.size());
 708                     peer = nodeIds.get(idx);
 709                 } while (peer.equals(self));
 710
 711                 LinkAntiEntropyAdvertisement ad = createAdvertisement();
 712
 713                 if (Thread.currentThread().isInterrupted()) {
 714                     log.debug("Interrupted, quitting");
 715                     return;
 716                 }
 717
 718                 try {
 719                     unicastMessage(peer, LINK_ANTI_ENTROPY_ADVERTISEMENT, ad);
 720                 } catch (IOException e) {
 721                     log.debug("Failed to send anti-entropy advertisement to {}", peer);
 722                     return;
 723                 }
 724             } catch (Exception e) {
 725                 // catch all Exception to avoid Scheduled task being suppressed.
 726                 log.error("Exception thrown while sending advertisement", e);
 727             }
 728         }
 729     }
 730
 731     private LinkAntiEntropyAdvertisement createAdvertisement() {
 732         final NodeId self = clusterService.getLocalNode().id();
 733
 734         Map<LinkFragmentId, Timestamp> linkTimestamps = new HashMap<>(linkDescs.size());
 735         Map<LinkKey, Timestamp> linkTombstones = new HashMap<>(removedLinks.size());
 736
 737         linkDescs.forEach((linkKey, linkDesc) -> {
 738             synchronized (linkDesc) {
 739                 for (Map.Entry<ProviderId, Timestamped<LinkDescription>> e : linkDesc.entrySet()) {
 740                     linkTimestamps.put(new LinkFragmentId(linkKey, e.getKey()), e.getValue().timestamp());
 741                 }
 742             }
 743         });
 744
 745         linkTombstones.putAll(removedLinks);
 746
 747         return new LinkAntiEntropyAdvertisement(self, linkTimestamps, linkTombstones);
 748     }
 749
 750     private void handleAntiEntropyAdvertisement(LinkAntiEntropyAdvertisement ad) {
 751
 752         final NodeId sender = ad.sender();
 753         boolean localOutdated = false;
 754
 755         for (Entry<LinkKey, Map<ProviderId, Timestamped<LinkDescription>>>
 756                 l : linkDescs.entrySet()) {
 757
 758             final LinkKey key = l.getKey();
 759             final Map<ProviderId, Timestamped<LinkDescription>> link = l.getValue();
 760             synchronized (link) {
 761                 Timestamp localLatest = removedLinks.get(key);
 762
 763                 for (Entry<ProviderId, Timestamped<LinkDescription>> p : link.entrySet()) {
 764                     final ProviderId providerId = p.getKey();
 765                     final Timestamped<LinkDescription> pDesc = p.getValue();
 766
 767                     final LinkFragmentId fragId = new LinkFragmentId(key, providerId);
 768                     // remote
 769                     Timestamp remoteTimestamp = ad.linkTimestamps().get(fragId);
 770                     if (remoteTimestamp == null) {
 771                         remoteTimestamp = ad.linkTombstones().get(key);
 772                     }
 773                     if (remoteTimestamp == null ||
 774                         pDesc.isNewerThan(remoteTimestamp)) {
 775                         // I have more recent link description. update peer.
 776                         notifyPeer(sender, new InternalLinkEvent(providerId, pDesc));
 777                     } else {
 778                         final Timestamp remoteLive = ad.linkTimestamps().get(fragId);
 779                         if (remoteLive != null &&
 780                             remoteLive.compareTo(pDesc.timestamp()) > 0) {
 781                             // I have something outdated
 782                             localOutdated = true;
 783                         }
 784                     }
 785
 786                     // search local latest along the way
 787                     if (localLatest == null ||
 788                         pDesc.isNewerThan(localLatest)) {
 789                         localLatest = pDesc.timestamp();
 790                     }
 791                 }
 792                 // Tests if remote remove is more recent then local latest.
 793                 final Timestamp remoteRemove = ad.linkTombstones().get(key);
 794                 if (remoteRemove != null) {
 795                     if (localLatest != null &&
 796                         localLatest.compareTo(remoteRemove) < 0) {
 797                         // remote remove is more recent
 798                         notifyDelegateIfNotNull(removeLinkInternal(key, remoteRemove));
 799                     }
 800                 }
 801             }
 802         }
 803
 804         // populate remove info if not known locally
 805         for (Entry<LinkKey, Timestamp> remoteRm : ad.linkTombstones().entrySet()) {
 806             final LinkKey key = remoteRm.getKey();
 807             final Timestamp remoteRemove = remoteRm.getValue();
 808             // relying on removeLinkInternal to ignore stale info
 809             notifyDelegateIfNotNull(removeLinkInternal(key, remoteRemove));
 810         }
 811
 812         if (localOutdated) {
 813             // send back advertisement to speed up convergence
 814             try {
 815                 unicastMessage(sender, LINK_ANTI_ENTROPY_ADVERTISEMENT,
 816                                 createAdvertisement());
 817             } catch (IOException e) {
 818                 log.debug("Failed to send back active advertisement");
 819             }
 820         }
 821     }
 822
 823     private final class InternalLinkEventListener
 824             implements ClusterMessageHandler {
 825         @Override
 826         public void handle(ClusterMessage message) {
 827
 828             log.trace("Received link event from peer: {}", message.sender());
 829             InternalLinkEvent event = (InternalLinkEvent) SERIALIZER.decode(message.payload());
 830
 831             ProviderId providerId = event.providerId();
 832             Timestamped<LinkDescription> linkDescription = event.linkDescription();
 833
 834             try {
 835                 notifyDelegateIfNotNull(createOrUpdateLinkInternal(providerId, linkDescription));
 836             } catch (Exception e) {
 837                 log.warn("Exception thrown handling link event", e);
 838             }
 839         }
 840     }
 841
 842     private final class InternalLinkRemovedEventListener
 843             implements ClusterMessageHandler {
 844         @Override
 845         public void handle(ClusterMessage message) {
 846
 847             log.trace("Received link removed event from peer: {}", message.sender());
 848             InternalLinkRemovedEvent event = (InternalLinkRemovedEvent) SERIALIZER.decode(message.payload());
 849
 850             LinkKey linkKey = event.linkKey();
 851             Timestamp timestamp = event.timestamp();
 852
 853             try {
 854                 notifyDelegateIfNotNull(removeLinkInternal(linkKey, timestamp));
 855             } catch (Exception e) {
 856                 log.warn("Exception thrown handling link removed", e);
 857             }
 858         }
 859     }
 860
 861     private final class InternalLinkAntiEntropyAdvertisementListener
 862             implements ClusterMessageHandler {
 863
 864         @Override
 865         public void handle(ClusterMessage message) {
 866             log.trace("Received Link Anti-Entropy advertisement from peer: {}", message.sender());
 867             LinkAntiEntropyAdvertisement advertisement = SERIALIZER.decode(message.payload());
 868             try {
 869                 handleAntiEntropyAdvertisement(advertisement);
 870             } catch (Exception e) {
 871                 log.warn("Exception thrown while handling Link advertisements", e);
 872                 throw e;
 873             }
 874         }
 875     }
 876
 877     private final class LinkInjectedEventListener
 878             implements ClusterMessageHandler {
 879         @Override
 880         public void handle(ClusterMessage message) {
 881
 882             log.trace("Received injected link event from peer: {}", message.sender());
 883             LinkInjectedEvent linkInjectedEvent = SERIALIZER.decode(message.payload());
 884
 885             ProviderId providerId = linkInjectedEvent.providerId();
 886             LinkDescription linkDescription = linkInjectedEvent.linkDescription();
 887
 888             final DeviceId deviceId = linkDescription.dst().deviceId();
 889             if (!deviceClockService.isTimestampAvailable(deviceId)) {
 890                 // workaround for ONOS-1208
 891                 log.warn("Not ready to accept update. Dropping {}", linkDescription);
 892                 return;
 893             }
 894
 895             try {
 896                 createOrUpdateLink(providerId, linkDescription);
 897             } catch (Exception e) {
 898                 log.warn("Exception thrown while handling link injected event", e);
 899             }
 900         }
 901     }
 902 }