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