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=98e115efc6270342e4f54ce99f5dcf0e95d99bc8;hp=b774fc47876dcf978adacb15cf4a6e9a6e0be4da;hb=5b69c8e66b12a29a36457955cac4a45affd7c73f;hpb=f7da1de9c83827bbb9c13632e5715c2e6720ae21 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 b774fc4787..98e115efc6 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,39 +7,38 @@ */ package org.opendaylight.controller.cluster.datastore; +import static com.google.common.base.Preconditions.checkState; +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.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.FluentFuture; 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.Optional; import java.util.Set; +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.ActorContext; +import org.opendaylight.controller.cluster.datastore.utils.ActorUtils; 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.mdsal.common.api.MappingCheckedFuture; +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; @@ -61,30 +60,28 @@ public class TransactionProxy extends AbstractDOMStoreTransaction txContextWrappers = new HashMap<>(); + private final Map txContextWrappers = new TreeMap<>(); private final AbstractTransactionContextFactory txContextFactory; private final TransactionType type; private TransactionState state = TransactionState.OPEN; @VisibleForTesting public TransactionProxy(final AbstractTransactionContextFactory txContextFactory, final TransactionType type) { - super(txContextFactory.nextIdentifier(), txContextFactory.getActorContext().getDatastoreContext() + super(txContextFactory.nextIdentifier(), txContextFactory.getActorUtils().getDatastoreContext() .isTransactionDebugContextEnabled()); this.txContextFactory = txContextFactory; - this.type = Preconditions.checkNotNull(type); + this.type = requireNonNull(type); LOG.debug("New {} Tx - {}", type, getIdentifier()); } @Override - public CheckedFuture exists(final YangInstanceIdentifier path) { + public FluentFuture exists(final YangInstanceIdentifier path) { return executeRead(shardNameFromIdentifier(path), new DataExists(path, DataStoreVersions.CURRENT_VERSION)); } - private CheckedFuture executeRead(final String shardName, - final AbstractRead readCmd) { - Preconditions.checkState(type != TransactionType.WRITE_ONLY, - "Reads from write-only transactions are not allowed"); + private FluentFuture executeRead(final String shardName, final AbstractRead readCmd) { + checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed"); LOG.trace("Tx {} {} {}", getIdentifier(), readCmd.getClass().getSimpleName(), readCmd.getPath()); @@ -92,109 +89,95 @@ public class TransactionProxy extends AbstractDOMStoreTransaction>, ReadFailedException> read(final YangInstanceIdentifier path) { - Preconditions.checkState(type != TransactionType.WRITE_ONLY, - "Reads from write-only transactions are not allowed"); + 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); - - if (YangInstanceIdentifier.EMPTY.equals(path)) { - return readAllData(); - } else { - return singleShardRead(shardNameFromIdentifier(path), path); - } + return path.isEmpty() ? readAllData() : singleShardRead(shardNameFromIdentifier(path), path); } - private CheckedFuture>, ReadFailedException> singleShardRead( + private FluentFuture>> singleShardRead( final String shardName, final YangInstanceIdentifier path) { 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()); + private FluentFuture>> readAllData() { + final Set allShardNames = txContextFactory.getActorUtils().getConfiguration().getAllShardNames(); + 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.getActorContext().getSchemaContext(), - txContextFactory.getActorContext().getDatastoreContext().getLogicalStoreType()); - } catch (DataValidationFailedException e) { - throw new IllegalArgumentException("Failed to aggregate", e); - } - }, MoreExecutors.directExecutor()); + 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 MappingCheckedFuture.create(aggregateFuture, ReadFailedException.MAPPER); + return FluentFuture.from(aggregateFuture); } @Override public void delete(final YangInstanceIdentifier path) { - executeModification(new DeleteModification(path)); + checkModificationState("delete", path); + + executeModification(new DeleteOperation(path)); } @Override public void merge(final YangInstanceIdentifier path, final NormalizedNode data) { - executeModification(new MergeModification(path, data)); + checkModificationState("merge", path); + + executeModification(new MergeOperation(path, data)); } @Override public void write(final YangInstanceIdentifier path, final NormalizedNode data) { - executeModification(new WriteModification(path, data)); - } + checkModificationState("write", path); - private void executeModification(final AbstractModification modification) { - checkModificationState(); - - LOG.trace("Tx {} executeModification {} {}", getIdentifier(), modification.getClass().getSimpleName(), - modification.getPath()); + executeModification(new WriteOperation(path, data)); + } - TransactionContextWrapper contextWrapper = getContextWrapper(modification.getPath()); - contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() { - @Override - protected void invoke(final TransactionContext transactionContext) { - transactionContext.executeModification(modification); - } - }); + private void executeModification(final TransactionModificationOperation operation) { + getContextWrapper(operation.path()).maybeExecuteTransactionOperation(operation); } - private void checkModificationState() { - Preconditions.checkState(type != TransactionType.READ_ONLY, - "Modification operation on read-only transaction is not allowed"); - Preconditions.checkState(state == TransactionState.OPEN, - "Transaction is sealed - further modifications are not allowed"); + 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 public final void close() { if (!seal(TransactionState.CLOSED)) { - Preconditions.checkState(state == TransactionState.CLOSED, "Transaction %s is ready, it cannot be closed", + checkState(state == TransactionState.CLOSED, "Transaction %s is ready, it cannot be closed", getIdentifier()); // Idempotent no-op as per AutoCloseable recommendation return; @@ -203,7 +186,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction ready() { - Preconditions.checkState(type != TransactionType.READ_ONLY, "Read-only transactions cannot be readied"); + checkState(type != TransactionType.READ_ONLY, "Read-only transactions cannot be readied"); final boolean success = seal(TransactionState.READY); - Preconditions.checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state); + checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state); LOG.debug("Tx {} Readying {} components for commit", getIdentifier(), txContextWrappers.size()); @@ -233,7 +216,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction getDirectCommitFuture(final TransactionContext transactionContext, - final OperationCallback.Reference operationCallbackRef) { + final OperationCallback.Reference operationCallbackRef, final Boolean havePermit) { TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback( - txContextFactory.getActorContext()); + txContextFactory.getActorUtils()); operationCallbackRef.set(rateLimitingCallback); rateLimitingCallback.run(); - return transactionContext.directCommit(); + return transactionContext.directCommit(havePermit); } - private AbstractThreePhaseCommitCohort createMultiCommitCohort( - final Set> txContextWrapperEntries) { + private AbstractThreePhaseCommitCohort createMultiCommitCohort() { - final List cohorts = new ArrayList<>(txContextWrapperEntries.size()); - for (Entry e : txContextWrapperEntries) { + final List cohorts = new ArrayList<>(txContextWrappers.size()); + final Optional> shardNames = Optional.of(new TreeSet<>(txContextWrappers.keySet())); + for (Entry e : txContextWrappers.entrySet()) { LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), e.getKey()); final TransactionContextWrapper wrapper = e.getValue(); @@ -292,16 +276,15 @@ public class TransactionProxy extends AbstractDOMStoreTransaction txVersionSupplier = () -> wrapper.getTransactionContext().getTransactionVersion(); - - cohorts.add(new ThreePhaseCommitCohortProxy.CohortInfo(wrapper.readyTransaction(), txVersionSupplier)); + cohorts.add(new ThreePhaseCommitCohortProxy.CohortInfo(wrapper.readyTransaction(shardNames), + () -> wrapper.getTransactionContext().getTransactionVersion())); } - return new ThreePhaseCommitCohortProxy(txContextFactory.getActorContext(), cohorts, getIdentifier()); + return new ThreePhaseCommitCohortProxy(txContextFactory.getActorUtils(), cohorts, getIdentifier()); } private String shardNameFromIdentifier(final YangInstanceIdentifier path) { - return txContextFactory.getActorContext().getShardStrategyFactory().getStrategy(path).findShard(path); + return txContextFactory.getActorUtils().getShardStrategyFactory().getStrategy(path).findShard(path); } private TransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) { @@ -327,7 +310,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction