initial DistributedDeviceStore

Change-Id: I8730f5c8f7706dafb245ee73d989e7a562d92187

@@ -33,6 +33,15 @@
             <groupId>org.apache.felix</groupId>
             <artifactId>org.apache.felix.scr.annotations</artifactId>
         </dependency>
+
+        <!-- TODO Consider removing store dependency.
+              Currently required for DistributedDeviceManagerTest. -->
+        <dependency>
+          <groupId>org.onlab.onos</groupId>
+          <artifactId>onos-core-store</artifactId>
+            <version>${project.version}</version>
+          <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

+package org.onlab.onos.net.device.impl;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.onlab.onos.event.Event;
+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.DefaultDeviceDescription;
+import org.onlab.onos.net.device.DefaultPortDescription;
+import org.onlab.onos.net.device.DeviceAdminService;
+import org.onlab.onos.net.device.DeviceDescription;
+import org.onlab.onos.net.device.DeviceEvent;
+import org.onlab.onos.net.device.DeviceListener;
+import org.onlab.onos.net.device.DeviceProvider;
+import org.onlab.onos.net.device.DeviceProviderRegistry;
+import org.onlab.onos.net.device.DeviceProviderService;
+import org.onlab.onos.net.device.DeviceService;
+import org.onlab.onos.net.device.PortDescription;
+import org.onlab.onos.net.provider.AbstractProvider;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.event.impl.TestEventDispatcher;
+import org.onlab.onos.store.device.impl.DistributedDeviceStore;
+
+import com.google.common.collect.Iterables;
+import com.hazelcast.config.Config;
+import com.hazelcast.core.Hazelcast;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+
+import static org.junit.Assert.*;
+import static org.onlab.onos.net.Device.Type.SWITCH;
+import static org.onlab.onos.net.DeviceId.deviceId;
+import static org.onlab.onos.net.device.DeviceEvent.Type.*;
+
+// FIXME This test is painfully slow starting up Hazelcast on each test cases,
+//       turning it off in repository for now.
+// FIXME DistributedDeviceStore should have it's own test cases.
+/**
+ * Test codifying the device service & device provider service contracts.
+ */
+@Ignore
+public class DistributedDeviceManagerTest {
+
+    private static final ProviderId PID = new ProviderId("of", "foo");
+    private static final DeviceId DID1 = deviceId("of:foo");
+    private static final DeviceId DID2 = deviceId("of:bar");
+    private static final String MFR = "whitebox";
+    private static final String HW = "1.1.x";
+    private static final String SW1 = "3.8.1";
+    private static final String SW2 = "3.9.5";
+    private static final String SN = "43311-12345";
+
+    private static final PortNumber P1 = PortNumber.portNumber(1);
+    private static final PortNumber P2 = PortNumber.portNumber(2);
+    private static final PortNumber P3 = PortNumber.portNumber(3);
+
+    private DeviceManager mgr;
+
+    protected DeviceService service;
+    protected DeviceAdminService admin;
+    protected DeviceProviderRegistry registry;
+    protected DeviceProviderService providerService;
+    protected TestProvider provider;
+    protected TestListener listener = new TestListener();
+    private DistributedDeviceStore dstore;
+
+    @Before
+    public void setUp() {
+        mgr = new DeviceManager();
+        service = mgr;
+        admin = mgr;
+        registry = mgr;
+        dstore = new DistributedDeviceStore();
+        // FIXME should be reading the hazelcast.xml
+        Config config = new Config();
+        // avoid accidentally joining other cluster
+        config.getGroupConfig().setName(UUID.randomUUID().toString());
+        // quickly form single node cluster
+        config.getNetworkConfig().getJoin().getMulticastConfig()
+            .setMulticastTimeoutSeconds(0);
+        dstore.theInstance = Hazelcast.newHazelcastInstance(config);
+        dstore.activate();
+        mgr.store = dstore;
+        mgr.eventDispatcher = new TestEventDispatcher();
+        mgr.activate();
+
+        service.addListener(listener);
+
+        provider = new TestProvider();
+        providerService = registry.register(provider);
+        assertTrue("provider should be registered",
+                   registry.getProviders().contains(provider.id()));
+    }
+
+    @After
+    public void tearDown() {
+        registry.unregister(provider);
+        assertFalse("provider should not be registered",
+                    registry.getProviders().contains(provider.id()));
+        service.removeListener(listener);
+        mgr.deactivate();
+
+        dstore.deactivate();
+        dstore.theInstance.shutdown();
+    }
+
+    private void connectDevice(DeviceId deviceId, String swVersion) {
+        DeviceDescription description =
+                new DefaultDeviceDescription(deviceId.uri(), SWITCH, MFR,
+                                             HW, swVersion, SN);
+        providerService.deviceConnected(deviceId, description);
+        assertNotNull("device should be found", service.getDevice(DID1));
+    }
+
+    @Test
+    public void deviceConnected() {
+        assertNull("device should not be found", service.getDevice(DID1));
+        connectDevice(DID1, SW1);
+        validateEvents(DEVICE_ADDED);
+
+        assertEquals("only one device expected", 1, Iterables.size(service.getDevices()));
+        Iterator<Device> it = service.getDevices().iterator();
+        assertNotNull("one device expected", it.next());
+        assertFalse("only one device expected", it.hasNext());
+
+        assertEquals("incorrect device count", 1, service.getDeviceCount());
+        assertTrue("device should be available", service.isAvailable(DID1));
+    }
+
+    @Test
+    public void deviceDisconnected() {
+        connectDevice(DID1, SW1);
+        connectDevice(DID2, SW1);
+        validateEvents(DEVICE_ADDED, DEVICE_ADDED);
+        assertTrue("device should be available", service.isAvailable(DID1));
+
+        // Disconnect
+        providerService.deviceDisconnected(DID1);
+        assertNotNull("device should not be found", service.getDevice(DID1));
+        assertFalse("device should not be available", service.isAvailable(DID1));
+        validateEvents(DEVICE_AVAILABILITY_CHANGED);
+
+        // Reconnect
+        connectDevice(DID1, SW1);
+        validateEvents(DEVICE_AVAILABILITY_CHANGED);
+
+        assertEquals("incorrect device count", 2, service.getDeviceCount());
+    }
+
+    @Test
+    public void deviceUpdated() {
+        connectDevice(DID1, SW1);
+        validateEvents(DEVICE_ADDED);
+
+        connectDevice(DID1, SW2);
+        validateEvents(DEVICE_UPDATED);
+    }
+
+    @Test
+    public void getRole() {
+        connectDevice(DID1, SW1);
+        assertEquals("incorrect role", MastershipRole.MASTER, service.getRole(DID1));
+    }
+
+    @Test
+    public void setRole() throws InterruptedException {
+        connectDevice(DID1, SW1);
+        admin.setRole(DID1, MastershipRole.STANDBY);
+        validateEvents(DEVICE_ADDED, DEVICE_MASTERSHIP_CHANGED);
+        assertEquals("incorrect role", MastershipRole.STANDBY, service.getRole(DID1));
+        assertEquals("incorrect device", DID1, provider.deviceReceived.id());
+        assertEquals("incorrect role", MastershipRole.STANDBY, provider.roleReceived);
+    }
+
+    @Test
+    public void updatePorts() {
+        connectDevice(DID1, SW1);
+        List<PortDescription> pds = new ArrayList<>();
+        pds.add(new DefaultPortDescription(P1, true));
+        pds.add(new DefaultPortDescription(P2, true));
+        pds.add(new DefaultPortDescription(P3, true));
+        providerService.updatePorts(DID1, pds);
+        validateEvents(DEVICE_ADDED, PORT_ADDED, PORT_ADDED, PORT_ADDED);
+        pds.clear();
+
+        pds.add(new DefaultPortDescription(P1, false));
+        pds.add(new DefaultPortDescription(P3, true));
+        providerService.updatePorts(DID1, pds);
+        validateEvents(PORT_UPDATED, PORT_REMOVED);
+    }
+
+    @Test
+    public void updatePortStatus() {
+        connectDevice(DID1, SW1);
+        List<PortDescription> pds = new ArrayList<>();
+        pds.add(new DefaultPortDescription(P1, true));
+        pds.add(new DefaultPortDescription(P2, true));
+        providerService.updatePorts(DID1, pds);
+        validateEvents(DEVICE_ADDED, PORT_ADDED, PORT_ADDED);
+
+        providerService.portStatusChanged(DID1, new DefaultPortDescription(P1, false));
+        validateEvents(PORT_UPDATED);
+        providerService.portStatusChanged(DID1, new DefaultPortDescription(P1, false));
+        assertTrue("no events expected", listener.events.isEmpty());
+    }
+
+    @Test
+    public void getPorts() {
+        connectDevice(DID1, SW1);
+        List<PortDescription> pds = new ArrayList<>();
+        pds.add(new DefaultPortDescription(P1, true));
+        pds.add(new DefaultPortDescription(P2, true));
+        providerService.updatePorts(DID1, pds);
+        validateEvents(DEVICE_ADDED, PORT_ADDED, PORT_ADDED);
+        assertEquals("wrong port count", 2, service.getPorts(DID1).size());
+
+        Port port = service.getPort(DID1, P1);
+        assertEquals("incorrect port", P1, service.getPort(DID1, P1).number());
+        assertEquals("incorrect state", true, service.getPort(DID1, P1).isEnabled());
+    }
+
+    @Test
+    public void removeDevice() {
+        connectDevice(DID1, SW1);
+        connectDevice(DID2, SW2);
+        assertEquals("incorrect device count", 2, service.getDeviceCount());
+        admin.removeDevice(DID1);
+        assertNull("device should not be found", service.getDevice(DID1));
+        assertNotNull("device should be found", service.getDevice(DID2));
+        assertEquals("incorrect device count", 1, service.getDeviceCount());
+
+    }
+
+    protected void validateEvents(Enum... types) {
+        int i = 0;
+        assertEquals("wrong events received", types.length, listener.events.size());
+        for (Event event : listener.events) {
+            assertEquals("incorrect event type", types[i], event.type());
+            i++;
+        }
+        listener.events.clear();
+    }
+
+
+    private class TestProvider extends AbstractProvider implements DeviceProvider {
+        private Device deviceReceived;
+        private MastershipRole roleReceived;
+
+        public TestProvider() {
+            super(PID);
+        }
+
+        @Override
+        public void triggerProbe(Device device) {
+        }
+
+        @Override
+        public void roleChanged(Device device, MastershipRole newRole) {
+            deviceReceived = device;
+            roleReceived = newRole;
+        }
+    }
+
+    private static class TestListener implements DeviceListener {
+        final List<DeviceEvent> events = new ArrayList<>();
+
+        @Override
+        public void event(DeviceEvent event) {
+            events.add(event);
+        }
+    }
+
+}

+package org.onlab.onos.store.device.impl;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+
+import com.google.common.base.Optional;
+import com.google.common.cache.ForwardingLoadingCache.SimpleForwardingLoadingCache;
+import com.google.common.cache.LoadingCache;
+
+public class AbsentInvalidatingLoadingCache<K, V> extends
+        SimpleForwardingLoadingCache<K, Optional<V>> {
+
+    public AbsentInvalidatingLoadingCache(LoadingCache<K, Optional<V>> delegate) {
+        super(delegate);
+    }
+
+    @Override
+    public Optional<V> get(K key) throws ExecutionException {
+        Optional<V> v = super.get(key);
+        if (!v.isPresent()) {
+            invalidate(key);
+        }
+        return v;
+    }
+
+    @Override
+    public Optional<V> getUnchecked(K key) {
+        Optional<V> v = super.getUnchecked(key);
+        if (!v.isPresent()) {
+            invalidate(key);
+        }
+        return v;
+    }
+
+    @Override
+    public Optional<V> apply(K key) {
+        return getUnchecked(key);
+    }
+
+    @Override
+    public Optional<V> getIfPresent(Object key) {
+        Optional<V> v = super.getIfPresent(key);
+        if (!v.isPresent()) {
+            invalidate(key);
+        }
+        return v;
+    }
+
+    @Override
+    public Optional<V> get(K key, Callable<? extends Optional<V>> valueLoader)
+            throws ExecutionException {
+
+        Optional<V> v = super.get(key, valueLoader);
+        if (!v.isPresent()) {
+            invalidate(key);
+        }
+        return v;
+    }
+
+    // TODO should we be also checking getAll, etc.
+}

 package org.onlab.onos.store.device.impl;
 
-import com.google.common.collect.ImmutableList;
+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;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
 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;
@@ -17,22 +40,25 @@ 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 java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
+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;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static org.onlab.onos.net.device.DeviceEvent.Type.*;
-import static org.slf4j.LoggerFactory.getLogger;
 
 /**
  * Manages inventory of infrastructure devices using Hazelcast-backed map.
@@ -41,18 +67,167 @@ import static org.slf4j.LoggerFactory.getLogger;
 @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 @@ public class DistributedDeviceStore implements DeviceStore {
 
     @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 @@ public class DistributedDeviceStore implements DeviceStore {
                                                  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);
     }
@@ -108,6 +307,7 @@ public class DistributedDeviceStore implements DeviceStore {
         // We allow only certain attributes to trigger update
         if (!Objects.equals(device.hwVersion(), desc.hwVersion()) ||
             !Objects.equals(device.swVersion(), desc.swVersion())) {
+
             DefaultDevice updated = new DefaultDevice(providerId, device.id(),
                                                       desc.type(),
                                                       desc.manufacturer(),
@@ -115,15 +315,15 @@ public class DistributedDeviceStore implements DeviceStore {
                                                       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 @@ public class DistributedDeviceStore implements DeviceStore {
     @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 @@ public class DistributedDeviceStore implements DeviceStore {
                                   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 @@ public class DistributedDeviceStore implements DeviceStore {
                 processed.add(portDescription.portNumber());
             }
 
+            updatePortMap(deviceId, ports);
+
             events.addAll(pruneOldPorts(device, ports, processed));
         }
         return events;
@@ -165,16 +367,19 @@ public class DistributedDeviceStore implements DeviceStore {
 
     // 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 @@ public class DistributedDeviceStore implements DeviceStore {
                     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 @@ public class DistributedDeviceStore implements DeviceStore {
 
     // 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 @@ public class DistributedDeviceStore implements DeviceStore {
                 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 @@ public class DistributedDeviceStore implements DeviceStore {
 
     @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 @@ public class DistributedDeviceStore implements DeviceStore {
         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 @@ public class DistributedDeviceStore implements DeviceStore {
     @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);
+    }
+
 }