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 com.google.common.base.MoreObjects.ToStringHelper;
  11 import com.google.common.base.Preconditions;
  12 import com.google.common.base.Verify;
  13 import com.google.common.primitives.UnsignedLong;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import java.util.Optional;
  16 import java.util.concurrent.CompletionStage;
  17 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  18 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  19 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateTip;
  20 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
  21 import org.slf4j.Logger;
  22 import org.slf4j.LoggerFactory;
  23
  24 final class SimpleShardDataTreeCohort extends ShardDataTreeCohort {
  25     private static final Logger LOG = LoggerFactory.getLogger(SimpleShardDataTreeCohort.class);
  26
  27     private final DataTreeModification transaction;
  28     private final ShardDataTree dataTree;
  29     private final TransactionIdentifier transactionId;
  30     private final CompositeDataTreeCohort userCohorts;
  31
  32     private State state = State.READY;
  33     private DataTreeCandidateTip candidate;
  34     private FutureCallback<?> callback;
  35     private Exception nextFailure;
  36
  37     SimpleShardDataTreeCohort(final ShardDataTree dataTree, final DataTreeModification transaction,
  38             final TransactionIdentifier transactionId, final CompositeDataTreeCohort userCohorts) {
  39         this.dataTree = Preconditions.checkNotNull(dataTree);
  40         this.transaction = Preconditions.checkNotNull(transaction);
  41         this.transactionId = Preconditions.checkNotNull(transactionId);
  42         this.userCohorts = Preconditions.checkNotNull(userCohorts);
  43     }
  44
  45     SimpleShardDataTreeCohort(final ShardDataTree dataTree, final DataTreeModification transaction,
  46         final TransactionIdentifier transactionId, final Exception nextFailure) {
  47         this.dataTree = Preconditions.checkNotNull(dataTree);
  48         this.transaction = Preconditions.checkNotNull(transaction);
  49         this.transactionId = Preconditions.checkNotNull(transactionId);
  50         this.userCohorts = null;
  51         this.nextFailure = Preconditions.checkNotNull(nextFailure);
  52     }
  53
  54     @Override
  55     public TransactionIdentifier getIdentifier() {
  56         return transactionId;
  57     }
  58
  59     @Override
  60     DataTreeCandidateTip getCandidate() {
  61         return candidate;
  62     }
  63
  64     @Override
  65     DataTreeModification getDataTreeModification() {
  66         return transaction;
  67     }
  68
  69     private void checkState(final State expected) {
  70         Preconditions.checkState(state == expected, "State %s does not match expected state %s", state, expected);
  71     }
  72
  73     @Override
  74     public void canCommit(final FutureCallback<Void> newCallback) {
  75         if (state == State.CAN_COMMIT_PENDING) {
  76             return;
  77         }
  78
  79         checkState(State.READY);
  80         this.callback = Preconditions.checkNotNull(newCallback);
  81         state = State.CAN_COMMIT_PENDING;
  82
  83         if (nextFailure == null) {
  84             dataTree.startCanCommit(this);
  85         } else {
  86             failedCanCommit(nextFailure);
  87         }
  88     }
  89
  90     @Override
  91     public void preCommit(final FutureCallback<DataTreeCandidate> newCallback) {
  92         checkState(State.CAN_COMMIT_COMPLETE);
  93         this.callback = Preconditions.checkNotNull(newCallback);
  94         state = State.PRE_COMMIT_PENDING;
  95
  96         if (nextFailure == null) {
  97             dataTree.startPreCommit(this);
  98         } else {
  99             failedPreCommit(nextFailure);
 100         }
 101     }
 102
 103     @Override
 104     public void abort(final FutureCallback<Void> abortCallback) {
 105         if (!dataTree.startAbort(this)) {
 106             abortCallback.onSuccess(null);
 107             return;
 108         }
 109
 110         candidate = null;
 111         state = State.ABORTED;
 112
 113         final Optional<CompletionStage<?>> maybeAborts = userCohorts.abort();
 114         if (!maybeAborts.isPresent()) {
 115             abortCallback.onSuccess(null);
 116             return;
 117         }
 118
 119         maybeAborts.get().whenComplete((noop, failure) -> {
 120             if (failure != null) {
 121                 abortCallback.onFailure(failure);
 122             } else {
 123                 abortCallback.onSuccess(null);
 124             }
 125         });
 126     }
 127
 128     @Override
 129     public void commit(final FutureCallback<UnsignedLong> newCallback) {
 130         checkState(State.PRE_COMMIT_COMPLETE);
 131         this.callback = Preconditions.checkNotNull(newCallback);
 132         state = State.COMMIT_PENDING;
 133
 134         if (nextFailure == null) {
 135             dataTree.startCommit(this, candidate);
 136         } else {
 137             failedCommit(nextFailure);
 138         }
 139     }
 140
 141     private <T> FutureCallback<T> switchState(final State newState) {
 142         @SuppressWarnings("unchecked")
 143         final FutureCallback<T> ret = (FutureCallback<T>) this.callback;
 144         this.callback = null;
 145         LOG.debug("Transaction {} changing state from {} to {}", transactionId, state, newState);
 146         this.state = newState;
 147         return ret;
 148     }
 149
 150     void setNewCandidate(final DataTreeCandidateTip dataTreeCandidate) {
 151         checkState(State.PRE_COMMIT_COMPLETE);
 152         this.candidate = Verify.verifyNotNull(dataTreeCandidate);
 153     }
 154
 155     void successfulCanCommit() {
 156         switchState(State.CAN_COMMIT_COMPLETE).onSuccess(null);
 157     }
 158
 159     void failedCanCommit(final Exception cause) {
 160         switchState(State.FAILED).onFailure(cause);
 161     }
 162
 163     /**
 164      * Run user-defined canCommit and preCommit hooks. We want to run these before we initiate persistence so that
 165      * any failure to validate is propagated before we record the transaction.
 166      *
 167      * @param dataTreeCandidate {@link DataTreeCandidate} under consideration
 168      * @param futureCallback the callback to invoke on completion, which may be immediate or async.
 169      */
 170     void userPreCommit(final DataTreeCandidate dataTreeCandidate, final FutureCallback<Void> futureCallback) {
 171         userCohorts.reset();
 172
 173         final Optional<CompletionStage<Void>> maybeCanCommitFuture = userCohorts.canCommit(dataTreeCandidate);
 174         if (!maybeCanCommitFuture.isPresent()) {
 175             doUserPreCommit(futureCallback);
 176             return;
 177         }
 178
 179         maybeCanCommitFuture.get().whenComplete((noop, failure) -> {
 180             if (failure != null) {
 181                 futureCallback.onFailure(failure);
 182             } else {
 183                 doUserPreCommit(futureCallback);
 184             }
 185         });
 186     }
 187
 188     private void doUserPreCommit(final FutureCallback<Void> futureCallback) {
 189         final Optional<CompletionStage<Void>> maybePreCommitFuture = userCohorts.preCommit();
 190         if (!maybePreCommitFuture.isPresent()) {
 191             futureCallback.onSuccess(null);
 192             return;
 193         }
 194
 195         maybePreCommitFuture.get().whenComplete((noop, failure) -> {
 196             if (failure != null) {
 197                 futureCallback.onFailure(failure);
 198             } else {
 199                 futureCallback.onSuccess(null);
 200             }
 201         });
 202     }
 203
 204     void successfulPreCommit(final DataTreeCandidateTip dataTreeCandidate) {
 205         LOG.trace("Transaction {} prepared candidate {}", transaction, dataTreeCandidate);
 206         this.candidate = Verify.verifyNotNull(dataTreeCandidate);
 207         switchState(State.PRE_COMMIT_COMPLETE).onSuccess(dataTreeCandidate);
 208     }
 209
 210     void failedPreCommit(final Throwable cause) {
 211         if (LOG.isTraceEnabled()) {
 212             LOG.trace("Transaction {} failed to prepare", transaction, cause);
 213         } else {
 214             LOG.error("Transaction {} failed to prepare", transactionId, cause);
 215         }
 216
 217         userCohorts.abort();
 218         switchState(State.FAILED).onFailure(cause);
 219     }
 220
 221     void successfulCommit(final UnsignedLong journalIndex, final Runnable onComplete) {
 222         final Optional<CompletionStage<Void>> maybeCommitFuture = userCohorts.commit();
 223         if (!maybeCommitFuture.isPresent()) {
 224             finishSuccessfulCommit(journalIndex, onComplete);
 225             return;
 226         }
 227
 228         maybeCommitFuture.get().whenComplete((noop, failure) -> {
 229             if (failure != null) {
 230                 LOG.error("User cohorts failed to commit", failure);
 231             }
 232
 233             finishSuccessfulCommit(journalIndex, onComplete);
 234         });
 235     }
 236
 237     private void finishSuccessfulCommit(final UnsignedLong journalIndex, final Runnable onComplete) {
 238         switchState(State.COMMITTED).onSuccess(journalIndex);
 239         onComplete.run();
 240     }
 241
 242     void failedCommit(final Exception cause) {
 243         if (LOG.isTraceEnabled()) {
 244             LOG.trace("Transaction {} failed to commit", transaction, cause);
 245         } else {
 246             LOG.error("Transaction failed to commit", cause);
 247         }
 248
 249         userCohorts.abort();
 250         switchState(State.FAILED).onFailure(cause);
 251     }
 252
 253     @Override
 254     public State getState() {
 255         return state;
 256     }
 257
 258     void reportFailure(final Exception cause) {
 259         if (nextFailure == null) {
 260             this.nextFailure = Preconditions.checkNotNull(cause);
 261         } else {
 262             LOG.debug("Transaction {} already has a set failure, not updating it", transactionId, cause);
 263         }
 264     }
 265
 266     @Override
 267     public boolean isFailed() {
 268         return state == State.FAILED || nextFailure != null;
 269     }
 270
 271     @Override
 272     ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
 273         return super.addToStringAttributes(toStringHelper).add("nextFailure", nextFailure);
 274     }
 275 }