dom/mdsal-dom-broker/src/main/java/org/opendaylight/mdsal/dom/broker/ShardedDOMDataTreeWriteTransaction.java

   1 /*
   2  * Copyright (c) 2015 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.mdsal.dom.broker;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.collect.ImmutableMap;
  12 import com.google.common.util.concurrent.FluentFuture;
  13 import com.google.common.util.concurrent.FutureCallback;
  14 import com.google.common.util.concurrent.Futures;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import com.google.common.util.concurrent.MoreExecutors;
  17 import com.google.common.util.concurrent.SettableFuture;
  18 import java.util.ArrayDeque;
  19 import java.util.Deque;
  20 import java.util.Map;
  21 import java.util.Map.Entry;
  22 import java.util.concurrent.atomic.AtomicLong;
  23 import java.util.function.BiConsumer;
  24 import java.util.function.Consumer;
  25 import java.util.stream.Collectors;
  26 import javax.annotation.Nonnull;
  27 import javax.annotation.concurrent.GuardedBy;
  28 import javax.annotation.concurrent.NotThreadSafe;
  29 import org.eclipse.jdt.annotation.NonNull;
  30 import org.opendaylight.mdsal.common.api.CommitInfo;
  31 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
  32 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  33 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
  34 import org.opendaylight.mdsal.dom.spi.shard.DOMDataTreeShardWriteTransaction;
  35 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
  36 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  37 import org.slf4j.Logger;
  38 import org.slf4j.LoggerFactory;
  39
  40 @NotThreadSafe
  41 final class ShardedDOMDataTreeWriteTransaction implements DOMDataTreeCursorAwareTransaction {
  42     private static final Logger LOG = LoggerFactory.getLogger(ShardedDOMDataTreeWriteTransaction.class);
  43     private static final AtomicLong COUNTER = new AtomicLong();
  44
  45     private final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> transactions;
  46     private final ShardedDOMDataTreeProducer producer;
  47     private final ProducerLayout layout;
  48     private final String identifier;
  49
  50     private final SettableFuture<CommitInfo> future = SettableFuture.create();
  51
  52     @GuardedBy("this")
  53     private boolean closed =  false;
  54
  55     @GuardedBy("this")
  56     private DOMDataTreeWriteCursor openCursor;
  57
  58     ShardedDOMDataTreeWriteTransaction(final ShardedDOMDataTreeProducer producer,
  59         final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> transactions, final ProducerLayout layout) {
  60         this.producer = Preconditions.checkNotNull(producer);
  61         this.transactions = ImmutableMap.copyOf(transactions);
  62         this.layout = Preconditions.checkNotNull(layout);
  63         this.identifier = "SHARDED-DOM-" + COUNTER.getAndIncrement();
  64         LOG.debug("Created new transaction {}", identifier);
  65     }
  66
  67     private DOMDataTreeShardWriteTransaction lookup(final DOMDataTreeIdentifier prefix) {
  68         final DOMDataTreeShardWriteTransaction fast = transactions.get(prefix);
  69         if (fast != null) {
  70             return fast;
  71         }
  72
  73         LOG.debug("Prefix {} not found in available subtrees {}, fallback to slow path", prefix, transactions.keySet());
  74         for (final Entry<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> e : transactions.entrySet()) {
  75             if (e.getKey().contains(prefix)) {
  76                 return e.getValue();
  77             }
  78         }
  79
  80         return null;
  81     }
  82
  83     @Override
  84     public String getIdentifier() {
  85         return identifier;
  86     }
  87
  88     @Override
  89     public synchronized boolean cancel() {
  90         if (closed) {
  91             return false;
  92         }
  93
  94         LOG.debug("Cancelling transaction {}", identifier);
  95         if (openCursor != null) {
  96             openCursor.close();
  97         }
  98         for (final DOMDataTreeShardWriteTransaction tx : transactions.values()) {
  99             tx.close();
 100         }
 101
 102         closed = true;
 103         producer.cancelTransaction(this);
 104         return true;
 105     }
 106
 107     @Override
 108     public synchronized DOMDataTreeWriteCursor createCursor(final DOMDataTreeIdentifier prefix) {
 109         Preconditions.checkState(!closed, "Transaction is closed already");
 110         Preconditions.checkState(openCursor == null, "There is still a cursor open");
 111         Preconditions.checkArgument(!producer.isDelegatedToChild(prefix), "Path %s is delegated to child producer.",
 112             prefix);
 113
 114         final DOMDataTreeShardWriteTransaction lookup = lookup(prefix);
 115         Preconditions.checkArgument(lookup != null, "Path %s is not accessible from transaction %s", prefix, this);
 116
 117         openCursor = new DelegatingCursor(lookup.createCursor(prefix), prefix);
 118         return openCursor;
 119     }
 120
 121     @Override
 122     public synchronized FluentFuture<? extends @NonNull CommitInfo> commit() {
 123         Preconditions.checkState(!closed, "Transaction %s is already closed", identifier);
 124         Preconditions.checkState(openCursor == null, "Cannot submit transaction while there is a cursor open");
 125
 126         producer.transactionSubmitted(this);
 127         return future;
 128     }
 129
 130     void doSubmit(final Consumer<ShardedDOMDataTreeWriteTransaction> success,
 131             final BiConsumer<ShardedDOMDataTreeWriteTransaction, Throwable> failure) {
 132         LOG.debug("Readying tx {}", identifier);
 133
 134         final ListenableFuture<?> internalFuture;
 135         switch (transactions.size()) {
 136             case 0:
 137                 success.accept(this);
 138                 return;
 139             case 1: {
 140                 final DOMDataTreeShardWriteTransaction tx = transactions.values().iterator().next();
 141                 tx.ready();
 142                 internalFuture = tx.submit();
 143                 break;
 144             }
 145             default:
 146                 internalFuture = Futures.allAsList(transactions.values().stream().map(tx -> {
 147                     tx.ready();
 148                     return tx.submit();
 149                 }).collect(Collectors.toList()));
 150         }
 151
 152         Futures.addCallback(internalFuture, new FutureCallback<Object>() {
 153             @Override
 154             public void onSuccess(final Object result) {
 155                 success.accept(ShardedDOMDataTreeWriteTransaction.this);
 156             }
 157
 158             @Override
 159             public void onFailure(final Throwable exp) {
 160                 failure.accept(ShardedDOMDataTreeWriteTransaction.this, exp);
 161             }
 162         }, MoreExecutors.directExecutor());
 163     }
 164
 165     void onTransactionSuccess(final CommitInfo commitInfo) {
 166         future.set(commitInfo);
 167     }
 168
 169     void onTransactionFailure(final Throwable throwable) {
 170         future.setException(throwable);
 171     }
 172
 173     synchronized void cursorClosed() {
 174         openCursor = null;
 175     }
 176
 177     private class DelegatingCursor implements DOMDataTreeWriteCursor {
 178         private final Deque<PathArgument> currentArgs = new ArrayDeque<>();
 179         private final DOMDataTreeWriteCursor delegate;
 180         private final DOMDataTreeIdentifier rootPosition;
 181
 182         DelegatingCursor(final DOMDataTreeWriteCursor delegate, final DOMDataTreeIdentifier rootPosition) {
 183             this.delegate = Preconditions.checkNotNull(delegate);
 184             this.rootPosition = Preconditions.checkNotNull(rootPosition);
 185             currentArgs.addAll(rootPosition.getRootIdentifier().getPathArguments());
 186         }
 187
 188         @Override
 189         public void enter(@Nonnull final PathArgument child) {
 190             checkAvailable(child);
 191             delegate.enter(child);
 192             currentArgs.push(child);
 193         }
 194
 195         @Override
 196         public void enter(@Nonnull final PathArgument... path) {
 197             for (final PathArgument pathArgument : path) {
 198                 enter(pathArgument);
 199             }
 200         }
 201
 202         @Override
 203         public void enter(@Nonnull final Iterable<PathArgument> path) {
 204             for (final PathArgument pathArgument : path) {
 205                 enter(pathArgument);
 206             }
 207         }
 208
 209         @Override
 210         public void exit() {
 211             delegate.exit();
 212             currentArgs.pop();
 213         }
 214
 215         @Override
 216         public void exit(final int depth) {
 217             delegate.exit(depth);
 218             for (int i = 0; i < depth; i++) {
 219                 currentArgs.pop();
 220             }
 221         }
 222
 223         @Override
 224         public void close() {
 225             int depthEntered = currentArgs.size() - rootPosition.getRootIdentifier().getPathArguments().size();
 226             if (depthEntered > 0) {
 227                 // clean up existing modification cursor in case this tx will be reused for batching
 228                 delegate.exit(depthEntered);
 229             }
 230
 231             delegate.close();
 232             cursorClosed();
 233         }
 234
 235         @Override
 236         public void delete(final PathArgument child) {
 237             checkAvailable(child);
 238             delegate.delete(child);
 239         }
 240
 241         @Override
 242         public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
 243             checkAvailable(child);
 244             delegate.merge(child, data);
 245         }
 246
 247         @Override
 248         public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
 249             checkAvailable(child);
 250             delegate.write(child, data);
 251         }
 252
 253         void checkAvailable(final PathArgument child) {
 254             layout.checkAvailable(currentArgs, child);
 255         }
 256     }
 257
 258     ProducerLayout getLayout() {
 259         return layout;
 260     }
 261 }