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