X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FCompositeDataTreeCohort.java;h=a84fa4610d5763c9ab84ae73687640f1236024f5;hb=4447f81c26b851e46acd3f111768bb498f0d553f;hp=8115473a0565fb2c3865f4bc03e7216a8029a579;hpb=a47dd7a5d21ca68804a6d0e2e3ca765f223c2ef4;p=controller.git
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/CompositeDataTreeCohort.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/CompositeDataTreeCohort.java
index 8115473a05..a84fa4610d 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/CompositeDataTreeCohort.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/CompositeDataTreeCohort.java
@@ -8,19 +8,24 @@
package org.opendaylight.controller.cluster.datastore;
+import akka.actor.ActorRef;
import akka.actor.Status;
import akka.actor.Status.Failure;
import akka.dispatch.ExecutionContexts;
import akka.dispatch.Futures;
import akka.dispatch.Recover;
-import akka.japi.Function;
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.ImmutableList;
+import com.google.common.collect.Lists;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
import java.util.Collection;
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;
@@ -29,18 +34,20 @@ import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CanComm
import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Success;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import scala.concurrent.Await;
import scala.concurrent.Future;
/**
- *
* Composite cohort, which coordinates multiple user-provided cohorts as if it was only one cohort.
- *
+ *
* It tracks current operation and list of cohorts which successfuly finished previous phase in
* case, if abort is necessary to invoke it only on cohort steps which are still active.
*
*/
class CompositeDataTreeCohort {
+ private static final Logger LOG = LoggerFactory.getLogger(CompositeDataTreeCohort.class);
private enum State {
/**
@@ -72,31 +79,28 @@ class CompositeDataTreeCohort {
*/
COMMITED,
/**
- * Some of cohorts responsed back with unsuccessful message.
- *
+ * Some of cohorts responded back with unsuccessful message.
*/
FAILED,
/**
- *
* Abort message was send to all cohorts which responded with success previously.
- *
*/
ABORTED
}
- protected static final Recover EXCEPTION_TO_MESSAGE = new Recover() {
+ static final Recover EXCEPTION_TO_MESSAGE = new Recover() {
@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 Timeout timeout;
- private Iterable successfulFromPrevious;
+
+ private List successfulFromPrevious;
private State state = State.IDLE;
CompositeDataTreeCohort(final DataTreeCohortActorRegistry registry, final TransactionIdentifier transactionID,
@@ -107,77 +111,162 @@ class CompositeDataTreeCohort {
this.timeout = Preconditions.checkNotNull(timeout);
}
+ void reset() {
+ switch (state) {
+ case CAN_COMMIT_SENT:
+ case CAN_COMMIT_SUCCESSFUL:
+ case PRE_COMMIT_SENT:
+ case PRE_COMMIT_SUCCESSFUL:
+ case COMMIT_SENT:
+ abort();
+ break;
+ case ABORTED:
+ case COMMITED:
+ case FAILED:
+ case IDLE:
+ break;
+ default:
+ throw new IllegalStateException("Unhandled state " + state);
+ }
+
+ successfulFromPrevious = null;
+ state = State.IDLE;
+ }
+
void canCommit(final DataTreeCandidate tip) throws ExecutionException, TimeoutException {
- Collection messages = registry.createCanCommitMessages(txId, tip, schema);
- // FIXME: Optimize empty collection list with pre-created futures, containing success.
- Future> canCommitsFuture =
- Futures.traverse(messages, new Function>() {
- @Override
- public Future apply(final CanCommit input) {
- return Patterns.ask(input.getCohort(), input, timeout).recover(EXCEPTION_TO_MESSAGE,
- ExecutionContexts.global());
- }
- }, ExecutionContexts.global());
+ LOG.debug("{}: canCommit - candidate: {}", txId, tip);
+
+ final List messages = registry.createCanCommitMessages(txId, tip, schema);
+ LOG.debug("{}: canCommit - messages: {}", txId, messages);
+ if (messages.isEmpty()) {
+ successfulFromPrevious = ImmutableList.of();
+ changeStateFrom(State.IDLE, State.CAN_COMMIT_SUCCESSFUL);
+ return;
+ }
+
+ final List>> futures = new ArrayList<>(messages.size());
+ for (CanCommit message : messages) {
+ final ActorRef actor = message.getCohort();
+ final Future 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));
+ }
+
changeStateFrom(State.IDLE, State.CAN_COMMIT_SENT);
- processResponses(canCommitsFuture, State.CAN_COMMIT_SENT, State.CAN_COMMIT_SUCCESSFUL);
+ processResponses(futures, State.CAN_COMMIT_SENT, State.CAN_COMMIT_SUCCESSFUL);
}
void preCommit() throws ExecutionException, TimeoutException {
+ LOG.debug("{}: preCommit - successfulFromPrevious: {}", txId, successfulFromPrevious);
+
Preconditions.checkState(successfulFromPrevious != null);
- Future> preCommitFutures = sendMesageToSuccessful(new DataTreeCohortActor.PreCommit(txId));
+ if (successfulFromPrevious.isEmpty()) {
+ changeStateFrom(State.CAN_COMMIT_SUCCESSFUL, State.PRE_COMMIT_SUCCESSFUL);
+ return;
+ }
+
+ final List>> 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);
+ processResponses(futures, State.PRE_COMMIT_SENT, State.PRE_COMMIT_SUCCESSFUL);
}
void commit() throws ExecutionException, TimeoutException {
+ LOG.debug("{}: commit - successfulFromPrevious: {}", txId, successfulFromPrevious);
+ if (successfulFromPrevious.isEmpty()) {
+ changeStateFrom(State.PRE_COMMIT_SUCCESSFUL, State.COMMITED);
+ return;
+ }
+
Preconditions.checkState(successfulFromPrevious != null);
- Future> commitsFuture = sendMesageToSuccessful(new DataTreeCohortActor.Commit(txId));
+ final List>> futures = sendMessageToSuccessful(
+ new DataTreeCohortActor.Commit(txId));
changeStateFrom(State.PRE_COMMIT_SUCCESSFUL, State.COMMIT_SENT);
- processResponses(commitsFuture, State.COMMIT_SENT, State.COMMITED);
+ processResponses(futures, State.COMMIT_SENT, State.COMMITED);
}
- Optional>> abort() {
- if (successfulFromPrevious != null) {
- return Optional.of(sendMesageToSuccessful(new DataTreeCohortActor.Abort(txId)));
+ Optional>> abort() {
+ LOG.debug("{}: abort - successfulFromPrevious: {}", txId, successfulFromPrevious);
+
+ state = State.ABORTED;
+ if (successfulFromPrevious == null || successfulFromPrevious.isEmpty()) {
+ return Optional.empty();
}
- return Optional.empty();
+ final DataTreeCohortActor.Abort message = new DataTreeCohortActor.Abort(txId);
+ final List> futures = new ArrayList<>(successfulFromPrevious.size());
+ for (Success s : successfulFromPrevious) {
+ futures.add(Patterns.ask(s.getCohort(), message, timeout));
+ }
+ return Optional.of(futures);
}
- private Future> sendMesageToSuccessful(final Object message) {
- return Futures.traverse(successfulFromPrevious, new Function>() {
-
- @Override
- public Future apply(final DataTreeCohortActor.Success cohortResponse) throws Exception {
- return Patterns.ask(cohortResponse.getCohort(), message, timeout);
- }
+ private List>> sendMessageToSuccessful(final Object message) {
+ LOG.debug("{}: sendMesageToSuccessful: {}", txId, message);
- }, ExecutionContexts.global());
+ final List>> 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;
}
- private void processResponses(final Future> resultsFuture, final State currentState, final State afterState)
- throws TimeoutException, ExecutionException {
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ private void processResponses(final List>> futures, final State currentState,
+ final State afterState) throws TimeoutException, ExecutionException {
+ LOG.debug("{}: processResponses - currentState: {}, afterState: {}", txId, currentState, afterState);
+
final Iterable results;
try {
- results = Await.result(resultsFuture, timeout.duration());
+ results = Await.result(Futures.sequence(Lists.transform(futures, e -> e.getValue()),
+ ExecutionContexts.global()), timeout.duration());
+ } catch (TimeoutException e) {
+ successfulFromPrevious = null;
+ LOG.debug("{}: processResponses - error from Future", txId, e);
+
+ for (Entry> f : futures) {
+ if (!f.getValue().isCompleted()) {
+ LOG.info("{}: actor {} failed to respond", txId, f.getKey());
+ }
+ }
+ throw e;
+ } catch (ExecutionException e) {
+ successfulFromPrevious = null;
+ LOG.debug("{}: processResponses - error from Future", txId, e);
+ throw e;
} catch (Exception e) {
successfulFromPrevious = null;
- Throwables.propagateIfInstanceOf(e, TimeoutException.class);
- throw Throwables.propagate(e);
+ LOG.debug("{}: processResponses - error from Future", txId, e);
+ throw new ExecutionException(e);
+ }
+
+ final Collection failed = new ArrayList<>(1);
+ final List successful = new ArrayList<>(futures.size());
+ 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", result);
+ }
}
- Iterable failed = Iterables.filter(results, Status.Failure.class);
- Iterable 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 it = failed.iterator();
- Throwable firstEx = it.next().cause();
+ final Iterator 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);
+ Throwables.propagateIfInstanceOf(firstEx, ExecutionException.class);
+ Throwables.propagateIfInstanceOf(firstEx, TimeoutException.class);
+ throw new ExecutionException(firstEx);
}
changeStateFrom(currentState, afterState);
}