Change DefaultRouteTable primitive to cached multimap.
Change-Id: I4b29f95894a5f761a59bebbd9aa6ea778a128f22
diff --git a/apps/route-service/app/src/main/java/org/onosproject/routeservice/store/DefaultRouteTable.java b/apps/route-service/app/src/main/java/org/onosproject/routeservice/store/DefaultRouteTable.java
index d42a05a..ad1ef854 100644
--- a/apps/route-service/app/src/main/java/org/onosproject/routeservice/store/DefaultRouteTable.java
+++ b/apps/route-service/app/src/main/java/org/onosproject/routeservice/store/DefaultRouteTable.java
@@ -16,6 +16,14 @@
package org.onosproject.routeservice.store;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.ImmutableSet;
import org.onlab.packet.IpAddress;
import org.onlab.packet.IpPrefix;
import org.onlab.util.KryoNamespace;
@@ -25,22 +33,14 @@
import org.onosproject.routeservice.RouteStoreDelegate;
import org.onosproject.routeservice.RouteTableId;
import org.onosproject.store.serializers.KryoNamespaces;
-import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.ConsistentMultimap;
import org.onosproject.store.service.DistributedPrimitive;
-import org.onosproject.store.service.MapEvent;
-import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.MultimapEvent;
+import org.onosproject.store.service.MultimapEventListener;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.StorageService;
import org.onosproject.store.service.Versioned;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
-
import static com.google.common.base.Preconditions.checkNotNull;
/**
@@ -49,7 +49,7 @@
public class DefaultRouteTable implements RouteTable {
private final RouteTableId id;
- private final ConsistentMap<IpPrefix, Set<Route>> routes;
+ private final ConsistentMultimap<IpPrefix, Route> routes;
private final RouteStoreDelegate delegate;
private final ExecutorService executor;
private final RouteTableListener listener = new RouteTableListener();
@@ -80,23 +80,21 @@
notifyExistingRoutes();
- routes.addListener(listener);
+ routes.addListener(listener, executor);
}
private void notifyExistingRoutes() {
- routes.entrySet().stream()
- .map(e -> new InternalRouteEvent(InternalRouteEvent.Type.ROUTE_ADDED,
- new RouteSet(id, e.getKey(), e.getValue().value())))
- .forEach(delegate::notify);
+ getRoutes().forEach(routeSet -> delegate.notify(
+ new InternalRouteEvent(InternalRouteEvent.Type.ROUTE_ADDED, routeSet)));
}
- private ConsistentMap<IpPrefix, Set<Route>> buildRouteMap(StorageService storageService) {
+ private ConsistentMultimap<IpPrefix, Route> buildRouteMap(StorageService storageService) {
KryoNamespace routeTableSerializer = KryoNamespace.newBuilder()
.register(KryoNamespaces.API)
.register(Route.class)
.register(Route.Source.class)
.build();
- return storageService.<IpPrefix, Set<Route>>consistentMapBuilder()
+ return storageService.<IpPrefix, Route>consistentMultimapBuilder()
.withName("onos-routes-" + id.name())
.withRelaxedReadConsistency()
.withSerializer(Serializer.using(routeTableSerializer))
@@ -122,88 +120,67 @@
@Override
public void update(Route route) {
- routes.compute(route.prefix(), (prefix, set) -> {
- if (set == null) {
- set = new HashSet<>();
- }
- set.add(route);
- return set;
- });
+ routes.put(route.prefix(), route);
}
@Override
public void remove(Route route) {
- routes.compute(route.prefix(), (prefix, set) -> {
- if (set != null) {
- set.remove(route);
- if (set.isEmpty()) {
- return null;
- }
- return set;
- }
- return null;
- });
+ routes.remove(route.prefix(), route);
}
@Override
public Collection<RouteSet> getRoutes() {
- return routes.entrySet().stream()
- .map(e -> new RouteSet(id, e.getKey(), e.getValue().value()))
- .collect(Collectors.toSet());
+ return routes.stream()
+ .map(Map.Entry::getValue)
+ .collect(Collectors.groupingBy(Route::prefix))
+ .entrySet()
+ .stream()
+ .map(entry -> new RouteSet(id, entry.getKey(), ImmutableSet.copyOf(entry.getValue())))
+ .collect(Collectors.toList());
}
@Override
public RouteSet getRoutes(IpPrefix prefix) {
- Versioned<Set<Route>> routeSet = routes.get(prefix);
+ Versioned<Collection<? extends Route>> routeSet = routes.get(prefix);
if (routeSet != null) {
- return new RouteSet(id, prefix, routeSet.value());
+ return new RouteSet(id, prefix, ImmutableSet.copyOf(routeSet.value()));
}
return null;
}
@Override
public Collection<Route> getRoutesForNextHop(IpAddress nextHop) {
- // TODO index
- return routes.values().stream()
- .flatMap(v -> v.value().stream())
- .filter(r -> r.nextHop().equals(nextHop))
- .collect(Collectors.toSet());
+ return routes.stream()
+ .map(Map.Entry::getValue)
+ .filter(r -> r.nextHop().equals(nextHop))
+ .collect(Collectors.toSet());
}
private class RouteTableListener
- implements MapEventListener<IpPrefix, Set<Route>> {
+ implements MultimapEventListener<IpPrefix, Route> {
private InternalRouteEvent createRouteEvent(
- InternalRouteEvent.Type type, MapEvent<IpPrefix, Set<Route>> event) {
- Set<Route> currentRoutes =
- (event.newValue() == null) ? Collections.emptySet() : event.newValue().value();
- return new InternalRouteEvent(type, new RouteSet(id, event.key(), currentRoutes));
+ InternalRouteEvent.Type type, MultimapEvent<IpPrefix, Route> event) {
+ Collection<? extends Route> currentRoutes = Versioned.valueOrNull(routes.get(event.key()));
+ return new InternalRouteEvent(type, new RouteSet(
+ id, event.key(), currentRoutes != null ? ImmutableSet.copyOf(currentRoutes) : Collections.emptySet()));
}
@Override
- public void event(MapEvent<IpPrefix, Set<Route>> event) {
+ public void event(MultimapEvent<IpPrefix, Route> event) {
InternalRouteEvent ire = null;
switch (event.type()) {
case INSERT:
ire = createRouteEvent(InternalRouteEvent.Type.ROUTE_ADDED, event);
break;
- case UPDATE:
- if (event.newValue().value().size() > event.oldValue().value().size()) {
- ire = createRouteEvent(InternalRouteEvent.Type.ROUTE_ADDED, event);
- } else {
- ire = createRouteEvent(InternalRouteEvent.Type.ROUTE_REMOVED, event);
- }
- break;
case REMOVE:
ire = createRouteEvent(InternalRouteEvent.Type.ROUTE_REMOVED, event);
break;
default:
break;
}
- if (ire != null) {
- delegate.notify(ire);
- }
+ delegate.notify(ire);
}
}
diff --git a/core/api/src/main/java/org/onosproject/store/service/AsyncIterable.java b/core/api/src/main/java/org/onosproject/store/service/AsyncIterable.java
index 53eb384..391efa5 100644
--- a/core/api/src/main/java/org/onosproject/store/service/AsyncIterable.java
+++ b/core/api/src/main/java/org/onosproject/store/service/AsyncIterable.java
@@ -28,5 +28,4 @@
* @return an asynchronous iterator
*/
CompletableFuture<AsyncIterator<T>> iterator();
-
}
diff --git a/core/api/src/main/java/org/onosproject/store/service/AsyncIterator.java b/core/api/src/main/java/org/onosproject/store/service/AsyncIterator.java
index 5159c6c..26c7ca7 100644
--- a/core/api/src/main/java/org/onosproject/store/service/AsyncIterator.java
+++ b/core/api/src/main/java/org/onosproject/store/service/AsyncIterator.java
@@ -35,5 +35,4 @@
* @return the next item in the iterator
*/
CompletableFuture<T> next();
-
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CachingAsyncConsistentMultimap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CachingAsyncConsistentMultimap.java
new file mode 100644
index 0000000..27cfd34
--- /dev/null
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/CachingAsyncConsistentMultimap.java
@@ -0,0 +1,180 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.store.primitives.impl;
+
+import java.util.Collection;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableSet;
+import org.onosproject.store.service.AsyncConsistentMultimap;
+import org.onosproject.store.service.MultimapEventListener;
+import org.onosproject.store.service.Versioned;
+import org.slf4j.Logger;
+
+import static org.onosproject.store.service.DistributedPrimitive.Status.INACTIVE;
+import static org.onosproject.store.service.DistributedPrimitive.Status.SUSPENDED;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Caching {@link AsyncConsistentMultimap} implementation.
+ */
+public class CachingAsyncConsistentMultimap<K, V> extends DelegatingAsyncConsistentMultimap<K, V> {
+ private static final int DEFAULT_CACHE_SIZE = 10000;
+ private final Logger log = getLogger(getClass());
+
+ private final LoadingCache<K, CompletableFuture<Versioned<Collection<? extends V>>>> cache;
+ private final MultimapEventListener<K, V> cacheUpdater;
+ private final Consumer<Status> statusListener;
+
+ /**
+ * Default constructor.
+ *
+ * @param backingMap a distributed, strongly consistent map for backing
+ */
+ public CachingAsyncConsistentMultimap(AsyncConsistentMultimap<K, V> backingMap) {
+ this(backingMap, DEFAULT_CACHE_SIZE);
+ }
+
+ /**
+ * Constructor to configure cache size.
+ *
+ * @param backingMap a distributed, strongly consistent map for backing
+ * @param cacheSize the maximum size of the cache
+ */
+ public CachingAsyncConsistentMultimap(AsyncConsistentMultimap<K, V> backingMap, int cacheSize) {
+ super(backingMap);
+ cache = CacheBuilder.newBuilder()
+ .maximumSize(cacheSize)
+ .build(CacheLoader.from(CachingAsyncConsistentMultimap.super::get));
+ cacheUpdater = event -> {
+ V oldValue = event.oldValue();
+ V newValue = event.newValue();
+ CompletableFuture<Versioned<Collection<? extends V>>> future = cache.getUnchecked(event.key());
+ switch (event.type()) {
+ case INSERT:
+ if (future.isDone()) {
+ Versioned<Collection<? extends V>> oldVersioned = future.join();
+ Versioned<Collection<? extends V>> newVersioned = new Versioned<>(
+ ImmutableSet.<V>builder().addAll(oldVersioned.value()).add(newValue).build(),
+ oldVersioned.version(),
+ oldVersioned.creationTime());
+ cache.put(event.key(), CompletableFuture.completedFuture(newVersioned));
+ } else {
+ cache.put(event.key(), future.thenApply(versioned -> new Versioned<>(
+ ImmutableSet.<V>builder().addAll(versioned.value()).add(newValue).build(),
+ versioned.version(),
+ versioned.creationTime())));
+ }
+ break;
+ case REMOVE:
+ if (future.isDone()) {
+ Versioned<Collection<? extends V>> oldVersioned = future.join();
+ cache.put(event.key(), CompletableFuture.completedFuture(new Versioned<>(oldVersioned.value()
+ .stream()
+ .filter(value -> !Objects.equals(value, oldValue))
+ .collect(Collectors.toSet()), oldVersioned.version(), oldVersioned.creationTime())));
+ } else {
+ cache.put(event.key(), future.thenApply(versioned -> new Versioned<>(versioned.value()
+ .stream()
+ .filter(value -> !Objects.equals(value, oldValue))
+ .collect(Collectors.toSet()), versioned.version(), versioned.creationTime())));
+ }
+ break;
+ default:
+ break;
+ }
+ };
+ statusListener = status -> {
+ log.debug("{} status changed to {}", this.name(), status);
+ // If the status of the underlying map is SUSPENDED or INACTIVE
+ // we can no longer guarantee that the cache will be in sync.
+ if (status == SUSPENDED || status == INACTIVE) {
+ cache.invalidateAll();
+ }
+ };
+ super.addListener(cacheUpdater);
+ super.addStatusChangeListener(statusListener);
+ }
+
+ @Override
+ public CompletableFuture<Boolean> containsKey(K key) {
+ return get(key).thenApply(value -> value != null && !value.value().isEmpty());
+ }
+
+ @Override
+ public CompletableFuture<Boolean> put(K key, V value) {
+ return super.put(key, value)
+ .whenComplete((r, e) -> cache.invalidate(key));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> remove(K key, V value) {
+ return super.remove(key, value)
+ .whenComplete((r, e) -> cache.invalidate(key));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> removeAll(K key, Collection<? extends V> values) {
+ return super.removeAll(key, values)
+ .whenComplete((r, e) -> cache.invalidate(key));
+ }
+
+ @Override
+ public CompletableFuture<Versioned<Collection<? extends V>>> removeAll(K key) {
+ return super.removeAll(key)
+ .whenComplete((r, e) -> cache.invalidate(key));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> putAll(K key, Collection<? extends V> values) {
+ return super.putAll(key, values)
+ .whenComplete((r, e) -> cache.invalidate(key));
+ }
+
+ @Override
+ public CompletableFuture<Versioned<Collection<? extends V>>> replaceValues(K key, Collection<V> values) {
+ return super.replaceValues(key, values)
+ .whenComplete((r, e) -> cache.invalidate(key));
+ }
+
+ @Override
+ public CompletableFuture<Void> clear() {
+ return super.clear()
+ .whenComplete((r, e) -> cache.invalidateAll());
+ }
+
+ @Override
+ public CompletableFuture<Versioned<Collection<? extends V>>> get(K key) {
+ return cache.getUnchecked(key)
+ .whenComplete((r, e) -> {
+ if (e != null) {
+ cache.invalidate(key);
+ }
+ });
+ }
+
+ @Override
+ public CompletableFuture<Void> destroy() {
+ super.removeStatusChangeListener(statusListener);
+ return super.destroy().thenCompose(v -> removeListener(cacheUpdater));
+ }
+}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMultimapBuilder.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMultimapBuilder.java
index f16d2c1..18ea1f1 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMultimapBuilder.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/DefaultConsistentMultimapBuilder.java
@@ -25,12 +25,12 @@
* Default {@link AsyncConsistentMultimap} builder.
*/
public class DefaultConsistentMultimapBuilder<K, V>
- extends ConsistentMultimapBuilder<K, V> {
+ extends ConsistentMultimapBuilder<K, V> {
private final DistributedPrimitiveCreator primitiveCreator;
public DefaultConsistentMultimapBuilder(
- DistributedPrimitiveCreator primitiveCreator) {
+ DistributedPrimitiveCreator primitiveCreator) {
this.primitiveCreator = primitiveCreator;
}
@@ -41,6 +41,10 @@
@Override
public ConsistentMultimap<K, V> build() {
- return buildMultimap().asMultimap();
+ AsyncConsistentMultimap<K, V> multimap = buildMultimap();
+ if (relaxedReadConsistency()) {
+ multimap = new CachingAsyncConsistentMultimap<>(multimap);
+ }
+ return multimap.asMultimap();
}
}
diff --git a/core/store/primitives/src/test/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentSetMultimapTest.java b/core/store/primitives/src/test/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentSetMultimapTest.java
index dbfd051..0247fca 100644
--- a/core/store/primitives/src/test/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentSetMultimapTest.java
+++ b/core/store/primitives/src/test/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentSetMultimapTest.java
@@ -167,7 +167,7 @@
//Check that contains key works properly for removed keys
map.containsKey(removedKey[0])
- .thenAccept(result -> assertFalse(result));
+ .thenAccept(result -> assertFalse(result)).join();
//Check that contains value works correctly for removed values
allValues.forEach(value -> {