Skip to content

Fix: #279

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
belliottsmith wants to merge 1 commit into
base: trunk
Choose a base branch
from
Open

Fix: #279

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion accord-core/src/main/java/accord/impl/AbstractFetchCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -267,7 +267,7 @@ protected AsyncChain<Data> beginRead(SafeCommandStore safeStore, Timestamp execu
// must be invoked by implementations some time after the read has started OR must override safeToReadAt()
protected void readStarted(SafeCommandStore safeStore)
{
safeToReadAfter = Timestamp.nonNullOrMax(Timestamp.NONE, Timestamp.nonNullOrMax(safeToReadAfter, safeStore.commandStore().unsafeGetMaxConflicts().foldl(MaxConflicts.Entry::get, Timestamp.NONE, TxnId.NONE)));
safeToReadAfter = Timestamp.nonNullOrMax(Timestamp.NONE, Timestamp.nonNullOrMax(safeToReadAfter, safeStore.commandStore().unsafeGetMaxConflicts().foldl(TxnId.NONE, (id, e, min) -> e.get(min, id), Timestamp.NONE)));
}

protected Timestamp safeToReadAfter()
Expand Down
25 changes: 3 additions & 22 deletions accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,33 +72,14 @@ public E get(RoutingKey key)
return BTree.find(tree, (RoutingKey k, Entry<?> e) -> Entry.compare(k, e), key);
}

public E foldl(BiFunction<E, E, E> reduce)
{
// TODO (expected): use BTree fold methods
require(!isEmpty());
Iterator<E> iter = iterator();
E result = iter.next();
while (iter.hasNext())
result = reduce.apply(result, iter.next());
return result;
}

public <V2> V2 foldl(BiFunction<E, V2, V2> reduce, V2 accumulator)
{
// TODO (expected): use BTree fold methods
require(!isEmpty());
for (E e : this)
accumulator = reduce.apply(e, accumulator);
return accumulator;
return BTree.foldl(tree, reduce, accumulator);
}

public <V2, P1> V2 foldl(TriFunction<E, V2, P1, V2> reduce, V2 accumulator, P1 p1)
public <V2, P1> V2 foldl(P1 p1, TriFunction<P1, E, V2, V2> reduce, V2 accumulator)
{
// TODO (expected): use BTree fold methods
require(!isEmpty());
for (E e : this)
accumulator = reduce.apply(e, accumulator, p1);
return accumulator;
return BTree.foldl(tree, p1, reduce, accumulator);
}

@Override
Expand Down
28 changes: 28 additions & 0 deletions accord-core/src/main/java/accord/utils/btree/BTree.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import accord.utils.AsymmetricComparator;
import accord.utils.Invariants;
import accord.utils.SortedArrays;
import accord.utils.TriFunction;
import accord.utils.btree.IntervalBTree.IntervalMaxIndex;
import accord.utils.btree.UpdateFunction.NoOp;

Expand Down Expand Up @@ -1848,6 +1849,33 @@ public static <V, A> void apply(Object[] btree, BiConsumer<A, V> function, A arg
}
}

public static <I, O> O foldl(Object[] btree, BiFunction<I, O, O> function, O accumulator)
{
return BTree.<I, BiFunction<I, O, O>, O>foldl(btree, function, BiFunction::apply, accumulator);
}

public static <I, P, O> O foldl(Object[] btree, P param, TriFunction<P, I, O, O> function, O accumulator)
{
if (isLeaf(btree))
return foldlLeaf(btree, function, param, accumulator);

int keys = getBranchKeyEnd(btree);
for (int i = 0; i < keys; i++)
{
accumulator = foldl((Object[]) btree[keys + i], param, function, accumulator);
accumulator = function.apply(param, (I) btree[i], accumulator);
}
return foldl((Object[]) btree[2 * keys], param, function, accumulator);
}

private static <I, P, O> O foldlLeaf(Object[] btree, TriFunction<P, I, O, O> function, P param, O accumulator)
{
int limit = sizeOfLeaf(btree);
for (int i = 0; i < limit; i++)
accumulator = function.apply(param, (I) btree[i], accumulator);
return accumulator;
}

/**
* Simple method to walk the btree forwards and apply a function till a stop condition is reached
* <p>
Expand Down
17 changes: 15 additions & 2 deletions accord-core/src/main/java/accord/utils/btree/ReducingBTree.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -573,7 +573,14 @@ public static <E extends Entry<E>, V2, P1, P2> V2 foldl(Object[] tree, AbstractR
{
ri = ranges.findNext(ri, to, ev.start(), ReducingBTree::compareWithEnd, FAST);
if (ri < 0) ri = childTo = -1 - ri;
else childTo = ri + 1;
else
{
childTo = ri + 1;

// we intersect another range; refresh rv to ensure we advance correctly
rv = ranges.get(ri);
ces = rv.end().compareTo(ev.start());
}
}
if (childTo > childFrom)
{
Expand Down Expand Up @@ -740,7 +747,13 @@ private static <E extends Entry<E>, V2, P1, P2> V2 foldlWithDefault(E lb, E ub,
{
ri = ranges.findNext(ri, to, ev.start(), ReducingBTree::compareWithEnd, FAST);
if (ri < 0) ri = childTo = -1 - ri;
else childTo = ri + 1;
else
{
childTo = ri + 1;
// we intersect another range but same tree child range; refresh rv to ensure we advance correctly
rv = ranges.get(ri);
ces = rv.end().compareTo(ev.start());
}
}
if (childTo > childFrom)
{
Expand Down
40 changes: 31 additions & 9 deletions accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;

import static accord.api.ProtocolModifiers.isRangeEndInclusive;
import static java.lang.Integer.MAX_VALUE;
import static java.lang.Integer.MIN_VALUE;

Expand Down Expand Up @@ -85,7 +86,10 @@ public static Entry mergeMax(RoutingKey start, RoutingKey end, Entry a, Entry b)
static final BTreeReducingRangeMap<Entry> EMPTY = new BTreeReducingRangeMap<>();
static final RoutingKey MINIMUM_EXCL = new IntKey.Routing(MIN_VALUE);
static final RoutingKey MAXIMUM_EXCL = new IntKey.Routing(MAX_VALUE);
static boolean END_INCLUSIVE = false;
// Must match the production setting: BTreeReducingRangeMap.get/foldl use isRangeEndInclusive(),
// so the canonical TreeMap (which models intervals as (prev, K] via ceilingEntry) only agrees
// with the map under test when END_INCLUSIVE has the same value.
static boolean END_INCLUSIVE = isRangeEndInclusive();

private static IntKey.Routing rk(int t)
{
Expand Down Expand Up @@ -169,8 +173,8 @@ public void testRandomAdds() throws ExecutionException, InterruptedException
{
ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
List<ListenableFuture<Void>> results = new ArrayList<>();
int count = 100000;
for (int numberOfAdditions : new int[] { 1, 10, 100 })
int count = 1000;
for (int numberOfAdditions : new int[] { 100, 10, 1 })
{
for (float maxCoveragePerRange : new float[] { 0.01f, 0.1f, 0.5f })
{
Expand Down Expand Up @@ -302,6 +306,11 @@ Timestamp get(RoutingKey rk)
return canonical.ceilingEntry(rk).getValue();
}

static Timestamp tsOrNull(Entry e)
{
return e == null ? null : e.timestamp;
}

RandomWithCanonical merge(Random random, RandomWithCanonical other)
{
RandomWithCanonical result = new RandomWithCanonical();
Expand Down Expand Up @@ -352,9 +361,9 @@ void validate(Random random, String id)
{
for (RoutingKey rk : canonical.keySet())
{
Assertions.assertEquals(get(decr(rk)), test.get(decr(rk)), id);
Assertions.assertEquals(get(rk), test.get(rk), id);
Assertions.assertEquals(get(incr(rk)), test.get(incr(rk)), id);
Assertions.assertEquals(get(decr(rk)), tsOrNull(test.get(decr(rk))), id);
Assertions.assertEquals(get(rk), tsOrNull(test.get(rk)), id);
Assertions.assertEquals(get(incr(rk)), tsOrNull(test.get(incr(rk))), id);
}

// check some random
Expand All @@ -363,7 +372,7 @@ void validate(Random random, String id)
while (remaining-- > 0)
{
RoutingKey routingKey = rk(random);
Assertions.assertEquals(get(routingKey), test.get(routingKey), id);
Assertions.assertEquals(get(routingKey), tsOrNull(test.get(routingKey)), id);
}
}

Expand Down Expand Up @@ -395,7 +404,7 @@ void validate(Random random, String id)
}

List<Timestamp> foldl = test.foldl(keys, (e, timestamps) -> {
if (timestamps.isEmpty() || !timestamps.get(timestamps.size() - 1).equals(e))
if (timestamps.isEmpty() || !timestamps.get(timestamps.size() - 1).equals(e.timestamp))
timestamps.add(e.timestamp);
return timestamps;
}, new ArrayList<>());
Expand All @@ -412,7 +421,7 @@ void validate(Random random, String id)
Assertions.assertEquals(canonFoldl, foldl, id);

foldl = test.foldl(ranges, (e, timestamps) -> {
if (timestamps.isEmpty() || !timestamps.get(timestamps.size() - 1).equals(e))
if (timestamps.isEmpty() || !timestamps.get(timestamps.size() - 1).equals(e.timestamp))
timestamps.add(e.timestamp);
return timestamps;
}, new ArrayList<>());
Expand All @@ -432,6 +441,19 @@ void validate(Random random, String id)
}
}
Assertions.assertEquals(canonFoldl, foldl, id);

foldl = test.foldl((e, timestamps) -> {
timestamps.add(e.timestamp);
return timestamps;
}, new ArrayList<>());

canonFoldl.clear();
for (Timestamp v : canonical.values())
{
if (v != null)
canonFoldl.add(v);
}
Assertions.assertEquals(canonFoldl, foldl, id);
}
}
}
Expand Down
14 changes: 6 additions & 8 deletions accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
import accord.primitives.Timestamp;
import org.opentest4j.AssertionFailedError;

import static accord.api.ProtocolModifiers.isRangeEndInclusive;
import static java.lang.Integer.MAX_VALUE;
import static java.lang.Integer.MIN_VALUE;

Expand All @@ -54,7 +55,10 @@ public class ReducingRangeMapTest
static final ReducingRangeMap<Timestamp> EMPTY = new ReducingRangeMap<>();
static final RoutingKey MINIMUM_EXCL = new IntKey.Routing(MIN_VALUE);
static final RoutingKey MAXIMUM_EXCL = new IntKey.Routing(MAX_VALUE);
static boolean END_INCLUSIVE = false;
// Must match the production setting: ReducingRangeMap.get/foldl use isRangeEndInclusive(),
// so the canonical TreeMap (which models intervals as (prev, K] via ceilingEntry) only agrees
// with the map under test when END_INCLUSIVE has the same value.
static boolean END_INCLUSIVE = isRangeEndInclusive();

private static RoutingKey rk(int t)
{
Expand Down Expand Up @@ -168,7 +172,7 @@ public void testRandomAdds() throws ExecutionException, InterruptedException
{
ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
List<ListenableFuture<Void>> results = new ArrayList<>();
int count = 100000;
int count = 1000;
for (int numberOfAdditions : new int[] { 1, 10, 100 })
{
for (float maxCoveragePerRange : new float[] { 0.01f, 0.1f, 0.5f })
Expand Down Expand Up @@ -336,12 +340,6 @@ protected Timestamp reduce(Timestamp a, Timestamp b)
return Timestamp.max(a, b);
}

@Override
protected Timestamp tryMergeEqual(Timestamp a, Timestamp b)
{
return a;
}

@Override
protected ReducingRangeMap<Timestamp> buildInternal()
{
Expand Down