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