initial DistributedDeviceStore
Change-Id: I8730f5c8f7706dafb245ee73d989e7a562d92187
diff --git a/core/net/pom.xml b/core/net/pom.xml
index c78643b..1e233ca 100644
--- a/core/net/pom.xml
+++ b/core/net/pom.xml
@@ -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>
diff --git a/core/net/src/test/java/org/onlab/onos/net/device/impl/DistributedDeviceManagerTest.java b/core/net/src/test/java/org/onlab/onos/net/device/impl/DistributedDeviceManagerTest.java
new file mode 100644
index 0000000..6502373
--- /dev/null
+++ b/core/net/src/test/java/org/onlab/onos/net/device/impl/DistributedDeviceManagerTest.java
@@ -0,0 +1,281 @@
+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);
+ }
+ }
+
+}
diff --git a/core/store/src/main/java/org/onlab/onos/store/device/impl/AbsentInvalidatingLoadingCache.java b/core/store/src/main/java/org/onlab/onos/store/device/impl/AbsentInvalidatingLoadingCache.java
new file mode 100644
index 0000000..df88c31
--- /dev/null
+++ b/core/store/src/main/java/org/onlab/onos/store/device/impl/AbsentInvalidatingLoadingCache.java
@@ -0,0 +1,61 @@
+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.
+}
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);
+ }
+
}