opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/SimpleShardDataTreeCohort.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import akka.dispatch.ExecutionContexts;
  11 import akka.dispatch.Futures;
  12 import akka.dispatch.OnComplete;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.base.Verify;
  15 import com.google.common.primitives.UnsignedLong;
  16 import com.google.common.util.concurrent.FutureCallback;
  17 import java.util.List;
  18 import java.util.Optional;
  19 import java.util.concurrent.ExecutionException;
  20 import java.util.concurrent.TimeoutException;
  21 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  22 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  23 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateTip;
  24 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27 import scala.concurrent.Future;
  28
  29 abstract class SimpleShardDataTreeCohort extends ShardDataTreeCohort {
  30     static final class DeadOnArrival extends SimpleShardDataTreeCohort {
  31         private final Exception failure;
  32
  33         DeadOnArrival(final ShardDataTree dataTree, final DataTreeModification transaction,
  34             final TransactionIdentifier transactionId, final Exception failure) {
  35             super(dataTree, transaction, transactionId, null);
  36             this.failure = Preconditions.checkNotNull(failure);
  37         }
  38
  39         @Override
  40         void throwCanCommitFailure() throws Exception {
  41             throw failure;
  42         }
  43     }
  44
  45     static final class Normal extends SimpleShardDataTreeCohort {
  46         Normal(final ShardDataTree dataTree, final DataTreeModification transaction,
  47             final TransactionIdentifier transactionId, final CompositeDataTreeCohort userCohorts) {
  48             super(dataTree, transaction, transactionId, Preconditions.checkNotNull(userCohorts));
  49         }
  50
  51         @Override
  52         void throwCanCommitFailure() {
  53             // No-op
  54         }
  55     }
  56
  57     private static final Logger LOG = LoggerFactory.getLogger(SimpleShardDataTreeCohort.class);
  58
  59     private final DataTreeModification transaction;
  60     private final ShardDataTree dataTree;
  61     private final TransactionIdentifier transactionId;
  62     private final CompositeDataTreeCohort userCohorts;
  63
  64     private State state = State.READY;
  65     private DataTreeCandidateTip candidate;
  66     private FutureCallback<?> callback;
  67     private Exception nextFailure;
  68
  69     SimpleShardDataTreeCohort(final ShardDataTree dataTree, final DataTreeModification transaction,
  70             final TransactionIdentifier transactionId, final CompositeDataTreeCohort userCohorts) {
  71         this.dataTree = Preconditions.checkNotNull(dataTree);
  72         this.transaction = Preconditions.checkNotNull(transaction);
  73         this.transactionId = Preconditions.checkNotNull(transactionId);
  74         this.userCohorts = userCohorts;
  75     }
  76
  77     @Override
  78     public TransactionIdentifier getIdentifier() {
  79         return transactionId;
  80     }
  81
  82     @Override
  83     DataTreeCandidateTip getCandidate() {
  84         return candidate;
  85     }
  86
  87     @Override
  88     DataTreeModification getDataTreeModification() {
  89         return transaction;
  90     }
  91
  92     private void checkState(final State expected) {
  93         Preconditions.checkState(state == expected, "State %s does not match expected state %s", state, expected);
  94     }
  95
  96     @Override
  97     public void canCommit(final FutureCallback<Void> newCallback) {
  98         if (state == State.CAN_COMMIT_PENDING) {
  99             return;
 100         }
 101
 102         checkState(State.READY);
 103         this.callback = Preconditions.checkNotNull(newCallback);
 104         state = State.CAN_COMMIT_PENDING;
 105         dataTree.startCanCommit(this);
 106     }
 107
 108     @Override
 109     public void preCommit(final FutureCallback<DataTreeCandidate> newCallback) {
 110         checkState(State.CAN_COMMIT_COMPLETE);
 111         this.callback = Preconditions.checkNotNull(newCallback);
 112         state = State.PRE_COMMIT_PENDING;
 113
 114         if (nextFailure == null) {
 115             dataTree.startPreCommit(this);
 116         } else {
 117             failedPreCommit(nextFailure);
 118         }
 119     }
 120
 121     @Override
 122     public void abort(final FutureCallback<Void> abortCallback) {
 123         if (!dataTree.startAbort(this)) {
 124             abortCallback.onSuccess(null);
 125             return;
 126         }
 127
 128         candidate = null;
 129         state = State.ABORTED;
 130
 131         final Optional<List<Future<Object>>> maybeAborts = userCohorts.abort();
 132         if (!maybeAborts.isPresent()) {
 133             abortCallback.onSuccess(null);
 134             return;
 135         }
 136
 137         final Future<Iterable<Object>> aborts = Futures.sequence(maybeAborts.get(), ExecutionContexts.global());
 138         if (aborts.isCompleted()) {
 139             abortCallback.onSuccess(null);
 140             return;
 141         }
 142
 143         aborts.onComplete(new OnComplete<Iterable<Object>>() {
 144             @Override
 145             public void onComplete(final Throwable failure, final Iterable<Object> objs) {
 146                 if (failure != null) {
 147                     abortCallback.onFailure(failure);
 148                 } else {
 149                     abortCallback.onSuccess(null);
 150                 }
 151             }
 152         }, ExecutionContexts.global());
 153     }
 154
 155     @Override
 156     public void commit(final FutureCallback<UnsignedLong> newCallback) {
 157         checkState(State.PRE_COMMIT_COMPLETE);
 158         this.callback = Preconditions.checkNotNull(newCallback);
 159         state = State.COMMIT_PENDING;
 160
 161         if (nextFailure == null) {
 162             dataTree.startCommit(this, candidate);
 163         } else {
 164             failedCommit(nextFailure);
 165         }
 166     }
 167
 168     private <T> FutureCallback<T> switchState(final State newState) {
 169         @SuppressWarnings("unchecked")
 170         final FutureCallback<T> ret = (FutureCallback<T>) this.callback;
 171         this.callback = null;
 172         LOG.debug("Transaction {} changing state from {} to {}", transactionId, state, newState);
 173         this.state = newState;
 174         return ret;
 175     }
 176
 177     void setNewCandidate(final DataTreeCandidateTip dataTreeCandidate) {
 178         checkState(State.PRE_COMMIT_COMPLETE);
 179         this.candidate = Verify.verifyNotNull(dataTreeCandidate);
 180     }
 181
 182     void successfulCanCommit() {
 183         switchState(State.CAN_COMMIT_COMPLETE).onSuccess(null);
 184     }
 185
 186     void failedCanCommit(final Exception cause) {
 187         switchState(State.FAILED).onFailure(cause);
 188     }
 189
 190     /**
 191      * Run user-defined canCommit and preCommit hooks. We want to run these before we initiate persistence so that
 192      * any failure to validate is propagated before we record the transaction.
 193      *
 194      * @param dataTreeCandidate {@link DataTreeCandidate} under consideration
 195      * @throws ExecutionException if the operation fails
 196      * @throws TimeoutException if the operation times out
 197      */
 198     // FIXME: this should be asynchronous
 199     void userPreCommit(final DataTreeCandidate dataTreeCandidate) throws ExecutionException, TimeoutException {
 200         userCohorts.reset();
 201         userCohorts.canCommit(dataTreeCandidate);
 202         userCohorts.preCommit();
 203     }
 204
 205     void successfulPreCommit(final DataTreeCandidateTip dataTreeCandidate) {
 206         LOG.trace("Transaction {} prepared candidate {}", transaction, dataTreeCandidate);
 207         this.candidate = Verify.verifyNotNull(dataTreeCandidate);
 208         switchState(State.PRE_COMMIT_COMPLETE).onSuccess(dataTreeCandidate);
 209     }
 210
 211     void failedPreCommit(final Exception cause) {
 212         if (LOG.isTraceEnabled()) {
 213             LOG.trace("Transaction {} failed to prepare", transaction, cause);
 214         } else {
 215             LOG.error("Transaction {} failed to prepare", transactionId, cause);
 216         }
 217
 218         userCohorts.abort();
 219         switchState(State.FAILED).onFailure(cause);
 220     }
 221
 222     void successfulCommit(final UnsignedLong journalIndex) {
 223         try {
 224             userCohorts.commit();
 225         } catch (TimeoutException | ExecutionException e) {
 226             // We are probably dead, depending on what the cohorts end up doing
 227             LOG.error("User cohorts failed to commit", e);
 228         }
 229
 230         switchState(State.COMMITTED).onSuccess(journalIndex);
 231     }
 232
 233     void failedCommit(final Exception cause) {
 234         if (LOG.isTraceEnabled()) {
 235             LOG.trace("Transaction {} failed to commit", transaction, cause);
 236         } else {
 237             LOG.error("Transaction failed to commit", cause);
 238         }
 239
 240         userCohorts.abort();
 241         switchState(State.FAILED).onFailure(cause);
 242     }
 243
 244     @Override
 245     public State getState() {
 246         return state;
 247     }
 248
 249     void reportFailure(final Exception cause) {
 250         this.nextFailure = Preconditions.checkNotNull(cause);
 251     }
 252
 253     /**
 254      * If there is an initial failure, throw it so the caller can process it.
 255      *
 256      * @throws Exception reported failure.
 257      */
 258     abstract void throwCanCommitFailure() throws Exception;
 259
 260     @Override
 261     public boolean isFailed() {
 262         return state == State.FAILED || nextFailure != null;
 263     }
 264 }