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%2Fdatabroker%2Factors%2Fdds%2FClientTransaction.java;h=b2ff5d5184d58753fa76bf90df457abfe6e685f0;hb=f40e99a51cd3fc2c9be3ac8aa0772bdb6b6ce479;hp=10d64ed9bb0a978f055e943544a73a032e1085b9;hpb=95208fa5d24f3d7c2362ee619c9a6a294a69f7cd;p=controller.git

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientTransaction.java
index 10d64ed9bb..b2ff5d5184 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientTransaction.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientTransaction.java
@@ -7,152 +7,132 @@
  */
 package org.opendaylight.controller.cluster.databroker.actors.dds;
 
+import static com.google.common.base.Preconditions.checkState;
+
 import com.google.common.annotations.Beta;
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Iterables;
-import com.google.common.util.concurrent.CheckedFuture;
-import java.util.HashMap;
+import com.google.common.util.concurrent.FluentFuture;
+import java.util.Collection;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.Optional;
+import org.eclipse.jdt.annotation.NonNull;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
-import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
-import org.opendaylight.yangtools.concepts.Identifiable;
+import org.opendaylight.controller.cluster.datastore.utils.RootScatterGather;
+import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
- * Client-side view of a free-standing transaction.
+ * Client-side view of a transaction.
  *
+ * <p>
  * This interface is used by the world outside of the actor system and in the actor system it is manifested via
  * its client actor. That requires some state transfer with {@link DistributedDataStoreClientBehavior}. In order to
  * reduce request latency, all messages are carbon-copied (and enqueued first) to the client actor.
  *
+ * <p>
  * It is internally composed of multiple {@link RemoteProxyTransaction}s, each responsible for a component shard.
  *
+ * <p>
  * Implementation is quite a bit complex, and involves cooperation with {@link AbstractClientHistory} for tracking
  * gaps in transaction identifiers seen by backends.
  *
+ * <p>
  * These gaps need to be accounted for in the transaction setup message sent to a particular backend, so it can verify
  * that the requested transaction is in-sequence. This is critical in ensuring that transactions (which are independent
  * entities from message queueing perspective) do not get reodered -- thus allowing multiple in-flight transactions.
  *
+ * <p>
  * Alternative would be to force visibility by sending an abort request to all potential backends, but that would mean
  * that even empty transactions increase load on all shards -- which would be a scalability issue.
  *
+ * <p>
  * Yet another alternative would be to introduce inter-transaction dependencies to the queueing layer in client actor,
  * but that would require additional indirection and complexity.
  *
  * @author Robert Varga
  */
 @Beta
-public final class ClientTransaction extends LocalAbortable implements Identifiable<TransactionIdentifier> {
-    private static final Logger LOG = LoggerFactory.getLogger(ClientTransaction.class);
-    private static final AtomicIntegerFieldUpdater<ClientTransaction> STATE_UPDATER =
-            AtomicIntegerFieldUpdater.newUpdater(ClientTransaction.class, "state");
-    private static final int OPEN_STATE = 0;
-    private static final int CLOSED_STATE = 1;
-
-    private final Map<Long, AbstractProxyTransaction> proxies = new HashMap<>();
-    private final TransactionIdentifier transactionId;
-    private final AbstractClientHistory parent;
-
-    private volatile int state = OPEN_STATE;
-
-    ClientTransaction(final DistributedDataStoreClientBehavior client, final AbstractClientHistory parent,
-        final TransactionIdentifier transactionId) {
-        this.transactionId = Preconditions.checkNotNull(transactionId);
-        this.parent = Preconditions.checkNotNull(parent);
-    }
-
-    private void checkNotClosed() {
-        Preconditions.checkState(state == OPEN_STATE, "Transaction %s is closed", transactionId);
+public class ClientTransaction extends AbstractClientHandle<AbstractProxyTransaction> {
+    ClientTransaction(final AbstractClientHistory parent, final TransactionIdentifier transactionId) {
+        super(parent, transactionId);
     }
 
-    private AbstractProxyTransaction createProxy(final Long shard) {
-        return parent.createTransactionProxy(transactionId, shard);
-    }
-
-    private AbstractProxyTransaction ensureProxy(final YangInstanceIdentifier path) {
-        checkNotClosed();
-
-        final ModuleShardBackendResolver resolver = parent.getClient().resolver();
-        final Long shard = resolver.resolveShardForPath(path);
-        return proxies.computeIfAbsent(shard, this::createProxy);
-    }
-
-    @Override
-    public TransactionIdentifier getIdentifier() {
-        return transactionId;
+    public FluentFuture<Boolean> exists(final YangInstanceIdentifier path) {
+        return ensureProxy(path).exists(path);
     }
 
-    public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
-        return ensureProxy(path).exists(path);
+    public FluentFuture<Optional<NormalizedNode>> read(final YangInstanceIdentifier path) {
+        return path.isEmpty() ? readRoot() : ensureProxy(path).read(path);
     }
 
-    public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
-        return ensureProxy(path).read(path);
+    private FluentFuture<Optional<NormalizedNode>> readRoot() {
+        return RootScatterGather.gather(parent().actorUtils(), ensureAllProxies()
+            .map(proxy -> proxy.read(YangInstanceIdentifier.of())));
     }
 
     public void delete(final YangInstanceIdentifier path) {
-        ensureProxy(path).delete(path);
+        if (path.isEmpty()) {
+            ensureAllProxies().forEach(proxy -> proxy.delete(YangInstanceIdentifier.of()));
+        } else {
+            ensureProxy(path).delete(path);
+        }
     }
 
-    public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        ensureProxy(path).merge(path, data);
+    public void merge(final YangInstanceIdentifier path, final NormalizedNode data) {
+        if (path.isEmpty()) {
+            mergeRoot(RootScatterGather.castRootNode(data));
+        } else {
+            ensureProxy(path).merge(path, data);
+        }
     }
 
-    public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        ensureProxy(path).write(path, data);
+    private void mergeRoot(final @NonNull ContainerNode rootData) {
+        if (!rootData.isEmpty()) {
+            RootScatterGather.scatterTouched(rootData, this::ensureProxy).forEach(
+                scattered -> scattered.shard().merge(YangInstanceIdentifier.of(), scattered.container()));
+        }
     }
 
-    private boolean ensureClosed() {
-        final int local = state;
-        if (local != CLOSED_STATE) {
-            final boolean success = STATE_UPDATER.compareAndSet(this, OPEN_STATE, CLOSED_STATE);
-            Preconditions.checkState(success, "Transaction %s raced during close", this);
-            return true;
+    public void write(final YangInstanceIdentifier path, final NormalizedNode data) {
+        if (path.isEmpty()) {
+            writeRoot(RootScatterGather.castRootNode(data));
         } else {
-            return false;
+            ensureProxy(path).write(path, data);
         }
     }
 
-    public DOMStoreThreePhaseCommitCohort ready() {
-        Preconditions.checkState(ensureClosed(), "Attempted to submit a closed transaction %s", this);
+    private void writeRoot(final @NonNull ContainerNode rootData) {
+        RootScatterGather.scatterAll(rootData, this::ensureProxy, ensureAllProxies()).forEach(
+            scattered -> scattered.shard().write(YangInstanceIdentifier.of(), scattered.container()));
+    }
 
-        for (AbstractProxyTransaction p : proxies.values()) {
-            p.seal();
-        }
-        parent.onTransactionReady(this);
-
-        switch (proxies.size()) {
-            case 0:
-                return EmptyTransactionCommitCohort.INSTANCE;
-            case 1:
-                return new DirectTransactionCommitCohort(Iterables.getOnlyElement(proxies.values()));
-            default:
-                return new ClientTransactionCommitCohort(proxies.values());
-        }
+    private AbstractProxyTransaction ensureProxy(final PathArgument childId) {
+        return ensureProxy(YangInstanceIdentifier.of(childId));
     }
 
-    /**
-     * Release all state associated with this transaction.
-     */
-    public void abort() {
-        if (ensureClosed()) {
-            for (AbstractProxyTransaction proxy : proxies.values()) {
-                proxy.abort();
-            }
-            proxies.clear();
-        }
+    public DOMStoreThreePhaseCommitCohort ready() {
+        final Map<Long, AbstractProxyTransaction> participants = ensureClosed();
+        checkState(participants != null, "Attempted to submit a closed transaction %s", this);
+
+        final Collection<AbstractProxyTransaction> toReady = participants.values();
+        toReady.forEach(AbstractProxyTransaction::seal);
+
+        final TransactionIdentifier txId = getIdentifier();
+        final AbstractClientHistory parent = parent();
+        parent.onTransactionShardsBound(txId, participants.keySet());
+
+        final AbstractTransactionCommitCohort cohort = switch (toReady.size()) {
+            case 0 -> new EmptyTransactionCommitCohort(parent, txId);
+            case 1 -> new DirectTransactionCommitCohort(parent, txId, toReady.iterator().next());
+            default -> new ClientTransactionCommitCohort(parent, txId, toReady);
+        };
+        return parent.onTransactionReady(this, cohort);
     }
 
     @Override
-    void localAbort(final Throwable cause) {
-        LOG.debug("Aborting transaction {}", getIdentifier(), cause);
-        abort();
+    final AbstractProxyTransaction createProxy(final Long shard) {
+        return parent().createTransactionProxy(getIdentifier(), shard);
     }
 }