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