import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.File;
import java.io.IOException;
-import java.util.AbstractMap.SimpleEntry;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
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.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.MetadataShardDataTreeSnapshot;
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.utils.DataTreeModificationOutput;
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.controller.md.sal.dom.store.impl.DataChangeListenerRegistration;
import org.opendaylight.yangtools.concepts.Identifier;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
private SchemaContext schemaContext;
- public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TipProducingDataTree dataTree,
+ ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TipProducingDataTree dataTree,
final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher,
final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext,
final ShardDataTreeMetadata<?>... metadata) {
tip = dataTree;
}
- public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType,
+ ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType,
final YangInstanceIdentifier root,
final ShardDataTreeChangeListenerPublisher treeChangeListenerPublisher,
- final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext) {
+ final ShardDataChangeListenerPublisher dataChangeListenerPublisher, final String logContext,
+ final ShardDataTreeMetadata<?>... metadata) {
this(shard, schemaContext, InMemoryDataTreeFactory.getInstance().create(treeType, root),
- treeChangeListenerPublisher, dataChangeListenerPublisher, logContext);
+ treeChangeListenerPublisher, dataChangeListenerPublisher, logContext, metadata);
}
@VisibleForTesting
((CommitTransactionPayload) payload).getCandidate();
applyRecoveryCandidate(e.getValue());
allMetadataCommittedTransaction(e.getKey());
+ } else if (payload instanceof AbortTransactionPayload) {
+ allMetadataAbortedTransaction(((AbortTransactionPayload) payload).getIdentifier());
+ } else if (payload instanceof PurgeTransactionPayload) {
+ allMetadataPurgedTransaction(((PurgeTransactionPayload) payload).getIdentifier());
} else if (payload instanceof CreateLocalHistoryPayload) {
allMetadataCreatedLocalHistory(((CreateLocalHistoryPayload) payload).getIdentifier());
} else if (payload instanceof CloseLocalHistoryPayload) {
Verify.verify(identifier instanceof TransactionIdentifier);
payloadReplicationComplete((TransactionIdentifier) identifier);
}
+ } else if (payload instanceof AbortTransactionPayload) {
+ if (identifier != null) {
+ payloadReplicationComplete((AbortTransactionPayload) payload);
+ } else {
+ allMetadataAbortedTransaction(((AbortTransactionPayload) payload).getIdentifier());
+ }
+ } else if (payload instanceof PurgeTransactionPayload) {
+ if (identifier != null) {
+ payloadReplicationComplete((PurgeTransactionPayload) payload);
+ } else {
+ allMetadataPurgedTransaction(((PurgeTransactionPayload) payload).getIdentifier());
+ }
} else if (payload instanceof CloseLocalHistoryPayload) {
if (identifier != null) {
payloadReplicationComplete((CloseLocalHistoryPayload) payload);
finishCommit(current.cohort);
}
+ private void allMetadataAbortedTransaction(final TransactionIdentifier txId) {
+ for (ShardDataTreeMetadata<?> m : metadata) {
+ m.onTransactionAborted(txId);
+ }
+ }
+
private void allMetadataCommittedTransaction(final TransactionIdentifier txId) {
for (ShardDataTreeMetadata<?> m : metadata) {
m.onTransactionCommitted(txId);
}
}
+ private void allMetadataPurgedTransaction(final TransactionIdentifier txId) {
+ for (ShardDataTreeMetadata<?> m : metadata) {
+ m.onTransactionPurged(txId);
+ }
+ }
+
private void allMetadataCreatedLocalHistory(final LocalHistoryIdentifier historyId) {
for (ShardDataTreeMetadata<?> m : metadata) {
m.onHistoryCreated(historyId);
}
}
+ /**
+ * Create a transaction chain for specified history. Unlike {@link #ensureTransactionChain(LocalHistoryIdentifier)},
+ * this method is used for re-establishing state when we are taking over
+ *
+ * @param historyId Local history identifier
+ * @param closed True if the chain should be created in closed state (i.e. pending purge)
+ * @return Transaction chain handle
+ */
+ ShardDataTreeTransactionChain recreateTransactionChain(final LocalHistoryIdentifier historyId,
+ 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);
+ return ret;
+ }
+
ShardDataTreeTransactionChain ensureTransactionChain(final LocalHistoryIdentifier historyId) {
ShardDataTreeTransactionChain chain = transactionChains.get(historyId);
if (chain == null) {
ReadOnlyShardDataTreeTransaction newReadOnlyTransaction(final TransactionIdentifier txId) {
if (txId.getHistoryId().getHistoryId() == 0) {
- return new ReadOnlyShardDataTreeTransaction(txId, dataTree.takeSnapshot());
+ return new ReadOnlyShardDataTreeTransaction(this, txId, dataTree.takeSnapshot());
}
return ensureTransactionChain(txId.getHistoryId()).newReadOnlyTransaction(txId);
dataChangeListenerPublisher.publishChanges(candidate, logContext);
}
- void notifyOfInitialData(final DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
- NormalizedNode<?, ?>>> listenerReg, final Optional<DataTreeCandidate> currentState) {
- if (currentState.isPresent()) {
- ShardDataChangeListenerPublisher localPublisher = dataChangeListenerPublisher.newInstance();
- localPublisher.registerDataChangeListener(listenerReg.getPath(), listenerReg.getInstance(),
- listenerReg.getScope());
- localPublisher.publishChanges(currentState.get(), logContext);
- }
- }
-
- void notifyOfInitialData(final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener,
- final Optional<DataTreeCandidate> currentState) {
- if (currentState.isPresent()) {
- ShardDataTreeChangeListenerPublisher localPublisher = treeChangeListenerPublisher.newInstance();
- localPublisher.registerTreeChangeListener(path, listener);
- localPublisher.publishChanges(currentState.get(), logContext);
- }
- }
-
/**
- * Immediately close all transaction chains.
+ * Immediately purge all state relevant to leader. This includes all transaction chains and any scheduled
+ * replication callbacks.
*/
- void closeAllTransactionChains() {
+ void purgeLeaderState() {
for (ShardDataTreeTransactionChain chain : transactionChains.values()) {
chain.close();
}
transactionChains.clear();
+ replicationCallbacks.clear();
}
/**
replicatePayload(id, PurgeLocalHistoryPayload.create(id), callback);
}
- Entry<DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>,
- Optional<DataTreeCandidate>> registerChangeListener(final YangInstanceIdentifier path,
- final AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener,
- final DataChangeScope scope) {
- DataChangeListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> reg =
- dataChangeListenerPublisher.registerDataChangeListener(path, listener, scope);
-
- return new SimpleEntry<>(reg, readCurrentData());
+ 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);
}
- private Optional<DataTreeCandidate> readCurrentData() {
+ 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();
}
- public Entry<ListenerRegistration<DOMDataTreeChangeListener>, Optional<DataTreeCandidate>>
- registerTreeChangeListener(final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener) {
- final ListenerRegistration<DOMDataTreeChangeListener> reg =
- treeChangeListenerPublisher.registerTreeChangeListener(path, listener);
-
- return new SimpleEntry<>(reg, readCurrentData());
+ public void registerTreeChangeListener(final YangInstanceIdentifier path, final DOMDataTreeChangeListener listener,
+ final Optional<DataTreeCandidate> initialState,
+ final Consumer<ListenerRegistration<DOMDataTreeChangeListener>> onRegistration) {
+ treeChangeListenerPublisher.registerTreeChangeListener(path, listener, initialState, onRegistration);
}
int getQueueSize() {
}
@Override
- void abortTransaction(final AbstractShardDataTreeTransaction<?> transaction) {
- // Intentional no-op
+ 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);
+ }
+
+ @Override
+ void abortFromTransactionActor(final AbstractShardDataTreeTransaction<?> transaction) {
+ // No-op for free-standing transactions
+
}
@Override
return createReadyCohort(transaction.getIdentifier(), snapshot);
}
+ void purgeTransaction(final TransactionIdentifier id, final Runnable callback) {
+ LOG.debug("{}: purging transaction {}", logContext, id);
+ replicatePayload(id, PurgeTransactionPayload.create(id), callback);
+ }
+
public Optional<NormalizedNode<?, ?>> readNode(final YangInstanceIdentifier path) {
return dataTree.takeSnapshot().readNode(path);
}
LOG.debug("{}: Validating transaction {}", logContext, cohort.getIdentifier());
Exception cause;
try {
+ cohort.throwCanCommitFailure();
+
tip.validate(modification);
LOG.debug("{}: Transaction {} validated", logContext, cohort.getIdentifier());
cohort.successfulCanCommit();
}
@Override
- ShardDataTreeCohort createReadyCohort(final TransactionIdentifier txId,
- final DataTreeModification modification) {
- SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(this, modification, txId,
+ ShardDataTreeCohort createFailedCohort(final TransactionIdentifier txId, final DataTreeModification mod,
+ final Exception failure) {
+ SimpleShardDataTreeCohort cohort = new SimpleShardDataTreeCohort.DeadOnArrival(this, mod, txId, failure);
+ pendingTransactions.add(new CommitEntry(cohort, ticker().read()));
+ 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));
pendingTransactions.add(new CommitEntry(cohort, ticker().read()));
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) {
+ if (txId.getHistoryId().getHistoryId() == 0) {
+ return createReadyCohort(txId, mod);
+ }
+
+ return ensureTransactionChain(txId.getHistoryId()).createReadyCohort(txId, mod);
+ }
+
@SuppressFBWarnings(value = "DB_DUPLICATE_SWITCH_CLAUSES", justification = "See inline comments below.")
void checkForExpiredTransactions(final long transactionCommitTimeoutMillis) {
final long timeout = TimeUnit.MILLISECONDS.toNanos(transactionCommitTimeoutMillis);
runOnPendingTransactionsComplete = null;
}
}
+
+ ShardStats getStats() {
+ return shard.getShardMBean();
+ }
}