*/
package org.opendaylight.controller.cluster.datastore;
-import static com.google.common.base.Verify.verify;
import static java.util.Objects.requireNonNull;
import com.google.common.base.MoreObjects;
import com.google.common.base.MoreObjects.ToStringHelper;
+import com.google.common.base.VerifyException;
import com.google.common.collect.Collections2;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableRangeSet;
-import com.google.common.collect.RangeSet;
-import com.google.common.primitives.UnsignedLong;
-import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.persisted.FrontendClientMetadata;
-import org.opendaylight.controller.cluster.datastore.persisted.FrontendHistoryMetadata;
-import org.opendaylight.controller.cluster.datastore.utils.UnsignedLongSet;
-import org.opendaylight.yangtools.concepts.Builder;
-import org.opendaylight.yangtools.concepts.Identifiable;
+import org.opendaylight.controller.cluster.datastore.utils.ImmutableUnsignedLongSet;
+import org.opendaylight.controller.cluster.datastore.utils.MutableUnsignedLongSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* This class is NOT thread-safe.
*/
-abstract class FrontendClientMetadataBuilder implements Builder<FrontendClientMetadata>,
- Identifiable<ClientIdentifier> {
+abstract sealed class FrontendClientMetadataBuilder {
static final class Disabled extends FrontendClientMetadataBuilder {
- Disabled(final String shardName, final ClientIdentifier identifier) {
- super(shardName, identifier);
+ Disabled(final String shardName, final ClientIdentifier clientId) {
+ super(shardName, clientId);
}
@Override
- public FrontendClientMetadata build() {
- return new FrontendClientMetadata(getIdentifier(), ImmutableRangeSet.of(), ImmutableList.of());
+ FrontendClientMetadata build() {
+ return new FrontendClientMetadata(clientId(), ImmutableUnsignedLongSet.of(), ImmutableList.of());
}
@Override
// No-op
}
+ @Override
+ void onTransactionsSkipped(final LocalHistoryIdentifier historyId, final ImmutableUnsignedLongSet txIds) {
+ // No-op
+ }
+
@Override
LeaderFrontendState toLeaderState(final Shard shard) {
- return new LeaderFrontendState.Disabled(shard.persistenceId(), getIdentifier(), shard.getDataStore());
+ return new LeaderFrontendState.Disabled(shard.persistenceId(), clientId(), shard.getDataStore());
}
}
static final class Enabled extends FrontendClientMetadataBuilder {
private final Map<LocalHistoryIdentifier, FrontendHistoryMetadataBuilder> currentHistories = new HashMap<>();
+ private final MutableUnsignedLongSet purgedHistories;
private final LocalHistoryIdentifier standaloneId;
- private final UnsignedLongSet purgedHistories;
- Enabled(final String shardName, final ClientIdentifier identifier) {
- super(shardName, identifier);
+ Enabled(final String shardName, final ClientIdentifier clientId) {
+ super(shardName, clientId);
- purgedHistories = UnsignedLongSet.of();
+ purgedHistories = MutableUnsignedLongSet.of();
// History for stand-alone transactions is always present
standaloneId = standaloneHistoryId();
}
Enabled(final String shardName, final FrontendClientMetadata meta) {
- super(shardName, meta.getIdentifier());
+ super(shardName, meta.clientId());
- purgedHistories = UnsignedLongSet.of(meta.getPurgedHistories());
- for (FrontendHistoryMetadata h : meta.getCurrentHistories()) {
- final FrontendHistoryMetadataBuilder b = new FrontendHistoryMetadataBuilder(getIdentifier(), h);
- currentHistories.put(b.getIdentifier(), b);
+ purgedHistories = meta.getPurgedHistories().mutableCopy();
+ for (var historyMeta : meta.getCurrentHistories()) {
+ final var builder = new FrontendHistoryMetadataBuilder(clientId(), historyMeta);
+ currentHistories.put(builder.getIdentifier(), builder);
}
// Sanity check and recovery
standaloneId = standaloneHistoryId();
if (!currentHistories.containsKey(standaloneId)) {
LOG.warn("{}: Client {} recovered histories {} do not contain stand-alone history, attempting recovery",
- shardName, getIdentifier(), currentHistories);
+ shardName, clientId(), currentHistories);
currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId));
}
}
@Override
- public FrontendClientMetadata build() {
- return new FrontendClientMetadata(getIdentifier(), purgedHistories.toRangeSet(),
+ FrontendClientMetadata build() {
+ return new FrontendClientMetadata(clientId(), purgedHistories.immutableCopy(),
Collections2.transform(currentHistories.values(), FrontendHistoryMetadataBuilder::build));
}
@Override
void onHistoryCreated(final LocalHistoryIdentifier historyId) {
- final FrontendHistoryMetadataBuilder newMeta = new FrontendHistoryMetadataBuilder(historyId);
- final FrontendHistoryMetadataBuilder oldMeta = currentHistories.putIfAbsent(historyId, newMeta);
+ final var newMeta = new FrontendHistoryMetadataBuilder(historyId);
+ final var oldMeta = currentHistories.putIfAbsent(historyId, newMeta);
if (oldMeta != null) {
// This should not be happening, warn about it
LOG.warn("{}: Reused local history {}", shardName(), historyId);
@Override
void onHistoryClosed(final LocalHistoryIdentifier historyId) {
- final FrontendHistoryMetadataBuilder builder = currentHistories.get(historyId);
+ final var builder = currentHistories.get(historyId);
if (builder != null) {
builder.onHistoryClosed();
LOG.debug("{}: Closed history {}", shardName(), historyId);
@Override
void onHistoryPurged(final LocalHistoryIdentifier historyId) {
- final FrontendHistoryMetadataBuilder history = currentHistories.remove(historyId);
+ final var history = currentHistories.remove(historyId);
final long historyBits = historyId.getHistoryId();
if (history == null) {
if (!purgedHistories.contains(historyBits)) {
@Override
void onTransactionAborted(final TransactionIdentifier txId) {
- final FrontendHistoryMetadataBuilder history = getHistory(txId);
+ final var history = getHistory(txId);
if (history != null) {
history.onTransactionAborted(txId);
LOG.debug("{}: Aborted transaction {}", shardName(), txId);
@Override
void onTransactionCommitted(final TransactionIdentifier txId) {
- final FrontendHistoryMetadataBuilder history = getHistory(txId);
+ final var history = getHistory(txId);
if (history != null) {
history.onTransactionCommitted(txId);
LOG.debug("{}: Committed transaction {}", shardName(), txId);
@Override
void onTransactionPurged(final TransactionIdentifier txId) {
- final FrontendHistoryMetadataBuilder history = getHistory(txId);
+ final var history = getHistory(txId);
if (history != null) {
history.onTransactionPurged(txId);
LOG.debug("{}: Purged transaction {}", shardName(), txId);
}
}
+ @Override
+ void onTransactionsSkipped(final LocalHistoryIdentifier historyId, final ImmutableUnsignedLongSet txIds) {
+ final FrontendHistoryMetadataBuilder history = getHistory(historyId);
+ if (history != null) {
+ history.onTransactionsSkipped(txIds);
+ LOG.debug("{}: History {} skipped transactions {}", shardName(), historyId, txIds);
+ } else {
+ LOG.warn("{}: Unknown history {} for skipped transactions, ignoring", shardName(), historyId);
+ }
+ }
+
@Override
LeaderFrontendState toLeaderState(final Shard shard) {
// Note: we have to make sure to *copy* all current state and not leak any views, otherwise leader/follower
// interactions would get intertwined leading to inconsistencies.
- final Map<LocalHistoryIdentifier, LocalFrontendHistory> histories = new HashMap<>();
- for (FrontendHistoryMetadataBuilder e : currentHistories.values()) {
- if (e.getIdentifier().getHistoryId() != 0) {
- final AbstractFrontendHistory state = e.toLeaderState(shard);
- verify(state instanceof LocalFrontendHistory, "Unexpected state %s", state);
- histories.put(e.getIdentifier(), (LocalFrontendHistory) state);
+ final var histories = new HashMap<LocalHistoryIdentifier, LocalFrontendHistory>();
+ for (var historyMetaBuilder : currentHistories.values()) {
+ final var historyId = historyMetaBuilder.getIdentifier();
+ if (historyId.getHistoryId() != 0) {
+ final var state = historyMetaBuilder.toLeaderState(shard);
+ if (state instanceof LocalFrontendHistory localState) {
+ histories.put(historyId, localState);
+ } else {
+ throw new VerifyException("Unexpected state " + state);
+ }
}
}
final AbstractFrontendHistory singleHistory;
- final FrontendHistoryMetadataBuilder singleHistoryMeta = currentHistories.get(
- new LocalHistoryIdentifier(getIdentifier(), 0));
+ final var singleHistoryMeta = currentHistories.get(new LocalHistoryIdentifier(clientId(), 0));
if (singleHistoryMeta == null) {
- final ShardDataTree tree = shard.getDataStore();
- singleHistory = StandaloneFrontendHistory.create(shard.persistenceId(), getIdentifier(), tree);
+ final var tree = shard.getDataStore();
+ singleHistory = StandaloneFrontendHistory.create(shard.persistenceId(), clientId(), tree);
} else {
singleHistory = singleHistoryMeta.toLeaderState(shard);
}
- return new LeaderFrontendState.Enabled(shard.persistenceId(), getIdentifier(), shard.getDataStore(),
- purgedHistories.copy(), singleHistory, histories);
+ return new LeaderFrontendState.Enabled(shard.persistenceId(), clientId(), shard.getDataStore(),
+ purgedHistories.mutableCopy(), singleHistory, histories);
}
@Override
}
private FrontendHistoryMetadataBuilder getHistory(final TransactionIdentifier txId) {
- LocalHistoryIdentifier historyId = txId.getHistoryId();
+ return getHistory(txId.getHistoryId());
+ }
+
+ private FrontendHistoryMetadataBuilder getHistory(final LocalHistoryIdentifier historyId) {
+ final LocalHistoryIdentifier local;
if (historyId.getHistoryId() == 0 && historyId.getCookie() != 0) {
// We are pre-creating the history for free-standing transactions with a zero cookie, hence our lookup
// needs to account for that.
LOG.debug("{}: looking up {} instead of {}", shardName(), standaloneId, historyId);
- historyId = standaloneId;
+ local = standaloneId;
+ } else {
+ local = historyId;
}
- return currentHistories.get(historyId);
+ return currentHistories.get(local);
}
private LocalHistoryIdentifier standaloneHistoryId() {
- return new LocalHistoryIdentifier(getIdentifier(), 0);
+ return new LocalHistoryIdentifier(clientId(), 0);
}
}
private static final Logger LOG = LoggerFactory.getLogger(FrontendClientMetadataBuilder.class);
- private final @NonNull ClientIdentifier identifier;
+ private final @NonNull ClientIdentifier clientId;
private final @NonNull String shardName;
- FrontendClientMetadataBuilder(final String shardName, final ClientIdentifier identifier) {
+ FrontendClientMetadataBuilder(final String shardName, final ClientIdentifier clientId) {
this.shardName = requireNonNull(shardName);
- this.identifier = requireNonNull(identifier);
+ this.clientId = requireNonNull(clientId);
}
static FrontendClientMetadataBuilder of(final String shardName, final FrontendClientMetadata meta) {
- final Collection<FrontendHistoryMetadata> current = meta.getCurrentHistories();
- final RangeSet<UnsignedLong> purged = meta.getPurgedHistories();
-
// Completely empty histories imply disabled state, as otherwise we'd have a record of the single history --
// either purged or active
- return current.isEmpty() && purged.isEmpty() ? new Disabled(shardName, meta.getIdentifier())
- : new Enabled(shardName, meta);
+ return meta.getCurrentHistories().isEmpty() && meta.getPurgedHistories().isEmpty()
+ ? new Disabled(shardName, meta.clientId()) : new Enabled(shardName, meta);
}
- @Override
- public final ClientIdentifier getIdentifier() {
- return identifier;
+ final ClientIdentifier clientId() {
+ return clientId;
}
final String shardName() {
return shardName;
}
+ abstract FrontendClientMetadata build();
+
abstract void onHistoryCreated(LocalHistoryIdentifier historyId);
abstract void onHistoryClosed(LocalHistoryIdentifier historyId);
abstract void onTransactionPurged(TransactionIdentifier txId);
+ abstract void onTransactionsSkipped(LocalHistoryIdentifier historyId, ImmutableUnsignedLongSet txIds);
+
/**
* Transform frontend metadata for a particular client into its {@link LeaderFrontendState} counterpart.
*
}
ToStringHelper addToStringAttributes(final ToStringHelper helper) {
- return helper.add("identifier", identifier);
+ return helper.add("clientId", clientId);
}
}
Code Review / controller.git / blobdiff