natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Accord: Fix minDecidedId calculation 

Also fix:
 - slowCoordinatorDelay calculation for locally truncated command throws ISE
 - Bad cast of Truncated during replay
 - Don't throw IllegalStateException in Invariants.expect if accord.testing == false
 - Replay of empty segment throws NPE
Also improve:
 - Support tracing of recovery and home progress

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20771
This commit is contained in:
Benedict Elliott Smith 2025-07-16 07:48:42 +01:00
parent e06b8eabb1
commit 37b6ade9c9
6 changed files with 67 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit ad26bb9df8f774300668a7099858e951e3f013cb Subproject commit 1c203efdd463f2c452e8072267370f9e8dfdb8f3

5
src/java/org/apache/cassandra/journal/Journal.java
View File

@ -1013,7 +1013,10 @@ public class Journal<K, V> implements Shutdownable
public StaticSegmentKeyIterator(K min, K max) 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<Iterator<Head>> iterators = new ArrayList<>(segments.count()); List<Iterator<Head>> iterators = new ArrayList<>(segments.count());
for (Segment<K, V> segment : segments.allSorted(true)) for (Segment<K, V> segment : segments.allSorted(true))

68
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
View File

@ -69,50 +69,46 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
{ {
static final IntervalComparators COMPARATORS = new IntervalComparators(); static final IntervalComparators COMPARATORS = new IntervalComparators();
static final IntervalKeyComparators KEY_COMPARATORS = new IntervalKeyComparators(); static final IntervalKeyComparators KEY_COMPARATORS = new IntervalKeyComparators();
static class TxnIdInterval implements Comparable<TxnIdInterval> static class TxnIdInterval extends TokenRange
{ {
final RoutingKey start, end;
final TxnId txnId; final TxnId txnId;
TxnIdInterval(RoutingKey start, RoutingKey end, TxnId txnId) TxnIdInterval(RoutingKey start, RoutingKey end, TxnId txnId)
{ {
this.start = start; super((TokenKey) start, (TokenKey) end);
this.end = end;
this.txnId = txnId; this.txnId = txnId;
} }
TxnIdInterval(Range range, TxnId txnId) TxnIdInterval(Range range, TxnId txnId)
{ {
this.start = range.start(); this(range.start(), range.end(), txnId);
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;
} }
} }
static class IntervalComparators implements IntervalBTree.IntervalComparators<TxnIdInterval> static class IntervalComparators implements IntervalBTree.IntervalComparators<TxnIdInterval>
{ {
@Override public Comparator<TxnIdInterval> totalOrder() { return TxnIdInterval::compareTo; } @Override
@Override public Comparator<TxnIdInterval> endWithEndSorter() { return (a, b) -> a.end.compareTo(b.end); } public Comparator<TxnIdInterval> 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<TxnIdInterval> endWithEndSorter() { return (a, b) -> a.end().compareTo(b.end()); }
@Override public SymmetricComparator<TxnIdInterval> startWithStartSeeker() { return (a, b) -> startWithStart(a.start.compareTo(b.start)); } @Override public SymmetricComparator<TxnIdInterval> startWithStartSeeker() { return (a, b) -> startWithStart(a.start().compareTo(b.start())); }
@Override public SymmetricComparator<TxnIdInterval> startWithEndSeeker() { return (a, b) -> startWithEnd(a.start.compareTo(b.end)); } @Override public SymmetricComparator<TxnIdInterval> startWithEndSeeker() { return (a, b) -> startWithEnd(a.start().compareTo(b.end())); }
@Override public SymmetricComparator<TxnIdInterval> endWithStartSeeker() { return (a, b) -> endWithStart(a.end.compareTo(b.start)); } @Override public SymmetricComparator<TxnIdInterval> endWithStartSeeker() { return (a, b) -> endWithStart(a.end().compareTo(b.start())); }
} }
static class IntervalKeyComparators implements IntervalBTree.WithIntervalComparators<RoutingKey, TxnIdInterval> static class IntervalKeyComparators implements IntervalBTree.WithIntervalComparators<RoutingKey, TxnIdInterval>
{ {
@Override public AsymmetricComparator<RoutingKey, TxnIdInterval> startWithStartSeeker() { return (a, b) -> keyStartWithStart(a.compareTo(b.start));} @Override public AsymmetricComparator<RoutingKey, TxnIdInterval> startWithStartSeeker() { return (a, b) -> keyStartWithStart(a.compareTo(b.start()));}
@Override public AsymmetricComparator<RoutingKey, TxnIdInterval> startWithEndSeeker() { return (a, b) -> keyStartWithEnd(a.compareTo(b.end)); } @Override public AsymmetricComparator<RoutingKey, TxnIdInterval> startWithEndSeeker() { return (a, b) -> keyStartWithEnd(a.compareTo(b.end())); }
@Override public AsymmetricComparator<RoutingKey, TxnIdInterval> endWithStartSeeker() { return (a, b) -> keyEndWithStart(a.compareTo(b.start)); } @Override public AsymmetricComparator<RoutingKey, TxnIdInterval> endWithStartSeeker() { return (a, b) -> keyEndWithStart(a.compareTo(b.start())); }
} }
public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m) public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
@ -217,7 +213,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
MaxDecidedRX maxDecidedRX = null; MaxDecidedRX maxDecidedRX = null;
if (primaryTxnId != null && primaryTxnId.is(Txn.Kind.ExclusiveSyncPoint) && findAsDep == null) if (primaryTxnId != null && primaryTxnId.is(Txn.Kind.ExclusiveSyncPoint) && findAsDep == null)
maxDecidedRX = commandStore.unsafeGetMaxDecidedRX(); 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<NavigableMap<TxnId, Ranges>> update) private void updateTransitive(UnaryOperator<NavigableMap<TxnId, Ranges>> update)
@ -267,14 +263,16 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
{ {
private final Manager manager; private final Manager manager;
private final MaxDecidedRX maxDecidedRX; private final MaxDecidedRX maxDecidedRX;
private final TxnId primaryTxnId;
private final TxnId minRelevantId; 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.manager = manager;
this.maxDecidedRX = maxDecidedRX; 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<TxnId> forEach) public void intersects(Consumer<TxnId> forEach)
@ -306,14 +304,22 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
{ {
if (maxDecidedRX == null) if (maxDecidedRX == null)
return true; return true;
if (txnIdInterval.txnId.compareTo(minRelevantId) < 0)
if (!isMaybeRelevant(txnIdInterval.txnId))
return false; 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) boolean isMaybeRelevant(TxnId txnId)
{ {
return maxDecidedRX == null || txnId.compareTo(minRelevantId) >= 0; return isRelevant(minRelevantId, txnId);
} }
public void forEachInCache(Unseekables<?> keysOrRanges, Consumer<Summary> forEach, AccordCommandStore.Caches caches) public void forEachInCache(Unseekables<?> keysOrRanges, Consumer<Summary> forEach, AccordCommandStore.Caches caches)

8
src/java/org/apache/cassandra/service/accord/TokenRange.java
View File

@ -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))); 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 // Don't make this public use create or createUnsafe
private TokenRange(TokenKey start, TokenKey end) protected TokenRange(TokenKey start, TokenKey end)
{ {
super(start, end); super(start, end);
} }
@ -59,19 +59,19 @@ public class TokenRange extends Range.EndInclusive
return new TokenRange(start, end); return new TokenRange(start, end);
} }
public TableId table() public final TableId table()
{ {
return start().table(); return start().table();
} }
@Override @Override
public TokenKey start() public final TokenKey start()
{ {
return (TokenKey) super.start(); return (TokenKey) super.start();
} }
@Override @Override
public TokenKey end() public final TokenKey end()
{ {
return (TokenKey) super.end(); return (TokenKey) super.end();
} }

5
src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
View File

@ -245,7 +245,10 @@ public class AccordAgent implements Agent
// TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy) // TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy)
long oneSecond = SECONDS.toMicros(1L); 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 waitMicros = recover(txnId).computeWait(retryCount, MICROSECONDS);
long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis()); long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
Invariants.expect(mostRecentStart <= nowMicros + SECONDS.toMicros(1L), "max(%s,%s)>%d", command.txnId(), command.promised(), nowMicros); Invariants.expect(mostRecentStart <= nowMicros + SECONDS.toMicros(1L), "max(%s,%s)>%d", command.txnId(), command.promised(), nowMicros);

34
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -150,31 +150,31 @@ public class AccordDebugKeyspaceTest extends CQLTester
TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); 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); 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"); execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
execute(SET_TRACE, 0, id.toString(), "PROGRESS"); execute(SET_TRACE, 0, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "PROGRESS"); execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
execute(UNSET_TRACE1, id.toString()); execute(UNSET_TRACE1, id.toString());
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "PROGRESS"); execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
execute(UNSET_TRACE2, id.toString(), "PROGRESS"); execute(UNSET_TRACE2, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "PROGRESS"); execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
accord.node().coordinate(id, txn); accord.node().coordinate(id, txn);
filter.preAccept.awaitThrowUncheckedOnInterrupt(); filter.preAccept.awaitThrowUncheckedOnInterrupt();
filter.apply.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_TRACES1, id.toString(), "FETCH", Long.MAX_VALUE);
execute(ERASE_TRACES2, id.toString(), "FETCH"); execute(ERASE_TRACES2, id.toString(), "FETCH");
execute(ERASE_TRACES1, id.toString(), "PROGRESS", Long.MAX_VALUE); execute(ERASE_TRACES1, id.toString(), "WAIT_PROGRESS", Long.MAX_VALUE);
Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "PROGRESS").size()).isEqualTo(0); Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WAIT_PROGRESS").size()).isEqualTo(0);
// just check other variants don't fail // 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()); execute(ERASE_TRACES3, id.toString());
} }
finally finally