X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionChainProxy.java;h=58ac1d8b8265bc50fb7d38dea1dd9c1b916211fc;hb=f97618f25dfc073d1de5d883f1794eefdb3e5c16;hp=b467ee4ddbf56c456c2e9f0f62381eefa173e31d;hpb=3c2de556e2b813b8da925199cb2dc4389c76ef39;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionChainProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionChainProxy.java index b467ee4ddb..58ac1d8b82 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionChainProxy.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionChainProxy.java @@ -9,71 +9,213 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorSelection; -import akka.dispatch.Futures; +import akka.dispatch.OnComplete; +import com.google.common.base.Preconditions; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; +import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; -import scala.concurrent.Await; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import scala.concurrent.Future; - -import java.util.Collections; -import java.util.List; +import scala.concurrent.Promise; /** * TransactionChainProxy acts as a proxy for a DOMStoreTransactionChain created on a remote shard */ -public class TransactionChainProxy implements DOMStoreTransactionChain{ +public class TransactionChainProxy implements DOMStoreTransactionChain { + + private static final Logger LOG = LoggerFactory.getLogger(TransactionChainProxy.class); + + private interface State { + boolean isReady(); + + List> getPreviousReadyFutures(); + } + + private static class Allocated implements State { + private final ChainedTransactionProxy transaction; + + Allocated(ChainedTransactionProxy transaction) { + this.transaction = transaction; + } + + @Override + public boolean isReady() { + return transaction.isReady(); + } + + @Override + public List> getPreviousReadyFutures() { + return transaction.getReadyFutures(); + } + } + + private static abstract class AbstractDefaultState implements State { + @Override + public List> getPreviousReadyFutures() { + return Collections.emptyList(); + } + } + + private static final State IDLE_STATE = new AbstractDefaultState() { + @Override + public boolean isReady() { + return true; + } + }; + + private static final State CLOSED_STATE = new AbstractDefaultState() { + @Override + public boolean isReady() { + throw new TransactionChainClosedException("Transaction chain has been closed"); + } + }; + + private static final AtomicInteger counter = new AtomicInteger(0); + private final ActorContext actorContext; private final String transactionChainId; - private volatile List> cohortFutures = Collections.emptyList(); + private volatile State currentState = IDLE_STATE; public TransactionChainProxy(ActorContext actorContext) { this.actorContext = actorContext; - transactionChainId = actorContext.getCurrentMemberName() + "-" + System.currentTimeMillis(); + transactionChainId = actorContext.getCurrentMemberName() + "-txn-chain-" + counter.incrementAndGet(); + } + + public String getTransactionChainId() { + return transactionChainId; } @Override public DOMStoreReadTransaction newReadOnlyTransaction() { - return new TransactionProxy(actorContext, - TransactionProxy.TransactionType.READ_ONLY, this); + State localState = currentState; + checkReadyState(localState); + + return new ChainedTransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY, + transactionChainId, localState.getPreviousReadyFutures()); } @Override public DOMStoreReadWriteTransaction newReadWriteTransaction() { - return new TransactionProxy(actorContext, - TransactionProxy.TransactionType.READ_WRITE, this); + actorContext.acquireTxCreationPermit(); + return allocateWriteTransaction(TransactionProxy.TransactionType.READ_WRITE); } @Override public DOMStoreWriteTransaction newWriteOnlyTransaction() { - return new TransactionProxy(actorContext, - TransactionProxy.TransactionType.WRITE_ONLY, this); + actorContext.acquireTxCreationPermit(); + return allocateWriteTransaction(TransactionProxy.TransactionType.WRITE_ONLY); } @Override public void close() { + currentState = CLOSED_STATE; + // Send a close transaction chain request to each and every shard actorContext.broadcast(new CloseTransactionChain(transactionChainId)); } - public String getTransactionChainId() { - return transactionChainId; + private ChainedTransactionProxy allocateWriteTransaction(TransactionProxy.TransactionType type) { + State localState = currentState; + + checkReadyState(localState); + + // Pass the ready Futures from the previous Tx. + ChainedTransactionProxy txProxy = new ChainedTransactionProxy(actorContext, type, + transactionChainId, localState.getPreviousReadyFutures()); + + currentState = new Allocated(txProxy); + + return txProxy; } - public void onTransactionReady(List> cohortFutures){ - this.cohortFutures = cohortFutures; + private void checkReadyState(State state) { + Preconditions.checkState(state.isReady(), "Previous transaction is not ready yet"); } - public void waitTillCurrentTransactionReady(){ - try { - Await.result(Futures - .sequence(this.cohortFutures, actorContext.getActorSystem().dispatcher()), - actorContext.getOperationDuration()); - } catch (Exception e) { - throw new IllegalStateException("Failed when waiting for transaction on a chain to become ready", e); + private static class ChainedTransactionProxy extends TransactionProxy { + + /** + * Stores the ready Futures from the previous Tx in the chain. + */ + private final List> previousReadyFutures; + + /** + * Stores the ready Futures from this transaction when it is readied. + */ + private volatile List> readyFutures; + + private ChainedTransactionProxy(ActorContext actorContext, TransactionType transactionType, + String transactionChainId, List> previousReadyFutures) { + super(actorContext, transactionType, transactionChainId); + this.previousReadyFutures = previousReadyFutures; + } + + List> getReadyFutures() { + return readyFutures; + } + + boolean isReady() { + return readyFutures != null; + } + + @Override + protected void onTransactionReady(List> readyFutures) { + LOG.debug("onTransactionReady {} pending readyFutures size {} chain {}", getIdentifier(), + readyFutures.size(), getTransactionChainId()); + this.readyFutures = readyFutures; + } + + /** + * This method is overridden to ensure the previous Tx's ready operations complete + * before we initiate the next Tx in the chain to avoid creation failures if the + * previous Tx's ready operations haven't completed yet. + */ + @Override + protected Future sendFindPrimaryShardAsync(final String shardName) { + // Check if there are any previous ready Futures, otherwise let the super class handle it. + if(previousReadyFutures.isEmpty()) { + return super.sendFindPrimaryShardAsync(shardName); + } + + if(LOG.isDebugEnabled()) { + LOG.debug("Waiting for {} previous ready futures for Tx {} on chain {}", + previousReadyFutures.size(), getIdentifier(), getTransactionChainId()); + } + + // Combine the ready Futures into 1. + Future> combinedFutures = akka.dispatch.Futures.sequence( + previousReadyFutures, getActorContext().getClientDispatcher()); + + // Add a callback for completion of the combined Futures. + final Promise returnPromise = akka.dispatch.Futures.promise(); + OnComplete> onComplete = new OnComplete>() { + @Override + public void onComplete(Throwable failure, Iterable notUsed) { + if(failure != null) { + // A Ready Future failed so fail the returned Promise. + returnPromise.failure(failure); + } else { + LOG.debug("Previous Tx readied - sending FindPrimaryShard for {} on chain {}", + getIdentifier(), getTransactionChainId()); + + // Send the FindPrimaryShard message and use the resulting Future to complete the + // returned Promise. + returnPromise.completeWith(ChainedTransactionProxy.super.sendFindPrimaryShardAsync(shardName)); + } + } + }; + + combinedFutures.onComplete(onComplete, getActorContext().getClientDispatcher()); + + return returnPromise.future(); } } }