Implementation of IdBlockStore on top of AtomicCounter
Change-Id: I019322a812262edeca20a55813183a63b4525b75
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseProxy.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseProxy.java
index 6c06d1f..72c2188 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseProxy.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseProxy.java
@@ -190,21 +190,30 @@
CompletableFuture<Result<Boolean>> replace(String tableName, K key, long oldVersion, V newValue);
/**
- * Returns the next value for the specified atomic counter after
- * incrementing the current value by one.
+ * Atomically add the given value to current value of the specified counter.
*
* @param counterName counter name
- * @return next value for the specified counter
+ * @param delta value to add
+ * @return updated value
*/
- CompletableFuture<Long> nextValue(String counterName);
+ CompletableFuture<Long> counterAddAndGet(String counterName, long delta);
/**
- * Returns the current value for the specified atomic counter.
+ * Atomically add the given value to current value of the specified counter.
*
* @param counterName counter name
- * @return current value for the specified counter
+ * @param delta value to add
+ * @return previous value
*/
- CompletableFuture<Long> currentValue(String counterName);
+ CompletableFuture<Long> counterGetAndAdd(String counterName, long delta);
+
+ /**
+ * Returns the current value of the specified atomic counter.
+ *
+ * @param counterName counter name
+ * @return current value
+ */
+ CompletableFuture<Long> counterGet(String counterName);
/**
* Prepare and commit the specified transaction.
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseState.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseState.java
index 5f955fd..250637d 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseState.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DatabaseState.java
@@ -98,10 +98,16 @@
Result<Boolean> replace(String tableName, K key, long oldVersion, V newValue);
@Command
- Long nextValue(String counterName);
+ Long counterIncrementAndGet(String counterName);
+
+ @Command
+ Long counterGetAndIncrement(String counterName);
+
+ @Command
+ Long counterGetAndAdd(String counterName, long delta);
@Query
- Long currentValue(String counterName);
+ Long counterGet(String counterName);
@Command
boolean prepareAndCommit(Transaction transaction);
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAsyncAtomicCounter.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAsyncAtomicCounter.java
index 3cad25f..a69c0cc 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAsyncAtomicCounter.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAsyncAtomicCounter.java
@@ -16,7 +16,9 @@
package org.onosproject.store.consistent.impl;
import java.util.concurrent.CompletableFuture;
+
import org.onosproject.store.service.AsyncAtomicCounter;
+
import static com.google.common.base.Preconditions.*;
/**
@@ -37,11 +39,26 @@
@Override
public CompletableFuture<Long> incrementAndGet() {
- return database.nextValue(name);
+ return addAndGet(1L);
}
@Override
public CompletableFuture<Long> get() {
- return database.currentValue(name);
+ return database.counterGet(name);
+ }
+
+ @Override
+ public CompletableFuture<Long> getAndIncrement() {
+ return getAndAdd(1L);
+ }
+
+ @Override
+ public CompletableFuture<Long> getAndAdd(long delta) {
+ return database.counterGetAndAdd(name, delta);
+ }
+
+ @Override
+ public CompletableFuture<Long> addAndGet(long delta) {
+ return database.counterAddAndGet(name, delta);
}
}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAtomicCounter.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAtomicCounter.java
index e479028..fee91fa 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAtomicCounter.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAtomicCounter.java
@@ -46,6 +46,21 @@
}
@Override
+ public long getAndIncrement() {
+ return complete(asyncCounter.getAndIncrement());
+ }
+
+ @Override
+ public long getAndAdd(long delta) {
+ return complete(asyncCounter.getAndAdd(delta));
+ }
+
+ @Override
+ public long addAndGet(long delta) {
+ return complete(asyncCounter.getAndAdd(delta));
+ }
+
+ @Override
public long get() {
return complete(asyncCounter.get());
}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabase.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabase.java
index c4f95bd..06c0ef4 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabase.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabase.java
@@ -150,13 +150,18 @@
}
@Override
- public CompletableFuture<Long> nextValue(String counterName) {
- return checkOpen(() -> proxy.nextValue(counterName));
+ public CompletableFuture<Long> counterGet(String counterName) {
+ return checkOpen(() -> proxy.counterGet(counterName));
}
@Override
- public CompletableFuture<Long> currentValue(String counterName) {
- return checkOpen(() -> proxy.currentValue(counterName));
+ public CompletableFuture<Long> counterAddAndGet(String counterName, long delta) {
+ return checkOpen(() -> proxy.counterAddAndGet(counterName, delta));
+ }
+
+ @Override
+ public CompletableFuture<Long> counterGetAndAdd(String counterName, long delta) {
+ return checkOpen(() -> proxy.counterGetAndAdd(counterName, delta));
}
@Override
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabaseState.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabaseState.java
index a907091..208f0e0 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabaseState.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultDatabaseState.java
@@ -225,12 +225,22 @@
}
@Override
- public Long nextValue(String counterName) {
+ public Long counterIncrementAndGet(String counterName) {
return getCounter(counterName).incrementAndGet();
}
@Override
- public Long currentValue(String counterName) {
+ public Long counterGetAndIncrement(String counterName) {
+ return getCounter(counterName).getAndIncrement();
+ }
+
+ @Override
+ public Long counterGetAndAdd(String counterName, long delta) {
+ return getCounter(counterName).getAndAdd(delta);
+ }
+
+ @Override
+ public Long counterGet(String counterName) {
return getCounter(counterName).get();
}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/PartitionedDatabase.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/PartitionedDatabase.java
index bc36419..1239475 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/PartitionedDatabase.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/PartitionedDatabase.java
@@ -235,15 +235,21 @@
}
@Override
- public CompletableFuture<Long> nextValue(String counterName) {
+ public CompletableFuture<Long> counterGet(String counterName) {
checkState(isOpen.get(), DB_NOT_OPEN);
- return partitioner.getPartition(counterName, counterName).nextValue(counterName);
+ return partitioner.getPartition(counterName, counterName).counterGet(counterName);
}
@Override
- public CompletableFuture<Long> currentValue(String counterName) {
+ public CompletableFuture<Long> counterAddAndGet(String counterName, long delta) {
checkState(isOpen.get(), DB_NOT_OPEN);
- return partitioner.getPartition(counterName, counterName).currentValue(counterName);
+ return partitioner.getPartition(counterName, counterName).counterAddAndGet(counterName, delta);
+ }
+
+ @Override
+ public CompletableFuture<Long> counterGetAndAdd(String counterName, long delta) {
+ checkState(isOpen.get(), DB_NOT_OPEN);
+ return partitioner.getPartition(counterName, counterName).counterGetAndAdd(counterName, delta);
}
@Override
diff --git a/core/store/dist/src/main/java/org/onosproject/store/core/impl/ConsistentIdBlockStore.java b/core/store/dist/src/main/java/org/onosproject/store/core/impl/ConsistentIdBlockStore.java
new file mode 100644
index 0000000..2dc3129
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onosproject/store/core/impl/ConsistentIdBlockStore.java
@@ -0,0 +1,55 @@
+package org.onosproject.store.core.impl;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.Map;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onosproject.core.IdBlock;
+import org.onosproject.core.IdBlockStore;
+import org.onosproject.store.service.AtomicCounter;
+import org.onosproject.store.service.StorageService;
+import org.slf4j.Logger;
+
+import com.google.common.collect.Maps;
+
+/**
+ * Implementation of {@code IdBlockStore} using {@code AtomicCounter}.
+ */
+@Component(immediate = true, enabled = true)
+@Service
+public class ConsistentIdBlockStore implements IdBlockStore {
+
+ private final Logger log = getLogger(getClass());
+ private final Map<String, AtomicCounter> topicCounters = Maps.newConcurrentMap();
+
+ private static final long DEFAULT_BLOCK_SIZE = 0x100000L;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected StorageService storageService;
+
+ @Activate
+ public void activate() {
+ log.info("Started");
+ }
+
+ @Deactivate
+ public void deactivate() {
+ log.info("Stopped");
+ }
+
+ @Override
+ public IdBlock getIdBlock(String topic) {
+ AtomicCounter counter = topicCounters.computeIfAbsent(topic,
+ name -> storageService.atomicCounterBuilder()
+ .withName(name)
+ .build());
+ Long blockBase = counter.getAndAdd(DEFAULT_BLOCK_SIZE);
+ return new IdBlock(blockBase, DEFAULT_BLOCK_SIZE);
+ }
+}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/core/impl/DistributedIdBlockStore.java b/core/store/dist/src/main/java/org/onosproject/store/core/impl/DistributedIdBlockStore.java
index e5098ba..260cef1 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/core/impl/DistributedIdBlockStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/core/impl/DistributedIdBlockStore.java
@@ -31,7 +31,7 @@
/**
* Distributed implementation of id block store using Hazelcast.
*/
-@Component(immediate = true)
+@Component(immediate = false, enabled = false)
@Service
public class DistributedIdBlockStore implements IdBlockStore {