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