[ONOS-6267] Detect and complete blocked futures on I/O threads.
Change-Id: I0488dc5096f9e610b97405ad05c02d0ff3854b5f
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicCounter.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicCounter.java
index 5d72838..504fa75 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicCounter.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicCounter.java
@@ -18,7 +18,6 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
-import org.onlab.util.Tools;
import org.onosproject.store.service.AsyncAtomicCounter;
/**
@@ -27,46 +26,45 @@
*/
public class ExecutingAsyncAtomicCounter extends ExecutingDistributedPrimitive implements AsyncAtomicCounter {
private final AsyncAtomicCounter delegateCounter;
- private final Executor executor;
- public ExecutingAsyncAtomicCounter(AsyncAtomicCounter delegateCounter, Executor executor) {
- super(delegateCounter, executor);
+ public ExecutingAsyncAtomicCounter(
+ AsyncAtomicCounter delegateCounter, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateCounter, orderedExecutor, threadPoolExecutor);
this.delegateCounter = delegateCounter;
- this.executor = executor;
}
@Override
public CompletableFuture<Long> incrementAndGet() {
- return Tools.asyncFuture(delegateCounter.incrementAndGet(), executor);
+ return asyncFuture(delegateCounter.incrementAndGet());
}
@Override
public CompletableFuture<Long> getAndIncrement() {
- return Tools.asyncFuture(delegateCounter.getAndIncrement(), executor);
+ return asyncFuture(delegateCounter.getAndIncrement());
}
@Override
public CompletableFuture<Long> getAndAdd(long delta) {
- return Tools.asyncFuture(delegateCounter.getAndAdd(delta), executor);
+ return asyncFuture(delegateCounter.getAndAdd(delta));
}
@Override
public CompletableFuture<Long> addAndGet(long delta) {
- return Tools.asyncFuture(delegateCounter.addAndGet(delta), executor);
+ return asyncFuture(delegateCounter.addAndGet(delta));
}
@Override
public CompletableFuture<Long> get() {
- return Tools.asyncFuture(delegateCounter.get(), executor);
+ return asyncFuture(delegateCounter.get());
}
@Override
public CompletableFuture<Void> set(long value) {
- return Tools.asyncFuture(delegateCounter.set(value), executor);
+ return asyncFuture(delegateCounter.set(value));
}
@Override
public CompletableFuture<Boolean> compareAndSet(long expectedValue, long updateValue) {
- return Tools.asyncFuture(delegateCounter.compareAndSet(expectedValue, updateValue), executor);
+ return asyncFuture(delegateCounter.compareAndSet(expectedValue, updateValue));
}
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicCounterMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicCounterMap.java
index 9a039f2..a17a2f0 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicCounterMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicCounterMap.java
@@ -18,7 +18,6 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
-import org.onlab.util.Tools;
import org.onosproject.store.service.AsyncAtomicCounterMap;
/**
@@ -28,86 +27,85 @@
public class ExecutingAsyncAtomicCounterMap<K>
extends ExecutingDistributedPrimitive implements AsyncAtomicCounterMap<K> {
private final AsyncAtomicCounterMap<K> delegateMap;
- private final Executor executor;
- public ExecutingAsyncAtomicCounterMap(AsyncAtomicCounterMap<K> delegateMap, Executor executor) {
- super(delegateMap, executor);
+ public ExecutingAsyncAtomicCounterMap(
+ AsyncAtomicCounterMap<K> delegateMap, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateMap, orderedExecutor, threadPoolExecutor);
this.delegateMap = delegateMap;
- this.executor = executor;
}
@Override
public CompletableFuture<Long> incrementAndGet(K key) {
- return Tools.asyncFuture(delegateMap.incrementAndGet(key), executor);
+ return asyncFuture(delegateMap.incrementAndGet(key));
}
@Override
public CompletableFuture<Long> decrementAndGet(K key) {
- return Tools.asyncFuture(delegateMap.decrementAndGet(key), executor);
+ return asyncFuture(delegateMap.decrementAndGet(key));
}
@Override
public CompletableFuture<Long> getAndIncrement(K key) {
- return Tools.asyncFuture(delegateMap.getAndIncrement(key), executor);
+ return asyncFuture(delegateMap.getAndIncrement(key));
}
@Override
public CompletableFuture<Long> getAndDecrement(K key) {
- return Tools.asyncFuture(delegateMap.getAndDecrement(key), executor);
+ return asyncFuture(delegateMap.getAndDecrement(key));
}
@Override
public CompletableFuture<Long> addAndGet(K key, long delta) {
- return Tools.asyncFuture(delegateMap.addAndGet(key, delta), executor);
+ return asyncFuture(delegateMap.addAndGet(key, delta));
}
@Override
public CompletableFuture<Long> getAndAdd(K key, long delta) {
- return Tools.asyncFuture(delegateMap.getAndAdd(key, delta), executor);
+ return asyncFuture(delegateMap.getAndAdd(key, delta));
}
@Override
public CompletableFuture<Long> get(K key) {
- return Tools.asyncFuture(delegateMap.get(key), executor);
+ return asyncFuture(delegateMap.get(key));
}
@Override
public CompletableFuture<Long> put(K key, long newValue) {
- return Tools.asyncFuture(delegateMap.put(key, newValue), executor);
+ return asyncFuture(delegateMap.put(key, newValue));
}
@Override
public CompletableFuture<Long> putIfAbsent(K key, long newValue) {
- return Tools.asyncFuture(delegateMap.putIfAbsent(key, newValue), executor);
+ return asyncFuture(delegateMap.putIfAbsent(key, newValue));
}
@Override
public CompletableFuture<Boolean> replace(K key, long expectedOldValue, long newValue) {
- return Tools.asyncFuture(delegateMap.replace(key, expectedOldValue, newValue), executor);
+ return asyncFuture(delegateMap.replace(key, expectedOldValue, newValue));
}
@Override
public CompletableFuture<Long> remove(K key) {
- return Tools.asyncFuture(delegateMap.remove(key), executor);
+ return asyncFuture(delegateMap.remove(key));
}
@Override
public CompletableFuture<Boolean> remove(K key, long value) {
- return Tools.asyncFuture(delegateMap.remove(key, value), executor);
+ return asyncFuture(delegateMap.remove(key, value));
}
@Override
public CompletableFuture<Integer> size() {
- return Tools.asyncFuture(delegateMap.size(), executor);
+ return asyncFuture(delegateMap.size());
}
@Override
public CompletableFuture<Boolean> isEmpty() {
- return Tools.asyncFuture(delegateMap.isEmpty(), executor);
+ return asyncFuture(delegateMap.isEmpty());
}
@Override
public CompletableFuture<Void> clear() {
- return Tools.asyncFuture(delegateMap.clear(), executor);
+ return asyncFuture(delegateMap.clear());
}
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicValue.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicValue.java
index 40eacc6..c8bba52 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicValue.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncAtomicValue.java
@@ -20,7 +20,6 @@
import java.util.concurrent.Executor;
import com.google.common.collect.Maps;
-import org.onlab.util.Tools;
import org.onosproject.store.service.AsyncAtomicValue;
import org.onosproject.store.service.AtomicValueEventListener;
@@ -30,47 +29,48 @@
*/
public class ExecutingAsyncAtomicValue<V> extends ExecutingDistributedPrimitive implements AsyncAtomicValue<V> {
private final AsyncAtomicValue<V> delegateValue;
- private final Executor executor;
+ private final Executor orderedExecutor;
private final Map<AtomicValueEventListener<V>, AtomicValueEventListener<V>> listenerMap = Maps.newConcurrentMap();
- public ExecutingAsyncAtomicValue(AsyncAtomicValue<V> delegateValue, Executor executor) {
- super(delegateValue, executor);
+ public ExecutingAsyncAtomicValue(
+ AsyncAtomicValue<V> delegateValue, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateValue, orderedExecutor, threadPoolExecutor);
this.delegateValue = delegateValue;
- this.executor = executor;
+ this.orderedExecutor = orderedExecutor;
}
@Override
public CompletableFuture<Boolean> compareAndSet(V expect, V update) {
- return Tools.asyncFuture(delegateValue.compareAndSet(expect, update), executor);
+ return asyncFuture(delegateValue.compareAndSet(expect, update));
}
@Override
public CompletableFuture<V> get() {
- return Tools.asyncFuture(delegateValue.get(), executor);
+ return asyncFuture(delegateValue.get());
}
@Override
public CompletableFuture<V> getAndSet(V value) {
- return Tools.asyncFuture(delegateValue.getAndSet(value), executor);
+ return asyncFuture(delegateValue.getAndSet(value));
}
@Override
public CompletableFuture<Void> set(V value) {
- return Tools.asyncFuture(delegateValue.set(value), executor);
+ return asyncFuture(delegateValue.set(value));
}
@Override
public CompletableFuture<Void> addListener(AtomicValueEventListener<V> listener) {
- AtomicValueEventListener<V> wrappedListener = e -> executor.execute(() -> listener.event(e));
+ AtomicValueEventListener<V> wrappedListener = e -> orderedExecutor.execute(() -> listener.event(e));
listenerMap.put(listener, wrappedListener);
- return Tools.asyncFuture(delegateValue.addListener(wrappedListener), executor);
+ return asyncFuture(delegateValue.addListener(wrappedListener));
}
@Override
public CompletableFuture<Void> removeListener(AtomicValueEventListener<V> listener) {
AtomicValueEventListener<V> wrappedListener = listenerMap.remove(listener);
if (wrappedListener != null) {
- return Tools.asyncFuture(delegateValue.removeListener(wrappedListener), executor);
+ return asyncFuture(delegateValue.removeListener(wrappedListener));
}
return CompletableFuture.completedFuture(null);
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentMap.java
index 07911af..d955121 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentMap.java
@@ -23,7 +23,6 @@
import java.util.function.BiFunction;
import java.util.function.Predicate;
-import org.onlab.util.Tools;
import org.onosproject.store.primitives.MapUpdate;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.service.AsyncConsistentMap;
@@ -39,147 +38,146 @@
public class ExecutingAsyncConsistentMap<K, V>
extends ExecutingDistributedPrimitive implements AsyncConsistentMap<K, V> {
private final AsyncConsistentMap<K, V> delegateMap;
- private final Executor executor;
- public ExecutingAsyncConsistentMap(AsyncConsistentMap<K, V> delegateMap, Executor executor) {
- super(delegateMap, executor);
+ public ExecutingAsyncConsistentMap(
+ AsyncConsistentMap<K, V> delegateMap, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateMap, orderedExecutor, threadPoolExecutor);
this.delegateMap = delegateMap;
- this.executor = executor;
}
@Override
public CompletableFuture<Integer> size() {
- return Tools.asyncFuture(delegateMap.size(), executor);
+ return asyncFuture(delegateMap.size());
}
@Override
public CompletableFuture<Boolean> containsKey(K key) {
- return Tools.asyncFuture(delegateMap.containsKey(key), executor);
+ return asyncFuture(delegateMap.containsKey(key));
}
@Override
public CompletableFuture<Boolean> containsValue(V value) {
- return Tools.asyncFuture(delegateMap.containsValue(value), executor);
+ return asyncFuture(delegateMap.containsValue(value));
}
@Override
public CompletableFuture<Versioned<V>> get(K key) {
- return Tools.asyncFuture(delegateMap.get(key), executor);
+ return asyncFuture(delegateMap.get(key));
}
@Override
public CompletableFuture<Versioned<V>> getOrDefault(K key, V defaultValue) {
- return Tools.asyncFuture(delegateMap.getOrDefault(key, defaultValue), executor);
+ return asyncFuture(delegateMap.getOrDefault(key, defaultValue));
}
@Override
public CompletableFuture<Versioned<V>> computeIf(
K key, Predicate<? super V> condition, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
- return Tools.asyncFuture(delegateMap.computeIf(key, condition, remappingFunction), executor);
+ return asyncFuture(delegateMap.computeIf(key, condition, remappingFunction));
}
@Override
public CompletableFuture<Versioned<V>> put(K key, V value) {
- return Tools.asyncFuture(delegateMap.put(key, value), executor);
+ return asyncFuture(delegateMap.put(key, value));
}
@Override
public CompletableFuture<Versioned<V>> putAndGet(K key, V value) {
- return Tools.asyncFuture(delegateMap.putAndGet(key, value), executor);
+ return asyncFuture(delegateMap.putAndGet(key, value));
}
@Override
public CompletableFuture<Versioned<V>> remove(K key) {
- return Tools.asyncFuture(delegateMap.remove(key), executor);
+ return asyncFuture(delegateMap.remove(key));
}
@Override
public CompletableFuture<Void> clear() {
- return Tools.asyncFuture(delegateMap.clear(), executor);
+ return asyncFuture(delegateMap.clear());
}
@Override
public CompletableFuture<Set<K>> keySet() {
- return Tools.asyncFuture(delegateMap.keySet(), executor);
+ return asyncFuture(delegateMap.keySet());
}
@Override
public CompletableFuture<Collection<Versioned<V>>> values() {
- return Tools.asyncFuture(delegateMap.values(), executor);
+ return asyncFuture(delegateMap.values());
}
@Override
public CompletableFuture<Set<Map.Entry<K, Versioned<V>>>> entrySet() {
- return Tools.asyncFuture(delegateMap.entrySet(), executor);
+ return asyncFuture(delegateMap.entrySet());
}
@Override
public CompletableFuture<Versioned<V>> putIfAbsent(K key, V value) {
- return Tools.asyncFuture(delegateMap.putIfAbsent(key, value), executor);
+ return asyncFuture(delegateMap.putIfAbsent(key, value));
}
@Override
public CompletableFuture<Boolean> remove(K key, V value) {
- return Tools.asyncFuture(delegateMap.remove(key, value), executor);
+ return asyncFuture(delegateMap.remove(key, value));
}
@Override
public CompletableFuture<Boolean> remove(K key, long version) {
- return Tools.asyncFuture(delegateMap.remove(key, version), executor);
+ return asyncFuture(delegateMap.remove(key, version));
}
@Override
public CompletableFuture<Versioned<V>> replace(K key, V value) {
- return Tools.asyncFuture(delegateMap.replace(key, value), executor);
+ return asyncFuture(delegateMap.replace(key, value));
}
@Override
public CompletableFuture<Boolean> replace(K key, V oldValue, V newValue) {
- return Tools.asyncFuture(delegateMap.replace(key, oldValue, newValue), executor);
+ return asyncFuture(delegateMap.replace(key, oldValue, newValue));
}
@Override
public CompletableFuture<Boolean> replace(K key, long oldVersion, V newValue) {
- return Tools.asyncFuture(delegateMap.replace(key, oldVersion, newValue), executor);
+ return asyncFuture(delegateMap.replace(key, oldVersion, newValue));
}
@Override
public CompletableFuture<Version> begin(TransactionId transactionId) {
- return Tools.asyncFuture(delegateMap.begin(transactionId), executor);
+ return asyncFuture(delegateMap.begin(transactionId));
}
@Override
public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<K, V>> transactionLog) {
- return Tools.asyncFuture(delegateMap.prepare(transactionLog), executor);
+ return asyncFuture(delegateMap.prepare(transactionLog));
}
@Override
public CompletableFuture<Void> commit(TransactionId transactionId) {
- return Tools.asyncFuture(delegateMap.commit(transactionId), executor);
+ return asyncFuture(delegateMap.commit(transactionId));
}
@Override
public CompletableFuture<Void> rollback(TransactionId transactionId) {
- return Tools.asyncFuture(delegateMap.rollback(transactionId), executor);
+ return asyncFuture(delegateMap.rollback(transactionId));
}
@Override
public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<K, V>> transactionLog) {
- return Tools.asyncFuture(delegateMap.prepareAndCommit(transactionLog), executor);
+ return asyncFuture(delegateMap.prepareAndCommit(transactionLog));
}
@Override
public CompletableFuture<Void> addListener(MapEventListener<K, V> listener) {
- return addListener(listener, executor);
+ return addListener(listener);
}
@Override
public CompletableFuture<Void> addListener(MapEventListener<K, V> listener, Executor executor) {
- return Tools.asyncFuture(delegateMap.addListener(listener, executor), this.executor);
+ return asyncFuture(delegateMap.addListener(listener, executor));
}
@Override
public CompletableFuture<Void> removeListener(MapEventListener<K, V> listener) {
- return Tools.asyncFuture(delegateMap.removeListener(listener), executor);
+ return asyncFuture(delegateMap.removeListener(listener));
}
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentMultimap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentMultimap.java
index a2b308a..6ec0a69 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentMultimap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentMultimap.java
@@ -22,7 +22,6 @@
import java.util.concurrent.Executor;
import com.google.common.collect.Multiset;
-import org.onlab.util.Tools;
import org.onosproject.store.service.AsyncConsistentMultimap;
import org.onosproject.store.service.Versioned;
@@ -33,101 +32,100 @@
public class ExecutingAsyncConsistentMultimap<K, V>
extends ExecutingDistributedPrimitive implements AsyncConsistentMultimap<K, V> {
private final AsyncConsistentMultimap<K, V> delegateMap;
- private final Executor executor;
- public ExecutingAsyncConsistentMultimap(AsyncConsistentMultimap<K, V> delegateMap, Executor executor) {
- super(delegateMap, executor);
+ public ExecutingAsyncConsistentMultimap(
+ AsyncConsistentMultimap<K, V> delegateMap, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateMap, orderedExecutor, threadPoolExecutor);
this.delegateMap = delegateMap;
- this.executor = executor;
}
@Override
public CompletableFuture<Integer> size() {
- return Tools.asyncFuture(delegateMap.size(), executor);
+ return asyncFuture(delegateMap.size());
}
@Override
public CompletableFuture<Boolean> isEmpty() {
- return Tools.asyncFuture(delegateMap.isEmpty(), executor);
+ return asyncFuture(delegateMap.isEmpty());
}
@Override
public CompletableFuture<Boolean> containsKey(K key) {
- return Tools.asyncFuture(delegateMap.containsKey(key), executor);
+ return asyncFuture(delegateMap.containsKey(key));
}
@Override
public CompletableFuture<Boolean> containsValue(V value) {
- return Tools.asyncFuture(delegateMap.containsValue(value), executor);
+ return asyncFuture(delegateMap.containsValue(value));
}
@Override
public CompletableFuture<Boolean> containsEntry(K key, V value) {
- return Tools.asyncFuture(delegateMap.containsEntry(key, value), executor);
+ return asyncFuture(delegateMap.containsEntry(key, value));
}
@Override
public CompletableFuture<Boolean> put(K key, V value) {
- return Tools.asyncFuture(delegateMap.put(key, value), executor);
+ return asyncFuture(delegateMap.put(key, value));
}
@Override
public CompletableFuture<Boolean> remove(K key, V value) {
- return Tools.asyncFuture(delegateMap.remove(key, value), executor);
+ return asyncFuture(delegateMap.remove(key, value));
}
@Override
public CompletableFuture<Boolean> removeAll(K key, Collection<? extends V> values) {
- return Tools.asyncFuture(delegateMap.removeAll(key, values), executor);
+ return asyncFuture(delegateMap.removeAll(key, values));
}
@Override
public CompletableFuture<Versioned<Collection<? extends V>>> removeAll(K key) {
- return Tools.asyncFuture(delegateMap.removeAll(key), executor);
+ return asyncFuture(delegateMap.removeAll(key));
}
@Override
public CompletableFuture<Boolean> putAll(K key, Collection<? extends V> values) {
- return Tools.asyncFuture(delegateMap.putAll(key, values), executor);
+ return asyncFuture(delegateMap.putAll(key, values));
}
@Override
public CompletableFuture<Versioned<Collection<? extends V>>> replaceValues(K key, Collection<V> values) {
- return Tools.asyncFuture(delegateMap.replaceValues(key, values), executor);
+ return asyncFuture(delegateMap.replaceValues(key, values));
}
@Override
public CompletableFuture<Void> clear() {
- return Tools.asyncFuture(delegateMap.clear(), executor);
+ return asyncFuture(delegateMap.clear());
}
@Override
public CompletableFuture<Versioned<Collection<? extends V>>> get(K key) {
- return Tools.asyncFuture(delegateMap.get(key), executor);
+ return asyncFuture(delegateMap.get(key));
}
@Override
public CompletableFuture<Set<K>> keySet() {
- return Tools.asyncFuture(delegateMap.keySet(), executor);
+ return asyncFuture(delegateMap.keySet());
}
@Override
public CompletableFuture<Multiset<K>> keys() {
- return Tools.asyncFuture(delegateMap.keys(), executor);
+ return asyncFuture(delegateMap.keys());
}
@Override
public CompletableFuture<Multiset<V>> values() {
- return Tools.asyncFuture(delegateMap.values(), executor);
+ return asyncFuture(delegateMap.values());
}
@Override
public CompletableFuture<Collection<Map.Entry<K, V>>> entries() {
- return Tools.asyncFuture(delegateMap.entries(), executor);
+ return asyncFuture(delegateMap.entries());
}
@Override
public CompletableFuture<Map<K, Collection<V>>> asMap() {
- return Tools.asyncFuture(delegateMap.asMap(), executor);
+ return asyncFuture(delegateMap.asMap());
}
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentTreeMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentTreeMap.java
index 1441941..3a4fe85 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentTreeMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncConsistentTreeMap.java
@@ -25,7 +25,6 @@
import java.util.function.BiFunction;
import java.util.function.Predicate;
-import org.onlab.util.Tools;
import org.onosproject.store.primitives.MapUpdate;
import org.onosproject.store.primitives.TransactionId;
import org.onosproject.store.service.AsyncConsistentTreeMap;
@@ -41,229 +40,228 @@
public class ExecutingAsyncConsistentTreeMap<V>
extends ExecutingDistributedPrimitive implements AsyncConsistentTreeMap<V> {
private final AsyncConsistentTreeMap<V> delegateMap;
- private final Executor executor;
- public ExecutingAsyncConsistentTreeMap(AsyncConsistentTreeMap<V> delegateMap, Executor executor) {
- super(delegateMap, executor);
+ public ExecutingAsyncConsistentTreeMap(
+ AsyncConsistentTreeMap<V> delegateMap, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateMap, orderedExecutor, threadPoolExecutor);
this.delegateMap = delegateMap;
- this.executor = executor;
}
@Override
public CompletableFuture<String> firstKey() {
- return Tools.asyncFuture(delegateMap.firstKey(), executor);
+ return asyncFuture(delegateMap.firstKey());
}
@Override
public CompletableFuture<String> lastKey() {
- return Tools.asyncFuture(delegateMap.lastKey(), executor);
+ return asyncFuture(delegateMap.lastKey());
}
@Override
public CompletableFuture<Map.Entry<String, Versioned<V>>> ceilingEntry(String key) {
- return Tools.asyncFuture(delegateMap.ceilingEntry(key), executor);
+ return asyncFuture(delegateMap.ceilingEntry(key));
}
@Override
public CompletableFuture<Map.Entry<String, Versioned<V>>> floorEntry(String key) {
- return Tools.asyncFuture(delegateMap.floorEntry(key), executor);
+ return asyncFuture(delegateMap.floorEntry(key));
}
@Override
public CompletableFuture<Map.Entry<String, Versioned<V>>> higherEntry(String key) {
- return Tools.asyncFuture(delegateMap.higherEntry(key), executor);
+ return asyncFuture(delegateMap.higherEntry(key));
}
@Override
public CompletableFuture<Map.Entry<String, Versioned<V>>> lowerEntry(String key) {
- return Tools.asyncFuture(delegateMap.lowerEntry(key), executor);
+ return asyncFuture(delegateMap.lowerEntry(key));
}
@Override
public CompletableFuture<Map.Entry<String, Versioned<V>>> firstEntry() {
- return Tools.asyncFuture(delegateMap.firstEntry(), executor);
+ return asyncFuture(delegateMap.firstEntry());
}
@Override
public CompletableFuture<Integer> size() {
- return Tools.asyncFuture(delegateMap.size(), executor);
+ return asyncFuture(delegateMap.size());
}
@Override
public CompletableFuture<Map.Entry<String, Versioned<V>>> lastEntry() {
- return Tools.asyncFuture(delegateMap.lastEntry(), executor);
+ return asyncFuture(delegateMap.lastEntry());
}
@Override
public CompletableFuture<Map.Entry<String, Versioned<V>>> pollFirstEntry() {
- return Tools.asyncFuture(delegateMap.pollFirstEntry(), executor);
+ return asyncFuture(delegateMap.pollFirstEntry());
}
@Override
public CompletableFuture<Boolean> containsKey(String key) {
- return Tools.asyncFuture(delegateMap.containsKey(key), executor);
+ return asyncFuture(delegateMap.containsKey(key));
}
@Override
public CompletableFuture<Map.Entry<String, Versioned<V>>> pollLastEntry() {
- return Tools.asyncFuture(delegateMap.pollLastEntry(), executor);
+ return asyncFuture(delegateMap.pollLastEntry());
}
@Override
public CompletableFuture<String> lowerKey(String key) {
- return Tools.asyncFuture(delegateMap.lowerKey(key), executor);
+ return asyncFuture(delegateMap.lowerKey(key));
}
@Override
public CompletableFuture<Boolean> containsValue(V value) {
- return Tools.asyncFuture(delegateMap.containsValue(value), executor);
+ return asyncFuture(delegateMap.containsValue(value));
}
@Override
public CompletableFuture<String> floorKey(String key) {
- return Tools.asyncFuture(delegateMap.floorKey(key), executor);
+ return asyncFuture(delegateMap.floorKey(key));
}
@Override
public CompletableFuture<String> ceilingKey(String key) {
- return Tools.asyncFuture(delegateMap.ceilingKey(key), executor);
+ return asyncFuture(delegateMap.ceilingKey(key));
}
@Override
public CompletableFuture<Versioned<V>> get(String key) {
- return Tools.asyncFuture(delegateMap.get(key), executor);
+ return asyncFuture(delegateMap.get(key));
}
@Override
public CompletableFuture<Versioned<V>> getOrDefault(String key, V defaultValue) {
- return Tools.asyncFuture(delegateMap.getOrDefault(key, defaultValue), executor);
+ return asyncFuture(delegateMap.getOrDefault(key, defaultValue));
}
@Override
public CompletableFuture<String> higherKey(String key) {
- return Tools.asyncFuture(delegateMap.higherKey(key), executor);
+ return asyncFuture(delegateMap.higherKey(key));
}
@Override
public CompletableFuture<NavigableSet<String>> navigableKeySet() {
- return Tools.asyncFuture(delegateMap.navigableKeySet(), executor);
+ return asyncFuture(delegateMap.navigableKeySet());
}
@Override
public CompletableFuture<NavigableMap<String, V>> subMap(
String upperKey, String lowerKey, boolean inclusiveUpper, boolean inclusiveLower) {
- return Tools.asyncFuture(delegateMap.subMap(upperKey, lowerKey, inclusiveUpper, inclusiveLower), executor);
+ return asyncFuture(delegateMap.subMap(upperKey, lowerKey, inclusiveUpper, inclusiveLower));
}
@Override
public CompletableFuture<Versioned<V>> computeIf(
String key, Predicate<? super V> condition,
BiFunction<? super String, ? super V, ? extends V> remappingFunction) {
- return Tools.asyncFuture(delegateMap.computeIf(key, condition, remappingFunction), executor);
+ return asyncFuture(delegateMap.computeIf(key, condition, remappingFunction));
}
@Override
public CompletableFuture<Versioned<V>> put(String key, V value) {
- return Tools.asyncFuture(delegateMap.put(key, value), executor);
+ return asyncFuture(delegateMap.put(key, value));
}
@Override
public CompletableFuture<Versioned<V>> putAndGet(String key, V value) {
- return Tools.asyncFuture(delegateMap.putAndGet(key, value), executor);
+ return asyncFuture(delegateMap.putAndGet(key, value));
}
@Override
public CompletableFuture<Versioned<V>> remove(String key) {
- return Tools.asyncFuture(delegateMap.remove(key), executor);
+ return asyncFuture(delegateMap.remove(key));
}
@Override
public CompletableFuture<Void> clear() {
- return Tools.asyncFuture(delegateMap.clear(), executor);
+ return asyncFuture(delegateMap.clear());
}
@Override
public CompletableFuture<Set<String>> keySet() {
- return Tools.asyncFuture(delegateMap.keySet(), executor);
+ return asyncFuture(delegateMap.keySet());
}
@Override
public CompletableFuture<Collection<Versioned<V>>> values() {
- return Tools.asyncFuture(delegateMap.values(), executor);
+ return asyncFuture(delegateMap.values());
}
@Override
public CompletableFuture<Set<Map.Entry<String, Versioned<V>>>> entrySet() {
- return Tools.asyncFuture(delegateMap.entrySet(), executor);
+ return asyncFuture(delegateMap.entrySet());
}
@Override
public CompletableFuture<Versioned<V>> putIfAbsent(String key, V value) {
- return Tools.asyncFuture(delegateMap.putIfAbsent(key, value), executor);
+ return asyncFuture(delegateMap.putIfAbsent(key, value));
}
@Override
public CompletableFuture<Boolean> remove(String key, V value) {
- return Tools.asyncFuture(delegateMap.remove(key, value), executor);
+ return asyncFuture(delegateMap.remove(key, value));
}
@Override
public CompletableFuture<Boolean> remove(String key, long version) {
- return Tools.asyncFuture(delegateMap.remove(key, version), executor);
+ return asyncFuture(delegateMap.remove(key, version));
}
@Override
public CompletableFuture<Versioned<V>> replace(String key, V value) {
- return Tools.asyncFuture(delegateMap.replace(key, value), executor);
+ return asyncFuture(delegateMap.replace(key, value));
}
@Override
public CompletableFuture<Boolean> replace(String key, V oldValue, V newValue) {
- return Tools.asyncFuture(delegateMap.replace(key, oldValue, newValue), executor);
+ return asyncFuture(delegateMap.replace(key, oldValue, newValue));
}
@Override
public CompletableFuture<Boolean> replace(String key, long oldVersion, V newValue) {
- return Tools.asyncFuture(delegateMap.replace(key, oldVersion, newValue), executor);
+ return asyncFuture(delegateMap.replace(key, oldVersion, newValue));
}
@Override
public CompletableFuture<Version> begin(TransactionId transactionId) {
- return Tools.asyncFuture(delegateMap.begin(transactionId), executor);
+ return asyncFuture(delegateMap.begin(transactionId));
}
@Override
public CompletableFuture<Boolean> prepare(TransactionLog<MapUpdate<String, V>> transactionLog) {
- return Tools.asyncFuture(delegateMap.prepare(transactionLog), executor);
+ return asyncFuture(delegateMap.prepare(transactionLog));
}
@Override
public CompletableFuture<Void> commit(TransactionId transactionId) {
- return Tools.asyncFuture(delegateMap.commit(transactionId), executor);
+ return asyncFuture(delegateMap.commit(transactionId));
}
@Override
public CompletableFuture<Void> rollback(TransactionId transactionId) {
- return Tools.asyncFuture(delegateMap.rollback(transactionId), executor);
+ return asyncFuture(delegateMap.rollback(transactionId));
}
@Override
public CompletableFuture<Boolean> prepareAndCommit(TransactionLog<MapUpdate<String, V>> transactionLog) {
- return Tools.asyncFuture(delegateMap.prepareAndCommit(transactionLog), executor);
+ return asyncFuture(delegateMap.prepareAndCommit(transactionLog));
}
@Override
public CompletableFuture<Void> addListener(MapEventListener<String, V> listener) {
- return addListener(listener, executor);
+ return addListener(listener);
}
@Override
public CompletableFuture<Void> addListener(MapEventListener<String, V> listener, Executor executor) {
- return Tools.asyncFuture(delegateMap.addListener(listener, executor), this.executor);
+ return asyncFuture(delegateMap.addListener(listener, executor));
}
@Override
public CompletableFuture<Void> removeListener(MapEventListener<String, V> listener) {
- return Tools.asyncFuture(delegateMap.removeListener(listener), executor);
+ return asyncFuture(delegateMap.removeListener(listener));
}
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncDocumentTree.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncDocumentTree.java
index 352ee7a..f6fc3d1 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncDocumentTree.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncDocumentTree.java
@@ -20,7 +20,6 @@
import java.util.concurrent.Executor;
import com.google.common.collect.Maps;
-import org.onlab.util.Tools;
import org.onosproject.store.service.AsyncDocumentTree;
import org.onosproject.store.service.DocumentPath;
import org.onosproject.store.service.DocumentTreeListener;
@@ -32,13 +31,14 @@
*/
public class ExecutingAsyncDocumentTree<V> extends ExecutingDistributedPrimitive implements AsyncDocumentTree<V> {
private final AsyncDocumentTree<V> delegateTree;
- private final Executor executor;
+ private final Executor orderedExecutor;
private final Map<DocumentTreeListener<V>, DocumentTreeListener<V>> listenerMap = Maps.newConcurrentMap();
- public ExecutingAsyncDocumentTree(AsyncDocumentTree<V> delegateTree, Executor executor) {
- super(delegateTree, executor);
+ public ExecutingAsyncDocumentTree(
+ AsyncDocumentTree<V> delegateTree, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateTree, orderedExecutor, threadPoolExecutor);
this.delegateTree = delegateTree;
- this.executor = executor;
+ this.orderedExecutor = orderedExecutor;
}
@Override
@@ -48,56 +48,56 @@
@Override
public CompletableFuture<Map<String, Versioned<V>>> getChildren(DocumentPath path) {
- return Tools.asyncFuture(delegateTree.getChildren(path), executor);
+ return asyncFuture(delegateTree.getChildren(path));
}
@Override
public CompletableFuture<Versioned<V>> get(DocumentPath path) {
- return Tools.asyncFuture(delegateTree.get(path), executor);
+ return asyncFuture(delegateTree.get(path));
}
@Override
public CompletableFuture<Versioned<V>> set(DocumentPath path, V value) {
- return Tools.asyncFuture(delegateTree.set(path, value), executor);
+ return asyncFuture(delegateTree.set(path, value));
}
@Override
public CompletableFuture<Boolean> create(DocumentPath path, V value) {
- return Tools.asyncFuture(delegateTree.create(path, value), executor);
+ return asyncFuture(delegateTree.create(path, value));
}
@Override
public CompletableFuture<Boolean> createRecursive(DocumentPath path, V value) {
- return Tools.asyncFuture(delegateTree.createRecursive(path, value), executor);
+ return asyncFuture(delegateTree.createRecursive(path, value));
}
@Override
public CompletableFuture<Boolean> replace(DocumentPath path, V newValue, long version) {
- return Tools.asyncFuture(delegateTree.replace(path, newValue, version), executor);
+ return asyncFuture(delegateTree.replace(path, newValue, version));
}
@Override
public CompletableFuture<Boolean> replace(DocumentPath path, V newValue, V currentValue) {
- return Tools.asyncFuture(delegateTree.replace(path, newValue, currentValue), executor);
+ return asyncFuture(delegateTree.replace(path, newValue, currentValue));
}
@Override
public CompletableFuture<Versioned<V>> removeNode(DocumentPath path) {
- return Tools.asyncFuture(delegateTree.removeNode(path), executor);
+ return asyncFuture(delegateTree.removeNode(path));
}
@Override
public CompletableFuture<Void> addListener(DocumentPath path, DocumentTreeListener<V> listener) {
- DocumentTreeListener<V> wrappedListener = e -> executor.execute(() -> listener.event(e));
+ DocumentTreeListener<V> wrappedListener = e -> orderedExecutor.execute(() -> listener.event(e));
listenerMap.put(listener, wrappedListener);
- return Tools.asyncFuture(delegateTree.addListener(path, wrappedListener), executor);
+ return asyncFuture(delegateTree.addListener(path, wrappedListener));
}
@Override
public CompletableFuture<Void> removeListener(DocumentTreeListener<V> listener) {
DocumentTreeListener<V> wrappedListener = listenerMap.remove(listener);
if (wrappedListener != null) {
- return Tools.asyncFuture(delegateTree.removeListener(wrappedListener), executor);
+ return asyncFuture(delegateTree.removeListener(wrappedListener));
}
return CompletableFuture.completedFuture(null);
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncLeaderElector.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncLeaderElector.java
index ecc5b8d..ba7cb81 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncLeaderElector.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingAsyncLeaderElector.java
@@ -21,7 +21,6 @@
import java.util.function.Consumer;
import com.google.common.collect.Maps;
-import org.onlab.util.Tools;
import org.onosproject.cluster.Leadership;
import org.onosproject.cluster.NodeId;
import org.onosproject.event.Change;
@@ -33,62 +32,63 @@
*/
public class ExecutingAsyncLeaderElector extends ExecutingDistributedPrimitive implements AsyncLeaderElector {
private final AsyncLeaderElector delegateElector;
- private final Executor executor;
+ private final Executor orderedExecutor;
private final Map<Consumer<Change<Leadership>>, Consumer<Change<Leadership>>> listenerMap = Maps.newConcurrentMap();
- public ExecutingAsyncLeaderElector(AsyncLeaderElector delegateElector, Executor executor) {
- super(delegateElector, executor);
+ public ExecutingAsyncLeaderElector(
+ AsyncLeaderElector delegateElector, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateElector, orderedExecutor, threadPoolExecutor);
this.delegateElector = delegateElector;
- this.executor = executor;
+ this.orderedExecutor = orderedExecutor;
}
@Override
public CompletableFuture<Leadership> run(String topic, NodeId nodeId) {
- return Tools.asyncFuture(delegateElector.run(topic, nodeId), executor);
+ return asyncFuture(delegateElector.run(topic, nodeId));
}
@Override
public CompletableFuture<Void> withdraw(String topic) {
- return Tools.asyncFuture(delegateElector.withdraw(topic), executor);
+ return asyncFuture(delegateElector.withdraw(topic));
}
@Override
public CompletableFuture<Boolean> anoint(String topic, NodeId nodeId) {
- return Tools.asyncFuture(delegateElector.anoint(topic, nodeId), executor);
+ return asyncFuture(delegateElector.anoint(topic, nodeId));
}
@Override
public CompletableFuture<Void> evict(NodeId nodeId) {
- return Tools.asyncFuture(delegateElector.evict(nodeId), executor);
+ return asyncFuture(delegateElector.evict(nodeId));
}
@Override
public CompletableFuture<Boolean> promote(String topic, NodeId nodeId) {
- return Tools.asyncFuture(delegateElector.promote(topic, nodeId), executor);
+ return asyncFuture(delegateElector.promote(topic, nodeId));
}
@Override
public CompletableFuture<Leadership> getLeadership(String topic) {
- return Tools.asyncFuture(delegateElector.getLeadership(topic), executor);
+ return asyncFuture(delegateElector.getLeadership(topic));
}
@Override
public CompletableFuture<Map<String, Leadership>> getLeaderships() {
- return Tools.asyncFuture(delegateElector.getLeaderships(), executor);
+ return asyncFuture(delegateElector.getLeaderships());
}
@Override
public CompletableFuture<Void> addChangeListener(Consumer<Change<Leadership>> listener) {
- Consumer<Change<Leadership>> wrappedListener = e -> executor.execute(() -> listener.accept(e));
+ Consumer<Change<Leadership>> wrappedListener = e -> orderedExecutor.execute(() -> listener.accept(e));
listenerMap.put(listener, wrappedListener);
- return Tools.asyncFuture(delegateElector.addChangeListener(wrappedListener), executor);
+ return asyncFuture(delegateElector.addChangeListener(wrappedListener));
}
@Override
public CompletableFuture<Void> removeChangeListener(Consumer<Change<Leadership>> listener) {
Consumer<Change<Leadership>> wrappedListener = listenerMap.remove(listener);
if (wrappedListener != null) {
- return Tools.asyncFuture(delegateElector.removeChangeListener(wrappedListener), executor);
+ return asyncFuture(delegateElector.removeChangeListener(wrappedListener));
}
return CompletableFuture.completedFuture(null);
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingDistributedPrimitive.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingDistributedPrimitive.java
index 021dbe5..836a682 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingDistributedPrimitive.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingDistributedPrimitive.java
@@ -32,24 +32,38 @@
public abstract class ExecutingDistributedPrimitive
extends DelegatingDistributedPrimitive {
private final DistributedPrimitive primitive;
- private final Executor executor;
+ private final Executor orderedExecutor;
+ private final Executor threadPoolExecutor;
private final Map<Consumer<Status>, Consumer<Status>> listenerMap = Maps.newConcurrentMap();
- protected ExecutingDistributedPrimitive(DistributedPrimitive primitive, Executor executor) {
+ protected ExecutingDistributedPrimitive(
+ DistributedPrimitive primitive, Executor orderedExecutor, Executor threadPoolExecutor) {
super(primitive);
this.primitive = primitive;
- this.executor = checkNotNull(executor);
+ this.orderedExecutor = checkNotNull(orderedExecutor);
+ this.threadPoolExecutor = checkNotNull(threadPoolExecutor);
+ }
+
+ /**
+ * Creates a future to be completed asynchronously on the provided ordered and thread pool executors.
+ *
+ * @param future the future to be completed asynchronously
+ * @param <T> future result type
+ * @return a new {@link CompletableFuture} to be completed asynchronously using the primitive thread model
+ */
+ protected <T> CompletableFuture<T> asyncFuture(CompletableFuture<T> future) {
+ return Tools.orderedFuture(future, orderedExecutor, threadPoolExecutor);
}
@Override
public CompletableFuture<Void> destroy() {
- return Tools.asyncFuture(primitive.destroy(), executor);
+ return asyncFuture(primitive.destroy());
}
@Override
public void addStatusChangeListener(Consumer<DistributedPrimitive.Status> listener) {
Consumer<DistributedPrimitive.Status> wrappedListener =
- status -> executor.execute(() -> listener.accept(status));
+ status -> orderedExecutor.execute(() -> listener.accept(status));
listenerMap.put(listener, wrappedListener);
primitive.addStatusChangeListener(wrappedListener);
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingWorkQueue.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingWorkQueue.java
index 301f73e..e6290b8 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingWorkQueue.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/ExecutingWorkQueue.java
@@ -20,7 +20,6 @@
import java.util.concurrent.Executor;
import java.util.function.Consumer;
-import org.onlab.util.Tools;
import org.onosproject.store.service.AsyncAtomicValue;
import org.onosproject.store.service.Task;
import org.onosproject.store.service.WorkQueue;
@@ -32,44 +31,40 @@
*/
public class ExecutingWorkQueue<E> extends ExecutingDistributedPrimitive implements WorkQueue<E> {
private final WorkQueue<E> delegateQueue;
- private final Executor executor;
- public ExecutingWorkQueue(WorkQueue<E> delegateQueue, Executor executor) {
- super(delegateQueue, executor);
+ public ExecutingWorkQueue(WorkQueue<E> delegateQueue, Executor orderedExecutor, Executor threadPoolExecutor) {
+ super(delegateQueue, orderedExecutor, threadPoolExecutor);
this.delegateQueue = delegateQueue;
- this.executor = executor;
}
@Override
public CompletableFuture<Void> addMultiple(Collection<E> items) {
- return Tools.asyncFuture(delegateQueue.addMultiple(items), executor);
+ return asyncFuture(delegateQueue.addMultiple(items));
}
@Override
public CompletableFuture<Collection<Task<E>>> take(int maxItems) {
- return Tools.asyncFuture(delegateQueue.take(maxItems), executor);
+ return asyncFuture(delegateQueue.take(maxItems));
}
@Override
public CompletableFuture<Void> complete(Collection<String> taskIds) {
- return Tools.asyncFuture(delegateQueue.complete(taskIds), executor);
+ return asyncFuture(delegateQueue.complete(taskIds));
}
@Override
public CompletableFuture<Void> registerTaskProcessor(
Consumer<E> taskProcessor, int parallelism, Executor executor) {
- return Tools.asyncFuture(
- delegateQueue.registerTaskProcessor(taskProcessor, parallelism, executor),
- this.executor);
+ return asyncFuture(delegateQueue.registerTaskProcessor(taskProcessor, parallelism, executor));
}
@Override
public CompletableFuture<Void> stopProcessing() {
- return Tools.asyncFuture(delegateQueue.stopProcessing(), executor);
+ return asyncFuture(delegateQueue.stopProcessing());
}
@Override
public CompletableFuture<WorkQueueStats> stats() {
- return Tools.asyncFuture(delegateQueue.stats(), executor);
+ return asyncFuture(delegateQueue.stats());
}
}
diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StoragePartitionClient.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StoragePartitionClient.java
index 83547ae..f06b052 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StoragePartitionClient.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/impl/StoragePartitionClient.java
@@ -15,6 +15,14 @@
*/
package org.onosproject.store.primitives.impl;
+import java.util.Collection;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
import com.google.common.base.Suppliers;
import io.atomix.AtomixClient;
import io.atomix.catalyst.transport.Transport;
@@ -30,7 +38,7 @@
import io.atomix.resource.ResourceType;
import io.atomix.variables.DistributedLong;
import org.onlab.util.HexString;
-import org.onlab.util.SerialExecutor;
+import org.onlab.util.OrderedExecutor;
import org.onosproject.store.primitives.DistributedPrimitiveCreator;
import org.onosproject.store.primitives.resources.impl.AtomixAtomicCounterMap;
import org.onosproject.store.primitives.resources.impl.AtomixConsistentMap;
@@ -56,14 +64,6 @@
import org.onosproject.store.service.WorkQueue;
import org.slf4j.Logger;
-import java.util.Collection;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
import static org.slf4j.LoggerFactory.getLogger;
/**
@@ -137,7 +137,7 @@
* @return the executor
*/
private Executor defaultExecutor(Supplier<Executor> executorSupplier) {
- return executorSupplier != null ? executorSupplier.get() : new SerialExecutor(sharedExecutor);
+ return executorSupplier != null ? executorSupplier.get() : new OrderedExecutor(sharedExecutor);
}
@Override
@@ -165,7 +165,7 @@
value -> value == null ? null : serializer.encode(value),
bytes -> serializer.decode(bytes));
- return new ExecutingAsyncConsistentMap<>(transcodedMap, defaultExecutor(executorSupplier));
+ return new ExecutingAsyncConsistentMap<>(transcodedMap, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override
@@ -193,7 +193,7 @@
value -> value == null ? null : serializer.encode(value),
bytes -> serializer.decode(bytes));
- return new ExecutingAsyncConsistentTreeMap<>(transcodedMap, defaultExecutor(executorSupplier));
+ return new ExecutingAsyncConsistentTreeMap<>(transcodedMap, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override
@@ -225,7 +225,7 @@
value -> serializer.encode(value),
bytes -> serializer.decode(bytes));
- return new ExecutingAsyncConsistentMultimap<>(transcodedMap, defaultExecutor(executorSupplier));
+ return new ExecutingAsyncConsistentMultimap<>(transcodedMap, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override
@@ -247,28 +247,28 @@
key -> HexString.toHexString(serializer.encode(key)),
string -> serializer.decode(HexString.fromHexString(string)));
- return new ExecutingAsyncAtomicCounterMap<>(transcodedMap, defaultExecutor(executorSupplier));
+ return new ExecutingAsyncAtomicCounterMap<>(transcodedMap, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override
public AsyncAtomicCounter newAsyncCounter(String name, Supplier<Executor> executorSupplier) {
DistributedLong distributedLong = client.getLong(name).join();
AsyncAtomicCounter asyncCounter = new AtomixCounter(name, distributedLong);
- return new ExecutingAsyncAtomicCounter(asyncCounter, defaultExecutor(executorSupplier));
+ return new ExecutingAsyncAtomicCounter(asyncCounter, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override
public <V> AsyncAtomicValue<V> newAsyncAtomicValue(
String name, Serializer serializer, Supplier<Executor> executorSupplier) {
AsyncAtomicValue<V> asyncValue = new DefaultAsyncAtomicValue<>(name, serializer, onosAtomicValuesMap.get());
- return new ExecutingAsyncAtomicValue<>(asyncValue, defaultExecutor(executorSupplier));
+ return new ExecutingAsyncAtomicValue<>(asyncValue, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override
public <E> WorkQueue<E> newWorkQueue(String name, Serializer serializer, Supplier<Executor> executorSupplier) {
AtomixWorkQueue atomixWorkQueue = client.getResource(name, AtomixWorkQueue.class).join();
WorkQueue<E> workQueue = new DefaultDistributedWorkQueue<>(atomixWorkQueue, serializer);
- return new ExecutingWorkQueue<>(workQueue, defaultExecutor(executorSupplier));
+ return new ExecutingWorkQueue<>(workQueue, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override
@@ -277,7 +277,7 @@
AtomixDocumentTree atomixDocumentTree = client.getResource(name, AtomixDocumentTree.class).join();
AsyncDocumentTree<V> asyncDocumentTree = new DefaultDistributedDocumentTree<>(
name, atomixDocumentTree, serializer);
- return new ExecutingAsyncDocumentTree<>(asyncDocumentTree, defaultExecutor(executorSupplier));
+ return new ExecutingAsyncDocumentTree<>(asyncDocumentTree, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override
@@ -285,12 +285,10 @@
AtomixLeaderElector leaderElector = client.getResource(name, AtomixLeaderElector.class)
.thenCompose(AtomixLeaderElector::setupCache)
.join();
- Consumer<State> statusListener = state -> {
- leaderElector.statusChangeListeners()
- .forEach(listener -> listener.accept(mapper.apply(state)));
- };
+ Consumer<State> statusListener = state -> leaderElector.statusChangeListeners()
+ .forEach(listener -> listener.accept(mapper.apply(state)));
resourceClient.client().onStateChange(statusListener);
- return new ExecutingAsyncLeaderElector(leaderElector, defaultExecutor(executorSupplier));
+ return new ExecutingAsyncLeaderElector(leaderElector, defaultExecutor(executorSupplier), sharedExecutor);
}
@Override