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.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
public ShardDataTree(final Shard shard, final SchemaContext schemaContext, final TreeType treeType) {
this(shard, schemaContext, treeType, YangInstanceIdentifier.EMPTY,
- new DefaultShardDataTreeChangeListenerPublisher(), new DefaultShardDataChangeListenerPublisher(), "");
+ new DefaultShardDataTreeChangeListenerPublisher(""),
+ new DefaultShardDataChangeListenerPublisher(""), "");
}
final String logContext() {
} else {
allMetadataClosedLocalHistory(((CloseLocalHistoryPayload) payload).getIdentifier());
}
- } else if (payload instanceof CloseLocalHistoryPayload) {
- if (identifier != null) {
- payloadReplicationComplete((CloseLocalHistoryPayload) payload);
- } else {
- allMetadataClosedLocalHistory(((CloseLocalHistoryPayload) payload).getIdentifier());
- }
} else if (payload instanceof CreateLocalHistoryPayload) {
if (identifier != null) {
payloadReplicationComplete((CreateLocalHistoryPayload)payload);
@VisibleForTesting
public void notifyListeners(final DataTreeCandidate candidate) {
- treeChangeListenerPublisher.publishChanges(candidate, logContext);
- 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);
- }
+ treeChangeListenerPublisher.publishChanges(candidate);
+ dataChangeListenerPublisher.publishChanges(candidate);
}
/**
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() {
replicatePayload(id, AbortTransactionPayload.create(id), callback);
}
-
@Override
- void purgeTransaction(final TransactionIdentifier id, final Runnable callback) {
- LOG.debug("{}: purging transaction {}", logContext, id);
- replicatePayload(id, PurgeTransactionPayload.create(id), callback);
+ 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();
+ }
}
Code Review / controller.git / blobdiff