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=405eb0f95bad82b2064240f524269260f93d5a8a;hp=e7a00042e4146c5ddc03a013e6eeda3198a38ea3;hb=95c296a7c1e8e186a88a0a0dc82e080b2185db33;hpb=233db109dea38ed45ca3c48eba04f302e38a9139 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 e7a00042e4..405eb0f95b 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 @@ -12,6 +12,7 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Function; import com.google.common.base.Optional; import com.google.common.base.Preconditions; +import com.google.common.base.Supplier; import com.google.common.collect.Iterables; import com.google.common.util.concurrent.CheckedFuture; import com.google.common.util.concurrent.Futures; @@ -24,10 +25,14 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Set; -import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; -import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier; -import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory; +import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; +import org.opendaylight.controller.cluster.datastore.messages.AbstractRead; +import org.opendaylight.controller.cluster.datastore.messages.DataExists; +import org.opendaylight.controller.cluster.datastore.messages.ReadData; +import org.opendaylight.controller.cluster.datastore.modification.AbstractModification; +import org.opendaylight.controller.cluster.datastore.modification.DeleteModification; +import org.opendaylight.controller.cluster.datastore.modification.MergeModification; +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; @@ -45,20 +50,20 @@ 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 txContextAdapters = new HashMap<>(); + private final Map txContextWrappers = new HashMap<>(); private final AbstractTransactionContextFactory txContextFactory; private final TransactionType type; private TransactionState state = TransactionState.OPEN; - private volatile OperationCompleter operationCompleter; - private volatile Semaphore operationLimiter; @VisibleForTesting public TransactionProxy(final AbstractTransactionContextFactory txContextFactory, final TransactionType type) { @@ -72,18 +77,21 @@ public class TransactionProxy extends AbstractDOMStoreTransaction exists(final YangInstanceIdentifier path) { - Preconditions.checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed"); + return executeRead(shardNameFromIdentifier(path), new DataExists(path, DataStoreVersions.CURRENT_VERSION)); + } - LOG.debug("Tx {} exists {}", getIdentifier(), path); + private CheckedFuture executeRead(String shardName, final AbstractRead readCmd) { + Preconditions.checkState(type != TransactionType.WRITE_ONLY, + "Reads from write-only transactions are not allowed"); - throttleOperation(); + LOG.debug("Tx {} {} {}", getIdentifier(), readCmd.getClass().getSimpleName(), readCmd.getPath()); - final SettableFuture proxyFuture = SettableFuture.create(); - TransactionContextWrapper contextAdapter = getContextAdapter(path); - contextAdapter.maybeExecuteTransactionOperation(new TransactionOperation() { + final SettableFuture proxyFuture = SettableFuture.create(); + TransactionContextWrapper contextWrapper = getContextWrapper(shardName); + contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() { @Override public void invoke(TransactionContext transactionContext) { - transactionContext.dataExists(path, proxyFuture); + transactionContext.executeRead(readCmd, proxyFuture); } }); @@ -92,36 +100,27 @@ 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"); LOG.debug("Tx {} read {}", getIdentifier(), path); if (YangInstanceIdentifier.EMPTY.equals(path)) { return readAllData(); } else { - throttleOperation(); - return singleShardRead(shardNameFromIdentifier(path), path); } } private CheckedFuture>, ReadFailedException> singleShardRead( final String shardName, final YangInstanceIdentifier path) { - final SettableFuture>> proxyFuture = SettableFuture.create(); - TransactionContextWrapper contextAdapter = getContextAdapter(shardName); - contextAdapter.maybeExecuteTransactionOperation(new TransactionOperation() { - @Override - public void invoke(TransactionContext transactionContext) { - transactionContext.readData(path, proxyFuture); - } - }); - - return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER); + return executeRead(shardName, new ReadData(path, DataStoreVersions.CURRENT_VERSION)); } private CheckedFuture>, 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)); @@ -130,67 +129,46 @@ 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); } - } - }); + }); return MappingCheckedFuture.create(aggregateFuture, ReadFailedException.MAPPER); } @Override public void delete(final YangInstanceIdentifier path) { - checkModificationState(); - - LOG.debug("Tx {} delete {}", getIdentifier(), path); - - throttleOperation(); - - TransactionContextWrapper contextAdapter = getContextAdapter(path); - contextAdapter.maybeExecuteTransactionOperation(new TransactionOperation() { - @Override - public void invoke(TransactionContext transactionContext) { - transactionContext.deleteData(path); - } - }); + executeModification(new DeleteModification(path)); } @Override public void merge(final YangInstanceIdentifier path, final NormalizedNode data) { - checkModificationState(); - - LOG.debug("Tx {} merge {}", getIdentifier(), path); - - throttleOperation(); - - TransactionContextWrapper contextAdapter = getContextAdapter(path); - contextAdapter.maybeExecuteTransactionOperation(new TransactionOperation() { - @Override - public void invoke(TransactionContext transactionContext) { - transactionContext.mergeData(path, data); - } - }); + executeModification(new MergeModification(path, data)); } @Override public void write(final YangInstanceIdentifier path, final NormalizedNode data) { - checkModificationState(); + executeModification(new WriteModification(path, data)); + } - LOG.debug("Tx {} write {}", getIdentifier(), path); + private void executeModification(final AbstractModification modification) { + checkModificationState(); - throttleOperation(); + LOG.debug("Tx {} executeModification {} {}", getIdentifier(), modification.getClass().getSimpleName(), + modification.getPath()); - TransactionContextWrapper contextAdapter = getContextAdapter(path); - contextAdapter.maybeExecuteTransactionOperation(new TransactionOperation() { + TransactionContextWrapper contextWrapper = getContextWrapper(modification.getPath()); + contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() { @Override - public void invoke(TransactionContext transactionContext) { - transactionContext.writeData(path, data); + protected void invoke(TransactionContext transactionContext) { + transactionContext.executeModification(modification); } }); } @@ -220,8 +198,8 @@ public class TransactionProxy extends AbstractDOMStoreTransaction ret; - switch (txContextAdapters.size()) { - case 0: - ret = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE; - break; - case 1: - final Entry e = Iterables.getOnlyElement(txContextAdapters.entrySet()); - ret = createSingleCommitCohort(e.getKey(), e.getValue()); - break; - default: - ret = createMultiCommitCohort(txContextAdapters.entrySet()); + 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()); } txContextFactory.onTransactionReady(getIdentifier(), ret.getCohortFutures()); @@ -261,96 +240,80 @@ public class TransactionProxy extends AbstractDOMStoreTransaction createSingleCommitCohort(final String shardName, - final TransactionContextWrapper contextAdapter) { - throttleOperation(); + final TransactionContextWrapper contextWrapper) { LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), shardName); final OperationCallback.Reference operationCallbackRef = new OperationCallback.Reference(OperationCallback.NO_OP_CALLBACK); - final TransactionContext transactionContext = contextAdapter.getTransactionContext(); + final TransactionContext transactionContext = contextWrapper.getTransactionContext(); final Future future; if (transactionContext == null) { final Promise promise = akka.dispatch.Futures.promise(); - contextAdapter.maybeExecuteTransactionOperation(new TransactionOperation() { + contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() { @Override - public void invoke(TransactionContext transactionContext) { - promise.completeWith(getReadyOrDirectCommitFuture(transactionContext, operationCallbackRef)); + public void invoke(TransactionContext newTransactionContext) { + promise.completeWith(getDirectCommitFuture(newTransactionContext, operationCallbackRef)); } }); future = promise.future(); } else { // avoid the creation of a promise and a TransactionOperation - future = getReadyOrDirectCommitFuture(transactionContext, operationCallbackRef); + future = getDirectCommitFuture(transactionContext, operationCallbackRef); } - return new SingleCommitCohortProxy(txContextFactory.getActorContext(), future, getIdentifier().toString(), - operationCallbackRef); + return new SingleCommitCohortProxy(txContextFactory.getActorContext(), future, getIdentifier(), + operationCallbackRef); } - private Future getReadyOrDirectCommitFuture(TransactionContext transactionContext, + private Future getDirectCommitFuture(TransactionContext transactionContext, OperationCallback.Reference operationCallbackRef) { - if (transactionContext.supportsDirectCommit()) { - TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback( - txContextFactory.getActorContext()); - operationCallbackRef.set(rateLimitingCallback); - rateLimitingCallback.run(); - return transactionContext.directCommit(); - } else { - return transactionContext.readyTransaction(); - } + TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback( + txContextFactory.getActorContext()); + operationCallbackRef.set(rateLimitingCallback); + rateLimitingCallback.run(); + return transactionContext.directCommit(); } private AbstractThreePhaseCommitCohort createMultiCommitCohort( - final Set> txContextAdapterEntries) { + final Set> txContextWrapperEntries) { - throttleOperation(); - final List> cohortFutures = new ArrayList<>(txContextAdapterEntries.size()); - for (Entry e : txContextAdapterEntries) { + final List cohorts = new ArrayList<>(txContextWrapperEntries.size()); + for (Entry e : txContextWrapperEntries) { LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), e.getKey()); - TransactionContextWrapper contextAdapter = e.getValue(); - final TransactionContext transactionContext = contextAdapter.getTransactionContext(); - Future future; - if (transactionContext != null) { - // avoid the creation of a promise and a TransactionOperation - future = transactionContext.readyTransaction(); - } else { - final Promise promise = akka.dispatch.Futures.promise(); - contextAdapter.maybeExecuteTransactionOperation(new TransactionOperation() { - @Override - public void invoke(TransactionContext transactionContext) { - promise.completeWith(transactionContext.readyTransaction()); - } - }); - - future = promise.future(); - } + 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(); - cohortFutures.add(future); + cohorts.add(new ThreePhaseCommitCohortProxy.CohortInfo(wrapper.readyTransaction(), txVersionSupplier)); } - return new ThreePhaseCommitCohortProxy(txContextFactory.getActorContext(), cohortFutures, getIdentifier().toString()); + return new ThreePhaseCommitCohortProxy(txContextFactory.getActorContext(), cohorts, getIdentifier()); } - private static String shardNameFromIdentifier(final YangInstanceIdentifier path) { - return ShardStrategyFactory.getStrategy(path).findShard(path); + private String shardNameFromIdentifier(final YangInstanceIdentifier path) { + return txContextFactory.getActorContext().getShardStrategyFactory().getStrategy(path).findShard(path); } - private TransactionContextWrapper getContextAdapter(final YangInstanceIdentifier path) { - return getContextAdapter(shardNameFromIdentifier(path)); + private TransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) { + return getContextWrapper(shardNameFromIdentifier(path)); } - private TransactionContextWrapper getContextAdapter(final String shardName) { - final TransactionContextWrapper existing = txContextAdapters.get(shardName); + private TransactionContextWrapper getContextWrapper(final String shardName) { + final TransactionContextWrapper existing = txContextWrappers.get(shardName); if (existing != null) { return existing; } - final TransactionContextWrapper fresh = txContextFactory.newTransactionAdapter(this, shardName); - txContextAdapters.put(shardName, fresh); + final TransactionContextWrapper fresh = txContextFactory.newTransactionContextWrapper(this, shardName); + txContextWrappers.put(shardName, fresh); return fresh; } @@ -365,44 +328,4 @@ public class TransactionProxy extends AbstractDOMStoreTransaction