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=9b4c5a962f9b62959ffc3464b475c27a7d198b59;hb=abaef4a5ae37f27542155457fe7306a4662b1eeb;hp=f5f545a48e85af1c6faa255ee031b08934d1dcdc;hpb=98d1c5606bad9633ce5549bcd691a98c75abdf6a;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 f5f545a48e..9b4c5a962f 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
@@ -8,153 +8,100 @@
 package org.opendaylight.controller.cluster.databroker.actors.dds;
 
 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 java.util.Map;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import com.google.common.util.concurrent.FluentFuture;
+import java.util.Collection;
+import java.util.Optional;
 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.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 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);
+public class ClientTransaction extends AbstractClientHandle<AbstractProxyTransaction> {
+    ClientTransaction(final AbstractClientHistory parent, final TransactionIdentifier transactionId) {
+        super(parent, transactionId);
     }
 
-    private void checkNotClosed() {
-        Preconditions.checkState(state == OPEN_STATE, "Transaction %s is closed", transactionId);
+    private AbstractProxyTransaction ensureTransactionProxy(final YangInstanceIdentifier path) {
+        return ensureProxy(path);
     }
 
-    private AbstractProxyTransaction ensureProxy(final YangInstanceIdentifier path) {
-        checkNotClosed();
-
-        final ModuleShardBackendResolver resolver = parent.getClient().resolver();
-        final Long shard = resolver.resolveShardForPath(path);
-        AbstractProxyTransaction ret = proxies.get(shard);
-        if (ret == null) {
-            ret = AbstractProxyTransaction.create(parent.getClient(), parent.getHistoryForCookie(shard),
-                transactionId.getTransactionId(), resolver.getFutureBackendInfo(shard));
-            proxies.put(shard, ret);
-        }
-        return ret;
+    public FluentFuture<Boolean> exists(final YangInstanceIdentifier path) {
+        return ensureTransactionProxy(path).exists(path);
     }
 
-    @Override
-    public TransactionIdentifier getIdentifier() {
-        return transactionId;
-    }
-
-    public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
-        return ensureProxy(path).exists(path);
-    }
-
-    public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
-        return ensureProxy(path).read(path);
+    public FluentFuture<Optional<NormalizedNode>> read(final YangInstanceIdentifier path) {
+        return ensureTransactionProxy(path).read(path);
     }
 
     public void delete(final YangInstanceIdentifier path) {
-        ensureProxy(path).delete(path);
-    }
-
-    public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        ensureProxy(path).merge(path, data);
+        ensureTransactionProxy(path).delete(path);
     }
 
-    public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
-        ensureProxy(path).write(path, data);
+    public void merge(final YangInstanceIdentifier path, final NormalizedNode data) {
+        ensureTransactionProxy(path).merge(path, data);
     }
 
-    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;
-        } else {
-            return false;
-        }
+    public void write(final YangInstanceIdentifier path, final NormalizedNode data) {
+        ensureTransactionProxy(path).write(path, data);
     }
 
     public DOMStoreThreePhaseCommitCohort ready() {
-        Preconditions.checkState(ensureClosed(), "Attempted to submit a closed transaction %s", this);
+        final Collection<AbstractProxyTransaction> toReady = ensureClosed();
+        Preconditions.checkState(toReady != null, "Attempted to submit a closed transaction %s", this);
 
-        for (AbstractProxyTransaction p : proxies.values()) {
-            p.seal();
-        }
-        parent.onTransactionReady(this);
-
-        switch (proxies.size()) {
+        toReady.forEach(AbstractProxyTransaction::seal);
+        final AbstractTransactionCommitCohort cohort;
+        switch (toReady.size()) {
             case 0:
-                return EmptyTransactionCommitCohort.INSTANCE;
+                cohort = new EmptyTransactionCommitCohort(parent(), getIdentifier());
+                break;
             case 1:
-                return new DirectTransactionCommitCohort(Iterables.getOnlyElement(proxies.values()));
+                cohort = new DirectTransactionCommitCohort(parent(), getIdentifier(),
+                    Iterables.getOnlyElement(toReady));
+                break;
             default:
-                return new ClientTransactionCommitCohort(proxies.values());
+                cohort = new ClientTransactionCommitCohort(parent(), getIdentifier(), toReady);
+                break;
         }
-    }
 
-    /**
-     * Release all state associated with this transaction.
-     */
-    public void abort() {
-        if (ensureClosed()) {
-            for (AbstractProxyTransaction proxy : proxies.values()) {
-                proxy.abort();
-            }
-            proxies.clear();
-        }
+        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);
     }
 }