X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FAbstractFrontendHistory.java;h=31bf000ac2123a0bcc0c573fbedc84fba7247bc3;hb=1d5ca4009be6c61d7b61989799037ad8f1ab7a75;hp=609751b4bdcdd5efd67b5031d138f914de1815a0;hpb=b93eba249632ea61c59c8182bf938bde6f2a34fe;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractFrontendHistory.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractFrontendHistory.java index 609751b4bd..31bf000ac2 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractFrontendHistory.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractFrontendHistory.java @@ -7,16 +7,17 @@ */ 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.collect.ImmutableMap; -import com.google.common.collect.Range; -import com.google.common.collect.RangeSet; import com.google.common.primitives.UnsignedLong; +import java.util.ArrayList; import java.util.HashMap; import java.util.Map; import java.util.Optional; -import javax.annotation.Nullable; +import java.util.SortedSet; +import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.controller.cluster.access.commands.AbstractReadTransactionRequest; import org.opendaylight.controller.cluster.access.commands.ClosedTransactionException; import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest; @@ -24,6 +25,8 @@ import org.opendaylight.controller.cluster.access.commands.DeadTransactionExcept import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest; import org.opendaylight.controller.cluster.access.commands.LocalHistorySuccess; import org.opendaylight.controller.cluster.access.commands.OutOfOrderRequestException; +import org.opendaylight.controller.cluster.access.commands.SkipTransactionsRequest; +import org.opendaylight.controller.cluster.access.commands.SkipTransactionsResponse; import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest; import org.opendaylight.controller.cluster.access.commands.TransactionPurgeResponse; import org.opendaylight.controller.cluster.access.commands.TransactionRequest; @@ -32,8 +35,9 @@ import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifie import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope; import org.opendaylight.controller.cluster.access.concepts.RequestException; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; +import org.opendaylight.controller.cluster.datastore.utils.MutableUnsignedLongSet; import org.opendaylight.yangtools.concepts.Identifiable; -import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; +import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -47,7 +51,7 @@ abstract class AbstractFrontendHistory implements Identifiable transactions = new HashMap<>(); - private final RangeSet purgedTransactions; + private final MutableUnsignedLongSet purgedTransactions; private final String persistenceId; private final ShardDataTree tree; @@ -58,11 +62,11 @@ abstract class AbstractFrontendHistory implements Identifiable closedTransactions; AbstractFrontendHistory(final String persistenceId, final ShardDataTree tree, - final Map closedTransactions, final RangeSet purgedTransactions) { - this.persistenceId = Preconditions.checkNotNull(persistenceId); - this.tree = Preconditions.checkNotNull(tree); - this.closedTransactions = Preconditions.checkNotNull(closedTransactions); - this.purgedTransactions = Preconditions.checkNotNull(purgedTransactions); + final Map closedTransactions, final MutableUnsignedLongSet purgedTransactions) { + this.persistenceId = requireNonNull(persistenceId); + this.tree = requireNonNull(tree); + this.closedTransactions = requireNonNull(closedTransactions); + this.purgedTransactions = requireNonNull(purgedTransactions); } final String persistenceId() { @@ -70,63 +74,27 @@ abstract class AbstractFrontendHistory implements Identifiable handleTransactionRequest(final TransactionRequest request, final RequestEnvelope envelope, final long now) throws RequestException { - final TransactionIdentifier id = request.getTarget(); - final UnsignedLong ul = UnsignedLong.fromLongBits(id.getTransactionId()); - if (request instanceof TransactionPurgeRequest) { - if (purgedTransactions.contains(ul)) { - // Retransmitted purge request: nothing to do - LOG.debug("{}: transaction {} already purged", persistenceId, id); - return new TransactionPurgeResponse(id, request.getSequence()); - } - - // We perform two lookups instead of a straight remove, because once the map becomes empty we switch it - // to an ImmutableMap, which does not allow remove(). - if (closedTransactions.containsKey(ul)) { - tree.purgeTransaction(id, () -> { - closedTransactions.remove(ul); - if (closedTransactions.isEmpty()) { - closedTransactions = ImmutableMap.of(); - } - - purgedTransactions.add(Range.singleton(ul)); - LOG.debug("{}: finished purging inherited transaction {}", persistenceId(), id); - envelope.sendSuccess(new TransactionPurgeResponse(id, request.getSequence()), readTime() - now); - }); - return null; - } - - final FrontendTransaction tx = transactions.get(id); - if (tx == null) { - // This should never happen because the purge callback removes the transaction and puts it into - // purged transactions in one go. If it does, we warn about the situation and - LOG.warn("{}: transaction {} not tracked in {}, but not present in active transactions", persistenceId, - id, purgedTransactions); - purgedTransactions.add(Range.singleton(ul)); - return new TransactionPurgeResponse(id, request.getSequence()); - } - - tree.purgeTransaction(id, () -> { - purgedTransactions.add(Range.singleton(ul)); - transactions.remove(id); - LOG.debug("{}: finished purging transaction {}", persistenceId(), id); - envelope.sendSuccess(new TransactionPurgeResponse(id, request.getSequence()), readTime() - now); - }); - return null; + return handleTransactionPurgeRequest((TransactionPurgeRequest) request, envelope, now); + } else if (request instanceof SkipTransactionsRequest) { + return handleSkipTransactionsRequest((SkipTransactionsRequest) request, envelope, now); } - if (purgedTransactions.contains(ul)) { + final TransactionIdentifier id = request.getTarget(); + final long txidBits = id.getTransactionId(); + if (purgedTransactions.contains(txidBits)) { LOG.warn("{}: Request {} is contained purged transactions {}", persistenceId, request, purgedTransactions); - throw new DeadTransactionException(purgedTransactions); + throw new DeadTransactionException(purgedTransactions.toRangeSet()); } - final Boolean closed = closedTransactions.get(ul); + + final Boolean closed = closedTransactions.get(UnsignedLong.fromLongBits(txidBits)); if (closed != null) { - final boolean successful = closed.booleanValue(); + final boolean successful = closed; LOG.debug("{}: Request {} refers to a {} transaction", persistenceId, request, successful ? "successful" : "failed"); throw new ClosedTransactionException(successful); @@ -154,31 +122,118 @@ abstract class AbstractFrontendHistory implements Identifiable { + closedTransactions.remove(ul); + if (closedTransactions.isEmpty()) { + closedTransactions = ImmutableMap.of(); + } + + purgedTransactions.add(txidBits); + LOG.debug("{}: finished purging inherited transaction {}", persistenceId(), id); + envelope.sendSuccess(new TransactionPurgeResponse(id, request.getSequence()), readTime() - now); + }); + return null; + } + + final FrontendTransaction tx = transactions.get(id); + if (tx == null) { + // This should never happen because the purge callback removes the transaction and puts it into + // purged transactions in one go. If it does, we warn about the situation and + LOG.warn("{}: transaction {} not tracked in {}, but not present in active transactions", persistenceId, + id, purgedTransactions); + purgedTransactions.add(txidBits); + return new TransactionPurgeResponse(id, request.getSequence()); + } + + tree.purgeTransaction(id, () -> { + purgedTransactions.add(txidBits); + transactions.remove(id); + LOG.debug("{}: finished purging transaction {}", persistenceId(), id); + envelope.sendSuccess(new TransactionPurgeResponse(id, request.getSequence()), readTime() - now); + }); + + return null; + } + + private SkipTransactionsResponse handleSkipTransactionsRequest(final SkipTransactionsRequest request, + final RequestEnvelope envelope, final long now) throws RequestException { + final var first = request.getTarget(); + final var others = request.getOthers(); + final var ids = new ArrayList(others.size() + 1); + ids.add(UnsignedLong.fromLongBits(first.getTransactionId())); + ids.addAll(others); + + final var it = ids.iterator(); + while (it.hasNext()) { + final var id = it.next(); + final long bits = id.longValue(); + if (purgedTransactions.contains(bits)) { + LOG.warn("{}: history {} tracks {} as purged", persistenceId(), getIdentifier(), id); + it.remove(); + } else if (transactions.containsKey(new TransactionIdentifier(getIdentifier(), bits))) { + LOG.warn("{}: history {} tracks {} as open", persistenceId(), getIdentifier(), id); + it.remove(); + } + } + + if (ids.isEmpty()) { + LOG.debug("{}: history {} completing empty skip request", persistenceId(), getIdentifier()); + return new SkipTransactionsResponse(first, now); + } + + final var transactionIds = MutableUnsignedLongSet.of(ids.stream().mapToLong(UnsignedLong::longValue).toArray()) + .immutableCopy(); + LOG.debug("{}: history {} skipping transactions {}", persistenceId(), getIdentifier(), transactionIds.ranges()); + + tree.skipTransactions(getIdentifier(), transactionIds, () -> { + purgedTransactions.addAll(transactionIds); + envelope.sendSuccess(new TransactionPurgeResponse(first, request.getSequence()), readTime() - now); + }); + return null; + } + + final void destroy(final long sequence, final RequestEnvelope envelope, final long now) { LOG.debug("{}: closing history {}", persistenceId(), getIdentifier()); tree.closeTransactionChain(getIdentifier(), () -> envelope.sendSuccess(new LocalHistorySuccess(getIdentifier(), sequence), readTime() - now)); } - void purge(final long sequence, final RequestEnvelope envelope, final long now) { + final void purge(final long sequence, final RequestEnvelope envelope, final long now) { LOG.debug("{}: purging history {}", persistenceId(), getIdentifier()); tree.purgeTransactionChain(getIdentifier(), () -> envelope.sendSuccess(new LocalHistorySuccess(getIdentifier(), sequence), readTime() - now)); } - private FrontendTransaction createTransaction(final TransactionRequest request, final TransactionIdentifier id) - throws RequestException { + final void retire() { + transactions.values().forEach(FrontendTransaction::retire); + tree.removeTransactionChain(getIdentifier()); + } + + private FrontendTransaction createTransaction(final TransactionRequest request, final TransactionIdentifier id) { if (request instanceof CommitLocalTransactionRequest) { LOG.debug("{}: allocating new ready transaction {}", persistenceId(), id); tree.getStats().incrementReadWriteTransactionCount(); return createReadyTransaction(id, ((CommitLocalTransactionRequest) request).getModification()); } - if (request instanceof AbstractReadTransactionRequest) { - if (((AbstractReadTransactionRequest) request).isSnapshotOnly()) { - LOG.debug("{}: allocating new open snapshot {}", persistenceId(), id); - tree.getStats().incrementReadOnlyTransactionCount(); - return createOpenSnapshot(id); - } + if (request instanceof AbstractReadTransactionRequest + && ((AbstractReadTransactionRequest) request).isSnapshotOnly()) { + LOG.debug("{}: allocating new open snapshot {}", persistenceId(), id); + tree.getStats().incrementReadOnlyTransactionCount(); + return createOpenSnapshot(id); } LOG.debug("{}: allocating new open transaction {}", persistenceId(), id); @@ -186,21 +241,24 @@ abstract class AbstractFrontendHistory implements Identifiable> participatingShardNames); @Override - public String toString() { - return MoreObjects.toStringHelper(this).omitNullValues().add("identifier", getIdentifier()) - .add("persistenceId", persistenceId).add("transactions", transactions).toString(); + public final String toString() { + return MoreObjects.toStringHelper(this).omitNullValues() + .add("identifier", getIdentifier()) + .add("persistenceId", persistenceId) + .add("transactions", transactions) + .toString(); } }