[ONOS-6267] Support configurable Executors for primitives
- Support user-provided Executors in primitive builders
- Implement default per-partition per-primitive serial executor using a shared thread pool
- Implement Executor wrappers for all primitive types
Change-Id: I53acfb173a9b49a992a9a388983791d9735ed54a
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/FederatedDistributedPrimitiveCreator.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/FederatedDistributedPrimitiveCreator.java
index 6dd2322..0358a7c 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/FederatedDistributedPrimitiveCreator.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/FederatedDistributedPrimitiveCreator.java
@@ -39,6 +39,8 @@
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.Executor;
+import java.util.function.Supplier;
import static com.google.common.base.Preconditions.checkNotNull;
@@ -57,12 +59,13 @@
}
@Override
- public <K, V> AsyncConsistentMap<K, V> newAsyncConsistentMap(String name, Serializer serializer) {
+ public <K, V> AsyncConsistentMap<K, V> newAsyncConsistentMap(
+ String name, Serializer serializer, Supplier<Executor> executorSupplier) {
checkNotNull(name);
checkNotNull(serializer);
Map<PartitionId, AsyncConsistentMap<K, V>> maps =
Maps.transformValues(members,
- partition -> partition.newAsyncConsistentMap(name, serializer));
+ partition -> partition.newAsyncConsistentMap(name, serializer, executorSupplier));
Hasher<K> hasher = key -> {
int hashCode = Hashing.sha256().hashBytes(serializer.encode(key)).asInt();
return sortedMemberPartitionIds.get(Math.abs(hashCode) % members.size());
@@ -71,44 +74,46 @@
}
@Override
- public <V> AsyncConsistentTreeMap<V> newAsyncConsistentTreeMap(String name,
- Serializer serializer) {
- return getCreator(name).newAsyncConsistentTreeMap(name, serializer);
+ public <V> AsyncConsistentTreeMap<V> newAsyncConsistentTreeMap(
+ String name, Serializer serializer, Supplier<Executor> executorSupplier) {
+ return getCreator(name).newAsyncConsistentTreeMap(name, serializer, executorSupplier);
}
@Override
public <K, V> AsyncConsistentMultimap<K, V> newAsyncConsistentSetMultimap(
- String name, Serializer serializer) {
- return getCreator(name).newAsyncConsistentSetMultimap(name,
- serializer);
+ String name, Serializer serializer, Supplier<Executor> executorSupplier) {
+ return getCreator(name).newAsyncConsistentSetMultimap(name, serializer, executorSupplier);
}
@Override
- public <E> AsyncDistributedSet<E> newAsyncDistributedSet(String name, Serializer serializer) {
- return DistributedPrimitives.newSetFromMap(newAsyncConsistentMap(name, serializer));
+ public <E> AsyncDistributedSet<E> newAsyncDistributedSet(
+ String name, Serializer serializer, Supplier<Executor> executorSupplier) {
+ return DistributedPrimitives.newSetFromMap(newAsyncConsistentMap(name, serializer, executorSupplier));
}
@Override
- public <K> AsyncAtomicCounterMap<K> newAsyncAtomicCounterMap(String name, Serializer serializer) {
- return getCreator(name).newAsyncAtomicCounterMap(name, serializer);
+ public <K> AsyncAtomicCounterMap<K> newAsyncAtomicCounterMap(
+ String name, Serializer serializer, Supplier<Executor> executorSupplier) {
+ return getCreator(name).newAsyncAtomicCounterMap(name, serializer, executorSupplier);
}
@Override
- public AsyncAtomicCounter newAsyncCounter(String name) {
- return getCreator(name).newAsyncCounter(name);
+ public AsyncAtomicCounter newAsyncCounter(String name, Supplier<Executor> executorSupplier) {
+ return getCreator(name).newAsyncCounter(name, executorSupplier);
}
@Override
- public <V> AsyncAtomicValue<V> newAsyncAtomicValue(String name, Serializer serializer) {
- return getCreator(name).newAsyncAtomicValue(name, serializer);
+ public <V> AsyncAtomicValue<V> newAsyncAtomicValue(
+ String name, Serializer serializer, Supplier<Executor> executorSupplier) {
+ return getCreator(name).newAsyncAtomicValue(name, serializer, executorSupplier);
}
@Override
- public AsyncLeaderElector newAsyncLeaderElector(String name) {
+ public AsyncLeaderElector newAsyncLeaderElector(String name, Supplier<Executor> executorSupplier) {
checkNotNull(name);
Map<PartitionId, AsyncLeaderElector> leaderElectors =
Maps.transformValues(members,
- partition -> partition.newAsyncLeaderElector(name));
+ partition -> partition.newAsyncLeaderElector(name, executorSupplier));
Hasher<String> hasher = topic -> {
int hashCode = Hashing.sha256().hashString(topic, Charsets.UTF_8).asInt();
return sortedMemberPartitionIds.get(Math.abs(hashCode) % members.size());
@@ -117,13 +122,14 @@
}
@Override
- public <E> WorkQueue<E> newWorkQueue(String name, Serializer serializer) {
- return getCreator(name).newWorkQueue(name, serializer);
+ public <E> WorkQueue<E> newWorkQueue(String name, Serializer serializer, Supplier<Executor> executorSupplier) {
+ return getCreator(name).newWorkQueue(name, serializer, executorSupplier);
}
@Override
- public <V> AsyncDocumentTree<V> newAsyncDocumentTree(String name, Serializer serializer) {
- return getCreator(name).newAsyncDocumentTree(name, serializer);
+ public <V> AsyncDocumentTree<V> newAsyncDocumentTree(
+ String name, Serializer serializer, Supplier<Executor> executorSupplier) {
+ return getCreator(name).newAsyncDocumentTree(name, serializer, executorSupplier);
}
@Override