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=2e671e3ce2f267807837225c95376847c6116eb8;hb=b131db5779e46e9555aa3358c5b6aa13109ef8f5;hp=93f9e6b7de1e2085f01d36c74a37b84ff7ecb4d2;hpb=0d318e4be80c368269172e915b2a193b6f0a8f09;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 93f9e6b7de..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 @@ -10,53 +10,141 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorSelection; 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 org.slf4j.Logger; +import org.slf4j.LoggerFactory; import scala.concurrent.Future; 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 SimpleEntry>> previousTxReadyFutures; + 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() { + checkReadyState(); return new ChainedTransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY); } @Override public DOMStoreReadWriteTransaction newReadWriteTransaction() { - return new ChainedTransactionProxy(actorContext, TransactionProxy.TransactionType.READ_WRITE); + return allocateWriteTransaction(TransactionProxy.TransactionType.READ_WRITE); } @Override public DOMStoreWriteTransaction newWriteOnlyTransaction() { - return new ChainedTransactionProxy(actorContext, TransactionProxy.TransactionType.WRITE_ONLY); + 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; + } + + private void checkReadyState() { + Preconditions.checkState(state.isReady(), "Previous transaction %s is not ready yet", + state.getReadyFutures().getKey()); } private class ChainedTransactionProxy extends TransactionProxy { @@ -66,12 +154,9 @@ public class TransactionChainProxy implements DOMStoreTransactionChain{ } @Override - protected void onTransactionReady(List> cohortFutures) { - if(!cohortFutures.isEmpty()) { - previousTxReadyFutures = new SimpleEntry<>(getIdentifier(), cohortFutures); - } else { - previousTxReadyFutures = null; - } + protected void onTransactionReady(List> readyFutures) { + LOG.debug("onTransactionReady {} pending readyFutures size {} chain {}", getIdentifier(), readyFutures.size(), TransactionChainProxy.this.transactionChainId); + state.setReadyFutures(getIdentifier(), readyFutures); } /** @@ -82,18 +167,34 @@ public class TransactionChainProxy implements DOMStoreTransactionChain{ @Override protected Future sendCreateTransaction(final ActorSelection shard, final Object serializedCreateMessage) { - // Check if there are any previous ready Futures. Also make sure the previous ready - // Futures aren't for this Tx as deadlock would occur if tried to wait on our own - // Futures. This may happen b/c the shard Tx creates are done async so it's possible - // for the client to ready this Tx before we've even attempted to create a shard Tx. - if(previousTxReadyFutures == null || - previousTxReadyFutures.getKey().equals(getIdentifier())) { + + // 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( - previousTxReadyFutures.getValue(), actorContext.getActorSystem().dispatcher()); + readyFutures, actorContext.getActorSystem().dispatcher()); // Add a callback for completion of the combined Futures. final Promise createTxPromise = akka.dispatch.Futures.promise();