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  * This interface is used by the world outside of the actor system and in the actor system it is manifested via
  31  * its client actor. That requires some state transfer with {@link DistributedDataStoreClientBehavior}. In order to
  32  * reduce request latency, all messages are carbon-copied (and enqueued first) to the client actor.
  33  *
  34  * It is internally composed of multiple {@link RemoteProxyTransaction}s, each responsible for a component shard.
  35  *
  36  * Implementation is quite a bit complex, and involves cooperation with {@link AbstractClientHistory} for tracking
  37  * gaps in transaction identifiers seen by backends.
  38  *
  39  * These gaps need to be accounted for in the transaction setup message sent to a particular backend, so it can verify
  40  * that the requested transaction is in-sequence. This is critical in ensuring that transactions (which are independent
  41  * entities from message queueing perspective) do not get reodered -- thus allowing multiple in-flight transactions.
  42  *
  43  * Alternative would be to force visibility by sending an abort request to all potential backends, but that would mean
  44  * that even empty transactions increase load on all shards -- which would be a scalability issue.
  45  *
  46  * Yet another alternative would be to introduce inter-transaction dependencies to the queueing layer in client actor,
  47  * but that would require additional indirection and complexity.
  48  *
  49  * @author Robert Varga
  50  */
  51 @Beta
  52 public final class ClientTransaction extends LocalAbortable implements Identifiable<TransactionIdentifier> {
  53     private static final Logger LOG = LoggerFactory.getLogger(ClientTransaction.class);
  54     private static final AtomicIntegerFieldUpdater<ClientTransaction> STATE_UPDATER =
  55             AtomicIntegerFieldUpdater.newUpdater(ClientTransaction.class, "state");
  56     private static final int OPEN_STATE = 0;
  57     private static final int CLOSED_STATE = 1;
  58
  59     private final Map<Long, AbstractProxyTransaction> proxies = new HashMap<>();
  60     private final TransactionIdentifier transactionId;
  61     private final AbstractClientHistory parent;
  62
  63     private volatile int state = OPEN_STATE;
  64
  65     ClientTransaction(final DistributedDataStoreClientBehavior client, final AbstractClientHistory parent,
  66         final TransactionIdentifier transactionId) {
  67         this.transactionId = Preconditions.checkNotNull(transactionId);
  68         this.parent = Preconditions.checkNotNull(parent);
  69     }
  70
  71     private void checkNotClosed() {
  72         Preconditions.checkState(state == OPEN_STATE, "Transaction %s is closed", transactionId);
  73     }
  74
  75     private AbstractProxyTransaction createProxy(final Long shard) {
  76         return parent.createTransactionProxy(transactionId, shard);
  77     }
  78
  79     private AbstractProxyTransaction ensureProxy(final YangInstanceIdentifier path) {
  80         checkNotClosed();
  81
  82         final ModuleShardBackendResolver resolver = parent.getClient().resolver();
  83         final Long shard = resolver.resolveShardForPath(path);
  84         return proxies.computeIfAbsent(shard, this::createProxy);
  85     }
  86
  87     @Override
  88     public TransactionIdentifier getIdentifier() {
  89         return transactionId;
  90     }
  91
  92     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
  93         return ensureProxy(path).exists(path);
  94     }
  95
  96     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
  97         return ensureProxy(path).read(path);
  98     }
  99
 100     public void delete(final YangInstanceIdentifier path) {
 101         ensureProxy(path).delete(path);
 102     }
 103
 104     public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 105         ensureProxy(path).merge(path, data);
 106     }
 107
 108     public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 109         ensureProxy(path).write(path, data);
 110     }
 111
 112     private boolean ensureClosed() {
 113         final int local = state;
 114         if (local != CLOSED_STATE) {
 115             final boolean success = STATE_UPDATER.compareAndSet(this, OPEN_STATE, CLOSED_STATE);
 116             Preconditions.checkState(success, "Transaction %s raced during close", this);
 117             return true;
 118         } else {
 119             return false;
 120         }
 121     }
 122
 123     public DOMStoreThreePhaseCommitCohort ready() {
 124         Preconditions.checkState(ensureClosed(), "Attempted to submit a closed transaction %s", this);
 125
 126         for (AbstractProxyTransaction p : proxies.values()) {
 127             p.seal();
 128         }
 129         parent.onTransactionReady(this);
 130
 131         switch (proxies.size()) {
 132             case 0:
 133                 return EmptyTransactionCommitCohort.INSTANCE;
 134             case 1:
 135                 return new DirectTransactionCommitCohort(Iterables.getOnlyElement(proxies.values()));
 136             default:
 137                 return new ClientTransactionCommitCohort(proxies.values());
 138         }
 139     }
 140
 141     /**
 142      * Release all state associated with this transaction.
 143      */
 144     public void abort() {
 145         if (ensureClosed()) {
 146             for (AbstractProxyTransaction proxy : proxies.values()) {
 147                 proxy.abort();
 148             }
 149             proxies.clear();
 150         }
 151     }
 152
 153     @Override
 154     void localAbort(final Throwable cause) {
 155         LOG.debug("Aborting transaction {}", getIdentifier(), cause);
 156         abort();
 157     }
 158 }