* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-
package org.opendaylight.controller.cluster.datastore;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+import akka.actor.ActorRef;
import akka.actor.Status;
import akka.actor.Status.Failure;
import akka.dispatch.ExecutionContexts;
import akka.dispatch.Futures;
+import akka.dispatch.OnComplete;
import akka.dispatch.Recover;
import akka.pattern.Patterns;
import akka.util.Timeout;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
-import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
import java.util.Optional;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CanCommit;
import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Success;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.concurrent.Await;
+import scala.compat.java8.FutureConverters;
import scala.concurrent.Future;
/**
ABORTED
}
- protected static final Recover<Object> EXCEPTION_TO_MESSAGE = new Recover<Object>() {
+ static final Recover<Object> EXCEPTION_TO_MESSAGE = new Recover<Object>() {
@Override
- public Failure recover(final Throwable error) throws Throwable {
+ public Failure recover(final Throwable error) {
return new Failure(error);
}
};
-
private final DataTreeCohortActorRegistry registry;
private final TransactionIdentifier txId;
private final SchemaContext schema;
+ private final Executor callbackExecutor;
private final Timeout timeout;
- private Iterable<Success> successfulFromPrevious;
+
+ private @NonNull List<Success> successfulFromPrevious = Collections.emptyList();
private State state = State.IDLE;
CompositeDataTreeCohort(final DataTreeCohortActorRegistry registry, final TransactionIdentifier transactionID,
- final SchemaContext schema, final Timeout timeout) {
- this.registry = Preconditions.checkNotNull(registry);
- this.txId = Preconditions.checkNotNull(transactionID);
- this.schema = Preconditions.checkNotNull(schema);
- this.timeout = Preconditions.checkNotNull(timeout);
+ final SchemaContext schema, final Executor callbackExecutor, final Timeout timeout) {
+ this.registry = requireNonNull(registry);
+ this.txId = requireNonNull(transactionID);
+ this.schema = requireNonNull(schema);
+ this.callbackExecutor = requireNonNull(callbackExecutor);
+ this.timeout = requireNonNull(timeout);
}
void reset() {
case COMMIT_SENT:
abort();
break;
- default :
+ case ABORTED:
+ case COMMITED:
+ case FAILED:
+ case IDLE:
break;
+ default:
+ throw new IllegalStateException("Unhandled state " + state);
}
- successfulFromPrevious = null;
+ successfulFromPrevious = Collections.emptyList();
state = State.IDLE;
}
- void canCommit(final DataTreeCandidate tip) throws ExecutionException, TimeoutException {
- LOG.debug("{}: canCommit - candidate: {}", txId, tip);
-
- Collection<CanCommit> messages = registry.createCanCommitMessages(txId, tip, schema);
+ Optional<CompletionStage<Void>> canCommit(final DataTreeCandidate tip) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("{}: canCommit - candidate: {}", txId, tip);
+ } else {
+ LOG.debug("{}: canCommit - candidate rootPath: {}", txId, tip.getRootPath());
+ }
+ final List<CanCommit> messages = registry.createCanCommitMessages(txId, tip, schema);
LOG.debug("{}: canCommit - messages: {}", txId, messages);
+ if (messages.isEmpty()) {
+ successfulFromPrevious = Collections.emptyList();
+ changeStateFrom(State.IDLE, State.CAN_COMMIT_SUCCESSFUL);
+ return Optional.empty();
+ }
+
+ final List<Entry<ActorRef, Future<Object>>> futures = new ArrayList<>(messages.size());
+ for (CanCommit message : messages) {
+ final ActorRef actor = message.getCohort();
+ final Future<Object> future = Patterns.ask(actor, message, timeout).recover(EXCEPTION_TO_MESSAGE,
+ ExecutionContexts.global());
+ LOG.trace("{}: requesting canCommit from {}", txId, actor);
+ futures.add(new SimpleImmutableEntry<>(actor, future));
+ }
- // FIXME: Optimize empty collection list with pre-created futures, containing success.
- Future<Iterable<Object>> canCommitsFuture = Futures.traverse(messages,
- input -> Patterns.ask(input.getCohort(), input, timeout).recover(EXCEPTION_TO_MESSAGE,
- ExecutionContexts.global()), ExecutionContexts.global());
changeStateFrom(State.IDLE, State.CAN_COMMIT_SENT);
- processResponses(canCommitsFuture, State.CAN_COMMIT_SENT, State.CAN_COMMIT_SUCCESSFUL);
+ return Optional.of(processResponses(futures, State.CAN_COMMIT_SENT, State.CAN_COMMIT_SUCCESSFUL));
}
- void preCommit() throws ExecutionException, TimeoutException {
+ Optional<CompletionStage<Void>> preCommit() {
LOG.debug("{}: preCommit - successfulFromPrevious: {}", txId, successfulFromPrevious);
- Preconditions.checkState(successfulFromPrevious != null);
- Future<Iterable<Object>> preCommitFutures = sendMesageToSuccessful(new DataTreeCohortActor.PreCommit(txId));
+ if (successfulFromPrevious.isEmpty()) {
+ changeStateFrom(State.CAN_COMMIT_SUCCESSFUL, State.PRE_COMMIT_SUCCESSFUL);
+ return Optional.empty();
+ }
+
+ final List<Entry<ActorRef, Future<Object>>> futures = sendMessageToSuccessful(
+ new DataTreeCohortActor.PreCommit(txId));
changeStateFrom(State.CAN_COMMIT_SUCCESSFUL, State.PRE_COMMIT_SENT);
- processResponses(preCommitFutures, State.PRE_COMMIT_SENT, State.PRE_COMMIT_SUCCESSFUL);
+ return Optional.of(processResponses(futures, State.PRE_COMMIT_SENT, State.PRE_COMMIT_SUCCESSFUL));
}
- void commit() throws ExecutionException, TimeoutException {
+ Optional<CompletionStage<Void>> commit() {
LOG.debug("{}: commit - successfulFromPrevious: {}", txId, successfulFromPrevious);
+ if (successfulFromPrevious.isEmpty()) {
+ changeStateFrom(State.PRE_COMMIT_SUCCESSFUL, State.COMMITED);
+ return Optional.empty();
+ }
- Preconditions.checkState(successfulFromPrevious != null);
- Future<Iterable<Object>> commitsFuture = sendMesageToSuccessful(new DataTreeCohortActor.Commit(txId));
+ final List<Entry<ActorRef, Future<Object>>> futures = sendMessageToSuccessful(
+ new DataTreeCohortActor.Commit(txId));
changeStateFrom(State.PRE_COMMIT_SUCCESSFUL, State.COMMIT_SENT);
- processResponses(commitsFuture, State.COMMIT_SENT, State.COMMITED);
+ return Optional.of(processResponses(futures, State.COMMIT_SENT, State.COMMITED));
}
- Optional<Future<Iterable<Object>>> abort() {
+ Optional<CompletionStage<?>> abort() {
LOG.debug("{}: abort - successfulFromPrevious: {}", txId, successfulFromPrevious);
state = State.ABORTED;
- if (successfulFromPrevious != null && !Iterables.isEmpty(successfulFromPrevious)) {
- return Optional.of(sendMesageToSuccessful(new DataTreeCohortActor.Abort(txId)));
+ if (successfulFromPrevious.isEmpty()) {
+ return Optional.empty();
}
- return Optional.empty();
+ final DataTreeCohortActor.Abort message = new DataTreeCohortActor.Abort(txId);
+ final List<Future<Object>> futures = new ArrayList<>(successfulFromPrevious.size());
+ for (Success s : successfulFromPrevious) {
+ futures.add(Patterns.ask(s.getCohort(), message, timeout));
+ }
+
+ return Optional.of(FutureConverters.toJava(Futures.sequence(futures, ExecutionContexts.global())));
}
- private Future<Iterable<Object>> sendMesageToSuccessful(final Object message) {
+ private List<Entry<ActorRef, Future<Object>>> sendMessageToSuccessful(final Object message) {
LOG.debug("{}: sendMesageToSuccessful: {}", txId, message);
- return Futures.traverse(successfulFromPrevious, cohortResponse -> Patterns.ask(
- cohortResponse.getCohort(), message, timeout), ExecutionContexts.global());
+ final List<Entry<ActorRef, Future<Object>>> ret = new ArrayList<>(successfulFromPrevious.size());
+ for (Success s : successfulFromPrevious) {
+ final ActorRef actor = s.getCohort();
+ ret.add(new SimpleImmutableEntry<>(actor, Patterns.ask(actor, message, timeout)));
+ }
+ return ret;
}
- @SuppressWarnings("checkstyle:IllegalCatch")
- private void processResponses(final Future<Iterable<Object>> resultsFuture, final State currentState,
- final State afterState) throws TimeoutException, ExecutionException {
+ private @NonNull CompletionStage<Void> processResponses(final List<Entry<ActorRef, Future<Object>>> futures,
+ final State currentState, final State afterState) {
LOG.debug("{}: processResponses - currentState: {}, afterState: {}", txId, currentState, afterState);
+ final CompletableFuture<Void> returnFuture = new CompletableFuture<>();
+ Future<Iterable<Object>> aggregateFuture = Futures.sequence(Lists.transform(futures, Entry::getValue),
+ ExecutionContexts.global());
- final Iterable<Object> results;
- try {
- results = Await.result(resultsFuture, timeout.duration());
- } catch (Exception e) {
- successfulFromPrevious = null;
- LOG.debug("{}: processResponses - error from Future", txId, e);
- Throwables.propagateIfInstanceOf(e, TimeoutException.class);
- throw Throwables.propagate(e);
+ aggregateFuture.onComplete(new OnComplete<Iterable<Object>>() {
+ @Override
+ public void onComplete(final Throwable failure, final Iterable<Object> results) {
+ callbackExecutor.execute(
+ () -> processResponses(failure, results, currentState, afterState, returnFuture));
+ }
+ }, ExecutionContexts.global());
+
+ return returnFuture;
+ }
+
+ // FB issues violation for passing null to CompletableFuture#complete but it is valid and necessary when the
+ // generic type is Void.
+ @SuppressFBWarnings(value = { "NP_NONNULL_PARAM_VIOLATION", "UPM_UNCALLED_PRIVATE_METHOD" },
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
+ private void processResponses(final Throwable failure, final Iterable<Object> results,
+ final State currentState, final State afterState, final CompletableFuture<Void> resultFuture) {
+ if (failure != null) {
+ successfulFromPrevious = Collections.emptyList();
+ resultFuture.completeExceptionally(failure);
+ return;
+ }
+
+ final Collection<Failure> failed = new ArrayList<>(1);
+ final List<Success> successful = new ArrayList<>();
+ for (Object result : results) {
+ if (result instanceof DataTreeCohortActor.Success) {
+ successful.add((Success) result);
+ } else if (result instanceof Status.Failure) {
+ failed.add((Failure) result);
+ } else {
+ LOG.warn("{}: unrecognized response {}, ignoring it", txId, result);
+ }
}
- Iterable<Failure> failed = Iterables.filter(results, Status.Failure.class);
- Iterable<Success> successful = Iterables.filter(results, DataTreeCohortActor.Success.class);
LOG.debug("{}: processResponses - successful: {}, failed: {}", txId, successful, failed);
- successfulFromPrevious = successful;
- if (!Iterables.isEmpty(failed)) {
+ if (!failed.isEmpty()) {
changeStateFrom(currentState, State.FAILED);
- Iterator<Failure> it = failed.iterator();
- Throwable firstEx = it.next().cause();
+ final Iterator<Failure> it = failed.iterator();
+ final Throwable firstEx = it.next().cause();
while (it.hasNext()) {
firstEx.addSuppressed(it.next().cause());
}
- Throwables.propagateIfPossible(firstEx, ExecutionException.class);
- Throwables.propagateIfPossible(firstEx, TimeoutException.class);
- throw Throwables.propagate(firstEx);
+
+ successfulFromPrevious = Collections.emptyList();
+ resultFuture.completeExceptionally(firstEx);
+ } else {
+ successfulFromPrevious = successful;
+ changeStateFrom(currentState, afterState);
+ resultFuture.complete(null);
}
- changeStateFrom(currentState, afterState);
}
void changeStateFrom(final State expected, final State followup) {
- Preconditions.checkState(state == expected);
+ checkState(state == expected);
state = followup;
}
}
Code Review / controller.git / blobdiff