Compaction of discrete resources with range based representation

This resolves ONOS-4281

Change-Id: I0739ba94cc0b3ce617e2db44307fef396dcfb942

@@ -23,6 +23,7 @@ import org.apache.felix.scr.annotations.Reference;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.Service;
 import org.onlab.util.Tools;
+import org.onlab.util.KryoNamespace;
 import org.onosproject.net.resource.ContinuousResource;
 import org.onosproject.net.resource.ContinuousResourceId;
 import org.onosproject.net.resource.DiscreteResource;
@@ -45,7 +46,6 @@ import org.onosproject.store.service.TransactionContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -70,11 +70,14 @@ public class ConsistentResourceStore extends AbstractStore<ResourceEvent, Resour
         implements ResourceStore {
     private static final Logger log = LoggerFactory.getLogger(ConsistentResourceStore.class);
 
-    static final Serializer SERIALIZER = Serializer.using(
-            Arrays.asList(KryoNamespaces.API),
-            UnifiedDiscreteResources.class,
-            NonEncodableDiscreteResources.class,
-            ContinuousResourceAllocation.class);
+    static final Serializer SERIALIZER = Serializer.using(KryoNamespace.newBuilder()
+            .register(KryoNamespaces.API)
+            .register(UnifiedDiscreteResources.class)
+            .register(new EncodableDiscreteResourcesSerializer(), EncodableDiscreteResources.class)
+            .register(NonEncodableDiscreteResources.class)
+            .register(EmptyDiscreteResources.class)
+            .register(ContinuousResourceAllocation.class)
+            .build());
 
     // TODO: We should provide centralized values for this
     static final int MAX_RETRIES = 5;

@@ -32,7 +32,7 @@ interface DiscreteResources {
      * @return a empty set.
      */
     static DiscreteResources empty() {
-        return UnifiedDiscreteResources.empty();
+        return EmptyDiscreteResources.INSTANCE;
     }
 
     /**
@@ -41,7 +41,7 @@ interface DiscreteResources {
      * @param resources resources
      * @return instance
      */
-    static DiscreteResources of(List<DiscreteResource> resources) {
+    static DiscreteResources of(Set<DiscreteResource> resources) {
         return UnifiedDiscreteResources.of(resources);
     }
 

@@ -43,6 +43,10 @@ final class EncodableDiscreteResources implements DiscreteResources {
     }
 
     static DiscreteResources of(Set<DiscreteResource> resources) {
+        if (resources.isEmpty()) {
+            return DiscreteResources.empty();
+        }
+
         DiscreteResource parent = resources.iterator().next().parent().get();
         return of(parent, resources);
     }

@@ -30,26 +30,23 @@ import java.util.Set;
 final class NonEncodableDiscreteResources implements DiscreteResources {
     private final Set<DiscreteResource> values;
 
-    static NonEncodableDiscreteResources empty() {
-        return new NonEncodableDiscreteResources();
-    }
+    static DiscreteResources of(Set<DiscreteResource> resources) {
+        if (resources.isEmpty()) {
+            return DiscreteResources.empty();
+        }
 
-    static NonEncodableDiscreteResources of(List<DiscreteResource> resources) {
         return new NonEncodableDiscreteResources(resources);
     }
 
-    private NonEncodableDiscreteResources() {
-        this.values = new LinkedHashSet<>();
-    }
-
-    private NonEncodableDiscreteResources(List<DiscreteResource> values) {
-        this.values = new LinkedHashSet<>(values);
-    }
-
     private NonEncodableDiscreteResources(Set<DiscreteResource> values) {
         this.values = values;
     }
 
+    // for serializer
+    private NonEncodableDiscreteResources() {
+        this.values = null;
+    }
+
     @Override
     public Optional<DiscreteResource> lookup(DiscreteResourceId id) {
         DiscreteResource resource = Resources.discrete(id).resource();

@@ -24,6 +24,7 @@ import org.onosproject.store.service.TransactionalMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Optional;
 
@@ -59,7 +60,7 @@ class TransactionalDiscreteResourceSubStore {
             return true;
         }
 
-        DiscreteResources requested = DiscreteResources.of(values);
+        DiscreteResources requested = DiscreteResources.of(new LinkedHashSet<>(values));
         DiscreteResources oldValues = childMap.putIfAbsent(key, requested);
         if (oldValues == null) {
             return true;

@@ -15,12 +15,18 @@
  */
 package org.onosproject.store.resource.impl;
 
+import com.google.common.collect.Sets;
 import org.onosproject.net.resource.DiscreteResource;
 import org.onosproject.net.resource.DiscreteResourceId;
+import org.onosproject.net.resource.Resources;
 
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * Represents a set of resources containing resources that can be encoded as integer
@@ -28,55 +34,66 @@ import java.util.Set;
  */
 final class UnifiedDiscreteResources implements DiscreteResources {
     private final DiscreteResources nonEncodables;
+    private final DiscreteResources encodables;
+    private static final Codecs CODECS = Codecs.getInstance();
 
-    static DiscreteResources empty() {
-        return new UnifiedDiscreteResources();
-    }
-
-    static DiscreteResources of(List<DiscreteResource> resources) {
-        return new UnifiedDiscreteResources(resources);
-    }
+    static DiscreteResources of(Set<DiscreteResource> resources) {
+        if (resources.isEmpty()) {
+            return DiscreteResources.empty();
+        }
 
-    private UnifiedDiscreteResources() {
-        this.nonEncodables = NonEncodableDiscreteResources.empty();
+        Map<Boolean, Set<DiscreteResource>> partitioned = resources.stream()
+                .collect(Collectors.partitioningBy(CODECS::isEncodable, Collectors.toCollection(LinkedHashSet::new)));
+        return new UnifiedDiscreteResources(
+                NonEncodableDiscreteResources.of(partitioned.get(false)),
+                EncodableDiscreteResources.of(partitioned.get(true))
+        );
     }
 
-    private UnifiedDiscreteResources(List<DiscreteResource> resources) {
-        this.nonEncodables = NonEncodableDiscreteResources.of(resources);
+    private UnifiedDiscreteResources(DiscreteResources nonEncodables, DiscreteResources encodables) {
+        this.nonEncodables = nonEncodables;
+        this.encodables = encodables;
     }
 
     @Override
     public Optional<DiscreteResource> lookup(DiscreteResourceId id) {
+        if (CODECS.isEncodable(Resources.discrete(id).resource())) {
+            return encodables.lookup(id);
+        }
+
         return nonEncodables.lookup(id);
     }
 
     @Override
     public DiscreteResources difference(DiscreteResources other) {
-        return nonEncodables.difference(other);
+        return of(Sets.difference(values(), other.values()));
     }
 
     @Override
     public boolean isEmpty() {
-        return nonEncodables.isEmpty();
+        return nonEncodables.isEmpty() && encodables.isEmpty();
     }
 
     @Override
     public boolean containsAny(List<DiscreteResource> other) {
-        return nonEncodables.containsAny(other);
+        Map<Boolean, List<DiscreteResource>> partitioned = other.stream()
+                .collect(Collectors.partitioningBy(CODECS::isEncodable));
+        return nonEncodables.containsAny(partitioned.get(false)) || encodables.containsAny(partitioned.get(true));
     }
 
     @Override
     public DiscreteResources add(DiscreteResources other) {
-        return nonEncodables.add(other);
+        return of(Sets.union(this.values(), other.values()));
     }
 
     @Override
     public DiscreteResources remove(List<DiscreteResource> removed) {
-        return nonEncodables.remove(removed);
+        return of(Sets.difference(values(), new LinkedHashSet<>(removed)));
     }
 
     @Override
     public Set<DiscreteResource> values() {
-        return nonEncodables.values();
+        return Stream.concat(encodables.values().stream(), nonEncodables.values().stream())
+                .collect(Collectors.toCollection(LinkedHashSet::new));
     }
 }