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=78b49a60ae5cef444b28633fd9b8fe397d117914;hp=a990b50a3222e5dbbc21ad3eaa8ee99b5279ddca;hb=823bd74f34ee1c651f1f90daeef386a35c68d431;hpb=92cbb07ef81943b0740ba7c8915001ac6785f560 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 a990b50a32..78b49a60ae 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 @@ -10,14 +10,18 @@ package org.opendaylight.controller.cluster.datastore; 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.Ticker; 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 edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.File; import java.io.IOException; import java.util.AbstractMap.SimpleEntry; @@ -32,6 +36,7 @@ import java.util.Queue; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.Consumer; import java.util.function.UnaryOperator; import javax.annotation.Nonnull; import javax.annotation.concurrent.NotThreadSafe; @@ -62,9 +67,11 @@ 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.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.ModificationType; 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.TreeType; import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory; import org.opendaylight.yangtools.yang.model.api.SchemaContext; @@ -77,6 +84,7 @@ import scala.concurrent.duration.Duration; * 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. */ @NotThreadSafe @@ -97,6 +105,8 @@ 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 Queue pendingCommits = new ArrayDeque<>(); + private final Queue pendingFinishCommits = new ArrayDeque<>(); private final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher; private final ShardDataChangeListenerPublisher dataChangeListenerPublisher; private final Collection> metadata; @@ -105,6 +115,14 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { 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 + * 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 SchemaContext schemaContext; public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TipProducingDataTree dataTree, @@ -119,25 +137,32 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { this.dataChangeListenerPublisher = Preconditions.checkNotNull(dataChangeListenerPublisher); this.logContext = Preconditions.checkNotNull(logContext); this.metadata = ImmutableList.copyOf(metadata); + tip = dataTree; } public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType, + final YangInstanceIdentifier root, final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher, final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext) { - this(shard, schemaContext, InMemoryDataTreeFactory.getInstance().create(treeType), + this(shard, schemaContext, InMemoryDataTreeFactory.getInstance().create(treeType, root), treeChangeListenerPublisher, dataChangeListenerPublisher, logContext); } @VisibleForTesting public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType) { - this(shard, schemaContext, treeType, new DefaultShardDataTreeChangeListenerPublisher(), + this(shard, schemaContext, treeType, YangInstanceIdentifier.EMPTY, + new DefaultShardDataTreeChangeListenerPublisher(), new DefaultShardDataChangeListenerPublisher(), ""); } - String logContext() { + final String logContext() { return logContext; } + final Ticker ticker() { + return shard.ticker(); + } + public TipProducingDataTree getDataTree() { return dataTree; } @@ -146,9 +171,9 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { return schemaContext; } - void updateSchemaContext(final SchemaContext schemaContext) { - dataTree.setSchemaContext(schemaContext); - this.schemaContext = Preconditions.checkNotNull(schemaContext); + void updateSchemaContext(final SchemaContext newSchemaContext) { + dataTree.setSchemaContext(newSchemaContext); + this.schemaContext = Preconditions.checkNotNull(newSchemaContext); } /** @@ -171,11 +196,15 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { return new MetadataShardDataTreeSnapshot(rootNode, metaBuilder.build()); } - private void applySnapshot(final @Nonnull ShardDataTreeSnapshot snapshot, + private boolean anyPendingTransactions() { + return !pendingTransactions.isEmpty() || !pendingCommits.isEmpty() || !pendingFinishCommits.isEmpty(); + } + + private void applySnapshot(@Nonnull final ShardDataTreeSnapshot snapshot, final UnaryOperator wrapper) throws DataValidationFailedException { final Stopwatch elapsed = Stopwatch.createStarted(); - if (!pendingTransactions.isEmpty()) { + if (anyPendingTransactions()) { LOG.warn("{}: applying state snapshot with pending transactions", logContext); } @@ -208,10 +237,24 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { final DataTreeModification unwrapped = unwrap(mod); dataTree.validate(unwrapped); - dataTree.commit(dataTree.prepare(unwrapped)); + DataTreeCandidateTip candidate = dataTree.prepare(unwrapped); + dataTree.commit(candidate); + notifyListeners(candidate); + LOG.debug("{}: state snapshot applied in %s", logContext, elapsed); } + /** + * Apply a snapshot coming from the leader. This method assumes the leader and follower SchemaContexts match and + * does not perform any pruning. + * + * @param snapshot Snapshot that needs to be applied + * @throws DataValidationFailedException when the snapshot fails to apply + */ + void applySnapshot(@Nonnull final ShardDataTreeSnapshot snapshot) throws DataValidationFailedException { + applySnapshot(snapshot, UnaryOperator.identity()); + } + private PruningDataTreeModification wrapWithPruning(final DataTreeModification delegate) { return new PruningDataTreeModification(delegate, dataTree, schemaContext); } @@ -234,18 +277,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { applySnapshot(snapshot, this::wrapWithPruning); } - - /** - * Apply a snapshot coming from the leader. This method assumes the leader and follower SchemaContexts match and - * does not perform any pruning. - * - * @param snapshot Snapshot that needs to be applied - * @throws DataValidationFailedException when the snapshot fails to apply - */ - void applySnapshot(final @Nonnull ShardDataTreeSnapshot snapshot) throws DataValidationFailedException { - applySnapshot(snapshot, UnaryOperator.identity()); - } - + @SuppressWarnings("checkstyle:IllegalCatch") private void applyRecoveryCandidate(final DataTreeCandidate candidate) throws DataValidationFailedException { final PruningDataTreeModification mod = wrapWithPruning(dataTree.takeSnapshot().newModification()); DataTreeCandidates.applyToModification(mod, candidate); @@ -277,7 +309,8 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { */ void applyRecoveryPayload(final @Nonnull Payload payload) throws IOException, DataValidationFailedException { if (payload instanceof CommitTransactionPayload) { - final Entry e = ((CommitTransactionPayload) payload).getCandidate(); + final Entry e = + ((CommitTransactionPayload) payload).getCandidate(); applyRecoveryCandidate(e.getValue()); allMetadataCommittedTransaction(e.getKey()); } else if (payload instanceof DataTreeCandidatePayload) { @@ -327,7 +360,8 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { */ if (payload instanceof CommitTransactionPayload) { if (identifier == null) { - final Entry e = ((CommitTransactionPayload) payload).getCandidate(); + final Entry e = + ((CommitTransactionPayload) payload).getCandidate(); applyReplicatedCandidate(e.getKey(), e.getValue()); allMetadataCommittedTransaction(e.getKey()); } else { @@ -340,14 +374,14 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { } private void payloadReplicationComplete(final TransactionIdentifier txId) { - final CommitEntry current = pendingTransactions.peek(); + final CommitEntry current = pendingFinishCommits.peek(); if (current == null) { LOG.warn("{}: No outstanding transactions, ignoring consensus on transaction {}", logContext, txId); return; } if (!current.cohort.getIdentifier().equals(txId)) { - LOG.warn("{}: Head of queue is {}, ignoring consensus on transaction {}", logContext, + LOG.debug("{}: Head of pendingFinishCommits queue is {}, ignoring consensus on transaction {}", logContext, current.cohort.getIdentifier(), txId); return; } @@ -361,7 +395,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { } } - private ShardDataTreeTransactionChain ensureTransactionChain(final LocalHistoryIdentifier localHistoryIdentifier) { + ShardDataTreeTransactionChain ensureTransactionChain(final LocalHistoryIdentifier localHistoryIdentifier) { ShardDataTreeTransactionChain chain = transactionChains.get(localHistoryIdentifier); if (chain == null) { chain = new ShardDataTreeTransactionChain(localHistoryIdentifier, this); @@ -434,28 +468,29 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { Optional> registerChangeListener(final YangInstanceIdentifier path, final AsyncDataChangeListener> listener, final DataChangeScope scope) { - final DataChangeListenerRegistration>> reg = + DataChangeListenerRegistration>> reg = dataChangeListenerPublisher.registerDataChangeListener(path, listener, scope); return new SimpleEntry<>(reg, readCurrentData()); } private Optional readCurrentData() { - final Optional> currentState = dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY); + final Optional> currentState = + dataTree.takeSnapshot().readNode(YangInstanceIdentifier.EMPTY); return currentState.isPresent() ? Optional.of(DataTreeCandidates.fromNormalizedNode( YangInstanceIdentifier.EMPTY, currentState.get())) : Optional.absent(); } - public Entry, Optional> registerTreeChangeListener( - final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener) { - final ListenerRegistration reg = treeChangeListenerPublisher.registerTreeChangeListener( - path, listener); + public Entry, Optional> + registerTreeChangeListener(final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener) { + final ListenerRegistration reg = + treeChangeListenerPublisher.registerTreeChangeListener(path, listener); return new SimpleEntry<>(reg, readCurrentData()); } int getQueueSize() { - return pendingTransactions.size(); + return pendingTransactions.size() + pendingCommits.size() + pendingFinishCommits.size(); } @Override @@ -468,7 +503,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { final DataTreeModification snapshot = transaction.getSnapshot(); snapshot.ready(); - return createReadyCohort(transaction.getId(), snapshot); + return createReadyCohort(transaction.getIdentifier(), snapshot); } public Optional> readNode(final YangInstanceIdentifier path) { @@ -485,11 +520,16 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { } /** + * 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); @@ -498,29 +538,37 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { } public Collection getAndClearPendingTransactions() { - Collection ret = new ArrayList<>(pendingTransactions.size()); - for(CommitEntry entry: pendingTransactions) { + Collection ret = new ArrayList<>(getQueueSize()); + + for (CommitEntry entry: pendingFinishCommits) { + ret.add(entry.cohort); + } + + for (CommitEntry entry: pendingCommits) { + ret.add(entry.cohort); + } + + for (CommitEntry entry: pendingTransactions) { ret.add(entry.cohort); } + pendingFinishCommits.clear(); + pendingCommits.clear(); pendingTransactions.clear(); + tip = dataTree; return ret; } - private void processNextTransaction() { - while (!pendingTransactions.isEmpty()) { - final CommitEntry entry = pendingTransactions.peek(); + @SuppressWarnings("checkstyle:IllegalCatch") + private void processNextPendingTransaction() { + processNextPending(pendingTransactions, State.CAN_COMMIT_PENDING, entry -> { final SimpleShardDataTreeCohort cohort = entry.cohort; final DataTreeModification modification = cohort.getDataTreeModification(); - if(cohort.getState() != State.CAN_COMMIT_PENDING) { - break; - } - LOG.debug("{}: Validating transaction {}", logContext, cohort.getIdentifier()); Exception cause; try { - dataTree.validate(modification); + tip.validate(modification); LOG.debug("{}: Transaction {} validated", logContext, cohort.getIdentifier()); cohort.successfulCanCommit(); entry.lastAccess = shard.ticker().read(); @@ -535,7 +583,8 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { // 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); + LOG.debug("{}: Store Tx {}: modifications: {} tree: {}", cohort.getIdentifier(), modification, + dataTree); cause = new TransactionCommitFailedException("Data did not pass validation.", e); } catch (Exception e) { LOG.warn("{}: Unexpected failure in validation phase", logContext, e); @@ -544,11 +593,51 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { // Failure path: propagate the failure, remove the transaction from the queue and loop to the next one pendingTransactions.poll().cohort.failedCanCommit(cause); + }); + } + + private void processNextPending() { + processNextPendingFinishCommit(); + processNextPendingCommit(); + processNextPendingTransaction(); + } + + private void processNextPending(Queue queue, State allowedState, Consumer processor) { + while (!queue.isEmpty()) { + final CommitEntry entry = queue.peek(); + final SimpleShardDataTreeCohort cohort = entry.cohort; + + if (cohort.isFailed()) { + LOG.debug("{}: Removing failed transaction {}", logContext, cohort.getIdentifier()); + queue.remove(); + continue; + } + + if (cohort.getState() == allowedState) { + processor.accept(entry); + } + + break; } maybeRunOperationOnPendingTransactionsComplete(); } + private void processNextPendingCommit() { + processNextPending(pendingCommits, State.COMMIT_PENDING, + entry -> startCommit(entry.cohort, entry.cohort.getCandidate())); + } + + private void processNextPendingFinishCommit() { + processNextPending(pendingFinishCommits, State.FINISH_COMMIT_PENDING, + entry -> payloadReplicationComplete(entry.cohort.getIdentifier())); + } + + private boolean peekNextPendingCommit() { + final CommitEntry first = pendingCommits.peek(); + return first != null && first.cohort.getState() == State.COMMIT_PENDING; + } + void startCanCommit(final SimpleShardDataTreeCohort cohort) { final SimpleShardDataTreeCohort current = pendingTransactions.peek().cohort; if (!cohort.equals(current)) { @@ -556,52 +645,67 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { return; } - processNextTransaction(); + processNextPendingTransaction(); } private void failPreCommit(final Exception cause) { shard.getShardMBean().incrementFailedTransactionsCount(); pendingTransactions.poll().cohort.failedPreCommit(cause); - processNextTransaction(); + 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); final SimpleShardDataTreeCohort current = entry.cohort; Verify.verify(cohort.equals(current), "Attempted to pre-commit %s while %s is pending", cohort, current); - final DataTreeCandidateTip candidate; - try { - candidate = dataTree.prepare(cohort.getDataTreeModification()); - } catch (Exception e) { - failPreCommit(e); - return; - } + LOG.debug("{}: Preparing transaction {}", logContext, current.getIdentifier()); + + final DataTreeCandidateTip candidate; try { + candidate = tip.prepare(cohort.getDataTreeModification()); cohort.userPreCommit(candidate); - } catch (ExecutionException | TimeoutException e) { + } catch (ExecutionException | TimeoutException | RuntimeException e) { failPreCommit(e); return; } + // Set the tip of the data tree. + tip = Verify.verifyNotNull(candidate); + entry.lastAccess = shard.ticker().read(); + + pendingTransactions.remove(); + pendingCommits.add(entry); + + LOG.debug("{}: Transaction {} prepared", logContext, current.getIdentifier()); + cohort.successfulPreCommit(candidate); + + processNextPendingTransaction(); } private void failCommit(final Exception cause) { shard.getShardMBean().incrementFailedTransactionsCount(); - pendingTransactions.poll().cohort.failedCommit(cause); - processNextTransaction(); + pendingFinishCommits.poll().cohort.failedCommit(cause); + processNextPending(); } + @SuppressWarnings("checkstyle:IllegalCatch") private void finishCommit(final SimpleShardDataTreeCohort cohort) { final TransactionIdentifier txId = cohort.getIdentifier(); final DataTreeCandidate candidate = cohort.getCandidate(); LOG.debug("{}: Resuming commit of transaction {}", logContext, txId); + if (tip == candidate) { + // All pending candidates have been committed, reset the tip to the data tree. + tip = dataTree; + } + try { dataTree.commit(candidate); } catch (Exception e) { @@ -614,47 +718,81 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { shard.getShardMBean().setLastCommittedTransactionTime(System.currentTimeMillis()); // FIXME: propagate journal index - pendingTransactions.poll().cohort.successfulCommit(UnsignedLong.ZERO); + pendingFinishCommits.poll().cohort.successfulCommit(UnsignedLong.ZERO); LOG.trace("{}: Transaction {} committed, proceeding to notify", logContext, txId); notifyListeners(candidate); - processNextTransaction(); + processNextPending(); } void startCommit(final SimpleShardDataTreeCohort cohort, final DataTreeCandidate candidate) { - final CommitEntry entry = pendingTransactions.peek(); + final CommitEntry entry = pendingCommits.peek(); Preconditions.checkState(entry != null, "Attempted to start commit of %s when no transactions pending", cohort); final SimpleShardDataTreeCohort current = entry.cohort; - Verify.verify(cohort.equals(current), "Attempted to commit %s while %s is pending", cohort, current); + if (!cohort.equals(current)) { + LOG.debug("{}: Transaction {} scheduled for commit step", logContext, cohort.getIdentifier()); + return; + } + + LOG.debug("{}: Starting commit for transaction {}", logContext, current.getIdentifier()); + final TransactionIdentifier txId = cohort.getIdentifier(); if (shard.canSkipPayload() || candidate.getRootNode().getModificationType() == ModificationType.UNMODIFIED) { LOG.debug("{}: No replication required, proceeding to finish commit", logContext); - finishCommit(cohort); + pendingCommits.remove(); + pendingFinishCommits.add(entry); + cohort.finishCommitPending(); + payloadReplicationComplete(txId); return; } - final TransactionIdentifier txId = cohort.getIdentifier(); final Payload payload; try { payload = CommitTransactionPayload.create(txId, candidate); } catch (IOException e) { LOG.error("{}: Failed to encode transaction {} candidate {}", logContext, txId, candidate, e); - pendingTransactions.poll().cohort.failedCommit(e); + pendingCommits.poll().cohort.failedCommit(e); + processNextPending(); return; } + // We process next transactions pending canCommit before we call persistPayload to possibly progress subsequent + // transactions to the COMMIT_PENDING state so the payloads can be batched for replication. This is done for + // single-shard transactions that immediately transition from canCommit to preCommit to commit. Note that + // if the next pending transaction is progressed to COMMIT_PENDING and this method (startCommit) is called, + // the next transaction will not attempt to replicate b/c the current transaction is still at the head of the + // pendingCommits queue. + processNextPendingTransaction(); + + // After processing next pending transactions, we can now remove the current transaction from pendingCommits. + // Note this must be done before the call to peekNextPendingCommit below so we check the next transaction + // in order to properly determine the batchHint flag for the call to persistPayload. + pendingCommits.remove(); + pendingFinishCommits.add(entry); + + // See if the next transaction is pending commit (ie in the COMMIT_PENDING state) so it can be batched with + // this transaction for replication. + boolean replicationBatchHint = peekNextPendingCommit(); + // Once completed, we will continue via payloadReplicationComplete + shard.persistPayload(txId, payload, replicationBatchHint); + entry.lastAccess = shard.ticker().read(); - shard.persistPayload(txId, payload); + LOG.debug("{}: Transaction {} submitted to persistence", logContext, txId); + + // Process the next transaction pending commit, if any. If there is one it will be batched with this + // transaction for replication. + processNextPendingCommit(); } void processCohortRegistryCommand(final ActorRef sender, final CohortRegistryCommand message) { cohortRegistry.process(sender, message); } + @Override ShardDataTreeCohort createReadyCohort(final TransactionIdentifier txId, final DataTreeModification modification) { SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(this, modification, txId, @@ -663,23 +801,30 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { return cohort; } + @SuppressFBWarnings(value = "DB_DUPLICATE_SWITCH_CLAUSES", justification = "See inline comments below.") void checkForExpiredTransactions(final long transactionCommitTimeoutMillis) { final long timeout = TimeUnit.MILLISECONDS.toNanos(transactionCommitTimeoutMillis); final long now = shard.ticker().read(); - final CommitEntry currentTx = pendingTransactions.peek(); + + final Queue 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: - pendingTransactions.poll().cohort.failedCanCommit(new TimeoutException()); + currentQueue.remove().cohort.failedCanCommit(new TimeoutException()); break; case CAN_COMMIT_COMPLETE: - pendingTransactions.poll().cohort.reportFailure(new TimeoutException()); + // 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: - pendingTransactions.poll().cohort.failedPreCommit(new TimeoutException()); + 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 @@ -699,7 +844,7 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { // 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. - pendingTransactions.poll().cohort.reportFailure(new TimeoutException()); + currentQueue.remove().cohort.reportFailure(new TimeoutException()); break; case COMMIT_PENDING: LOG.warn("{}: Transaction {} is still committing, cannot abort", logContext, @@ -712,47 +857,94 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { case FAILED: case READY: default: - pendingTransactions.poll(); + currentQueue.remove(); } if (processNext) { - processNextTransaction(); + processNextPending(); } } } - void startAbort(final SimpleShardDataTreeCohort cohort) { - final Iterator it = pendingTransactions.iterator(); + boolean startAbort(final SimpleShardDataTreeCohort cohort) { + final Iterator it = Iterables.concat(pendingFinishCommits, pendingCommits, + pendingTransactions).iterator(); if (!it.hasNext()) { LOG.debug("{}: no open transaction while attempting to abort {}", logContext, cohort.getIdentifier()); - return; + return true; } // First entry is special, as it may already be committing final CommitEntry first = it.next(); if (cohort.equals(first.cohort)) { if (cohort.getState() != State.COMMIT_PENDING) { - LOG.debug("{}: aborted head of queue {} in state {}", logContext, cohort.getIdentifier(), + LOG.debug("{}: aborting head of queue {} in state {}", logContext, cohort.getIdentifier(), cohort.getIdentifier()); - pendingTransactions.poll(); - processNextTransaction(); - } else { - LOG.warn("{}: transaction {} is committing, skipping abort", logContext, cohort.getIdentifier()); + + it.remove(); + if (cohort.getCandidate() != null) { + rebaseTransactions(it, dataTree); + } + + processNextPending(); + return true; } - return; + LOG.warn("{}: transaction {} is committing, skipping abort", logContext, cohort.getIdentifier()); + return false; } + TipProducingDataTreeTip newTip = MoreObjects.firstNonNull(first.cohort.getCandidate(), dataTree); while (it.hasNext()) { final CommitEntry e = it.next(); if (cohort.equals(e.cohort)) { LOG.debug("{}: aborting queued transaction {}", logContext, cohort.getIdentifier()); + it.remove(); - return; + if (cohort.getCandidate() != null) { + rebaseTransactions(it, newTip); + } + + return true; + } else { + newTip = MoreObjects.firstNonNull(e.cohort.getCandidate(), newTip); } } LOG.debug("{}: aborted transaction {} not found in the queue", logContext, cohort.getIdentifier()); + return true; + } + + @SuppressWarnings("checkstyle:IllegalCatch") + private void rebaseTransactions(Iterator iter, @Nonnull TipProducingDataTreeTip newTip) { + tip = Preconditions.checkNotNull(newTip); + while (iter.hasNext()) { + final SimpleShardDataTreeCohort cohort = iter.next().cohort; + if (cohort.getState() == State.CAN_COMMIT_COMPLETE) { + LOG.debug("{}: Revalidating queued transaction {}", logContext, cohort.getIdentifier()); + + try { + tip.validate(cohort.getDataTreeModification()); + } catch (DataValidationFailedException | RuntimeException e) { + LOG.debug("{}: Failed to revalidate queued transaction {}", logContext, cohort.getIdentifier(), e); + cohort.reportFailure(e); + } + } else if (cohort.getState() == State.PRE_COMMIT_COMPLETE) { + LOG.debug("{}: Repreparing queued transaction {}", logContext, cohort.getIdentifier()); + + 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) { + LOG.debug("{}: Failed to reprepare queued transaction {}", logContext, cohort.getIdentifier(), e); + cohort.reportFailure(e); + } + } + } } void setRunOnPendingTransactionsComplete(final Runnable operation) { @@ -761,12 +953,12 @@ public class ShardDataTree extends ShardDataTreeTransactionParent { } private void maybeRunOperationOnPendingTransactionsComplete() { - if (runOnPendingTransactionsComplete != null && pendingTransactions.isEmpty()) { - LOG.debug("{}: Pending transactions complete - running operation {}", logContext, - runOnPendingTransactionsComplete); - - runOnPendingTransactionsComplete.run(); - runOnPendingTransactionsComplete = null; - } - } + if (runOnPendingTransactionsComplete != null && !anyPendingTransactions()) { + LOG.debug("{}: Pending transactions complete - running operation {}", logContext, + runOnPendingTransactionsComplete); + + runOnPendingTransactionsComplete.run(); + runOnPendingTransactionsComplete = null; + } + } }