package org.opendaylight.mdsal.dom.broker;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.FluentFuture;
+import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import java.util.Collection;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
+import java.util.ArrayDeque;
import java.util.Deque;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;
-import javax.annotation.Nonnull;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.NotThreadSafe;
-import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
+import org.eclipse.jdt.annotation.NonNull;
+import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
-import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
-import org.opendaylight.mdsal.dom.store.inmemory.DOMDataTreeShardProducer;
-import org.opendaylight.mdsal.dom.store.inmemory.DOMDataTreeShardWriteTransaction;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.mdsal.dom.spi.shard.DOMDataTreeShardWriteTransaction;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.slf4j.Logger;
final class ShardedDOMDataTreeWriteTransaction implements DOMDataTreeCursorAwareTransaction {
private static final Logger LOG = LoggerFactory.getLogger(ShardedDOMDataTreeWriteTransaction.class);
private static final AtomicLong COUNTER = new AtomicLong();
- private final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> idToTransaction;
+
+ private final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> transactions;
private final ShardedDOMDataTreeProducer producer;
+ private final ProducerLayout layout;
private final String identifier;
- private final Set<YangInstanceIdentifier> childBoundaries = new HashSet<>();
+
+ private final SettableFuture<CommitInfo> future = SettableFuture.create();
+
@GuardedBy("this")
private boolean closed = false;
private DOMDataTreeWriteCursor openCursor;
ShardedDOMDataTreeWriteTransaction(final ShardedDOMDataTreeProducer producer,
- final Map<DOMDataTreeIdentifier, DOMDataTreeShardProducer> idToProducer,
- final Map<DOMDataTreeIdentifier, DOMDataTreeProducer> childProducers) {
+ final Map<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> transactions, final ProducerLayout layout) {
this.producer = Preconditions.checkNotNull(producer);
- idToTransaction = new HashMap<>();
- Preconditions.checkNotNull(idToProducer).forEach((id, prod) -> idToTransaction.put(id, prod.createTransaction()));
+ this.transactions = ImmutableMap.copyOf(transactions);
+ this.layout = Preconditions.checkNotNull(layout);
this.identifier = "SHARDED-DOM-" + COUNTER.getAndIncrement();
- childProducers.forEach((id, prod) -> childBoundaries.add(id.getRootIdentifier()));
+ LOG.debug("Created new transaction {}", identifier);
}
- // FIXME: use atomic operations
- @GuardedBy("this")
private DOMDataTreeShardWriteTransaction lookup(final DOMDataTreeIdentifier prefix) {
- for (final Entry<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> e : idToTransaction.entrySet()) {
+ final DOMDataTreeShardWriteTransaction fast = transactions.get(prefix);
+ if (fast != null) {
+ return fast;
+ }
+
+ LOG.debug("Prefix {} not found in available subtrees {}, fallback to slow path", prefix, transactions.keySet());
+ for (final Entry<DOMDataTreeIdentifier, DOMDataTreeShardWriteTransaction> e : transactions.entrySet()) {
if (e.getKey().contains(prefix)) {
- Preconditions.checkArgument(!producer.isDelegatedToChild(prefix),
- "Path %s is delegated to child producer.",
- prefix);
return e.getValue();
}
}
- throw new IllegalArgumentException(String.format("Path %s is not accessible from transaction %s", prefix, this));
+
+ return null;
}
@Override
if (openCursor != null) {
openCursor.close();
}
- for (final DOMDataTreeShardWriteTransaction tx : ImmutableSet.copyOf(idToTransaction.values())) {
+ for (final DOMDataTreeShardWriteTransaction tx : transactions.values()) {
tx.close();
}
public synchronized DOMDataTreeWriteCursor createCursor(final DOMDataTreeIdentifier prefix) {
Preconditions.checkState(!closed, "Transaction is closed already");
Preconditions.checkState(openCursor == null, "There is still a cursor open");
+ Preconditions.checkArgument(!producer.isDelegatedToChild(prefix), "Path %s is delegated to child producer.",
+ prefix);
+
final DOMDataTreeShardWriteTransaction lookup = lookup(prefix);
+ Preconditions.checkArgument(lookup != null, "Path %s is not accessible from transaction %s", prefix, this);
+
openCursor = new DelegatingCursor(lookup.createCursor(prefix), prefix);
return openCursor;
}
@Override
- public synchronized CheckedFuture<Void, TransactionCommitFailedException> submit() {
+ public synchronized FluentFuture<? extends @NonNull CommitInfo> commit() {
Preconditions.checkState(!closed, "Transaction %s is already closed", identifier);
Preconditions.checkState(openCursor == null, "Cannot submit transaction while there is a cursor open");
- final Set<DOMDataTreeShardWriteTransaction> txns = ImmutableSet.copyOf(idToTransaction.values());
- for (final DOMDataTreeShardWriteTransaction tx : txns) {
- tx.ready();
- }
producer.transactionSubmitted(this);
- try {
- return Futures.immediateCheckedFuture(new SubmitCoordinationTask(identifier, txns).call());
- } catch (final TransactionCommitFailedException e) {
- return Futures.immediateFailedCheckedFuture(e);
+ return future;
+ }
+
+ void doSubmit(final Consumer<ShardedDOMDataTreeWriteTransaction> success,
+ final BiConsumer<ShardedDOMDataTreeWriteTransaction, Throwable> failure) {
+ LOG.debug("Readying tx {}", identifier);
+
+ final ListenableFuture<?> internalFuture;
+ switch (transactions.size()) {
+ case 0:
+ success.accept(this);
+ return;
+ case 1: {
+ final DOMDataTreeShardWriteTransaction tx = transactions.values().iterator().next();
+ tx.ready();
+ internalFuture = tx.submit();
+ break;
+ }
+ default:
+ internalFuture = Futures.allAsList(transactions.values().stream().map(tx -> {
+ tx.ready();
+ return tx.submit();
+ }).collect(Collectors.toList()));
}
+
+ Futures.addCallback(internalFuture, new FutureCallback<Object>() {
+ @Override
+ public void onSuccess(final Object result) {
+ success.accept(ShardedDOMDataTreeWriteTransaction.this);
+ }
+
+ @Override
+ public void onFailure(final Throwable exp) {
+ failure.accept(ShardedDOMDataTreeWriteTransaction.this, exp);
+ }
+ }, MoreExecutors.directExecutor());
+ }
+
+ void onTransactionSuccess(final CommitInfo commitInfo) {
+ future.set(commitInfo);
+ }
+
+ void onTransactionFailure(final Throwable throwable) {
+ future.setException(throwable);
}
synchronized void cursorClosed() {
}
private class DelegatingCursor implements DOMDataTreeWriteCursor {
-
+ private final Deque<PathArgument> currentArgs = new ArrayDeque<>();
private final DOMDataTreeWriteCursor delegate;
- private final Deque<PathArgument> path = new LinkedList<>();
+ private final DOMDataTreeIdentifier rootPosition;
- public DelegatingCursor(final DOMDataTreeWriteCursor delegate, final DOMDataTreeIdentifier rootPosition) {
- this.delegate = delegate;
- path.addAll(rootPosition.getRootIdentifier().getPathArguments());
+ DelegatingCursor(final DOMDataTreeWriteCursor delegate, final DOMDataTreeIdentifier rootPosition) {
+ this.delegate = Preconditions.checkNotNull(delegate);
+ this.rootPosition = Preconditions.checkNotNull(rootPosition);
+ currentArgs.addAll(rootPosition.getRootIdentifier().getPathArguments());
}
@Override
- public void enter(@Nonnull final PathArgument child) {
+ public void enter(final PathArgument child) {
checkAvailable(child);
- path.push(child);
delegate.enter(child);
+ currentArgs.push(child);
}
@Override
- public void enter(@Nonnull final PathArgument... path) {
+ public void enter(final PathArgument... path) {
for (final PathArgument pathArgument : path) {
enter(pathArgument);
}
}
@Override
- public void enter(
@Nonnull final Iterable<PathArgument> path) {
+ public void enter(final Iterable<PathArgument> path) {
for (final PathArgument pathArgument : path) {
enter(pathArgument);
}
@Override
public void exit() {
- path.pop();
delegate.exit();
+ currentArgs.pop();
}
@Override
public void exit(final int depth) {
+ delegate.exit(depth);
for (int i = 0; i < depth; i++) {
- path.pop();
+ currentArgs.pop();
}
- delegate.exit(depth);
}
@Override
public void close() {
+ int depthEntered = currentArgs.size() - rootPosition.getRootIdentifier().getPathArguments().size();
+ if (depthEntered > 0) {
+ // clean up existing modification cursor in case this tx will be reused for batching
+ delegate.exit(depthEntered);
+ }
+
delegate.close();
cursorClosed();
}
}
void checkAvailable(final PathArgument child) {
- path.add(child);
- final YangInstanceIdentifier yid = YangInstanceIdentifier.create(path);
- childBoundaries.forEach(id -> {
- if (id.contains(yid)) {
- path.removeLast();
- throw new IllegalArgumentException("Path {" + yid + "} is not available to this cursor since it's already claimed by a child producer");
- }
- });
- path.removeLast();
+ layout.checkAvailable(currentArgs, child);
}
}
- private static class SubmitCoordinationTask implements Callable<Void> {
-
- private static final Logger LOG = LoggerFactory.getLogger(SubmitCoordinationTask.class);
-
- private final String identifier;
- private final Collection<DOMDataTreeShardWriteTransaction> transactions;
-
- SubmitCoordinationTask(final String identifier,
- final Collection<DOMDataTreeShardWriteTransaction> transactions) {
- this.identifier = identifier;
- this.transactions = transactions;
- }
-
- @Override
- public Void call() throws TransactionCommitFailedException {
-
- try {
- LOG.debug("Producer {}, submit started", identifier);
- submitBlocking();
-
- return null;
- } catch (final TransactionCommitFailedException e) {
- LOG.warn("Failure while submitting transaction for producer {}", identifier, e);
- //FIXME abort here
- throw e;
- }
- }
-
- void submitBlocking() throws TransactionCommitFailedException {
- for (final ListenableFuture<?> commit : submitAll()) {
- try {
- commit.get();
- } catch (InterruptedException | ExecutionException e) {
- throw new TransactionCommitFailedException("Submit failed", e);
- }
- }
- }
-
- private ListenableFuture<?>[] submitAll() {
- final ListenableFuture<?>[] ops = new ListenableFuture<?>[transactions.size()];
- int i = 0;
- for (final DOMDataTreeShardWriteTransaction tx : transactions) {
- ops[i++] = tx.submit();
- }
- return ops;
- }
+ ProducerLayout getLayout() {
+ return layout;
}
}
Code Review / mdsal.git / blobdiff