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.FutureCallback;
  22 import com.google.common.util.concurrent.Futures;
  23 import com.google.common.util.concurrent.SettableFuture;
  24 import java.util.HashMap;
  25 import java.util.List;
  26 import java.util.Map;
  27 import java.util.concurrent.ConcurrentHashMap;
  28 import java.util.concurrent.TimeUnit;
  29 import java.util.concurrent.atomic.AtomicBoolean;
  30 import java.util.concurrent.atomic.AtomicLong;
  31 import javax.annotation.concurrent.GuardedBy;
  32 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  33 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
  34 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  35 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  36 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
  37 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
  38 import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
  39 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
  40 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
  41 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  42 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
  43 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
  44 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  45 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
  46 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
  47 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  48 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  49 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  50 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  51 import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
  52 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  53 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  54 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  55 import org.slf4j.Logger;
  56 import org.slf4j.LoggerFactory;
  57 import scala.concurrent.Future;
  58 import scala.concurrent.Promise;
  59 import scala.concurrent.duration.FiniteDuration;
  60
  61 /**
  62  * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
  63  * <p>
  64  * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
  65  * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
  66  * be created on each of those shards by the TransactionProxy
  67  *</p>
  68  * <p>
  69  * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
  70  * shards will be executed.
  71  * </p>
  72  */
  73 public class TransactionProxy implements DOMStoreReadWriteTransaction {
  74
  75     public static enum TransactionType {
  76         READ_ONLY,
  77         WRITE_ONLY,
  78         READ_WRITE
  79     }
  80
  81     static final Mapper<Throwable, Throwable> SAME_FAILURE_TRANSFORMER =
  82                                                               new Mapper<Throwable, Throwable>() {
  83         @Override
  84         public Throwable apply(Throwable failure) {
  85             return failure;
  86         }
  87     };
  88
  89     private static final AtomicLong counter = new AtomicLong();
  90
  91     private static final Logger LOG = LoggerFactory.getLogger(TransactionProxy.class);
  92
  93     /**
  94      * Time interval in between transaction create retries.
  95      */
  96     private static final FiniteDuration CREATE_TX_TRY_INTERVAL =
  97             FiniteDuration.create(1, TimeUnit.SECONDS);
  98
  99     /**
 100      * Used to enqueue the PhantomReferences for read-only TransactionProxy instances. The
 101      * FinalizableReferenceQueue is safe to use statically in an OSGi environment as it uses some
 102      * trickery to clean up its internal thread when the bundle is unloaded.
 103      */
 104     private static final FinalizableReferenceQueue phantomReferenceQueue =
 105                                                                   new FinalizableReferenceQueue();
 106
 107     /**
 108      * This stores the TransactionProxyCleanupPhantomReference instances statically, This is
 109      * necessary because PhantomReferences need a hard reference so they're not garbage collected.
 110      * Once finalized, the TransactionProxyCleanupPhantomReference removes itself from this map
 111      * and thus becomes eligible for garbage collection.
 112      */
 113     private static final Map<TransactionProxyCleanupPhantomReference,
 114                              TransactionProxyCleanupPhantomReference> phantomReferenceCache =
 115                                                                         new ConcurrentHashMap<>();
 116
 117     /**
 118      * A PhantomReference that closes remote transactions for a TransactionProxy when it's
 119      * garbage collected. This is used for read-only transactions as they're not explicitly closed
 120      * by clients. So the only way to detect that a transaction is no longer in use and it's safe
 121      * to clean up is when it's garbage collected. It's inexact as to when an instance will be GC'ed
 122      * but TransactionProxy instances should generally be short-lived enough to avoid being moved
 123      * to the old generation space and thus should be cleaned up in a timely manner as the GC
 124      * runs on the young generation (eden, swap1...) space much more frequently.
 125      */
 126     private static class TransactionProxyCleanupPhantomReference
 127                                            extends FinalizablePhantomReference<TransactionProxy> {
 128
 129         private final List<ActorSelection> remoteTransactionActors;
 130         private final AtomicBoolean remoteTransactionActorsMB;
 131         private final ActorContext actorContext;
 132         private final TransactionIdentifier identifier;
 133
 134         protected TransactionProxyCleanupPhantomReference(TransactionProxy referent) {
 135             super(referent, phantomReferenceQueue);
 136
 137             // Note we need to cache the relevant fields from the TransactionProxy as we can't
 138             // have a hard reference to the TransactionProxy instance itself.
 139
 140             remoteTransactionActors = referent.remoteTransactionActors;
 141             remoteTransactionActorsMB = referent.remoteTransactionActorsMB;
 142             actorContext = referent.actorContext;
 143             identifier = referent.identifier;
 144         }
 145
 146         @Override
 147         public void finalizeReferent() {
 148             LOG.trace("Cleaning up {} Tx actors for TransactionProxy {}",
 149                     remoteTransactionActors.size(), identifier);
 150
 151             phantomReferenceCache.remove(this);
 152
 153             // Access the memory barrier volatile to ensure all previous updates to the
 154             // remoteTransactionActors list are visible to this thread.
 155
 156             if(remoteTransactionActorsMB.get()) {
 157                 for(ActorSelection actor : remoteTransactionActors) {
 158                     LOG.trace("Sending CloseTransaction to {}", actor);
 159                     actorContext.sendOperationAsync(actor,
 160                             new CloseTransaction().toSerializable());
 161                 }
 162             }
 163         }
 164     }
 165
 166     /**
 167      * Stores the remote Tx actors for each requested data store path to be used by the
 168      * PhantomReference to close the remote Tx's. This is only used for read-only Tx's. The
 169      * remoteTransactionActorsMB volatile serves as a memory barrier to publish updates to the
 170      * remoteTransactionActors list so they will be visible to the thread accessing the
 171      * PhantomReference.
 172      */
 173     private List<ActorSelection> remoteTransactionActors;
 174     private AtomicBoolean remoteTransactionActorsMB;
 175
 176     /**
 177      * Stores the create transaction results per shard.
 178      */
 179     private final Map<String, TransactionFutureCallback> txFutureCallbackMap = new HashMap<>();
 180
 181     private final TransactionType transactionType;
 182     private final ActorContext actorContext;
 183     private final TransactionIdentifier identifier;
 184     private final String transactionChainId;
 185     private final SchemaContext schemaContext;
 186     private boolean inReadyState;
 187
 188     public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
 189         this(actorContext, transactionType, "");
 190     }
 191
 192     public TransactionProxy(ActorContext actorContext, TransactionType transactionType,
 193             String transactionChainId) {
 194         this.actorContext = Preconditions.checkNotNull(actorContext,
 195             "actorContext should not be null");
 196         this.transactionType = Preconditions.checkNotNull(transactionType,
 197             "transactionType should not be null");
 198         this.schemaContext = Preconditions.checkNotNull(actorContext.getSchemaContext(),
 199             "schemaContext should not be null");
 200         this.transactionChainId = transactionChainId;
 201
 202         String memberName = actorContext.getCurrentMemberName();
 203         if(memberName == null){
 204             memberName = "UNKNOWN-MEMBER";
 205         }
 206
 207         this.identifier = TransactionIdentifier.builder().memberName(memberName).counter(
 208             counter.getAndIncrement()).build();
 209
 210         if(transactionType == TransactionType.READ_ONLY) {
 211             // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference
 212             // to close the remote Tx's when this instance is no longer in use and is garbage
 213             // collected.
 214
 215             remoteTransactionActors = Lists.newArrayList();
 216             remoteTransactionActorsMB = new AtomicBoolean();
 217
 218             TransactionProxyCleanupPhantomReference cleanup =
 219                 new TransactionProxyCleanupPhantomReference(this);
 220             phantomReferenceCache.put(cleanup, cleanup);
 221         }
 222
 223         LOG.debug("Created txn {} of type {}", identifier, transactionType);
 224     }
 225
 226     @VisibleForTesting
 227     List<Future<Object>> getRecordedOperationFutures() {
 228         List<Future<Object>> recordedOperationFutures = Lists.newArrayList();
 229         for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
 230             TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 231             if(transactionContext != null) {
 232                 recordedOperationFutures.addAll(transactionContext.getRecordedOperationFutures());
 233             }
 234         }
 235
 236         return recordedOperationFutures;
 237     }
 238
 239     @Override
 240     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(
 241             final YangInstanceIdentifier path) {
 242
 243         Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
 244                 "Read operation on write-only transaction is not allowed");
 245
 246         LOG.debug("Tx {} read {}", identifier, path);
 247
 248         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 249         TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 250
 251         CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> future;
 252         if(transactionContext != null) {
 253             future = transactionContext.readData(path);
 254         } else {
 255             // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
 256             // callback to be executed after the Tx is created.
 257             final SettableFuture<Optional<NormalizedNode<?, ?>>> proxyFuture = SettableFuture.create();
 258             txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
 259                 @Override
 260                 public void invoke(TransactionContext transactionContext) {
 261                     Futures.addCallback(transactionContext.readData(path),
 262                         new FutureCallback<Optional<NormalizedNode<?, ?>>>() {
 263                             @Override
 264                             public void onSuccess(Optional<NormalizedNode<?, ?>> data) {
 265                                 proxyFuture.set(data);
 266                             }
 267
 268                             @Override
 269                             public void onFailure(Throwable t) {
 270                                 proxyFuture.setException(t);
 271                             }
 272                         });
 273                 }
 274             });
 275
 276             future = MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
 277         }
 278
 279         return future;
 280     }
 281
 282     @Override
 283     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
 284
 285         Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
 286                 "Exists operation on write-only transaction is not allowed");
 287
 288         LOG.debug("Tx {} exists {}", identifier, path);
 289
 290         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 291         TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 292
 293         CheckedFuture<Boolean, ReadFailedException> future;
 294         if(transactionContext != null) {
 295             future = transactionContext.dataExists(path);
 296         } else {
 297             // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
 298             // callback to be executed after the Tx is created.
 299             final SettableFuture<Boolean> proxyFuture = SettableFuture.create();
 300             txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
 301                 @Override
 302                 public void invoke(TransactionContext transactionContext) {
 303                     Futures.addCallback(transactionContext.dataExists(path),
 304                         new FutureCallback<Boolean>() {
 305                             @Override
 306                             public void onSuccess(Boolean exists) {
 307                                 proxyFuture.set(exists);
 308                             }
 309
 310                             @Override
 311                             public void onFailure(Throwable t) {
 312                                 proxyFuture.setException(t);
 313                             }
 314                         });
 315                 }
 316             });
 317
 318             future = MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
 319         }
 320
 321         return future;
 322     }
 323
 324     private void checkModificationState() {
 325         Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
 326                 "Modification operation on read-only transaction is not allowed");
 327         Preconditions.checkState(!inReadyState,
 328                 "Transaction is sealed - further modifications are not allowed");
 329     }
 330
 331     @Override
 332     public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 333
 334         checkModificationState();
 335
 336         LOG.debug("Tx {} write {}", identifier, path);
 337
 338         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 339         TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 340         if(transactionContext != null) {
 341             transactionContext.writeData(path, data);
 342         } else {
 343             // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
 344             // callback to be executed after the Tx is created.
 345             txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
 346                 @Override
 347                 public void invoke(TransactionContext transactionContext) {
 348                     transactionContext.writeData(path, data);
 349                 }
 350             });
 351         }
 352     }
 353
 354     @Override
 355     public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 356
 357         checkModificationState();
 358
 359         LOG.debug("Tx {} merge {}", identifier, path);
 360
 361         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 362         TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 363         if(transactionContext != null) {
 364             transactionContext.mergeData(path, data);
 365         } else {
 366             // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
 367             // callback to be executed after the Tx is created.
 368             txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
 369                 @Override
 370                 public void invoke(TransactionContext transactionContext) {
 371                     transactionContext.mergeData(path, data);
 372                 }
 373             });
 374         }
 375     }
 376
 377     @Override
 378     public void delete(final YangInstanceIdentifier path) {
 379
 380         checkModificationState();
 381
 382         LOG.debug("Tx {} delete {}", identifier, path);
 383
 384         TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
 385         TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 386         if(transactionContext != null) {
 387             transactionContext.deleteData(path);
 388         } else {
 389             // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
 390             // callback to be executed after the Tx is created.
 391             txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
 392                 @Override
 393                 public void invoke(TransactionContext transactionContext) {
 394                     transactionContext.deleteData(path);
 395                 }
 396             });
 397         }
 398     }
 399
 400     @Override
 401     public DOMStoreThreePhaseCommitCohort ready() {
 402
 403         checkModificationState();
 404
 405         inReadyState = true;
 406
 407         LOG.debug("Tx {} Readying {} transactions for commit", identifier,
 408                     txFutureCallbackMap.size());
 409
 410         List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
 411
 412         for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
 413
 414             LOG.debug("Tx {} Readying transaction for shard {}", identifier,
 415                         txFutureCallback.getShardName());
 416
 417             TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 418             if(transactionContext != null) {
 419                 cohortFutures.add(transactionContext.readyTransaction());
 420             } else {
 421                 // The shard Tx hasn't been created yet so create a promise to ready the Tx later
 422                 // after it's created.
 423                 final Promise<ActorSelection> cohortPromise = akka.dispatch.Futures.promise();
 424                 txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
 425                     @Override
 426                     public void invoke(TransactionContext transactionContext) {
 427                         cohortPromise.completeWith(transactionContext.readyTransaction());
 428                     }
 429                 });
 430
 431                 cohortFutures.add(cohortPromise.future());
 432             }
 433         }
 434
 435         onTransactionReady(cohortFutures);
 436
 437         return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
 438                 identifier.toString());
 439     }
 440
 441     /**
 442      * Method for derived classes to be notified when the transaction has been readied.
 443      *
 444      * @param cohortFutures the cohort Futures for each shard transaction.
 445      */
 446     protected void onTransactionReady(List<Future<ActorSelection>> cohortFutures) {
 447     }
 448
 449     /**
 450      * Method called to send a CreateTransaction message to a shard.
 451      *
 452      * @param shard the shard actor to send to
 453      * @param serializedCreateMessage the serialized message to send
 454      * @return the response Future
 455      */
 456     protected Future<Object> sendCreateTransaction(ActorSelection shard,
 457             Object serializedCreateMessage) {
 458         return actorContext.executeOperationAsync(shard, serializedCreateMessage);
 459     }
 460
 461     @Override
 462     public Object getIdentifier() {
 463         return this.identifier;
 464     }
 465
 466     @Override
 467     public void close() {
 468         for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
 469             TransactionContext transactionContext = txFutureCallback.getTransactionContext();
 470             if(transactionContext != null) {
 471                 transactionContext.closeTransaction();
 472             } else {
 473                 txFutureCallback.addTxOperationOnComplete(new TransactionOperation() {
 474                     @Override
 475                     public void invoke(TransactionContext transactionContext) {
 476                         transactionContext.closeTransaction();
 477                     }
 478                 });
 479             }
 480         }
 481
 482         txFutureCallbackMap.clear();
 483
 484         if(transactionType == TransactionType.READ_ONLY) {
 485             remoteTransactionActors.clear();
 486             remoteTransactionActorsMB.set(true);
 487         }
 488     }
 489
 490     private String shardNameFromIdentifier(YangInstanceIdentifier path){
 491         return ShardStrategyFactory.getStrategy(path).findShard(path);
 492     }
 493
 494     private TransactionFutureCallback getOrCreateTxFutureCallback(YangInstanceIdentifier path) {
 495         String shardName = shardNameFromIdentifier(path);
 496         TransactionFutureCallback txFutureCallback = txFutureCallbackMap.get(shardName);
 497         if(txFutureCallback == null) {
 498             Future<ActorSelection> findPrimaryFuture = actorContext.findPrimaryShardAsync(shardName);
 499
 500             final TransactionFutureCallback newTxFutureCallback =
 501                     new TransactionFutureCallback(shardName);
 502
 503             txFutureCallback = newTxFutureCallback;
 504             txFutureCallbackMap.put(shardName, txFutureCallback);
 505
 506             findPrimaryFuture.onComplete(new OnComplete<ActorSelection>() {
 507                 @Override
 508                 public void onComplete(Throwable failure, ActorSelection primaryShard) {
 509                     if(failure != null) {
 510                         newTxFutureCallback.onComplete(failure, null);
 511                     } else {
 512                         newTxFutureCallback.setPrimaryShard(primaryShard);
 513                     }
 514                 }
 515             }, actorContext.getActorSystem().dispatcher());
 516         }
 517
 518         return txFutureCallback;
 519     }
 520
 521     public String getTransactionChainId() {
 522         return transactionChainId;
 523     }
 524
 525     /**
 526      * Interface for a transaction operation to be invoked later.
 527      */
 528     private static interface TransactionOperation {
 529         void invoke(TransactionContext transactionContext);
 530     }
 531
 532     /**
 533      * Implements a Future OnComplete callback for a CreateTransaction message. This class handles
 534      * retries, up to a limit, if the shard doesn't have a leader yet. This is done by scheduling a
 535      * retry task after a short delay.
 536      * <p>
 537      * The end result from a completed CreateTransaction message is a TransactionContext that is
 538      * used to perform transaction operations. Transaction operations that occur before the
 539      * CreateTransaction completes are cache and executed once the CreateTransaction completes,
 540      * successfully or not.
 541      */
 542     private class TransactionFutureCallback extends OnComplete<Object> {
 543
 544         /**
 545          * The list of transaction operations to execute once the CreateTransaction completes.
 546          */
 547         @GuardedBy("txOperationsOnComplete")
 548         private final List<TransactionOperation> txOperationsOnComplete = Lists.newArrayList();
 549
 550         /**
 551          * The TransactionContext resulting from the CreateTransaction reply.
 552          */
 553         private volatile TransactionContext transactionContext;
 554
 555         /**
 556          * The target primary shard.
 557          */
 558         private volatile ActorSelection primaryShard;
 559
 560         private volatile int createTxTries = (int) (actorContext.getDatastoreContext().
 561                 getShardLeaderElectionTimeout().duration().toMillis() /
 562                 CREATE_TX_TRY_INTERVAL.toMillis());
 563
 564         private final String shardName;
 565
 566         TransactionFutureCallback(String shardName) {
 567             this.shardName = shardName;
 568         }
 569
 570         String getShardName() {
 571             return shardName;
 572         }
 573
 574         TransactionContext getTransactionContext() {
 575             return transactionContext;
 576         }
 577
 578
 579         /**
 580          * Sets the target primary shard and initiates a CreateTransaction try.
 581          */
 582         void setPrimaryShard(ActorSelection primaryShard) {
 583             LOG.debug("Tx {} Primary shard found - trying create transaction", identifier);
 584
 585             this.primaryShard = primaryShard;
 586             tryCreateTransaction();
 587         }
 588
 589         /**
 590          * Adds a TransactionOperation to be executed after the CreateTransaction completes.
 591          */
 592         void addTxOperationOnComplete(TransactionOperation operation) {
 593             synchronized(txOperationsOnComplete) {
 594                 if(transactionContext == null) {
 595                     LOG.debug("Tx {} Adding operation on complete {}", identifier);
 596
 597                     txOperationsOnComplete.add(operation);
 598                 } else {
 599                     operation.invoke(transactionContext);
 600                 }
 601             }
 602         }
 603
 604         /**
 605          * Performs a CreateTransaction try async.
 606          */
 607         private void tryCreateTransaction() {
 608             Future<Object> createTxFuture = sendCreateTransaction(primaryShard,
 609                     new CreateTransaction(identifier.toString(),
 610                             TransactionProxy.this.transactionType.ordinal(),
 611                             getTransactionChainId()).toSerializable());
 612
 613             createTxFuture.onComplete(this, actorContext.getActorSystem().dispatcher());
 614         }
 615
 616         @Override
 617         public void onComplete(Throwable failure, Object response) {
 618             if(failure instanceof NoShardLeaderException) {
 619                 // There's no leader for the shard yet - schedule and try again, unless we're out
 620                 // of retries. Note: createTxTries is volatile as it may be written by different
 621                 // threads however not concurrently, therefore decrementing it non-atomically here
 622                 // is ok.
 623                 if(--createTxTries > 0) {
 624                     LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
 625                             identifier, shardName);
 626
 627                     actorContext.getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
 628                             new Runnable() {
 629                                 @Override
 630                                 public void run() {
 631                                     tryCreateTransaction();
 632                                 }
 633                             }, actorContext.getActorSystem().dispatcher());
 634                     return;
 635                 }
 636             }
 637
 638             // Create the TransactionContext from the response or failure and execute delayed
 639             // TransactionOperations. This entire section is done atomically (ie synchronized) with
 640             // respect to #addTxOperationOnComplete to handle timing issues and ensure no
 641             // TransactionOperation is missed and that they are processed in the order they occurred.
 642             synchronized(txOperationsOnComplete) {
 643                 if(failure != null) {
 644                     LOG.debug("Tx {} Creating NoOpTransaction because of error: {}", identifier,
 645                             failure.getMessage());
 646
 647                     transactionContext = new NoOpTransactionContext(failure, identifier);
 648                 } else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
 649                     createValidTransactionContext(CreateTransactionReply.fromSerializable(response));
 650                 } else {
 651                     IllegalArgumentException exception = new IllegalArgumentException(String.format(
 652                         "Invalid reply type %s for CreateTransaction", response.getClass()));
 653
 654                     transactionContext = new NoOpTransactionContext(exception, identifier);
 655                 }
 656
 657                 for(TransactionOperation oper: txOperationsOnComplete) {
 658                     oper.invoke(transactionContext);
 659                 }
 660
 661                 txOperationsOnComplete.clear();
 662             }
 663         }
 664
 665         private void createValidTransactionContext(CreateTransactionReply reply) {
 666             String transactionPath = reply.getTransactionPath();
 667
 668             LOG.debug("Tx {} Received transaction actor path {}", identifier, transactionPath);
 669
 670             ActorSelection transactionActor = actorContext.actorSelection(transactionPath);
 671
 672             if (transactionType == TransactionType.READ_ONLY) {
 673                 // Add the actor to the remoteTransactionActors list for access by the
 674                 // cleanup PhantonReference.
 675                 remoteTransactionActors.add(transactionActor);
 676
 677                 // Write to the memory barrier volatile to publish the above update to the
 678                 // remoteTransactionActors list for thread visibility.
 679                 remoteTransactionActorsMB.set(true);
 680             }
 681
 682             // TxActor is always created where the leader of the shard is.
 683             // Check if TxActor is created in the same node
 684             boolean isTxActorLocal = actorContext.isLocalPath(transactionPath);
 685
 686             transactionContext = new TransactionContextImpl(transactionPath, transactionActor, identifier,
 687                 actorContext, schemaContext, isTxActorLocal, reply.getVersion());
 688         }
 689     }
 690
 691     private interface TransactionContext {
 692         void closeTransaction();
 693
 694         Future<ActorSelection> readyTransaction();
 695
 696         void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 697
 698         void deleteData(YangInstanceIdentifier path);
 699
 700         void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 701
 702         CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readData(
 703                 final YangInstanceIdentifier path);
 704
 705         CheckedFuture<Boolean, ReadFailedException> dataExists(YangInstanceIdentifier path);
 706
 707         List<Future<Object>> getRecordedOperationFutures();
 708     }
 709
 710     private static abstract class AbstractTransactionContext implements TransactionContext {
 711
 712         protected final TransactionIdentifier identifier;
 713         protected final List<Future<Object>> recordedOperationFutures = Lists.newArrayList();
 714
 715         AbstractTransactionContext(TransactionIdentifier identifier) {
 716             this.identifier = identifier;
 717         }
 718
 719         @Override
 720         public List<Future<Object>> getRecordedOperationFutures() {
 721             return recordedOperationFutures;
 722         }
 723     }
 724
 725     private static class TransactionContextImpl extends AbstractTransactionContext {
 726         private final Logger LOG = LoggerFactory.getLogger(TransactionContextImpl.class);
 727
 728         private final ActorContext actorContext;
 729         private final SchemaContext schemaContext;
 730         private final String transactionPath;
 731         private final ActorSelection actor;
 732         private final boolean isTxActorLocal;
 733         private final int remoteTransactionVersion;
 734
 735         private TransactionContextImpl(String transactionPath, ActorSelection actor, TransactionIdentifier identifier,
 736                 ActorContext actorContext, SchemaContext schemaContext,
 737                 boolean isTxActorLocal, int remoteTransactionVersion) {
 738             super(identifier);
 739             this.transactionPath = transactionPath;
 740             this.actor = actor;
 741             this.actorContext = actorContext;
 742             this.schemaContext = schemaContext;
 743             this.isTxActorLocal = isTxActorLocal;
 744             this.remoteTransactionVersion = remoteTransactionVersion;
 745         }
 746
 747         private ActorSelection getActor() {
 748             return actor;
 749         }
 750
 751         @Override
 752         public void closeTransaction() {
 753             LOG.debug("Tx {} closeTransaction called", identifier);
 754
 755             actorContext.sendOperationAsync(getActor(), new CloseTransaction().toSerializable());
 756         }
 757
 758         @Override
 759         public Future<ActorSelection> readyTransaction() {
 760             LOG.debug("Tx {} readyTransaction called with {} previous recorded operations pending",
 761                     identifier, recordedOperationFutures.size());
 762
 763             // Send the ReadyTransaction message to the Tx actor.
 764
 765             ReadyTransaction readyTransaction = new ReadyTransaction();
 766             final Future<Object> replyFuture = actorContext.executeOperationAsync(getActor(),
 767                 isTxActorLocal ? readyTransaction : readyTransaction.toSerializable());
 768
 769             // Combine all the previously recorded put/merge/delete operation reply Futures and the
 770             // ReadyTransactionReply Future into one Future. If any one fails then the combined
 771             // Future will fail. We need all prior operations and the ready operation to succeed
 772             // in order to attempt commit.
 773
 774             List<Future<Object>> futureList =
 775                     Lists.newArrayListWithCapacity(recordedOperationFutures.size() + 1);
 776             futureList.addAll(recordedOperationFutures);
 777             futureList.add(replyFuture);
 778
 779             Future<Iterable<Object>> combinedFutures = akka.dispatch.Futures.sequence(futureList,
 780                     actorContext.getActorSystem().dispatcher());
 781
 782             // Transform the combined Future into a Future that returns the cohort actor path from
 783             // the ReadyTransactionReply. That's the end result of the ready operation.
 784
 785             return combinedFutures.transform(new Mapper<Iterable<Object>, ActorSelection>() {
 786                 @Override
 787                 public ActorSelection checkedApply(Iterable<Object> notUsed) {
 788                     LOG.debug("Tx {} readyTransaction: pending recorded operations succeeded",
 789                             identifier);
 790
 791                     // At this point all the Futures succeeded and we need to extract the cohort
 792                     // actor path from the ReadyTransactionReply. For the recorded operations, they
 793                     // don't return any data so we're only interested that they completed
 794                     // successfully. We could be paranoid and verify the correct reply types but
 795                     // that really should never happen so it's not worth the overhead of
 796                     // de-serializing each reply.
 797
 798                     // Note the Future get call here won't block as it's complete.
 799                     Object serializedReadyReply = replyFuture.value().get().get();
 800                     if (serializedReadyReply instanceof ReadyTransactionReply) {
 801                         return actorContext.actorSelection(((ReadyTransactionReply)serializedReadyReply).getCohortPath());
 802
 803                     } else if(serializedReadyReply.getClass().equals(ReadyTransactionReply.SERIALIZABLE_CLASS)) {
 804                         ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(serializedReadyReply);
 805                         String cohortPath = reply.getCohortPath();
 806
 807                         // In Helium we used to return the local path of the actor which represented
 808                         // a remote ThreePhaseCommitCohort. The local path would then be converted to
 809                         // a remote path using this resolvePath method. To maintain compatibility with
 810                         // a Helium node we need to continue to do this conversion.
 811                         // At some point in the future when upgrades from Helium are not supported
 812                         // we could remove this code to resolvePath and just use the cohortPath as the
 813                         // resolved cohortPath
 814                         if(TransactionContextImpl.this.remoteTransactionVersion < CreateTransaction.HELIUM_1_VERSION) {
 815                             cohortPath = actorContext.resolvePath(transactionPath, cohortPath);
 816                         }
 817
 818                         return actorContext.actorSelection(cohortPath);
 819
 820                     } else {
 821                         // Throwing an exception here will fail the Future.
 822                         throw new IllegalArgumentException(String.format("Invalid reply type {}",
 823                                 serializedReadyReply.getClass()));
 824                     }
 825                 }
 826             }, SAME_FAILURE_TRANSFORMER, actorContext.getActorSystem().dispatcher());
 827         }
 828
 829         @Override
 830         public void deleteData(YangInstanceIdentifier path) {
 831             LOG.debug("Tx {} deleteData called path = {}", identifier, path);
 832
 833             DeleteData deleteData = new DeleteData(path);
 834             recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
 835                 isTxActorLocal ? deleteData : deleteData.toSerializable()));
 836         }
 837
 838         @Override
 839         public void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
 840             LOG.debug("Tx {} mergeData called path = {}", identifier, path);
 841
 842             MergeData mergeData = new MergeData(path, data, schemaContext);
 843             recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
 844                 isTxActorLocal ? mergeData : mergeData.toSerializable()));
 845         }
 846
 847         @Override
 848         public void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
 849             LOG.debug("Tx {} writeData called path = {}", identifier, path);
 850
 851             WriteData writeData = new WriteData(path, data, schemaContext);
 852             recordedOperationFutures.add(actorContext.executeOperationAsync(getActor(),
 853                 isTxActorLocal ? writeData : writeData.toSerializable()));
 854         }
 855
 856         @Override
 857         public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readData(
 858                 final YangInstanceIdentifier path) {
 859
 860             LOG.debug("Tx {} readData called path = {}", identifier, path);
 861
 862             final SettableFuture<Optional<NormalizedNode<?, ?>>> returnFuture = SettableFuture.create();
 863
 864             // If there were any previous recorded put/merge/delete operation reply Futures then we
 865             // must wait for them to successfully complete. This is necessary to honor the read
 866             // uncommitted semantics of the public API contract. If any one fails then fail the read.
 867
 868             if(recordedOperationFutures.isEmpty()) {
 869                 finishReadData(path, returnFuture);
 870             } else {
 871                 LOG.debug("Tx {} readData: verifying {} previous recorded operations",
 872                         identifier, recordedOperationFutures.size());
 873
 874                 // Note: we make a copy of recordedOperationFutures to be on the safe side in case
 875                 // Futures#sequence accesses the passed List on a different thread, as
 876                 // recordedOperationFutures is not synchronized.
 877
 878                 Future<Iterable<Object>> combinedFutures = akka.dispatch.Futures.sequence(
 879                         Lists.newArrayList(recordedOperationFutures),
 880                         actorContext.getActorSystem().dispatcher());
 881
 882                 OnComplete<Iterable<Object>> onComplete = new OnComplete<Iterable<Object>>() {
 883                     @Override
 884                     public void onComplete(Throwable failure, Iterable<Object> notUsed)
 885                             throws Throwable {
 886                         if(failure != null) {
 887                             LOG.debug("Tx {} readData: a recorded operation failed: {}",
 888                                     identifier, failure);
 889                             returnFuture.setException(new ReadFailedException(
 890                                     "The read could not be performed because a previous put, merge,"
 891                                     + "or delete operation failed", failure));
 892                         } else {
 893                             finishReadData(path, returnFuture);
 894                         }
 895                     }
 896                 };
 897
 898                 combinedFutures.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
 899             }
 900
 901             return MappingCheckedFuture.create(returnFuture, ReadFailedException.MAPPER);
 902         }
 903
 904         private void finishReadData(final YangInstanceIdentifier path,
 905                 final SettableFuture<Optional<NormalizedNode<?, ?>>> returnFuture) {
 906
 907             LOG.debug("Tx {} finishReadData called path = {}", identifier, path);
 908
 909             OnComplete<Object> onComplete = new OnComplete<Object>() {
 910                 @Override
 911                 public void onComplete(Throwable failure, Object readResponse) throws Throwable {
 912                     if(failure != null) {
 913                         LOG.debug("Tx {} read operation failed: {}", identifier, failure);
 914                         returnFuture.setException(new ReadFailedException(
 915                                 "Error reading data for path " + path, failure));
 916
 917                     } else {
 918                         LOG.debug("Tx {} read operation succeeded", identifier, failure);
 919
 920                         if (readResponse instanceof ReadDataReply) {
 921                             ReadDataReply reply = (ReadDataReply) readResponse;
 922                             returnFuture.set(Optional.<NormalizedNode<?, ?>>fromNullable(reply.getNormalizedNode()));
 923
 924                         } else if (readResponse.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
 925                             ReadDataReply reply = ReadDataReply.fromSerializable(schemaContext, path, readResponse);
 926                             returnFuture.set(Optional.<NormalizedNode<?, ?>>fromNullable(reply.getNormalizedNode()));
 927
 928                         } else {
 929                             returnFuture.setException(new ReadFailedException(
 930                                 "Invalid response reading data for path " + path));
 931                         }
 932                     }
 933                 }
 934             };
 935
 936             ReadData readData = new ReadData(path);
 937             Future<Object> readFuture = actorContext.executeOperationAsync(getActor(),
 938                 isTxActorLocal ? readData : readData.toSerializable());
 939
 940             readFuture.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
 941         }
 942
 943         @Override
 944         public CheckedFuture<Boolean, ReadFailedException> dataExists(
 945                 final YangInstanceIdentifier path) {
 946
 947             LOG.debug("Tx {} dataExists called path = {}", identifier, path);
 948
 949             final SettableFuture<Boolean> returnFuture = SettableFuture.create();
 950
 951             // If there were any previous recorded put/merge/delete operation reply Futures then we
 952             // must wait for them to successfully complete. This is necessary to honor the read
 953             // uncommitted semantics of the public API contract. If any one fails then fail this
 954             // request.
 955
 956             if(recordedOperationFutures.isEmpty()) {
 957                 finishDataExists(path, returnFuture);
 958             } else {
 959                 LOG.debug("Tx {} dataExists: verifying {} previous recorded operations",
 960                         identifier, recordedOperationFutures.size());
 961
 962                 // Note: we make a copy of recordedOperationFutures to be on the safe side in case
 963                 // Futures#sequence accesses the passed List on a different thread, as
 964                 // recordedOperationFutures is not synchronized.
 965
 966                 Future<Iterable<Object>> combinedFutures = akka.dispatch.Futures.sequence(
 967                         Lists.newArrayList(recordedOperationFutures),
 968                         actorContext.getActorSystem().dispatcher());
 969                 OnComplete<Iterable<Object>> onComplete = new OnComplete<Iterable<Object>>() {
 970                     @Override
 971                     public void onComplete(Throwable failure, Iterable<Object> notUsed)
 972                             throws Throwable {
 973                         if(failure != null) {
 974                             LOG.debug("Tx {} dataExists: a recorded operation failed: {}",
 975                                     identifier, failure);
 976                             returnFuture.setException(new ReadFailedException(
 977                                     "The data exists could not be performed because a previous "
 978                                     + "put, merge, or delete operation failed", failure));
 979                         } else {
 980                             finishDataExists(path, returnFuture);
 981                         }
 982                     }
 983                 };
 984
 985                 combinedFutures.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
 986             }
 987
 988             return MappingCheckedFuture.create(returnFuture, ReadFailedException.MAPPER);
 989         }
 990
 991         private void finishDataExists(final YangInstanceIdentifier path,
 992                 final SettableFuture<Boolean> returnFuture) {
 993
 994             LOG.debug("Tx {} finishDataExists called path = {}", identifier, path);
 995
 996             OnComplete<Object> onComplete = new OnComplete<Object>() {
 997                 @Override
 998                 public void onComplete(Throwable failure, Object response) throws Throwable {
 999                     if(failure != null) {
1000                         LOG.debug("Tx {} dataExists operation failed: {}", identifier, failure);
1001                         returnFuture.setException(new ReadFailedException(
1002                                 "Error checking data exists for path " + path, failure));
1003                     } else {
1004                         LOG.debug("Tx {} dataExists operation succeeded", identifier, failure);
1005
1006                         if (response instanceof DataExistsReply) {
1007                             returnFuture.set(Boolean.valueOf(((DataExistsReply) response).exists()));
1008
1009                         } else if (response.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) {
1010                             returnFuture.set(Boolean.valueOf(DataExistsReply.fromSerializable(response).exists()));
1011
1012                         } else {
1013                             returnFuture.setException(new ReadFailedException(
1014                                     "Invalid response checking exists for path " + path));
1015                         }
1016                     }
1017                 }
1018             };
1019
1020             DataExists dataExists = new DataExists(path);
1021             Future<Object> future = actorContext.executeOperationAsync(getActor(),
1022                 isTxActorLocal ? dataExists : dataExists.toSerializable());
1023
1024             future.onComplete(onComplete, actorContext.getActorSystem().dispatcher());
1025         }
1026     }
1027
1028     private static class NoOpTransactionContext extends AbstractTransactionContext {
1029
1030         private final Logger LOG = LoggerFactory.getLogger(NoOpTransactionContext.class);
1031
1032         private final Throwable failure;
1033
1034         public NoOpTransactionContext(Throwable failure, TransactionIdentifier identifier){
1035             super(identifier);
1036             this.failure = failure;
1037         }
1038
1039         @Override
1040         public void closeTransaction() {
1041             LOG.debug("NoOpTransactionContext {} closeTransaction called", identifier);
1042         }
1043
1044         @Override
1045         public Future<ActorSelection> readyTransaction() {
1046             LOG.debug("Tx {} readyTransaction called", identifier);
1047             return akka.dispatch.Futures.failed(failure);
1048         }
1049
1050         @Override
1051         public void deleteData(YangInstanceIdentifier path) {
1052             LOG.debug("Tx {} deleteData called path = {}", identifier, path);
1053         }
1054
1055         @Override
1056         public void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
1057             LOG.debug("Tx {} mergeData called path = {}", identifier, path);
1058         }
1059
1060         @Override
1061         public void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
1062             LOG.debug("Tx {} writeData called path = {}", identifier, path);
1063         }
1064
1065         @Override
1066         public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readData(
1067                 YangInstanceIdentifier path) {
1068             LOG.debug("Tx {} readData called path = {}", identifier, path);
1069             return Futures.immediateFailedCheckedFuture(new ReadFailedException(
1070                     "Error reading data for path " + path, failure));
1071         }
1072
1073         @Override
1074         public CheckedFuture<Boolean, ReadFailedException> dataExists(
1075                 YangInstanceIdentifier path) {
1076             LOG.debug("Tx {} dataExists called path = {}", identifier, path);
1077             return Futures.immediateFailedCheckedFuture(new ReadFailedException(
1078                     "Error checking exists for path " + path, failure));
1079         }
1080     }
1081 }