GossipDeviceStore: switch inner Map to non-Concurrent

Change-Id: I3f3a36f2249c1ba36175ad81fa374f7d8ab89af1

GossipDeviceStore: switch inner Map to non-Concurrent
Change-Id: I3f3a36f2249c1ba36175ad81fa374f7d8ab89af1
Yuta HIGUCHI
Commit bf71dffe47192ff30ad7d62313dbed429758527c bf71dffe 1 parent ad861cf9
Showing 1 changed file with 66 additions and 40 deletions
core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
--- a/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
View file @bf71dff
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/device/impl/GossipDeviceStore.java
View file @bf71dff
@@ -86,14 +86,11 @@ public class GossipDeviceStore
     private final Logger log = getLogger(getClass());
-    public static final String DEVICE_NOT_FOUND = "Device with ID %s not found";
+    private static final String DEVICE_NOT_FOUND = "Device with ID %s not found";
-    // TODO: Check if inner Map can be replaced with plain Map.
     // innerMap is used to lock a Device, thus instance should never be replaced.
-
     // collection of Description given from various providers
-    private final ConcurrentMap<DeviceId,
+    private final ConcurrentMap<DeviceId, Map<ProviderId, DeviceDescriptions>>
-                            ConcurrentMap<ProviderId, DeviceDescriptions>>
                                 deviceDescs = Maps.newConcurrentMap();
     // cache of Device and Ports generated by compositing descriptions from providers
@@ -208,9 +205,9 @@ public class GossipDeviceStore
         final Timestamped<DeviceDescription> deltaDesc = new Timestamped<>(deviceDescription, newTimestamp);
         final DeviceEvent event;
         final Timestamped<DeviceDescription> mergedDesc;
-        synchronized (getDeviceDescriptions(deviceId)) {
+        synchronized (getOrCreateDeviceDescriptionsMap(deviceId)) {
             event = createOrUpdateDeviceInternal(providerId, deviceId, deltaDesc);
-            mergedDesc = getDeviceDescriptions(deviceId).get(providerId).getDeviceDesc();
+            mergedDesc = getOrCreateDeviceDescriptionsMap(deviceId).get(providerId).getDeviceDesc();
         }
         if (event != null) {
             log.info("Notifying peers of a device update topology event for providerId: {} and deviceId: {}",
@@ -230,8 +227,8 @@ public class GossipDeviceStore
                                     Timestamped<DeviceDescription> deltaDesc) {
         // Collection of DeviceDescriptions for a Device
-        ConcurrentMap<ProviderId, DeviceDescriptions> providerDescs
+        Map<ProviderId, DeviceDescriptions> providerDescs
-            = getDeviceDescriptions(deviceId);
+            = getOrCreateDeviceDescriptionsMap(deviceId);
         synchronized (providerDescs) {
             // locking per device
@@ -241,9 +238,7 @@ public class GossipDeviceStore
                 return null;
             }
-            DeviceDescriptions descs
+            DeviceDescriptions descs = getOrCreateProviderDeviceDescriptions(providerDescs, providerId, deltaDesc);
-                = createIfAbsentUnchecked(providerDescs, providerId,
-                    new InitDeviceDescs(deltaDesc));
             final Device oldDevice = devices.get(deviceId);
             final Device newDevice;
@@ -338,7 +333,7 @@ public class GossipDeviceStore
     private DeviceEvent markOfflineInternal(DeviceId deviceId, Timestamp timestamp) {
         Map<ProviderId, DeviceDescriptions> providerDescs
-            = getDeviceDescriptions(deviceId);
+            = getOrCreateDeviceDescriptionsMap(deviceId);
         // locking device
         synchronized (providerDescs) {
@@ -401,9 +396,9 @@ public class GossipDeviceStore
         final List<DeviceEvent> events;
         final Timestamped<List<PortDescription>> merged;
-        synchronized (getDeviceDescriptions(deviceId)) {
+        synchronized (getOrCreateDeviceDescriptionsMap(deviceId)) {
             events = updatePortsInternal(providerId, deviceId, timestampedInput);
-            final DeviceDescriptions descs = getDeviceDescriptions(deviceId).get(providerId);
+            final DeviceDescriptions descs = getOrCreateDeviceDescriptionsMap(deviceId).get(providerId);
             List<PortDescription> mergedList =
                     FluentIterable.from(portDescriptions)
                 .transform(new Function<PortDescription, PortDescription>() {
@@ -435,7 +430,7 @@ public class GossipDeviceStore
         Device device = devices.get(deviceId);
         checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
-        ConcurrentMap<ProviderId, DeviceDescriptions> descsMap = deviceDescs.get(deviceId);
+        Map<ProviderId, DeviceDescriptions> descsMap = deviceDescs.get(deviceId);
         checkArgument(descsMap != null, DEVICE_NOT_FOUND, deviceId);
         List<DeviceEvent> events = new ArrayList<>();
@@ -539,10 +534,34 @@ public class GossipDeviceStore
                 NewConcurrentHashMap.<PortNumber, Port>ifNeeded());
     }
-    private ConcurrentMap<ProviderId, DeviceDescriptions> getDeviceDescriptions(
+    private Map<ProviderId, DeviceDescriptions> getOrCreateDeviceDescriptionsMap(
             DeviceId deviceId) {
-        return createIfAbsentUnchecked(deviceDescs, deviceId,
+        Map<ProviderId, DeviceDescriptions> r;
-                NewConcurrentHashMap.<ProviderId, DeviceDescriptions>ifNeeded());
+        r = deviceDescs.get(deviceId);
+        if (r == null) {
+            r = new HashMap<ProviderId, DeviceDescriptions>();
+            final Map<ProviderId, DeviceDescriptions> concurrentlyAdded;
+            concurrentlyAdded = deviceDescs.putIfAbsent(deviceId, r);
+            if (concurrentlyAdded != null) {
+                r = concurrentlyAdded;
+            }
+        }
+        return r;
+    }
+
+    // Guarded by deviceDescs value (=Device lock)
+    private DeviceDescriptions getOrCreateProviderDeviceDescriptions(
+            Map<ProviderId, DeviceDescriptions> device,
+            ProviderId providerId, Timestamped<DeviceDescription> deltaDesc) {
+
+        synchronized (device) {
+            DeviceDescriptions r = device.get(providerId);
+            if (r == null) {
+                r = new DeviceDescriptions(deltaDesc);
+                device.put(providerId, r);
+            }
+            return r;
+        }
     }
     @Override
@@ -555,9 +574,9 @@ public class GossipDeviceStore
             = new Timestamped<>(portDescription, newTimestamp);
         final DeviceEvent event;
         final Timestamped<PortDescription> mergedDesc;
-        synchronized (getDeviceDescriptions(deviceId)) {
+        synchronized (getOrCreateDeviceDescriptionsMap(deviceId)) {
             event = updatePortStatusInternal(providerId, deviceId, deltaDesc);
-            mergedDesc = getDeviceDescriptions(deviceId).get(providerId)
+            mergedDesc = getOrCreateDeviceDescriptionsMap(deviceId).get(providerId)
                             .getPortDesc(portDescription.portNumber());
         }
         if (event != null) {
@@ -579,7 +598,7 @@ public class GossipDeviceStore
         Device device = devices.get(deviceId);
         checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
-        ConcurrentMap<ProviderId, DeviceDescriptions> descsMap = deviceDescs.get(deviceId);
+        Map<ProviderId, DeviceDescriptions> descsMap = deviceDescs.get(deviceId);
         checkArgument(descsMap != null, DEVICE_NOT_FOUND, deviceId);
         synchronized (descsMap) {
@@ -591,7 +610,7 @@ public class GossipDeviceStore
             DeviceDescriptions descs = descsMap.get(providerId);
             // assuming all providers must to give DeviceDescription
-            checkArgument(descs != null,
+            verify(descs != null,
                     "Device description for Device ID %s from Provider %s was not found",
                     deviceId, providerId);
@@ -661,7 +680,7 @@ public class GossipDeviceStore
     private DeviceEvent removeDeviceInternal(DeviceId deviceId,
                                              Timestamp timestamp) {
-        Map<ProviderId, DeviceDescriptions> descs = getDeviceDescriptions(deviceId);
+        Map<ProviderId, DeviceDescriptions> descs = getOrCreateDeviceDescriptionsMap(deviceId);
         synchronized (descs) {
             // accept removal request if given timestamp is newer than
             // the latest Timestamp from Primary provider
@@ -751,14 +770,14 @@ public class GossipDeviceStore
      *
      * @param device device the port is on
      * @param number port number
-     * @param providerDescs Collection of Descriptions from multiple providers
+     * @param descsMap Collection of Descriptions from multiple providers
      * @return Port instance
      */
     private Port composePort(Device device, PortNumber number,
-                ConcurrentMap<ProviderId, DeviceDescriptions> providerDescs) {
+                Map<ProviderId, DeviceDescriptions> descsMap) {
-        ProviderId primary = pickPrimaryPID(providerDescs);
+        ProviderId primary = pickPrimaryPID(descsMap);
-        DeviceDescriptions primDescs = providerDescs.get(primary);
+        DeviceDescriptions primDescs = descsMap.get(primary);
         // if no primary, assume not enabled
         // TODO: revisit this default port enabled/disabled behavior
         boolean isEnabled = false;
@@ -770,7 +789,7 @@ public class GossipDeviceStore
             annotations = merge(annotations, portDesc.value().annotations());
         }
-        for (Entry<ProviderId, DeviceDescriptions> e : providerDescs.entrySet()) {
+        for (Entry<ProviderId, DeviceDescriptions> e : descsMap.entrySet()) {
             if (e.getKey().equals(primary)) {
                 continue;
             }
@@ -893,41 +912,48 @@ public class GossipDeviceStore
     private DeviceAntiEntropyAdvertisement createAdvertisement() {
         final NodeId self = clusterService.getLocalNode().id();
-        Map<DeviceFragmentId, Timestamp> devices = new HashMap<>(deviceDescs.size());
+        final int numDevices = deviceDescs.size();
-        final int portsPerDevice = 8; // random guess to minimize reallocation
+        Map<DeviceFragmentId, Timestamp> adDevices = new HashMap<>(numDevices);
-        Map<PortFragmentId, Timestamp> ports = new HashMap<>(devices.size() * portsPerDevice);
+        final int portsPerDevice = 8; // random factor to minimize reallocation
-        Map<DeviceId, Timestamp> offline = new HashMap<>(devices.size());
+        Map<PortFragmentId, Timestamp> adPorts = new HashMap<>(numDevices * portsPerDevice);
+        Map<DeviceId, Timestamp> adOffline = new HashMap<>(numDevices);
-        for (Entry<DeviceId, ConcurrentMap<ProviderId, DeviceDescriptions>>
+        for (Entry<DeviceId, Map<ProviderId, DeviceDescriptions>>
             provs : deviceDescs.entrySet()) {
+            // for each Device...
             final DeviceId deviceId = provs.getKey();
-            final ConcurrentMap<ProviderId, DeviceDescriptions> devDescs = provs.getValue();
+            final Map<ProviderId, DeviceDescriptions> devDescs = provs.getValue();
             synchronized (devDescs) {
-                offline.put(deviceId, this.offline.get(deviceId));
+                // send device offline timestamp
+                Timestamp lOffline = this.offline.get(deviceId);
+                if (lOffline != null) {
+                    adOffline.put(deviceId, lOffline);
+                }
                 for (Entry<ProviderId, DeviceDescriptions>
                         prov : devDescs.entrySet()) {
+                    // for each Provider Descriptions...
                     final ProviderId provId = prov.getKey();
                     final DeviceDescriptions descs = prov.getValue();
-                    devices.put(new DeviceFragmentId(deviceId, provId),
+                    adDevices.put(new DeviceFragmentId(deviceId, provId),
                             descs.getDeviceDesc().timestamp());
                     for (Entry<PortNumber, Timestamped<PortDescription>>
                             portDesc : descs.getPortDescs().entrySet()) {
                         final PortNumber number = portDesc.getKey();
-                        ports.put(new PortFragmentId(deviceId, provId, number),
+                        adPorts.put(new PortFragmentId(deviceId, provId, number),
                                 portDesc.getValue().timestamp());
                     }
                 }
             }
         }
-        return new DeviceAntiEntropyAdvertisement(self, devices, ports, offline);
+        return new DeviceAntiEntropyAdvertisement(self, adDevices, adPorts, adOffline);
     }
     /**
@@ -950,7 +976,7 @@ public class GossipDeviceStore
         Collection<DeviceFragmentId> reqDevices = new ArrayList<>();
         Collection<PortFragmentId> reqPorts = new ArrayList<>();
-        for (Entry<DeviceId, ConcurrentMap<ProviderId, DeviceDescriptions>> de : deviceDescs.entrySet()) {
+        for (Entry<DeviceId, Map<ProviderId, DeviceDescriptions>> de : deviceDescs.entrySet()) {
             final DeviceId deviceId = de.getKey();
             final Map<ProviderId, DeviceDescriptions> lDevice = de.getValue();