Added RetryingFunction for simplified retry support.
Moved retry logic out of primitives (AtomicCounter) to the caller site.
Change-Id: I319d61f153f98d421baf32a1b5cd69d20dc63427
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 7ff6b89..8a9f884 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,15 +16,9 @@
package org.onosproject.store.consistent.impl;
import org.onosproject.store.service.AsyncAtomicCounter;
-import org.slf4j.Logger;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.function.BiFunction;
-
import static com.google.common.base.Preconditions.checkNotNull;
-import static org.slf4j.LoggerFactory.getLogger;
/**
* Default implementation for a distributed AsyncAtomicCounter backed by
@@ -36,11 +30,6 @@
private final String name;
private final Database database;
- private final boolean retryOnFailure;
- private final ScheduledExecutorService retryExecutor;
- // TODO: configure delay via builder
- private static final int DELAY_BETWEEN_RETRY_SEC = 1;
- private final Logger log = getLogger(getClass());
private final MeteringAgent monitor;
private static final String PRIMITIVE_NAME = "atomicCounter";
@@ -52,13 +41,9 @@
public DefaultAsyncAtomicCounter(String name,
Database database,
- boolean retryOnException,
- boolean meteringEnabled,
- ScheduledExecutorService retryExecutor) {
+ boolean meteringEnabled) {
this.name = checkNotNull(name);
this.database = checkNotNull(database);
- this.retryOnFailure = retryOnException;
- this.retryExecutor = retryExecutor;
this.monitor = new MeteringAgent(PRIMITIVE_NAME, name, meteringEnabled);
}
@@ -86,77 +71,14 @@
@Override
public CompletableFuture<Long> getAndAdd(long delta) {
final MeteringAgent.Context timer = monitor.startTimer(GET_AND_ADD);
- CompletableFuture<Long> result = database.counterGetAndAdd(name, delta);
- if (!retryOnFailure) {
- return result
- .whenComplete((r, e) -> timer.stop());
- }
-
- CompletableFuture<Long> future = new CompletableFuture<>();
- return result.whenComplete((r, e) -> {
- timer.stop();
- // TODO : Account for retries
- if (e != null) {
- log.warn("getAndAdd failed due to {}. Will retry", e.getMessage());
- retryExecutor.schedule(new RetryTask(database::counterGetAndAdd, delta, future),
- DELAY_BETWEEN_RETRY_SEC,
- TimeUnit.SECONDS);
- } else {
- future.complete(r);
- }
- }).thenCompose(v -> future);
+ return database.counterGetAndAdd(name, delta)
+ .whenComplete((r, e) -> timer.stop());
}
@Override
public CompletableFuture<Long> addAndGet(long delta) {
final MeteringAgent.Context timer = monitor.startTimer(ADD_AND_GET);
- CompletableFuture<Long> result = database.counterAddAndGet(name, delta);
- if (!retryOnFailure) {
- return result
- .whenComplete((r, e) -> timer.stop());
- }
-
- CompletableFuture<Long> future = new CompletableFuture<>();
- return result.whenComplete((r, e) -> {
- timer.stop();
- // TODO : Account for retries
- if (e != null) {
- log.warn("addAndGet failed due to {}. Will retry", e.getMessage());
- retryExecutor.schedule(new RetryTask(database::counterAddAndGet, delta, future),
- DELAY_BETWEEN_RETRY_SEC,
- TimeUnit.SECONDS);
- } else {
- future.complete(r);
- }
- }).thenCompose(v -> future);
- }
-
- private class RetryTask implements Runnable {
-
- private final BiFunction<String, Long, CompletableFuture<Long>> function;
- private final Long delta;
- private final CompletableFuture<Long> result;
-
- public RetryTask(BiFunction<String, Long, CompletableFuture<Long>> function,
- Long delta,
- CompletableFuture<Long> result) {
- this.function = function;
- this.delta = delta;
- this.result = result;
- }
-
- @Override
- public void run() {
- function.apply(name, delta).whenComplete((r, e) -> {
- if (e == null) {
- result.complete(r);
- } else {
- log.warn("{} retry failed due to {}. Will try again...", function, e.getMessage());
- // TODO: Exponential backoff
- // TODO: limit retries
- retryExecutor.schedule(this, DELAY_BETWEEN_RETRY_SEC, TimeUnit.SECONDS);
- }
- });
- }
+ return database.counterAddAndGet(name, delta)
+ .whenComplete((r, e) -> timer.stop());
}
}
\ No newline at end of file
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 483dbc4..64886e4 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
@@ -21,7 +21,6 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -39,10 +38,8 @@
public DefaultAtomicCounter(String name,
Database database,
- boolean retryOnException,
- boolean meteringEnabled,
- ScheduledExecutorService retryExecutor) {
- asyncCounter = new DefaultAsyncAtomicCounter(name, database, retryOnException, meteringEnabled, retryExecutor);
+ boolean meteringEnabled) {
+ asyncCounter = new DefaultAsyncAtomicCounter(name, database, meteringEnabled);
}
@Override
diff --git a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAtomicCounterBuilder.java b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAtomicCounterBuilder.java
index 4f0a296..dba4443 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAtomicCounterBuilder.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/consistent/impl/DefaultAtomicCounterBuilder.java
@@ -19,9 +19,8 @@
import org.onosproject.store.service.AtomicCounter;
import org.onosproject.store.service.AtomicCounterBuilder;
-import java.util.concurrent.ScheduledExecutorService;
-
import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
/**
* Default implementation of AtomicCounterBuilder.
@@ -32,9 +31,7 @@
private boolean partitionsEnabled = true;
private final Database partitionedDatabase;
private final Database inMemoryDatabase;
- private boolean retryOnFailure = false;
private boolean metering = true;
- private ScheduledExecutorService retryExecutor = null;
public DefaultAtomicCounterBuilder(Database inMemoryDatabase, Database partitionedDatabase) {
this.inMemoryDatabase = inMemoryDatabase;
@@ -58,20 +55,14 @@
public AtomicCounter build() {
validateInputs();
Database database = partitionsEnabled ? partitionedDatabase : inMemoryDatabase;
- return new DefaultAtomicCounter(name, database, retryOnFailure, metering, retryExecutor);
+ return new DefaultAtomicCounter(name, database, metering);
}
@Override
public AsyncAtomicCounter buildAsyncCounter() {
validateInputs();
Database database = partitionsEnabled ? partitionedDatabase : inMemoryDatabase;
- return new DefaultAsyncAtomicCounter(name, database, retryOnFailure, metering, retryExecutor);
- }
-
- @Override
- public AtomicCounterBuilder withRetryOnFailure() {
- retryOnFailure = true;
- return this;
+ return new DefaultAsyncAtomicCounter(name, database, metering);
}
@Override
@@ -80,17 +71,7 @@
return this;
}
- @Override
- public AtomicCounterBuilder withRetryExecutor(ScheduledExecutorService executor) {
- this.retryExecutor = executor;
- return this;
- }
-
private void validateInputs() {
- if (retryOnFailure) {
- if (retryExecutor == null) {
- throw new IllegalArgumentException("RetryExecutor must be specified when retries are enabled");
- }
- }
+ checkState(name != null, "name must be specified");
}
}