initial DistributedDeviceStore

Change-Id: I8730f5c8f7706dafb245ee73d989e7a562d92187
diff --git a/core/store/src/main/java/org/onlab/onos/store/device/impl/DistributedDeviceStore.java b/core/store/src/main/java/org/onlab/onos/store/device/impl/DistributedDeviceStore.java
index fb0e641..5f81aef 100644
--- a/core/store/src/main/java/org/onlab/onos/store/device/impl/DistributedDeviceStore.java
+++ b/core/store/src/main/java/org/onlab/onos/store/device/impl/DistributedDeviceStore.java
@@ -1,24 +1,16 @@
 package org.onlab.onos.store.device.impl;
 
-import com.google.common.collect.ImmutableList;
-import org.apache.felix.scr.annotations.Activate;
-import org.apache.felix.scr.annotations.Component;
-import org.apache.felix.scr.annotations.Deactivate;
-import org.apache.felix.scr.annotations.Service;
-import org.onlab.onos.net.DefaultDevice;
-import org.onlab.onos.net.DefaultPort;
-import org.onlab.onos.net.Device;
-import org.onlab.onos.net.DeviceId;
-import org.onlab.onos.net.MastershipRole;
-import org.onlab.onos.net.Port;
-import org.onlab.onos.net.PortNumber;
-import org.onlab.onos.net.device.DeviceDescription;
-import org.onlab.onos.net.device.DeviceEvent;
-import org.onlab.onos.net.device.DeviceStore;
-import org.onlab.onos.net.device.PortDescription;
-import org.onlab.onos.net.provider.ProviderId;
-import org.slf4j.Logger;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onlab.onos.net.device.DeviceEvent.Type.DEVICE_AVAILABILITY_CHANGED;
+import static org.onlab.onos.net.device.DeviceEvent.Type.DEVICE_MASTERSHIP_CHANGED;
+import static org.onlab.onos.net.device.DeviceEvent.Type.DEVICE_REMOVED;
+import static org.onlab.onos.net.device.DeviceEvent.Type.PORT_ADDED;
+import static org.onlab.onos.net.device.DeviceEvent.Type.PORT_REMOVED;
+import static org.onlab.onos.net.device.DeviceEvent.Type.PORT_UPDATED;
+import static org.slf4j.LoggerFactory.getLogger;
 
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -28,11 +20,45 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static org.onlab.onos.net.device.DeviceEvent.Type.*;
-import static org.slf4j.LoggerFactory.getLogger;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.net.DefaultDevice;
+import org.onlab.onos.net.DefaultPort;
+import org.onlab.onos.net.Device;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Element;
+import org.onlab.onos.net.MastershipRole;
+import org.onlab.onos.net.Port;
+import org.onlab.onos.net.PortNumber;
+import org.onlab.onos.net.device.DeviceDescription;
+import org.onlab.onos.net.device.DeviceEvent;
+import org.onlab.onos.net.device.DeviceStore;
+import org.onlab.onos.net.device.PortDescription;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.util.KryoPool;
+import org.slf4j.Logger;
+
+import com.google.common.base.Optional;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableSet.Builder;
+import com.hazelcast.core.EntryAdapter;
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.IMap;
+import com.hazelcast.core.ISet;
+import com.hazelcast.core.MapEvent;
+
+import de.javakaffee.kryoserializers.URISerializer;
+
 
 /**
  * Manages inventory of infrastructure devices using Hazelcast-backed map.
@@ -41,18 +67,167 @@
 @Service
 public class DistributedDeviceStore implements DeviceStore {
 
+    /**
+     * An IMap EntryListener, which reflects each remote event to cache.
+     *
+     * @param <K> IMap key type after deserialization
+     * @param <V> IMap value type after deserialization
+     */
+    public static final class RemoteEventHandler<K, V> extends
+            EntryAdapter<byte[], byte[]> {
+
+        private LoadingCache<K, Optional<V>> cache;
+
+        /**
+         * Constructor.
+         *
+         * @param cache cache to update
+         */
+        public RemoteEventHandler(
+                LoadingCache<K, Optional<V>> cache) {
+            this.cache = checkNotNull(cache);
+        }
+
+        @Override
+        public void mapCleared(MapEvent event) {
+            cache.invalidateAll();
+        }
+
+        @Override
+        public void entryUpdated(EntryEvent<byte[], byte[]> event) {
+            cache.put(POOL.<K>deserialize(event.getKey()),
+                        Optional.of(POOL.<V>deserialize(
+                                        event.getValue())));
+        }
+
+        @Override
+        public void entryRemoved(EntryEvent<byte[], byte[]> event) {
+            cache.invalidate(POOL.<DeviceId>deserialize(event.getKey()));
+        }
+
+        @Override
+        public void entryAdded(EntryEvent<byte[], byte[]> event) {
+            entryUpdated(event);
+        }
+    }
+
+    /**
+     * CacheLoader to wrap Map value with Optional,
+     * to handle negative hit on underlying IMap.
+     *
+     * @param <K> IMap key type after deserialization
+     * @param <V> IMap value type after deserialization
+     */
+    public static final class OptionalCacheLoader<K, V> extends
+            CacheLoader<K, Optional<V>> {
+
+        private IMap<byte[], byte[]> rawMap;
+
+        /**
+         * Constructor.
+         *
+         * @param rawMap underlying IMap
+         */
+        public OptionalCacheLoader(IMap<byte[], byte[]> rawMap) {
+            this.rawMap = checkNotNull(rawMap);
+        }
+
+        @Override
+        public Optional<V> load(K key) throws Exception {
+            byte[] keyBytes = serialize(key);
+            byte[] valBytes = rawMap.get(keyBytes);
+            if (valBytes == null) {
+                return Optional.absent();
+            }
+            V dev = deserialize(valBytes);
+            return Optional.of(dev);
+        }
+    }
+
     private final Logger log = getLogger(getClass());
 
     public static final String DEVICE_NOT_FOUND = "Device with ID %s not found";
 
-    private final Map<DeviceId, DefaultDevice> devices = new ConcurrentHashMap<>();
-    private final Map<DeviceId, MastershipRole> roles = new ConcurrentHashMap<>();
-    private final Set<DeviceId> availableDevices = new HashSet<>();
-    private final Map<DeviceId, Map<PortNumber, Port>> devicePorts = new HashMap<>();
+    // FIXME Slice out types used in common to separate pool/namespace.
+    private static final KryoPool POOL = KryoPool.newBuilder()
+            .register(URI.class, new URISerializer())
+            .register(
+                    ArrayList.class,
+
+                    ProviderId.class,
+                    Device.Type.class,
+
+                    DeviceId.class,
+                    DefaultDevice.class,
+                    MastershipRole.class,
+                    HashMap.class,
+                    Port.class,
+                    Element.class
+                    )
+            .register(PortNumber.class, new PortNumberSerializer())
+            .register(DefaultPort.class, new DefaultPortSerializer())
+            .build()
+            .populate(10);
+
+    // private IMap<DeviceId, DefaultDevice> cache;
+    private IMap<byte[], byte[]> rawDevices;
+    private LoadingCache<DeviceId, Optional<DefaultDevice>> devices;
+
+    // private IMap<DeviceId, MastershipRole> roles;
+    private IMap<byte[], byte[]> rawRoles;
+    private LoadingCache<DeviceId, Optional<MastershipRole>> roles;
+
+    // private ISet<DeviceId> availableDevices;
+    private ISet<byte[]> availableDevices;
+
+    // TODO DevicePorts is very inefficient consider restructuring.
+    // private IMap<DeviceId, Map<PortNumber, Port>> devicePorts;
+    private IMap<byte[], byte[]> rawDevicePorts;
+    private LoadingCache<DeviceId, Optional<Map<PortNumber, Port>>> devicePorts;
+
+    // FIXME change to protected once we remove DistributedDeviceManagerTest.
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    /*protected*/public HazelcastInstance theInstance;
+
 
     @Activate
     public void activate() {
         log.info("Started");
+
+        // IMap event handler needs value
+        final boolean includeValue = true;
+
+        // TODO decide on Map name scheme to avoid collision
+        rawDevices = theInstance.getMap("devices");
+        devices = new AbsentInvalidatingLoadingCache<DeviceId, DefaultDevice>(
+                CacheBuilder.newBuilder()
+                .build(new OptionalCacheLoader<DeviceId, DefaultDevice>(rawDevices)));
+        // refresh/populate cache based on notification from other instance
+        rawDevices.addEntryListener(
+                new RemoteEventHandler<DeviceId, DefaultDevice>(devices),
+                includeValue);
+
+        rawRoles = theInstance.getMap("roles");
+        roles = new AbsentInvalidatingLoadingCache<DeviceId, MastershipRole>(
+                CacheBuilder.newBuilder()
+                .build(new OptionalCacheLoader<DeviceId, MastershipRole>(rawRoles)));
+        // refresh/populate cache based on notification from other instance
+        rawRoles.addEntryListener(
+                new RemoteEventHandler<DeviceId, MastershipRole>(roles),
+                includeValue);
+
+        // TODO cache avai
+        availableDevices = theInstance.getSet("availableDevices");
+
+        rawDevicePorts = theInstance.getMap("devicePorts");
+        devicePorts = new AbsentInvalidatingLoadingCache<DeviceId, Map<PortNumber, Port>>(
+                CacheBuilder.newBuilder()
+                .build(new OptionalCacheLoader<DeviceId, Map<PortNumber, Port>>(rawDevicePorts)));
+        // refresh/populate cache based on notification from other instance
+        rawDevicePorts.addEntryListener(
+                new RemoteEventHandler<DeviceId, Map<PortNumber, Port>>(devicePorts),
+                includeValue);
+
     }
 
     @Deactivate
@@ -62,23 +237,42 @@
 
     @Override
     public int getDeviceCount() {
-        return devices.size();
+        // TODO IMap size or cache size?
+        return rawDevices.size();
     }
 
     @Override
     public Iterable<Device> getDevices() {
-        return Collections.unmodifiableSet(new HashSet<Device>(devices.values()));
+// TODO Revisit if we ever need to do this.
+//        log.info("{}:{}", rawMap.size(), cache.size());
+//        if (rawMap.size() != cache.size()) {
+//            for (Entry<byte[], byte[]> e : rawMap.entrySet()) {
+//                final DeviceId key = deserialize(e.getKey());
+//                final DefaultDevice val = deserialize(e.getValue());
+//                cache.put(key, val);
+//            }
+//        }
+
+        // TODO builder v.s. copyOf. Guava semms to be using copyOf?
+        Builder<Device> builder = ImmutableSet.<Device>builder();
+        for (Optional<DefaultDevice> e : devices.asMap().values()) {
+            if (e.isPresent()) {
+                builder.add(e.get());
+            }
+        }
+        return builder.build();
     }
 
     @Override
     public Device getDevice(DeviceId deviceId) {
-        return devices.get(deviceId);
+        // TODO revisit if ignoring exception is safe.
+        return devices.getUnchecked(deviceId).orNull();
     }
 
     @Override
     public DeviceEvent createOrUpdateDevice(ProviderId providerId, DeviceId deviceId,
                                      DeviceDescription deviceDescription) {
-        DefaultDevice device = devices.get(deviceId);
+        DefaultDevice device = devices.getUnchecked(deviceId).orNull();
         if (device == null) {
             return createDevice(providerId, deviceId, deviceDescription);
         }
@@ -92,12 +286,17 @@
                                                  desc.manufacturer(),
                                                  desc.hwVersion(), desc.swVersion(),
                                                  desc.serialNumber());
+
         synchronized (this) {
-            devices.put(deviceId, device);
-            availableDevices.add(deviceId);
+            final byte[] deviceIdBytes = serialize(deviceId);
+            rawDevices.put(deviceIdBytes, serialize(device));
+            devices.put(deviceId, Optional.of(device));
+
+            availableDevices.add(deviceIdBytes);
 
             // For now claim the device as a master automatically.
-            roles.put(deviceId, MastershipRole.MASTER);
+            rawRoles.put(deviceIdBytes, serialize(MastershipRole.MASTER));
+            roles.put(deviceId,  Optional.of(MastershipRole.MASTER));
         }
         return new DeviceEvent(DeviceEvent.Type.DEVICE_ADDED, device, null);
     }
@@ -107,7 +306,8 @@
                                      DeviceDescription desc) {
         // We allow only certain attributes to trigger update
         if (!Objects.equals(device.hwVersion(), desc.hwVersion()) ||
-                !Objects.equals(device.swVersion(), desc.swVersion())) {
+            !Objects.equals(device.swVersion(), desc.swVersion())) {
+
             DefaultDevice updated = new DefaultDevice(providerId, device.id(),
                                                       desc.type(),
                                                       desc.manufacturer(),
@@ -115,15 +315,15 @@
                                                       desc.swVersion(),
                                                       desc.serialNumber());
             synchronized (this) {
-                devices.put(device.id(), updated);
-                availableDevices.add(device.id());
+                devices.put(device.id(), Optional.of(updated));
+                availableDevices.add(serialize(device.id()));
             }
             return new DeviceEvent(DeviceEvent.Type.DEVICE_UPDATED, device, null);
         }
 
         // Otherwise merely attempt to change availability
         synchronized (this) {
-            boolean added = availableDevices.add(device.id());
+            boolean added = availableDevices.add(serialize(device.id()));
             return !added ? null :
                     new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, device, null);
         }
@@ -132,8 +332,8 @@
     @Override
     public DeviceEvent markOffline(DeviceId deviceId) {
         synchronized (this) {
-            Device device = devices.get(deviceId);
-            boolean removed = device != null && availableDevices.remove(deviceId);
+            Device device = devices.getUnchecked(deviceId).orNull();
+            boolean removed = device != null && availableDevices.remove(serialize(deviceId));
             return !removed ? null :
                     new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, device, null);
         }
@@ -144,7 +344,7 @@
                                   List<PortDescription> portDescriptions) {
         List<DeviceEvent> events = new ArrayList<>();
         synchronized (this) {
-            Device device = devices.get(deviceId);
+            Device device = devices.getUnchecked(deviceId).orNull();
             checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
             Map<PortNumber, Port> ports = getPortMap(deviceId);
 
@@ -158,6 +358,8 @@
                 processed.add(portDescription.portNumber());
             }
 
+            updatePortMap(deviceId, ports);
+
             events.addAll(pruneOldPorts(device, ports, processed));
         }
         return events;
@@ -165,16 +367,19 @@
 
     // Creates a new port based on the port description adds it to the map and
     // Returns corresponding event.
+    //@GuardedBy("this")
     private DeviceEvent createPort(Device device, PortDescription portDescription,
                                    Map<PortNumber, Port> ports) {
         DefaultPort port = new DefaultPort(device, portDescription.portNumber(),
                                            portDescription.isEnabled());
         ports.put(port.number(), port);
+        updatePortMap(device.id(), ports);
         return new DeviceEvent(PORT_ADDED, device, port);
     }
 
-    // CHecks if the specified port requires update and if so, it replaces the
+    // Checks if the specified port requires update and if so, it replaces the
     // existing entry in the map and returns corresponding event.
+    //@GuardedBy("this")
     private DeviceEvent updatePort(Device device, Port port,
                                    PortDescription portDescription,
                                    Map<PortNumber, Port> ports) {
@@ -183,6 +388,7 @@
                     new DefaultPort(device, portDescription.portNumber(),
                                     portDescription.isEnabled());
             ports.put(port.number(), updatedPort);
+            updatePortMap(device.id(), ports);
             return new DeviceEvent(PORT_UPDATED, device, port);
         }
         return null;
@@ -190,6 +396,7 @@
 
     // Prunes the specified list of ports based on which ports are in the
     // processed list and returns list of corresponding events.
+    //@GuardedBy("this")
     private List<DeviceEvent> pruneOldPorts(Device device,
                                             Map<PortNumber, Port> ports,
                                             Set<PortNumber> processed) {
@@ -203,25 +410,38 @@
                 iterator.remove();
             }
         }
+        if (!events.isEmpty()) {
+            updatePortMap(device.id(), ports);
+        }
         return events;
     }
 
     // Gets the map of ports for the specified device; if one does not already
     // exist, it creates and registers a new one.
+    // WARN: returned value is a copy, changes made to the Map
+    //       needs to be written back using updatePortMap
+    //@GuardedBy("this")
     private Map<PortNumber, Port> getPortMap(DeviceId deviceId) {
-        Map<PortNumber, Port> ports = devicePorts.get(deviceId);
+        Map<PortNumber, Port> ports = devicePorts.getUnchecked(deviceId).orNull();
         if (ports == null) {
             ports = new HashMap<>();
-            devicePorts.put(deviceId, ports);
+            // this probably is waste of time in most cases.
+            updatePortMap(deviceId, ports);
         }
         return ports;
     }
 
+    //@GuardedBy("this")
+    private void updatePortMap(DeviceId deviceId, Map<PortNumber, Port> ports) {
+        rawDevicePorts.put(serialize(deviceId), serialize(ports));
+        devicePorts.put(deviceId, Optional.of(ports));
+    }
+
     @Override
     public DeviceEvent updatePortStatus(DeviceId deviceId,
                                  PortDescription portDescription) {
         synchronized (this) {
-            Device device = devices.get(deviceId);
+            Device device = devices.getUnchecked(deviceId).orNull();
             checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
             Map<PortNumber, Port> ports = getPortMap(deviceId);
             Port port = ports.get(portDescription.portNumber());
@@ -231,24 +451,24 @@
 
     @Override
     public List<Port> getPorts(DeviceId deviceId) {
-        Map<PortNumber, Port> ports = devicePorts.get(deviceId);
-        return ports == null ? new ArrayList<Port>() : ImmutableList.copyOf(ports.values());
+        Map<PortNumber, Port> ports = devicePorts.getUnchecked(deviceId).orNull();
+        return ports == null ? Collections.<Port>emptyList() : ImmutableList.copyOf(ports.values());
     }
 
     @Override
     public Port getPort(DeviceId deviceId, PortNumber portNumber) {
-        Map<PortNumber, Port> ports = devicePorts.get(deviceId);
+        Map<PortNumber, Port> ports = devicePorts.getUnchecked(deviceId).orNull();
         return ports == null ? null : ports.get(portNumber);
     }
 
     @Override
     public boolean isAvailable(DeviceId deviceId) {
-        return availableDevices.contains(deviceId);
+        return availableDevices.contains(serialize(deviceId));
     }
 
     @Override
     public MastershipRole getRole(DeviceId deviceId) {
-        MastershipRole role = roles.get(deviceId);
+        MastershipRole role = roles.getUnchecked(deviceId).orNull();
         return role != null ? role : MastershipRole.NONE;
     }
 
@@ -257,7 +477,9 @@
         synchronized (this) {
             Device device = getDevice(deviceId);
             checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
-            MastershipRole oldRole = roles.put(deviceId, role);
+            MastershipRole oldRole = deserialize(
+                    rawRoles.put(serialize(deviceId), serialize(role)));
+            roles.put(deviceId, Optional.of(role));
             return oldRole == role ? null :
                     new DeviceEvent(DEVICE_MASTERSHIP_CHANGED, device, null);
         }
@@ -266,10 +488,29 @@
     @Override
     public DeviceEvent removeDevice(DeviceId deviceId) {
         synchronized (this) {
-            roles.remove(deviceId);
-            Device device = devices.remove(deviceId);
+            byte[] deviceIdBytes = serialize(deviceId);
+            rawRoles.remove(deviceIdBytes);
+            roles.invalidate(deviceId);
+
+            // TODO conditional remove?
+            Device device = deserialize(rawDevices.remove(deviceIdBytes));
+            devices.invalidate(deviceId);
             return device == null ? null :
                     new DeviceEvent(DEVICE_REMOVED, device, null);
         }
     }
+
+    // TODO cache serialized DeviceID if we suffer from serialization cost
+
+    private static byte[] serialize(final Object obj) {
+        return POOL.serialize(obj);
+    }
+
+    private static <T> T deserialize(final byte[] bytes) {
+        if (bytes == null) {
+            return null;
+        }
+        return POOL.deserialize(bytes);
+    }
+
 }