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.ImmutableSet;
12 import com.google.common.util.concurrent.CheckedFuture;
13 import com.google.common.util.concurrent.Futures;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import java.util.Collection;
16 import java.util.Deque;
17 import java.util.HashMap;
18 import java.util.HashSet;
19 import java.util.LinkedList;
21 import java.util.Map.Entry;
23 import java.util.concurrent.Callable;
24 import java.util.concurrent.ExecutionException;
25 import java.util.concurrent.atomic.AtomicLong;
26 import javax.annotation.Nonnull;
27 import javax.annotation.concurrent.GuardedBy;
28 import javax.annotation.concurrent.NotThreadSafe;
29 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
30 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
31 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
32 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
33 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
34 import org.opendaylight.mdsal.dom.store.inmemory.DOMDataTreeShardProducer;
35 import org.opendaylight.mdsal.dom.store.inmemory.DOMDataTreeShardWriteTransaction;
36 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
37 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
38 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
43 final class ShardedDOMDataTreeWriteTransaction implements DOMDataTreeCursorAwareTransaction {
44 private static final Logger LOG = LoggerFactory.getLogger(ShardedDOMDataTreeWriteTransaction.class);
45 private static final AtomicLong COUNTER = new AtomicLong();
46 private final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> idToTransaction;
47 private final ShardedDOMDataTreeProducer producer;
48 private final String identifier;
49 private final Set<YangInstanceIdentifier> childBoundaries = new HashSet<>();
51 private boolean closed = false;
54 private DOMDataTreeWriteCursor openCursor;
56 ShardedDOMDataTreeWriteTransaction(final ShardedDOMDataTreeProducer producer,
57 final Map<DOMDataTreeIdentifier, DOMDataTreeShardProducer> idToProducer,
58 final Map<DOMDataTreeIdentifier, DOMDataTreeProducer> childProducers) {
59 this.producer = Preconditions.checkNotNull(producer);
60 idToTransaction = new HashMap<>();
61 Preconditions.checkNotNull(idToProducer).forEach((id, prod) -> idToTransaction.put(id, prod.createTransaction()));
62 this.identifier = "SHARDED-DOM-" + COUNTER.getAndIncrement();
63 childProducers.forEach((id, prod) -> childBoundaries.add(id.getRootIdentifier()));
66 // FIXME: use atomic operations
68 private DOMDataTreeShardWriteTransaction lookup(final DOMDataTreeIdentifier prefix) {
69 for (final Entry<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> e : idToTransaction.entrySet()) {
70 if (e.getKey().contains(prefix)) {
71 Preconditions.checkArgument(!producer.isDelegatedToChild(prefix),
72 "Path %s is delegated to child producer.",
77 throw new IllegalArgumentException(String.format("Path %s is not accessible from transaction %s", prefix, this));
81 public String getIdentifier() {
86 public synchronized boolean cancel() {
91 LOG.debug("Cancelling transaction {}", identifier);
92 if (openCursor != null) {
95 for (final DOMDataTreeShardWriteTransaction tx : ImmutableSet.copyOf(idToTransaction.values())) {
100 producer.cancelTransaction(this);
105 public synchronized DOMDataTreeWriteCursor createCursor(final DOMDataTreeIdentifier prefix) {
106 Preconditions.checkState(!closed, "Transaction is closed already");
107 Preconditions.checkState(openCursor == null, "There is still a cursor open");
108 final DOMDataTreeShardWriteTransaction lookup = lookup(prefix);
109 openCursor = new DelegatingCursor(lookup.createCursor(prefix), prefix);
114 public synchronized CheckedFuture<Void, TransactionCommitFailedException> submit() {
115 Preconditions.checkState(!closed, "Transaction %s is already closed", identifier);
116 Preconditions.checkState(openCursor == null, "Cannot submit transaction while there is a cursor open");
118 final Set<DOMDataTreeShardWriteTransaction> txns = ImmutableSet.copyOf(idToTransaction.values());
119 for (final DOMDataTreeShardWriteTransaction tx : txns) {
122 producer.transactionSubmitted(this);
124 return Futures.immediateCheckedFuture(new SubmitCoordinationTask(identifier, txns).call());
125 } catch (final TransactionCommitFailedException e) {
126 return Futures.immediateFailedCheckedFuture(e);
130 synchronized void cursorClosed() {
134 private class DelegatingCursor implements DOMDataTreeWriteCursor {
136 private final DOMDataTreeWriteCursor delegate;
137 private final Deque<PathArgument> path = new LinkedList<>();
139 public DelegatingCursor(final DOMDataTreeWriteCursor delegate, final DOMDataTreeIdentifier rootPosition) {
140 this.delegate = delegate;
141 path.addAll(rootPosition.getRootIdentifier().getPathArguments());
145 public void enter(@Nonnull final PathArgument child) {
146 checkAvailable(child);
148 delegate.enter(child);
152 public void enter(@Nonnull final PathArgument... path) {
153 for (final PathArgument pathArgument : path) {
159 public void enter(@Nonnull final Iterable<PathArgument> path) {
160 for (final PathArgument pathArgument : path) {
172 public void exit(final int depth) {
173 for (int i = 0; i < depth; i++) {
176 delegate.exit(depth);
180 public void close() {
186 public void delete(final PathArgument child) {
187 checkAvailable(child);
188 delegate.delete(child);
192 public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
193 checkAvailable(child);
194 delegate.merge(child, data);
198 public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
199 checkAvailable(child);
200 delegate.write(child, data);
203 void checkAvailable(final PathArgument child) {
205 final YangInstanceIdentifier yid = YangInstanceIdentifier.create(path);
206 childBoundaries.forEach(id -> {
207 if (id.contains(yid)) {
209 throw new IllegalArgumentException("Path {" + yid + "} is not available to this cursor since it's already claimed by a child producer");
216 private static class SubmitCoordinationTask implements Callable<Void> {
218 private static final Logger LOG = LoggerFactory.getLogger(SubmitCoordinationTask.class);
220 private final String identifier;
221 private final Collection<DOMDataTreeShardWriteTransaction> transactions;
223 SubmitCoordinationTask(final String identifier,
224 final Collection<DOMDataTreeShardWriteTransaction> transactions) {
225 this.identifier = identifier;
226 this.transactions = transactions;
230 public Void call() throws TransactionCommitFailedException {
233 LOG.debug("Producer {}, submit started", identifier);
237 } catch (final TransactionCommitFailedException e) {
238 LOG.warn("Failure while submitting transaction for producer {}", identifier, e);
244 void submitBlocking() throws TransactionCommitFailedException {
245 for (final ListenableFuture<?> commit : submitAll()) {
248 } catch (InterruptedException | ExecutionException e) {
249 throw new TransactionCommitFailedException("Submit failed", e);
254 private ListenableFuture<?>[] submitAll() {
255 final ListenableFuture<?>[] ops = new ListenableFuture<?>[transactions.size()];
257 for (final DOMDataTreeShardWriteTransaction tx : transactions) {
258 ops[i++] = tx.submit();