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