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

   1 /*
   2  * Copyright (c) 2014 Brocade Communications 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 package org.opendaylight.controller.cluster.datastore;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.Status;
  12 import akka.serialization.Serialization;
  13 import com.google.common.annotations.VisibleForTesting;
  14 import com.google.common.base.Preconditions;
  15 import com.google.common.cache.Cache;
  16 import com.google.common.cache.CacheBuilder;
  17 import com.google.common.cache.RemovalCause;
  18 import com.google.common.cache.RemovalListener;
  19 import com.google.common.cache.RemovalNotification;
  20 import java.util.LinkedList;
  21 import java.util.Queue;
  22 import java.util.concurrent.ExecutionException;
  23 import java.util.concurrent.TimeUnit;
  24 import org.opendaylight.controller.cluster.datastore.compat.BackwardsCompatibleThreePhaseCommitCohort;
  25 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  26 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
  27 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
  28 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
  29 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
  30 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  31 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  32 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
  33 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  34 import org.slf4j.Logger;
  35
  36 /**
  37  * Coordinates commits for a shard ensuring only one concurrent 3-phase commit.
  38  *
  39  * @author Thomas Pantelis
  40  */
  41 public class ShardCommitCoordinator {
  42
  43     // Interface hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
  44     public interface CohortDecorator {
  45         ShardDataTreeCohort decorate(String transactionID, ShardDataTreeCohort actual);
  46     }
  47
  48     private final Cache<String, CohortEntry> cohortCache;
  49
  50     private CohortEntry currentCohortEntry;
  51
  52     private final ShardDataTree dataTree;
  53
  54     private final Queue<CohortEntry> queuedCohortEntries;
  55
  56     private int queueCapacity;
  57
  58     private final Logger log;
  59
  60     private final String name;
  61
  62     private final RemovalListener<String, CohortEntry> cacheRemovalListener =
  63             new RemovalListener<String, CohortEntry>() {
  64                 @Override
  65                 public void onRemoval(RemovalNotification<String, CohortEntry> notification) {
  66                     if(notification.getCause() == RemovalCause.EXPIRED) {
  67                         log.warn("{}: Transaction {} was timed out of the cache", name, notification.getKey());
  68                     }
  69                 }
  70             };
  71
  72     // This is a hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
  73     private CohortDecorator cohortDecorator;
  74
  75     private ReadyTransactionReply readyTransactionReply;
  76
  77     public ShardCommitCoordinator(ShardDataTree dataTree,
  78             long cacheExpiryTimeoutInSec, int queueCapacity, ActorRef shardActor, Logger log, String name) {
  79
  80         this.queueCapacity = queueCapacity;
  81         this.log = log;
  82         this.name = name;
  83         this.dataTree = Preconditions.checkNotNull(dataTree);
  84
  85         cohortCache = CacheBuilder.newBuilder().expireAfterAccess(cacheExpiryTimeoutInSec, TimeUnit.SECONDS).
  86                 removalListener(cacheRemovalListener).build();
  87
  88         // We use a LinkedList here to avoid synchronization overhead with concurrent queue impls
  89         // since this should only be accessed on the shard's dispatcher.
  90         queuedCohortEntries = new LinkedList<>();
  91     }
  92
  93     public void setQueueCapacity(int queueCapacity) {
  94         this.queueCapacity = queueCapacity;
  95     }
  96
  97     private ReadyTransactionReply readyTransactionReply(Shard shard) {
  98         if(readyTransactionReply == null) {
  99             readyTransactionReply = new ReadyTransactionReply(Serialization.serializedActorPath(shard.self()));
 100         }
 101
 102         return readyTransactionReply;
 103     }
 104
 105     /**
 106      * This method is called to ready a transaction that was prepared by ShardTransaction actor. It caches
 107      * the prepared cohort entry for the given transactions ID in preparation for the subsequent 3-phase commit.
 108      */
 109     public void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard) {
 110         log.debug("{}: Readying transaction {}, client version {}", name,
 111                 ready.getTransactionID(), ready.getTxnClientVersion());
 112
 113         CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), ready.getCohort(),
 114                 (MutableCompositeModification) ready.getModification());
 115         cohortCache.put(ready.getTransactionID(), cohortEntry);
 116
 117         if(ready.getTxnClientVersion() < DataStoreVersions.LITHIUM_VERSION) {
 118             // Return our actor path as we'll handle the three phase commit except if the Tx client
 119             // version < Helium-1 version which means the Tx was initiated by a base Helium version node.
 120             // In that case, the subsequent 3-phase commit messages won't contain the transactionId so to
 121             // maintain backwards compatibility, we create a separate cohort actor to provide the compatible behavior.
 122             ActorRef replyActorPath = shard.self();
 123             if(ready.getTxnClientVersion() < DataStoreVersions.HELIUM_1_VERSION) {
 124                 log.debug("{}: Creating BackwardsCompatibleThreePhaseCommitCohort", name);
 125                 replyActorPath = shard.getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props(
 126                         ready.getTransactionID()));
 127             }
 128
 129             ReadyTransactionReply readyTransactionReply =
 130                     new ReadyTransactionReply(Serialization.serializedActorPath(replyActorPath),
 131                             ready.getTxnClientVersion());
 132             sender.tell(ready.isReturnSerialized() ? readyTransactionReply.toSerializable() :
 133                 readyTransactionReply, shard.self());
 134         } else {
 135             if(ready.isDoImmediateCommit()) {
 136                 cohortEntry.setDoImmediateCommit(true);
 137                 cohortEntry.setReplySender(sender);
 138                 cohortEntry.setShard(shard);
 139                 handleCanCommit(cohortEntry);
 140             } else {
 141                 // The caller does not want immediate commit - the 3-phase commit will be coordinated by the
 142                 // front-end so send back a ReadyTransactionReply with our actor path.
 143                 sender.tell(readyTransactionReply(shard), shard.self());
 144             }
 145         }
 146     }
 147
 148     /**
 149      * This method handles a BatchedModifications message for a transaction being prepared directly on the
 150      * Shard actor instead of via a ShardTransaction actor. If there's no currently cached
 151      * DOMStoreWriteTransaction, one is created. The batched modifications are applied to the write Tx. If
 152      * the BatchedModifications is ready to commit then a DOMStoreThreePhaseCommitCohort is created.
 153      *
 154      * @param batched the BatchedModifications
 155      * @param shardActor the transaction's shard actor
 156      *
 157      * @throws ExecutionException if an error occurs loading the cache
 158      */
 159     boolean handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard)
 160             throws ExecutionException {
 161         CohortEntry cohortEntry = cohortCache.getIfPresent(batched.getTransactionID());
 162         if(cohortEntry == null) {
 163             cohortEntry = new CohortEntry(batched.getTransactionID(),
 164                     dataTree.newReadWriteTransaction(batched.getTransactionID(),
 165                         batched.getTransactionChainID()));
 166             cohortCache.put(batched.getTransactionID(), cohortEntry);
 167         }
 168
 169         if(log.isDebugEnabled()) {
 170             log.debug("{}: Applying {} batched modifications for Tx {}", name,
 171                     batched.getModifications().size(), batched.getTransactionID());
 172         }
 173
 174         cohortEntry.applyModifications(batched.getModifications());
 175
 176         if(batched.isReady()) {
 177             if(log.isDebugEnabled()) {
 178                 log.debug("{}: Readying Tx {}, client version {}", name,
 179                         batched.getTransactionID(), batched.getVersion());
 180             }
 181
 182             cohortEntry.ready(cohortDecorator, batched.isDoCommitOnReady());
 183
 184             if(batched.isDoCommitOnReady()) {
 185                 cohortEntry.setReplySender(sender);
 186                 cohortEntry.setShard(shard);
 187                 handleCanCommit(cohortEntry);
 188             } else {
 189                 sender.tell(readyTransactionReply(shard), shard.self());
 190             }
 191         } else {
 192             sender.tell(new BatchedModificationsReply(batched.getModifications().size()), shard.self());
 193         }
 194
 195         return batched.isReady();
 196     }
 197
 198     /**
 199      * This method handles {@link ReadyLocalTransaction} message. All transaction modifications have
 200      * been prepared beforehand by the sender and we just need to drive them through into the dataTree.
 201      *
 202      * @param message
 203      * @param sender
 204      * @param shard
 205      */
 206     void handleReadyLocalTransaction(ReadyLocalTransaction message, ActorRef sender, Shard shard) {
 207         final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification());
 208         final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort);
 209         cohortCache.put(message.getTransactionID(), cohortEntry);
 210         cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
 211         log.debug("{}: Applying local modifications for Tx {}", name, message.getTransactionID());
 212
 213         if (message.isDoCommitOnReady()) {
 214             cohortEntry.setReplySender(sender);
 215             cohortEntry.setShard(shard);
 216             handleCanCommit(cohortEntry);
 217         } else {
 218             sender.tell(readyTransactionReply(shard), shard.self());
 219         }
 220     }
 221
 222     private void handleCanCommit(CohortEntry cohortEntry) {
 223         String transactionID = cohortEntry.getTransactionID();
 224
 225         if(log.isDebugEnabled()) {
 226             log.debug("{}: Processing canCommit for transaction {} for shard {}",
 227                     name, transactionID, cohortEntry.getShard().self().path());
 228         }
 229
 230         if(currentCohortEntry != null) {
 231             // There's already a Tx commit in progress - attempt to queue this entry to be
 232             // committed after the current Tx completes.
 233             log.debug("{}: Transaction {} is already in progress - queueing transaction {}",
 234                     name, currentCohortEntry.getTransactionID(), transactionID);
 235
 236             if(queuedCohortEntries.size() < queueCapacity) {
 237                 queuedCohortEntries.offer(cohortEntry);
 238             } else {
 239                 removeCohortEntry(transactionID);
 240
 241                 RuntimeException ex = new RuntimeException(
 242                         String.format("%s: Could not enqueue transaction %s - the maximum commit queue"+
 243                                       " capacity %d has been reached.",
 244                                       name, transactionID, queueCapacity));
 245                 log.error(ex.getMessage());
 246                 cohortEntry.getReplySender().tell(new Status.Failure(ex), cohortEntry.getShard().self());
 247             }
 248         } else {
 249             // No Tx commit currently in progress - make this the current entry and proceed with
 250             // canCommit.
 251             cohortEntry.updateLastAccessTime();
 252             currentCohortEntry = cohortEntry;
 253
 254             doCanCommit(cohortEntry);
 255         }
 256     }
 257
 258     /**
 259      * This method handles the canCommit phase for a transaction.
 260      *
 261      * @param canCommit the CanCommitTransaction message
 262      * @param sender the actor that sent the message
 263      * @param shard the transaction's shard actor
 264      */
 265     public void handleCanCommit(String transactionID, final ActorRef sender, final Shard shard) {
 266         // Lookup the cohort entry that was cached previously (or should have been) by
 267         // transactionReady (via the ForwardedReadyTransaction message).
 268         final CohortEntry cohortEntry = cohortCache.getIfPresent(transactionID);
 269         if(cohortEntry == null) {
 270             // Either canCommit was invoked before ready(shouldn't happen)  or a long time passed
 271             // between canCommit and ready and the entry was expired from the cache.
 272             IllegalStateException ex = new IllegalStateException(
 273                     String.format("%s: No cohort entry found for transaction %s", name, transactionID));
 274             log.error(ex.getMessage());
 275             sender.tell(new Status.Failure(ex), shard.self());
 276             return;
 277         }
 278
 279         cohortEntry.setReplySender(sender);
 280         cohortEntry.setShard(shard);
 281
 282         handleCanCommit(cohortEntry);
 283     }
 284
 285     private void doCanCommit(final CohortEntry cohortEntry) {
 286
 287         boolean canCommit = false;
 288         try {
 289             // We block on the future here so we don't have to worry about possibly accessing our
 290             // state on a different thread outside of our dispatcher. Also, the data store
 291             // currently uses a same thread executor anyway.
 292             canCommit = cohortEntry.getCohort().canCommit().get();
 293
 294             if(cohortEntry.isDoImmediateCommit()) {
 295                 if(canCommit) {
 296                     doCommit(cohortEntry);
 297                 } else {
 298                     cohortEntry.getReplySender().tell(new Status.Failure(new TransactionCommitFailedException(
 299                                 "Can Commit failed, no detailed cause available.")), cohortEntry.getShard().self());
 300                 }
 301             } else {
 302                 cohortEntry.getReplySender().tell(
 303                         canCommit ? CanCommitTransactionReply.YES.toSerializable() :
 304                             CanCommitTransactionReply.NO.toSerializable(), cohortEntry.getShard().self());
 305             }
 306         } catch (Exception e) {
 307             log.debug("{}: An exception occurred during canCommit: {}", name, e);
 308
 309             Throwable failure = e;
 310             if(e instanceof ExecutionException) {
 311                 failure = e.getCause();
 312             }
 313
 314             cohortEntry.getReplySender().tell(new Status.Failure(failure), cohortEntry.getShard().self());
 315         } finally {
 316             if(!canCommit) {
 317                 // Remove the entry from the cache now.
 318                 currentTransactionComplete(cohortEntry.getTransactionID(), true);
 319             }
 320         }
 321     }
 322
 323     private boolean doCommit(CohortEntry cohortEntry) {
 324         log.debug("{}: Committing transaction {}", name, cohortEntry.getTransactionID());
 325
 326         boolean success = false;
 327
 328         // We perform the preCommit phase here atomically with the commit phase. This is an
 329         // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
 330         // coordination of preCommit across shards in case of failure but preCommit should not
 331         // normally fail since we ensure only one concurrent 3-phase commit.
 332
 333         try {
 334             // We block on the future here so we don't have to worry about possibly accessing our
 335             // state on a different thread outside of our dispatcher. Also, the data store
 336             // currently uses a same thread executor anyway.
 337             cohortEntry.getCohort().preCommit().get();
 338
 339             cohortEntry.getShard().continueCommit(cohortEntry);
 340
 341             cohortEntry.updateLastAccessTime();
 342
 343             success = true;
 344         } catch (Exception e) {
 345             log.error("{} An exception occurred while preCommitting transaction {}",
 346                     name, cohortEntry.getTransactionID(), e);
 347             cohortEntry.getReplySender().tell(new akka.actor.Status.Failure(e), cohortEntry.getShard().self());
 348
 349             currentTransactionComplete(cohortEntry.getTransactionID(), true);
 350         }
 351
 352         return success;
 353     }
 354
 355     boolean handleCommit(final String transactionID, final ActorRef sender, final Shard shard) {
 356         // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
 357         // this transaction.
 358         final CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
 359         if(cohortEntry == null) {
 360             // We're not the current Tx - the Tx was likely expired b/c it took too long in
 361             // between the canCommit and commit messages.
 362             IllegalStateException ex = new IllegalStateException(
 363                     String.format("%s: Cannot commit transaction %s - it is not the current transaction",
 364                             name, transactionID));
 365             log.error(ex.getMessage());
 366             sender.tell(new akka.actor.Status.Failure(ex), shard.self());
 367             return false;
 368         }
 369
 370         return doCommit(cohortEntry);
 371     }
 372
 373     /**
 374      * Returns the cohort entry for the Tx commit currently in progress if the given transaction ID
 375      * matches the current entry.
 376      *
 377      * @param transactionID the ID of the transaction
 378      * @return the current CohortEntry or null if the given transaction ID does not match the
 379      *         current entry.
 380      */
 381     public CohortEntry getCohortEntryIfCurrent(String transactionID) {
 382         if(isCurrentTransaction(transactionID)) {
 383             return currentCohortEntry;
 384         }
 385
 386         return null;
 387     }
 388
 389     public CohortEntry getCurrentCohortEntry() {
 390         return currentCohortEntry;
 391     }
 392
 393     public CohortEntry getAndRemoveCohortEntry(String transactionID) {
 394         CohortEntry cohortEntry = cohortCache.getIfPresent(transactionID);
 395         cohortCache.invalidate(transactionID);
 396         return cohortEntry;
 397     }
 398
 399     public void removeCohortEntry(String transactionID) {
 400         cohortCache.invalidate(transactionID);
 401     }
 402
 403     public boolean isCurrentTransaction(String transactionID) {
 404         return currentCohortEntry != null &&
 405                 currentCohortEntry.getTransactionID().equals(transactionID);
 406     }
 407
 408     /**
 409      * This method is called when a transaction is complete, successful or not. If the given
 410      * given transaction ID matches the current in-progress transaction, the next cohort entry,
 411      * if any, is dequeued and processed.
 412      *
 413      * @param transactionID the ID of the completed transaction
 414      * @param removeCohortEntry if true the CohortEntry for the transaction is also removed from
 415      *        the cache.
 416      */
 417     public void currentTransactionComplete(String transactionID, boolean removeCohortEntry) {
 418         if(removeCohortEntry) {
 419             removeCohortEntry(transactionID);
 420         }
 421
 422         if(isCurrentTransaction(transactionID)) {
 423             // Dequeue the next cohort entry waiting in the queue.
 424             currentCohortEntry = queuedCohortEntries.poll();
 425             if(currentCohortEntry != null) {
 426                 currentCohortEntry.updateLastAccessTime();
 427                 doCanCommit(currentCohortEntry);
 428             }
 429         }
 430     }
 431
 432     @VisibleForTesting
 433     void setCohortDecorator(CohortDecorator cohortDecorator) {
 434         this.cohortDecorator = cohortDecorator;
 435     }
 436
 437
 438     static class CohortEntry {
 439         private final String transactionID;
 440         private ShardDataTreeCohort cohort;
 441         private final ReadWriteShardDataTreeTransaction transaction;
 442         private ActorRef replySender;
 443         private Shard shard;
 444         private long lastAccessTime;
 445         private boolean doImmediateCommit;
 446
 447         CohortEntry(String transactionID, ReadWriteShardDataTreeTransaction transaction) {
 448             this.transaction = Preconditions.checkNotNull(transaction);
 449             this.transactionID = transactionID;
 450         }
 451
 452         CohortEntry(String transactionID, ShardDataTreeCohort cohort,
 453                 MutableCompositeModification compositeModification) {
 454             this.transactionID = transactionID;
 455             this.cohort = cohort;
 456             this.transaction = null;
 457         }
 458
 459         CohortEntry(String transactionID, ShardDataTreeCohort cohort) {
 460             this.transactionID = transactionID;
 461             this.cohort = cohort;
 462             this.transaction = null;
 463         }
 464
 465         void updateLastAccessTime() {
 466             lastAccessTime = System.currentTimeMillis();
 467         }
 468
 469         long getLastAccessTime() {
 470             return lastAccessTime;
 471         }
 472
 473         String getTransactionID() {
 474             return transactionID;
 475         }
 476
 477         ShardDataTreeCohort getCohort() {
 478             return cohort;
 479         }
 480
 481         void applyModifications(Iterable<Modification> modifications) {
 482             for (Modification modification : modifications) {
 483                 modification.apply(transaction.getSnapshot());
 484             }
 485         }
 486
 487         void ready(CohortDecorator cohortDecorator, boolean doImmediateCommit) {
 488             Preconditions.checkState(cohort == null, "cohort was already set");
 489
 490             setDoImmediateCommit(doImmediateCommit);
 491
 492             cohort = transaction.ready();
 493
 494             if(cohortDecorator != null) {
 495                 // Call the hook for unit tests.
 496                 cohort = cohortDecorator.decorate(transactionID, cohort);
 497             }
 498         }
 499
 500         boolean isDoImmediateCommit() {
 501             return doImmediateCommit;
 502         }
 503
 504         void setDoImmediateCommit(boolean doImmediateCommit) {
 505             this.doImmediateCommit = doImmediateCommit;
 506         }
 507
 508         ActorRef getReplySender() {
 509             return replySender;
 510         }
 511
 512         void setReplySender(ActorRef replySender) {
 513             this.replySender = replySender;
 514         }
 515
 516         Shard getShard() {
 517             return shard;
 518         }
 519
 520         void setShard(Shard shard) {
 521             this.shard = shard;
 522         }
 523     }
 524 }