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