X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionChainProxy.java;h=2e671e3ce2f267807837225c95376847c6116eb8;hp=b467ee4ddbf56c456c2e9f0f62381eefa173e31d;hb=0c6b6b36b116a3b01e8c110252fbaa4035500bb6;hpb=b3c034675957f963c5878ce1e5e183ec2de8b5e2 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..2e671e3ce2 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.AbstractMap.SimpleEntry; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; 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(); + + SimpleEntry>> getReadyFutures(); + + void setReadyFutures(Object txIdentifier, List> readyFutures); + } + + private static class Allocated implements State { + private volatile SimpleEntry>> readyFutures; + + @Override + public boolean isReady() { + return readyFutures != null; + } + + @Override + public SimpleEntry>> getReadyFutures() { + return readyFutures != null ? readyFutures : EMPTY_READY_FUTURES; + } + + @Override + public void setReadyFutures(Object txIdentifier, List> readyFutures) { + this.readyFutures = new SimpleEntry<>(txIdentifier, readyFutures); + } + } + + private static abstract class AbstractDefaultState implements State { + @Override + public SimpleEntry>> getReadyFutures() { + return EMPTY_READY_FUTURES; + } + + @Override + public void setReadyFutures(Object txIdentifier, List> readyFutures) { + throw new IllegalStateException("No transaction is allocated"); + } + } + + 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 SimpleEntry>> EMPTY_READY_FUTURES = + new SimpleEntry>>("", + Collections.>emptyList()); + + private static final AtomicReferenceFieldUpdater STATE_UPDATER = + AtomicReferenceFieldUpdater.newUpdater(TransactionChainProxy.class, State.class, "state"); + private final ActorContext actorContext; private final String transactionChainId; - private volatile List> cohortFutures = Collections.emptyList(); + private volatile State state = IDLE_STATE; + private static final AtomicInteger counter = new AtomicInteger(0); public TransactionChainProxy(ActorContext actorContext) { this.actorContext = actorContext; - transactionChainId = actorContext.getCurrentMemberName() + "-" + System.currentTimeMillis(); + transactionChainId = actorContext.getCurrentMemberName() + "-transaction-chain-" + counter.incrementAndGet(); + } + + public String getTransactionChainId() { + return transactionChainId; } @Override public DOMStoreReadTransaction newReadOnlyTransaction() { - return new TransactionProxy(actorContext, - TransactionProxy.TransactionType.READ_ONLY, this); + checkReadyState(); + return new ChainedTransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY); } @Override public DOMStoreReadWriteTransaction newReadWriteTransaction() { - return new TransactionProxy(actorContext, - TransactionProxy.TransactionType.READ_WRITE, this); + return allocateWriteTransaction(TransactionProxy.TransactionType.READ_WRITE); } @Override public DOMStoreWriteTransaction newWriteOnlyTransaction() { - return new TransactionProxy(actorContext, - TransactionProxy.TransactionType.WRITE_ONLY, this); + return allocateWriteTransaction(TransactionProxy.TransactionType.WRITE_ONLY); } @Override public void close() { + state = 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) { + checkReadyState(); + + ChainedTransactionProxy txProxy = new ChainedTransactionProxy(actorContext, type); + STATE_UPDATER.compareAndSet(this, IDLE_STATE, new Allocated()); + + return txProxy; } - public void onTransactionReady(List> cohortFutures){ - this.cohortFutures = cohortFutures; + private void checkReadyState() { + Preconditions.checkState(state.isReady(), "Previous transaction %s is not ready yet", + state.getReadyFutures().getKey()); } - 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 class ChainedTransactionProxy extends TransactionProxy { + + ChainedTransactionProxy(ActorContext actorContext, TransactionType transactionType) { + super(actorContext, transactionType, transactionChainId); + } + + @Override + protected void onTransactionReady(List> readyFutures) { + LOG.debug("onTransactionReady {} pending readyFutures size {} chain {}", getIdentifier(), readyFutures.size(), TransactionChainProxy.this.transactionChainId); + state.setReadyFutures(getIdentifier(), readyFutures); + } + + /** + * This method is overridden to ensure the previous Tx's ready operations complete + * before we create the next shard Tx in the chain to avoid creation failures if the + * previous Tx's ready operations haven't completed yet. + */ + @Override + protected Future sendCreateTransaction(final ActorSelection shard, + final Object serializedCreateMessage) { + + // Check if there are any previous ready Futures, otherwise let the super class handle it. + // The second check is done to ensure the the previous ready Futures aren't for this + // Tx instance as deadlock would occur if we tried to wait on our own Futures. This can + // occur in this scenario: + // + // - the TransactionProxy is created and the client does a write. + // + // - the TransactionProxy then attempts to create the shard Tx. However it first + // sends a FindPrimaryShard message to the shard manager to find the local shard + // This call is done async. + // + // - the client submits the Tx and the TransactionProxy is readied and we cache + // the ready Futures here. + // + // - then the FindPrimaryShard call completes and this method is called to create + // the shard Tx. However the cached Futures were from the ready on this Tx. If we + // tried to wait on them, it would cause a form of deadlock as the ready Future + // would be waiting on the Tx create Future and vice versa. + SimpleEntry>> readyFuturesEntry = state.getReadyFutures(); + List> readyFutures = readyFuturesEntry.getValue(); + if(readyFutures.isEmpty() || getIdentifier().equals(readyFuturesEntry.getKey())) { + return super.sendCreateTransaction(shard, serializedCreateMessage); + } + + // Combine the ready Futures into 1. + Future> combinedFutures = akka.dispatch.Futures.sequence( + readyFutures, actorContext.getActorSystem().dispatcher()); + + // Add a callback for completion of the combined Futures. + final Promise createTxPromise = 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. + createTxPromise.failure(failure); + } else { + // Send the CreateTx message and use the resulting Future to complete the + // returned Promise. + createTxPromise.completeWith(actorContext.executeOperationAsync(shard, + serializedCreateMessage)); + } + } + }; + + combinedFutures.onComplete(onComplete, actorContext.getActorSystem().dispatcher()); + + return createTxPromise.future(); } } }