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

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  * Copyright (c) 2015 Cisco 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 java.util.concurrent.Semaphore;
  15 import java.util.concurrent.TimeUnit;
  16 import org.opendaylight.controller.cluster.datastore.compat.PreLithiumTransactionContextImpl;
  17 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  18 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
  19 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  20 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
  21 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  22 import org.slf4j.Logger;
  23 import org.slf4j.LoggerFactory;
  24 import scala.concurrent.Future;
  25 import scala.concurrent.duration.FiniteDuration;
  26
  27 /**
  28  * Handles creation of TransactionContext instances for remote transactions. This class creates
  29  * remote transactions, if necessary, by sending CreateTransaction messages with retries, up to a limit,
  30  * if the shard doesn't have a leader yet. This is done by scheduling a retry task after a short delay.
  31  * <p>
  32  * The end result from a completed CreateTransaction message is a TransactionContext that is
  33  * used to perform transaction operations. Transaction operations that occur before the
  34  * CreateTransaction completes are cache via a TransactionContextWrapper and executed once the
  35  * CreateTransaction completes, successfully or not.
  36  */
  37 final class RemoteTransactionContextSupport {
  38     private static final Logger LOG = LoggerFactory.getLogger(RemoteTransactionContextSupport.class);
  39
  40     /**
  41      * Time interval in between transaction create retries.
  42      */
  43     private static final FiniteDuration CREATE_TX_TRY_INTERVAL = FiniteDuration.create(1, TimeUnit.SECONDS);
  44
  45     private final TransactionProxy parent;
  46     private final String shardName;
  47
  48     /**
  49      * The target primary shard.
  50      */
  51     private volatile ActorSelection primaryShard;
  52     private volatile int createTxTries;
  53
  54     private final TransactionContextWrapper transactionContextAdapter;
  55
  56     RemoteTransactionContextSupport(final TransactionContextWrapper transactionContextAdapter, final TransactionProxy parent,
  57             final String shardName) {
  58         this.parent = Preconditions.checkNotNull(parent);
  59         this.shardName = shardName;
  60         this.transactionContextAdapter = transactionContextAdapter;
  61         createTxTries = (int) (parent.getActorContext().getDatastoreContext().
  62                 getShardLeaderElectionTimeout().duration().toMillis() /
  63                 CREATE_TX_TRY_INTERVAL.toMillis());
  64     }
  65
  66     String getShardName() {
  67         return shardName;
  68     }
  69
  70     private TransactionType getTransactionType() {
  71         return parent.getType();
  72     }
  73
  74     private ActorContext getActorContext() {
  75         return parent.getActorContext();
  76     }
  77
  78     private Semaphore getOperationLimiter() {
  79         return parent.getLimiter();
  80     }
  81
  82     private TransactionIdentifier getIdentifier() {
  83         return parent.getIdentifier();
  84     }
  85
  86     /**
  87      * Sets the target primary shard and initiates a CreateTransaction try.
  88      */
  89     void setPrimaryShard(ActorSelection primaryShard) {
  90         this.primaryShard = primaryShard;
  91
  92         if (getTransactionType() == TransactionType.WRITE_ONLY &&
  93                 getActorContext().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
  94             LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context",
  95                 getIdentifier(), primaryShard);
  96
  97             // For write-only Tx's we prepare the transaction modifications directly on the shard actor
  98             // to avoid the overhead of creating a separate transaction actor.
  99             // FIXME: can't assume the shard version is LITHIUM_VERSION - need to obtain it somehow.
 100             transactionContextAdapter.executePriorTransactionOperations(createValidTransactionContext(this.primaryShard,
 101                     this.primaryShard.path().toString(), DataStoreVersions.LITHIUM_VERSION));
 102         } else {
 103             tryCreateTransaction();
 104         }
 105     }
 106
 107     /**
 108      * Performs a CreateTransaction try async.
 109      */
 110     private void tryCreateTransaction() {
 111         if(LOG.isDebugEnabled()) {
 112             LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(), primaryShard);
 113         }
 114
 115         Object serializedCreateMessage = new CreateTransaction(getIdentifier().toString(),
 116             getTransactionType().ordinal(), getIdentifier().getChainId()).toSerializable();
 117
 118         Future<Object> createTxFuture = getActorContext().executeOperationAsync(primaryShard, serializedCreateMessage);
 119
 120         createTxFuture.onComplete(new OnComplete<Object>() {
 121             @Override
 122             public void onComplete(Throwable failure, Object response) {
 123                 onCreateTransactionComplete(failure, response);
 124             }
 125         }, getActorContext().getClientDispatcher());
 126     }
 127
 128     private void onCreateTransactionComplete(Throwable failure, Object response) {
 129         if(failure instanceof NoShardLeaderException) {
 130             // There's no leader for the shard yet - schedule and try again, unless we're out
 131             // of retries. Note: createTxTries is volatile as it may be written by different
 132             // threads however not concurrently, therefore decrementing it non-atomically here
 133             // is ok.
 134             if(--createTxTries > 0) {
 135                 LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
 136                     getIdentifier(), shardName);
 137
 138                 getActorContext().getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
 139                         new Runnable() {
 140                             @Override
 141                             public void run() {
 142                                 tryCreateTransaction();
 143                             }
 144                         }, getActorContext().getClientDispatcher());
 145                 return;
 146             }
 147         }
 148
 149         createTransactionContext(failure, response);
 150     }
 151
 152     private void createTransactionContext(Throwable failure, Object response) {
 153         // Create the TransactionContext from the response or failure. Store the new
 154         // TransactionContext locally until we've completed invoking the
 155         // TransactionOperations. This avoids thread timing issues which could cause
 156         // out-of-order TransactionOperations. Eg, on a modification operation, if the
 157         // TransactionContext is non-null, then we directly call the TransactionContext.
 158         // However, at the same time, the code may be executing the cached
 159         // TransactionOperations. So to avoid thus timing, we don't publish the
 160         // TransactionContext until after we've executed all cached TransactionOperations.
 161         TransactionContext localTransactionContext;
 162         if(failure != null) {
 163             LOG.debug("Tx {} Creating NoOpTransaction because of error", getIdentifier(), failure);
 164
 165             localTransactionContext = new NoOpTransactionContext(failure, getIdentifier(), getOperationLimiter());
 166         } else if (CreateTransactionReply.SERIALIZABLE_CLASS.equals(response.getClass())) {
 167             localTransactionContext = createValidTransactionContext(
 168                     CreateTransactionReply.fromSerializable(response));
 169         } else {
 170             IllegalArgumentException exception = new IllegalArgumentException(String.format(
 171                     "Invalid reply type %s for CreateTransaction", response.getClass()));
 172
 173             localTransactionContext = new NoOpTransactionContext(exception, getIdentifier(), getOperationLimiter());
 174         }
 175
 176         transactionContextAdapter.executePriorTransactionOperations(localTransactionContext);
 177     }
 178
 179     private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
 180         LOG.debug("Tx {} Received {}", getIdentifier(), reply);
 181
 182         return createValidTransactionContext(getActorContext().actorSelection(reply.getTransactionPath()),
 183                 reply.getTransactionPath(), reply.getVersion());
 184     }
 185
 186     private TransactionContext createValidTransactionContext(ActorSelection transactionActor, String transactionPath,
 187             short remoteTransactionVersion) {
 188         // TxActor is always created where the leader of the shard is.
 189         // Check if TxActor is created in the same node
 190         boolean isTxActorLocal = getActorContext().isPathLocal(transactionPath);
 191         final TransactionContext ret;
 192
 193         if (remoteTransactionVersion < DataStoreVersions.LITHIUM_VERSION) {
 194             ret = new PreLithiumTransactionContextImpl(transactionPath, transactionActor, getIdentifier(),
 195                 getActorContext(), isTxActorLocal, remoteTransactionVersion, parent.getCompleter());
 196         } else {
 197             ret = new RemoteTransactionContext(transactionActor, getIdentifier(), getActorContext(),
 198                 isTxActorLocal, remoteTransactionVersion, parent.getCompleter());
 199         }
 200
 201         TransactionContextCleanup.track(this, ret);
 202         return ret;
 203     }
 204 }
 205