*/
package org.opendaylight.controller.cluster.datastore;
+import static akka.actor.ActorRef.noSender;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Verify.verify;
+import static com.google.common.base.Verify.verifyNotNull;
+import static java.util.Objects.requireNonNull;
+
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;
import com.google.common.collect.ImmutableList;
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;
import org.opendaylight.controller.cluster.datastore.ShardDataTreeCohort.State;
-import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
+import org.opendaylight.controller.cluster.datastore.node.utils.transformer.ReusableNormalizedNodePruner;
import org.opendaylight.controller.cluster.datastore.persisted.AbortTransactionPayload;
import org.opendaylight.controller.cluster.datastore.persisted.AbstractIdentifiablePayload;
import org.opendaylight.controller.cluster.datastore.persisted.CloseLocalHistoryPayload;
import org.opendaylight.controller.cluster.datastore.persisted.CommitTransactionPayload;
import org.opendaylight.controller.cluster.datastore.persisted.CreateLocalHistoryPayload;
+import org.opendaylight.controller.cluster.datastore.persisted.DataTreeCandidateInputOutput.DataTreeCandidateWithVersion;
import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
+import org.opendaylight.controller.cluster.datastore.persisted.PayloadVersion;
import org.opendaylight.controller.cluster.datastore.persisted.PurgeLocalHistoryPayload;
import org.opendaylight.controller.cluster.datastore.persisted.PurgeTransactionPayload;
import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshot;
import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshotMetadata;
+import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
import org.opendaylight.controller.cluster.datastore.utils.DataTreeModificationOutput;
import org.opendaylight.controller.cluster.datastore.utils.PruningDataTreeModification;
+import org.opendaylight.controller.cluster.raft.base.messages.InitiateCaptureSnapshot;
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;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
import org.opendaylight.yangtools.yang.data.api.schema.tree.ConflictingModificationAppliedException;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateTip;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
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.TipProducingDataTree;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTreeTip;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
+import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeStreamVersion;
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.EffectiveModelContext;
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.
*
* <p>
- * 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;
long lastAccess;
CommitEntry(final SimpleShardDataTreeCohort cohort, final long now) {
- this.cohort = Preconditions.checkNotNull(cohort);
+ this.cohort = requireNonNull(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);
/**
private final Map<LocalHistoryIdentifier, ShardDataTreeTransactionChain> transactionChains = new HashMap<>();
private final DataTreeCohortActorRegistry cohortRegistry = new DataTreeCohortActorRegistry();
- private final Queue<CommitEntry> pendingTransactions = new ArrayDeque<>();
+ private final Deque<CommitEntry> pendingTransactions = new ArrayDeque<>();
private final Queue<CommitEntry> pendingCommits = new ArrayDeque<>();
private final Queue<CommitEntry> pendingFinishCommits = new ArrayDeque<>();
private final Map<Payload, Runnable> replicationCallbacks = new HashMap<>();
private final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher;
- private final ShardDataChangeListenerPublisher dataChangeListenerPublisher;
private final Collection<ShardDataTreeMetadata<?>> metadata;
- private final TipProducingDataTree dataTree;
+ private final DataTree dataTree;
private final String logContext;
private final Shard shard;
private Runnable runOnPendingTransactionsComplete;
/**
* Optimistic {@link DataTreeCandidate} preparation. Since our DataTree implementation is a
- * {@link TipProducingDataTree}, each {@link DataTreeCandidate} is also a {@link DataTreeTip}, e.g. another
+ * {@link DataTree}, each {@link DataTreeCandidate} is also a {@link DataTreeTip}, e.g. another
* candidate can be prepared on top of it. They still need to be committed in sequence. Here we track the current
* tip of the data tree, which is the last DataTreeCandidate we have in flight, or the DataTree itself.
*/
- private TipProducingDataTreeTip tip;
+ private DataTreeTip tip;
private SchemaContext schemaContext;
+ private DataSchemaContextTree dataSchemaContext;
private int currentTransactionBatch;
- ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TipProducingDataTree dataTree,
+ ShardDataTree(final Shard shard, final EffectiveModelContext schemaContext, final DataTree dataTree,
final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher,
- final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext,
+ final String logContext,
final ShardDataTreeMetadata<?>... metadata) {
- this.dataTree = Preconditions.checkNotNull(dataTree);
+ this.dataTree = requireNonNull(dataTree);
updateSchemaContext(schemaContext);
- this.shard = Preconditions.checkNotNull(shard);
- this.treeChangeListenerPublisher = Preconditions.checkNotNull(treeChangeListenerPublisher);
- this.dataChangeListenerPublisher = Preconditions.checkNotNull(dataChangeListenerPublisher);
- this.logContext = Preconditions.checkNotNull(logContext);
+ this.shard = requireNonNull(shard);
+ this.treeChangeListenerPublisher = requireNonNull(treeChangeListenerPublisher);
+ this.logContext = requireNonNull(logContext);
this.metadata = ImmutableList.copyOf(metadata);
tip = dataTree;
}
- ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType,
+ ShardDataTree(final Shard shard, final EffectiveModelContext 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, InMemoryDataTreeFactory.getInstance().create(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) {
+ final DataTreeConfiguration baseConfig = DataTreeConfiguration.getDefault(treeType);
+ return new InMemoryDataTreeFactory().create(new DataTreeConfiguration.Builder(baseConfig.getTreeType())
+ .setMandatoryNodesValidation(baseConfig.isMandatoryNodesValidationEnabled())
+ .setUniqueIndexes(baseConfig.isUniqueIndexEnabled())
+ .setRootPath(root)
+ .build());
}
@VisibleForTesting
- public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType) {
- this(shard, schemaContext, treeType, YangInstanceIdentifier.EMPTY,
- new DefaultShardDataTreeChangeListenerPublisher(""),
- new DefaultShardDataChangeListenerPublisher(""), "");
+ public ShardDataTree(final Shard shard, final EffectiveModelContext schemaContext, final TreeType treeType) {
+ this(shard, schemaContext, treeType, YangInstanceIdentifier.empty(),
+ new DefaultShardDataTreeChangeListenerPublisher(""), "");
}
final String logContext() {
return shard.ticker().read();
}
- public TipProducingDataTree getDataTree() {
+ public DataTree getDataTree() {
return dataTree;
}
return schemaContext;
}
- void updateSchemaContext(final SchemaContext newSchemaContext) {
- dataTree.setSchemaContext(newSchemaContext);
- this.schemaContext = Preconditions.checkNotNull(newSchemaContext);
+ void updateSchemaContext(final @NonNull EffectiveModelContext newSchemaContext) {
+ dataTree.setEffectiveModelContext(newSchemaContext);
+ this.schemaContext = newSchemaContext;
+ this.dataSchemaContext = DataSchemaContextTree.from(newSchemaContext);
}
void resetTransactionBatch() {
*
* @return A state snapshot
*/
- @Nonnull ShardDataTreeSnapshot takeStateSnapshot() {
- final NormalizedNode<?, ?> rootNode = dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY).get();
+ @NonNull ShardDataTreeSnapshot takeStateSnapshot() {
+ final NormalizedNode<?, ?> rootNode = dataTree.takeSnapshot().readNode(YangInstanceIdentifier.empty()).get();
final Builder<Class<? extends ShardDataTreeSnapshotMetadata<?>>, ShardDataTreeSnapshotMetadata<?>> metaBuilder =
ImmutableMap.builder();
return !pendingTransactions.isEmpty() || !pendingCommits.isEmpty() || !pendingFinishCommits.isEmpty();
}
- private void applySnapshot(@Nonnull final ShardDataTreeSnapshot snapshot,
+ private void applySnapshot(final @NonNull ShardDataTreeSnapshot snapshot,
final UnaryOperator<DataTreeModification> wrapper) throws DataValidationFailedException {
final Stopwatch elapsed = Stopwatch.createStarted();
}
}
- final DataTreeModification mod = wrapper.apply(dataTree.takeSnapshot().newModification());
+ final DataTreeModification unwrapped = dataTree.takeSnapshot().newModification();
+ final DataTreeModification mod = wrapper.apply(unwrapped);
// delete everything first
- mod.delete(YangInstanceIdentifier.EMPTY);
+ mod.delete(YangInstanceIdentifier.empty());
- final java.util.Optional<NormalizedNode<?, ?>> maybeNode = snapshot.getRootNode();
+ final Optional<NormalizedNode<?, ?>> maybeNode = snapshot.getRootNode();
if (maybeNode.isPresent()) {
// Add everything from the remote node back
- mod.write(YangInstanceIdentifier.EMPTY, maybeNode.get());
+ mod.write(YangInstanceIdentifier.empty(), maybeNode.get());
}
mod.ready();
- final DataTreeModification unwrapped = unwrap(mod);
dataTree.validate(unwrapped);
DataTreeCandidateTip candidate = dataTree.prepare(unwrapped);
dataTree.commit(candidate);
* @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 {
+ // TODO: we should be taking ShardSnapshotState here and performing forward-compatibility translation
applySnapshot(snapshot, UnaryOperator.identity());
}
- private PruningDataTreeModification wrapWithPruning(final DataTreeModification delegate) {
- return new PruningDataTreeModification(delegate, dataTree, schemaContext);
- }
-
- private static DataTreeModification unwrap(final DataTreeModification modification) {
- if (modification instanceof PruningDataTreeModification) {
- return ((PruningDataTreeModification)modification).delegate();
- }
- return modification;
- }
-
/**
* Apply a snapshot coming from recovery. This method does not assume the SchemaContexts match and performs data
* pruning in an attempt to adjust the state to our current SchemaContext.
* @param snapshot Snapshot that needs to be applied
* @throws DataValidationFailedException when the snapshot fails to apply
*/
- void applyRecoverySnapshot(final @Nonnull ShardDataTreeSnapshot snapshot) throws DataValidationFailedException {
- applySnapshot(snapshot, this::wrapWithPruning);
+ void applyRecoverySnapshot(final @NonNull ShardSnapshotState snapshot) throws DataValidationFailedException {
+ // TODO: we should be able to reuse the pruner, provided we are not reentrant
+ final ReusableNormalizedNodePruner pruner = ReusableNormalizedNodePruner.forDataSchemaContext(
+ dataSchemaContext);
+ if (snapshot.needsMigration()) {
+ final ReusableNormalizedNodePruner uintPruner = pruner.withUintAdaption();
+ applySnapshot(snapshot.getSnapshot(),
+ delegate -> new PruningDataTreeModification.Proactive(delegate, dataTree, uintPruner));
+ } else {
+ applySnapshot(snapshot.getSnapshot(),
+ delegate -> new PruningDataTreeModification.Reactive(delegate, dataTree, pruner));
+ }
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private void applyRecoveryCandidate(final DataTreeCandidate candidate) throws DataValidationFailedException {
- final PruningDataTreeModification mod = wrapWithPruning(dataTree.takeSnapshot().newModification());
- DataTreeCandidates.applyToModification(mod, candidate);
- mod.ready();
+ private void applyRecoveryCandidate(final CommitTransactionPayload payload) throws IOException {
+ final Entry<TransactionIdentifier, DataTreeCandidateWithVersion> entry = payload.acquireCandidate();
+ final DataTreeModification unwrapped = dataTree.takeSnapshot().newModification();
+ final PruningDataTreeModification mod = createPruningModification(unwrapped,
+ NormalizedNodeStreamVersion.MAGNESIUM.compareTo(entry.getValue().getVersion()) > 0);
- final DataTreeModification unwrapped = mod.delegate();
+ DataTreeCandidates.applyToModification(mod, entry.getValue().getCandidate());
+ mod.ready();
LOG.trace("{}: Applying recovery modification {}", logContext, unwrapped);
try {
"%s: Failed to apply recovery payload. Modification data was written to file %s",
logContext, file), e);
}
+
+ allMetadataCommittedTransaction(entry.getKey());
+ }
+
+ private PruningDataTreeModification createPruningModification(final DataTreeModification unwrapped,
+ final boolean uintAdapting) {
+ // TODO: we should be able to reuse the pruner, provided we are not reentrant
+ final ReusableNormalizedNodePruner pruner = ReusableNormalizedNodePruner.forDataSchemaContext(
+ dataSchemaContext);
+ return uintAdapting ? new PruningDataTreeModification.Proactive(unwrapped, dataTree, pruner.withUintAdaption())
+ : new PruningDataTreeModification.Reactive(unwrapped, dataTree, pruner);
}
/**
* @throws IOException when the snapshot fails to deserialize
* @throws DataValidationFailedException when the snapshot fails to apply
*/
- void applyRecoveryPayload(final @Nonnull Payload payload) throws IOException, DataValidationFailedException {
+ void applyRecoveryPayload(final @NonNull Payload payload) throws IOException {
if (payload instanceof CommitTransactionPayload) {
- final Entry<TransactionIdentifier, DataTreeCandidate> e =
- ((CommitTransactionPayload) payload).getCandidate();
- applyRecoveryCandidate(e.getValue());
- allMetadataCommittedTransaction(e.getKey());
+ applyRecoveryCandidate((CommitTransactionPayload) payload);
} else if (payload instanceof AbortTransactionPayload) {
allMetadataAbortedTransaction(((AbortTransactionPayload) payload).getIdentifier());
} else if (payload instanceof PurgeTransactionPayload) {
}
}
- private void applyReplicatedCandidate(final Identifier identifier, final DataTreeCandidate foreign)
- throws DataValidationFailedException {
+ private void applyReplicatedCandidate(final CommitTransactionPayload payload)
+ throws DataValidationFailedException, IOException {
+ final Entry<TransactionIdentifier, DataTreeCandidateWithVersion> entry = payload.acquireCandidate();
+ final TransactionIdentifier identifier = entry.getKey();
LOG.debug("{}: Applying foreign transaction {}", logContext, identifier);
final DataTreeModification mod = dataTree.takeSnapshot().newModification();
- DataTreeCandidates.applyToModification(mod, foreign);
+ // TODO: check version here, which will enable us to perform forward-compatibility transformations
+ DataTreeCandidates.applyToModification(mod, entry.getValue().getCandidate());
mod.ready();
LOG.trace("{}: Applying foreign modification {}", logContext, mod);
final DataTreeCandidate candidate = dataTree.prepare(mod);
dataTree.commit(candidate);
+ allMetadataCommittedTransaction(identifier);
notifyListeners(candidate);
}
* pre-Boron state -- which limits the number of options here.
*/
if (payload instanceof CommitTransactionPayload) {
- final TransactionIdentifier txId;
if (identifier == null) {
- final Entry<TransactionIdentifier, DataTreeCandidate> e =
- ((CommitTransactionPayload) payload).getCandidate();
- txId = e.getKey();
- applyReplicatedCandidate(txId, e.getValue());
+ applyReplicatedCandidate((CommitTransactionPayload) payload);
} else {
- Verify.verify(identifier instanceof TransactionIdentifier);
- txId = (TransactionIdentifier) identifier;
- payloadReplicationComplete(txId);
+ verify(identifier instanceof TransactionIdentifier);
+ // if we did not track this transaction before, it means that it came from another leader and we are in
+ // the process of commiting it while in PreLeader state. That means that it hasnt yet been committed to
+ // the local DataTree and would be lost if it was only applied via payloadReplicationComplete().
+ if (!payloadReplicationComplete((TransactionIdentifier) identifier)) {
+ applyReplicatedCandidate((CommitTransactionPayload) payload);
+ }
}
- allMetadataCommittedTransaction(txId);
+
+ // make sure acquireCandidate() is the last call touching the payload data as we want it to be GC-ed.
+ checkRootOverwrite(((CommitTransactionPayload) payload).acquireCandidate().getValue()
+ .getCandidate());
} else if (payload instanceof AbortTransactionPayload) {
if (identifier != null) {
payloadReplicationComplete((AbortTransactionPayload) payload);
}
}
- private void replicatePayload(final Identifier id, final Payload payload, @Nullable final Runnable callback) {
+ private void checkRootOverwrite(DataTreeCandidate candidate) {
+ final DatastoreContext datastoreContext = shard.getDatastoreContext();
+ if (!datastoreContext.isSnapshotOnRootOverwrite()) {
+ return;
+ }
+
+ if (!datastoreContext.isPersistent()) {
+ return;
+ }
+
+ if (candidate.getRootNode().getModificationType().equals(ModificationType.UNMODIFIED)) {
+ return;
+ }
+
+ // top level container ie "/"
+ if ((candidate.getRootPath().equals(YangInstanceIdentifier.empty())
+ && candidate.getRootNode().getModificationType().equals(ModificationType.WRITE))) {
+ LOG.debug("{}: shard root overwritten, enqueuing snapshot", logContext);
+ shard.self().tell(new InitiateCaptureSnapshot(), noSender());
+ return;
+ }
+ }
+
+ private void replicatePayload(final Identifier id, final Payload payload, final @Nullable Runnable callback) {
if (callback != null) {
replicationCallbacks.put(payload, callback);
}
}
}
- private void payloadReplicationComplete(final TransactionIdentifier txId) {
+ private boolean payloadReplicationComplete(final TransactionIdentifier txId) {
final CommitEntry current = pendingFinishCommits.peek();
if (current == null) {
LOG.warn("{}: No outstanding transactions, ignoring consensus on transaction {}", logContext, txId);
- return;
+ allMetadataCommittedTransaction(txId);
+ return false;
}
if (!current.cohort.getIdentifier().equals(txId)) {
LOG.debug("{}: Head of pendingFinishCommits queue is {}, ignoring consensus on transaction {}", logContext,
current.cohort.getIdentifier(), txId);
- return;
+ allMetadataCommittedTransaction(txId);
+ return false;
}
finishCommit(current.cohort);
+ return true;
}
private void allMetadataAbortedTransaction(final TransactionIdentifier txId) {
final boolean closed) {
final ShardDataTreeTransactionChain ret = new ShardDataTreeTransactionChain(historyId, this);
final ShardDataTreeTransactionChain existing = transactionChains.putIfAbsent(historyId, ret);
- Preconditions.checkState(existing == null, "Attempted to recreate chain %s, but %s already exists", historyId,
- existing);
+ checkState(existing == null, "Attempted to recreate chain %s, but %s already exists", historyId, existing);
return ret;
}
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();
}
}
ReadOnlyShardDataTreeTransaction newReadOnlyTransaction(final TransactionIdentifier txId) {
+ shard.getShardMBean().incrementReadOnlyTransactionCount();
+
if (txId.getHistoryId().getHistoryId() == 0) {
return new ReadOnlyShardDataTreeTransaction(this, txId, dataTree.takeSnapshot());
}
}
ReadWriteShardDataTreeTransaction newReadWriteTransaction(final TransactionIdentifier txId) {
+ shard.getShardMBean().incrementReadWriteTransactionCount();
+
if (txId.getHistoryId().getHistoryId() == 0) {
return new ReadWriteShardDataTreeTransaction(ShardDataTree.this, txId, dataTree.takeSnapshot()
.newModification());
@VisibleForTesting
public void notifyListeners(final DataTreeCandidate candidate) {
treeChangeListenerPublisher.publishChanges(candidate);
- dataChangeListenerPublisher.publishChanges(candidate);
}
/**
* @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;
}
/**
* @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);
return;
}
- replicatePayload(id, PurgeLocalHistoryPayload.create(id), callback);
- }
-
- void registerDataChangeListener(final YangInstanceIdentifier path,
- final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener,
- final DataChangeScope scope, final Optional<DataTreeCandidate> initialState,
- final Consumer<ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>>
- onRegistration) {
- dataChangeListenerPublisher.registerDataChangeListener(path, listener, scope, initialState, onRegistration);
+ replicatePayload(id, PurgeLocalHistoryPayload.create(
+ id, shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity()), callback);
}
Optional<DataTreeCandidate> readCurrentData() {
- final Optional<NormalizedNode<?, ?>> currentState =
- dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY);
- return currentState.isPresent() ? Optional.of(DataTreeCandidates.fromNormalizedNode(
- YangInstanceIdentifier.EMPTY, currentState.get())) : Optional.<DataTreeCandidate>absent();
+ return dataTree.takeSnapshot().readNode(YangInstanceIdentifier.empty())
+ .map(state -> DataTreeCandidates.fromNormalizedNode(YangInstanceIdentifier.empty(), state));
}
public void registerTreeChangeListener(final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener,
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
}
@Override
- ShardDataTreeCohort finishTransaction(final ReadWriteShardDataTreeTransaction transaction) {
+ ShardDataTreeCohort finishTransaction(final ReadWriteShardDataTreeTransaction transaction,
+ final Optional<SortedSet<String>> 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<NormalizedNode<?, ?>> readNode(final YangInstanceIdentifier path) {
return dataTree.takeSnapshot().newModification();
}
- /**
- * Commits a modification.
- *
- * @deprecated This method violates DataTree containment and will be removed.
- */
- @VisibleForTesting
- @Deprecated
- public DataTreeCandidate commit(final DataTreeModification modification) throws DataValidationFailedException {
- // Direct modification commit is a utility, which cannot be used while we have transactions in-flight
- Preconditions.checkState(tip == dataTree, "Cannot modify data tree while transacgitons are pending");
-
- modification.ready();
- dataTree.validate(modification);
- DataTreeCandidate candidate = dataTree.prepare(modification);
- dataTree.commit(candidate);
- return candidate;
- }
-
public Collection<ShardDataTreeCohort> getAndClearPendingTransactions() {
Collection<ShardDataTreeCohort> ret = new ArrayList<>(getQueueSize());
LOG.debug("{}: Validating transaction {}", logContext, cohort.getIdentifier());
Exception cause;
try {
- cohort.throwCanCommitFailure();
-
tip.validate(modification);
LOG.debug("{}: Transaction {} validated", logContext, cohort.getIdentifier());
cohort.successfulCanCommit();
} 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;
}
void startCanCommit(final SimpleShardDataTreeCohort cohort) {
- final SimpleShardDataTreeCohort current = pendingTransactions.peek().cohort;
- if (!cohort.equals(current)) {
- LOG.debug("{}: Transaction {} scheduled for canCommit step", logContext, cohort.getIdentifier());
+ final CommitEntry head = pendingTransactions.peek();
+ if (head == null) {
+ LOG.warn("{}: No transactions enqueued while attempting to start canCommit on {}", logContext, cohort);
return;
}
+ if (!cohort.equals(head.cohort)) {
+ // 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<String> 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<CommitEntry> 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<String> 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<CommitEntry> queue, final CommitEntry entry, final int atIndex) {
+ if (atIndex == 0) {
+ queue.addFirst(entry);
+ return;
+ }
+
+ LOG.trace("Inserting into Deque at index {}", atIndex);
+
+ Deque<CommitEntry> tempStack = new ArrayDeque<>(atIndex);
+ for (int i = 0; i < atIndex; i++) {
+ tempStack.push(queue.poll());
+ }
+
+ queue.addFirst(entry);
+
+ tempStack.forEach(queue::addFirst);
+ }
+
+ private Collection<String> extractPrecedingShardNames(final Optional<SortedSet<String>> participatingShardNames) {
+ return participatingShardNames.map((Function<SortedSet<String>, Collection<String>>)
+ set -> set.headSet(shard.getShardName())).orElse(Collections.<String>emptyList());
+ }
+
+ private void failPreCommit(final Throwable cause) {
shard.getShardMBean().incrementFailedTransactionsCount();
pendingTransactions.poll().cohort.failedPreCommit(cause);
processNextPendingTransaction();
@SuppressWarnings("checkstyle:IllegalCatch")
void startPreCommit(final SimpleShardDataTreeCohort cohort) {
final CommitEntry entry = pendingTransactions.peek();
- Preconditions.checkState(entry != null, "Attempted to pre-commit of %s when no transactions pending", cohort);
+ checkState(entry != null, "Attempted to pre-commit of %s when no transactions pending", cohort);
final SimpleShardDataTreeCohort current = entry.cohort;
- Verify.verify(cohort.equals(current), "Attempted to pre-commit %s while %s is pending", cohort, current);
+ 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 (DataValidationFailedException | RuntimeException e) {
failPreCommit(e);
return;
}
- // Set the tip of the data tree.
- tip = Verify.verifyNotNull(candidate);
+ cohort.userPreCommit(candidate, new FutureCallback<Void>() {
+ @Override
+ public void onSuccess(final Void noop) {
+ // Set the tip of the data tree.
+ tip = 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) {
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) {
final CommitEntry entry = pendingCommits.peek();
- Preconditions.checkState(entry != null, "Attempted to start commit of %s when no transactions pending", cohort);
+ checkState(entry != null, "Attempted to start commit of %s when no transactions pending", cohort);
final SimpleShardDataTreeCohort current = entry.cohort;
if (!cohort.equals(current)) {
final TransactionIdentifier txId = cohort.getIdentifier();
final Payload payload;
try {
- payload = CommitTransactionPayload.create(txId, candidate);
+ payload = CommitTransactionPayload.create(txId, candidate, PayloadVersion.current(),
+ shard.getDatastoreContext().getInitialPayloadSerializedBufferCapacity());
} catch (IOException e) {
LOG.error("{}: Failed to encode transaction {} candidate {}", logContext, txId, candidate, e);
pendingCommits.poll().cohort.failedCommit(e);
@Override
ShardDataTreeCohort createFailedCohort(final TransactionIdentifier txId, final DataTreeModification mod,
final Exception failure) {
- SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort.DeadOnArrival(this, mod, txId, failure);
+ final SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(this, mod, txId, failure);
pendingTransactions.add(new CommitEntry(cohort, readTime()));
return cohort;
}
@Override
- ShardDataTreeCohort createReadyCohort(final TransactionIdentifier txId,
- final DataTreeModification mod) {
- SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort.Normal(this, mod, txId,
- cohortRegistry.createCohort(schemaContext, txId, COMMIT_STEP_TIMEOUT));
+ ShardDataTreeCohort createReadyCohort(final TransactionIdentifier txId, final DataTreeModification mod,
+ final Optional<SortedSet<String>> participatingShardNames) {
+ SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(this, mod, txId,
+ 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<SortedSet<String>> 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) {
+ void checkForExpiredTransactions(final long transactionCommitTimeoutMillis,
+ final Function<SimpleShardDataTreeCohort, OptionalLong> accessTimeUpdater) {
final long timeout = TimeUnit.MILLISECONDS.toNanos(transactionCommitTimeoutMillis);
final long now = readTime();
final Queue<CommitEntry> currentQueue = !pendingFinishCommits.isEmpty() ? pendingFinishCommits :
!pendingCommits.isEmpty() ? pendingCommits : pendingTransactions;
final CommitEntry currentTx = currentQueue.peek();
- if (currentTx != null && currentTx.lastAccess + timeout < now) {
- LOG.warn("{}: Current transaction {} has timed out after {} ms in state {}", logContext,
- currentTx.cohort.getIdentifier(), transactionCommitTimeoutMillis, currentTx.cohort.getState());
- boolean processNext = true;
- switch (currentTx.cohort.getState()) {
- case CAN_COMMIT_PENDING:
- currentQueue.remove().cohort.failedCanCommit(new TimeoutException());
- break;
- case CAN_COMMIT_COMPLETE:
- // The suppression of the FindBugs "DB_DUPLICATE_SWITCH_CLAUSES" warning pertains to this clause
- // whose code is duplicated with PRE_COMMIT_COMPLETE. The clauses aren't combined in case the code
- // in PRE_COMMIT_COMPLETE is changed.
- currentQueue.remove().cohort.reportFailure(new TimeoutException());
- break;
- case PRE_COMMIT_PENDING:
- currentQueue.remove().cohort.failedPreCommit(new TimeoutException());
- break;
- case PRE_COMMIT_COMPLETE:
- // FIXME: this is a legacy behavior problem. Three-phase commit protocol specifies that after we
- // are ready we should commit the transaction, not abort it. Our current software stack does
- // not allow us to do that consistently, because we persist at the time of commit, hence
- // we can end up in a state where we have pre-committed a transaction, then a leader failover
- // occurred ... the new leader does not see the pre-committed transaction and does not have
- // a running timer. To fix this we really need two persistence events.
- //
- // The first one, done at pre-commit time will hold the transaction payload. When consensus
- // is reached, we exit the pre-commit phase and start the pre-commit timer. Followers do not
- // apply the state in this event.
- //
- // The second one, done at commit (or abort) time holds only the transaction identifier and
- // signals to followers that the state should (or should not) be applied.
- //
- // In order to make the pre-commit timer working across failovers, though, we need
- // a per-shard cluster-wide monotonic time, so a follower becoming the leader can accurately
- // restart the timer.
- currentQueue.remove().cohort.reportFailure(new TimeoutException());
- break;
- case COMMIT_PENDING:
- LOG.warn("{}: Transaction {} is still committing, cannot abort", logContext,
- currentTx.cohort.getIdentifier());
- currentTx.lastAccess = now;
- processNext = false;
- return;
- case ABORTED:
- case COMMITTED:
- case FAILED:
- case READY:
- default:
- currentQueue.remove();
+ if (currentTx == null) {
+ // Empty queue, no-op
+ return;
+ }
+
+ long delta = now - currentTx.lastAccess;
+ if (delta < timeout) {
+ // Not expired yet, bail
+ return;
+ }
+
+ final OptionalLong updateOpt = accessTimeUpdater.apply(currentTx.cohort);
+ if (updateOpt.isPresent()) {
+ final long newAccess = updateOpt.getAsLong();
+ final long newDelta = now - newAccess;
+ if (newDelta < delta) {
+ LOG.debug("{}: Updated current transaction {} access time", logContext,
+ currentTx.cohort.getIdentifier());
+ currentTx.lastAccess = newAccess;
+ delta = newDelta;
}
- if (processNext) {
- processNextPending();
+ if (delta < timeout) {
+ // Not expired yet, bail
+ return;
}
}
+
+ final long deltaMillis = TimeUnit.NANOSECONDS.toMillis(delta);
+ final State state = currentTx.cohort.getState();
+
+ LOG.warn("{}: Current transaction {} has timed out after {} ms in state {}", logContext,
+ currentTx.cohort.getIdentifier(), deltaMillis, state);
+ boolean processNext = true;
+ final TimeoutException cohortFailure = new TimeoutException("Backend timeout in state " + state + " after "
+ + deltaMillis + "ms");
+
+ switch (state) {
+ case CAN_COMMIT_PENDING:
+ currentQueue.remove().cohort.failedCanCommit(cohortFailure);
+ break;
+ case CAN_COMMIT_COMPLETE:
+ // The suppression of the FindBugs "DB_DUPLICATE_SWITCH_CLAUSES" warning pertains to this clause
+ // whose code is duplicated with PRE_COMMIT_COMPLETE. The clauses aren't combined in case the code
+ // in PRE_COMMIT_COMPLETE is changed.
+ currentQueue.remove().cohort.reportFailure(cohortFailure);
+ break;
+ case PRE_COMMIT_PENDING:
+ currentQueue.remove().cohort.failedPreCommit(cohortFailure);
+ break;
+ case PRE_COMMIT_COMPLETE:
+ // FIXME: this is a legacy behavior problem. Three-phase commit protocol specifies that after we
+ // are ready we should commit the transaction, not abort it. Our current software stack does
+ // not allow us to do that consistently, because we persist at the time of commit, hence
+ // we can end up in a state where we have pre-committed a transaction, then a leader failover
+ // occurred ... the new leader does not see the pre-committed transaction and does not have
+ // a running timer. To fix this we really need two persistence events.
+ //
+ // The first one, done at pre-commit time will hold the transaction payload. When consensus
+ // is reached, we exit the pre-commit phase and start the pre-commit timer. Followers do not
+ // apply the state in this event.
+ //
+ // The second one, done at commit (or abort) time holds only the transaction identifier and
+ // signals to followers that the state should (or should not) be applied.
+ //
+ // In order to make the pre-commit timer working across failovers, though, we need
+ // a per-shard cluster-wide monotonic time, so a follower becoming the leader can accurately
+ // restart the timer.
+ currentQueue.remove().cohort.reportFailure(cohortFailure);
+ break;
+ case COMMIT_PENDING:
+ LOG.warn("{}: Transaction {} is still committing, cannot abort", logContext,
+ currentTx.cohort.getIdentifier());
+ currentTx.lastAccess = now;
+ processNext = false;
+ return;
+ case READY:
+ currentQueue.remove().cohort.reportFailure(cohortFailure);
+ break;
+ case ABORTED:
+ case COMMITTED:
+ case FAILED:
+ default:
+ currentQueue.remove();
+ }
+
+ if (processNext) {
+ processNextPending();
+ }
}
boolean startAbort(final SimpleShardDataTreeCohort cohort) {
return false;
}
- TipProducingDataTreeTip newTip = MoreObjects.firstNonNull(first.cohort.getCandidate(), dataTree);
+ DataTreeTip newTip = MoreObjects.firstNonNull(first.cohort.getCandidate(), dataTree);
while (it.hasNext()) {
final CommitEntry e = it.next();
if (cohort.equals(e.cohort)) {
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private void rebaseTransactions(final Iterator<CommitEntry> iter, @Nonnull final TipProducingDataTreeTip newTip) {
- tip = Preconditions.checkNotNull(newTip);
+ private void rebaseTransactions(final Iterator<CommitEntry> iter, final @NonNull DataTreeTip newTip) {
+ tip = requireNonNull(newTip);
while (iter.hasNext()) {
final SimpleShardDataTreeCohort cohort = iter.next().cohort;
if (cohort.getState() == State.CAN_COMMIT_COMPLETE) {
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);
}
ShardStats getStats() {
return shard.getShardMBean();
}
+
+ Iterator<SimpleShardDataTreeCohort> cohortIterator() {
+ return Iterables.transform(Iterables.concat(pendingFinishCommits, pendingCommits, pendingTransactions),
+ e -> e.cohort).iterator();
+ }
+
+ void removeTransactionChain(final LocalHistoryIdentifier id) {
+ if (transactionChains.remove(id) != null) {
+ LOG.debug("{}: Removed transaction chain {}", logContext, id);
+ }
+ }
}
Code Review / controller.git / blobdiff