You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ko...@apache.org on 2026/03/08 16:15:33 UTC
(cassandra-accord) 07/07: Fix DurableBefore.merge
This is an automated email from the ASF dual-hosted git repository.
konstantinov pushed a commit to branch fixes-260226
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit 437b4d81a30665a912c5b221d1a24f8388b0c8f2
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Sun Mar 1 08:22:00 2026 +0000
Fix DurableBefore.merge
---
.../src/main/java/accord/local/CommandStore.java | 2 +
.../main/java/accord/topology/TopologyRange.java | 2 +-
.../java/accord/utils/BTreeReducingRangeMap.java | 5 +
.../src/main/java/accord/utils/btree/BTree.java | 190 ++++++++++++++-------
.../java/accord/utils/btree/ReducingBTree.java | 76 +++++++--
.../test/java/accord/local/DurableBeforeTest.java | 23 ++-
...urableBeforeTest.java => MaxConflictsTest.java} | 177 +++++++++++--------
.../src/test/java/accord/utils/AccordGens.java | 26 +++
8 files changed, 341 insertions(+), 160 deletions(-)
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index ed9277b0..c84a3472 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -454,6 +454,8 @@ public abstract class CommandStore implements AbstractAsyncExecutor, SequentialA
if (prev != null && prev.executeAt() != null && prev.executeAt().compareToStrict(executeAt) >= 0 && !force) return;
executeAt = executeAt.flattenUniqueHlc(); // this is what guarantees a bootstrap recipient can compute uniqueHlc safely
MaxConflicts updatedMaxConflicts = maxConflicts.update(updated.participants().hasTouched(), executeAt);
+ if (Invariants.isParanoid())
+ Invariants.require(updatedMaxConflicts.get(updated.participants().hasTouched()).compareTo(executeAt) >= 0);
updateMaxConflicts(executeAt, updatedMaxConflicts);
}
diff --git a/accord-core/src/main/java/accord/topology/TopologyRange.java b/accord-core/src/main/java/accord/topology/TopologyRange.java
index 4540e39d..4f4b9fbc 100644
--- a/accord-core/src/main/java/accord/topology/TopologyRange.java
+++ b/accord-core/src/main/java/accord/topology/TopologyRange.java
@@ -45,7 +45,7 @@ public class TopologyRange
public void forEach(Consumer<Topology> forEach, long minEpoch, int count)
{
if (minEpoch == 0) // Bootstrap
- minEpoch = this.min;
+ minEpoch = Math.max(1, this.min);
long emptyUpTo = firstNonEmpty == -1 ? current : firstNonEmpty - 1;
// Report empty epochs
diff --git a/accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java b/accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java
index f1b3399a..e5515aad 100644
--- a/accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java
+++ b/accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java
@@ -332,4 +332,9 @@ public class BTreeReducingRangeMap<E extends Entry<E>> implements Iterable<E>
{
public abstract M build();
}
+
+ public static boolean isWellFormed(BTreeReducingRangeMap<?> map)
+ {
+ return ReducingBTree.isWellFormed(map.tree);
+ }
}
diff --git a/accord-core/src/main/java/accord/utils/btree/BTree.java b/accord-core/src/main/java/accord/utils/btree/BTree.java
index f2dbb1b1..f7c364c9 100644
--- a/accord-core/src/main/java/accord/utils/btree/BTree.java
+++ b/accord-core/src/main/java/accord/utils/btree/BTree.java
@@ -2311,7 +2311,7 @@ public class BTree
*/
public final boolean isEmpty()
{
- return count == 0 && savedNextKey == null;
+ return count == 0 && savedNextKey == null && (getClass() != BranchBuilder.class || !((BranchBuilder)this).hasRightChild);
}
abstract void initialiseCopy(Object[] unode);
@@ -3374,9 +3374,29 @@ public class BTree
private static abstract class AbstractUpdater extends AbstractFastBuilder implements AutoCloseable
{
+ protected void requireEmpty()
+ {
+ Invariants.require(leaf().isEmpty());
+ BranchBuilder branch = leaf().parent;
+ while (branch != null && branch.inUse)
+ {
+ Invariants.require(branch.isEmpty());
+ branch = branch.parent;
+ }
+ Invariants.require(branch == null || branch.isEmpty());
+ if (Invariants.isParanoid() && branch != null)
+ {
+ while (branch != null)
+ {
+ Invariants.require(!branch.inUse);
+ Invariants.require(branch.sourceNode == null);
+ branch = branch.parent;
+ }
+ }
+ }
+
void reset()
{
- assert leaf().count == 0;
clearLeafBuffer(leaf().buffer);
if (leaf().savedBuffer != null)
Arrays.fill(leaf().savedBuffer, null);
@@ -3393,16 +3413,6 @@ public class BTree
branch.inUse = false;
branch = branch.parent;
}
- Invariants.require(branch == null || (branch.count == 0 && !branch.hasRightChild));
- if (Invariants.isParanoid() && branch != null)
- {
- while (branch != null)
- {
- Invariants.require(!branch.inUse);
- Invariants.require(branch.sourceNode == null);
- branch = branch.parent;
- }
- }
}
/**
@@ -3497,7 +3507,7 @@ public class BTree
ik = updateRecursive(ik, update, null, builder);
assert ik == null;
Object[] result = builder.completeBuild();
-
+ requireEmpty();
return result;
}
@@ -3680,7 +3690,8 @@ public class BTree
parent.inUse = false;
}
- abstract O apply(LeafOrBranchBuilder level, I in);
+ abstract O apply(I in);
+ O applyNoInput() { return null; }
LeafOrBranchBuilder initRoot(Object[] root)
{
@@ -3784,7 +3795,7 @@ public class BTree
level = level.parent;
}
- O next = apply(level, (I) unode[upos++]);
+ O next = apply((I) unode[upos++]);
if (next == null)
{
// next has been filtered, so look for the (unfiltered) successor node
@@ -3797,22 +3808,32 @@ public class BTree
while (upos < usz)
{
- if (null != (next = apply(level, (I)unode[upos++])))
+ if (null != (next = apply((I)unode[upos++])))
break successor;
}
- do
+ int ascendTo = update.ascendIfUnfinishedParent();
+ if (ascendTo >= 0)
{
- if (!update.ascendToParent())
- return finishAndDrain(level);
-
+ Invariants.require(ascendTo > 0);
unode = update.node();
upos = update.position();
usz = shallowSizeOfBranch(unode);
- level = level.parent;
- } while (upos >= usz);
+ while (ascendTo-- > 0)
+ level = level.parent;
+ }
+ else
+ {
+ if (null != (next = applyNoInput()))
+ break;
+
+ while (update.ascendToParent())
+ level = level.parent;
- next = apply(level, (I) unode[upos++]);
+ return finishAndDrain(level);
+ }
+
+ next = apply((I) unode[upos++]);
if (next != null)
break;
}
@@ -3852,7 +3873,7 @@ public class BTree
unode = update.node();
upos = update.position();
usz = shallowSize(unode);
- from = from.child;
+ from = Invariants.nonNull(from.child);
from.setSourceNode(unode);
}
return to;
@@ -4027,64 +4048,71 @@ public class BTree
fill.prepend(predecessor, predecessorNextKey);
}
- void overwritePrev(LeafOrBranchBuilder level, Object with)
+ void overwritePrev(Object with)
{
- if (level.getClass() == BranchBuilder.class) overwritePrevInBranch((BranchBuilder) level, with);
- else overwritePrevInLeafOrBranch(level, with);
- }
-
- void overwritePrevInBranch(BranchBuilder branch, Object with)
- {
- if (branch.child.count > 0)
- {
- branch.child.buffer[branch.child.count - 1] = with;
- return;
- }
- else if (branch.hasRightChild)
+ LeafOrBranchBuilder level = leaf();
+ if (level.isEmpty())
{
- setRightMost(branch, with);
- return;
+ BranchBuilder parent = leaf().parent;
+ while (parent != null && parent.inUse && parent.isEmpty())
+ parent = parent.parent;
+
+ if (parent != null)
+ {
+ if (parent.hasRightChild)
+ {
+ setRightMost(parent, with);
+ return;
+ }
+ level = parent;
+ }
}
- overwritePrevInLeafOrBranch(branch, with);
+ overwriteLast(level, with);
}
- void overwritePrevInLeafOrBranch(LeafOrBranchBuilder level, Object with)
+ private void overwriteLast(LeafOrBranchBuilder level, Object with)
{
- if (level.count > 0) level.buffer[level.count - 1] = with;
- else if (level.savedNextKey != null) level.savedNextKey = with;
- else if (level.parent != null && level.parent.inUse) overwritePrevInBranch(level.parent, with);
+ if (level.count > 0) level.buffer[level.count - 1] = validateOverwritePrev(level.buffer[level.count - 1], with);
+ else if (level.savedNextKey != null) level.savedNextKey = validateOverwritePrev(level.savedNextKey, with);
else throw new IllegalStateException("Nothing written yet");
}
- Object prev(LeafOrBranchBuilder level)
+ Object validateOverwritePrev(Object overwriting, Object with)
{
- if (level.getClass() == BranchBuilder.class) return prevInBranch((BranchBuilder) level);
- else return prevInLeafOrBranch(level);
+ return with;
}
- private Object prevInBranch(BranchBuilder branch)
+ Object prev()
{
- if (branch.child.count > 0)
- return branch.child.buffer[branch.child.count - 1];
- if (branch.hasRightChild)
- return rightMost(branch);
- return prevInLeafOrBranch(branch);
+ if (!leaf().isEmpty())
+ return last(leaf());
+
+ BranchBuilder parent = leaf().parent;
+ while (parent != null && parent.inUse && parent.isEmpty())
+ parent = parent.parent;
+
+ if (parent == null || parent.isEmpty())
+ return null;
+
+ if (parent.hasRightChild)
+ return rightMost(parent);
+
+ return last(parent);
}
- private Object prevInLeafOrBranch(LeafOrBranchBuilder level)
+ private Object last(LeafOrBranchBuilder level)
{
if (level.count > 0) return level.buffer[level.count - 1];
else if (level.savedNextKey != null) return level.savedNextKey;
- else if (level.parent != null && level.parent.inUse) return prevInBranch(level.parent);
else return null;
}
- private Object rightMost(LeafOrBranchBuilder level)
+ private Object rightMost(BranchBuilder level)
{
Object[] node = (Object[]) level.buffer[31 + level.count];
while (!BTree.isLeaf(node))
node = (Object[]) node[node.length - 2];
- return node[BTree.size(node) - 1];
+ return node[BTree.sizeOfLeaf(node) - 1];
}
private void setRightMost(LeafOrBranchBuilder level, Object set)
@@ -4096,7 +4124,7 @@ public class BTree
private Object[] replaceRightMost(Object[] node, Object set)
{
Object[] result = node.clone();
- if (BTree.isLeaf(result)) result[BTree.size(result) - 1] = set;
+ if (BTree.isLeaf(result)) result[BTree.sizeOfLeaf(result) - 1] = validateOverwritePrev(result[BTree.sizeOfLeaf(result) - 1], set);
else result[result.length - 2] = replaceRightMost((Object[])result[result.length - 2], set);
return result;
}
@@ -4121,7 +4149,9 @@ public class BTree
Object[] subtract(Object[] update, PeekingSearchIterator<K, ? extends K> remove)
{
this.remove = remove;
- return apply(update);
+ Object[] result = apply(update);
+ requireEmpty();
+ return result;
}
Object[] subtract(Object[] update, Object[] remove)
@@ -4130,7 +4160,7 @@ public class BTree
}
@Override
- T apply(LeafOrBranchBuilder level, T v)
+ T apply(T v)
{
return remove.next(v) == null ? v : null;
}
@@ -4218,13 +4248,21 @@ public class BTree
return -1 - upos;
}
+ @Override
+ Object[] apply(Object[] root)
+ {
+ Object[] result = super.apply(root);
+ requireEmpty();
+ return result;
+ }
+
protected final boolean transformLeaf(Object[] unode, int upos, int usz)
{
if (leaf().count == 0)
{
while (upos < usz)
{
- O v = apply(leaf(), (I) unode[upos++]);
+ O v = apply((I) unode[upos++]);
leaf().maybeAddKeyNoOverflow(v);
}
}
@@ -4232,7 +4270,7 @@ public class BTree
{
while (upos < usz)
{
- O v = apply(leaf(), (I) unode[upos++]);
+ O v = apply((I) unode[upos++]);
leaf().maybeAddKey(v);
}
}
@@ -4247,7 +4285,7 @@ public class BTree
Function<? super I, ? extends O> apply;
- O apply(LeafOrBranchBuilder level, I v)
+ O apply(I v)
{
return apply.apply(v);
}
@@ -4280,7 +4318,7 @@ public class BTree
I2 i2;
TinyThreadLocalPool.TinyPool<BiTransformer> pool;
- O apply(LeafOrBranchBuilder level, I i1)
+ O apply(I i1)
{
return apply.apply(i1, i2);
}
@@ -4427,6 +4465,30 @@ public class BTree
return false;
return --depth >= 0;
}
+
+ int ascendIfUnfinishedParent()
+ {
+ int d = depth;
+ while (--d >= 0)
+ {
+ Object[] unode = nodes[d];
+ int upos = positions[d];
+ int usz = shallowSizeOfBranch(unode);
+ if (upos < usz)
+ {
+ int levels = depth - d;
+ depth = d;
+ return levels;
+ }
+ }
+
+ return -1;
+ }
+
+ void setDepth(int depth)
+ {
+ this.depth = depth;
+ }
}
private static class SimpleTreeKeysIterator<Compare, Insert extends Compare>
diff --git a/accord-core/src/main/java/accord/utils/btree/ReducingBTree.java b/accord-core/src/main/java/accord/utils/btree/ReducingBTree.java
index 36f3b4ef..313faaed 100644
--- a/accord-core/src/main/java/accord/utils/btree/ReducingBTree.java
+++ b/accord-core/src/main/java/accord/utils/btree/ReducingBTree.java
@@ -34,6 +34,7 @@ import static accord.utils.SortedArrays.Search.CEIL;
import static accord.utils.SortedArrays.Search.FAST;
import static accord.utils.btree.BTree.Dir.ASC;
import static accord.utils.btree.BTree.isLeaf;
+import static accord.utils.btree.BTree.iterable;
import static accord.utils.btree.BTree.slice;
public class ReducingBTree
@@ -125,6 +126,7 @@ public class ReducingBTree
Object[] result = apply(update);
Invariants.require(pendingInputs.isEmpty());
Invariants.require(!merge.hasNext());
+ requireEmpty();
return result;
}
@@ -134,11 +136,11 @@ public class ReducingBTree
}
@Override
- E apply(LeafOrBranchBuilder level, E in)
+ E apply(E in)
{
while (true)
{
- E result = coalesce(level, mergeInputOrPending(level, in));
+ E result = coalesce(mergeInputOrPending(in));
if (result != null)
return result;
@@ -148,24 +150,46 @@ public class ReducingBTree
}
}
- E coalesce(LeafOrBranchBuilder level, E out)
+ E applyNoInput()
{
- E prev = prev(level);
+ while (true)
+ {
+ E result = pendingMerge;
+ if (result == null)
+ return null;
+
+ advancePendingMerge();
+ result = coalesce(result);
+ if (result != null)
+ return result;
+ }
+ }
+
+ E coalesce(E out)
+ {
+ E prev = prev();
if (prev != null && out.equalsIgnoreRange(prev) && prev.end().compareTo(out.start()) >= 0)
{
prev = prev.with(prev.start(), out.end());
- overwritePrev(level, prev);
+ overwritePrev(prev);
return null;
}
return out;
}
- E prev(LeafOrBranchBuilder level)
+ E prev()
+ {
+ return (E) super.prev();
+ }
+
+ @Override
+ Object validateOverwritePrev(Object overwriting, Object with)
{
- return (E) super.prev(level);
+ Invariants.require(((Entry)overwriting).start().equals(((Entry)with).start()));
+ return with;
}
- E mergeInputOrPending(LeafOrBranchBuilder level, E input)
+ E mergeInputOrPending(E input)
{
if (!pendingInputs.isEmpty())
{
@@ -173,12 +197,12 @@ public class ReducingBTree
pendingInputs.addLast(input);
input = tmp;
}
- return merge(level, input);
+ return merge(input);
}
- E merge(LeafOrBranchBuilder level, E input)
+ E merge(E input)
{
- E prev = prev(level);
+ E prev = prev();
if (pendingMerge == null)
return maybeMoveStart(prev, input);
@@ -221,15 +245,21 @@ public class ReducingBTree
pendingInputs.addFirst(input);
if (cs < 0)
{
+ // ces > 0 -> is.end() > ms.start()
return input.with(is, ms);
}
else
{
int cse = is.compareTo(me);
if (cse >= 0)
+ {
advancePendingMerge();
-
- return merge.with(ms, is);
+ return merge.with(ms, me);
+ }
+ else
+ {
+ return merge.with(ms, is);
+ }
}
}
}
@@ -274,7 +304,7 @@ public class ReducingBTree
private void mergeToLeaf(E input)
{
- E next = coalesce(leaf(), mergeInputOrPending(leaf(), input));
+ E next = coalesce(mergeInputOrPending(input));
if (next != null)
leaf().addKey(next);
drainPendingInputs();
@@ -284,7 +314,7 @@ public class ReducingBTree
{
while (!pendingInputs.isEmpty())
{
- E next = coalesce(leaf(), merge(leaf(), pendingInputs.poll()));
+ E next = coalesce(merge(pendingInputs.poll()));
if (next != null)
leaf().addKey(next);
}
@@ -357,7 +387,7 @@ public class ReducingBTree
RoutingKey pendingStart = next.start(), nextStart = pendingStart;
RoutingKey pendingEnd = next.end(), nextEnd = pendingEnd;
- E prev = copyFrom != copyTo ? (E) copy[copyTo - 1] : prev(leaf());
+ E prev = copyFrom != copyTo ? (E) copy[copyTo - 1] : prev();
if (prev != null && prev.end().compareTo(nextStart) > 0)
nextStart = prev.end();
@@ -375,7 +405,7 @@ public class ReducingBTree
leaf().copy(copy, copyFrom, copyTo - copyFrom);
copyFrom = copyTo;
- next = coalesce(leaf(), next);
+ next = coalesce(next);
if (next != null)
leaf().addKey(next);
@@ -735,4 +765,16 @@ public class ReducingBTree
if (end.compareTo(range.start()) <= 0) return -1;
return 0;
}
+
+ public static boolean isWellFormed(Object[] btree)
+ {
+ Entry<?> prev = null;
+ for (Entry<?> e : BTree.<Entry<?>>iterable(btree))
+ {
+ if (prev != null && prev.end().compareTo(e.start()) > 0)
+ return false;
+ prev = e;
+ }
+ return true;
+ }
}
diff --git a/accord-core/src/test/java/accord/local/DurableBeforeTest.java b/accord-core/src/test/java/accord/local/DurableBeforeTest.java
index f4916437..db1b08af 100644
--- a/accord-core/src/test/java/accord/local/DurableBeforeTest.java
+++ b/accord-core/src/test/java/accord/local/DurableBeforeTest.java
@@ -30,6 +30,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import accord.api.RoutingKey;
+import accord.impl.PrefixedIntHashKey;
import accord.primitives.AbstractRanges;
import accord.primitives.Range;
import accord.primitives.Ranges;
@@ -47,6 +48,7 @@ import accord.utils.ReducingRangeMap;
import static accord.primitives.Status.Durability.HasOutcome.None;
import static accord.primitives.Status.Durability.HasOutcome.Quorum;
import static accord.primitives.Status.Durability.HasOutcome.Universal;
+import static accord.utils.BTreeReducingRangeMap.isWellFormed;
import static accord.utils.Functions.alwaysFalse;
public class DurableBeforeTest
@@ -60,15 +62,28 @@ public class DurableBeforeTest
public void test()
{
test(1L, 100000, 5);
- for (int i = 0 ; i < 100 ; ++i)
+ for (int i = 0 ; i < 1000 ; ++i)
test(ThreadLocalRandom.current().nextLong(), 100000, 5);
}
private void test(long seed, int actionCount, int maxDurableBefores)
{
+ int MAX_HASH = 1 << 20;
RandomTestRunner.test().withSeed(seed).check(rs -> {
+ int maxHashRange = 1 + rs.nextInt(MAX_HASH - 1);
Gen<TxnId> genSyncIds = AccordGens.txnIds(Gens.pick(Txn.Kind.VisibilitySyncPoint, Txn.Kind.ExclusiveSyncPoint));
- Gen<Ranges> genRanges = AccordGens.ranges(Gens.pickInt(1,2,3), AccordGens.intRoutingKey(), Range::of);
+ Gen<Ranges> genRanges = rs2 -> {
+ int size = 1 + rs.nextInt(2);
+ int prefix = 1 + rs.nextInt(3);
+ Range[] ranges = new Range[size];
+ for (int i = 0 ; i < size ; ++i)
+ {
+ int startHash = rs.nextInt(MAX_HASH - (1 + maxHashRange));
+ int endHash = startHash + 1 + rs.nextInt(maxHashRange);
+ ranges[i] = Range.of(PrefixedIntHashKey.forHash(prefix, startHash), PrefixedIntHashKey.forHash(prefix, endHash));
+ }
+ return Ranges.of(ranges);
+ };
List<DurableBeforeLinear> as = new ArrayList<>();
List<DurableBefore> bs = new ArrayList<>();
@@ -108,13 +123,14 @@ public class DurableBeforeTest
DurableBefore new2 = bs.get(j);
DurableBeforeLinear oldmerge = DurableBeforeLinear.merge(old1, old2);
DurableBefore newmerge = DurableBefore.merge(new1, new2);
+ Assertions.assertTrue(isWellFormed(newmerge));
+ Assertions.assertTrue(equals(oldmerge, newmerge));
as.set(i, oldmerge);
bs.set(i, newmerge);
as.set(j, as.get(as.size() - 1));
bs.set(j, bs.get(bs.size() - 1));
as.remove(as.size() - 1);
bs.remove(bs.size() - 1);
- Assertions.assertTrue(equals(oldmerge, newmerge));
}
break;
case UPDATE:
@@ -129,6 +145,7 @@ public class DurableBeforeTest
DurableBeforeLinear nextold = DurableBeforeLinear.merge(prevold, DurableBeforeLinear.create(ranges, q, u));
DurableBefore nextnew = prevnew.update(ranges, q, u);
+ Assertions.assertTrue(isWellFormed(nextnew));
Assertions.assertTrue(equals(nextold, nextnew));
as.set(i, nextold);
bs.set(i, nextnew);
diff --git a/accord-core/src/test/java/accord/local/DurableBeforeTest.java b/accord-core/src/test/java/accord/local/MaxConflictsTest.java
similarity index 57%
copy from accord-core/src/test/java/accord/local/DurableBeforeTest.java
copy to accord-core/src/test/java/accord/local/MaxConflictsTest.java
index f4916437..75206d48 100644
--- a/accord-core/src/test/java/accord/local/DurableBeforeTest.java
+++ b/accord-core/src/test/java/accord/local/MaxConflictsTest.java
@@ -18,22 +18,22 @@
package accord.local;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Objects;
import java.util.concurrent.ThreadLocalRandom;
-import java.util.function.Supplier;
-
+import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nonnull;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import accord.api.RoutingKey;
+import accord.impl.PrefixedIntHashKey;
import accord.primitives.AbstractRanges;
import accord.primitives.Range;
import accord.primitives.Ranges;
+import accord.primitives.RoutingKeys;
import accord.primitives.Status;
+import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
@@ -47,96 +47,123 @@ import accord.utils.ReducingRangeMap;
import static accord.primitives.Status.Durability.HasOutcome.None;
import static accord.primitives.Status.Durability.HasOutcome.Quorum;
import static accord.primitives.Status.Durability.HasOutcome.Universal;
+import static accord.utils.BTreeReducingRangeMap.isWellFormed;
import static accord.utils.Functions.alwaysFalse;
-public class DurableBeforeTest
+public class MaxConflictsTest
{
- enum TestAction
- {
- NEW, UPDATE, MERGE
- }
-
@Test
public void test()
{
- test(1L, 100000, 5);
+ test(1L, 1000);
+ for (int i = 0 ; i < 10000 ; ++i)
+ test(ThreadLocalRandom.current().nextLong(), 10);
+ for (int i = 0 ; i < 1000 ; ++i)
+ test(ThreadLocalRandom.current().nextLong(), 100);
for (int i = 0 ; i < 100 ; ++i)
- test(ThreadLocalRandom.current().nextLong(), 100000, 5);
+ test(ThreadLocalRandom.current().nextLong(), 1000);
}
- private void test(long seed, int actionCount, int maxDurableBefores)
+ private void test(long seed, int actionCount)
{
+ System.out.println("Seed: " + seed + ", count: " + actionCount);
+ int MAX_HASH = 1 << 20;
RandomTestRunner.test().withSeed(seed).check(rs -> {
- Gen<TxnId> genSyncIds = AccordGens.txnIds(Gens.pick(Txn.Kind.VisibilitySyncPoint, Txn.Kind.ExclusiveSyncPoint));
- Gen<Ranges> genRanges = AccordGens.ranges(Gens.pickInt(1,2,3), AccordGens.intRoutingKey(), Range::of);
+ int maxSyncHashRange = 1 + rs.nextInt(MAX_HASH - 1);
+ int maxPruneHashRange = (MAX_HASH-1) / (1 + rs.nextInt(15));
+ int prefixCount = rs.nextInt(1, 3);
+ int kwrxRatio = 1 + rs.nextInt(255);
+ int pruneCount = rs.nextInt(1, 32);
+ AtomicLong epoch = new AtomicLong(1), minHlc = new AtomicLong(), maxHlc = new AtomicLong(10000);
+ Gen.LongGen hlcs = rs2 -> rs2.nextLong(minHlc.get(), maxHlc.get());
+ Gen<TxnId> genSyncIds = AccordGens.txnIds(rs2 -> epoch.get(), hlcs, rs2 -> rs2.nextInt(10), Gens.pick(Txn.Kind.VisibilitySyncPoint, Txn.Kind.ExclusiveSyncPoint));
+ Gen<TxnId> genWriteIds = AccordGens.txnIds(rs2 -> epoch.get(), hlcs, rs2 -> rs2.nextInt(10), Gens.pick(Txn.Kind.Write));
+ Gen<RoutingKeys> genKeys = rs2 -> {
+ int size = 1 + rs.nextInt(2);
+ int prefix = 1 + rs.nextInt(prefixCount);
+ RoutingKey[] keys = new RoutingKey[size];
+ for (int i = 0 ; i < size ; ++i)
+ keys[i] = PrefixedIntHashKey.forHash(prefix, rs.nextInt(MAX_HASH));
+ return RoutingKeys.of(keys);
+ };
+ Gen<Ranges> genRanges = genRanges(MAX_HASH, maxSyncHashRange, prefixCount);
+ Gen<Ranges> genPruneRanges = genRanges(MAX_HASH, maxPruneHashRange, prefixCount);
+
+ MaxConflicts conflicts = MaxConflicts.EMPTY;
+ for (int actionCounter = actionCount ; actionCounter > 0 ; --actionCounter)
+ {
+ MaxConflicts next;
+ if (rs.nextInt(kwrxRatio) == 0)
+ {
+ TxnId syncId = genSyncIds.next(rs);
+ Ranges ranges = genRanges.next(rs);
+ next = conflicts.update(ranges, syncId);
+ Assertions.assertTrue(syncId.compareTo(next.get(ranges)) <= 0);
+ }
+ else
+ {
+ TxnId txnId = genWriteIds.next(rs);
+ RoutingKeys keys = genKeys.next(rs);
+ next = conflicts.update(keys, txnId);
+ Assertions.assertTrue(txnId.compareTo(next.get(keys)) <= 0);
+ }
- List<DurableBeforeLinear> as = new ArrayList<>();
- List<DurableBefore> bs = new ArrayList<>();
- as.add(DurableBeforeLinear.EMPTY);
- bs.add(DurableBefore.EMPTY);
+ Assertions.assertTrue(isWellFormed(next));
+ conflicts = next;
- Supplier<TestAction> nextAction = rs.weightedPicker(TestAction.values(), new float[]{ 1f, rs.nextInt(10, 100), 1f });
- for (int actionCounter = 0 ; actionCounter < actionCount ; ++actionCounter)
- {
- switch (nextAction.get())
+ if (pruneCount > 0 && (pruneCount >= actionCounter || rs.nextInt((actionCounter + pruneCount - 1) / pruneCount) == 0))
{
- case NEW:
- if (as.size() >= maxDurableBefores)
- {
- --actionCounter;
- }
- else
- {
- as.add(DurableBeforeLinear.EMPTY);
- bs.add(DurableBefore.EMPTY);
- }
- break;
- case MERGE:
- if (as.size() == 1)
- {
- --actionCounter;
- }
- else
- {
- int i = rs.nextInt(as.size());
- int j = rs.nextInt(as.size());
- while (i == j) j = rs.nextInt(as.size());
-
- DurableBeforeLinear old1 = as.get(i);
- DurableBeforeLinear old2 = as.get(j);
- DurableBefore new1 = bs.get(i);
- DurableBefore new2 = bs.get(j);
- DurableBeforeLinear oldmerge = DurableBeforeLinear.merge(old1, old2);
- DurableBefore newmerge = DurableBefore.merge(new1, new2);
- as.set(i, oldmerge);
- bs.set(i, newmerge);
- as.set(j, as.get(as.size() - 1));
- bs.set(j, bs.get(bs.size() - 1));
- as.remove(as.size() - 1);
- bs.remove(bs.size() - 1);
- Assertions.assertTrue(equals(oldmerge, newmerge));
- }
- break;
- case UPDATE:
- int i = rs.nextInt(as.size());
- DurableBeforeLinear prevold = as.get(i);
- DurableBefore prevnew = bs.get(i);
- TxnId syncId1 = genSyncIds.next(rs);
- TxnId syncId2 = genSyncIds.next(rs);
- TxnId q = syncId1.compareTo(syncId2) > 0 ? syncId1 : syncId2;
- TxnId u = q == syncId1 ? syncId2 : syncId1;
- Ranges ranges = genRanges.next(rs);
-
- DurableBeforeLinear nextold = DurableBeforeLinear.merge(prevold, DurableBeforeLinear.create(ranges, q, u));
- DurableBefore nextnew = prevnew.update(ranges, q, u);
- Assertions.assertTrue(equals(nextold, nextnew));
- as.set(i, nextold);
- bs.set(i, nextnew);
+ Ranges ranges = genPruneRanges.next(rs);
+ if (rs.decide(0.5f))
+ {
+ next = next.update(ranges, Timestamp.minForEpoch(epoch.incrementAndGet()));
+ }
+ else
+ {
+ long hlc = rs.nextLong(minHlc.get(), minHlc.get() + maxHlc.get() / 2);
+ long newMinHlc = rs.nextLong(minHlc.get(), hlc + 1);
+ maxHlc.addAndGet(newMinHlc - minHlc.get());
+ minHlc.set(newMinHlc);
+ next = next.update(ranges, Timestamp.fromValues(epoch.get(), hlc, Node.Id.NONE));
+ }
+
+ for (MaxConflicts.Entry e : conflicts)
+ {
+ Assertions.assertTrue(next.get(Ranges.of(e.toPlainRange())).compareTo(e.all) >= 0);
+ }
+ Assertions.assertTrue(isWellFormed(next));
+ conflicts = next;
+ --pruneCount;
}
}
});
}
+ private Gen<Ranges> genRanges(int maxHash, int maxHashRange, int prefixCount)
+ {
+ if (maxHashRange + 1 >= maxHash)
+ {
+ return rs ->
+ {
+ int prefix = 1 + rs.nextInt(prefixCount);
+ return Ranges.of(Range.of(PrefixedIntHashKey.forHash(prefix, 0), PrefixedIntHashKey.forHash(prefix, maxHash)));
+ };
+ }
+
+ return rs -> {
+ int size = 1 + rs.nextInt(2);
+ int prefix = 1 + rs.nextInt(3);
+ Range[] ranges = new Range[size];
+ for (int i = 0 ; i < size ; ++i)
+ {
+ int startHash = rs.nextInt(maxHash - (1 + maxHashRange));
+ int endHash = startHash + 1 + rs.nextInt(maxHashRange);
+ ranges[i] = Range.of(PrefixedIntHashKey.forHash(prefix, startHash), PrefixedIntHashKey.forHash(prefix, endHash));
+ }
+ return Ranges.of(ranges);
+ };
+ }
+
static boolean equals(DurableBeforeLinear as, DurableBefore bs)
{
if (as.isEmpty())
diff --git a/accord-core/src/test/java/accord/utils/AccordGens.java b/accord-core/src/test/java/accord/utils/AccordGens.java
index 563ca2d3..d4040a74 100644
--- a/accord-core/src/test/java/accord/utils/AccordGens.java
+++ b/accord-core/src/test/java/accord/utils/AccordGens.java
@@ -449,6 +449,20 @@ public class AccordGens
};
}
+ public static <T extends RoutingKey> Gen<Range> prefixRanges(Gen<? extends Gen<? extends T>> keyGen)
+ {
+ List<T> keys = Arrays.asList(null, null);
+ return rs -> {
+ Gen<? extends T> gen = keyGen.next(rs);
+ keys.set(0, gen.next(rs));
+ // range doesn't allow a=b
+ do keys.set(1, gen.next(rs));
+ while (Objects.equals(keys.get(0), keys.get(1)));
+ keys.sort(Comparator.naturalOrder());
+ return Range.of(keys.get(0), keys.get(1));
+ };
+ }
+
public static <T extends RoutingKey> Gen<Ranges> ranges(Gen.IntGen sizeGen, Gen<T> keyGen, RangeFactory<T> factory)
{
Gen<Range> rangeGen = ranges(keyGen, factory);
@@ -461,6 +475,18 @@ public class AccordGens
};
}
+ public static <T extends RoutingKey> Gen<Ranges> prefixRanges(Gen.IntGen sizeGen, Gen<Gen<T>> keyGen)
+ {
+ Gen<Range> rangeGen = prefixRanges(keyGen);
+ return rs -> {
+ int size = sizeGen.nextInt(rs);
+ Range[] ranges = new Range[size];
+ for (int i = 0; i < size; i++)
+ ranges[i] = rangeGen.next(rs);
+ return Ranges.of(ranges);
+ };
+ }
+
public static <T extends RoutingKey> Gen<Ranges> ranges(Gen.IntGen sizeGen, Gen<T> keyGen, BiFunction<? super T, ? super T, ? extends Range> factory)
{
return ranges(sizeGen, keyGen, (ignore, a, b) -> factory.apply(a, b));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org