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

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.datastore;
  10
  11 import akka.actor.ActorSelection;
  12 import akka.dispatch.Mapper;
  13 import akka.dispatch.OnComplete;
  14 import com.google.common.annotations.VisibleForTesting;
  15 import com.google.common.base.FinalizablePhantomReference;
  16 import com.google.common.base.FinalizableReferenceQueue;
  17 import com.google.common.base.Optional;
  18 import com.google.common.base.Preconditions;
  19 import com.google.common.collect.Lists;
  20 import com.google.common.util.concurrent.CheckedFuture;
  21 import com.google.common.util.concurrent.ListenableFuture;
  22 import com.google.common.util.concurrent.SettableFuture;
  23 import java.util.ArrayList;
  24 import java.util.Collection;
  25 import java.util.Collections;
  26 import java.util.HashMap;
  27 import java.util.List;
  28 import java.util.Map;
  29 import java.util.concurrent.ConcurrentHashMap;
  30 import java.util.concurrent.Semaphore;
  31 import java.util.concurrent.TimeUnit;
  32 import java.util.concurrent.atomic.AtomicBoolean;
  33 import java.util.concurrent.atomic.AtomicLong;
  34 import javax.annotation.concurrent.GuardedBy;
  35 import org.opendaylight.controller.cluster.datastore.compat.PreLithiumTransactionContextImpl;
  36 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  37 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
  38 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  39 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  40 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
  41 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
  42 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  43 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  44 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  45 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  46 import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
  47 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  48 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  49 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  50 import org.slf4j.Logger;
  51 import org.slf4j.LoggerFactory;
  52 import scala.concurrent.Future;
  53 import scala.concurrent.Promise;
  54 import scala.concurrent.duration.FiniteDuration;
  55
  56 /**
  57  * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
  58  * <p>
  59  * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
  60  * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
  61  * be created on each of those shards by the TransactionProxy
  62  *</p>
  63  * <p>
  64  * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
  65  * shards will be executed.
  66  * </p>
  67  */
  68 public class TransactionProxy implements DOMStoreReadWriteTransaction {
  69
  70     public static enum TransactionType {
  71         READ_ONLY,
  72         WRITE_ONLY,
  73         READ_WRITE;
  74
  75         public static TransactionType fromInt(int type) {
  76             if(type == WRITE_ONLY.ordinal()) {
  77                 return WRITE_ONLY;
  78             } else if(type == READ_WRITE.ordinal()) {
  79                 return READ_WRITE;
  80             } else if(type == READ_ONLY.ordinal()) {
  81                 return READ_ONLY;
  82             } else {
  83                 throw new IllegalArgumentException("In TransactionType enum value" + type);
  84             }
  85         }
  86     }
  87
  88     static final Mapper<Throwable, Throwable> SAME_FAILURE_TRANSFORMER =
  89                                                               new Mapper<Throwable, Throwable>() {
  90         @Override
  91         public Throwable apply(Throwable failure) {
  92             return failure;
  93         }
  94     };
  95
  96     private static final AtomicLong counter = new AtomicLong();
  97
  98     private static final Logger LOG = LoggerFactory.getLogger(TransactionProxy.class);
  99
 100     /**
 101      * Time interval in between transaction create retries.
 102      */
 103     private static final FiniteDuration CREATE_TX_TRY_INTERVAL =
 104             FiniteDuration.create(1, TimeUnit.SECONDS);
 105
 106     /**
 107      * Used to enqueue the PhantomReferences for read-only TransactionProxy instances. The
 108      * FinalizableReferenceQueue is safe to use statically in an OSGi environment as it uses some
 109      * trickery to clean up its internal thread when the bundle is unloaded.
 110      */
 111     private static final FinalizableReferenceQueue phantomReferenceQueue =
 112                                                                   new FinalizableReferenceQueue();
 113
 114     /**
 115      * This stores the TransactionProxyCleanupPhantomReference instances statically, This is
 116      * necessary because PhantomReferences need a hard reference so they're not garbage collected.
 117      * Once finalized, the TransactionProxyCleanupPhantomReference removes itself from this map
 118      * and thus becomes eligible for garbage collection.
 119      */
 120     private static final Map<TransactionProxyCleanupPhantomReference,
 121                              TransactionProxyCleanupPhantomReference> phantomReferenceCache =
 122                                                                         new ConcurrentHashMap<>();
 123
 124     /**
 125      * A PhantomReference that closes remote transactions for a TransactionProxy when it's
 126      * garbage collected. This is used for read-only transactions as they're not explicitly closed
 127      * by clients. So the only way to detect that a transaction is no longer in use and it's safe
 128      * to clean up is when it's garbage collected. It's inexact as to when an instance will be GC'ed
 129      * but TransactionProxy instances should generally be short-lived enough to avoid being moved
 130      * to the old generation space and thus should be cleaned up in a timely manner as the GC
 131      * runs on the young generation (eden, swap1...) space much more frequently.
 132      */
 133     private static class TransactionProxyCleanupPhantomReference
 134                                            extends FinalizablePhantomReference<TransactionProxy> {
 135
 136         private final List<ActorSelection> remoteTransactionActors;
 137         private final AtomicBoolean remoteTransactionActorsMB;
 138         private final ActorContext actorContext;
 139         private final TransactionIdentifier identifier;
 140
 141         protected TransactionProxyCleanupPhantomReference(TransactionProxy referent) {
 142             super(referent, phantomReferenceQueue);
 143
 144             // Note we need to cache the relevant fields from the TransactionProxy as we can't
 145             // have a hard reference to the TransactionProxy instance itself.
 146
 147             remoteTransactionActors = referent.remoteTransactionActors;
 148             remoteTransactionActorsMB = referent.remoteTransactionActorsMB;
 149             actorContext = referent.actorContext;
 150             identifier = referent.identifier;
 151         }
 152
 153         @Override
 154         public void finalizeReferent() {
 155             LOG.trace("Cleaning up {} Tx actors for TransactionProxy {}",
 156                     remoteTransactionActors.size(), identifier);
 157
 158             phantomReferenceCache.remove(this);
 159
 160             // Access the memory barrier volatile to ensure all previous updates to the
 161             // remoteTransactionActors list are visible to this thread.
 162
 163             if(remoteTransactionActorsMB.get()) {
 164                 for(ActorSelection actor : remoteTransactionActors) {
 165                     LOG.trace("Sending CloseTransaction to {}", actor);
 166                     actorContext.sendOperationAsync(actor, CloseTransaction.INSTANCE.toSerializable());
 167                 }
 168             }
 169         }
 170     }
 171
 172     /**
 173      * Stores the remote Tx actors for each requested data store path to be used by the
 174      * PhantomReference to close the remote Tx's. This is only used for read-only Tx's. The
 175      * remoteTransactionActorsMB volatile serves as a memory barrier to publish updates to the
 176      * remoteTransactionActors list so they will be visible to the thread accessing the
 177      * PhantomReference.
 178      */
 179     private List<ActorSelection> remoteTransactionActors;
 180     private volatile AtomicBoolean remoteTransactionActorsMB;
 181
 182     /**
 183      * Stores the create transaction results per shard.
 184      */
 185     private final Map<String, TransactionFutureCallback> txFutureCallbackMap = new HashMap<>();
 186
 187     private final TransactionType transactionType;
 188     private final ActorContext actorContext;
 189     private final TransactionIdentifier identifier;
 190     private final String transactionChainId;
 191     private final SchemaContext schemaContext;
 192     private boolean inReadyState;
 193
 194     private volatile boolean initialized;
 195     private Semaphore operationLimiter;
 196     private OperationCompleter operationCompleter;
 197
 198     public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
 199         this(actorContext, transactionType, "");
 200     }
 201
 202     public TransactionProxy(ActorContext actorContext, TransactionType transactionType,
 203             String transactionChainId) {
 204         this.actorContext = Preconditions.checkNotNull(actorContext,
 205             "actorContext should not be null");
 206         this.transactionType = Preconditions.checkNotNull(transactionType,
 207             "transactionType should not be null");
 208         this.schemaContext = Preconditions.checkNotNull(actorContext.getSchemaContext(),
 209             "schemaContext should not be null");
 210         this.transactionChainId = transactionChainId;
 211
 212         String memberName = actorContext.getCurrentMemberName();
 213         if(memberName == null){
 214             memberName = "UNKNOWN-MEMBER";
 215         }
 216
 217         this.identifier = new TransactionIdentifier(memberName, counter.getAndIncrement());
 218
 219         LOG.debug("Created txn {} of type {} on chain {}", identifier, transactionType, transactionChainId);
 220     }
 221
 222     @VisibleForTesting
 223     List<Future<Object>> getRecordedOperationFutures() {
 224         List<Future<Object>> recordedOperationFutures = Lists.newArrayList();
 225         for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
 226             TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 227             if(transactionContext != null) {
 228                 recordedOperationFutures.addAll(transactionContext.getRecordedOperationFutures());
 229             }
 230         }
 231
 232         return recordedOperationFutures;
 233     }
 234
 235     @VisibleForTesting
 236     boolean hasTransactionContext() {
 237         for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
 238             TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 239             if(transactionContext != null) {
 240                 return true;
 241             }
 242         }
 243
 244         return false;
 245     }
 246
 247     @Override
 248     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
 249
 250         Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
 251                 "Read operation on write-only transaction is not allowed");
 252
 253         LOG.debug("Tx {} read {}", identifier, path);
 254
 255         throttleOperation();
 256
 257         final SettableFuture<Optional<NormalizedNode<?, ?>>> proxyFuture = SettableFuture.create();
 258
 259         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 260         txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
 261             @Override
 262             public void invoke(TransactionContext transactionContext) {
 263                 transactionContext.readData(path, proxyFuture);
 264             }
 265         });
 266
 267         return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
 268     }
 269
 270     @Override
 271     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
 272
 273         Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
 274                 "Exists operation on write-only transaction is not allowed");
 275
 276         LOG.debug("Tx {} exists {}", identifier, path);
 277
 278         throttleOperation();
 279
 280         final SettableFuture<Boolean> proxyFuture = SettableFuture.create();
 281
 282         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 283         txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
 284             @Override
 285             public void invoke(TransactionContext transactionContext) {
 286                 transactionContext.dataExists(path, proxyFuture);
 287             }
 288         });
 289
 290         return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
 291     }
 292
 293     private void checkModificationState() {
 294         Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
 295                 "Modification operation on read-only transaction is not allowed");
 296         Preconditions.checkState(!inReadyState,
 297                 "Transaction is sealed - further modifications are not allowed");
 298     }
 299
 300     private void throttleOperation() {
 301         throttleOperation(1);
 302     }
 303
 304     private void throttleOperation(int acquirePermits) {
 305         if(!initialized) {
 306             // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
 307             operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
 308             operationCompleter = new OperationCompleter(operationLimiter);
 309
 310             // Make sure we write this last because it's volatile and will also publish the non-volatile writes
 311             // above as well so they'll be visible to other threads.
 312             initialized = true;
 313         }
 314
 315         try {
 316             if(!operationLimiter.tryAcquire(acquirePermits,
 317                     actorContext.getDatastoreContext().getOperationTimeoutInSeconds(), TimeUnit.SECONDS)){
 318                 LOG.warn("Failed to acquire operation permit for transaction {}", getIdentifier());
 319             }
 320         } catch (InterruptedException e) {
 321             if(LOG.isDebugEnabled()) {
 322                 LOG.debug("Interrupted when trying to acquire operation permit for transaction " + getIdentifier().toString(), e);
 323             } else {
 324                 LOG.warn("Interrupted when trying to acquire operation permit for transaction {}", getIdentifier());
 325             }
 326         }
 327     }
 328
 329
 330     @Override
 331     public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 332
 333         checkModificationState();
 334
 335         LOG.debug("Tx {} write {}", identifier, path);
 336
 337         throttleOperation();
 338
 339         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 340         txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
 341             @Override
 342             public void invoke(TransactionContext transactionContext) {
 343                 transactionContext.writeData(path, data);
 344             }
 345         });
 346     }
 347
 348     @Override
 349     public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 350
 351         checkModificationState();
 352
 353         LOG.debug("Tx {} merge {}", identifier, path);
 354
 355         throttleOperation();
 356
 357         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 358         txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
 359             @Override
 360             public void invoke(TransactionContext transactionContext) {
 361                 transactionContext.mergeData(path, data);
 362             }
 363         });
 364     }
 365
 366     @Override
 367     public void delete(final YangInstanceIdentifier path) {
 368
 369         checkModificationState();
 370
 371         LOG.debug("Tx {} delete {}", identifier, path);
 372
 373         throttleOperation();
 374
 375         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 376         txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
 377             @Override
 378             public void invoke(TransactionContext transactionContext) {
 379                 transactionContext.deleteData(path);
 380             }
 381         });
 382     }
 383
 384     @Override
 385     public DOMStoreThreePhaseCommitCohort ready() {
 386
 387         checkModificationState();
 388
 389         inReadyState = true;
 390
 391         LOG.debug("Tx {} Readying {} transactions for commit", identifier,
 392                     txFutureCallbackMap.size());
 393
 394         if(txFutureCallbackMap.size() == 0) {
 395             onTransactionReady(Collections.<Future<ActorSelection>>emptyList());
 396             return NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
 397         }
 398
 399         throttleOperation(txFutureCallbackMap.size());
 400
 401         List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
 402
 403         for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
 404
 405             LOG.debug("Tx {} Readying transaction for shard {} chain {}", identifier,
 406                         txFutureCallback.getShardName(), transactionChainId);
 407
 408             final TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 409             final Future<ActorSelection> future;
 410             if (transactionContext != null) {
 411                 // avoid the creation of a promise and a TransactionOperation
 412                 future = transactionContext.readyTransaction();
 413             } else {
 414                 final Promise<ActorSelection> promise = akka.dispatch.Futures.promise();
 415                 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
 416                     @Override
 417                     public void invoke(TransactionContext transactionContext) {
 418                         promise.completeWith(transactionContext.readyTransaction());
 419                     }
 420                 });
 421                 future = promise.future();
 422             }
 423
 424             cohortFutures.add(future);
 425         }
 426
 427         onTransactionReady(cohortFutures);
 428
 429         return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
 430                 identifier.toString());
 431     }
 432
 433     /**
 434      * Method for derived classes to be notified when the transaction has been readied.
 435      *
 436      * @param cohortFutures the cohort Futures for each shard transaction.
 437      */
 438     protected void onTransactionReady(List<Future<ActorSelection>> cohortFutures) {
 439     }
 440
 441     @Override
 442     public Object getIdentifier() {
 443         return this.identifier;
 444     }
 445
 446     @Override
 447     public void close() {
 448         for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
 449             txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
 450                 @Override
 451                 public void invoke(TransactionContext transactionContext) {
 452                     transactionContext.closeTransaction();
 453                 }
 454             });
 455         }
 456
 457         txFutureCallbackMap.clear();
 458
 459         if(remoteTransactionActorsMB != null) {
 460             remoteTransactionActors.clear();
 461             remoteTransactionActorsMB.set(true);
 462         }
 463     }
 464
 465     private String shardNameFromIdentifier(YangInstanceIdentifier path){
 466         return ShardStrategyFactory.getStrategy(path).findShard(path);
 467     }
 468
 469     protected Future<ActorSelection> sendFindPrimaryShardAsync(String shardName) {
 470         return actorContext.findPrimaryShardAsync(shardName);
 471     }
 472
 473     private TransactionFutureCallback getOrCreateTxFutureCallback(YangInstanceIdentifier path) {
 474         String shardName = shardNameFromIdentifier(path);
 475         TransactionFutureCallback txFutureCallback = txFutureCallbackMap.get(shardName);
 476         if(txFutureCallback == null) {
 477             Future<ActorSelection> findPrimaryFuture = sendFindPrimaryShardAsync(shardName);
 478
 479             final TransactionFutureCallback newTxFutureCallback = new TransactionFutureCallback(shardName);
 480
 481             txFutureCallback = newTxFutureCallback;
 482             txFutureCallbackMap.put(shardName, txFutureCallback);
 483
 484             findPrimaryFuture.onComplete(new OnComplete<ActorSelection>() {
 485                 @Override
 486                 public void onComplete(Throwable failure, ActorSelection primaryShard) {
 487                     if(failure != null) {
 488                         newTxFutureCallback.createTransactionContext(failure, null);
 489                     } else {
 490                         newTxFutureCallback.setPrimaryShard(primaryShard);
 491                     }
 492                 }
 493             }, actorContext.getClientDispatcher());
 494         }
 495
 496         return txFutureCallback;
 497     }
 498
 499     public String getTransactionChainId() {
 500         return transactionChainId;
 501     }
 502
 503     protected ActorContext getActorContext() {
 504         return actorContext;
 505     }
 506
 507     /**
 508      * Interfaces for transaction operations to be invoked later.
 509      */
 510     private static interface TransactionOperation {
 511         void invoke(TransactionContext transactionContext);
 512     }
 513
 514     /**
 515      * Implements a Future OnComplete callback for a CreateTransaction message. This class handles
 516      * retries, up to a limit, if the shard doesn't have a leader yet. This is done by scheduling a
 517      * retry task after a short delay.
 518      * <p>
 519      * The end result from a completed CreateTransaction message is a TransactionContext that is
 520      * used to perform transaction operations. Transaction operations that occur before the
 521      * CreateTransaction completes are cache and executed once the CreateTransaction completes,
 522      * successfully or not.
 523      */
 524     private class TransactionFutureCallback extends OnComplete<Object> {
 525
 526         /**
 527          * The list of transaction operations to execute once the CreateTransaction completes.
 528          */
 529         @GuardedBy("txOperationsOnComplete")
 530         private final List<TransactionOperation> txOperationsOnComplete = Lists.newArrayList();
 531
 532         /**
 533          * The TransactionContext resulting from the CreateTransaction reply.
 534          */
 535         private volatile TransactionContext transactionContext;
 536
 537         /**
 538          * The target primary shard.
 539          */
 540         private volatile ActorSelection primaryShard;
 541
 542         private volatile int createTxTries = (int) (actorContext.getDatastoreContext().
 543                 getShardLeaderElectionTimeout().duration().toMillis() /
 544                 CREATE_TX_TRY_INTERVAL.toMillis());
 545
 546         private final String shardName;
 547
 548         TransactionFutureCallback(String shardName) {
 549             this.shardName = shardName;
 550         }
 551
 552         String getShardName() {
 553             return shardName;
 554         }
 555
 556         TransactionContext getTransactionContext() {
 557             return transactionContext;
 558         }
 559
 560
 561         /**
 562          * Sets the target primary shard and initiates a CreateTransaction try.
 563          */
 564         void setPrimaryShard(ActorSelection primaryShard) {
 565             this.primaryShard = primaryShard;
 566
 567             if(transactionType == TransactionType.WRITE_ONLY &&
 568                     actorContext.getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
 569                 LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context",
 570                         identifier, primaryShard);
 571
 572                 // For write-only Tx's we prepare the transaction modifications directly on the shard actor
 573                 // to avoid the overhead of creating a separate transaction actor.
 574                 // FIXME: can't assume the shard version is LITHIUM_VERSION - need to obtain it somehow.
 575                 executeTxOperatonsOnComplete(createValidTransactionContext(this.primaryShard,
 576                         this.primaryShard.path().toString(), DataStoreVersions.LITHIUM_VERSION));
 577             } else {
 578                 tryCreateTransaction();
 579             }
 580         }
 581
 582         /**
 583          * Adds a TransactionOperation to be executed after the CreateTransaction completes.
 584          */
 585         void addTxOperationOnComplete(TransactionOperation operation) {
 586             boolean invokeOperation = true;
 587             synchronized(txOperationsOnComplete) {
 588                 if(transactionContext == null) {
 589                     LOG.debug("Tx {} Adding operation on complete", identifier);
 590
 591                     invokeOperation = false;
 592                     txOperationsOnComplete.add(operation);
 593                 }
 594             }
 595
 596             if(invokeOperation) {
 597                 operation.invoke(transactionContext);
 598             }
 599         }
 600
 601         void enqueueTransactionOperation(final TransactionOperation op) {
 602
 603             if (transactionContext != null) {
 604                 op.invoke(transactionContext);
 605             } else {
 606                 // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
 607                 // callback to be executed after the Tx is created.
 608                 addTxOperationOnComplete(op);
 609             }
 610         }
 611
 612         /**
 613          * Performs a CreateTransaction try async.
 614          */
 615         private void tryCreateTransaction() {
 616             if(LOG.isDebugEnabled()) {
 617                 LOG.debug("Tx {} Primary shard {} found - trying create transaction", identifier, primaryShard);
 618             }
 619
 620             Object serializedCreateMessage = new CreateTransaction(identifier.toString(),
 621                     TransactionProxy.this.transactionType.ordinal(),
 622                     getTransactionChainId()).toSerializable();
 623
 624             Future<Object> createTxFuture = actorContext.executeOperationAsync(primaryShard, serializedCreateMessage);
 625
 626             createTxFuture.onComplete(this, actorContext.getClientDispatcher());
 627         }
 628
 629         @Override
 630         public void onComplete(Throwable failure, Object response) {
 631             if(failure instanceof NoShardLeaderException) {
 632                 // There's no leader for the shard yet - schedule and try again, unless we're out
 633                 // of retries. Note: createTxTries is volatile as it may be written by different
 634                 // threads however not concurrently, therefore decrementing it non-atomically here
 635                 // is ok.
 636                 if(--createTxTries > 0) {
 637                     LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
 638                             identifier, shardName);
 639
 640                     actorContext.getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
 641                             new Runnable() {
 642                                 @Override
 643                                 public void run() {
 644                                     tryCreateTransaction();
 645                                 }
 646                             }, actorContext.getClientDispatcher());
 647                     return;
 648                 }
 649             }
 650
 651             createTransactionContext(failure, response);
 652         }
 653
 654         private void createTransactionContext(Throwable failure, Object response) {
 655             // Mainly checking for state violation here to perform a volatile read of "initialized" to
 656             // ensure updates to operationLimter et al are visible to this thread (ie we're doing
 657             // "piggy-back" synchronization here).
 658             Preconditions.checkState(initialized, "Tx was not propertly initialized.");
 659
 660             // Create the TransactionContext from the response or failure. Store the new
 661             // TransactionContext locally until we've completed invoking the
 662             // TransactionOperations. This avoids thread timing issues which could cause
 663             // out-of-order TransactionOperations. Eg, on a modification operation, if the
 664             // TransactionContext is non-null, then we directly call the TransactionContext.
 665             // However, at the same time, the code may be executing the cached
 666             // TransactionOperations. So to avoid thus timing, we don't publish the
 667             // TransactionContext until after we've executed all cached TransactionOperations.
 668             TransactionContext localTransactionContext;
 669             if(failure != null) {
 670                 LOG.debug("Tx {} Creating NoOpTransaction because of error", identifier, failure);
 671
 672                 localTransactionContext = new NoOpTransactionContext(failure, identifier, operationLimiter);
 673             } else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
 674                 localTransactionContext = createValidTransactionContext(
 675                         CreateTransactionReply.fromSerializable(response));
 676             } else {
 677                 IllegalArgumentException exception = new IllegalArgumentException(String.format(
 678                         "Invalid reply type %s for CreateTransaction", response.getClass()));
 679
 680                 localTransactionContext = new NoOpTransactionContext(exception, identifier, operationLimiter);
 681             }
 682
 683             executeTxOperatonsOnComplete(localTransactionContext);
 684         }
 685
 686         private void executeTxOperatonsOnComplete(TransactionContext localTransactionContext) {
 687             while(true) {
 688                 // Access to txOperationsOnComplete and transactionContext must be protected and atomic
 689                 // (ie synchronized) with respect to #addTxOperationOnComplete to handle timing
 690                 // issues and ensure no TransactionOperation is missed and that they are processed
 691                 // in the order they occurred.
 692
 693                 // We'll make a local copy of the txOperationsOnComplete list to handle re-entrancy
 694                 // in case a TransactionOperation results in another transaction operation being
 695                 // queued (eg a put operation from a client read Future callback that is notified
 696                 // synchronously).
 697                 Collection<TransactionOperation> operationsBatch = null;
 698                 synchronized(txOperationsOnComplete) {
 699                     if(txOperationsOnComplete.isEmpty()) {
 700                         // We're done invoking the TransactionOperations so we can now publish the
 701                         // TransactionContext.
 702                         transactionContext = localTransactionContext;
 703                         break;
 704                     }
 705
 706                     operationsBatch = new ArrayList<>(txOperationsOnComplete);
 707                     txOperationsOnComplete.clear();
 708                 }
 709
 710                 // Invoke TransactionOperations outside the sync block to avoid unnecessary blocking.
 711                 // A slight down-side is that we need to re-acquire the lock below but this should
 712                 // be negligible.
 713                 for(TransactionOperation oper: operationsBatch) {
 714                     oper.invoke(localTransactionContext);
 715                 }
 716             }
 717         }
 718
 719         private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
 720             LOG.debug("Tx {} Received {}", identifier, reply);
 721
 722             return createValidTransactionContext(actorContext.actorSelection(reply.getTransactionPath()),
 723                     reply.getTransactionPath(), reply.getVersion());
 724         }
 725
 726         private TransactionContext createValidTransactionContext(ActorSelection transactionActor,
 727                 String transactionPath, short remoteTransactionVersion) {
 728
 729             if (transactionType == TransactionType.READ_ONLY) {
 730                 // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference
 731                 // to close the remote Tx's when this instance is no longer in use and is garbage
 732                 // collected.
 733
 734                 if(remoteTransactionActorsMB == null) {
 735                     remoteTransactionActors = Lists.newArrayList();
 736                     remoteTransactionActorsMB = new AtomicBoolean();
 737
 738                     TransactionProxyCleanupPhantomReference cleanup =
 739                             new TransactionProxyCleanupPhantomReference(TransactionProxy.this);
 740                     phantomReferenceCache.put(cleanup, cleanup);
 741                 }
 742
 743                 // Add the actor to the remoteTransactionActors list for access by the
 744                 // cleanup PhantonReference.
 745                 remoteTransactionActors.add(transactionActor);
 746
 747                 // Write to the memory barrier volatile to publish the above update to the
 748                 // remoteTransactionActors list for thread visibility.
 749                 remoteTransactionActorsMB.set(true);
 750             }
 751
 752             // TxActor is always created where the leader of the shard is.
 753             // Check if TxActor is created in the same node
 754             boolean isTxActorLocal = actorContext.isPathLocal(transactionPath);
 755
 756             if(remoteTransactionVersion < DataStoreVersions.LITHIUM_VERSION) {
 757                 return new PreLithiumTransactionContextImpl(transactionPath, transactionActor, identifier,
 758                         transactionChainId, actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion,
 759                         operationCompleter);
 760             } else if (transactionType == TransactionType.WRITE_ONLY &&
 761                     actorContext.getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
 762                 return new WriteOnlyTransactionContextImpl(transactionActor, identifier, transactionChainId,
 763                     actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter);
 764             } else {
 765                 return new TransactionContextImpl(transactionActor, identifier, transactionChainId,
 766                         actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter);
 767             }
 768         }
 769     }
 770
 771     private static class NoOpDOMStoreThreePhaseCommitCohort implements DOMStoreThreePhaseCommitCohort {
 772         static NoOpDOMStoreThreePhaseCommitCohort INSTANCE = new NoOpDOMStoreThreePhaseCommitCohort();
 773
 774         private static final ListenableFuture<Void> IMMEDIATE_VOID_SUCCESS =
 775                 com.google.common.util.concurrent.Futures.immediateFuture(null);
 776         private static final ListenableFuture<Boolean> IMMEDIATE_BOOLEAN_SUCCESS =
 777                 com.google.common.util.concurrent.Futures.immediateFuture(Boolean.TRUE);
 778
 779         private NoOpDOMStoreThreePhaseCommitCohort() {
 780         }
 781
 782         @Override
 783         public ListenableFuture<Boolean> canCommit() {
 784             return IMMEDIATE_BOOLEAN_SUCCESS;
 785         }
 786
 787         @Override
 788         public ListenableFuture<Void> preCommit() {
 789             return IMMEDIATE_VOID_SUCCESS;
 790         }
 791
 792         @Override
 793         public ListenableFuture<Void> abort() {
 794             return IMMEDIATE_VOID_SUCCESS;
 795         }
 796
 797         @Override
 798         public ListenableFuture<Void> commit() {
 799             return IMMEDIATE_VOID_SUCCESS;
 800         }
 801     }
 802 }