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.OnComplete;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.SettableFuture;
  15 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
  16 import org.opendaylight.controller.cluster.datastore.messages.AbstractRead;
  17 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  18 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  19 import org.opendaylight.controller.cluster.datastore.messages.SerializableMessage;
  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 short remoteTransactionVersion;
  40     private final OperationLimiter limiter;
  41
  42     private BatchedModifications batchedModifications;
  43     private int totalBatchedModificationsSent;
  44
  45     protected RemoteTransactionContext(TransactionIdentifier identifier, ActorSelection actor,
  46             ActorContext actorContext, short remoteTransactionVersion, OperationLimiter limiter) {
  47         super(identifier);
  48         this.limiter = Preconditions.checkNotNull(limiter);
  49         this.actor = actor;
  50         this.actorContext = actorContext;
  51         this.remoteTransactionVersion = remoteTransactionVersion;
  52     }
  53
  54     private Future<Object> completeOperation(Future<Object> operationFuture){
  55         operationFuture.onComplete(limiter, actorContext.getClientDispatcher());
  56         return operationFuture;
  57     }
  58
  59     private ActorSelection getActor() {
  60         return actor;
  61     }
  62
  63     protected ActorContext getActorContext() {
  64         return actorContext;
  65     }
  66
  67     protected short getRemoteTransactionVersion() {
  68         return remoteTransactionVersion;
  69     }
  70
  71     protected Future<Object> executeOperationAsync(SerializableMessage msg) {
  72         return completeOperation(actorContext.executeOperationAsync(getActor(), msg.toSerializable()));
  73     }
  74
  75     @Override
  76     public void closeTransaction() {
  77         LOG.debug("Tx {} closeTransaction called", getIdentifier());
  78         TransactionContextCleanup.untrack(this);
  79
  80         actorContext.sendOperationAsync(getActor(), new CloseTransaction(remoteTransactionVersion).toSerializable());
  81     }
  82
  83     @Override
  84     public boolean supportsDirectCommit() {
  85         return true;
  86     }
  87
  88     @Override
  89     public Future<Object> directCommit() {
  90         LOG.debug("Tx {} directCommit called", getIdentifier());
  91
  92         // Send the remaining batched modifications, if any, with the ready flag set.
  93
  94         return sendBatchedModifications(true, true);
  95     }
  96
  97     @Override
  98     public Future<ActorSelection> readyTransaction() {
  99         logModificationCount();
 100
 101         LOG.debug("Tx {} readyTransaction called", getIdentifier());
 102
 103         // Send the remaining batched modifications, if any, with the ready flag set.
 104
 105         Future<Object> lastModificationsFuture = sendBatchedModifications(true, false);
 106
 107         return transformReadyReply(lastModificationsFuture);
 108     }
 109
 110     protected Future<ActorSelection> transformReadyReply(final Future<Object> readyReplyFuture) {
 111         // Transform the last reply Future into a Future that returns the cohort actor path from
 112         // the last reply message. That's the end result of the ready operation.
 113
 114         return TransactionReadyReplyMapper.transform(readyReplyFuture, actorContext, getIdentifier());
 115     }
 116
 117     private BatchedModifications newBatchedModifications() {
 118         return new BatchedModifications(getIdentifier().toString(), remoteTransactionVersion, getIdentifier().getChainId());
 119     }
 120
 121     private void batchModification(Modification modification) {
 122         incrementModificationCount();
 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(boolean ready, 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             if(LOG.isDebugEnabled()) {
 147                 LOG.debug("Tx {} sending {} batched modifications, ready: {}", getIdentifier(),
 148                         batchedModifications.getModifications().size(), ready);
 149             }
 150
 151             batchedModifications.setReady(ready);
 152             batchedModifications.setDoCommitOnReady(doCommitOnReady);
 153             batchedModifications.setTotalMessagesSent(++totalBatchedModificationsSent);
 154             sent = executeOperationAsync(batchedModifications);
 155
 156             if(ready) {
 157                 batchedModifications = null;
 158             } else {
 159                 batchedModifications = newBatchedModifications();
 160             }
 161         }
 162
 163         return sent;
 164     }
 165
 166     @Override
 167     public void executeModification(AbstractModification modification) {
 168         if(LOG.isDebugEnabled()) {
 169             LOG.debug("Tx {} executeModification {} called path = {}", getIdentifier(), modification.getClass()
 170                     .getSimpleName(), modification.getPath());
 171         }
 172
 173         acquireOperation();
 174         batchModification(modification);
 175     }
 176
 177     @Override
 178     public <T> void executeRead(final AbstractRead<T> readCmd, final SettableFuture<T> returnFuture) {
 179         if(LOG.isDebugEnabled()) {
 180             LOG.debug("Tx {} executeRead {} called path = {}", getIdentifier(), readCmd.getClass().getSimpleName(),
 181                     readCmd.getPath());
 182         }
 183
 184         // Send any batched modifications. This is necessary to honor the read uncommitted semantics of the
 185         // public API contract.
 186
 187         acquireOperation();
 188         sendBatchedModifications();
 189
 190         OnComplete<Object> onComplete = new OnComplete<Object>() {
 191             @Override
 192             public void onComplete(Throwable failure, Object response) throws Throwable {
 193                 if(failure != null) {
 194                     if(LOG.isDebugEnabled()) {
 195                         LOG.debug("Tx {} {} operation failed: {}", getIdentifier(), readCmd.getClass().getSimpleName(),
 196                                 failure);
 197                     }
 198                     returnFuture.setException(new ReadFailedException("Error checking " + readCmd.getClass().getSimpleName()
 199                             + " for path " + readCmd.getPath(), failure));
 200                 } else {
 201                     if(LOG.isDebugEnabled()) {
 202                         LOG.debug("Tx {} {} operation succeeded", getIdentifier(), readCmd.getClass().getSimpleName());
 203                     }
 204                     readCmd.processResponse(response, returnFuture);
 205                 }
 206             }
 207         };
 208
 209         Future<Object> future = executeOperationAsync(readCmd.asVersion(remoteTransactionVersion));
 210
 211         future.onComplete(onComplete, actorContext.getClientDispatcher());
 212     }
 213
 214     /**
 215      * Acquire operation from the limiter if the hand-off has completed. If
 216      * the hand-off is still ongoing, this method does nothing.
 217      */
 218     private final void acquireOperation() {
 219         if (isOperationHandOffComplete()) {
 220             limiter.acquire();
 221         }
 222     }
 223
 224     @Override
 225     public boolean usesOperationLimiting() {
 226         return true;
 227     }
 228 }