opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java

   1 /*
   2  * Copyright (c) 2014 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.datastore;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static com.google.common.base.Preconditions.checkState;
  12 import static com.google.common.base.Verify.verifyNotNull;
  13 import static java.util.Objects.requireNonNull;
  14
  15 import akka.actor.ActorSelection;
  16 import com.google.common.annotations.VisibleForTesting;
  17 import com.google.common.collect.Iterables;
  18 import com.google.common.util.concurrent.FluentFuture;
  19 import com.google.common.util.concurrent.Futures;
  20 import com.google.common.util.concurrent.ListenableFuture;
  21 import com.google.common.util.concurrent.MoreExecutors;
  22 import com.google.common.util.concurrent.SettableFuture;
  23 import java.util.ArrayList;
  24 import java.util.Collection;
  25 import java.util.HashMap;
  26 import java.util.List;
  27 import java.util.Map;
  28 import java.util.Map.Entry;
  29 import java.util.Optional;
  30 import java.util.Set;
  31 import java.util.SortedSet;
  32 import java.util.TreeMap;
  33 import java.util.TreeSet;
  34 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  35 import org.opendaylight.controller.cluster.datastore.TransactionModificationOperation.DeleteOperation;
  36 import org.opendaylight.controller.cluster.datastore.TransactionModificationOperation.MergeOperation;
  37 import org.opendaylight.controller.cluster.datastore.TransactionModificationOperation.WriteOperation;
  38 import org.opendaylight.controller.cluster.datastore.messages.AbstractRead;
  39 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
  40 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
  41 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
  42 import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregator;
  43 import org.opendaylight.mdsal.dom.spi.store.AbstractDOMStoreTransaction;
  44 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
  45 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  46 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  47 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  48 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  49 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  50 import org.opendaylight.yangtools.yang.data.api.schema.builder.DataContainerNodeBuilder;
  51 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
  52 import org.opendaylight.yangtools.yang.data.tree.api.DataValidationFailedException;
  53 import org.slf4j.Logger;
  54 import org.slf4j.LoggerFactory;
  55 import scala.concurrent.Future;
  56 import scala.concurrent.Promise;
  57
  58 /**
  59  * A transaction potentially spanning multiple backend shards.
  60  */
  61 public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIdentifier>
  62         implements DOMStoreReadWriteTransaction {
  63     private enum TransactionState {
  64         OPEN,
  65         READY,
  66         CLOSED,
  67     }
  68
  69     private static final Logger LOG = LoggerFactory.getLogger(TransactionProxy.class);
  70     private static final DeleteOperation ROOT_DELETE_OPERATION = new DeleteOperation(YangInstanceIdentifier.empty());
  71
  72     private final Map<String, AbstractTransactionContextWrapper> txContextWrappers = new TreeMap<>();
  73     private final AbstractTransactionContextFactory<?> txContextFactory;
  74     private final TransactionType type;
  75     private TransactionState state = TransactionState.OPEN;
  76
  77     @VisibleForTesting
  78     public TransactionProxy(final AbstractTransactionContextFactory<?> txContextFactory, final TransactionType type) {
  79         super(txContextFactory.nextIdentifier(), txContextFactory.getActorUtils().getDatastoreContext()
  80                 .isTransactionDebugContextEnabled());
  81         this.txContextFactory = txContextFactory;
  82         this.type = requireNonNull(type);
  83
  84         LOG.debug("New {} Tx - {}", type, getIdentifier());
  85     }
  86
  87     @Override
  88     public FluentFuture<Boolean> exists(final YangInstanceIdentifier path) {
  89         return executeRead(shardNameFromIdentifier(path), new DataExists(path, DataStoreVersions.CURRENT_VERSION));
  90     }
  91
  92     private <T> FluentFuture<T> executeRead(final String shardName, final AbstractRead<T> readCmd) {
  93         checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed");
  94
  95         LOG.trace("Tx {} {} {}", getIdentifier(), readCmd.getClass().getSimpleName(), readCmd.getPath());
  96
  97         final SettableFuture<T> proxyFuture = SettableFuture.create();
  98         AbstractTransactionContextWrapper contextWrapper = getContextWrapper(shardName);
  99         contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() {
 100             @Override
 101             public void invoke(final TransactionContext transactionContext, final Boolean havePermit) {
 102                 transactionContext.executeRead(readCmd, proxyFuture, havePermit);
 103             }
 104         });
 105
 106         return FluentFuture.from(proxyFuture);
 107     }
 108
 109     @Override
 110     public FluentFuture<Optional<NormalizedNode>> read(final YangInstanceIdentifier path) {
 111         checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed");
 112         requireNonNull(path, "path should not be null");
 113
 114         LOG.trace("Tx {} read {}", getIdentifier(), path);
 115         return path.isEmpty() ? readAllData() : singleShardRead(shardNameFromIdentifier(path), path);
 116     }
 117
 118     private FluentFuture<Optional<NormalizedNode>> singleShardRead(final String shardName,
 119             final YangInstanceIdentifier path) {
 120         return executeRead(shardName, new ReadData(path, DataStoreVersions.CURRENT_VERSION));
 121     }
 122
 123     private FluentFuture<Optional<NormalizedNode>> readAllData() {
 124         final Set<String> allShardNames = txContextFactory.getActorUtils().getConfiguration().getAllShardNames();
 125         final Collection<FluentFuture<Optional<NormalizedNode>>> futures = new ArrayList<>(allShardNames.size());
 126
 127         for (String shardName : allShardNames) {
 128             futures.add(singleShardRead(shardName, YangInstanceIdentifier.empty()));
 129         }
 130
 131         final ListenableFuture<List<Optional<NormalizedNode>>> listFuture = Futures.allAsList(futures);
 132         final ListenableFuture<Optional<NormalizedNode>> aggregateFuture;
 133
 134         aggregateFuture = Futures.transform(listFuture, input -> {
 135             try {
 136                 return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.empty(), input,
 137                         txContextFactory.getActorUtils().getSchemaContext(),
 138                         txContextFactory.getActorUtils().getDatastoreContext().getLogicalStoreType());
 139             } catch (DataValidationFailedException e) {
 140                 throw new IllegalArgumentException("Failed to aggregate", e);
 141             }
 142         }, MoreExecutors.directExecutor());
 143
 144         return FluentFuture.from(aggregateFuture);
 145     }
 146
 147     @Override
 148     public void delete(final YangInstanceIdentifier path) {
 149         checkModificationState("delete", path);
 150
 151         if (path.isEmpty()) {
 152             deleteAllData();
 153         } else {
 154             executeModification(new DeleteOperation(path));
 155         }
 156     }
 157
 158     private void deleteAllData() {
 159         for (String shardName : getActorUtils().getConfiguration().getAllShardNames()) {
 160             getContextWrapper(shardName).maybeExecuteTransactionOperation(ROOT_DELETE_OPERATION);
 161         }
 162     }
 163
 164     @Override
 165     public void merge(final YangInstanceIdentifier path, final NormalizedNode data) {
 166         checkModificationState("merge", path);
 167
 168         if (path.isEmpty()) {
 169             mergeAllData(checkRootData(data));
 170         } else {
 171             executeModification(new MergeOperation(path, data));
 172         }
 173     }
 174
 175     private void mergeAllData(final ContainerNode rootData) {
 176         // Populate requests for individual shards that are being touched
 177         final Map<String, DataContainerNodeBuilder<NodeIdentifier, ContainerNode>> rootBuilders = new HashMap<>();
 178         for (DataContainerChild child : rootData.body()) {
 179             final String shardName = shardNameFromRootChild(child);
 180             rootBuilders.computeIfAbsent(shardName,
 181                 unused -> Builders.containerBuilder().withNodeIdentifier(rootData.getIdentifier()))
 182                 .addChild(child);
 183         }
 184
 185         // Now dispatch all merges
 186         for (Entry<String, DataContainerNodeBuilder<NodeIdentifier, ContainerNode>> entry : rootBuilders.entrySet()) {
 187             getContextWrapper(entry.getKey()).maybeExecuteTransactionOperation(new MergeOperation(
 188                 YangInstanceIdentifier.empty(), entry.getValue().build()));
 189         }
 190     }
 191
 192     @Override
 193     public void write(final YangInstanceIdentifier path, final NormalizedNode data) {
 194         checkModificationState("write", path);
 195
 196         if (path.isEmpty()) {
 197             writeAllData(checkRootData(data));
 198         } else {
 199             executeModification(new WriteOperation(path, data));
 200         }
 201     }
 202
 203     private void writeAllData(final ContainerNode rootData) {
 204         // Open builders for all shards
 205         final Map<String, DataContainerNodeBuilder<NodeIdentifier, ContainerNode>> rootBuilders = new HashMap<>();
 206         for (String shardName : getActorUtils().getConfiguration().getAllShardNames()) {
 207             rootBuilders.put(shardName, Builders.containerBuilder().withNodeIdentifier(rootData.getIdentifier()));
 208         }
 209
 210         // Now distribute children as needed
 211         for (DataContainerChild child : rootData.body()) {
 212             final String shardName = shardNameFromRootChild(child);
 213             verifyNotNull(rootBuilders.get(shardName), "Failed to find builder for %s", shardName).addChild(child);
 214         }
 215
 216         // Now dispatch all writes
 217         for (Entry<String, DataContainerNodeBuilder<NodeIdentifier, ContainerNode>> entry : rootBuilders.entrySet()) {
 218             getContextWrapper(entry.getKey()).maybeExecuteTransactionOperation(new WriteOperation(
 219                 YangInstanceIdentifier.empty(), entry.getValue().build()));
 220         }
 221     }
 222
 223     private void executeModification(final TransactionModificationOperation operation) {
 224         getContextWrapper(operation.path()).maybeExecuteTransactionOperation(operation);
 225     }
 226
 227     private static ContainerNode checkRootData(final NormalizedNode data) {
 228         // Root has to be a container
 229         checkArgument(data instanceof ContainerNode, "Invalid root data %s", data);
 230         return (ContainerNode) data;
 231     }
 232
 233     private void checkModificationState(final String opName, final YangInstanceIdentifier path) {
 234         checkState(type != TransactionType.READ_ONLY, "Modification operation on read-only transaction is not allowed");
 235         checkState(state == TransactionState.OPEN, "Transaction is sealed - further modifications are not allowed");
 236         LOG.trace("Tx {} {} {}", getIdentifier(), opName, path);
 237     }
 238
 239     private boolean seal(final TransactionState newState) {
 240         if (state == TransactionState.OPEN) {
 241             state = newState;
 242             return true;
 243         }
 244         return false;
 245     }
 246
 247     @Override
 248     public final void close() {
 249         if (!seal(TransactionState.CLOSED)) {
 250             checkState(state == TransactionState.CLOSED, "Transaction %s is ready, it cannot be closed",
 251                 getIdentifier());
 252             // Idempotent no-op as per AutoCloseable recommendation
 253             return;
 254         }
 255
 256         for (AbstractTransactionContextWrapper contextWrapper : txContextWrappers.values()) {
 257             contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() {
 258                 @Override
 259                 public void invoke(final TransactionContext transactionContext, final Boolean havePermit) {
 260                     transactionContext.closeTransaction();
 261                 }
 262             });
 263         }
 264
 265
 266         txContextWrappers.clear();
 267     }
 268
 269     @Override
 270     public final AbstractThreePhaseCommitCohort<?> ready() {
 271         checkState(type != TransactionType.READ_ONLY, "Read-only transactions cannot be readied");
 272
 273         final boolean success = seal(TransactionState.READY);
 274         checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state);
 275
 276         LOG.debug("Tx {} Readying {} components for commit", getIdentifier(), txContextWrappers.size());
 277
 278         final AbstractThreePhaseCommitCohort<?> ret;
 279         switch (txContextWrappers.size()) {
 280             case 0:
 281                 ret = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
 282                 break;
 283             case 1:
 284                 final Entry<String, AbstractTransactionContextWrapper> e = Iterables.getOnlyElement(
 285                         txContextWrappers.entrySet());
 286                 ret = createSingleCommitCohort(e.getKey(), e.getValue());
 287                 break;
 288             default:
 289                 ret = createMultiCommitCohort();
 290         }
 291
 292         txContextFactory.onTransactionReady(getIdentifier(), ret.getCohortFutures());
 293
 294         final Throwable debugContext = getDebugContext();
 295         return debugContext == null ? ret : new DebugThreePhaseCommitCohort(getIdentifier(), ret, debugContext);
 296     }
 297
 298     @SuppressWarnings({ "rawtypes", "unchecked" })
 299     private AbstractThreePhaseCommitCohort<?> createSingleCommitCohort(final String shardName,
 300             final AbstractTransactionContextWrapper contextWrapper) {
 301
 302         LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), shardName);
 303
 304         final OperationCallback.Reference operationCallbackRef =
 305                 new OperationCallback.Reference(OperationCallback.NO_OP_CALLBACK);
 306
 307         final TransactionContext transactionContext = contextWrapper.getTransactionContext();
 308         final Future future;
 309         if (transactionContext == null) {
 310             final Promise promise = akka.dispatch.Futures.promise();
 311             contextWrapper.maybeExecuteTransactionOperation(new TransactionOperation() {
 312                 @Override
 313                 public void invoke(final TransactionContext newTransactionContext, final Boolean havePermit) {
 314                     promise.completeWith(getDirectCommitFuture(newTransactionContext, operationCallbackRef,
 315                         havePermit));
 316                 }
 317             });
 318             future = promise.future();
 319         } else {
 320             // avoid the creation of a promise and a TransactionOperation
 321             future = getDirectCommitFuture(transactionContext, operationCallbackRef, null);
 322         }
 323
 324         return new SingleCommitCohortProxy(txContextFactory.getActorUtils(), future, getIdentifier(),
 325             operationCallbackRef);
 326     }
 327
 328     private Future<?> getDirectCommitFuture(final TransactionContext transactionContext,
 329             final OperationCallback.Reference operationCallbackRef, final Boolean havePermit) {
 330         TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback(
 331                 txContextFactory.getActorUtils());
 332         operationCallbackRef.set(rateLimitingCallback);
 333         rateLimitingCallback.run();
 334         return transactionContext.directCommit(havePermit);
 335     }
 336
 337     private AbstractThreePhaseCommitCohort<ActorSelection> createMultiCommitCohort() {
 338
 339         final List<ThreePhaseCommitCohortProxy.CohortInfo> cohorts = new ArrayList<>(txContextWrappers.size());
 340         final Optional<SortedSet<String>> shardNames = Optional.of(new TreeSet<>(txContextWrappers.keySet()));
 341         for (Entry<String, AbstractTransactionContextWrapper> e : txContextWrappers.entrySet()) {
 342             LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), e.getKey());
 343
 344             final AbstractTransactionContextWrapper wrapper = e.getValue();
 345
 346             // The remote tx version is obtained the via TransactionContext which may not be available yet so
 347             // we pass a Supplier to dynamically obtain it. Once the ready Future is resolved the
 348             // TransactionContext is available.
 349             cohorts.add(new ThreePhaseCommitCohortProxy.CohortInfo(wrapper.readyTransaction(shardNames),
 350                 () -> wrapper.getTransactionContext().getTransactionVersion()));
 351         }
 352
 353         return new ThreePhaseCommitCohortProxy(txContextFactory.getActorUtils(), cohorts, getIdentifier());
 354     }
 355
 356     private String shardNameFromRootChild(final DataContainerChild child) {
 357         return shardNameFromIdentifier(YangInstanceIdentifier.create(child.getIdentifier()));
 358     }
 359
 360     private String shardNameFromIdentifier(final YangInstanceIdentifier path) {
 361         return getActorUtils().getShardStrategyFactory().getStrategy(path).findShard(path);
 362     }
 363
 364     private AbstractTransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) {
 365         return getContextWrapper(shardNameFromIdentifier(path));
 366     }
 367
 368     private AbstractTransactionContextWrapper getContextWrapper(final String shardName) {
 369         final AbstractTransactionContextWrapper existing = txContextWrappers.get(shardName);
 370         if (existing != null) {
 371             return existing;
 372         }
 373
 374         final AbstractTransactionContextWrapper fresh = txContextFactory.newTransactionContextWrapper(this, shardName);
 375         txContextWrappers.put(shardName, fresh);
 376         return fresh;
 377     }
 378
 379     TransactionType getType() {
 380         return type;
 381     }
 382
 383     boolean isReady() {
 384         return state != TransactionState.OPEN;
 385     }
 386
 387     final ActorUtils getActorUtils() {
 388         return txContextFactory.getActorUtils();
 389     }
 390 }