X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardDataTree.java;h=6e4ee7d9e13f7dd4899c24833f65538eb255a42f;hp=9aca517bb110d2f28381167962f2ad8f3de288af;hb=cd648b7a1662577cd998670f5b39bc839e0f9070;hpb=20a32e6459fd1e27e7669bf1ebc7742b96787b94
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java
index 9aca517bb1..6e4ee7d9e1 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardDataTree.java
@@ -11,7 +11,6 @@ import akka.actor.ActorRef;
import akka.util.Timeout;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
-import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.google.common.base.Verify;
@@ -20,26 +19,30 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMap.Builder;
import com.google.common.collect.Iterables;
import com.google.common.primitives.UnsignedLong;
+import com.google.common.util.concurrent.FutureCallback;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.File;
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.Deque;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.OptionalLong;
import java.util.Queue;
-import java.util.concurrent.ExecutionException;
+import java.util.SortedSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.UnaryOperator;
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-import javax.annotation.concurrent.NotThreadSafe;
+import org.eclipse.jdt.annotation.NonNull;
+import org.eclipse.jdt.annotation.Nullable;
import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.DataTreeCohortActorRegistry.CohortRegistryCommand;
@@ -58,11 +61,9 @@ import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnap
import org.opendaylight.controller.cluster.datastore.utils.DataTreeModificationOutput;
import org.opendaylight.controller.cluster.datastore.utils.PruningDataTreeModification;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
-import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
-import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
+import org.opendaylight.mdsal.common.api.OptimisticLockFailedException;
+import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
+import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
import org.opendaylight.yangtools.concepts.Identifier;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
@@ -79,20 +80,19 @@ import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeTip;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
+import org.opendaylight.yangtools.yang.data.util.DataSchemaContextTree;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.concurrent.duration.Duration;
+import scala.concurrent.duration.FiniteDuration;
/**
- * Internal shard state, similar to a DOMStore, but optimized for use in the actor system,
- * e.g. it does not expose public interfaces and assumes it is only ever called from a
- * single thread.
+ * Internal shard state, similar to a DOMStore, but optimized for use in the actor system, e.g. it does not expose
+ * public interfaces and assumes it is only ever called from a single thread.
*
*
- * This class is not part of the API contract and is subject to change at any time.
+ * This class is not part of the API contract and is subject to change at any time. It is NOT thread-safe.
*/
-@NotThreadSafe
public class ShardDataTree extends ShardDataTreeTransactionParent {
private static final class CommitEntry {
final SimpleShardDataTreeCohort cohort;
@@ -102,9 +102,14 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
this.cohort = Preconditions.checkNotNull(cohort);
lastAccess = now;
}
+
+ @Override
+ public String toString() {
+ return "CommitEntry [tx=" + cohort.getIdentifier() + ", state=" + cohort.getState() + "]";
+ }
}
- private static final Timeout COMMIT_STEP_TIMEOUT = new Timeout(Duration.create(5, TimeUnit.SECONDS));
+ private static final Timeout COMMIT_STEP_TIMEOUT = new Timeout(FiniteDuration.create(5, TimeUnit.SECONDS));
private static final Logger LOG = LoggerFactory.getLogger(ShardDataTree.class);
/**
@@ -117,7 +122,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
private final Map transactionChains = new HashMap<>();
private final DataTreeCohortActorRegistry cohortRegistry = new DataTreeCohortActorRegistry();
- private final Queue pendingTransactions = new ArrayDeque<>();
+ private final Deque pendingTransactions = new ArrayDeque<>();
private final Queue pendingCommits = new ArrayDeque<>();
private final Queue pendingFinishCommits = new ArrayDeque<>();
@@ -127,7 +132,6 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
private final Map replicationCallbacks = new HashMap<>();
private final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher;
- private final ShardDataChangeListenerPublisher dataChangeListenerPublisher;
private final Collection> metadata;
private final DataTree dataTree;
private final String logContext;
@@ -143,19 +147,19 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
private DataTreeTip tip;
private SchemaContext schemaContext;
+ private DataSchemaContextTree dataSchemaContext;
private int currentTransactionBatch;
ShardDataTree(final Shard shard, final SchemaContext schemaContext, final DataTree dataTree,
final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher,
- final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext,
+ final String logContext,
final ShardDataTreeMetadata>... metadata) {
this.dataTree = Preconditions.checkNotNull(dataTree);
updateSchemaContext(schemaContext);
this.shard = Preconditions.checkNotNull(shard);
this.treeChangeListenerPublisher = Preconditions.checkNotNull(treeChangeListenerPublisher);
- this.dataChangeListenerPublisher = Preconditions.checkNotNull(dataChangeListenerPublisher);
this.logContext = Preconditions.checkNotNull(logContext);
this.metadata = ImmutableList.copyOf(metadata);
tip = dataTree;
@@ -164,10 +168,9 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType,
final YangInstanceIdentifier root,
final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher,
- final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext,
+ final String logContext,
final ShardDataTreeMetadata>... metadata) {
- this(shard, schemaContext, createDataTree(treeType, root), treeChangeListenerPublisher,
- dataChangeListenerPublisher, logContext, metadata);
+ this(shard, schemaContext, createDataTree(treeType, root), treeChangeListenerPublisher, logContext, metadata);
}
private static DataTree createDataTree(final TreeType treeType, final YangInstanceIdentifier root) {
@@ -182,8 +185,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
@VisibleForTesting
public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType) {
this(shard, schemaContext, treeType, YangInstanceIdentifier.EMPTY,
- new DefaultShardDataTreeChangeListenerPublisher(""),
- new DefaultShardDataChangeListenerPublisher(""), "");
+ new DefaultShardDataTreeChangeListenerPublisher(""), "");
}
final String logContext() {
@@ -205,6 +207,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
void updateSchemaContext(final SchemaContext newSchemaContext) {
dataTree.setSchemaContext(newSchemaContext);
this.schemaContext = Preconditions.checkNotNull(newSchemaContext);
+ this.dataSchemaContext = DataSchemaContextTree.from(newSchemaContext);
}
void resetTransactionBatch() {
@@ -216,7 +219,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
*
* @return A state snapshot
*/
- @Nonnull ShardDataTreeSnapshot takeStateSnapshot() {
+ @NonNull ShardDataTreeSnapshot takeStateSnapshot() {
final NormalizedNode, ?> rootNode = dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY).get();
final Builder>, ShardDataTreeSnapshotMetadata>> metaBuilder =
ImmutableMap.builder();
@@ -235,7 +238,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
return !pendingTransactions.isEmpty() || !pendingCommits.isEmpty() || !pendingFinishCommits.isEmpty();
}
- private void applySnapshot(@Nonnull final ShardDataTreeSnapshot snapshot,
+ private void applySnapshot(final @NonNull ShardDataTreeSnapshot snapshot,
final UnaryOperator wrapper) throws DataValidationFailedException {
final Stopwatch elapsed = Stopwatch.createStarted();
@@ -263,7 +266,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
// delete everything first
mod.delete(YangInstanceIdentifier.EMPTY);
- final java.util.Optional> maybeNode = snapshot.getRootNode();
+ final Optional> maybeNode = snapshot.getRootNode();
if (maybeNode.isPresent()) {
// Add everything from the remote node back
mod.write(YangInstanceIdentifier.EMPTY, maybeNode.get());
@@ -286,12 +289,12 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
* @param snapshot Snapshot that needs to be applied
* @throws DataValidationFailedException when the snapshot fails to apply
*/
- void applySnapshot(@Nonnull final ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
+ void applySnapshot(final @NonNull ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
applySnapshot(snapshot, UnaryOperator.identity());
}
private PruningDataTreeModification wrapWithPruning(final DataTreeModification delegate) {
- return new PruningDataTreeModification(delegate, dataTree, schemaContext);
+ return new PruningDataTreeModification(delegate, dataTree, dataSchemaContext);
}
private static DataTreeModification unwrap(final DataTreeModification modification) {
@@ -308,12 +311,12 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
* @param snapshot Snapshot that needs to be applied
* @throws DataValidationFailedException when the snapshot fails to apply
*/
- void applyRecoverySnapshot(@Nonnull final ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
+ void applyRecoverySnapshot(final @NonNull ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
applySnapshot(snapshot, this::wrapWithPruning);
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private void applyRecoveryCandidate(final DataTreeCandidate candidate) throws DataValidationFailedException {
+ private void applyRecoveryCandidate(final DataTreeCandidate candidate) {
final PruningDataTreeModification mod = wrapWithPruning(dataTree.takeSnapshot().newModification());
DataTreeCandidates.applyToModification(mod, candidate);
mod.ready();
@@ -342,7 +345,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
* @throws IOException when the snapshot fails to deserialize
* @throws DataValidationFailedException when the snapshot fails to apply
*/
- void applyRecoveryPayload(@Nonnull final Payload payload) throws IOException, DataValidationFailedException {
+ void applyRecoveryPayload(final @NonNull Payload payload) throws IOException {
if (payload instanceof CommitTransactionPayload) {
final Entry e =
((CommitTransactionPayload) payload).getCandidate();
@@ -363,7 +366,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
}
}
- private void applyReplicatedCandidate(final Identifier identifier, final DataTreeCandidate foreign)
+ private void applyReplicatedCandidate(final TransactionIdentifier identifier, final DataTreeCandidate foreign)
throws DataValidationFailedException {
LOG.debug("{}: Applying foreign transaction {}", logContext, identifier);
@@ -376,6 +379,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
final DataTreeCandidate candidate = dataTree.prepare(mod);
dataTree.commit(candidate);
+ allMetadataCommittedTransaction(identifier);
notifyListeners(candidate);
}
@@ -402,18 +406,14 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
* pre-Boron state -- which limits the number of options here.
*/
if (payload instanceof CommitTransactionPayload) {
- final TransactionIdentifier txId;
if (identifier == null) {
final Entry e =
((CommitTransactionPayload) payload).getCandidate();
- txId = e.getKey();
- applyReplicatedCandidate(txId, e.getValue());
+ applyReplicatedCandidate(e.getKey(), e.getValue());
} else {
Verify.verify(identifier instanceof TransactionIdentifier);
- txId = (TransactionIdentifier) identifier;
- payloadReplicationComplete(txId);
+ payloadReplicationComplete((TransactionIdentifier) identifier);
}
- allMetadataCommittedTransaction(txId);
} else if (payload instanceof AbortTransactionPayload) {
if (identifier != null) {
payloadReplicationComplete((AbortTransactionPayload) payload);
@@ -444,7 +444,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
}
}
- private void replicatePayload(final Identifier id, final Payload payload, @Nullable final Runnable callback) {
+ private void replicatePayload(final Identifier id, final Payload payload, final @Nullable Runnable callback) {
if (callback != null) {
replicationCallbacks.put(payload, callback);
}
@@ -465,12 +465,14 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
final CommitEntry current = pendingFinishCommits.peek();
if (current == null) {
LOG.warn("{}: No outstanding transactions, ignoring consensus on transaction {}", logContext, txId);
+ allMetadataCommittedTransaction(txId);
return;
}
if (!current.cohort.getIdentifier().equals(txId)) {
LOG.debug("{}: Head of pendingFinishCommits queue is {}, ignoring consensus on transaction {}", logContext,
current.cohort.getIdentifier(), txId);
+ allMetadataCommittedTransaction(txId);
return;
}
@@ -531,12 +533,13 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
}
ShardDataTreeTransactionChain ensureTransactionChain(final LocalHistoryIdentifier historyId,
- @Nullable final Runnable callback) {
+ final @Nullable Runnable callback) {
ShardDataTreeTransactionChain chain = transactionChains.get(historyId);
if (chain == null) {
chain = new ShardDataTreeTransactionChain(historyId, this);
transactionChains.put(historyId, chain);
- replicatePayload(historyId, CreateLocalHistoryPayload.create(historyId), callback);
+ replicatePayload(historyId, CreateLocalHistoryPayload.create(
+ historyId, shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity()), callback);
} else if (callback != null) {
callback.run();
}
@@ -545,6 +548,8 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
}
ReadOnlyShardDataTreeTransaction newReadOnlyTransaction(final TransactionIdentifier txId) {
+ shard.getShardMBean().incrementReadOnlyTransactionCount();
+
if (txId.getHistoryId().getHistoryId() == 0) {
return new ReadOnlyShardDataTreeTransaction(this, txId, dataTree.takeSnapshot());
}
@@ -553,6 +558,8 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
}
ReadWriteShardDataTreeTransaction newReadWriteTransaction(final TransactionIdentifier txId) {
+ shard.getShardMBean().incrementReadWriteTransactionCount();
+
if (txId.getHistoryId().getHistoryId() == 0) {
return new ReadWriteShardDataTreeTransaction(ShardDataTree.this, txId, dataTree.takeSnapshot()
.newModification());
@@ -564,7 +571,6 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
@VisibleForTesting
public void notifyListeners(final DataTreeCandidate candidate) {
treeChangeListenerPublisher.publishChanges(candidate);
- dataChangeListenerPublisher.publishChanges(candidate);
}
/**
@@ -586,18 +592,34 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
* @param id History identifier
* @param callback Callback to invoke upon completion, may be null
*/
- void closeTransactionChain(final LocalHistoryIdentifier id, @Nullable final Runnable callback) {
+ void closeTransactionChain(final LocalHistoryIdentifier id, final @Nullable Runnable callback) {
+ if (commonCloseTransactionChain(id, callback)) {
+ replicatePayload(id, CloseLocalHistoryPayload.create(id,
+ shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity()), callback);
+ }
+ }
+
+ /**
+ * Close a single transaction chain which is received through ask-based protocol. It does not keep a commit record.
+ *
+ * @param id History identifier
+ */
+ void closeTransactionChain(final LocalHistoryIdentifier id) {
+ commonCloseTransactionChain(id, null);
+ }
+
+ private boolean commonCloseTransactionChain(final LocalHistoryIdentifier id, final @Nullable Runnable callback) {
final ShardDataTreeTransactionChain chain = transactionChains.get(id);
if (chain == null) {
LOG.debug("{}: Closing non-existent transaction chain {}", logContext, id);
if (callback != null) {
callback.run();
}
- return;
+ return false;
}
chain.close();
- replicatePayload(id, CloseLocalHistoryPayload.create(id), callback);
+ return true;
}
/**
@@ -606,7 +628,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
* @param id History identifier
* @param callback Callback to invoke upon completion, may be null
*/
- void purgeTransactionChain(final LocalHistoryIdentifier id, @Nullable final Runnable callback) {
+ void purgeTransactionChain(final LocalHistoryIdentifier id, final @Nullable Runnable callback) {
final ShardDataTreeTransactionChain chain = transactionChains.remove(id);
if (chain == null) {
LOG.debug("{}: Purging non-existent transaction chain {}", logContext, id);
@@ -616,22 +638,13 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
return;
}
- replicatePayload(id, PurgeLocalHistoryPayload.create(id), callback);
- }
-
- void registerDataChangeListener(final YangInstanceIdentifier path,
- final AsyncDataChangeListener> listener,
- final DataChangeScope scope, final Optional initialState,
- final Consumer>>>
- onRegistration) {
- dataChangeListenerPublisher.registerDataChangeListener(path, listener, scope, initialState, onRegistration);
+ replicatePayload(id, PurgeLocalHistoryPayload.create(
+ id, shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity()), callback);
}
Optional readCurrentData() {
- final java.util.Optional> currentState =
- dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY);
- return currentState.isPresent() ? Optional.of(DataTreeCandidates.fromNormalizedNode(
- YangInstanceIdentifier.EMPTY, currentState.get())) : Optional.absent();
+ return dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY)
+ .map(state -> DataTreeCandidates.fromNormalizedNode(YangInstanceIdentifier.EMPTY, state));
}
public void registerTreeChangeListener(final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener,
@@ -648,7 +661,8 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
void abortTransaction(final AbstractShardDataTreeTransaction> transaction, final Runnable callback) {
final TransactionIdentifier id = transaction.getIdentifier();
LOG.debug("{}: aborting transaction {}", logContext, id);
- replicatePayload(id, AbortTransactionPayload.create(id), callback);
+ replicatePayload(id, AbortTransactionPayload.create(
+ id, shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity()), callback);
}
@Override
@@ -658,20 +672,25 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
}
@Override
- ShardDataTreeCohort finishTransaction(final ReadWriteShardDataTreeTransaction transaction) {
+ ShardDataTreeCohort finishTransaction(final ReadWriteShardDataTreeTransaction transaction,
+ final Optional> participatingShardNames) {
final DataTreeModification snapshot = transaction.getSnapshot();
+ final TransactionIdentifier id = transaction.getIdentifier();
+ LOG.debug("{}: readying transaction {}", logContext, id);
snapshot.ready();
+ LOG.debug("{}: transaction {} ready", logContext, id);
- return createReadyCohort(transaction.getIdentifier(), snapshot);
+ return createReadyCohort(transaction.getIdentifier(), snapshot, participatingShardNames);
}
void purgeTransaction(final TransactionIdentifier id, final Runnable callback) {
LOG.debug("{}: purging transaction {}", logContext, id);
- replicatePayload(id, PurgeTransactionPayload.create(id), callback);
+ replicatePayload(id, PurgeTransactionPayload.create(
+ id, shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity()), callback);
}
public Optional> readNode(final YangInstanceIdentifier path) {
- return Optional.fromJavaUtil(dataTree.takeSnapshot().readNode(path));
+ return dataTree.takeSnapshot().readNode(path);
}
DataTreeSnapshot takeSnapshot() {
@@ -737,16 +756,16 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
} catch (ConflictingModificationAppliedException e) {
LOG.warn("{}: Store Tx {}: Conflicting modification for path {}.", logContext, cohort.getIdentifier(),
e.getPath());
- cause = new OptimisticLockFailedException("Optimistic lock failed.", e);
+ cause = new OptimisticLockFailedException("Optimistic lock failed for path " + e.getPath(), e);
} catch (DataValidationFailedException e) {
LOG.warn("{}: Store Tx {}: Data validation failed for path {}.", logContext, cohort.getIdentifier(),
e.getPath(), e);
// For debugging purposes, allow dumping of the modification. Coupled with the above
// precondition log, it should allow us to understand what went on.
- LOG.debug("{}: Store Tx {}: modifications: {} tree: {}", cohort.getIdentifier(), modification,
- dataTree);
- cause = new TransactionCommitFailedException("Data did not pass validation.", e);
+ LOG.debug("{}: Store Tx {}: modifications: {}", logContext, cohort.getIdentifier(), modification);
+ LOG.trace("{}: Current tree: {}", logContext, dataTree);
+ cause = new TransactionCommitFailedException("Data did not pass validation for path " + e.getPath(), e);
} catch (Exception e) {
LOG.warn("{}: Unexpected failure in validation phase", logContext, e);
cause = e;
@@ -801,14 +820,108 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
return;
}
if (!cohort.equals(head.cohort)) {
- LOG.debug("{}: Transaction {} scheduled for canCommit step", logContext, cohort.getIdentifier());
- return;
+ // The tx isn't at the head of the queue so we can't start canCommit at this point. Here we check if this
+ // tx should be moved ahead of other tx's in the READY state in the pendingTransactions queue. If this tx
+ // has other participating shards, it could deadlock with other tx's accessing the same shards
+ // depending on the order the tx's are readied on each shard
+ // (see https://jira.opendaylight.org/browse/CONTROLLER-1836). Therefore, if the preceding participating
+ // shard names for a preceding pending tx, call it A, in the queue matches that of this tx, then this tx
+ // is allowed to be moved ahead of tx A in the queue so it is processed first to avoid potential deadlock
+ // if tx A is behind this tx in the pendingTransactions queue for a preceding shard. In other words, since
+ // canCommmit for this tx was requested before tx A, honor that request. If this tx is moved to the head of
+ // the queue as a result, then proceed with canCommit.
+
+ Collection precedingShardNames = extractPrecedingShardNames(cohort.getParticipatingShardNames());
+ if (precedingShardNames.isEmpty()) {
+ LOG.debug("{}: Tx {} is scheduled for canCommit step", logContext, cohort.getIdentifier());
+ return;
+ }
+
+ LOG.debug("{}: Evaluating tx {} for canCommit - preceding participating shard names {}",
+ logContext, cohort.getIdentifier(), precedingShardNames);
+ final Iterator iter = pendingTransactions.iterator();
+ int index = -1;
+ int moveToIndex = -1;
+ while (iter.hasNext()) {
+ final CommitEntry entry = iter.next();
+ ++index;
+
+ if (cohort.equals(entry.cohort)) {
+ if (moveToIndex < 0) {
+ LOG.debug("{}: Not moving tx {} - cannot proceed with canCommit",
+ logContext, cohort.getIdentifier());
+ return;
+ }
+
+ LOG.debug("{}: Moving {} to index {} in the pendingTransactions queue",
+ logContext, cohort.getIdentifier(), moveToIndex);
+ iter.remove();
+ insertEntry(pendingTransactions, entry, moveToIndex);
+
+ if (!cohort.equals(pendingTransactions.peek().cohort)) {
+ LOG.debug("{}: Tx {} is not at the head of the queue - cannot proceed with canCommit",
+ logContext, cohort.getIdentifier());
+ return;
+ }
+
+ LOG.debug("{}: Tx {} is now at the head of the queue - proceeding with canCommit",
+ logContext, cohort.getIdentifier());
+ break;
+ }
+
+ if (entry.cohort.getState() != State.READY) {
+ LOG.debug("{}: Skipping pending transaction {} in state {}",
+ logContext, entry.cohort.getIdentifier(), entry.cohort.getState());
+ continue;
+ }
+
+ final Collection pendingPrecedingShardNames = extractPrecedingShardNames(
+ entry.cohort.getParticipatingShardNames());
+
+ if (precedingShardNames.equals(pendingPrecedingShardNames)) {
+ if (moveToIndex < 0) {
+ LOG.debug("{}: Preceding shard names {} for pending tx {} match - saving moveToIndex {}",
+ logContext, pendingPrecedingShardNames, entry.cohort.getIdentifier(), index);
+ moveToIndex = index;
+ } else {
+ LOG.debug(
+ "{}: Preceding shard names {} for pending tx {} match but moveToIndex already set to {}",
+ logContext, pendingPrecedingShardNames, entry.cohort.getIdentifier(), moveToIndex);
+ }
+ } else {
+ LOG.debug("{}: Preceding shard names {} for pending tx {} differ - skipping",
+ logContext, pendingPrecedingShardNames, entry.cohort.getIdentifier());
+ }
+ }
}
processNextPendingTransaction();
}
- private void failPreCommit(final Exception cause) {
+ private static void insertEntry(final Deque queue, final CommitEntry entry, final int atIndex) {
+ if (atIndex == 0) {
+ queue.addFirst(entry);
+ return;
+ }
+
+ LOG.trace("Inserting into Deque at index {}", atIndex);
+
+ Deque tempStack = new ArrayDeque<>(atIndex);
+ for (int i = 0; i < atIndex; i++) {
+ tempStack.push(queue.poll());
+ }
+
+ queue.addFirst(entry);
+
+ tempStack.forEach(queue::addFirst);
+ }
+
+ private Collection extractPrecedingShardNames(final Optional> participatingShardNames) {
+ return participatingShardNames.map((Function, Collection>)
+ set -> set.headSet(shard.getShardName())).orElse(Collections.emptyList());
+ }
+
+ private void failPreCommit(final Throwable cause) {
shard.getShardMBean().incrementFailedTransactionsCount();
pendingTransactions.poll().cohort.failedPreCommit(cause);
processNextPendingTransaction();
@@ -822,30 +935,41 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
final SimpleShardDataTreeCohort current = entry.cohort;
Verify.verify(cohort.equals(current), "Attempted to pre-commit %s while %s is pending", cohort, current);
- LOG.debug("{}: Preparing transaction {}", logContext, current.getIdentifier());
+ final TransactionIdentifier currentId = current.getIdentifier();
+ LOG.debug("{}: Preparing transaction {}", logContext, currentId);
final DataTreeCandidateTip candidate;
try {
candidate = tip.prepare(cohort.getDataTreeModification());
- cohort.userPreCommit(candidate);
- } catch (ExecutionException | TimeoutException | RuntimeException e) {
+ LOG.debug("{}: Transaction {} candidate ready", logContext, currentId);
+ } catch (RuntimeException e) {
failPreCommit(e);
return;
}
- // Set the tip of the data tree.
- tip = Verify.verifyNotNull(candidate);
+ cohort.userPreCommit(candidate, new FutureCallback() {
+ @Override
+ public void onSuccess(final Void noop) {
+ // Set the tip of the data tree.
+ tip = Verify.verifyNotNull(candidate);
- entry.lastAccess = readTime();
+ entry.lastAccess = readTime();
- pendingTransactions.remove();
- pendingCommits.add(entry);
+ pendingTransactions.remove();
+ pendingCommits.add(entry);
- LOG.debug("{}: Transaction {} prepared", logContext, current.getIdentifier());
+ LOG.debug("{}: Transaction {} prepared", logContext, currentId);
- cohort.successfulPreCommit(candidate);
+ cohort.successfulPreCommit(candidate);
- processNextPendingTransaction();
+ processNextPendingTransaction();
+ }
+
+ @Override
+ public void onFailure(final Throwable failure) {
+ failPreCommit(failure);
+ }
+ });
}
private void failCommit(final Exception cause) {
@@ -874,16 +998,17 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
return;
}
+ allMetadataCommittedTransaction(txId);
shard.getShardMBean().incrementCommittedTransactionCount();
shard.getShardMBean().setLastCommittedTransactionTime(System.currentTimeMillis());
// FIXME: propagate journal index
- pendingFinishCommits.poll().cohort.successfulCommit(UnsignedLong.ZERO);
-
- LOG.trace("{}: Transaction {} committed, proceeding to notify", logContext, txId);
- notifyListeners(candidate);
+ pendingFinishCommits.poll().cohort.successfulCommit(UnsignedLong.ZERO, () -> {
+ LOG.trace("{}: Transaction {} committed, proceeding to notify", logContext, txId);
+ notifyListeners(candidate);
- processNextPending();
+ processNextPending();
+ });
}
void startCommit(final SimpleShardDataTreeCohort cohort, final DataTreeCandidate candidate) {
@@ -901,7 +1026,8 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
final TransactionIdentifier txId = cohort.getIdentifier();
final Payload payload;
try {
- payload = CommitTransactionPayload.create(txId, candidate);
+ payload = CommitTransactionPayload.create(txId, candidate,
+ shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity());
} catch (IOException e) {
LOG.error("{}: Failed to encode transaction {} candidate {}", logContext, txId, candidate, e);
pendingCommits.poll().cohort.failedCommit(e);
@@ -956,26 +1082,29 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
}
@Override
- ShardDataTreeCohort createReadyCohort(final TransactionIdentifier txId, final DataTreeModification mod) {
+ ShardDataTreeCohort createReadyCohort(final TransactionIdentifier txId, final DataTreeModification mod,
+ final Optional> participatingShardNames) {
SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(this, mod, txId,
- cohortRegistry.createCohort(schemaContext, txId, COMMIT_STEP_TIMEOUT));
+ cohortRegistry.createCohort(schemaContext, txId, shard::executeInSelf,
+ COMMIT_STEP_TIMEOUT), participatingShardNames);
pendingTransactions.add(new CommitEntry(cohort, readTime()));
return cohort;
}
// Exposed for ShardCommitCoordinator so it does not have deal with local histories (it does not care), this mimics
// the newReadWriteTransaction()
- ShardDataTreeCohort newReadyCohort(final TransactionIdentifier txId, final DataTreeModification mod) {
+ ShardDataTreeCohort newReadyCohort(final TransactionIdentifier txId, final DataTreeModification mod,
+ final Optional> participatingShardNames) {
if (txId.getHistoryId().getHistoryId() == 0) {
- return createReadyCohort(txId, mod);
+ return createReadyCohort(txId, mod, participatingShardNames);
}
- return ensureTransactionChain(txId.getHistoryId(), null).createReadyCohort(txId, mod);
+ return ensureTransactionChain(txId.getHistoryId(), null).createReadyCohort(txId, mod, participatingShardNames);
}
@SuppressFBWarnings(value = "DB_DUPLICATE_SWITCH_CLAUSES", justification = "See inline comments below.")
void checkForExpiredTransactions(final long transactionCommitTimeoutMillis,
- final Function> accessTimeUpdater) {
+ final Function accessTimeUpdater) {
final long timeout = TimeUnit.MILLISECONDS.toNanos(transactionCommitTimeoutMillis);
final long now = readTime();
@@ -993,9 +1122,9 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
return;
}
- final Optional updateOpt = accessTimeUpdater.apply(currentTx.cohort);
+ final OptionalLong updateOpt = accessTimeUpdater.apply(currentTx.cohort);
if (updateOpt.isPresent()) {
- final long newAccess = updateOpt.get().longValue();
+ final long newAccess = updateOpt.getAsLong();
final long newDelta = now - newAccess;
if (newDelta < delta) {
LOG.debug("{}: Updated current transaction {} access time", logContext,
@@ -1123,7 +1252,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private void rebaseTransactions(final Iterator iter, @Nonnull final DataTreeTip newTip) {
+ private void rebaseTransactions(final Iterator iter, final @NonNull DataTreeTip newTip) {
tip = Preconditions.checkNotNull(newTip);
while (iter.hasNext()) {
final SimpleShardDataTreeCohort cohort = iter.next().cohort;
@@ -1142,11 +1271,10 @@ public class ShardDataTree extends ShardDataTreeTransactionParent {
try {
tip.validate(cohort.getDataTreeModification());
DataTreeCandidateTip candidate = tip.prepare(cohort.getDataTreeModification());
- cohort.userPreCommit(candidate);
cohort.setNewCandidate(candidate);
tip = candidate;
- } catch (ExecutionException | TimeoutException | RuntimeException | DataValidationFailedException e) {
+ } catch (RuntimeException | DataValidationFailedException e) {
LOG.debug("{}: Failed to reprepare queued transaction {}", logContext, cohort.getIdentifier(), e);
cohort.reportFailure(e);
}