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.OnComplete;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Verify;
  14 import com.google.common.primitives.UnsignedLong;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  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 final class SimpleShardDataTreeCohort extends ShardDataTreeCohort {
  30     private static final Logger LOG = LoggerFactory.getLogger(SimpleShardDataTreeCohort.class);
  31     private static final ListenableFuture<Void> VOID_FUTURE = Futures.immediateFuture(null);
  32     private final DataTreeModification transaction;
  33     private final ShardDataTree dataTree;
  34     private final TransactionIdentifier transactionId;
  35     private final CompositeDataTreeCohort userCohorts;
  36
  37     private State state = State.READY;
  38     private DataTreeCandidateTip candidate;
  39     private FutureCallback<?> callback;
  40     private Exception nextFailure;
  41
  42     SimpleShardDataTreeCohort(final ShardDataTree dataTree, final DataTreeModification transaction,
  43             final TransactionIdentifier transactionId, final CompositeDataTreeCohort userCohorts) {
  44         this.dataTree = Preconditions.checkNotNull(dataTree);
  45         this.transaction = Preconditions.checkNotNull(transaction);
  46         this.transactionId = Preconditions.checkNotNull(transactionId);
  47         this.userCohorts = Preconditions.checkNotNull(userCohorts);
  48     }
  49
  50     @Override
  51     public TransactionIdentifier getIdentifier() {
  52         return transactionId;
  53     }
  54
  55     @Override
  56     DataTreeCandidateTip getCandidate() {
  57         return candidate;
  58     }
  59
  60     @Override
  61
  62     DataTreeModification getDataTreeModification() {
  63         return transaction;
  64     }
  65
  66     private void checkState(final State expected) {
  67         Preconditions.checkState(state == expected, "State %s does not match expected state %s", state, expected);
  68     }
  69
  70     @Override
  71     public void canCommit(final FutureCallback<Void> newCallback) {
  72         if (state == State.CAN_COMMIT_PENDING) {
  73             return;
  74         }
  75
  76         checkState(State.READY);
  77         this.callback = Preconditions.checkNotNull(newCallback);
  78         state = State.CAN_COMMIT_PENDING;
  79         dataTree.startCanCommit(this);
  80     }
  81
  82     @Override
  83     public void preCommit(final FutureCallback<DataTreeCandidate> newCallback) {
  84         checkState(State.CAN_COMMIT_COMPLETE);
  85         this.callback = Preconditions.checkNotNull(newCallback);
  86         state = State.PRE_COMMIT_PENDING;
  87
  88         if (nextFailure == null) {
  89             dataTree.startPreCommit(this);
  90         } else {
  91             failedPreCommit(nextFailure);
  92         }
  93     }
  94
  95
  96     @Override
  97     public void abort(final FutureCallback<Void> callback) {
  98         dataTree.startAbort(this);
  99         state = State.ABORTED;
 100
 101         final Optional<Future<Iterable<Object>>> maybeAborts = userCohorts.abort();
 102         if (!maybeAborts.isPresent()) {
 103             callback.onSuccess(null);
 104             return;
 105         }
 106
 107         final Future<Iterable<Object>> aborts = maybeAborts.get();
 108         if (aborts.isCompleted()) {
 109             callback.onSuccess(null);
 110             return;
 111         }
 112
 113         aborts.onComplete(new OnComplete<Iterable<Object>>() {
 114             @Override
 115             public void onComplete(final Throwable failure, final Iterable<Object> objs) {
 116                 if (failure != null) {
 117                     callback.onFailure(failure);
 118                 } else {
 119                     callback.onSuccess(null);
 120                 }
 121             }
 122         }, ExecutionContexts.global());
 123     }
 124
 125     @Override
 126     public void commit(final FutureCallback<UnsignedLong> newCallback) {
 127         checkState(State.PRE_COMMIT_COMPLETE);
 128         this.callback = Preconditions.checkNotNull(newCallback);
 129         state = State.COMMIT_PENDING;
 130         dataTree.startCommit(this, candidate);
 131     }
 132
 133     private <T> FutureCallback<T> switchState(final State newState) {
 134         @SuppressWarnings("unchecked")
 135         final FutureCallback<T> ret = (FutureCallback<T>) this.callback;
 136         this.callback = null;
 137         LOG.debug("Transaction {} changing state from {} to {}", transactionId, state, newState);
 138         this.state = newState;
 139         return ret;
 140     }
 141
 142     void successfulCanCommit() {
 143         switchState(State.CAN_COMMIT_COMPLETE).onSuccess(null);
 144     }
 145
 146     void failedCanCommit(final Exception cause) {
 147         switchState(State.FAILED).onFailure(cause);
 148     }
 149
 150     /**
 151      * Run user-defined canCommit and preCommit hooks. We want to run these before we initiate persistence so that
 152      * any failure to validate is propagated before we record the transaction.
 153      *
 154      * @param dataTreeCandidate {@link DataTreeCandidate} under consideration
 155      * @throws ExecutionException if the operation fails
 156      * @throws TimeoutException if the operation times out
 157      */
 158     // FIXME: this should be asynchronous
 159     void userPreCommit(final DataTreeCandidate dataTreeCandidate) throws ExecutionException, TimeoutException {
 160         userCohorts.canCommit(dataTreeCandidate);
 161         userCohorts.preCommit();
 162     }
 163
 164     void successfulPreCommit(final DataTreeCandidateTip dataTreeCandidate) {
 165         LOG.trace("Transaction {} prepared candidate {}", transaction, dataTreeCandidate);
 166         this.candidate = Verify.verifyNotNull(dataTreeCandidate);
 167         switchState(State.PRE_COMMIT_COMPLETE).onSuccess(dataTreeCandidate);
 168     }
 169
 170     void failedPreCommit(final Exception cause) {
 171         if (LOG.isTraceEnabled()) {
 172             LOG.trace("Transaction {} failed to prepare", transaction, cause);
 173         } else {
 174             LOG.error("Transaction {} failed to prepare", transactionId, cause);
 175         }
 176
 177         userCohorts.abort();
 178         switchState(State.FAILED).onFailure(cause);
 179     }
 180
 181     void successfulCommit(final UnsignedLong journalIndex) {
 182         try {
 183             userCohorts.commit();
 184         } catch (TimeoutException | ExecutionException e) {
 185             // We are probably dead, depending on what the cohorts end up doing
 186             LOG.error("User cohorts failed to commit", e);
 187         }
 188
 189         switchState(State.COMMITTED).onSuccess(journalIndex);
 190     }
 191
 192     void failedCommit(final Exception cause) {
 193         if (LOG.isTraceEnabled()) {
 194             LOG.trace("Transaction {} failed to commit", transaction, cause);
 195         } else {
 196             LOG.error("Transaction failed to commit", cause);
 197         }
 198
 199         userCohorts.abort();
 200         switchState(State.FAILED).onFailure(cause);
 201     }
 202
 203     @Override
 204     public State getState() {
 205         return state;
 206     }
 207
 208     void reportFailure(final Exception cause) {
 209         this.nextFailure = Preconditions.checkNotNull(cause);
 210     }
 211
 212     @Override
 213     public boolean isFailed() {
 214         return state == State.FAILED || nextFailure != null;
 215     }
 216 }