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 AbstractClientHistory parent, final TransactionIdentifier transactionId) {
  66         this.transactionId = Preconditions.checkNotNull(transactionId);
  67         this.parent = Preconditions.checkNotNull(parent);
  68     }
  69
  70     private void checkNotClosed() {
  71         Preconditions.checkState(state == OPEN_STATE, "Transaction %s is closed", transactionId);
  72     }
  73
  74     private AbstractProxyTransaction createProxy(final Long shard) {
  75         return parent.createTransactionProxy(transactionId, shard);
  76     }
  77
  78     private AbstractProxyTransaction ensureProxy(final YangInstanceIdentifier path) {
  79         checkNotClosed();
  80
  81         final ModuleShardBackendResolver resolver = parent.getClient().resolver();
  82         final Long shard = resolver.resolveShardForPath(path);
  83         return proxies.computeIfAbsent(shard, this::createProxy);
  84     }
  85
  86     @Override
  87     public TransactionIdentifier getIdentifier() {
  88         return transactionId;
  89     }
  90
  91     public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
  92         return ensureProxy(path).exists(path);
  93     }
  94
  95     public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
  96         return ensureProxy(path).read(path);
  97     }
  98
  99     public void delete(final YangInstanceIdentifier path) {
 100         ensureProxy(path).delete(path);
 101     }
 102
 103     public void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 104         ensureProxy(path).merge(path, data);
 105     }
 106
 107     public void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
 108         ensureProxy(path).write(path, data);
 109     }
 110
 111     private boolean ensureClosed() {
 112         final int local = state;
 113         if (local != CLOSED_STATE) {
 114             final boolean success = STATE_UPDATER.compareAndSet(this, OPEN_STATE, CLOSED_STATE);
 115             Preconditions.checkState(success, "Transaction %s raced during close", this);
 116             return true;
 117         } else {
 118             return false;
 119         }
 120     }
 121
 122     public DOMStoreThreePhaseCommitCohort ready() {
 123         Preconditions.checkState(ensureClosed(), "Attempted to submit a closed transaction %s", this);
 124
 125         for (AbstractProxyTransaction p : proxies.values()) {
 126             p.seal();
 127         }
 128         parent.onTransactionReady(this);
 129
 130         switch (proxies.size()) {
 131             case 0:
 132                 return EmptyTransactionCommitCohort.INSTANCE;
 133             case 1:
 134                 return new DirectTransactionCommitCohort(Iterables.getOnlyElement(proxies.values()));
 135             default:
 136                 return new ClientTransactionCommitCohort(proxies.values());
 137         }
 138     }
 139
 140     /**
 141      * Release all state associated with this transaction.
 142      */
 143     public void abort() {
 144         if (ensureClosed()) {
 145             for (AbstractProxyTransaction proxy : proxies.values()) {
 146                 proxy.abort();
 147             }
 148             proxies.clear();
 149         }
 150     }
 151
 152     @Override
 153     void localAbort(final Throwable cause) {
 154         LOG.debug("Aborting transaction {}", getIdentifier(), cause);
 155         abort();
 156     }
 157 }