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=da9789693cdca41e1be07f428775a2c74403bb29;hp=155bb1fd3e303cfe8aa929fb44322bad90e2e949;hb=1d5ca4009be6c61d7b61989799037ad8f1ab7a75;hpb=3859df9beca8f13f1ff2b2744ed3470a1715bec3 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 155bb1fd3e..da9789693c 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 @@ -7,13 +7,13 @@ */ package org.opendaylight.controller.cluster.datastore; +import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkState; +import static com.google.common.base.Verify.verifyNotNull; import static java.util.Objects.requireNonNull; import akka.actor.ActorSelection; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Function; -import com.google.common.base.Supplier; import com.google.common.collect.Iterables; import com.google.common.util.concurrent.FluentFuture; import com.google.common.util.concurrent.Futures; @@ -22,6 +22,7 @@ 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; @@ -31,20 +32,24 @@ import java.util.SortedSet; import java.util.TreeMap; import java.util.TreeSet; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; +import org.opendaylight.controller.cluster.datastore.TransactionModificationOperation.DeleteOperation; +import org.opendaylight.controller.cluster.datastore.TransactionModificationOperation.MergeOperation; +import org.opendaylight.controller.cluster.datastore.TransactionModificationOperation.WriteOperation; 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.ActorUtils; import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregator; 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.YangInstanceIdentifier.NodeIdentifier; +import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; +import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException; +import org.opendaylight.yangtools.yang.data.api.schema.builder.DataContainerNodeBuilder; +import org.opendaylight.yangtools.yang.data.impl.schema.Builders; +import org.opendaylight.yangtools.yang.data.tree.api.DataValidationFailedException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.Future; @@ -62,8 +67,9 @@ public class TransactionProxy extends AbstractDOMStoreTransaction txContextWrappers = new TreeMap<>(); + private final Map txContextWrappers = new TreeMap<>(); private final AbstractTransactionContextFactory txContextFactory; private final TransactionType type; private TransactionState state = TransactionState.OPEN; @@ -89,7 +95,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction proxyFuture = SettableFuture.create(); - TransactionContextWrapper contextWrapper = getContextWrapper(shardName); + AbstractTransactionContextWrapper contextWrapper = getContextWrapper(shardName); contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() { @Override public void invoke(final TransactionContext transactionContext, final Boolean havePermit) { @@ -101,86 +107,141 @@ public class TransactionProxy extends AbstractDOMStoreTransaction>> read(final YangInstanceIdentifier path) { + public FluentFuture> read(final YangInstanceIdentifier path) { checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed"); requireNonNull(path, "path should not be null"); LOG.trace("Tx {} read {}", getIdentifier(), path); - return path.isEmpty() ? readAllData() : singleShardRead(shardNameFromIdentifier(path), path); + return path.isEmpty() ? readAllData() : singleShardRead(shardNameFromIdentifier(path), path); } - private FluentFuture>> singleShardRead( - final String shardName, final YangInstanceIdentifier path) { + private FluentFuture> singleShardRead(final String shardName, + final YangInstanceIdentifier path) { return executeRead(shardName, new ReadData(path, DataStoreVersions.CURRENT_VERSION)); } - private FluentFuture>> readAllData() { + private FluentFuture> readAllData() { final Set allShardNames = txContextFactory.getActorUtils().getConfiguration().getAllShardNames(); - final Collection>>> futures = new ArrayList<>(allShardNames.size()); + final Collection>> futures = new ArrayList<>(allShardNames.size()); for (String shardName : allShardNames) { - futures.add(singleShardRead(shardName, YangInstanceIdentifier.EMPTY)); + futures.add(singleShardRead(shardName, YangInstanceIdentifier.empty())); } - final ListenableFuture>>> listFuture = Futures.allAsList(futures); - final ListenableFuture>> aggregateFuture; - - aggregateFuture = Futures.transform(listFuture, - (Function>>, Optional>>) input -> { - try { - return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.EMPTY, input, - txContextFactory.getActorUtils().getSchemaContext(), - txContextFactory.getActorUtils().getDatastoreContext().getLogicalStoreType()); - } catch (DataValidationFailedException e) { - throw new IllegalArgumentException("Failed to aggregate", e); - } - }, MoreExecutors.directExecutor()); + final ListenableFuture>> listFuture = Futures.allAsList(futures); + final ListenableFuture> aggregateFuture; + + aggregateFuture = Futures.transform(listFuture, input -> { + try { + return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.empty(), input, + txContextFactory.getActorUtils().getSchemaContext(), + txContextFactory.getActorUtils().getDatastoreContext().getLogicalStoreType()); + } catch (DataValidationFailedException e) { + throw new IllegalArgumentException("Failed to aggregate", e); + } + }, MoreExecutors.directExecutor()); return FluentFuture.from(aggregateFuture); } @Override public void delete(final YangInstanceIdentifier path) { - executeModification(new DeleteModification(path)); + checkModificationState("delete", path); + + if (path.isEmpty()) { + deleteAllData(); + } else { + executeModification(new DeleteOperation(path)); + } + } + + private void deleteAllData() { + for (String shardName : getActorUtils().getConfiguration().getAllShardNames()) { + getContextWrapper(shardName).maybeExecuteTransactionOperation(ROOT_DELETE_OPERATION); + } } @Override - public void merge(final YangInstanceIdentifier path, final NormalizedNode data) { - executeModification(new MergeModification(path, data)); + public void merge(final YangInstanceIdentifier path, final NormalizedNode data) { + checkModificationState("merge", path); + + if (path.isEmpty()) { + mergeAllData(checkRootData(data)); + } else { + executeModification(new MergeOperation(path, data)); + } + } + + private void mergeAllData(final ContainerNode rootData) { + // Populate requests for individual shards that are being touched + final Map> rootBuilders = new HashMap<>(); + for (DataContainerChild child : rootData.body()) { + final String shardName = shardNameFromRootChild(child); + rootBuilders.computeIfAbsent(shardName, + unused -> Builders.containerBuilder().withNodeIdentifier(rootData.getIdentifier())) + .addChild(child); + } + + // Now dispatch all merges + for (Entry> entry : rootBuilders.entrySet()) { + getContextWrapper(entry.getKey()).maybeExecuteTransactionOperation(new MergeOperation( + YangInstanceIdentifier.empty(), entry.getValue().build())); + } } @Override - public void write(final YangInstanceIdentifier path, final NormalizedNode data) { - executeModification(new WriteModification(path, data)); + public void write(final YangInstanceIdentifier path, final NormalizedNode data) { + checkModificationState("write", path); + + if (path.isEmpty()) { + writeAllData(checkRootData(data)); + } else { + executeModification(new WriteOperation(path, data)); + } } - private void executeModification(final AbstractModification modification) { - checkModificationState(); + private void writeAllData(final ContainerNode rootData) { + // Open builders for all shards + final Map> rootBuilders = new HashMap<>(); + for (String shardName : getActorUtils().getConfiguration().getAllShardNames()) { + rootBuilders.put(shardName, Builders.containerBuilder().withNodeIdentifier(rootData.getIdentifier())); + } - LOG.trace("Tx {} executeModification {} {}", getIdentifier(), modification.getClass().getSimpleName(), - modification.getPath()); + // Now distribute children as needed + for (DataContainerChild child : rootData.body()) { + final String shardName = shardNameFromRootChild(child); + verifyNotNull(rootBuilders.get(shardName), "Failed to find builder for %s", shardName).addChild(child); + } - TransactionContextWrapper contextWrapper = getContextWrapper(modification.getPath()); - contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() { - @Override - protected void invoke(final TransactionContext transactionContext, final Boolean havePermit) { - transactionContext.executeModification(modification, havePermit); - } - }); + // Now dispatch all writes + for (Entry> entry : rootBuilders.entrySet()) { + getContextWrapper(entry.getKey()).maybeExecuteTransactionOperation(new WriteOperation( + YangInstanceIdentifier.empty(), entry.getValue().build())); + } } - private void checkModificationState() { + private void executeModification(final TransactionModificationOperation operation) { + getContextWrapper(operation.path()).maybeExecuteTransactionOperation(operation); + } + + private static ContainerNode checkRootData(final NormalizedNode data) { + // Root has to be a container + checkArgument(data instanceof ContainerNode, "Invalid root data %s", data); + return (ContainerNode) data; + } + + private void checkModificationState(final String opName, final YangInstanceIdentifier path) { checkState(type != TransactionType.READ_ONLY, "Modification operation on read-only transaction is not allowed"); checkState(state == TransactionState.OPEN, "Transaction is sealed - further modifications are not allowed"); + LOG.trace("Tx {} {} {}", getIdentifier(), opName, path); } private boolean seal(final TransactionState newState) { if (state == TransactionState.OPEN) { state = newState; return true; - } else { - return false; } + return false; } @Override @@ -192,7 +253,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction e = Iterables.getOnlyElement( + final Entry e = Iterables.getOnlyElement( txContextWrappers.entrySet()); ret = createSingleCommitCohort(e.getKey(), e.getValue()); break; @@ -236,7 +297,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction createSingleCommitCohort(final String shardName, - final TransactionContextWrapper contextWrapper) { + final AbstractTransactionContextWrapper contextWrapper) { LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), shardName); @@ -277,38 +338,40 @@ public class TransactionProxy extends AbstractDOMStoreTransaction cohorts = new ArrayList<>(txContextWrappers.size()); final Optional> shardNames = Optional.of(new TreeSet<>(txContextWrappers.keySet())); - for (Entry e : txContextWrappers.entrySet()) { + for (Entry e : txContextWrappers.entrySet()) { LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), e.getKey()); - final TransactionContextWrapper wrapper = e.getValue(); + final AbstractTransactionContextWrapper 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(shardNames), - txVersionSupplier)); + () -> wrapper.getTransactionContext().getTransactionVersion())); } return new ThreePhaseCommitCohortProxy(txContextFactory.getActorUtils(), cohorts, getIdentifier()); } + private String shardNameFromRootChild(final DataContainerChild child) { + return shardNameFromIdentifier(YangInstanceIdentifier.create(child.getIdentifier())); + } + private String shardNameFromIdentifier(final YangInstanceIdentifier path) { - return txContextFactory.getActorUtils().getShardStrategyFactory().getStrategy(path).findShard(path); + return getActorUtils().getShardStrategyFactory().getStrategy(path).findShard(path); } - private TransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) { + private AbstractTransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) { return getContextWrapper(shardNameFromIdentifier(path)); } - private TransactionContextWrapper getContextWrapper(final String shardName) { - final TransactionContextWrapper existing = txContextWrappers.get(shardName); + private AbstractTransactionContextWrapper getContextWrapper(final String shardName) { + final AbstractTransactionContextWrapper existing = txContextWrappers.get(shardName); if (existing != null) { return existing; } - final TransactionContextWrapper fresh = txContextFactory.newTransactionContextWrapper(this, shardName); + final AbstractTransactionContextWrapper fresh = txContextFactory.newTransactionContextWrapper(this, shardName); txContextWrappers.put(shardName, fresh); return fresh; } @@ -321,7 +384,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction