distributed link fixes

Change-Id: Iefede001a76834599a5629d843a4325283e42711

 package org.onlab.onos.net.link;
 package org.onlab.onos.net.link;
 
 
+import com.google.common.base.MoreObjects;
 import org.onlab.onos.net.AbstractDescription;
 import org.onlab.onos.net.AbstractDescription;
 import org.onlab.onos.net.ConnectPoint;
 import org.onlab.onos.net.ConnectPoint;
 import org.onlab.onos.net.Link;
 import org.onlab.onos.net.Link;
@@ -46,4 +47,11 @@ public class DefaultLinkDescription extends AbstractDescription
         return type;
         return type;
     }
     }
 
 
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper("Link").add("src", src())
+                                .add("dst", dst())
+                                .add("type", type()).toString();
+    }
+
 }
 }

@@ -201,10 +201,10 @@ public class LinkManager
             ConnectPoint dst = linkDescription.dst();
             ConnectPoint dst = linkDescription.dst();
             // if we aren't master for the device associated with the ConnectPoint
             // if we aren't master for the device associated with the ConnectPoint
             // we probably shouldn't be doing this.
             // we probably shouldn't be doing this.
-            if ((deviceService.getRole(src.deviceId()) != MastershipRole.MASTER) ||
-                    (deviceService.getRole(dst.deviceId()) != MastershipRole.MASTER)) {
-                return;
-            }
+
+//            if (deviceService.getRole(dst.deviceId()) != MastershipRole.MASTER) {
+//                return;
+//            }
             LinkEvent event = store.createOrUpdateLink(provider().id(),
             LinkEvent event = store.createOrUpdateLink(provider().id(),
                                                        linkDescription);
                                                        linkDescription);
             if (event != null) {
             if (event != null) {
@@ -220,14 +220,8 @@ public class LinkManager
 
 
             ConnectPoint src = linkDescription.src();
             ConnectPoint src = linkDescription.src();
             ConnectPoint dst = linkDescription.dst();
             ConnectPoint dst = linkDescription.dst();
-            // if we aren't master for the device associated with the ConnectPoint
-            // we probably shouldn't be doing this.
-            if ((deviceService.getRole(src.deviceId()) != MastershipRole.MASTER) ||
-                    (deviceService.getRole(dst.deviceId()) != MastershipRole.MASTER)) {
-                return;
-            }
-            LinkEvent event = store.removeLink(linkDescription.src(),
-                                               linkDescription.dst());
+
+            LinkEvent event = store.removeLink(src, dst);
             if (event != null) {
             if (event != null) {
                 log.info("Link {} vanished", linkDescription);
                 log.info("Link {} vanished", linkDescription);
                 post(event);
                 post(event);

@@ -239,11 +239,14 @@ public class GossipLinkStore
         LinkKey key = linkKey(linkDescription.src(), linkDescription.dst());
         LinkKey key = linkKey(linkDescription.src(), linkDescription.dst());
         final LinkEvent event;
         final LinkEvent event;
         final Timestamped<LinkDescription> mergedDesc;
         final Timestamped<LinkDescription> mergedDesc;
-        synchronized (getOrCreateLinkDescriptions(key)) {
+        Map<ProviderId, Timestamped<LinkDescription>> map = getOrCreateLinkDescriptions(key);
+        synchronized (map) {
             event = createOrUpdateLinkInternal(providerId, deltaDesc);
             event = createOrUpdateLinkInternal(providerId, deltaDesc);
-            mergedDesc = getOrCreateLinkDescriptions(key).get(providerId);
+            mergedDesc = map.get(providerId);
         }
         }
 
 
+
+
         if (event != null) {
         if (event != null) {
             log.info("Notifying peers of a link update topology event from providerId: "
             log.info("Notifying peers of a link update topology event from providerId: "
                     + "{}  between src: {} and dst: {}",
                     + "{}  between src: {} and dst: {}",
@@ -252,8 +255,8 @@ public class GossipLinkStore
                 notifyPeers(new InternalLinkEvent(providerId, mergedDesc));
                 notifyPeers(new InternalLinkEvent(providerId, mergedDesc));
             } catch (IOException e) {
             } catch (IOException e) {
                 log.info("Failed to notify peers of a link update topology event from providerId: "
                 log.info("Failed to notify peers of a link update topology event from providerId: "
-                        + "{}  between src: {} and dst: {}",
-                        providerId, linkDescription.src(), linkDescription.dst());
+                                 + "{}  between src: {} and dst: {}",
+                         providerId, linkDescription.src(), linkDescription.dst());
             }
             }
         }
         }
         return event;
         return event;

@@ -288,6 +288,7 @@ public abstract class AbstractOpenFlowSwitch implements OpenFlowSwitchDriver {
             // The message wasn't really a Nicira role reply. We just
             // The message wasn't really a Nicira role reply. We just
             // dispatch it to the OFMessage listeners in this case.
             // dispatch it to the OFMessage listeners in this case.
             this.handleMessage(m);
             this.handleMessage(m);
+            return;
         }
         }
 
 
         RoleRecvStatus rrs = this.roleMan.deliverRoleReply(
         RoleRecvStatus rrs = this.roleMan.deliverRoleReply(
@@ -301,8 +302,7 @@ public abstract class AbstractOpenFlowSwitch implements OpenFlowSwitchDriver {
                 this.transitionToEqualSwitch();
                 this.transitionToEqualSwitch();
             }
             }
         } else {
         } else {
-            return;
-            //TODO: tell people that we failed.
+            this.disconnectSwitch();
         }
         }
     }
     }
 
 

@@ -71,6 +71,15 @@ public class LLDPLinkProvider extends AbstractProvider implements LinkProvider {
         providerService = providerRegistry.register(this);
         providerService = providerRegistry.register(this);
         deviceService.addListener(listener);
         deviceService.addListener(listener);
         packetSevice.addProcessor(listener, 0);
         packetSevice.addProcessor(listener, 0);
+        LinkDiscovery ld;
+        for (Device device : deviceService.getDevices()) {
+            ld = new LinkDiscovery(device, packetSevice, masterService,
+                              providerService, useBDDP);
+            discoverers.put(device.id(), ld);
+            for (Port p : deviceService.getPorts(device.id())) {
+                ld.addPort(p);
+            }
+        }
 
 
         log.info("Started");
         log.info("Started");
     }
     }
@@ -96,6 +105,10 @@ public class LLDPLinkProvider extends AbstractProvider implements LinkProvider {
             LinkDiscovery ld = null;
             LinkDiscovery ld = null;
             Device device = event.subject();
             Device device = event.subject();
             Port port = event.port();
             Port port = event.port();
+            if (device == null) {
+                log.error("Device is null.");
+                return;
+            }
             switch (event.type()) {
             switch (event.type()) {
                 case DEVICE_ADDED:
                 case DEVICE_ADDED:
                     discoverers.put(device.id(),
                     discoverers.put(device.id(),
@@ -144,6 +157,11 @@ public class LLDPLinkProvider extends AbstractProvider implements LinkProvider {
                     break;
                     break;
                 case DEVICE_UPDATED:
                 case DEVICE_UPDATED:
                 case DEVICE_MASTERSHIP_CHANGED:
                 case DEVICE_MASTERSHIP_CHANGED:
+                    if (!discoverers.containsKey(device.id())) {
+                        discoverers.put(device.id(),
+                               new LinkDiscovery(device, packetSevice, masterService,
+                                      providerService, useBDDP));
+                    }
                     break;
                     break;
                 default:
                 default:
                     log.debug("Unknown event {}", event);
                     log.debug("Unknown event {}", event);

@@ -257,10 +257,10 @@ public class LinkDiscovery implements TimerTask {
                     sendProbes(portNumber);
                     sendProbes(portNumber);
                 } else {
                 } else {
                     // Update fast and slow ports
                     // Update fast and slow ports
-                    //fastIterator.remove();
-                    //this.slowPorts.add(portNumber);
-                    //this.portProbeCount.remove(portNumber);
-                    this.portProbeCount.get(portNumber).set(0);
+                    fastIterator.remove();
+                    this.slowPorts.add(portNumber);
+                    this.portProbeCount.remove(portNumber);
+
 
 
                     ConnectPoint cp = new ConnectPoint(
                     ConnectPoint cp = new ConnectPoint(
                             device.id(),
                             device.id(),

@@ -134,7 +134,7 @@ public class LLDPLinkProviderTest {
         deviceListener.event(portEvent(DeviceEvent.Type.PORT_ADDED, DID2, port(DID2, 1, false)));
         deviceListener.event(portEvent(DeviceEvent.Type.PORT_ADDED, DID2, port(DID2, 1, false)));
 
 
 
 
-        assertNull("DPID exists",
+        assertNull("DeviceId exists",
                    provider.discoverers.get(DID2));
                    provider.discoverers.get(DID2));
     }
     }
 
 
@@ -394,7 +394,7 @@ public class LLDPLinkProviderTest {
 
 
         @Override
         @Override
         public Iterable<Device> getDevices() {
         public Iterable<Device> getDevices() {
-            return devices.values();
+            return Collections.EMPTY_LIST;
         }
         }
 
 
         @Override
         @Override

@@ -205,7 +205,7 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
             pendingFMs.put(xid, installation);
             pendingFMs.put(xid, installation);
         }
         }
         pendingFutures.put(U32.f(batch.hashCode()), installation);
         pendingFutures.put(U32.f(batch.hashCode()), installation);
-        installation.verify(batch.hashCode());
+        installation.verify(U32.f(batch.hashCode()));
         return installation;
         return installation;
     }
     }
 
 
@@ -226,7 +226,10 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
 
 
         @Override
         @Override
         public void switchRemoved(Dpid dpid) {
         public void switchRemoved(Dpid dpid) {
-            collectors.remove(dpid).stop();
+            FlowStatsCollector collector = collectors.remove(dpid);
+            if (collector != null) {
+                collector.stop();
+            }
         }
         }
 
 
         @Override
         @Override
@@ -321,7 +324,7 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
         private final List<FlowEntry> offendingFlowMods = Lists.newLinkedList();
         private final List<FlowEntry> offendingFlowMods = Lists.newLinkedList();
 
 
         private final CountDownLatch countDownLatch;
         private final CountDownLatch countDownLatch;
-        private Integer pendingXid;
+        private Long pendingXid;
         private BatchState state;
         private BatchState state;
 
 
         public InstallationFuture(Set<Dpid> sws, Map<Long, FlowRuleBatchEntry> fmXids) {
         public InstallationFuture(Set<Dpid> sws, Map<Long, FlowRuleBatchEntry> fmXids) {
@@ -391,7 +394,7 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
         }
         }
 
 
 
 
-        public void verify(Integer id) {
+        public void verify(Long id) {
             pendingXid = id;
             pendingXid = id;
             for (Dpid dpid : sws) {
             for (Dpid dpid : sws) {
                 OpenFlowSwitch sw = controller.getSwitch(dpid);
                 OpenFlowSwitch sw = controller.getSwitch(dpid);
@@ -449,7 +452,9 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
 
 
         private void cleanUp() {
         private void cleanUp() {
             if (isDone() || isCancelled()) {
             if (isDone() || isCancelled()) {
-                pendingFutures.remove(pendingXid);
+                if (pendingXid != null) {
+                    pendingFutures.remove(pendingXid);
+                }
                 for (Long xid : fms.keySet()) {
                 for (Long xid : fms.keySet()) {
                     pendingFMs.remove(xid);
                     pendingFMs.remove(xid);
                 }
                 }