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