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

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   4  *
   5  * This program and the accompanying materials are made available under the
   6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  * and is available at http://www.eclipse.org/legal/epl-v10.html
   8  */
   9 package org.opendaylight.controller.cluster.datastore;
  10
  11 import akka.actor.ActorSelection;
  12 import akka.dispatch.Futures;
  13 import akka.dispatch.OnComplete;
  14 import com.google.common.base.Preconditions;
  15 import com.google.common.util.concurrent.SettableFuture;
  16 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  17 import org.opendaylight.controller.cluster.datastore.messages.AbstractRead;
  18 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  19 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  20 import org.opendaylight.controller.cluster.datastore.modification.AbstractModification;
  21 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  22 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  23 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26 import scala.concurrent.Future;
  27
  28 /**
  29  * Redirects front-end transaction operations to a shard for processing. Instances of this class are used
  30  * when the destination shard is remote to the caller.
  31  *
  32  * @author Thomas Pantelis
  33  */
  34 public class RemoteTransactionContext extends AbstractTransactionContext {
  35     private static final Logger LOG = LoggerFactory.getLogger(RemoteTransactionContext.class);
  36
  37     private final ActorContext actorContext;
  38     private final ActorSelection actor;
  39     private final OperationLimiter limiter;
  40
  41     private BatchedModifications batchedModifications;
  42     private int totalBatchedModificationsSent;
  43     private int batchPermits;
  44
  45     /**
  46      * We have observed a failed modification batch. This transaction context is effectively doomed, as the backend
  47      * does not have a correct view of the world. If this happens, we do not limit operations but rather short-cut them
  48      * to a either a no-op (modifications) or a failure (reads). Once the transaction is ready, though, we send the
  49      * message to resynchronize with the backend, sharing a 'lost message' failure path.
  50      */
  51     private volatile Throwable failedModification;
  52
  53     protected RemoteTransactionContext(final TransactionIdentifier identifier, final ActorSelection actor,
  54             final ActorContext actorContext, final short remoteTransactionVersion, final OperationLimiter limiter) {
  55         super(identifier, remoteTransactionVersion);
  56         this.limiter = Preconditions.checkNotNull(limiter);
  57         this.actor = actor;
  58         this.actorContext = actorContext;
  59     }
  60
  61     private ActorSelection getActor() {
  62         return actor;
  63     }
  64
  65     protected ActorContext getActorContext() {
  66         return actorContext;
  67     }
  68
  69     @Override
  70     public void closeTransaction() {
  71         LOG.debug("Tx {} closeTransaction called", getIdentifier());
  72         TransactionContextCleanup.untrack(this);
  73
  74         actorContext.sendOperationAsync(getActor(), new CloseTransaction(getTransactionVersion()).toSerializable());
  75     }
  76
  77     @Override
  78     public Future<Object> directCommit(final Boolean havePermit) {
  79         LOG.debug("Tx {} directCommit called", getIdentifier());
  80
  81         // Send the remaining batched modifications, if any, with the ready flag set.
  82         bumpPermits(havePermit);
  83         return sendBatchedModifications(true, true);
  84     }
  85
  86     @Override
  87     public Future<ActorSelection> readyTransaction(final Boolean havePermit) {
  88         logModificationCount();
  89
  90         LOG.debug("Tx {} readyTransaction called", getIdentifier());
  91
  92         // Send the remaining batched modifications, if any, with the ready flag set.
  93
  94         bumpPermits(havePermit);
  95         Future<Object> lastModificationsFuture = sendBatchedModifications(true, false);
  96
  97         return transformReadyReply(lastModificationsFuture);
  98     }
  99
 100     private void bumpPermits(final Boolean havePermit) {
 101         if (Boolean.TRUE.equals(havePermit)) {
 102             ++batchPermits;
 103         }
 104     }
 105
 106     protected Future<ActorSelection> transformReadyReply(final Future<Object> readyReplyFuture) {
 107         // Transform the last reply Future into a Future that returns the cohort actor path from
 108         // the last reply message. That's the end result of the ready operation.
 109
 110         return TransactionReadyReplyMapper.transform(readyReplyFuture, actorContext, getIdentifier());
 111     }
 112
 113     private BatchedModifications newBatchedModifications() {
 114         return new BatchedModifications(getIdentifier(), getTransactionVersion());
 115     }
 116
 117     private void batchModification(final Modification modification, final boolean havePermit) {
 118         incrementModificationCount();
 119         if (havePermit) {
 120             ++batchPermits;
 121         }
 122
 123         if (batchedModifications == null) {
 124             batchedModifications = newBatchedModifications();
 125         }
 126
 127         batchedModifications.addModification(modification);
 128
 129         if (batchedModifications.getModifications().size()
 130                 >= actorContext.getDatastoreContext().getShardBatchedModificationCount()) {
 131             sendBatchedModifications();
 132         }
 133     }
 134
 135     protected Future<Object> sendBatchedModifications() {
 136         return sendBatchedModifications(false, false);
 137     }
 138
 139     protected Future<Object> sendBatchedModifications(final boolean ready, final boolean doCommitOnReady) {
 140         Future<Object> sent = null;
 141         if (ready || batchedModifications != null && !batchedModifications.getModifications().isEmpty()) {
 142             if (batchedModifications == null) {
 143                 batchedModifications = newBatchedModifications();
 144             }
 145
 146             LOG.debug("Tx {} sending {} batched modifications, ready: {}", getIdentifier(),
 147                     batchedModifications.getModifications().size(), ready);
 148
 149             batchedModifications.setReady(ready);
 150             batchedModifications.setDoCommitOnReady(doCommitOnReady);
 151             batchedModifications.setTotalMessagesSent(++totalBatchedModificationsSent);
 152
 153             final BatchedModifications toSend = batchedModifications;
 154             final int permitsToRelease = batchPermits;
 155             batchPermits = 0;
 156
 157             if (ready) {
 158                 batchedModifications = null;
 159             } else {
 160                 batchedModifications = newBatchedModifications();
 161
 162                 final Throwable failure = failedModification;
 163                 if (failure != null) {
 164                     // We have observed a modification failure, it does not make sense to send this batch. This speeds
 165                     // up the time when the application could be blocked due to messages timing out and operation
 166                     // limiter kicking in.
 167                     LOG.debug("Tx {} modifications previously failed, not sending a non-ready batch", getIdentifier());
 168                     limiter.release(permitsToRelease);
 169                     return Futures.failed(failure);
 170                 }
 171             }
 172
 173             sent = actorContext.executeOperationAsync(getActor(), toSend.toSerializable(),
 174                 actorContext.getTransactionCommitOperationTimeout());
 175             sent.onComplete(new OnComplete<Object>() {
 176                 @Override
 177                 public void onComplete(final Throwable failure, final Object success) {
 178                     if (failure != null) {
 179                         LOG.debug("Tx {} modifications failed", getIdentifier(), failure);
 180                         failedModification = failure;
 181                     } else {
 182                         LOG.debug("Tx {} modifications completed with {}", getIdentifier(), success);
 183                     }
 184                     limiter.release(permitsToRelease);
 185                 }
 186             }, actorContext.getClientDispatcher());
 187         }
 188
 189         return sent;
 190     }
 191
 192     @Override
 193     public void executeModification(final AbstractModification modification, final Boolean havePermit) {
 194         LOG.debug("Tx {} executeModification {} called path = {}", getIdentifier(),
 195                 modification.getClass().getSimpleName(), modification.getPath());
 196
 197         final boolean permitToRelease;
 198         if (havePermit == null) {
 199             permitToRelease = failedModification == null && acquireOperation();
 200         } else {
 201             permitToRelease = havePermit.booleanValue();
 202         }
 203
 204         batchModification(modification, permitToRelease);
 205     }
 206
 207     @Override
 208     public <T> void executeRead(final AbstractRead<T> readCmd, final SettableFuture<T> returnFuture,
 209             final Boolean havePermit) {
 210         LOG.debug("Tx {} executeRead {} called path = {}", getIdentifier(), readCmd.getClass().getSimpleName(),
 211                 readCmd.getPath());
 212
 213         final Throwable failure = failedModification;
 214         if (failure != null) {
 215             // If we know there was a previous modification failure, we must not send a read request, as it risks
 216             // returning incorrect data. We check this before acquiring an operation simply because we want the app
 217             // to complete this transaction as soon as possible.
 218             returnFuture.setException(new ReadFailedException("Previous modification failed, cannot "
 219                     + readCmd.getClass().getSimpleName() + " for path " + readCmd.getPath(), failure));
 220             return;
 221         }
 222
 223         // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the
 224         // public API contract.
 225
 226         final boolean permitToRelease = havePermit == null ? acquireOperation() : havePermit.booleanValue();
 227         sendBatchedModifications();
 228
 229         OnComplete<Object> onComplete = new OnComplete<Object>() {
 230             @Override
 231             public void onComplete(final Throwable failure, final Object response) {
 232                 // We have previously acquired an operation, now release it, no matter what happened
 233                 if (permitToRelease) {
 234                     limiter.release();
 235                 }
 236
 237                 if (failure != null) {
 238                     LOG.debug("Tx {} {} operation failed: {}", getIdentifier(), readCmd.getClass().getSimpleName(),
 239                             failure);
 240
 241                     returnFuture.setException(new ReadFailedException("Error checking "
 242                         + readCmd.getClass().getSimpleName() + " for path " + readCmd.getPath(), failure));
 243                 } else {
 244                     LOG.debug("Tx {} {} operation succeeded", getIdentifier(), readCmd.getClass().getSimpleName());
 245                     readCmd.processResponse(response, returnFuture);
 246                 }
 247             }
 248         };
 249
 250         final Future<Object> future = actorContext.executeOperationAsync(getActor(),
 251             readCmd.asVersion(getTransactionVersion()).toSerializable(), actorContext.getOperationTimeout());
 252         future.onComplete(onComplete, actorContext.getClientDispatcher());
 253     }
 254
 255     /**
 256      * Acquire operation from the limiter if the hand-off has completed. If the hand-off is still ongoing, this method
 257      * does nothing.
 258      *
 259      * @return True if a permit was successfully acquired, false otherwise
 260      */
 261     private boolean acquireOperation() {
 262         Preconditions.checkState(isOperationHandOffComplete(),
 263             "Attempted to acquire execute operation permit for transaction %s on actor %s during handoff",
 264             getIdentifier(), actor);
 265
 266         if (limiter.acquire()) {
 267             return true;
 268         }
 269
 270         LOG.warn("Failed to acquire execute operation permit for transaction {} on actor {}", getIdentifier(), actor);
 271         return false;
 272     }
 273
 274     @Override
 275     public boolean usesOperationLimiting() {
 276         return true;
 277     }
 278 }