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