Preliminary checkin for implementing version aware topology state.

@@ -16,6 +16,7 @@ import org.onlab.onos.cluster.ClusterService;
 import org.onlab.onos.cluster.MastershipEvent;
 import org.onlab.onos.cluster.MastershipListener;
 import org.onlab.onos.cluster.MastershipService;
+import org.onlab.onos.cluster.MastershipTerm;
 import org.onlab.onos.event.AbstractListenerRegistry;
 import org.onlab.onos.event.EventDeliveryService;
 import org.onlab.onos.net.Device;
@@ -36,6 +37,7 @@ import org.onlab.onos.net.device.DeviceStoreDelegate;
 import org.onlab.onos.net.device.PortDescription;
 import org.onlab.onos.net.provider.AbstractProviderRegistry;
 import org.onlab.onos.net.provider.AbstractProviderService;
+import org.onlab.onos.store.common.ClockService;
 import org.slf4j.Logger;
 
 /**
@@ -74,6 +76,9 @@ implements DeviceService, DeviceAdminService, DeviceProviderRegistry {
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected MastershipService mastershipService;
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ClockService clockService;
+
     @Activate
     public void activate() {
         store.setDelegate(delegate);
@@ -250,6 +255,8 @@ implements DeviceService, DeviceAdminService, DeviceProviderRegistry {
         @Override
         public void event(MastershipEvent event) {
             if (event.master().equals(clusterService.getLocalNode().id())) {
+                MastershipTerm term = mastershipService.requestTermService().getMastershipTerm(event.subject());
+                clockService.setMastershipTerm(event.subject(), term);
                 applyRole(event.subject(), MastershipRole.MASTER);
             } else {
                 applyRole(event.subject(), MastershipRole.STANDBY);

+package org.onlab.onos.store.common;
+
+import org.onlab.onos.cluster.MastershipTerm;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.store.Timestamp;
+
+/**
+ * Interface for a logical clock service that vends per device timestamps.
+ */
+public interface ClockService {
+    
+    /**
+     * Returns a new timestamp for the specified deviceId.
+     * @param deviceId device identifier.
+     * @return timestamp.
+     */
+    public Timestamp getTimestamp(DeviceId deviceId);
+    
+    /**
+     * Updates the mastership term for the specified deviceId.
+     * @param deviceId device identifier.
+     * @param term mastership term.
+     */
+    public void setMastershipTerm(DeviceId deviceId, MastershipTerm term);
+}

+package org.onlab.onos.store.device.impl;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+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.cluster.MastershipTerm;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.store.Timestamp;
+import org.onlab.onos.store.common.ClockService;
+import org.onlab.onos.store.impl.OnosTimestamp;
+import org.slf4j.Logger;
+
+@Component(immediate = true)
+@Service
+public class OnosClockService implements ClockService {
+    
+    private final Logger log = getLogger(getClass());
+
+    // TODO: Implement per device ticker that is reset to 0 at the beginning of a new term.
+    private final AtomicInteger ticker = new AtomicInteger(0);
+    private ConcurrentMap<DeviceId, MastershipTerm> deviceMastershipTerms = new ConcurrentHashMap<>();
+    
+    @Activate
+    public void activate() {
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        log.info("Stopped");
+    }
+    
+    @Override
+    public Timestamp getTimestamp(DeviceId deviceId) {
+        MastershipTerm term = deviceMastershipTerms.get(deviceId);
+        if (term == null) {
+            throw new IllegalStateException("Requesting timestamp for a deviceId without mastership");
+        }
+        return new OnosTimestamp(term.termNumber(), ticker.incrementAndGet());
+    }
+
+    @Override
+    public void setMastershipTerm(DeviceId deviceId, MastershipTerm term) {
+        deviceMastershipTerms.put(deviceId, term);
+    }
+}

+package org.onlab.onos.store.device.impl;
+
+import static com.google.common.base.Predicates.notNull;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.FluentIterable;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableSet.Builder;
+
+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.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.DeviceStoreDelegate;
+import org.onlab.onos.net.device.PortDescription;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.store.Timestamp;
+import org.onlab.onos.store.common.ClockService;
+import org.onlab.onos.store.impl.AbstractDistributedStore;
+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 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 a protocol that takes into consideration
+ * the order in which device events occur.
+ */
+@Component(immediate = true)
+@Service
+public class OnosDistributedDeviceStore
+        extends AbstractDistributedStore<DeviceEvent, DeviceStoreDelegate>
+        implements DeviceStore {
+
+    private final Logger log = getLogger(getClass());
+
+    public static final String DEVICE_NOT_FOUND = "Device with ID %s not found";
+
+    private ConcurrentHashMap<DeviceId, VersionedValue<Device>> devices;
+    private ConcurrentHashMap<DeviceId, Map<PortNumber, VersionedValue<Port>>> devicePorts;
+    
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ClockService clockService;
+
+    @Override
+    @Activate
+    public void activate() {
+        super.activate();
+
+        devices = new ConcurrentHashMap<>();
+        devicePorts = new ConcurrentHashMap<>();
+
+        log.info("Started");
+    }
+    
+    @Deactivate
+    public void deactivate() {
+        log.info("Stopped");
+    }
+
+    @Override
+    public int getDeviceCount() {
+        return devices.size();
+    }
+
+    @Override
+    public Iterable<Device> getDevices() {
+        // TODO builder v.s. copyOf. Guava semms to be using copyOf?
+        // FIXME: synchronize.
+        Builder<Device> builder = ImmutableSet.builder();
+        for (VersionedValue<? extends Device> device : devices.values()) {
+            builder.add(device.entity());
+        }
+        return builder.build();
+    }
+
+    @Override
+    public Device getDevice(DeviceId deviceId) {
+        return devices.get(deviceId).entity();
+    }
+
+    @Override
+    public DeviceEvent createOrUpdateDevice(ProviderId providerId, DeviceId deviceId,
+                                            DeviceDescription deviceDescription) {
+        Timestamp now = clockService.getTimestamp(deviceId);
+        VersionedValue<Device> device = devices.get(deviceId);
+        
+        if (device == null) {
+            return createDevice(providerId, deviceId, deviceDescription, now);
+        }
+        
+        Preconditions.checkState(now.compareTo(device.timestamp()) > 0, "Existing device has a timestamp in the future!");
+
+        return updateDevice(providerId, device.entity(), deviceDescription, now);
+    }
+
+    // Creates the device and returns the appropriate event if necessary.
+    private DeviceEvent createDevice(ProviderId providerId, DeviceId deviceId,
+                                     DeviceDescription desc, Timestamp timestamp) {
+        DefaultDevice device = new DefaultDevice(providerId, deviceId, desc.type(),
+                                                 desc.manufacturer(),
+                                                 desc.hwVersion(), desc.swVersion(),
+                                                 desc.serialNumber());
+
+        devices.put(deviceId, new VersionedValue<Device>(device, true, timestamp));
+        // FIXME: broadcast a message telling peers of a device event.
+        return new DeviceEvent(DEVICE_ADDED, device, null);
+    }
+
+    // Updates the device and returns the appropriate event if necessary.
+    private DeviceEvent updateDevice(ProviderId providerId, Device device,
+                                     DeviceDescription desc, Timestamp timestamp) {
+        // We allow only certain attributes to trigger update
+        if (!Objects.equals(device.hwVersion(), desc.hwVersion()) ||
+                !Objects.equals(device.swVersion(), desc.swVersion())) {
+
+            Device updated = new DefaultDevice(providerId, device.id(),
+                                                      desc.type(),
+                                                      desc.manufacturer(),
+                                                      desc.hwVersion(),
+                                                      desc.swVersion(),
+                                                      desc.serialNumber());
+            devices.put(device.id(), new VersionedValue<Device>(updated, true, timestamp));
+            // FIXME: broadcast a message telling peers of a device event.
+            return new DeviceEvent(DeviceEvent.Type.DEVICE_UPDATED, updated, null);
+        }
+
+        // Otherwise merely attempt to change availability
+        DefaultDevice updated = new DefaultDevice(providerId, device.id(),
+                desc.type(),
+                desc.manufacturer(),
+                desc.hwVersion(),
+                desc.swVersion(),
+                desc.serialNumber());
+
+        VersionedValue<Device> oldDevice = devices.put(device.id(), new VersionedValue<Device>(updated, true, timestamp));
+        if (!oldDevice.isUp()) {
+            return new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, device, null);
+        } else {
+            return null;
+        }
+    }
+
+    @Override
+    public DeviceEvent markOffline(DeviceId deviceId) {
+        VersionedValue<Device> device = devices.get(deviceId);
+        boolean willRemove = device != null && device.isUp();
+        if (!willRemove) return null;
+        Timestamp timestamp = clockService.getTimestamp(deviceId);
+        if (replaceIfLatest(device.entity(), false, timestamp))
+        {
+            return new DeviceEvent(DEVICE_AVAILABILITY_CHANGED, device.entity(), null);
+        }
+        return null;
+    }
+    
+    // Replace existing value if its timestamp is older.
+    private synchronized boolean replaceIfLatest(Device device, boolean isUp, Timestamp timestamp)
+    {
+        VersionedValue<Device> existingValue = devices.get(device.id());
+        if (timestamp.compareTo(existingValue.timestamp()) > 0)
+        {
+            devices.put(device.id(), new VersionedValue<Device>(device, isUp, timestamp));
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public List<DeviceEvent> updatePorts(DeviceId deviceId,
+                                         List<PortDescription> portDescriptions) {
+        List<DeviceEvent> events = new ArrayList<>();
+        synchronized (this) {
+            VersionedValue<Device> device = devices.get(deviceId);
+            checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
+            Map<PortNumber, VersionedValue<Port>> ports = getPortMap(deviceId);
+            Timestamp timestamp = clockService.getTimestamp(deviceId);
+
+            // Add new ports
+            Set<PortNumber> processed = new HashSet<>();
+            for (PortDescription portDescription : portDescriptions) {
+                VersionedValue<Port> port = ports.get(portDescription.portNumber());
+                if (port == null) events.add(createPort(device, portDescription, ports, timestamp));
+                Preconditions.checkState(timestamp.compareTo(port.timestamp()) > 0, "Existing port state has a timestamp in the future!");
+                events.add(updatePort(device, port, portDescription, ports, timestamp));
+                processed.add(portDescription.portNumber());
+            }
+
+            updatePortMap(deviceId, ports);
+
+            events.addAll(pruneOldPorts(device.entity(), ports, processed));
+        }
+        return FluentIterable.from(events).filter(notNull()).toList();
+    }
+
+    // Creates a new port based on the port description adds it to the map and
+    // Returns corresponding event.
+    //@GuardedBy("this")
+    private DeviceEvent createPort(VersionedValue<Device> device, PortDescription portDescription,
+                                   Map<PortNumber, VersionedValue<Port>> ports, Timestamp timestamp) {
+        Port port = new DefaultPort(device.entity(), portDescription.portNumber(),
+                                           portDescription.isEnabled());
+        ports.put(port.number(), new VersionedValue<Port>(port, true, timestamp));
+        updatePortMap(device.entity().id(), ports);
+        return new DeviceEvent(PORT_ADDED, device.entity(), port);
+    }
+
+    // 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(VersionedValue<Device> device, VersionedValue<Port> port,
+                                   PortDescription portDescription,
+                                   Map<PortNumber, VersionedValue<Port>> ports,
+                                   Timestamp timestamp) {
+        if (port.entity().isEnabled() != portDescription.isEnabled()) {
+            VersionedValue<Port> updatedPort = new VersionedValue<Port>(
+                    new DefaultPort(device.entity(), portDescription.portNumber(),
+                                    portDescription.isEnabled()),
+                    portDescription.isEnabled(),
+                    timestamp);
+            ports.put(port.entity().number(), updatedPort);
+            updatePortMap(device.entity().id(), ports);
+            return new DeviceEvent(PORT_UPDATED, device.entity(), updatedPort.entity());
+        }
+        return null;
+    }
+
+    // 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, VersionedValue<Port>> ports,
+                                            Set<PortNumber> processed) {
+        List<DeviceEvent> events = new ArrayList<>();
+        Iterator<PortNumber> iterator = ports.keySet().iterator();
+        while (iterator.hasNext()) {
+            PortNumber portNumber = iterator.next();
+            if (!processed.contains(portNumber)) {
+                events.add(new DeviceEvent(PORT_REMOVED, device,
+                                           ports.get(portNumber).entity()));
+                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, VersionedValue<Port>> getPortMap(DeviceId deviceId) {
+        Map<PortNumber, VersionedValue<Port>> ports = devicePorts.get(deviceId);
+        if (ports == null) {
+            ports = new HashMap<>();
+            // this probably is waste of time in most cases.
+            updatePortMap(deviceId, ports);
+        }
+        return ports;
+    }
+
+    //@GuardedBy("this")
+    private void updatePortMap(DeviceId deviceId, Map<PortNumber, VersionedValue<Port>> ports) {
+        devicePorts.put(deviceId, ports);
+    }
+
+    @Override
+    public DeviceEvent updatePortStatus(DeviceId deviceId,
+                                        PortDescription portDescription) {
+        VersionedValue<Device> device = devices.get(deviceId);
+        checkArgument(device != null, DEVICE_NOT_FOUND, deviceId);
+        Map<PortNumber, VersionedValue<Port>> ports = getPortMap(deviceId);
+        VersionedValue<Port> port = ports.get(portDescription.portNumber());
+        Timestamp timestamp = clockService.getTimestamp(deviceId);
+        return updatePort(device, port, portDescription, ports, timestamp);
+    }
+
+    @Override
+    public List<Port> getPorts(DeviceId deviceId) {
+        Map<PortNumber, VersionedValue<Port>> versionedPorts = devicePorts.get(deviceId);
+        if (versionedPorts == null) return Collections.emptyList();
+        List<Port> ports = new ArrayList<Port>(); 
+        for (VersionedValue<Port> port : versionedPorts.values()) {
+            ports.add(port.entity());
+        }
+        return ports;
+    }
+
+    @Override
+    public Port getPort(DeviceId deviceId, PortNumber portNumber) {
+        Map<PortNumber, VersionedValue<Port>> ports = devicePorts.get(deviceId);
+        return ports == null ? null : ports.get(portNumber).entity();
+    }
+
+    @Override
+    public boolean isAvailable(DeviceId deviceId) {
+        return devices.get(deviceId).isUp();
+    }
+
+    @Override
+    public DeviceEvent removeDevice(DeviceId deviceId) {
+        VersionedValue<Device> previousDevice = devices.remove(deviceId);
+        return previousDevice == null ? null :
+            new DeviceEvent(DEVICE_REMOVED, previousDevice.entity(), null);
+    }
+}

+package org.onlab.onos.store.device.impl;
+
+import org.onlab.onos.store.Timestamp;
+
+/**
+ * Wrapper class for a entity that is versioned 
+ * and can either be up or down.
+ *
+ * @param <T> type of the value.
+ */
+public class VersionedValue<T> {
+    private final T entity;
+    private final Timestamp timestamp;
+    private final boolean isUp;
+    
+    public VersionedValue(T entity, boolean isUp, Timestamp timestamp) {
+        this.entity = entity;
+        this.isUp = isUp;
+        this.timestamp = timestamp;
+    }
+    
+    /**
+     * Returns the value.
+     * @return value.
+     */
+    public T entity() {
+        return entity;
+    }
+    
+    /**
+     * Tells whether the entity is up or down.
+     * @return true if up, false otherwise.
+     */
+    public boolean isUp() {
+        return isUp;
+    }
+    
+    /**
+     * Returns the timestamp (version) associated with this entity.
+     * @return timestamp.
+     */
+    public Timestamp timestamp() {
+        return timestamp;
+    }
+}

 package org.onlab.onos.store.impl;
 
-import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkArgument;
 
 import java.util.Objects;
 
-import org.onlab.onos.net.ElementId;
 import org.onlab.onos.store.Timestamp;
 
 import com.google.common.base.MoreObjects;
@@ -14,22 +12,20 @@ import com.google.common.collect.ComparisonChain;
 // If it is store specific, implement serializable interfaces?
 /**
  * Default implementation of Timestamp.
+ * TODO: Better documentation.
  */
 public final class OnosTimestamp implements Timestamp {
 
-    private final ElementId id;
     private final int termNumber;
     private final int sequenceNumber;
 
     /**
      * Default version tuple.
      *
-     * @param id identifier of the element
      * @param termNumber the mastership termNumber
      * @param sequenceNumber  the sequenceNumber number within the termNumber
      */
-    public OnosTimestamp(ElementId id, int termNumber, int sequenceNumber) {
-        this.id = checkNotNull(id);
+    public OnosTimestamp(int termNumber, int sequenceNumber) {
         this.termNumber = termNumber;
         this.sequenceNumber = sequenceNumber;
     }
@@ -38,9 +34,6 @@ public final class OnosTimestamp implements Timestamp {
     public int compareTo(Timestamp o) {
         checkArgument(o instanceof OnosTimestamp, "Must be OnosTimestamp", o);
         OnosTimestamp that = (OnosTimestamp) o;
-        checkArgument(this.id.equals(that.id),
-                "Cannot compare version for different element this:%s, that:%s",
-                    this, that);
 
         return ComparisonChain.start()
                 .compare(this.termNumber, that.termNumber)
@@ -50,7 +43,7 @@ public final class OnosTimestamp implements Timestamp {
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, termNumber, sequenceNumber);
+        return Objects.hash(termNumber, sequenceNumber);
     }
 
     @Override
@@ -62,30 +55,19 @@ public final class OnosTimestamp implements Timestamp {
             return false;
         }
         OnosTimestamp that = (OnosTimestamp) obj;
-        return Objects.equals(this.id, that.id) &&
-                Objects.equals(this.termNumber, that.termNumber) &&
+        return Objects.equals(this.termNumber, that.termNumber) &&
                 Objects.equals(this.sequenceNumber, that.sequenceNumber);
     }
 
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(getClass())
-                    .add("id", id)
                     .add("termNumber", termNumber)
                     .add("sequenceNumber", sequenceNumber)
                     .toString();
     }
 
     /**
-     * Returns the element.
-     *
-     * @return element identifier
-     */
-    public ElementId id() {
-        return id;
-    }
-
-    /**
      * Returns the termNumber.
      *
      * @return termNumber
@@ -102,4 +84,4 @@ public final class OnosTimestamp implements Timestamp {
     public int sequenceNumber() {
         return sequenceNumber;
     }
-}
+}
\ No newline at end of file

 package org.onlab.onos.store.serializers;
 
-import org.onlab.onos.net.ElementId;
 import org.onlab.onos.store.impl.OnosTimestamp;
 
 import com.esotericsoftware.kryo.Kryo;
@@ -20,18 +19,17 @@ public class OnosTimestampSerializer extends Serializer<OnosTimestamp> {
         // non-null, immutable
         super(false, true);
     }
+
     @Override
     public void write(Kryo kryo, Output output, OnosTimestamp object) {
-        kryo.writeClassAndObject(output, object.id());
         output.writeInt(object.termNumber());
         output.writeInt(object.sequenceNumber());
     }
 
     @Override
     public OnosTimestamp read(Kryo kryo, Input input, Class<OnosTimestamp> type) {
-        ElementId id = (ElementId) kryo.readClassAndObject(input);
         final int term = input.readInt();
         final int sequence = input.readInt();
-        return new OnosTimestamp(id, term, sequence);
+        return new OnosTimestamp(term, sequence);
     }
 }