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%2FTransactionProxy.java;h=5a1cb6740d0229b4e9918f8280f73299eab82727;hp=af4c899e568767ea347639fef8cdf354ce510cd2;hb=17b8a7d3f93e8030c2fe22612f4112f680660fd8;hpb=4d1709660b7af992d4c382a2a38debb5c7d64fb9 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java index af4c899e56..5a1cb6740d 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java @@ -17,14 +17,15 @@ import com.google.common.collect.Iterables; import com.google.common.util.concurrent.CheckedFuture; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.SettableFuture; import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import java.util.TreeMap; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; import org.opendaylight.controller.cluster.datastore.messages.AbstractRead; import org.opendaylight.controller.cluster.datastore.messages.DataExists; @@ -35,10 +36,10 @@ import org.opendaylight.controller.cluster.datastore.modification.MergeModificat import org.opendaylight.controller.cluster.datastore.modification.WriteModification; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregator; -import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; -import org.opendaylight.controller.sal.core.spi.data.AbstractDOMStoreTransaction; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction; -import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture; +import org.opendaylight.mdsal.common.api.MappingCheckedFuture; +import org.opendaylight.mdsal.common.api.ReadFailedException; +import org.opendaylight.mdsal.dom.spi.store.AbstractDOMStoreTransaction; +import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException; @@ -50,15 +51,39 @@ import scala.concurrent.Promise; /** * A transaction potentially spanning multiple backend shards. */ -public class TransactionProxy extends AbstractDOMStoreTransaction implements DOMStoreReadWriteTransaction { - private static enum TransactionState { +public class TransactionProxy extends AbstractDOMStoreTransaction + implements DOMStoreReadWriteTransaction { + private enum TransactionState { OPEN, READY, CLOSED, } + private static final Logger LOG = LoggerFactory.getLogger(TransactionProxy.class); - private final Map txContextWrappers = new HashMap<>(); + // Global lock used for transactions spanning multiple shards - synchronizes sending of the ready messages + // for atomicity to avoid potential deadlock with concurrent transactions spanning the same shards as outlined + // in the following scenario: + // + // - Tx1 sends ready message to shard A + // - Tx2 sends ready message to shard A + // - Tx2 sends ready message to shard B + // - Tx1 sends ready message to shard B + // + // This scenario results in deadlock: after Tx1 canCommits to shard A, it can't proceed with shard B until Tx2 + // completes as Tx2 was readied first on shard B. However Tx2 cannot make progress because it's waiting to canCommit + // on shard A which is blocked by Tx1. + // + // The global lock avoids this as it forces the ready messages to be sent in a predictable order: + // + // - Tx1 sends ready message to shard A + // - Tx1 sends ready message to shard B + // - Tx2 sends ready message to shard A + // - Tx2 sends ready message to shard B + // + private static final Object GLOBAL_TX_READY_LOCK = new Object(); + + private final Map txContextWrappers = new TreeMap<>(); private final AbstractTransactionContextFactory txContextFactory; private final TransactionType type; private TransactionState state = TransactionState.OPEN; @@ -78,19 +103,19 @@ public class TransactionProxy extends AbstractDOMStoreTransaction CheckedFuture executeRead(String shardName, final AbstractRead readCmd) { - Preconditions.checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed"); + private CheckedFuture executeRead(final String shardName, + final AbstractRead readCmd) { + Preconditions.checkState(type != TransactionType.WRITE_ONLY, + "Reads from write-only transactions are not allowed"); - if(LOG.isDebugEnabled()) { - LOG.debug("Tx {} {} {}", getIdentifier(), readCmd.getClass().getSimpleName(), readCmd.getPath()); - } + LOG.trace("Tx {} {} {}", getIdentifier(), readCmd.getClass().getSimpleName(), readCmd.getPath()); final SettableFuture proxyFuture = SettableFuture.create(); TransactionContextWrapper contextWrapper = getContextWrapper(shardName); contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() { @Override - public void invoke(TransactionContext transactionContext) { - transactionContext.executeRead(readCmd, proxyFuture); + public void invoke(final TransactionContext transactionContext, final Boolean havePermit) { + transactionContext.executeRead(readCmd, proxyFuture, havePermit); } }); @@ -99,15 +124,12 @@ public class TransactionProxy extends AbstractDOMStoreTransaction>, ReadFailedException> read(final YangInstanceIdentifier path) { - Preconditions.checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed"); + Preconditions.checkState(type != TransactionType.WRITE_ONLY, + "Reads from write-only transactions are not allowed"); + Preconditions.checkNotNull(path, "path should not be null"); - LOG.debug("Tx {} read {}", getIdentifier(), path); - - if (YangInstanceIdentifier.EMPTY.equals(path)) { - return readAllData(); - } else { - return singleShardRead(shardNameFromIdentifier(path), path); - } + LOG.trace("Tx {} read {}", getIdentifier(), path); + return path.isEmpty() ? readAllData() : singleShardRead(shardNameFromIdentifier(path), path); } private CheckedFuture>, ReadFailedException> singleShardRead( @@ -117,7 +139,8 @@ public class TransactionProxy extends AbstractDOMStoreTransaction>, ReadFailedException> readAllData() { final Set allShardNames = txContextFactory.getActorContext().getConfiguration().getAllShardNames(); - final Collection>, ReadFailedException>> futures = new ArrayList<>(allShardNames.size()); + final Collection>, ReadFailedException>> futures = + new ArrayList<>(allShardNames.size()); for (String shardName : allShardNames) { futures.add(singleShardRead(shardName, YangInstanceIdentifier.EMPTY)); @@ -126,16 +149,16 @@ public class TransactionProxy extends AbstractDOMStoreTransaction>>> listFuture = Futures.allAsList(futures); final ListenableFuture>> aggregateFuture; - aggregateFuture = Futures.transform(listFuture, new Function>>, Optional>>() { - @Override - public Optional> apply(final List>> input) { + aggregateFuture = Futures.transform(listFuture, + (Function>>, Optional>>) input -> { try { - return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.EMPTY, input, txContextFactory.getActorContext().getSchemaContext()); + return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.EMPTY, input, + txContextFactory.getActorContext().getSchemaContext(), + txContextFactory.getActorContext().getDatastoreContext().getLogicalStoreType()); } catch (DataValidationFailedException e) { throw new IllegalArgumentException("Failed to aggregate", e); } - } - }); + }, MoreExecutors.directExecutor()); return MappingCheckedFuture.create(aggregateFuture, ReadFailedException.MAPPER); } @@ -158,16 +181,14 @@ public class TransactionProxy extends AbstractDOMStoreTransaction ret; switch (txContextWrappers.size()) { - case 0: - ret = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE; - break; - case 1: - final Entry e = Iterables.getOnlyElement(txContextWrappers.entrySet()); - ret = createSingleCommitCohort(e.getKey(), e.getValue()); - break; - default: - ret = createMultiCommitCohort(txContextWrappers.entrySet()); + case 0: + ret = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE; + break; + case 1: + final Entry e = Iterables.getOnlyElement( + txContextWrappers.entrySet()); + ret = createSingleCommitCohort(e.getKey(), e.getValue()); + break; + default: + ret = createMultiCommitCohort(txContextWrappers.entrySet()); } txContextFactory.onTransactionReady(getIdentifier(), ret.getCohortFutures()); @@ -253,49 +275,48 @@ public class TransactionProxy extends AbstractDOMStoreTransaction getDirectCommitFuture(TransactionContext transactionContext, - OperationCallback.Reference operationCallbackRef) { + private Future getDirectCommitFuture(final TransactionContext transactionContext, + final OperationCallback.Reference operationCallbackRef, final Boolean havePermit) { TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback( txContextFactory.getActorContext()); operationCallbackRef.set(rateLimitingCallback); rateLimitingCallback.run(); - return transactionContext.directCommit(); + return transactionContext.directCommit(havePermit); } private AbstractThreePhaseCommitCohort createMultiCommitCohort( final Set> txContextWrapperEntries) { final List cohorts = new ArrayList<>(txContextWrapperEntries.size()); - for (Entry e : txContextWrapperEntries) { - LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), e.getKey()); - final TransactionContextWrapper wrapper = e.getValue(); + synchronized (GLOBAL_TX_READY_LOCK) { + for (Entry e : txContextWrapperEntries) { + LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), e.getKey()); - // The remote tx version is obtained the via TransactionContext which may not be available yet so - // we pass a Supplier to dynamically obtain it. Once the ready Future is resolved the - // TransactionContext is available. - Supplier txVersionSupplier = new Supplier() { - @Override - public Short get() { - return wrapper.getTransactionContext().getTransactionVersion(); - } - }; + final TransactionContextWrapper wrapper = e.getValue(); + + // The remote tx version is obtained the via TransactionContext which may not be available yet so + // we pass a Supplier to dynamically obtain it. Once the ready Future is resolved the + // TransactionContext is available. + Supplier txVersionSupplier = () -> wrapper.getTransactionContext().getTransactionVersion(); - cohorts.add(new ThreePhaseCommitCohortProxy.CohortInfo(wrapper.readyTransaction(), txVersionSupplier)); + cohorts.add(new ThreePhaseCommitCohortProxy.CohortInfo(wrapper.readyTransaction(), txVersionSupplier)); + } } return new ThreePhaseCommitCohortProxy(txContextFactory.getActorContext(), cohorts, getIdentifier());