ONOS-2456 Added usage metrics to Atomic Counter and Distributed Queue plus refactored the code a bit
Refactored code plus instrumented AtomicValue and DistributedSet
Change-Id: I9c5f7c9f23d530131f15d3c98250ea33238dd2ec
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDistributedQueue.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDistributedQueue.java
index 0bcbdc4..c27774a 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDistributedQueue.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDistributedQueue.java
@@ -15,25 +15,24 @@
*/
package org.onosproject.store.consistent.impl;
-import static com.google.common.base.Preconditions.checkNotNull;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Futures;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.store.service.DistributedQueue;
+import org.onosproject.store.service.Serializer;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
-import org.onosproject.cluster.NodeId;
-import org.onosproject.store.service.DistributedQueue;
-import org.onosproject.store.service.Serializer;
-
-import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.Futures;
+import static com.google.common.base.Preconditions.checkNotNull;
/**
* DistributedQueue implementation that provides FIFO ordering semantics.
*
* @param <E> queue entry type
*/
-public class DefaultDistributedQueue<E> implements DistributedQueue<E> {
+public class DefaultDistributedQueue<E> implements DistributedQueue<E> {
private final String name;
private final Database database;
@@ -42,35 +41,56 @@
private final Set<CompletableFuture<E>> pendingFutures = Sets.newIdentityHashSet();
private final Consumer<Set<NodeId>> notifyConsumers;
+ private static final String PRIMITIVE_NAME = "distributedQueue";
+ private static final String SIZE = "size";
+ private static final String PUSH = "push";
+ private static final String POP = "pop";
+ private static final String PEEK = "peek";
+
private static final String ERROR_NULL_ENTRY = "Null entries are not allowed";
+ private final MeteringAgent monitor;
public DefaultDistributedQueue(String name,
- Database database,
- Serializer serializer,
- NodeId localNodeId,
- Consumer<Set<NodeId>> notifyConsumers) {
+ Database database,
+ Serializer serializer,
+ NodeId localNodeId,
+ boolean meteringEnabled,
+ Consumer<Set<NodeId>> notifyConsumers) {
this.name = checkNotNull(name, "queue name cannot be null");
this.database = checkNotNull(database, "database cannot be null");
this.serializer = checkNotNull(serializer, "serializer cannot be null");
this.localNodeId = localNodeId;
this.notifyConsumers = notifyConsumers;
+ this.monitor = new MeteringAgent(PRIMITIVE_NAME, name, meteringEnabled);
+
}
@Override
public long size() {
- return Futures.getUnchecked(database.queueSize(name));
+ final MeteringAgent.Context timer = monitor.startTimer(SIZE);
+ try {
+ return Futures.getUnchecked(database.queueSize(name));
+ } finally {
+ timer.stop();
+ }
}
@Override
public void push(E entry) {
- checkNotNull(entry, ERROR_NULL_ENTRY);
- Futures.getUnchecked(database.queuePush(name, serializer.encode(entry))
- .thenAccept(notifyConsumers)
- .thenApply(v -> null));
+ final MeteringAgent.Context timer = monitor.startTimer(PUSH);
+ try {
+ checkNotNull(entry, ERROR_NULL_ENTRY);
+ Futures.getUnchecked(database.queuePush(name, serializer.encode(entry))
+ .thenAccept(notifyConsumers)
+ .thenApply(v -> null));
+ } finally {
+ timer.stop();
+ }
}
@Override
public CompletableFuture<E> pop() {
+ final MeteringAgent.Context timer = monitor.startTimer(POP);
return database.queuePop(name, localNodeId)
.thenCompose(v -> {
if (v != null) {
@@ -80,13 +100,19 @@
pendingFutures.add(newPendingFuture);
return newPendingFuture;
}
- });
+ })
+ .whenComplete((r, e) -> timer.stop());
}
@Override
public E peek() {
- return Futures.getUnchecked(database.queuePeek(name)
- .thenApply(v -> v != null ? serializer.decode(v) : null));
+ final MeteringAgent.Context timer = monitor.startTimer(PEEK);
+ try {
+ return Futures.getUnchecked(database.queuePeek(name)
+ .thenApply(v -> v != null ? serializer.decode(v) : null));
+ } finally {
+ timer.stop();
+ }
}
public String name() {