2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.mdsal.dom.broker;
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.SettableFuture;
17 import java.util.ArrayDeque;
18 import java.util.Deque;
20 import java.util.Map.Entry;
21 import java.util.concurrent.atomic.AtomicLong;
22 import java.util.function.BiConsumer;
23 import java.util.function.Consumer;
24 import java.util.stream.Collectors;
25 import javax.annotation.Nonnull;
26 import javax.annotation.concurrent.GuardedBy;
27 import javax.annotation.concurrent.NotThreadSafe;
28 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
29 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
30 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
31 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
32 import org.opendaylight.mdsal.dom.spi.shard.DOMDataTreeShardWriteTransaction;
33 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
34 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
35 import org.slf4j.Logger;
36 import org.slf4j.LoggerFactory;
39 final class ShardedDOMDataTreeWriteTransaction implements DOMDataTreeCursorAwareTransaction {
40 private static final Logger LOG = LoggerFactory.getLogger(ShardedDOMDataTreeWriteTransaction.class);
41 private static final TransactionCommitFailedExceptionMapper SUBMIT_FAILED_MAPPER =
42 TransactionCommitFailedExceptionMapper.create("submit");
43 private static final AtomicLong COUNTER = new AtomicLong();
45 private final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> transactions;
46 private final ShardedDOMDataTreeProducer producer;
47 private final ProducerLayout layout;
48 private final String identifier;
50 private final SettableFuture<Void> future = SettableFuture.create();
51 private final CheckedFuture<Void, TransactionCommitFailedException> submitFuture =
52 Futures.makeChecked(future, SUBMIT_FAILED_MAPPER);
55 private boolean closed = false;
58 private DOMDataTreeWriteCursor openCursor;
60 ShardedDOMDataTreeWriteTransaction(final ShardedDOMDataTreeProducer producer,
61 final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> transactions, final ProducerLayout layout) {
62 this.producer = Preconditions.checkNotNull(producer);
63 this.transactions = ImmutableMap.copyOf(transactions);
64 this.layout = Preconditions.checkNotNull(layout);
65 this.identifier = "SHARDED-DOM-" + COUNTER.getAndIncrement();
66 LOG.debug("Created new transaction {}", identifier);
69 private DOMDataTreeShardWriteTransaction lookup(final DOMDataTreeIdentifier prefix) {
70 final DOMDataTreeShardWriteTransaction fast = transactions.get(prefix);
75 LOG.debug("Prefix {} not found in available subtrees {}, fallback to slow path", prefix, transactions.keySet());
76 for (final Entry<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> e : transactions.entrySet()) {
77 if (e.getKey().contains(prefix)) {
86 public String getIdentifier() {
91 public synchronized boolean cancel() {
96 LOG.debug("Cancelling transaction {}", identifier);
97 if (openCursor != null) {
100 for (final DOMDataTreeShardWriteTransaction tx : transactions.values()) {
105 producer.cancelTransaction(this);
110 public synchronized DOMDataTreeWriteCursor createCursor(final DOMDataTreeIdentifier prefix) {
111 Preconditions.checkState(!closed, "Transaction is closed already");
112 Preconditions.checkState(openCursor == null, "There is still a cursor open");
113 Preconditions.checkArgument(!producer.isDelegatedToChild(prefix), "Path %s is delegated to child producer.",
116 final DOMDataTreeShardWriteTransaction lookup = lookup(prefix);
117 Preconditions.checkArgument(lookup != null, "Path %s is not accessible from transaction %s", prefix, this);
119 openCursor = new DelegatingCursor(lookup.createCursor(prefix), prefix);
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");
128 producer.transactionSubmitted(this);
132 void doSubmit(final Consumer<ShardedDOMDataTreeWriteTransaction> success,
133 final BiConsumer<ShardedDOMDataTreeWriteTransaction, Throwable> failure) {
134 LOG.debug("Readying tx {}", identifier);
136 final ListenableFuture<?> future;
137 switch (transactions.size()) {
139 success.accept(this);
142 final DOMDataTreeShardWriteTransaction tx = transactions.values().iterator().next();
144 future = tx.submit();
148 future = Futures.allAsList(transactions.values().stream().map(tx -> {
151 }).collect(Collectors.toList()));
154 Futures.addCallback(future, new FutureCallback<Object>() {
156 public void onSuccess(final Object result) {
157 success.accept(ShardedDOMDataTreeWriteTransaction.this);
161 public void onFailure(final Throwable exp) {
162 failure.accept(ShardedDOMDataTreeWriteTransaction.this, exp);
167 void onTransactionSuccess(final Void result) {
171 void onTransactionFailure(final Throwable throwable) {
172 future.setException(throwable);
175 synchronized void cursorClosed() {
179 private class DelegatingCursor implements DOMDataTreeWriteCursor {
180 private final Deque<PathArgument> path = new ArrayDeque<>();
181 private final DOMDataTreeWriteCursor delegate;
182 private final DOMDataTreeIdentifier rootPosition;
184 DelegatingCursor(final DOMDataTreeWriteCursor delegate, final DOMDataTreeIdentifier rootPosition) {
185 this.delegate = Preconditions.checkNotNull(delegate);
186 this.rootPosition = Preconditions.checkNotNull(rootPosition);
187 path.addAll(rootPosition.getRootIdentifier().getPathArguments());
191 public void enter(@Nonnull final PathArgument child) {
192 checkAvailable(child);
193 delegate.enter(child);
198 public void enter(@Nonnull final PathArgument... path) {
199 for (final PathArgument pathArgument : path) {
205 public void enter(@Nonnull final Iterable<PathArgument> path) {
206 for (final PathArgument pathArgument : path) {
218 public void exit(final int depth) {
219 delegate.exit(depth);
220 for (int i = 0; i < depth; i++) {
226 public void close() {
227 int depthEntered = path.size() - rootPosition.getRootIdentifier().getPathArguments().size();
228 if (depthEntered > 0) {
229 // clean up existing modification cursor in case this tx will be reused for batching
230 delegate.exit(depthEntered);
238 public void delete(final PathArgument child) {
239 checkAvailable(child);
240 delegate.delete(child);
244 public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
245 checkAvailable(child);
246 delegate.merge(child, data);
250 public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
251 checkAvailable(child);
252 delegate.write(child, data);
255 void checkAvailable(final PathArgument child) {
256 layout.checkAvailable(path, child);
260 ProducerLayout getLayout() {