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(TransactionIdentifier identifier, ActorSelection actor,
  54             ActorContext actorContext, short remoteTransactionVersion, 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() {
  79         LOG.debug("Tx {} directCommit called", getIdentifier());
  80
  81         // Send the remaining batched modifications, if any, with the ready flag set.
  82
  83         return sendBatchedModifications(true, true);
  84     }
  85
  86     @Override
  87     public Future<ActorSelection> readyTransaction() {
  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         Future<Object> lastModificationsFuture = sendBatchedModifications(true, false);
  95
  96         return transformReadyReply(lastModificationsFuture);
  97     }
  98
  99     protected Future<ActorSelection> transformReadyReply(final Future<Object> readyReplyFuture) {
 100         // Transform the last reply Future into a Future that returns the cohort actor path from
 101         // the last reply message. That's the end result of the ready operation.
 102
 103         return TransactionReadyReplyMapper.transform(readyReplyFuture, actorContext, getIdentifier());
 104     }
 105
 106     private BatchedModifications newBatchedModifications() {
 107         return new BatchedModifications(getIdentifier(), getTransactionVersion());
 108     }
 109
 110     private void batchModification(Modification modification, boolean havePermit) {
 111         incrementModificationCount();
 112         if (havePermit) {
 113             ++batchPermits;
 114         }
 115
 116         if (batchedModifications == null) {
 117             batchedModifications = newBatchedModifications();
 118         }
 119
 120         batchedModifications.addModification(modification);
 121
 122         if (batchedModifications.getModifications().size()
 123                 >= actorContext.getDatastoreContext().getShardBatchedModificationCount()) {
 124             sendBatchedModifications();
 125         }
 126     }
 127
 128     protected Future<Object> sendBatchedModifications() {
 129         return sendBatchedModifications(false, false);
 130     }
 131
 132     protected Future<Object> sendBatchedModifications(boolean ready, boolean doCommitOnReady) {
 133         Future<Object> sent = null;
 134         if (ready || batchedModifications != null && !batchedModifications.getModifications().isEmpty()) {
 135             if (batchedModifications == null) {
 136                 batchedModifications = newBatchedModifications();
 137             }
 138
 139             LOG.debug("Tx {} sending {} batched modifications, ready: {}", getIdentifier(),
 140                     batchedModifications.getModifications().size(), ready);
 141
 142             batchedModifications.setReady(ready);
 143             batchedModifications.setDoCommitOnReady(doCommitOnReady);
 144             batchedModifications.setTotalMessagesSent(++totalBatchedModificationsSent);
 145
 146             final BatchedModifications toSend = batchedModifications;
 147             final int permitsToRelease = batchPermits;
 148             batchPermits = 0;
 149
 150             if (ready) {
 151                 batchedModifications = null;
 152             } else {
 153                 batchedModifications = newBatchedModifications();
 154
 155                 final Throwable failure = failedModification;
 156                 if (failure != null) {
 157                     // We have observed a modification failure, it does not make sense to send this batch. This speeds
 158                     // up the time when the application could be blocked due to messages timing out and operation
 159                     // limiter kicking in.
 160                     LOG.debug("Tx {} modifications previously failed, not sending a non-ready batch", getIdentifier());
 161                     limiter.release(permitsToRelease);
 162                     return Futures.failed(failure);
 163                 }
 164             }
 165
 166             sent = actorContext.executeOperationAsync(getActor(), toSend.toSerializable(),
 167                 actorContext.getTransactionCommitOperationTimeout());
 168             sent.onComplete(new OnComplete<Object>() {
 169                 @Override
 170                 public void onComplete(Throwable failure, Object success) {
 171                     if (failure != null) {
 172                         LOG.debug("Tx {} modifications failed", getIdentifier(), failure);
 173                         failedModification = failure;
 174                     } else {
 175                         LOG.debug("Tx {} modifications completed with {}", getIdentifier(), success);
 176                     }
 177                     limiter.release(permitsToRelease);
 178                 }
 179             }, actorContext.getClientDispatcher());
 180         }
 181
 182         return sent;
 183     }
 184
 185     @Override
 186     public void executeModification(AbstractModification modification) {
 187         LOG.debug("Tx {} executeModification {} called path = {}", getIdentifier(),
 188                 modification.getClass().getSimpleName(), modification.getPath());
 189
 190         final boolean havePermit = failedModification == null && acquireOperation();
 191         batchModification(modification, havePermit);
 192     }
 193
 194     @Override
 195     public <T> void executeRead(final AbstractRead<T> readCmd, final SettableFuture<T> returnFuture) {
 196         LOG.debug("Tx {} executeRead {} called path = {}", getIdentifier(), readCmd.getClass().getSimpleName(),
 197                 readCmd.getPath());
 198
 199         final Throwable failure = failedModification;
 200         if (failure != null) {
 201             // If we know there was a previous modification failure, we must not send a read request, as it risks
 202             // returning incorrect data. We check this before acquiring an operation simply because we want the app
 203             // to complete this transaction as soon as possible.
 204             returnFuture.setException(new ReadFailedException("Previous modification failed, cannot "
 205                     + readCmd.getClass().getSimpleName() + " for path " + readCmd.getPath(), failure));
 206             return;
 207         }
 208
 209         // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the
 210         // public API contract.
 211
 212         final boolean havePermit = acquireOperation();
 213         sendBatchedModifications();
 214
 215         OnComplete<Object> onComplete = new OnComplete<Object>() {
 216             @Override
 217             public void onComplete(Throwable failure, Object response) {
 218                 // We have previously acquired an operation, now release it, no matter what happened
 219                 if (havePermit) {
 220                     limiter.release();
 221                 }
 222
 223                 if (failure != null) {
 224                     LOG.debug("Tx {} {} operation failed: {}", getIdentifier(), readCmd.getClass().getSimpleName(),
 225                             failure);
 226
 227                     returnFuture.setException(new ReadFailedException("Error checking "
 228                         + readCmd.getClass().getSimpleName() + " for path " + readCmd.getPath(), failure));
 229                 } else {
 230                     LOG.debug("Tx {} {} operation succeeded", getIdentifier(), readCmd.getClass().getSimpleName());
 231                     readCmd.processResponse(response, returnFuture);
 232                 }
 233             }
 234         };
 235
 236         final Future<Object> future = actorContext.executeOperationAsync(getActor(),
 237             readCmd.asVersion(getTransactionVersion()).toSerializable(), actorContext.getOperationTimeout());
 238         future.onComplete(onComplete, actorContext.getClientDispatcher());
 239     }
 240
 241     /**
 242      * Acquire operation from the limiter if the hand-off has completed. If the hand-off is still ongoing, this method
 243      * does nothing.
 244      *
 245      * @return True if a permit was successfully acquired, false otherwise
 246      */
 247     private boolean acquireOperation() {
 248         return isOperationHandOffComplete() && limiter.acquire();
 249     }
 250
 251     @Override
 252     public boolean usesOperationLimiting() {
 253         return true;
 254     }
 255 }