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.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;
27 import java.util.concurrent.ConcurrentHashMap;
28 import java.util.concurrent.Semaphore;
29 import java.util.concurrent.TimeUnit;
30 import java.util.concurrent.atomic.AtomicBoolean;
31 import java.util.concurrent.atomic.AtomicLong;
32 import javax.annotation.concurrent.GuardedBy;
33 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
34 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
35 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
36 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
37 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
38 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
39 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
40 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
41 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
42 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
43 import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
44 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
45 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
46 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
47 import org.slf4j.Logger;
48 import org.slf4j.LoggerFactory;
49 import scala.concurrent.Future;
50 import scala.concurrent.Promise;
51 import scala.concurrent.duration.FiniteDuration;
54 * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
56 * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
57 * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
58 * be created on each of those shards by the TransactionProxy
61 * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
62 * shards will be executed.
65 public class TransactionProxy implements DOMStoreReadWriteTransaction {
67 public static enum TransactionType {
73 static final Mapper<Throwable, Throwable> SAME_FAILURE_TRANSFORMER =
74 new Mapper<Throwable, Throwable>() {
76 public Throwable apply(Throwable failure) {
81 private static final AtomicLong counter = new AtomicLong();
83 private static final Logger LOG = LoggerFactory.getLogger(TransactionProxy.class);
86 * Time interval in between transaction create retries.
88 private static final FiniteDuration CREATE_TX_TRY_INTERVAL =
89 FiniteDuration.create(1, TimeUnit.SECONDS);
92 * Used to enqueue the PhantomReferences for read-only TransactionProxy instances. The
93 * FinalizableReferenceQueue is safe to use statically in an OSGi environment as it uses some
94 * trickery to clean up its internal thread when the bundle is unloaded.
96 private static final FinalizableReferenceQueue phantomReferenceQueue =
97 new FinalizableReferenceQueue();
100 * This stores the TransactionProxyCleanupPhantomReference instances statically, This is
101 * necessary because PhantomReferences need a hard reference so they're not garbage collected.
102 * Once finalized, the TransactionProxyCleanupPhantomReference removes itself from this map
103 * and thus becomes eligible for garbage collection.
105 private static final Map<TransactionProxyCleanupPhantomReference,
106 TransactionProxyCleanupPhantomReference> phantomReferenceCache =
107 new ConcurrentHashMap<>();
110 * A PhantomReference that closes remote transactions for a TransactionProxy when it's
111 * garbage collected. This is used for read-only transactions as they're not explicitly closed
112 * by clients. So the only way to detect that a transaction is no longer in use and it's safe
113 * to clean up is when it's garbage collected. It's inexact as to when an instance will be GC'ed
114 * but TransactionProxy instances should generally be short-lived enough to avoid being moved
115 * to the old generation space and thus should be cleaned up in a timely manner as the GC
116 * runs on the young generation (eden, swap1...) space much more frequently.
118 private static class TransactionProxyCleanupPhantomReference
119 extends FinalizablePhantomReference<TransactionProxy> {
121 private final List<ActorSelection> remoteTransactionActors;
122 private final AtomicBoolean remoteTransactionActorsMB;
123 private final ActorContext actorContext;
124 private final TransactionIdentifier identifier;
126 protected TransactionProxyCleanupPhantomReference(TransactionProxy referent) {
127 super(referent, phantomReferenceQueue);
129 // Note we need to cache the relevant fields from the TransactionProxy as we can't
130 // have a hard reference to the TransactionProxy instance itself.
132 remoteTransactionActors = referent.remoteTransactionActors;
133 remoteTransactionActorsMB = referent.remoteTransactionActorsMB;
134 actorContext = referent.actorContext;
135 identifier = referent.identifier;
139 public void finalizeReferent() {
140 LOG.trace("Cleaning up {} Tx actors for TransactionProxy {}",
141 remoteTransactionActors.size(), identifier);
143 phantomReferenceCache.remove(this);
145 // Access the memory barrier volatile to ensure all previous updates to the
146 // remoteTransactionActors list are visible to this thread.
148 if(remoteTransactionActorsMB.get()) {
149 for(ActorSelection actor : remoteTransactionActors) {
150 LOG.trace("Sending CloseTransaction to {}", actor);
151 actorContext.sendOperationAsync(actor, CloseTransaction.INSTANCE.toSerializable());
158 * Stores the remote Tx actors for each requested data store path to be used by the
159 * PhantomReference to close the remote Tx's. This is only used for read-only Tx's. The
160 * remoteTransactionActorsMB volatile serves as a memory barrier to publish updates to the
161 * remoteTransactionActors list so they will be visible to the thread accessing the
164 private List<ActorSelection> remoteTransactionActors;
165 private AtomicBoolean remoteTransactionActorsMB;
168 * Stores the create transaction results per shard.
170 private final Map<String, TransactionFutureCallback> txFutureCallbackMap = new HashMap<>();
172 private final TransactionType transactionType;
173 private final ActorContext actorContext;
174 private final TransactionIdentifier identifier;
175 private final String transactionChainId;
176 private final SchemaContext schemaContext;
177 private boolean inReadyState;
178 private final Semaphore operationLimiter;
179 private final OperationCompleter operationCompleter;
181 public TransactionProxy(ActorContext actorContext, TransactionType transactionType) {
182 this(actorContext, transactionType, "");
185 public TransactionProxy(ActorContext actorContext, TransactionType transactionType,
186 String transactionChainId) {
187 this.actorContext = Preconditions.checkNotNull(actorContext,
188 "actorContext should not be null");
189 this.transactionType = Preconditions.checkNotNull(transactionType,
190 "transactionType should not be null");
191 this.schemaContext = Preconditions.checkNotNull(actorContext.getSchemaContext(),
192 "schemaContext should not be null");
193 this.transactionChainId = transactionChainId;
195 String memberName = actorContext.getCurrentMemberName();
196 if(memberName == null){
197 memberName = "UNKNOWN-MEMBER";
200 this.identifier = TransactionIdentifier.builder().memberName(memberName).counter(
201 counter.getAndIncrement()).build();
203 if(transactionType == TransactionType.READ_ONLY) {
204 // Read-only Tx's aren't explicitly closed by the client so we create a PhantomReference
205 // to close the remote Tx's when this instance is no longer in use and is garbage
208 remoteTransactionActors = Lists.newArrayList();
209 remoteTransactionActorsMB = new AtomicBoolean();
211 TransactionProxyCleanupPhantomReference cleanup =
212 new TransactionProxyCleanupPhantomReference(this);
213 phantomReferenceCache.put(cleanup, cleanup);
216 // Note : Currently mailbox-capacity comes from akka.conf and not from the config-subsystem
217 this.operationLimiter = new Semaphore(actorContext.getTransactionOutstandingOperationLimit());
218 this.operationCompleter = new OperationCompleter(operationLimiter);
220 LOG.debug("Created txn {} of type {} on chain {}", identifier, transactionType, transactionChainId);
224 List<Future<Object>> getRecordedOperationFutures() {
225 List<Future<Object>> recordedOperationFutures = Lists.newArrayList();
226 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
227 TransactionContext transactionContext = txFutureCallback.getTransactionContext();
228 if(transactionContext != null) {
229 recordedOperationFutures.addAll(transactionContext.getRecordedOperationFutures());
233 return recordedOperationFutures;
237 boolean hasTransactionContext() {
238 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
239 TransactionContext transactionContext = txFutureCallback.getTransactionContext();
240 if(transactionContext != null) {
249 public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
251 Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
252 "Read operation on write-only transaction is not allowed");
254 LOG.debug("Tx {} read {}", identifier, path);
258 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
259 return txFutureCallback.enqueueReadOperation(new ReadOperation<Optional<NormalizedNode<?, ?>>>() {
261 public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> invoke(
262 TransactionContext transactionContext) {
263 return transactionContext.readData(path);
269 public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
271 Preconditions.checkState(transactionType != TransactionType.WRITE_ONLY,
272 "Exists operation on write-only transaction is not allowed");
274 LOG.debug("Tx {} exists {}", identifier, path);
278 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
279 return txFutureCallback.enqueueReadOperation(new ReadOperation<Boolean>() {
281 public CheckedFuture<Boolean, ReadFailedException> invoke(TransactionContext transactionContext) {
282 return transactionContext.dataExists(path);
288 private void checkModificationState() {
289 Preconditions.checkState(transactionType != TransactionType.READ_ONLY,
290 "Modification operation on read-only transaction is not allowed");
291 Preconditions.checkState(!inReadyState,
292 "Transaction is sealed - further modifications are not allowed");
295 private void throttleOperation() {
296 throttleOperation(1);
299 private void throttleOperation(int acquirePermits) {
301 if(!operationLimiter.tryAcquire(acquirePermits, actorContext.getDatastoreContext().getOperationTimeoutInSeconds(), TimeUnit.SECONDS)){
302 LOG.warn("Failed to acquire operation permit for transaction {}", getIdentifier());
304 } catch (InterruptedException e) {
305 if(LOG.isDebugEnabled()) {
306 LOG.debug("Interrupted when trying to acquire operation permit for transaction " + getIdentifier().toString(), e);
308 LOG.warn("Interrupted when trying to acquire operation permit for transaction {}", getIdentifier());
315 public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
317 checkModificationState();
319 LOG.debug("Tx {} write {}", identifier, path);
323 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
324 txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
326 public void invoke(TransactionContext transactionContext) {
327 transactionContext.writeData(path, data);
333 public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
335 checkModificationState();
337 LOG.debug("Tx {} merge {}", identifier, path);
341 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
342 txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
344 public void invoke(TransactionContext transactionContext) {
345 transactionContext.mergeData(path, data);
351 public void delete(final YangInstanceIdentifier path) {
353 checkModificationState();
355 LOG.debug("Tx {} delete {}", identifier, path);
359 TransactionFutureCallback txFutureCallback = getOrCreateTxFutureCallback(path);
360 txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
362 public void invoke(TransactionContext transactionContext) {
363 transactionContext.deleteData(path);
369 public DOMStoreThreePhaseCommitCohort ready() {
371 checkModificationState();
373 throttleOperation(txFutureCallbackMap.size());
377 LOG.debug("Tx {} Readying {} transactions for commit", identifier,
378 txFutureCallbackMap.size());
380 List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
382 for(TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
384 LOG.debug("Tx {} Readying transaction for shard {} chain {}", identifier,
385 txFutureCallback.getShardName(), transactionChainId);
387 Future<ActorSelection> future = txFutureCallback.enqueueFutureOperation(new FutureOperation<ActorSelection>() {
389 public Future<ActorSelection> invoke(TransactionContext transactionContext) {
390 return transactionContext.readyTransaction();
394 cohortFutures.add(future);
397 onTransactionReady(cohortFutures);
399 return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures,
400 identifier.toString());
404 * Method for derived classes to be notified when the transaction has been readied.
406 * @param cohortFutures the cohort Futures for each shard transaction.
408 protected void onTransactionReady(List<Future<ActorSelection>> cohortFutures) {
412 * Method called to send a CreateTransaction message to a shard.
414 * @param shard the shard actor to send to
415 * @param serializedCreateMessage the serialized message to send
416 * @return the response Future
418 protected Future<Object> sendCreateTransaction(ActorSelection shard,
419 Object serializedCreateMessage) {
420 return actorContext.executeOperationAsync(shard, serializedCreateMessage);
424 public Object getIdentifier() {
425 return this.identifier;
429 public void close() {
430 for (TransactionFutureCallback txFutureCallback : txFutureCallbackMap.values()) {
431 txFutureCallback.enqueueModifyOperation(new TransactionOperation() {
433 public void invoke(TransactionContext transactionContext) {
434 transactionContext.closeTransaction();
439 txFutureCallbackMap.clear();
441 if(transactionType == TransactionType.READ_ONLY) {
442 remoteTransactionActors.clear();
443 remoteTransactionActorsMB.set(true);
447 private String shardNameFromIdentifier(YangInstanceIdentifier path){
448 return ShardStrategyFactory.getStrategy(path).findShard(path);
451 private TransactionFutureCallback getOrCreateTxFutureCallback(YangInstanceIdentifier path) {
452 String shardName = shardNameFromIdentifier(path);
453 TransactionFutureCallback txFutureCallback = txFutureCallbackMap.get(shardName);
454 if(txFutureCallback == null) {
455 Future<ActorSelection> findPrimaryFuture = actorContext.findPrimaryShardAsync(shardName);
457 final TransactionFutureCallback newTxFutureCallback =
458 new TransactionFutureCallback(shardName);
460 txFutureCallback = newTxFutureCallback;
461 txFutureCallbackMap.put(shardName, txFutureCallback);
463 findPrimaryFuture.onComplete(new OnComplete<ActorSelection>() {
465 public void onComplete(Throwable failure, ActorSelection primaryShard) {
466 if(failure != null) {
467 newTxFutureCallback.onComplete(failure, null);
469 newTxFutureCallback.setPrimaryShard(primaryShard);
472 }, actorContext.getActorSystem().dispatcher());
475 return txFutureCallback;
478 public String getTransactionChainId() {
479 return transactionChainId;
482 protected ActorContext getActorContext() {
487 * Interfaces for transaction operations to be invoked later.
489 private static interface TransactionOperation {
490 void invoke(TransactionContext transactionContext);
494 * This interface returns a Guava Future
496 private static interface ReadOperation<T> {
497 CheckedFuture<T, ReadFailedException> invoke(TransactionContext transactionContext);
501 * This interface returns a Scala Future
503 private static interface FutureOperation<T> {
504 Future<T> invoke(TransactionContext transactionContext);
508 * Implements a Future OnComplete callback for a CreateTransaction message. This class handles
509 * retries, up to a limit, if the shard doesn't have a leader yet. This is done by scheduling a
510 * retry task after a short delay.
512 * The end result from a completed CreateTransaction message is a TransactionContext that is
513 * used to perform transaction operations. Transaction operations that occur before the
514 * CreateTransaction completes are cache and executed once the CreateTransaction completes,
515 * successfully or not.
517 private class TransactionFutureCallback extends OnComplete<Object> {
520 * The list of transaction operations to execute once the CreateTransaction completes.
522 @GuardedBy("txOperationsOnComplete")
523 private final List<TransactionOperation> txOperationsOnComplete = Lists.newArrayList();
526 * The TransactionContext resulting from the CreateTransaction reply.
528 private volatile TransactionContext transactionContext;
531 * The target primary shard.
533 private volatile ActorSelection primaryShard;
535 private volatile int createTxTries = (int) (actorContext.getDatastoreContext().
536 getShardLeaderElectionTimeout().duration().toMillis() /
537 CREATE_TX_TRY_INTERVAL.toMillis());
539 private final String shardName;
541 TransactionFutureCallback(String shardName) {
542 this.shardName = shardName;
545 String getShardName() {
549 TransactionContext getTransactionContext() {
550 return transactionContext;
555 * Sets the target primary shard and initiates a CreateTransaction try.
557 void setPrimaryShard(ActorSelection primaryShard) {
558 LOG.debug("Tx {} Primary shard found - trying create transaction", identifier);
560 this.primaryShard = primaryShard;
561 tryCreateTransaction();
565 * Adds a TransactionOperation to be executed after the CreateTransaction completes.
567 void addTxOperationOnComplete(TransactionOperation operation) {
568 synchronized(txOperationsOnComplete) {
569 if(transactionContext == null) {
570 LOG.debug("Tx {} Adding operation on complete {}", identifier);
572 txOperationsOnComplete.add(operation);
574 operation.invoke(transactionContext);
580 <T> Future<T> enqueueFutureOperation(final FutureOperation<T> op) {
584 if (transactionContext != null) {
585 future = op.invoke(transactionContext);
587 // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
588 // callback to be executed after the Tx is created.
589 final Promise<T> promise = akka.dispatch.Futures.promise();
590 addTxOperationOnComplete(new TransactionOperation() {
592 public void invoke(TransactionContext transactionContext) {
593 promise.completeWith(op.invoke(transactionContext));
597 future = promise.future();
603 <T> CheckedFuture<T, ReadFailedException> enqueueReadOperation(final ReadOperation<T> op) {
605 CheckedFuture<T, ReadFailedException> future;
607 if (transactionContext != null) {
608 future = op.invoke(transactionContext);
610 // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
611 // callback to be executed after the Tx is created.
612 final SettableFuture<T> proxyFuture = SettableFuture.create();
613 addTxOperationOnComplete(new TransactionOperation() {
615 public void invoke(TransactionContext transactionContext) {
616 Futures.addCallback(op.invoke(transactionContext), new FutureCallback<T>() {
618 public void onSuccess(T data) {
619 proxyFuture.set(data);
623 public void onFailure(Throwable t) {
624 proxyFuture.setException(t);
630 future = MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
636 void enqueueModifyOperation(final TransactionOperation op) {
638 if (transactionContext != null) {
639 op.invoke(transactionContext);
641 // The shard Tx hasn't been created yet so add the Tx operation to the Tx Future
642 // callback to be executed after the Tx is created.
643 addTxOperationOnComplete(op);
648 * Performs a CreateTransaction try async.
650 private void tryCreateTransaction() {
651 Future<Object> createTxFuture = sendCreateTransaction(primaryShard,
652 new CreateTransaction(identifier.toString(),
653 TransactionProxy.this.transactionType.ordinal(),
654 getTransactionChainId()).toSerializable());
656 createTxFuture.onComplete(this, actorContext.getActorSystem().dispatcher());
660 public void onComplete(Throwable failure, Object response) {
661 if(failure instanceof NoShardLeaderException) {
662 // There's no leader for the shard yet - schedule and try again, unless we're out
663 // of retries. Note: createTxTries is volatile as it may be written by different
664 // threads however not concurrently, therefore decrementing it non-atomically here
666 if(--createTxTries > 0) {
667 LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
668 identifier, shardName);
670 actorContext.getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
674 tryCreateTransaction();
676 }, actorContext.getActorSystem().dispatcher());
681 // Create the TransactionContext from the response or failure and execute delayed
682 // TransactionOperations. This entire section is done atomically (ie synchronized) with
683 // respect to #addTxOperationOnComplete to handle timing issues and ensure no
684 // TransactionOperation is missed and that they are processed in the order they occurred.
685 synchronized(txOperationsOnComplete) {
686 // Store the new TransactionContext locally until we've completed invoking the
687 // TransactionOperations. This avoids thread timing issues which could cause
688 // out-of-order TransactionOperations. Eg, on a modification operation, if the
689 // TransactionContext is non-null, then we directly call the TransactionContext.
690 // However, at the same time, the code may be executing the cached
691 // TransactionOperations. So to avoid thus timing, we don't publish the
692 // TransactionContext until after we've executed all cached TransactionOperations.
693 TransactionContext localTransactionContext;
694 if(failure != null) {
695 LOG.debug("Tx {} Creating NoOpTransaction because of error: {}", identifier,
696 failure.getMessage());
698 localTransactionContext = new NoOpTransactionContext(failure, identifier, operationLimiter);
699 } else if (response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
700 localTransactionContext = createValidTransactionContext(
701 CreateTransactionReply.fromSerializable(response));
703 IllegalArgumentException exception = new IllegalArgumentException(String.format(
704 "Invalid reply type %s for CreateTransaction", response.getClass()));
706 localTransactionContext = new NoOpTransactionContext(exception, identifier, operationLimiter);
709 for(TransactionOperation oper: txOperationsOnComplete) {
710 oper.invoke(localTransactionContext);
713 txOperationsOnComplete.clear();
715 // We're done invoking the TransactionOperations so we can now publish the
716 // TransactionContext.
717 transactionContext = localTransactionContext;
721 private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
722 String transactionPath = reply.getTransactionPath();
724 LOG.debug("Tx {} Received transaction actor path {}", identifier, transactionPath);
726 ActorSelection transactionActor = actorContext.actorSelection(transactionPath);
728 if (transactionType == TransactionType.READ_ONLY) {
729 // Add the actor to the remoteTransactionActors list for access by the
730 // cleanup PhantonReference.
731 remoteTransactionActors.add(transactionActor);
733 // Write to the memory barrier volatile to publish the above update to the
734 // remoteTransactionActors list for thread visibility.
735 remoteTransactionActorsMB.set(true);
738 // TxActor is always created where the leader of the shard is.
739 // Check if TxActor is created in the same node
740 boolean isTxActorLocal = actorContext.isPathLocal(transactionPath);
742 return new TransactionContextImpl(transactionPath, transactionActor, identifier,
743 actorContext, schemaContext, isTxActorLocal, reply.getVersion(), operationCompleter);