git.opendaylight Code Review - mdsal.git/blob

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