opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientTransaction.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.databroker.actors.dds;
   9
  10 import com.google.common.annotations.Beta;
  11 import com.google.common.base.Optional;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.collect.Iterables;
  14 import com.google.common.util.concurrent.CheckedFuture;
  15 import java.util.HashMap;
  16 import java.util.Map;
  17 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
  18 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  19 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  20 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  21 import org.opendaylight.yangtools.concepts.Identifiable;
  22 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  23 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 /**
  28  * Client-side view of a free-standing transaction.
  29  *
  30  * <p>
  31  * This interface is used by the world outside of the actor system and in the actor system it is manifested via
  32  * its client actor. That requires some state transfer with {@link DistributedDataStoreClientBehavior}. In order to
  33  * reduce request latency, all messages are carbon-copied (and enqueued first) to the client actor.
  34  *
  35  * <p>
  36  * It is internally composed of multiple {@link RemoteProxyTransaction}s, each responsible for a component shard.
  37  *
  38  * <p>
  39  * Implementation is quite a bit complex, and involves cooperation with {@link AbstractClientHistory} for tracking
  40  * gaps in transaction identifiers seen by backends.
  41  *
  42  * <p>
  43  * These gaps need to be accounted for in the transaction setup message sent to a particular backend, so it can verify
  44  * that the requested transaction is in-sequence. This is critical in ensuring that transactions (which are independent
  45  * entities from message queueing perspective) do not get reodered -- thus allowing multiple in-flight transactions.
  46  *
  47  * <p>
  48  * Alternative would be to force visibility by sending an abort request to all potential backends, but that would mean
  49  * that even empty transactions increase load on all shards -- which would be a scalability issue.
  50  *
  51  * <p>
  52  * Yet another alternative would be to introduce inter-transaction dependencies to the queueing layer in client actor,
  53  * but that would require additional indirection and complexity.
  54  *
  55  * @author Robert Varga
  56  */
  57 @Beta
  58 public final class ClientTransaction extends LocalAbortable implements Identifiable<TransactionIdentifier> {
  59     private static final Logger LOG = LoggerFactory.getLogger(ClientTransaction.class);
  60     private static final AtomicIntegerFieldUpdater<ClientTransaction> STATE_UPDATER =
  61             AtomicIntegerFieldUpdater.newUpdater(ClientTransaction.class, "state");
  62     private static final int OPEN_STATE = 0;
  63     private static final int CLOSED_STATE = 1;
  64
  65     private final Map<Long, AbstractProxyTransaction> proxies = new HashMap<>();
  66     private final TransactionIdentifier transactionId;
  67     private final AbstractClientHistory parent;
  68
  69     private volatile int state = OPEN_STATE;
  70
  71     ClientTransaction(final AbstractClientHistory parent, final TransactionIdentifier transactionId) {
  72         this.transactionId = Preconditions.checkNotNull(transactionId);
  73         this.parent = Preconditions.checkNotNull(parent);
  74     }
  75
  76     private void checkNotClosed() {
  77         Preconditions.checkState(state == OPEN_STATE, "Transaction %s is closed", transactionId);
  78     }
  79
  80     private AbstractProxyTransaction createProxy(final Long shard) {
  81         return parent.createTransactionProxy(transactionId, shard);
  82     }
  83
  84     private AbstractProxyTransaction ensureProxy(final YangInstanceIdentifier path) {
  85         checkNotClosed();
  86
  87         final ModuleShardBackendResolver resolver = parent.getClient().resolver();
  88         final Long shard = resolver.resolveShardForPath(path);
  89         return proxies.computeIfAbsent(shard, this::createProxy);
  90     }
  91
  92     @Override
  93     public TransactionIdentifier getIdentifier() {
  94         return transactionId;
  95     }
  96
  97     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
  98         return ensureProxy(path).exists(path);
  99     }
 100
 101     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
 102         return ensureProxy(path).read(path);
 103     }
 104
 105     public void delete(final YangInstanceIdentifier path) {
 106         ensureProxy(path).delete(path);
 107     }
 108
 109     public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 110         ensureProxy(path).merge(path, data);
 111     }
 112
 113     public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 114         ensureProxy(path).write(path, data);
 115     }
 116
 117     private boolean ensureClosed() {
 118         final int local = state;
 119         if (local != CLOSED_STATE) {
 120             final boolean success = STATE_UPDATER.compareAndSet(this, OPEN_STATE, CLOSED_STATE);
 121             Preconditions.checkState(success, "Transaction %s raced during close", this);
 122             return true;
 123         } else {
 124             return false;
 125         }
 126     }
 127
 128     public DOMStoreThreePhaseCommitCohort ready() {
 129         Preconditions.checkState(ensureClosed(), "Attempted to submit a closed transaction %s", this);
 130
 131         for (AbstractProxyTransaction p : proxies.values()) {
 132             p.seal();
 133         }
 134         parent.onTransactionReady(this);
 135
 136         switch (proxies.size()) {
 137             case 0:
 138                 return EmptyTransactionCommitCohort.INSTANCE;
 139             case 1:
 140                 return new DirectTransactionCommitCohort(Iterables.getOnlyElement(proxies.values()));
 141             default:
 142                 return new ClientTransactionCommitCohort(proxies.values());
 143         }
 144     }
 145
 146     /**
 147      * Release all state associated with this transaction.
 148      */
 149     public void abort() {
 150         if (ensureClosed()) {
 151             for (AbstractProxyTransaction proxy : proxies.values()) {
 152                 proxy.abort();
 153             }
 154             proxies.clear();
 155         }
 156     }
 157
 158     @Override
 159     void localAbort(final Throwable cause) {
 160         LOG.debug("Aborting transaction {}", getIdentifier(), cause);
 161         abort();
 162     }
 163 }