Implementation of IdBlockStore on top of AtomicCounter

Change-Id: I019322a812262edeca20a55813183a63b4525b75

@@ -30,6 +30,29 @@ public interface AsyncAtomicCounter {
     CompletableFuture<Long> incrementAndGet();
     CompletableFuture<Long> incrementAndGet();
 
 
     /**
     /**
+     * Atomically increment by one the current value.
+     *
+     * @return previous value
+     */
+    CompletableFuture<Long> getAndIncrement();
+
+    /**
+     * Atomically adds the given value to the current value.
+     *
+     * @param delta the value to add
+     * @return previous value
+     */
+    CompletableFuture<Long> getAndAdd(long delta);
+
+    /**
+     * Atomically adds the given value to the current value.
+     *
+     * @param delta the value to add
+     * @return updated value
+     */
+    CompletableFuture<Long> addAndGet(long delta);
+
+    /**
      * Returns the current value of the counter without modifying it.
      * Returns the current value of the counter without modifying it.
      *
      *
      * @return current value
      * @return current value

@@ -28,6 +28,29 @@ public interface AtomicCounter {
     long incrementAndGet();
     long incrementAndGet();
 
 
     /**
     /**
+     * Atomically increment by one the current value.
+     *
+     * @return previous value
+     */
+    long getAndIncrement();
+
+    /**
+     * Atomically adds the given value to the current value.
+     *
+     * @param delta the value to add
+     * @return previous value
+     */
+    long getAndAdd(long delta);
+
+    /**
+     * Atomically adds the given value to the current value.
+     *
+     * @param delta the value to add
+     * @return updated value
+     */
+    long addAndGet(long delta);
+
+    /**
      * Returns the current value of the counter without modifying it.
      * Returns the current value of the counter without modifying it.
      *
      *
      * @return current value
      * @return current value

@@ -190,21 +190,30 @@ public interface DatabaseProxy<K, V> {
     CompletableFuture<Result<Boolean>> replace(String tableName, K key, long oldVersion, V newValue);
     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
      * @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
      * @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.
      * Prepare and commit the specified transaction.

@@ -98,10 +98,16 @@ public interface DatabaseState<K, V> {
   Result<Boolean> replace(String tableName, K key, long oldVersion, V newValue);
   Result<Boolean> replace(String tableName, K key, long oldVersion, V newValue);
 
 
   @Command
   @Command
-  Long nextValue(String counterName);
+  Long counterIncrementAndGet(String counterName);
+
+  @Command
+  Long counterGetAndIncrement(String counterName);
+
+  @Command
+  Long counterGetAndAdd(String counterName, long delta);
 
 
   @Query
   @Query
-  Long currentValue(String counterName);
+  Long counterGet(String counterName);
 
 
   @Command
   @Command
   boolean prepareAndCommit(Transaction transaction);
   boolean prepareAndCommit(Transaction transaction);

@@ -16,7 +16,9 @@
 package org.onosproject.store.consistent.impl;
 package org.onosproject.store.consistent.impl;
 
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletableFuture;
+
 import org.onosproject.store.service.AsyncAtomicCounter;
 import org.onosproject.store.service.AsyncAtomicCounter;
+
 import static com.google.common.base.Preconditions.*;
 import static com.google.common.base.Preconditions.*;
 
 
 /**
 /**
@@ -37,11 +39,26 @@ public class DefaultAsyncAtomicCounter implements AsyncAtomicCounter {
 
 
     @Override
     @Override
     public CompletableFuture<Long> incrementAndGet() {
     public CompletableFuture<Long> incrementAndGet() {
-        return database.nextValue(name);
+        return addAndGet(1L);
     }
     }
 
 
     @Override
     @Override
     public CompletableFuture<Long> get() {
     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);
     }
     }
 }
 }

@@ -46,6 +46,21 @@ public class DefaultAtomicCounter implements AtomicCounter {
     }
     }
 
 
     @Override
     @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() {
     public long get() {
         return complete(asyncCounter.get());
         return complete(asyncCounter.get());
     }
     }

@@ -150,13 +150,18 @@ public class DefaultDatabase extends AbstractResource<Database> implements Datab
     }
     }
 
 
     @Override
     @Override
-    public CompletableFuture<Long> nextValue(String counterName) {
-        return checkOpen(() -> proxy.nextValue(counterName));
+    public CompletableFuture<Long> counterGet(String counterName) {
+        return checkOpen(() -> proxy.counterGet(counterName));
     }
     }
 
 
     @Override
     @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
     @Override

@@ -225,12 +225,22 @@ public class DefaultDatabaseState implements DatabaseState<String, byte[]> {
     }
     }
 
 
     @Override
     @Override
-    public Long nextValue(String counterName) {
+    public Long counterIncrementAndGet(String counterName) {
         return getCounter(counterName).incrementAndGet();
         return getCounter(counterName).incrementAndGet();
     }
     }
 
 
     @Override
     @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();
         return getCounter(counterName).get();
     }
     }
 
 

@@ -235,15 +235,21 @@ public class PartitionedDatabase implements Database {
     }
     }
 
 
     @Override
     @Override
-    public CompletableFuture<Long> nextValue(String counterName) {
+    public CompletableFuture<Long> counterGet(String counterName) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         checkState(isOpen.get(), DB_NOT_OPEN);
-        return partitioner.getPartition(counterName, counterName).nextValue(counterName);
+        return partitioner.getPartition(counterName, counterName).counterGet(counterName);
     }
     }
 
 
     @Override
     @Override
-    public CompletableFuture<Long> currentValue(String counterName) {
+    public CompletableFuture<Long> counterAddAndGet(String counterName, long delta) {
         checkState(isOpen.get(), DB_NOT_OPEN);
         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
     @Override

+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);
+    }
+}

@@ -31,7 +31,7 @@ import java.util.Map;
 /**
 /**
  * Distributed implementation of id block store using Hazelcast.
  * Distributed implementation of id block store using Hazelcast.
  */
  */
-@Component(immediate = true)
+@Component(immediate = false, enabled = false)
 @Service
 @Service
 public class DistributedIdBlockStore implements IdBlockStore {
 public class DistributedIdBlockStore implements IdBlockStore {