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.Map;
  16 import java.util.concurrent.ConcurrentHashMap;
  17 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
  18 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  19 import org.opendaylight.mdsal.common.api.ReadFailedException;
  20 import org.opendaylight.mdsal.dom.spi.store.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 ConcurrentHashMap<>();
  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 Long shard = parent.resolveShardForPath(path);
  88         return proxies.computeIfAbsent(shard, this::createProxy);
  89     }
  90
  91     @Override
  92     public TransactionIdentifier getIdentifier() {
  93         return transactionId;
  94     }
  95
  96     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
  97         return ensureProxy(path).exists(path);
  98     }
  99
 100     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
 101         return ensureProxy(path).read(path);
 102     }
 103
 104     public void delete(final YangInstanceIdentifier path) {
 105         ensureProxy(path).delete(path);
 106     }
 107
 108     public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 109         ensureProxy(path).merge(path, data);
 110     }
 111
 112     public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 113         ensureProxy(path).write(path, data);
 114     }
 115
 116     private boolean ensureClosed() {
 117         final int local = state;
 118         if (local == CLOSED_STATE) {
 119             return false;
 120         }
 121
 122         final boolean success = STATE_UPDATER.compareAndSet(this, OPEN_STATE, CLOSED_STATE);
 123         Preconditions.checkState(success, "Transaction %s raced during close", this);
 124         return true;
 125     }
 126
 127     public DOMStoreThreePhaseCommitCohort ready() {
 128         Preconditions.checkState(ensureClosed(), "Attempted to submit a closed transaction %s", this);
 129
 130         for (AbstractProxyTransaction p : proxies.values()) {
 131             p.seal();
 132         }
 133
 134         final AbstractTransactionCommitCohort cohort;
 135         switch (proxies.size()) {
 136             case 0:
 137                 cohort = new EmptyTransactionCommitCohort(parent, transactionId);
 138                 break;
 139             case 1:
 140                 cohort = new DirectTransactionCommitCohort(parent, transactionId,
 141                     Iterables.getOnlyElement(proxies.values()));
 142                 break;
 143             default:
 144                 cohort = new ClientTransactionCommitCohort(parent, transactionId, proxies.values());
 145                 break;
 146         }
 147
 148         return parent.onTransactionReady(transactionId, cohort);
 149     }
 150
 151     /**
 152      * Release all state associated with this transaction.
 153      */
 154     public void abort() {
 155         if (commonAbort()) {
 156             parent.onTransactionAbort(transactionId);
 157         }
 158     }
 159
 160     private boolean commonAbort() {
 161         if (!ensureClosed()) {
 162             return false;
 163         }
 164
 165         for (AbstractProxyTransaction proxy : proxies.values()) {
 166             proxy.abort();
 167         }
 168         proxies.clear();
 169         return true;
 170     }
 171
 172     @Override
 173     void localAbort(final Throwable cause) {
 174         LOG.debug("Local abort of transaction {}", getIdentifier(), cause);
 175         commonAbort();
 176     }
 177
 178     Map<Long, AbstractProxyTransaction> getProxies() {
 179         return proxies;
 180     }
 181 }