diff --git a/modules/accord b/modules/accord index ad26bb9df8..1c203efdd4 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit ad26bb9df8f774300668a7099858e951e3f013cb +Subproject commit 1c203efdd463f2c452e8072267370f9e8dfdb8f3 diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java index 2545f622f1..5f765b48f2 100644 --- a/src/java/org/apache/cassandra/journal/Journal.java +++ b/src/java/org/apache/cassandra/journal/Journal.java @@ -1013,7 +1013,10 @@ public class Journal implements Shutdownable public StaticSegmentKeyIterator(K min, K max) { - this.segments = selectAndReference(s -> s.isStatic() && (min == null || keySupport.compare(s.index().lastId(), min) >= 0) && (max == null || keySupport.compare(s.index().firstId(), max) <= 0)); + this.segments = selectAndReference(s -> s.isStatic() + && s.asStatic().index().entryCount() > 0 + && (min == null || keySupport.compare(s.index().lastId(), min) >= 0) + && (max == null || keySupport.compare(s.index().firstId(), max) <= 0)); List> iterators = new ArrayList<>(segments.count()); for (Segment segment : segments.allSorted(true)) diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java index aa331fbda4..ff43778d69 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java @@ -69,50 +69,46 @@ public class CommandsForRanges extends TreeMap implements Co { static final IntervalComparators COMPARATORS = new IntervalComparators(); static final IntervalKeyComparators KEY_COMPARATORS = new IntervalKeyComparators(); - static class TxnIdInterval implements Comparable + static class TxnIdInterval extends TokenRange { - final RoutingKey start, end; final TxnId txnId; TxnIdInterval(RoutingKey start, RoutingKey end, TxnId txnId) { - this.start = start; - this.end = end; + super((TokenKey) start, (TokenKey) end); this.txnId = txnId; } TxnIdInterval(Range range, TxnId txnId) { - this.start = range.start(); - this.end = range.end(); - this.txnId = txnId; - } - - @Override - public int compareTo(TxnIdInterval that) - { - int c = this.start.compareTo(that.start); - if (c == 0) c = this.end.compareTo(that.end); - if (c == 0) c = this.txnId.compareTo(that.txnId); - return c; + this(range.start(), range.end(), txnId); } } static class IntervalComparators implements IntervalBTree.IntervalComparators { - @Override public Comparator totalOrder() { return TxnIdInterval::compareTo; } - @Override public Comparator endWithEndSorter() { return (a, b) -> a.end.compareTo(b.end); } + @Override + public Comparator totalOrder() + { + return (a, b) -> { + int c = a.start().compareTo(b.start()); + if (c == 0) c = a.end().compareTo(b.end()); + if (c == 0) c = a.txnId.compareTo(b.txnId); + return c; + }; + } + @Override public Comparator endWithEndSorter() { return (a, b) -> a.end().compareTo(b.end()); } - @Override public SymmetricComparator startWithStartSeeker() { return (a, b) -> startWithStart(a.start.compareTo(b.start)); } - @Override public SymmetricComparator startWithEndSeeker() { return (a, b) -> startWithEnd(a.start.compareTo(b.end)); } - @Override public SymmetricComparator endWithStartSeeker() { return (a, b) -> endWithStart(a.end.compareTo(b.start)); } + @Override public SymmetricComparator startWithStartSeeker() { return (a, b) -> startWithStart(a.start().compareTo(b.start())); } + @Override public SymmetricComparator startWithEndSeeker() { return (a, b) -> startWithEnd(a.start().compareTo(b.end())); } + @Override public SymmetricComparator endWithStartSeeker() { return (a, b) -> endWithStart(a.end().compareTo(b.start())); } } static class IntervalKeyComparators implements IntervalBTree.WithIntervalComparators { - @Override public AsymmetricComparator startWithStartSeeker() { return (a, b) -> keyStartWithStart(a.compareTo(b.start));} - @Override public AsymmetricComparator startWithEndSeeker() { return (a, b) -> keyStartWithEnd(a.compareTo(b.end)); } - @Override public AsymmetricComparator endWithStartSeeker() { return (a, b) -> keyEndWithStart(a.compareTo(b.start)); } + @Override public AsymmetricComparator startWithStartSeeker() { return (a, b) -> keyStartWithStart(a.compareTo(b.start()));} + @Override public AsymmetricComparator startWithEndSeeker() { return (a, b) -> keyStartWithEnd(a.compareTo(b.end())); } + @Override public AsymmetricComparator endWithStartSeeker() { return (a, b) -> keyEndWithStart(a.compareTo(b.start())); } } public CommandsForRanges(Map m) @@ -217,7 +213,7 @@ public class CommandsForRanges extends TreeMap implements Co MaxDecidedRX maxDecidedRX = null; if (primaryTxnId != null && primaryTxnId.is(Txn.Kind.ExclusiveSyncPoint) && findAsDep == null) maxDecidedRX = commandStore.unsafeGetMaxDecidedRX(); - return new Loader(this, searchKeysOrRanges, redundantBefore, testKind, minTxnId, maxTxnId, findAsDep, maxDecidedRX); + return new Loader(this, primaryTxnId, searchKeysOrRanges, redundantBefore, testKind, minTxnId, maxTxnId, findAsDep, maxDecidedRX); } private void updateTransitive(UnaryOperator> update) @@ -267,14 +263,16 @@ public class CommandsForRanges extends TreeMap implements Co { private final Manager manager; private final MaxDecidedRX maxDecidedRX; + private final TxnId primaryTxnId; private final TxnId minRelevantId; - public Loader(Manager manager, Unseekables searchKeysOrRanges, RedundantBefore redundantBefore, Kinds testKinds, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep, MaxDecidedRX maxDecidedRX) + public Loader(Manager manager, TxnId primaryTxnId, Unseekables searchKeysOrRanges, RedundantBefore redundantBefore, Kinds testKinds, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep, MaxDecidedRX maxDecidedRX) { - super(null, searchKeysOrRanges, redundantBefore, testKinds, minTxnId, maxTxnId, findAsDep); + super(primaryTxnId, searchKeysOrRanges, redundantBefore, testKinds, minTxnId, maxTxnId, findAsDep); this.manager = manager; this.maxDecidedRX = maxDecidedRX; - this.minRelevantId = maxDecidedRX == null ? null : TxnId.nonNullOrMax(TxnId.NONE, maxDecidedRX.foldl(searchKeysOrRanges, TxnId::nonNullOrMin, null)); + this.primaryTxnId = primaryTxnId; + this.minRelevantId = MaxDecidedRX.minDecidedDependencyId(maxDecidedRX, searchKeysOrRanges, primaryTxnId); } public void intersects(Consumer forEach) @@ -306,14 +304,22 @@ public class CommandsForRanges extends TreeMap implements Co { if (maxDecidedRX == null) return true; - if (txnIdInterval.txnId.compareTo(minRelevantId) < 0) + + if (!isMaybeRelevant(txnIdInterval.txnId)) return false; - return maxDecidedRX.foldl(txnIdInterval.start, txnIdInterval.end, (decided, anyUndecided, test, ignore) -> test.compareTo(decided) >= 0, false, txnIdInterval.txnId, null); + + TxnId minRelevantId = MaxDecidedRX.minDecidedDependencyId(maxDecidedRX, Ranges.of(txnIdInterval), primaryTxnId); + return isRelevant(minRelevantId, primaryTxnId); + } + + private boolean isRelevant(@Nullable TxnId minRelevantId, TxnId txnId) + { + return minRelevantId == null || minRelevantId.compareTo(txnId) <= 0; } boolean isMaybeRelevant(TxnId txnId) { - return maxDecidedRX == null || txnId.compareTo(minRelevantId) >= 0; + return isRelevant(minRelevantId, txnId); } public void forEachInCache(Unseekables keysOrRanges, Consumer forEach, AccordCommandStore.Caches caches) diff --git a/src/java/org/apache/cassandra/service/accord/TokenRange.java b/src/java/org/apache/cassandra/service/accord/TokenRange.java index 4d0294af7a..08750a841f 100644 --- a/src/java/org/apache/cassandra/service/accord/TokenRange.java +++ b/src/java/org/apache/cassandra/service/accord/TokenRange.java @@ -41,7 +41,7 @@ public class TokenRange extends Range.EndInclusive public static final long EMPTY_SIZE = ObjectSizes.measure(new TokenRange(TokenKey.min(TableId.fromLong(0), Murmur3Partitioner.instance), TokenKey.max(TableId.fromLong(0), Murmur3Partitioner.instance))); // Don't make this public use create or createUnsafe - private TokenRange(TokenKey start, TokenKey end) + protected TokenRange(TokenKey start, TokenKey end) { super(start, end); } @@ -59,19 +59,19 @@ public class TokenRange extends Range.EndInclusive return new TokenRange(start, end); } - public TableId table() + public final TableId table() { return start().table(); } @Override - public TokenKey start() + public final TokenKey start() { return (TokenKey) super.start(); } @Override - public TokenKey end() + public final TokenKey end() { return (TokenKey) super.end(); } diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index 78da816fd7..ed39bbad8d 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -245,7 +245,10 @@ public class AccordAgent implements Agent // TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy) long oneSecond = SECONDS.toMicros(1L); - long mostRecentStart = Math.max(command.txnId().hlc(), command.promised().hlc()); + long promisedHlc = command.promised().hlc(); + if (promisedHlc == Long.MAX_VALUE) + promisedHlc = 0; + long mostRecentStart = Math.max(command.txnId().hlc(), promisedHlc); long waitMicros = recover(txnId).computeWait(retryCount, MICROSECONDS); long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis()); Invariants.expect(mostRecentStart <= nowMicros + SECONDS.toMicros(1L), "max(%s,%s)>%d", command.txnId(), command.promised(), nowMicros); diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index 4d4067786a..08163de516 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -150,31 +150,31 @@ public class AccordDebugKeyspaceTest extends CQLTester TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0); - execute(SET_TRACE, 1, id.toString(), "PROGRESS"); - assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); - execute(SET_TRACE, 0, id.toString(), "PROGRESS"); - assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); - execute(SET_TRACE, 1, id.toString(), "PROGRESS"); - assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); + execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1)); + execute(SET_TRACE, 0, id.toString(), "WAIT_PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS")); + execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1)); execute(UNSET_TRACE1, id.toString()); - assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); - execute(SET_TRACE, 1, id.toString(), "PROGRESS"); - assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); - execute(UNSET_TRACE2, id.toString(), "PROGRESS"); - assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); - execute(SET_TRACE, 1, id.toString(), "PROGRESS"); - assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); + assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS")); + execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1)); + execute(UNSET_TRACE2, id.toString(), "WAIT_PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS")); + execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1)); accord.node().coordinate(id, txn); filter.preAccept.awaitThrowUncheckedOnInterrupt(); filter.apply.awaitThrowUncheckedOnInterrupt(); - spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "PROGRESS").size()).isGreaterThan(0)); + spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WAIT_PROGRESS").size()).isGreaterThan(0)); execute(ERASE_TRACES1, id.toString(), "FETCH", Long.MAX_VALUE); execute(ERASE_TRACES2, id.toString(), "FETCH"); - execute(ERASE_TRACES1, id.toString(), "PROGRESS", Long.MAX_VALUE); - Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "PROGRESS").size()).isEqualTo(0); + execute(ERASE_TRACES1, id.toString(), "WAIT_PROGRESS", Long.MAX_VALUE); + Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WAIT_PROGRESS").size()).isEqualTo(0); // just check other variants don't fail - execute(ERASE_TRACES2, id.toString(), "PROGRESS"); + execute(ERASE_TRACES2, id.toString(), "WAIT_PROGRESS"); execute(ERASE_TRACES3, id.toString()); } finally