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%2FFrontendClientMetadataBuilder.java;h=7e6eced779bdf4026dca36134b2b13e7509f1bbb;hp=4ffa27c60dcdb771767d836490d2af333b2c5f9f;hb=HEAD;hpb=a7aab5b8d7692d12928944a0d3da818571e324cc diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendClientMetadataBuilder.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendClientMetadataBuilder.java index 4ffa27c60d..c89627800f 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendClientMetadataBuilder.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendClientMetadataBuilder.java @@ -7,179 +7,316 @@ */ package org.opendaylight.controller.cluster.datastore; +import static java.util.Objects.requireNonNull; + import com.google.common.base.MoreObjects; -import com.google.common.base.Preconditions; -import com.google.common.base.Verify; +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 java.util.HashMap; import java.util.Map; -import javax.annotation.Nonnull; -import javax.annotation.concurrent.NotThreadSafe; +import org.eclipse.jdt.annotation.NonNull; import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier; 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.UnsignedLongRangeSet; -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; -@NotThreadSafe -final class FrontendClientMetadataBuilder implements Builder, Identifiable { - private static final Logger LOG = LoggerFactory.getLogger(FrontendClientMetadataBuilder.class); +/** + * This class is NOT thread-safe. + */ +abstract sealed class FrontendClientMetadataBuilder { + static final class Disabled extends FrontendClientMetadataBuilder { + Disabled(final String shardName, final ClientIdentifier clientId) { + super(shardName, clientId); + } - private final Map currentHistories = new HashMap<>(); - private final UnsignedLongRangeSet purgedHistories; - private final ClientIdentifier identifier; - private final String shardName; + @Override + FrontendClientMetadata build() { + return new FrontendClientMetadata(clientId(), ImmutableUnsignedLongSet.of(), ImmutableList.of()); + } - FrontendClientMetadataBuilder(final String shardName, final ClientIdentifier identifier) { - this.shardName = Preconditions.checkNotNull(shardName); - this.identifier = Preconditions.checkNotNull(identifier); - purgedHistories = UnsignedLongRangeSet.create(); + @Override + void onHistoryCreated(final LocalHistoryIdentifier historyId) { + // No-op + } - // History for stand-alone transactions is always present - final LocalHistoryIdentifier standaloneId = standaloneHistoryId(); - currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId)); - } + @Override + void onHistoryClosed(final LocalHistoryIdentifier historyId) { + // No-op + } - FrontendClientMetadataBuilder(final String shardName, final FrontendClientMetadata meta) { - this.shardName = Preconditions.checkNotNull(shardName); - this.identifier = Preconditions.checkNotNull(meta.getIdentifier()); - purgedHistories = UnsignedLongRangeSet.create(meta.getPurgedHistories()); + @Override + void onHistoryPurged(final LocalHistoryIdentifier historyId) { + // No-op + } - for (FrontendHistoryMetadata h : meta.getCurrentHistories()) { - final FrontendHistoryMetadataBuilder b = new FrontendHistoryMetadataBuilder(identifier, h); - currentHistories.put(b.getIdentifier(), b); + @Override + void onTransactionAborted(final TransactionIdentifier txId) { + // No-op } - // Sanity check and recovery - final LocalHistoryIdentifier standaloneId = standaloneHistoryId(); - if (!currentHistories.containsKey(standaloneId)) { - LOG.warn("{}: Client {} recovered histories {} do not contain stand-alone history, attempting recovery", - shardName, identifier, currentHistories); - currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId)); + @Override + void onTransactionCommitted(final TransactionIdentifier txId) { + // No-op } - } - private LocalHistoryIdentifier standaloneHistoryId() { - return new LocalHistoryIdentifier(identifier, 0); - } + @Override + void onTransactionPurged(final TransactionIdentifier txId) { + // No-op + } - @Override - public FrontendClientMetadata build() { - return new FrontendClientMetadata(identifier, purgedHistories.toImmutable(), - Collections2.transform(currentHistories.values(), FrontendHistoryMetadataBuilder::build)); - } + @Override + void onTransactionsSkipped(final LocalHistoryIdentifier historyId, final ImmutableUnsignedLongSet txIds) { + // No-op + } - @Override - public ClientIdentifier getIdentifier() { - return identifier; + @Override + LeaderFrontendState toLeaderState(final Shard shard) { + return new LeaderFrontendState.Disabled(shard.persistenceId(), clientId(), shard.getDataStore()); + } } - void onHistoryCreated(final LocalHistoryIdentifier historyId) { - final FrontendHistoryMetadataBuilder newMeta = new FrontendHistoryMetadataBuilder(historyId); - final FrontendHistoryMetadataBuilder oldMeta = currentHistories.putIfAbsent(historyId, newMeta); - if (oldMeta != null) { - // This should not be happening, warn about it - LOG.warn("{}: Reused local history {}", shardName, historyId); - } else { - LOG.debug("{}: Created local history {}", shardName, historyId); + static final class Enabled extends FrontendClientMetadataBuilder { + private final Map currentHistories = new HashMap<>(); + private final MutableUnsignedLongSet purgedHistories; + private final LocalHistoryIdentifier standaloneId; + + Enabled(final String shardName, final ClientIdentifier clientId) { + super(shardName, clientId); + + purgedHistories = MutableUnsignedLongSet.of(); + + // History for stand-alone transactions is always present + standaloneId = standaloneHistoryId(); + currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId)); } - } - void onHistoryClosed(final LocalHistoryIdentifier historyId) { - final FrontendHistoryMetadataBuilder builder = currentHistories.get(historyId); - if (builder != null) { - builder.onHistoryClosed(); - LOG.debug("{}: Closed history {}", shardName, historyId); - } else { - LOG.warn("{}: Closed unknown history {}, ignoring", shardName, historyId); + Enabled(final String shardName, final FrontendClientMetadata meta) { + super(shardName, meta.clientId()); + + 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, clientId(), currentHistories); + currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId)); + } } - } - void onHistoryPurged(final LocalHistoryIdentifier historyId) { - final FrontendHistoryMetadataBuilder history = currentHistories.remove(historyId); - if (history == null) { - LOG.warn("{}: Purging unknown history {}", shardName, historyId); + @Override + FrontendClientMetadata build() { + return new FrontendClientMetadata(clientId(), purgedHistories.immutableCopy(), + Collections2.transform(currentHistories.values(), FrontendHistoryMetadataBuilder::build)); } - // XXX: do we need to account for cookies? - purgedHistories.add(historyId.getHistoryId()); - LOG.debug("{}: Purged history {}", historyId); - } + @Override + void onHistoryCreated(final LocalHistoryIdentifier historyId) { + 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); + } else { + LOG.debug("{}: Created local history {}", shardName(), historyId); + } + } - void onTransactionAborted(final TransactionIdentifier txId) { - final FrontendHistoryMetadataBuilder history = getHistory(txId); - if (history != null) { - history.onTransactionAborted(txId); - LOG.debug("{}: Committed transaction {}", shardName, txId); - } else { - LOG.warn("{}: Unknown history for aborted transaction {}, ignoring", shardName, txId); + @Override + void onHistoryClosed(final LocalHistoryIdentifier historyId) { + final var builder = currentHistories.get(historyId); + if (builder != null) { + builder.onHistoryClosed(); + LOG.debug("{}: Closed history {}", shardName(), historyId); + } else { + LOG.warn("{}: Closed unknown history {}, ignoring", shardName(), historyId); + } } - } - void onTransactionCommitted(final TransactionIdentifier txId) { - final FrontendHistoryMetadataBuilder history = getHistory(txId); - if (history != null) { - history.onTransactionCommitted(txId); - LOG.debug("{}: Aborted transaction {}", shardName, txId); - } else { - LOG.warn("{}: Unknown history for commited transaction {}, ignoring", shardName, txId); + @Override + void onHistoryPurged(final LocalHistoryIdentifier historyId) { + final var history = currentHistories.remove(historyId); + final long historyBits = historyId.getHistoryId(); + if (history == null) { + if (!purgedHistories.contains(historyBits)) { + purgedHistories.add(historyBits); + LOG.warn("{}: Purging unknown history {}", shardName(), historyId); + } else { + LOG.warn("{}: Duplicate purge of history {}", shardName(), historyId); + } + } else { + purgedHistories.add(historyBits); + LOG.debug("{}: Purged history {}", shardName(), historyId); + } } - } - void onTransactionPurged(final TransactionIdentifier txId) { - final FrontendHistoryMetadataBuilder history = getHistory(txId); - if (history != null) { - history.onTransactionPurged(txId); - LOG.debug("{}: Purged transaction {}", shardName, txId); - } else { - LOG.warn("{}: Unknown history for purged transaction {}, ignoring", shardName, txId); + @Override + void onTransactionAborted(final TransactionIdentifier txId) { + final var history = getHistory(txId); + if (history != null) { + history.onTransactionAborted(txId); + LOG.debug("{}: Aborted transaction {}", shardName(), txId); + } else { + LOG.warn("{}: Unknown history for aborted transaction {}, ignoring", shardName(), txId); + } } - } - /** - * Transform frontend metadata for a particular client into its {@link LeaderFrontendState} counterpart. - * - * @param shard parent shard - * @return Leader frontend state - */ - @Nonnull LeaderFrontendState toLeaderState(@Nonnull 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 histories = new HashMap<>(); - for (FrontendHistoryMetadataBuilder e : currentHistories.values()) { - if (e.getIdentifier().getHistoryId() != 0) { - final AbstractFrontendHistory state = e.toLeaderState(shard); - Verify.verify(state instanceof LocalFrontendHistory); - histories.put(e.getIdentifier(), (LocalFrontendHistory) state); + @Override + void onTransactionCommitted(final TransactionIdentifier txId) { + final var history = getHistory(txId); + if (history != null) { + history.onTransactionCommitted(txId); + LOG.debug("{}: Committed transaction {}", shardName(), txId); + } else { + LOG.warn("{}: Unknown history for commited transaction {}, ignoring", shardName(), txId); + } + } + + @Override + void onTransactionPurged(final TransactionIdentifier txId) { + final var history = getHistory(txId); + if (history != null) { + history.onTransactionPurged(txId); + LOG.debug("{}: Purged transaction {}", shardName(), txId); + } else { + LOG.warn("{}: Unknown history for purged transaction {}, ignoring", 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 var histories = new HashMap(); + 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 var singleHistoryMeta = currentHistories.get(new LocalHistoryIdentifier(clientId(), 0)); + if (singleHistoryMeta == null) { + final var tree = shard.getDataStore(); + singleHistory = StandaloneFrontendHistory.create(shard.persistenceId(), clientId(), tree); + } else { + singleHistory = singleHistoryMeta.toLeaderState(shard); + } + + return new LeaderFrontendState.Enabled(shard.persistenceId(), clientId(), shard.getDataStore(), + purgedHistories.mutableCopy(), singleHistory, histories); + } + + @Override + ToStringHelper addToStringAttributes(final ToStringHelper helper) { + return super.addToStringAttributes(helper).add("current", currentHistories).add("purged", purgedHistories); + } + + private FrontendHistoryMetadataBuilder getHistory(final TransactionIdentifier txId) { + 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); + local = standaloneId; + } else { + local = historyId; } + + return currentHistories.get(local); } - final AbstractFrontendHistory singleHistory; - final FrontendHistoryMetadataBuilder singleHistoryMeta = currentHistories.get( - new LocalHistoryIdentifier(identifier, 0)); - if (singleHistoryMeta == null) { - final ShardDataTree tree = shard.getDataStore(); - singleHistory = StandaloneFrontendHistory.create(shard.persistenceId(), getIdentifier(), tree); - } else { - singleHistory = singleHistoryMeta.toLeaderState(shard); + private LocalHistoryIdentifier standaloneHistoryId() { + return new LocalHistoryIdentifier(clientId(), 0); } + } + + private static final Logger LOG = LoggerFactory.getLogger(FrontendClientMetadataBuilder.class); - return new LeaderFrontendState(shard.persistenceId(), getIdentifier(), shard.getDataStore(), - purgedHistories.copy(), singleHistory, histories); + private final @NonNull ClientIdentifier clientId; + private final @NonNull String shardName; + + FrontendClientMetadataBuilder(final String shardName, final ClientIdentifier clientId) { + this.shardName = requireNonNull(shardName); + this.clientId = requireNonNull(clientId); + } + + static FrontendClientMetadataBuilder of(final String shardName, final FrontendClientMetadata meta) { + // Completely empty histories imply disabled state, as otherwise we'd have a record of the single history -- + // either purged or active + return meta.getCurrentHistories().isEmpty() && meta.getPurgedHistories().isEmpty() + ? new Disabled(shardName, meta.clientId()) : new Enabled(shardName, meta); } - private FrontendHistoryMetadataBuilder getHistory(final TransactionIdentifier txId) { - return currentHistories.get(txId.getHistoryId()); + 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 onHistoryPurged(LocalHistoryIdentifier historyId); + + abstract void onTransactionAborted(TransactionIdentifier txId); + + abstract void onTransactionCommitted(TransactionIdentifier txId); + + abstract void onTransactionPurged(TransactionIdentifier txId); + + abstract void onTransactionsSkipped(LocalHistoryIdentifier historyId, ImmutableUnsignedLongSet txIds); + + /** + * Transform frontend metadata for a particular client into its {@link LeaderFrontendState} counterpart. + * + * @param shard parent shard + * @return Leader frontend state + */ + abstract @NonNull LeaderFrontendState toLeaderState(@NonNull Shard shard); + @Override - public String toString() { - return MoreObjects.toStringHelper(this).add("identifier", identifier).add("current", currentHistories) - .add("purged", purgedHistories).toString(); + public final String toString() { + return addToStringAttributes(MoreObjects.toStringHelper(this)).toString(); + } + + ToStringHelper addToStringAttributes(final ToStringHelper helper) { + return helper.add("clientId", clientId); } }