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