2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.cluster.datastore;
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.Optional;
16 import com.google.common.base.Preconditions;
17 import com.google.common.collect.Lists;
18 import com.google.common.util.concurrent.CheckedFuture;
19 import com.google.common.util.concurrent.SettableFuture;
20 import java.util.ArrayList;
21 import java.util.Collection;
22 import java.util.HashMap;
23 import java.util.List;
25 import java.util.concurrent.Semaphore;
26 import java.util.concurrent.TimeUnit;
27 import java.util.concurrent.atomic.AtomicBoolean;
28 import java.util.concurrent.atomic.AtomicLong;
29 import javax.annotation.concurrent.GuardedBy;
30 import org.opendaylight.controller.cluster.datastore.compat.PreLithiumTransactionContextImpl;
31 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
32 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
33 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
34 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
35 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
36 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
37 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
38 import org.opendaylight.controller.sal.core.spi.data.AbstractDOMStoreTransaction;
39 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
40 import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
41 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
42 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
43 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
44 import org.slf4j.Logger;
45 import org.slf4j.LoggerFactory;
46 import scala.concurrent.Future;
47 import scala.concurrent.Promise;
48 import scala.concurrent.duration.FiniteDuration;
51 * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
53 * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
54 * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
55 * be created on each of those shards by the TransactionProxy
58 * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
59 * shards will be executed.
62 public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIdentifier> implements DOMStoreReadWriteTransaction {
64 public static enum TransactionType {
70 private static final TransactionType[] VALUES = values();
72 public static TransactionType fromInt(final int type) {
75 } catch (IndexOutOfBoundsException e) {
76 throw new IllegalArgumentException("In TransactionType enum value " + type, e);
81 private static enum TransactionState {
87 static final Mapper<Throwable, Throwable> SAME_FAILURE_TRANSFORMER =
88 new Mapper<Throwable, Throwable>() {
90 public Throwable apply(Throwable failure) {
95 private static final AtomicLong counter = new AtomicLong();
97 private static final Logger LOG = LoggerFactory.getLogger(TransactionProxy.class);
100 * Time interval in between transaction create retries.
102 private static final FiniteDuration CREATE_TX_TRY_INTERVAL =
103 FiniteDuration.create(1, TimeUnit.SECONDS);
106 * Stores the remote Tx actors for each requested data store path to be used by the
107 * PhantomReference to close the remote Tx's. This is only used for read-only Tx's. The
108 * remoteTransactionActorsMB volatile serves as a memory barrier to publish updates to the
109 * remoteTransactionActors list so they will be visible to the thread accessing the
112 List<ActorSelection> remoteTransactionActors;
113 volatile AtomicBoolean remoteTransactionActorsMB;
116 * Stores the create transaction results per shard.
118 private final Map<String, TransactionFutureCallback> txFutureCallbackMap = new HashMap<>();
120 private final TransactionType transactionType;
121 final ActorContext actorContext;
122 private final String transactionChainId;
123 private final SchemaContext schemaContext;
124 private TransactionState state = TransactionState.OPEN;
126 private volatile boolean initialized;
127 private Semaphore operationLimiter;
128 private OperationCompleter operationCompleter;
130 public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
131 this(actorContext, transactionType, "");
134 public TransactionProxy(ActorContext actorContext, TransactionType transactionType, String transactionChainId) {
135 super(createIdentifier(actorContext));
136 this.actorContext = Preconditions.checkNotNull(actorContext,
137 "actorContext should not be null");
138 this.transactionType = Preconditions.checkNotNull(transactionType,
139 "transactionType should not be null");
140 this.schemaContext = Preconditions.checkNotNull(actorContext.getSchemaContext(),
141 "schemaContext should not be null");
142 this.transactionChainId = transactionChainId;
144 LOG.debug("Created txn {} of type {} on chain {}", getIdentifier(), transactionType, transactionChainId);
147 private static TransactionIdentifier createIdentifier(ActorContext actorContext) {
148 String memberName = actorContext.getCurrentMemberName();
149 if (memberName == null) {
150 memberName = "UNKNOWN-MEMBER";
153 return new TransactionIdentifier(memberName, counter.getAndIncrement());
157 List<Future<Object>> getRecordedOperationFutures() {
158 List<Future<Object>> recordedOperationFutures = Lists.newArrayList();
159 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
160 TransactionContext transactionContext = txFutureCallback.getTransactionContext();
161 if (transactionContext != null) {
162 transactionContext.copyRecordedOperationFutures(recordedOperationFutures);
166 return recordedOperationFutures;
170 boolean hasTransactionContext() {
171 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
172 TransactionContext transactionContext = txFutureCallback.getTransactionContext();
173 if(transactionContext != null) {
182 public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
184 Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
185 "Read operation on write-only transaction is not allowed");
187 LOG.debug("Tx {} read {}", getIdentifier(), path);
191 final SettableFuture<Optional<NormalizedNode<?, ?>>> proxyFuture = SettableFuture.create();
193 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
194 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
196 public void invoke(TransactionContext transactionContext) {
197 transactionContext.readData(path, proxyFuture);
201 return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
205 public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
207 Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
208 "Exists operation on write-only transaction is not allowed");
210 LOG.debug("Tx {} exists {}", getIdentifier(), path);
214 final SettableFuture<Boolean> proxyFuture = SettableFuture.create();
216 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
217 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
219 public void invoke(TransactionContext transactionContext) {
220 transactionContext.dataExists(path, proxyFuture);
224 return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
227 private void checkModificationState() {
228 Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
229 "Modification operation on read-only transaction is not allowed");
230 Preconditions.checkState(state == TransactionState.OPEN,
231 "Transaction is sealed - further modifications are not allowed");
234 private void throttleOperation() {
235 throttleOperation(1);
238 private void throttleOperation(int acquirePermits) {
240 // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
241 operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
242 operationCompleter = new OperationCompleter(operationLimiter);
244 // Make sure we write this last because it's volatile and will also publish the non-volatile writes
245 // above as well so they'll be visible to other threads.
250 if(!operationLimiter.tryAcquire(acquirePermits,
251 actorContext.getDatastoreContext().getOperationTimeoutInSeconds(), TimeUnit.SECONDS)){
252 LOG.warn("Failed to acquire operation permit for transaction {}", getIdentifier());
254 } catch (InterruptedException e) {
255 if(LOG.isDebugEnabled()) {
256 LOG.debug("Interrupted when trying to acquire operation permit for transaction " + getIdentifier().toString(), e);
258 LOG.warn("Interrupted when trying to acquire operation permit for transaction {}", getIdentifier());
264 public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
266 checkModificationState();
268 LOG.debug("Tx {} write {}", getIdentifier(), path);
272 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
273 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
275 public void invoke(TransactionContext transactionContext) {
276 transactionContext.writeData(path, data);
282 public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
284 checkModificationState();
286 LOG.debug("Tx {} merge {}", getIdentifier(), path);
290 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
291 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
293 public void invoke(TransactionContext transactionContext) {
294 transactionContext.mergeData(path, data);
300 public void delete(final YangInstanceIdentifier path) {
302 checkModificationState();
304 LOG.debug("Tx {} delete {}", getIdentifier(), path);
308 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
309 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
311 public void invoke(TransactionContext transactionContext) {
312 transactionContext.deleteData(path);
317 private boolean seal(final TransactionState newState) {
318 if (state == TransactionState.OPEN) {
327 public AbstractThreePhaseCommitCohort ready() {
328 Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
329 "Read-only transactions cannot be readied");
331 final boolean success = seal(TransactionState.READY);
332 Preconditions.checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state);
334 LOG.debug("Tx {} Readying {} transactions for commit", getIdentifier(),
335 txFutureCallbackMap.size());
337 if (txFutureCallbackMap.isEmpty()) {
338 TransactionRateLimitingCallback.adjustRateLimitForUnusedTransaction(actorContext);
339 return NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
342 throttleOperation(txFutureCallbackMap.size());
344 List<Future<ActorSelection>> cohortFutures = new ArrayList<>(txFutureCallbackMap.size());
345 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
347 LOG.debug("Tx {} Readying transaction for shard {} chain {}", getIdentifier(),
348 txFutureCallback.getShardName(), transactionChainId);
350 final TransactionContext transactionContext = txFutureCallback.getTransactionContext();
351 final Future<ActorSelection> future;
352 if (transactionContext != null) {
353 // avoid the creation of a promise and a TransactionOperation
354 future = transactionContext.readyTransaction();
356 final Promise<ActorSelection> promise = akka.dispatch.Futures.promise();
357 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
359 public void invoke(TransactionContext transactionContext) {
360 promise.completeWith(transactionContext.readyTransaction());
363 future = promise.future();
366 cohortFutures.add(future);
369 return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
370 getIdentifier().toString());
374 public void close() {
375 if (!seal(TransactionState.CLOSED)) {
376 if (state == TransactionState.CLOSED) {
377 // Idempotent no-op as per AutoCloseable recommendation
381 throw new IllegalStateException(String.format("Transaction %s is ready, it cannot be closed",
385 for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
386 txFutureCallback.enqueueTransactionOperation(new TransactionOperation() {
388 public void invoke(TransactionContext transactionContext) {
389 transactionContext.closeTransaction();
394 txFutureCallbackMap.clear();
396 if(remoteTransactionActorsMB != null) {
397 remoteTransactionActors.clear();
398 remoteTransactionActorsMB.set(true);
402 private String shardNameFromIdentifier(YangInstanceIdentifier path){
403 return ShardStrategyFactory.getStrategy(path).findShard(path);
406 protected Future<ActorSelection> sendFindPrimaryShardAsync(String shardName) {
407 return actorContext.findPrimaryShardAsync(shardName);
410 private TransactionFutureCallback getOrCreateTxFutureCallback(YangInstanceIdentifier path) {
411 String shardName = shardNameFromIdentifier(path);
412 TransactionFutureCallback txFutureCallback = txFutureCallbackMap.get(shardName);
413 if(txFutureCallback == null) {
414 Future<ActorSelection> findPrimaryFuture = sendFindPrimaryShardAsync(shardName);
416 final TransactionFutureCallback newTxFutureCallback = new TransactionFutureCallback(shardName);
418 txFutureCallback = newTxFutureCallback;
419 txFutureCallbackMap.put(shardName, txFutureCallback);
421 findPrimaryFuture.onComplete(new OnComplete<ActorSelection>() {
423 public void onComplete(Throwable failure, ActorSelection primaryShard) {
424 if(failure != null) {
425 newTxFutureCallback.createTransactionContext(failure, null);
427 newTxFutureCallback.setPrimaryShard(primaryShard);
430 }, actorContext.getClientDispatcher());
433 return txFutureCallback;
436 public String getTransactionChainId() {
437 return transactionChainId;
440 protected ActorContext getActorContext() {
445 * Implements a Future OnComplete callback for a CreateTransaction message. This class handles
446 * retries, up to a limit, if the shard doesn't have a leader yet. This is done by scheduling a
447 * retry task after a short delay.
449 * The end result from a completed CreateTransaction message is a TransactionContext that is
450 * used to perform transaction operations. Transaction operations that occur before the
451 * CreateTransaction completes are cache and executed once the CreateTransaction completes,
452 * successfully or not.
454 private class TransactionFutureCallback extends OnComplete<Object> {
457 * The list of transaction operations to execute once the CreateTransaction completes.
459 @GuardedBy("txOperationsOnComplete")
460 private final List<TransactionOperation> txOperationsOnComplete = Lists.newArrayList();
463 * The TransactionContext resulting from the CreateTransaction reply.
465 private volatile TransactionContext transactionContext;
468 * The target primary shard.
470 private volatile ActorSelection primaryShard;
472 private volatile int createTxTries = (int) (actorContext.getDatastoreContext().
473 getShardLeaderElectionTimeout().duration().toMillis() /
474 CREATE_TX_TRY_INTERVAL.toMillis());
476 private final String shardName;
478 TransactionFutureCallback(String shardName) {
479 this.shardName = shardName;
482 String getShardName() {
486 TransactionContext getTransactionContext() {
487 return transactionContext;
492 * Sets the target primary shard and initiates a CreateTransaction try.
494 void setPrimaryShard(ActorSelection primaryShard) {
495 this.primaryShard = primaryShard;
497 if(transactionType == TransactionType.WRITE_ONLY &&
498 actorContext.getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
499 LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context",
500 getIdentifier(), primaryShard);
502 // For write-only Tx's we prepare the transaction modifications directly on the shard actor
503 // to avoid the overhead of creating a separate transaction actor.
504 // FIXME: can't assume the shard version is LITHIUM_VERSION - need to obtain it somehow.
505 executeTxOperatonsOnComplete(createValidTransactionContext(this.primaryShard,
506 this.primaryShard.path().toString(), DataStoreVersions.LITHIUM_VERSION));
508 tryCreateTransaction();
513 * Adds a TransactionOperation to be executed after the CreateTransaction completes.
515 void addTxOperationOnComplete(TransactionOperation operation) {
516 boolean invokeOperation = true;
517 synchronized(txOperationsOnComplete) {
518 if(transactionContext == null) {
519 LOG.debug("Tx {} Adding operation on complete", getIdentifier());
521 invokeOperation = false;
522 txOperationsOnComplete.add(operation);
526 if(invokeOperation) {
527 operation.invoke(transactionContext);
531 void enqueueTransactionOperation(final TransactionOperation op) {
533 if (transactionContext != null) {
534 op.invoke(transactionContext);
536 // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
537 // callback to be executed after the Tx is created.
538 addTxOperationOnComplete(op);
543 * Performs a CreateTransaction try async.
545 private void tryCreateTransaction() {
546 if(LOG.isDebugEnabled()) {
547 LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(), primaryShard);
550 Object serializedCreateMessage = new CreateTransaction(getIdentifier().toString(),
551 TransactionProxy.this.transactionType.ordinal(),
552 getTransactionChainId()).toSerializable();
554 Future<Object> createTxFuture = actorContext.executeOperationAsync(primaryShard, serializedCreateMessage);
556 createTxFuture.onComplete(this, actorContext.getClientDispatcher());
560 public void onComplete(Throwable failure, Object response) {
561 if(failure instanceof NoShardLeaderException) {
562 // There's no leader for the shard yet - schedule and try again, unless we're out
563 // of retries. Note: createTxTries is volatile as it may be written by different
564 // threads however not concurrently, therefore decrementing it non-atomically here
566 if(--createTxTries > 0) {
567 LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
568 getIdentifier(), shardName);
570 actorContext.getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
574 tryCreateTransaction();
576 }, actorContext.getClientDispatcher());
581 createTransactionContext(failure, response);
584 private void createTransactionContext(Throwable failure, Object response) {
585 // Mainly checking for state violation here to perform a volatile read of "initialized" to
586 // ensure updates to operationLimter et al are visible to this thread (ie we're doing
587 // "piggy-back" synchronization here).
588 Preconditions.checkState(initialized, "Tx was not propertly initialized.");
590 // Create the TransactionContext from the response or failure. Store the new
591 // TransactionContext locally until we've completed invoking the
592 // TransactionOperations. This avoids thread timing issues which could cause
593 // out-of-order TransactionOperations. Eg, on a modification operation, if the
594 // TransactionContext is non-null, then we directly call the TransactionContext.
595 // However, at the same time, the code may be executing the cached
596 // TransactionOperations. So to avoid thus timing, we don't publish the
597 // TransactionContext until after we've executed all cached TransactionOperations.
598 TransactionContext localTransactionContext;
599 if(failure != null) {
600 LOG.debug("Tx {} Creating NoOpTransaction because of error", getIdentifier(), failure);
602 localTransactionContext = new NoOpTransactionContext(failure, getIdentifier(), operationLimiter);
603 } else if (CreateTransactionReply.SERIALIZABLE_CLASS.equals(response.getClass())) {
604 localTransactionContext = createValidTransactionContext(
605 CreateTransactionReply.fromSerializable(response));
607 IllegalArgumentException exception = new IllegalArgumentException(String.format(
608 "Invalid reply type %s for CreateTransaction", response.getClass()));
610 localTransactionContext = new NoOpTransactionContext(exception, getIdentifier(), operationLimiter);
613 executeTxOperatonsOnComplete(localTransactionContext);
616 private void executeTxOperatonsOnComplete(TransactionContext localTransactionContext) {
618 // Access to txOperationsOnComplete and transactionContext must be protected and atomic
619 // (ie synchronized) with respect to #addTxOperationOnComplete to handle timing
620 // issues and ensure no TransactionOperation is missed and that they are processed
621 // in the order they occurred.
623 // We'll make a local copy of the txOperationsOnComplete list to handle re-entrancy
624 // in case a TransactionOperation results in another transaction operation being
625 // queued (eg a put operation from a client read Future callback that is notified
627 Collection<TransactionOperation> operationsBatch = null;
628 synchronized(txOperationsOnComplete) {
629 if(txOperationsOnComplete.isEmpty()) {
630 // We're done invoking the TransactionOperations so we can now publish the
631 // TransactionContext.
632 transactionContext = localTransactionContext;
636 operationsBatch = new ArrayList<>(txOperationsOnComplete);
637 txOperationsOnComplete.clear();
640 // Invoke TransactionOperations outside the sync block to avoid unnecessary blocking.
641 // A slight down-side is that we need to re-acquire the lock below but this should
643 for(TransactionOperation oper: operationsBatch) {
644 oper.invoke(localTransactionContext);
649 private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
650 LOG.debug("Tx {} Received {}", getIdentifier(), reply);
652 return createValidTransactionContext(actorContext.actorSelection(reply.getTransactionPath()),
653 reply.getTransactionPath(), reply.getVersion());
656 private TransactionContext createValidTransactionContext(ActorSelection transactionActor,
657 String transactionPath, short remoteTransactionVersion) {
659 if (transactionType == TransactionType.READ_ONLY) {
660 // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference
661 // to close the remote Tx's when this instance is no longer in use and is garbage
664 if(remoteTransactionActorsMB == null) {
665 remoteTransactionActors = Lists.newArrayList();
666 remoteTransactionActorsMB = new AtomicBoolean();
668 TransactionProxyCleanupPhantomReference.track(TransactionProxy.this);
671 // Add the actor to the remoteTransactionActors list for access by the
672 // cleanup PhantonReference.
673 remoteTransactionActors.add(transactionActor);
675 // Write to the memory barrier volatile to publish the above update to the
676 // remoteTransactionActors list for thread visibility.
677 remoteTransactionActorsMB.set(true);
680 // TxActor is always created where the leader of the shard is.
681 // Check if TxActor is created in the same node
682 boolean isTxActorLocal = actorContext.isPathLocal(transactionPath);
684 if(remoteTransactionVersion < DataStoreVersions.LITHIUM_VERSION) {
685 return new PreLithiumTransactionContextImpl(transactionPath, transactionActor, getIdentifier(),
686 transactionChainId, actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion,
688 } else if (transactionType == TransactionType.WRITE_ONLY &&
689 actorContext.getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
690 return new WriteOnlyTransactionContextImpl(transactionActor, getIdentifier(), transactionChainId,
691 actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter);
693 return new TransactionContextImpl(transactionActor, getIdentifier(), transactionChainId,
694 actorContext, schemaContext, isTxActorLocal, remoteTransactionVersion, operationCompleter);