You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by GitBox <gi...@apache.org> on 2023/01/13 01:18:15 UTC
[GitHub] [cassandra-accord] aweisberg commented on a diff in pull request #23: Introduce range transactions, including burn test support for for read-only transactions
aweisberg commented on code in PR #23:
URL: https://github.com/apache/cassandra-accord/pull/23#discussion_r1067413860
##########
accord-core/src/main/java/accord/coordinate/CheckOn.java:
##########
@@ -136,21 +138,28 @@ class OnDone implements MapReduceConsume<SafeCommandStore, Void>
public OnDone()
{
- Ranges localRanges = node.topology().localRangesForEpochs(txnId.epoch, untilLocalEpoch);
- PartialRoute<?> selfRoute = route().slice(localRanges);
+ Ranges sliceRanges = node.topology().localRangesForEpochs(txnId.epoch(), untilLocalEpoch);
+ PartialRoute<?> selfRoute = route().slice(sliceRanges, Maximal);
Review Comment:
Maximal is `/** Overlapping ranges are extended to the union of the overlaps */`
What does union of the overlaps mean? It seems like this behavior can cause the route to include ranges not in `sliceRanges`? Why is that useful vs just where they intersect?
##########
accord-core/src/main/java/accord/impl/InMemoryCommandStore.java:
##########
@@ -181,7 +197,14 @@ public NodeTimeService time()
private Timestamp maxConflict(Seekables<?, ?> keysOrRanges, Ranges slice)
{
- return mapReduce(keysOrRanges, slice, CommandsForKey::max, (a, b) -> a.compareTo(b) >= 0 ? a : b, Timestamp.NONE);
+ Timestamp timestamp = mapReduceForKey(keysOrRanges, slice, (forKey, prev) -> Timestamp.max(forKey.max(), prev), Timestamp.NONE, null);
Review Comment:
Some commands are indexed by key, some are indexed by range so we compare with every range command for now to find the max conflicting timestamp?
##########
accord-core/src/main/java/accord/impl/InMemoryCommandsForKey.java:
##########
@@ -74,25 +74,27 @@ public boolean isEmpty()
}
@Override
- public Stream<T> before(@Nonnull Timestamp timestamp, @Nonnull TestKind testKind, @Nonnull TestDep testDep, @Nullable TxnId depId, @Nonnull TestStatus testStatus, @Nullable Status status)
+ public <T> T mapReduce(TestKind testKind, TestTimestamp testTimestamp, Timestamp timestamp,
+ TestDep testDep, @Nullable TxnId depId,
+ @Nullable Status minStatus, @Nullable Status maxStatus,
+ CommandFunction<T, T> map, T initialValue, T terminalValue)
{
- return commands.headMap(timestamp, false).values().stream()
- .filter(cmd -> testKind == RorWs || cmd.txnId().isWrite())
- // If we don't have any dependencies, we treat a dependency filter as a mismatch
- .filter(cmd -> testDep == ANY_DEPS || (cmd.known().deps != DepsUnknown && (cmd.partialDeps().contains(depId) ^ (testDep == WITHOUT))))
- .filter(cmd -> TestStatus.test(cmd.status(), testStatus, status))
- .map(map);
- }
- @Override
- public Stream<T> after(@Nonnull Timestamp timestamp, @Nonnull TestKind testKind, @Nonnull TestDep testDep, @Nullable TxnId depId, @Nonnull TestStatus testStatus, @Nullable Status status)
- {
- return commands.tailMap(timestamp, false).values().stream()
- .filter(cmd -> testKind == RorWs || cmd.txnId().isWrite())
+ for (Command cmd : (testTimestamp == TestTimestamp.BEFORE ? commands.headMap(timestamp, false) : commands.tailMap(timestamp, false)).values())
+ {
+ if (testKind == Ws && cmd.txnId().isRead()) continue;
// If we don't have any dependencies, we treat a dependency filter as a mismatch
- .filter(cmd -> testDep == ANY_DEPS || (cmd.known().deps != DepsUnknown && (cmd.partialDeps().contains(depId) ^ (testDep == WITHOUT))))
- .filter(cmd -> TestStatus.test(cmd.status(), testStatus, status))
- .map(map);
+ if (testDep != ANY_DEPS && (cmd.known().deps == DepsUnknown || (cmd.partialDeps().contains(depId) != (testDep == WITH))))
Review Comment:
Subtle difference with how this is done in `InMemoryCommandStore` for range transactions is `hasProposedOrDecidedDeps` vs `deps == DepsUnkown` which will behave differently for `NoDeps`.
Should `hasProposedOrDecidedDeps` return true for `NoDeps`?
##########
accord-core/src/main/java/accord/messages/CheckStatus.java:
##########
@@ -37,6 +37,7 @@
import static accord.local.Status.*;
import static accord.local.Status.Durability.NotDurable;
import static accord.messages.TxnRequest.computeScope;
+import static accord.primitives.Routables.Slice.Overlapping;
Review Comment:
Unused
##########
accord-core/src/main/java/accord/coordinate/ReadCoordinator.java:
##########
@@ -16,7 +16,7 @@
abstract class ReadCoordinator<Reply extends accord.messages.Reply> extends ReadTracker implements Callback<Reply>
{
- private static final boolean DEBUG = false;
+ private static final boolean DEBUG = true;
Review Comment:
Don't forget to turn this off
##########
accord-core/src/main/java/accord/messages/WaitOnCommit.java:
##########
@@ -29,6 +29,7 @@
import org.slf4j.LoggerFactory;
import static accord.local.Status.Committed;
+import static accord.primitives.Routables.Slice.Overlapping;
Review Comment:
Unused
##########
accord-core/src/main/java/accord/coordinate/CheckOn.java:
##########
@@ -136,21 +138,28 @@ class OnDone implements MapReduceConsume<SafeCommandStore, Void>
public OnDone()
{
- Ranges localRanges = node.topology().localRangesForEpochs(txnId.epoch, untilLocalEpoch);
- PartialRoute<?> selfRoute = route().slice(localRanges);
+ Ranges sliceRanges = node.topology().localRangesForEpochs(txnId.epoch(), untilLocalEpoch);
+ PartialRoute<?> selfRoute = route().slice(sliceRanges, Maximal);
+ if (selfRoute.domain().isRange())
+ {
+ // for range queries, we want to "cover" the entirety of every range
Review Comment:
Why do we want to cover the entire thing if we don't replicate the data?
##########
accord-core/src/main/java/accord/impl/InMemoryCommand.java:
##########
@@ -129,7 +131,7 @@ public Route<?> route()
@Override
protected void setRoute(Route<?> route)
{
- this.route = route;
+ this.route = Invariants.checkArgument(route, !route.isEmpty());
Review Comment:
##########
accord-core/src/main/java/accord/impl/InMemoryCommandStore.java:
##########
@@ -214,36 +237,176 @@ public void forCommittedInEpoch(Ranges ranges, long epoch, Consumer<Command> con
range.endInclusive()).values();
for (InMemoryCommandsForKey commands : rangeCommands)
{
- commands.committedByExecuteAt()
+ commands.byExecuteAt()
.between(minTimestamp, maxTimestamp)
+ .filter(command -> command.hasBeen(Committed))
.forEach(consumer);
}
}
}
- public <T> T mapReduce(Routables<?, ?> keysOrRanges, Ranges slice, Function<CommandsForKey, T> map, BinaryOperator<T> reduce, T initialValue)
+ @Override
+ public <T> T mapReduce(Seekables<?, ?> keysOrRanges, Ranges slice, TestKind testKind, TestTimestamp testTimestamp, Timestamp timestamp, TestDep testDep, @Nullable TxnId depId, @Nullable Status minStatus, @Nullable Status maxStatus, CommandFunction<T, T> map, T accumulate, T terminalValue)
{
- switch (keysOrRanges.kindOfContents()) {
+ accumulate = mapReduceForKey(keysOrRanges, slice, (forKey, prev) -> {
+ CommandTimeseries timeseries;
+ switch (testTimestamp)
+ {
+ default: throw new AssertionError();
+ case STARTED_AFTER:
+ case STARTED_BEFORE:
+ timeseries = forKey.byId();
+ break;
+ case EXECUTES_AFTER:
+ case MAY_EXECUTE_BEFORE:
+ timeseries = forKey.byExecuteAt();
+ }
+ CommandTimeseries.TestTimestamp remapTestTimestamp;
+ switch (testTimestamp)
+ {
+ default: throw new AssertionError();
+ case STARTED_AFTER:
+ case EXECUTES_AFTER:
+ remapTestTimestamp = CommandTimeseries.TestTimestamp.AFTER;
+ break;
+ case STARTED_BEFORE:
+ case MAY_EXECUTE_BEFORE:
+ remapTestTimestamp = CommandTimeseries.TestTimestamp.BEFORE;
+ }
+ return timeseries.mapReduce(testKind, remapTestTimestamp, timestamp, testDep, depId, minStatus, maxStatus, map, prev, terminalValue);
+ }, accumulate, terminalValue);
+
+ if (accumulate.equals(terminalValue))
+ return accumulate;
+
+ // TODO (find lib, efficiency): this is super inefficient, need to store Command in something queryable
+ Seekables<?, ?> sliced = keysOrRanges.slice(slice, Minimal);
+ Map<Range, List<Command>> collect = new TreeMap<>(Range::compare);
+ for (RangeCommand rangeCommand : rangeCommands.values())
+ {
+ Command command = rangeCommand.command;
+ switch (testTimestamp)
+ {
+ default: throw new AssertionError();
+ case STARTED_AFTER:
+ if (command.txnId().compareTo(timestamp) < 0) continue;
+ else break;
+ case STARTED_BEFORE:
+ if (command.txnId().compareTo(timestamp) > 0) continue;
+ else break;
+ case EXECUTES_AFTER:
+ if (command.executeAt().compareTo(timestamp) < 0) continue;
+ else break;
+ case MAY_EXECUTE_BEFORE:
+ Timestamp compareTo = command.known().executeAt.hasDecidedExecuteAt() ? command.executeAt() : command.txnId();
+ if (compareTo.compareTo(timestamp) > 0) continue;
+ else break;
+ }
+
+ if (minStatus != null && command.status().compareTo(minStatus) < 0)
+ continue;
+
+ if (maxStatus != null && command.status().compareTo(maxStatus) > 0)
+ continue;
+
+ if (testKind == Ws && command.txnId().rw().isRead())
+ continue;
+
+ if (testDep != ANY_DEPS)
+ {
+ if (!command.known().deps.hasProposedOrDecidedDeps())
+ continue;
+
+ if ((testDep == WITH) == !command.partialDeps().contains(depId))
+ continue;
+ }
+
+ if (!rangeCommand.ranges.intersects(sliced))
+ continue;
+
+ Routables.foldl(rangeCommand.ranges, sliced, (r, in, i) -> {
+ // TODO (easy, efficiency): pass command as a parameter to Fold
+ List<Command> list = in.computeIfAbsent(r, ignore -> new ArrayList<>());
+ if (list.isEmpty() || list.get(list.size() - 1) != command)
+ list.add(command);
+ return in;
+ }, collect);
+ }
+
+ for (Map.Entry<Range, List<Command>> e : collect.entrySet())
+ {
+ for (Command command : e.getValue())
+ accumulate = map.apply(e.getKey(), command.txnId(), command.executeAt(), accumulate);
+ }
+
+ return accumulate;
+ }
+
+ @Override
+ public void register(Seekables<?, ?> keysOrRanges, Ranges slice, Command command)
+ {
+ switch (keysOrRanges.domain())
+ {
+ default: throw new AssertionError();
+ case Key:
+ forEach(keysOrRanges, slice, forKey -> forKey.register(command));
+ break;
+ case Range:
+ rangeCommands.computeIfAbsent(command.txnId(), ignore -> new RangeCommand(command))
+ .update((Ranges)keysOrRanges);
+ }
+ }
+
+ @Override
+ public void register(Seekable keyOrRange, Ranges slice, Command command)
+ {
+ switch (keyOrRange.domain())
+ {
+ default: throw new AssertionError();
+ case Key:
+ forEach(keyOrRange, slice, forKey -> forKey.register(command));
+ break;
+ case Range:
+ rangeCommands.computeIfAbsent(command.txnId(), ignore -> new RangeCommand(command))
+ .update(Ranges.of((Range)keyOrRange));
+ }
+ }
+
+ public <O> O mapReduceForKey(Routables<?, ?> keysOrRanges, Ranges slice, BiFunction<CommandsForKey, O, O> map, O accumulate, O terminalValue)
+ {
+ switch (keysOrRanges.domain()) {
default:
throw new AssertionError();
case Key:
AbstractKeys<Key, ?> keys = (AbstractKeys<Key, ?>) keysOrRanges;
- return keys.stream()
- .filter(slice::contains)
- .map(this::commandsForKey)
- .map(map)
- .reduce(initialValue, reduce);
+ for (Key key : keys)
Review Comment:
De-streaming for performance?
##########
accord-core/src/main/java/accord/local/Command.java:
##########
@@ -601,14 +603,18 @@ private boolean maybeExecute(SafeCommandStore safeStore, ProgressShard shard, bo
switch (status())
{
case Committed:
+ // TODO (desirable, efficiency): maintain distinct ReadyToRead and ReadyToWrite states
setStatus(ReadyToExecute);
logger.trace("{}: set to ReadyToExecute", txnId());
safeStore.progressLog().readyToExecute(this, shard);
notifyListeners(safeStore);
break;
case PreApplied:
- if (executeRanges(safeStore, executeAt()).intersects(writes().keys))
+ Ranges executeRanges = executeRanges(safeStore, executeAt());
+ boolean intersects = switchOnDomain(writes().keys, executeRanges, AbstractKeys::intersects, AbstractRanges::intersects);
Review Comment:
Why is `switchOnDomain` necessary? Can intersection be done without having the caller use `switchOnDomain`?
##########
accord-core/src/main/java/accord/local/Command.java:
##########
@@ -911,57 +917,50 @@ enum EnsureAction { Ignore, Check, Add, TrySet, Set }
* Validate we have sufficient information for the route, partialTxn and partialDeps fields, and if so update them;
* otherwise return false (or throw an exception if an illegal state is encountered)
*/
- private boolean validate(Ranges existingRanges, Ranges additionalRanges, ProgressShard shard,
- Route<?> route, EnsureAction ensureRoute,
+ private boolean validate(Ranges existingRanges, Ranges additionalRanges,
+ ProgressShard shard, Route<?> route, EnsureAction ensureRoute,
+ @Nullable PartialTxn partialTxn, EnsureAction ensurePartialTxn,
+ @Nullable PartialDeps partialDeps, EnsureAction ensurePartialDeps)
+ {
+ return validate(existingRanges, additionalRanges, additionalRanges, shard, route, ensureRoute, partialTxn, ensurePartialTxn, partialDeps, ensurePartialDeps);
+ }
+ private boolean validate(Ranges existingRanges, Ranges additionalRouteRanges, Ranges additionalTxnRanges,
Review Comment:
Additional txn ranges looks like it is always `Ranges.EMPTY` or the same as `acceptRanges`
It's also really not clear what it is. `validate` could use documentation of what its parameters are.
##########
accord-core/src/main/java/accord/local/Command.java:
##########
@@ -292,30 +294,28 @@ public AcceptOutcome accept(SafeCommandStore safeStore, Ballot ballot, Kind kind
return AcceptOutcome.Redundant;
}
- if (known().isDefinitionKnown() && txnId().rw() != kind)
Review Comment:
Kind is not worth plumbing through everywhere for just this check?
##########
accord-core/src/main/java/accord/primitives/AbstractRanges.java:
##########
@@ -210,9 +246,102 @@ static <RS extends AbstractRanges<?>, P> RS slice(Ranges covering, AbstractRange
ri = (int) (lri >>> 32);
Range l = covering.ranges[li], r = input.ranges[ri];
- buffer[bufferCount++] = Range.slice(l, r);
- if (l.end().compareTo(r.end()) >= 0) ri++;
- else li++;
+ buffer[bufferCount++] = r;
+ if (l.end().compareTo(r.end()) <= 0) li++;
+ ri++;
+ }
+ Range[] result = cachedRanges.complete(buffer, bufferCount);
+ cachedRanges.discard(buffer, bufferCount);
+ return constructor.construct(covering, param, result);
+ }
+ catch (Throwable t)
+ {
+ cachedRanges.forceDiscard(buffer, bufferCount);
+ throw t;
+ }
+ }
+
+ static <RS extends AbstractRanges<?>, P> RS sliceMinimal(Ranges covering, AbstractRanges<?> input, P param, SliceConstructor<P, RS> constructor)
+ {
+ ObjectBuffers<Range> cachedRanges = cachedRanges();
+
+ Range[] buffer = cachedRanges.get(covering.ranges.length + input.ranges.length);
+ int bufferCount = 0;
+ try
+ {
+ int li = 0, ri = 0;
+ while (true)
+ {
+ long lri = covering.findNextIntersection(li, input, ri);
+ if (lri < 0)
+ break;
+
+ if (bufferCount == buffer.length)
+ buffer = cachedRanges.resize(buffer, bufferCount, bufferCount + 1 + (bufferCount/2));
+
+ li = (int) (lri);
+ ri = (int) (lri >>> 32);
+
+ Range l = covering.ranges[li], r = input.ranges[ri];
+ RoutingKey ls = l.start(), rs = r.start(), le = l.end(), re = r.end();
+ int cs = rs.compareTo(ls), ce = re.compareTo(le);
+ if (cs >= 0 && ce <= 0)
+ {
+ buffer[bufferCount++] = r;
+ ++ri;
+ }
+ else
+ {
+ buffer[bufferCount++] = r.subRange(cs >= 0 ? rs : ls, ce <= 0 ? re : le);
+ if (ce <= 0) ++ri;
+ }
+ if (ce >= 0) li++; // le <= re
+ }
+ Range[] result = cachedRanges.complete(buffer, bufferCount);
+ cachedRanges.discard(buffer, bufferCount);
+ return constructor.construct(covering, param, result);
+ }
+ catch (Throwable t)
+ {
+ cachedRanges.forceDiscard(buffer, bufferCount);
+ throw t;
+ }
+ }
+
+ static <RS extends AbstractRanges<?>, P> RS sliceMaximal(Ranges covering, AbstractRanges<?> input, P param, SliceConstructor<P, RS> constructor)
+ {
+ ObjectBuffers<Range> cachedRanges = cachedRanges();
+
+ Range[] buffer = cachedRanges.get(covering.ranges.length + input.ranges.length);
+ int bufferCount = 0;
+ try
+ {
+ int li = 0, ri = 0;
+ while (true)
+ {
+ long lri = covering.findNextIntersection(li, input, ri);
+ if (lri < 0)
+ break;
+
+ if (bufferCount == buffer.length)
+ buffer = cachedRanges.resize(buffer, bufferCount, bufferCount + 1 + (bufferCount/2));
+
+ li = (int) (lri);
+ ri = (int) (lri >>> 32);
+
+ Range l = covering.ranges[li], r = input.ranges[ri]; // l(eft), r(right)
+ RoutingKey ls = l.start(), rs = r.start(), le = l.end(), re = r.end(); // l(eft),r(ight) s(tart),e(nd)
+ int cs = rs.compareTo(ls), ce = re.compareTo(le); // c(ompare) s(tart),e(nd)
+ if (cs <= 0 && ce >= 0)
+ {
+ buffer[bufferCount++] = r;
+ }
+ else
+ {
+ buffer[bufferCount++] = r.subRange(cs <= 0 ? rs : ls, ce >= 0 ? re : le);
Review Comment:
It was confusing that `subRange` constructs a new range, but it isn't actually a `subRange` of the range you invoked `subRange` on.
##########
accord-core/src/main/java/accord/local/Node.java:
##########
@@ -363,7 +363,7 @@ public TxnId nextTxnId(Txn.Kind rw, Domain domain)
public Future<Result> coordinate(Txn txn)
{
- return coordinate(nextTxnId(txn.kind(), Key), txn);
+ return coordinate(nextTxnId(txn.kind(), txn.keys().domain()), txn);
Review Comment:
Surprised to see domain here that is only used once in `SomeRoute`. The home key itself determines the domain so I don't see why the TxnId needs it?
##########
accord-core/src/main/java/accord/messages/Accept.java:
##########
@@ -39,51 +39,55 @@
import javax.annotation.Nullable;
import static accord.local.Command.AcceptOutcome.*;
-import static accord.messages.PreAccept.calculatePartialDeps;
// TODO (low priority, efficiency): use different objects for send and receive, so can be more efficient
// (e.g. serialize without slicing, and without unnecessary fields)
public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
{
public static class SerializerSupport
{
- public static Accept create(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, PartialDeps partialDeps, Txn.Kind kind)
+ public static Accept create(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, PartialDeps partialDeps)
{
- return new Accept(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey, ballot, executeAt, keys, partialDeps, kind);
+ return new Accept(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey, ballot, executeAt, keys, partialDeps);
}
}
public final Ballot ballot;
public final Timestamp executeAt;
public final Seekables<?, ?> keys;
public final PartialDeps partialDeps;
- public final Txn.Kind kind;
- public Accept(Id to, Topologies topologies, Ballot ballot, TxnId txnId, FullRoute<?> route, Timestamp executeAt, Seekables<?, ?> keys, Deps deps, Txn.Kind kind)
+ public Accept(Id to, Topologies topologies, Ballot ballot, TxnId txnId, FullRoute<?> route, Timestamp executeAt, Seekables<?, ?> keys, Deps deps)
{
super(to, topologies, txnId, route);
this.ballot = ballot;
this.executeAt = executeAt;
this.keys = keys.slice(scope.covering());
this.partialDeps = deps.slice(scope.covering());
- this.kind = kind;
}
- private Accept(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, PartialDeps partialDeps, Txn.Kind kind)
+ private Accept(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, PartialDeps partialDeps)
{
super(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey);
this.ballot = ballot;
this.executeAt = executeAt;
this.keys = keys;
this.partialDeps = partialDeps;
- this.kind = kind;
}
@Override
public synchronized AcceptReply apply(SafeCommandStore safeStore)
{
+ if (minEpoch < txnId.epoch())
Review Comment:
How does `minEpoch` end up < `txnId.epoch` if we calculate `minEpoch` from a `Topologies` that was itself calculated using `TxnId.Epoch`?
##########
accord-core/src/main/java/accord/primitives/Ranges.java:
##########
@@ -89,10 +96,29 @@ public Ranges union(Ranges that)
return union(MERGE_OVERLAPPING, that);
}
+ @Override
+ public Ranges with(Unseekables<Range, ?> that)
+ {
+ return with((AbstractRanges<?>) that);
+ }
+
+ public Ranges with(Ranges that)
+ {
+ return union(MERGE_OVERLAPPING, that);
+ }
+
+ public Ranges with(AbstractRanges<?> that)
Review Comment:
Why have multiple overlapping versions of the same thing? `union` ends up being over` AbstractRanges`. Is it some kind of monomorphic call site thing?
##########
accord-core/src/main/java/accord/primitives/Unseekables.java:
##########
@@ -45,7 +49,7 @@ public boolean isFullRoute()
return right;
// non-route types can always accept route types as input, so just call its union method on the other
- return leftKind.isRoute() ? ((Unseekables)right).union(left) : ((Unseekables)left).union(right);
+ return left.with(right);
Review Comment:
Up above where it says `// non-route types can always accept route types as input, so just call its union method on the other`
How has it established that `left` is not a route?
Really it looks like `with` just doesn't care. I'm not even clear on what the difference between `with` and `union` so maybe it needs a name change or documentation?
##########
accord-core/src/main/java/accord/local/PreLoadContext.java:
##########
@@ -33,11 +34,19 @@
{
/**
* @return ids of the {@link Command} objects that need to be loaded into memory before this operation is run
+ *
+ * TODO (expected): this is used for Apply, NotifyWaitingOn and listenerContexts; others only use a single txnId
+ * firstly, it would be nice to simply have that txnId as a single value.
+ * In the case of Apply, we can likely avoid loading all
Review Comment:
all transactions?
##########
accord-core/src/main/java/accord/messages/Accept.java:
##########
@@ -39,51 +39,55 @@
import javax.annotation.Nullable;
import static accord.local.Command.AcceptOutcome.*;
-import static accord.messages.PreAccept.calculatePartialDeps;
// TODO (low priority, efficiency): use different objects for send and receive, so can be more efficient
// (e.g. serialize without slicing, and without unnecessary fields)
public class Accept extends TxnRequest.WithUnsynced<Accept.AcceptReply>
{
public static class SerializerSupport
{
- public static Accept create(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, PartialDeps partialDeps, Txn.Kind kind)
+ public static Accept create(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, PartialDeps partialDeps)
{
- return new Accept(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey, ballot, executeAt, keys, partialDeps, kind);
+ return new Accept(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey, ballot, executeAt, keys, partialDeps);
}
}
public final Ballot ballot;
public final Timestamp executeAt;
public final Seekables<?, ?> keys;
public final PartialDeps partialDeps;
- public final Txn.Kind kind;
- public Accept(Id to, Topologies topologies, Ballot ballot, TxnId txnId, FullRoute<?> route, Timestamp executeAt, Seekables<?, ?> keys, Deps deps, Txn.Kind kind)
+ public Accept(Id to, Topologies topologies, Ballot ballot, TxnId txnId, FullRoute<?> route, Timestamp executeAt, Seekables<?, ?> keys, Deps deps)
{
super(to, topologies, txnId, route);
this.ballot = ballot;
this.executeAt = executeAt;
this.keys = keys.slice(scope.covering());
this.partialDeps = deps.slice(scope.covering());
- this.kind = kind;
}
- private Accept(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, PartialDeps partialDeps, Txn.Kind kind)
+ private Accept(TxnId txnId, PartialRoute<?> scope, long waitForEpoch, long minEpoch, boolean doNotComputeProgressKey, Ballot ballot, Timestamp executeAt, Seekables<?, ?> keys, PartialDeps partialDeps)
{
super(txnId, scope, waitForEpoch, minEpoch, doNotComputeProgressKey);
this.ballot = ballot;
this.executeAt = executeAt;
this.keys = keys;
this.partialDeps = partialDeps;
- this.kind = kind;
}
@Override
public synchronized AcceptReply apply(SafeCommandStore safeStore)
{
+ if (minEpoch < txnId.epoch())
+ {
+ Ranges acceptRanges = safeStore.ranges().between(txnId.epoch(), executeAt.epoch());
+ if (!acceptRanges.intersects(scope))
+ return new AcceptReply(calculatePartialDeps(safeStore));
Review Comment:
Why are the locally known deps relevant if none of the scope of the txn was replicated locally during the transactions epochs?
##########
accord-core/src/main/java/accord/messages/BeginInvalidation.java:
##########
@@ -10,6 +10,7 @@
import java.util.Collections;
import java.util.List;
+import static accord.primitives.Routables.Slice.Overlapping;
Review Comment:
Unused
##########
accord-core/src/main/java/accord/impl/InMemoryCommandsForKey.java:
##########
@@ -108,15 +110,16 @@ public Stream<Command> all()
// TODO (now): add validation that anything inserted into *committedBy* has everything prior in its dependencies
private final Key key;
- private final InMemoryCommandTimeseries<TxnIdWithExecuteAt> uncommitted = new InMemoryCommandTimeseries<>(cmd -> new TxnIdWithExecuteAt(cmd.txnId(), cmd.executeAt()));
Review Comment:
I got to the end and never figured out why `uncommitted` could be replaced by `byId`. Does the `TestTimestamp` take care of that now?
##########
accord-core/src/main/java/accord/primitives/Unseekables.java:
##########
@@ -45,7 +49,7 @@ public boolean isFullRoute()
return right;
// non-route types can always accept route types as input, so just call its union method on the other
- return leftKind.isRoute() ? ((Unseekables)right).union(left) : ((Unseekables)left).union(right);
+ return left.with(right);
Review Comment:
I just realized `union`, `with`, and `merge` are all the same thing?
Why `with` here and `union` at the end of the function?
##########
accord-core/src/main/java/accord/primitives/Deps.java:
##########
@@ -18,1311 +18,220 @@
package accord.primitives;
-import java.util.*;
-import java.util.function.BiConsumer;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Predicate;
-
-import java.util.stream.Collectors;
-
import accord.api.Key;
-import accord.utils.ArrayBuffers;
-import accord.api.RoutingKey;
-import accord.utils.SortedArrays;
import accord.utils.Invariants;
-import static accord.utils.ArrayBuffers.*;
-import static accord.utils.SortedArrays.*;
-import static accord.utils.SortedArrays.Search.FAST;
+import java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
/**
- * A collection of dependencies for a transaction, organised by the key the dependency is adopted via.
- * An inverse map from TxnId to Key may also be constructed and stored in this collection.
+ * A collection of transaction dependencies, keyed by the key or range on which they were adopted
*/
-// TODO (desired, consider): switch to RoutingKey? Would mean adopting execution dependencies less precisely, but saving ser/deser of large keys
-public class Deps implements Iterable<Map.Entry<Key, TxnId>>
+public class Deps
{
- private static final boolean DEBUG_CHECKS = true;
-
- private static final TxnId[] NO_TXNIDS = new TxnId[0];
- private static final int[] NO_INTS = new int[0];
- public static final Deps NONE = new Deps(Keys.EMPTY, NO_TXNIDS, NO_INTS);
+ public static final Deps NONE = new Deps(KeyDeps.NONE, RangeDeps.NONE);
- public static class SerializerSupport
+ public static Builder builder()
{
- private SerializerSupport() {}
-
- public static int keyToTxnIdCount(Deps deps)
- {
- return deps.keyToTxnId.length;
- }
-
- public static int keyToTxnId(Deps deps, int idx)
- {
- return deps.keyToTxnId[idx];
- }
-
- public static Deps create(Keys keys, TxnId[] txnIds, int[] keyToTxnId)
- {
- return new Deps(keys, txnIds, keyToTxnId);
- }
- }
-
- public static Deps none(Keys keys)
- {
- int[] keysToTxnId = new int[keys.size()];
- Arrays.fill(keysToTxnId, keys.size());
- return new Deps(keys, NO_TXNIDS, keysToTxnId);
- }
-
- /**
- * Expects Command to be provided in TxnId order
- */
- public static OrderedBuilder orderedBuilder(boolean hasOrderedTxnId)
- {
- return new OrderedBuilder(hasOrderedTxnId);
+ return new Builder();
}
// TODO (expected, efficiency): cache this object per thread
- public static abstract class AbstractOrderedBuilder<T extends Deps> implements AutoCloseable
+ public static abstract class AbstractBuilder<T extends Deps> implements AutoCloseable
{
- final ObjectBuffers<TxnId> cachedTxnIds = cachedTxnIds();
- final ObjectBuffers<Key> cachedKeys = cachedKeys();
- final IntBuffers cachedInts = cachedInts();
-
- final boolean hasOrderedTxnId;
- Key[] keys;
- int[] keyLimits;
- // txnId -> Offset
- TxnId[] keyToTxnId;
- int keyCount;
- int keyOffset;
- int totalCount;
-
- public AbstractOrderedBuilder(boolean hasOrderedTxnId)
- {
- this.keys = cachedKeys.get(16);
- this.keyLimits = cachedInts.getInts(keys.length);
- this.hasOrderedTxnId = hasOrderedTxnId;
- this.keyToTxnId = cachedTxnIds.get(16);
- }
-
- public boolean isEmpty()
- {
- return totalCount() == 0;
- }
-
- private int totalCount()
- {
- return totalCount;
- }
+ final KeyDeps.Builder keyBuilder;
+ RangeDeps.Builder rangeBuilder;
- public void nextKey(Key key)
+ AbstractBuilder()
{
- if (keyCount > 0 && keys[keyCount - 1].compareTo(key) >= 0)
- {
- throw new IllegalArgumentException("Key " + key + " has already been visited or was provided out of order ("
- + Arrays.toString(Arrays.copyOf(keys, keyCount)) + ")");
- }
-
- finishKey();
-
- if (keyCount == keys.length)
- {
- Key[] newKeys = cachedKeys.get(keyCount * 2);
- System.arraycopy(keys, 0, newKeys, 0, keyCount);
- cachedKeys.forceDiscard(keys, keyCount);
- keys = newKeys;
- int[] newKeyLimits = cachedInts.getInts(keyCount * 2);
- System.arraycopy(keyLimits, 0, newKeyLimits, 0, keyCount);
- cachedInts.forceDiscard(keyLimits, keyCount);
- keyLimits = newKeyLimits;
- }
- keys[keyCount++] = key;
+ this.keyBuilder = KeyDeps.builder();
}
- private void finishKey()
+ public AbstractBuilder<T> add(Seekable keyOrRange, TxnId txnId)
{
- if (totalCount == keyOffset && keyCount > 0)
- --keyCount; // remove this key; no data
-
- if (keyCount == 0)
- return;
-
- if (totalCount != keyOffset && !hasOrderedTxnId)
+ switch (keyOrRange.domain())
{
- // TODO (low priority, efficiency): this allocates a significant amount of memory: would be preferable to be able to sort using a pre-defined scratch buffer
- Arrays.sort(keyToTxnId, keyOffset, totalCount);
- for (int i = keyOffset + 1 ; i < totalCount ; ++i)
- {
- if (keyToTxnId[i - 1].equals(keyToTxnId[i]))
- throw new IllegalArgumentException("TxnId for " + keys[keyCount - 1] + " are not unique: " + Arrays.asList(keyToTxnId).subList(keyOffset, totalCount));
- }
+ default: throw new AssertionError();
+ case Key:
+ keyBuilder.add(keyOrRange.asKey(), txnId);
+ break;
+ case Range:
+ if (rangeBuilder == null)
+ rangeBuilder = RangeDeps.builder();
+ rangeBuilder.add(keyOrRange.asRange(), txnId);
+ break;
}
-
- keyLimits[keyCount - 1] = totalCount;
- keyOffset = totalCount;
- }
-
- public void add(Key key, TxnId txnId)
- {
- if (keyCount == 0 || !keys[keyCount - 1].equals(key))
- nextKey(key);
- add(txnId);
- }
-
- /**
- * Add this command as a dependency for each intersecting key
- */
- public void add(TxnId txnId)
- {
- if (hasOrderedTxnId && totalCount > keyOffset && keyToTxnId[totalCount - 1].compareTo(txnId) >= 0)
- throw new IllegalArgumentException("TxnId provided out of order");
-
- if (totalCount >= keyToTxnId.length)
- {
- TxnId[] newTxnIds = cachedTxnIds.get(keyToTxnId.length * 2);
- System.arraycopy(keyToTxnId, 0, newTxnIds, 0, totalCount);
- cachedTxnIds.forceDiscard(keyToTxnId, totalCount);
- keyToTxnId = newTxnIds;
- }
-
- keyToTxnId[totalCount++] = txnId;
- }
-
- public T build()
- {
- if (totalCount == 0)
- return build(Keys.EMPTY, NO_TXNIDS, NO_INTS);
-
- finishKey();
-
- TxnId[] uniqueTxnId = cachedTxnIds.get(totalCount);
- System.arraycopy(keyToTxnId, 0, uniqueTxnId, 0, totalCount);
- Arrays.sort(uniqueTxnId, 0, totalCount);
- int txnIdCount = 1;
- for (int i = 1 ; i < totalCount ; ++i)
- {
- if (!uniqueTxnId[txnIdCount - 1].equals(uniqueTxnId[i]))
- uniqueTxnId[txnIdCount++] = uniqueTxnId[i];
- }
-
- TxnId[] txnIds = cachedTxnIds.complete(uniqueTxnId, txnIdCount);
- cachedTxnIds.discard(uniqueTxnId, totalCount);
-
- int[] result = new int[keyCount + totalCount];
- int offset = keyCount;
- for (int k = 0 ; k < keyCount ; ++k)
- {
- result[k] = keyCount + keyLimits[k];
- int from = k == 0 ? 0 : keyLimits[k - 1];
- int to = keyLimits[k];
- offset = (int)SortedArrays.foldlIntersection(txnIds, 0, txnIdCount, keyToTxnId, from, to, (key, p, v, li, ri) -> {
- result[(int)v] = li;
- return v + 1;
- }, keyCount, offset, -1);
- }
-
- return build(Keys.ofSortedUnchecked(cachedKeys.complete(keys, keyCount)), txnIds, result);
+ return this;
}
- abstract T build(Keys keys, TxnId[] txnIds, int[] keyToTxnId);
+ public abstract T build();
@Override
public void close()
{
- cachedKeys.discard(keys, keyCount);
- cachedInts.forceDiscard(keyLimits, keyCount);
- cachedTxnIds.forceDiscard(keyToTxnId, totalCount);
- }
- }
-
- public static class OrderedBuilder extends AbstractOrderedBuilder<Deps>
- {
- public OrderedBuilder(boolean hasOrderedTxnId)
- {
- super(hasOrderedTxnId);
- }
-
- @Override
- Deps build(Keys keys, TxnId[] txnIds, int[] keysToTxnIds)
- {
- return new Deps(keys, txnIds, keysToTxnIds);
+ keyBuilder.close();
+ if (rangeBuilder != null)
+ rangeBuilder.close();
}
}
- /**
- * An object for managing a sequence of efficient linear merges Deps objects.
- * Its primary purpose is to manage input and output buffers, so that we reuse output buffers
- * as input to the next merge, and if any input is a superset of the other inputs that this input
- * is returned unmodified.
- *
- * This is achieved by using PassThroughXBuffers so that the result buffers (and their sizes) are returned
- * unmodified, and the buffers are cached as far as possible. In general, the buffers should be taken
- * out of pre-existing caches, but if the buffers are too large then we cache any additional buffers we
- * allocate for the duration of the merge.
- */
- private static class LinearMerger extends PassThroughObjectAndIntBuffers<TxnId> implements DepsConstructor<Key, TxnId, Object>
+ public static class Builder extends AbstractBuilder<Deps>
{
- final PassThroughObjectBuffers<Key> keyBuffers;
- Key[] bufKeys;
- TxnId[] bufTxnIds;
- int[] buf = null;
- int bufKeysLength, bufTxnIdsLength = 0, bufLength = 0;
- Deps from = null;
-
- LinearMerger()
- {
- super(cachedTxnIds(), cachedInts());
- keyBuffers = new PassThroughObjectBuffers<>(cachedKeys());
- }
-
- @Override
- public Object construct(Key[] keys, int keysLength, TxnId[] txnIds, int txnIdsLength, int[] out, int outLength)
- {
- if (from == null)
- {
- // if our input buffers were themselves buffers, we want to discard them unless they have been returned back to us
- discard(keys, txnIds, out);
- }
- else if (buf != out)
- {
- // the output is not equal to a prior input
- from = null;
- }
-
- if (from == null)
- {
- bufKeys = keys;
- bufKeysLength = keysLength;
- bufTxnIds = txnIds;
- bufTxnIdsLength = txnIdsLength;
- buf = out;
- bufLength = outLength;
- }
- else
- {
- Invariants.checkState(keys == bufKeys && keysLength == bufKeysLength);
- Invariants.checkState(txnIds == bufTxnIds && txnIdsLength == bufTxnIdsLength);
- Invariants.checkState(outLength == bufLength);
- }
- return null;
- }
-
- void update(Deps deps)
+ public Builder()
{
- if (buf == null)
- {
- bufKeys = deps.keys.keys;
- bufKeysLength = deps.keys.keys.length;
- bufTxnIds = deps.txnIds;
- bufTxnIdsLength = deps.txnIds.length;
- buf = deps.keyToTxnId;
- bufLength = deps.keyToTxnId.length;
- from = deps;
- return;
- }
-
- linearUnion(
- bufKeys, bufKeysLength, bufTxnIds, bufTxnIdsLength, buf, bufLength,
- deps.keys.keys, deps.keys.keys.length, deps.txnIds, deps.txnIds.length, deps.keyToTxnId, deps.keyToTxnId.length,
- keyBuffers, this, this, this
- );
- if (buf == deps.keyToTxnId)
- {
- Invariants.checkState(deps.keys.keys == bufKeys && deps.keys.keys.length == bufKeysLength);
- Invariants.checkState(deps.txnIds == bufTxnIds && deps.txnIds.length == bufTxnIdsLength);
- Invariants.checkState(deps.keyToTxnId.length == bufLength);
- from = deps;
- }
+ super();
}
- Deps get()
+ public Deps build()
{
- if (buf == null)
- return NONE;
-
- if (from != null)
- return from;
-
- return new Deps(
- Keys.ofSortedUnchecked(keyBuffers.realComplete(bufKeys, bufKeysLength)),
- realComplete(bufTxnIds, bufTxnIdsLength),
- realComplete(buf, bufLength));
- }
-
- /**
- * Free any buffers we no longer need
- */
- void discard()
- {
- if (from == null)
- discard(null, null, null);
- }
-
- /**
- * Free buffers unless they are equal to the corresponding parameter
- */
- void discard(Key[] freeKeysIfNot, TxnId[] freeTxnIdsIfNot, int[] freeBufIfNot)
- {
- if (from != null)
- return;
-
- if (bufKeys != freeKeysIfNot)
- {
- keyBuffers.realDiscard(bufKeys, bufKeysLength);
- bufKeys = null;
- }
- if (bufTxnIds != freeTxnIdsIfNot)
- {
- realDiscard(bufTxnIds, bufTxnIdsLength);
- bufTxnIds = null;
- }
- if (buf != freeBufIfNot)
- {
- realDiscard(buf, bufLength);
- buf = null;
- }
+ return new Deps(keyBuilder.build(), rangeBuilder == null ? RangeDeps.NONE : rangeBuilder.build());
}
}
- public static <T> Deps merge(List<T> merge, Function<T, Deps> getter)
- {
- LinearMerger linearMerger = new LinearMerger();
- try
- {
- int mergeIndex = 0, mergeSize = merge.size();
- while (mergeIndex < mergeSize)
- {
- Deps deps = getter.apply(merge.get(mergeIndex++));
- if (deps == null || deps.isEmpty())
- continue;
-
- linearMerger.update(deps);
- }
-
- return linearMerger.get();
- }
- finally
- {
- linearMerger.discard();
- }
- }
-
- final Keys keys; // unique Keys
- final TxnId[] txnIds; // unique TxnId TODO (low priority, efficiency): this could be a BTree?
+ public final KeyDeps keyDeps;
+ public final RangeDeps rangeDeps;
- /**
- * This represents a map of {@code Key -> [TxnId] } where each TxnId is actually a pointer into the txnIds array.
- * The beginning of the array (the first keys.size() entries) are offsets into this array.
- * <p/>
- * Example:
- * <p/>
- * {@code
- * int keyIdx = keys.indexOf(key);
- * int startOfTxnOffset = keyIdx == 0 ? keys.size() : keyToTxnId[keyIdx - 1];
- * int endOfTxnOffset = keyToTxnId[keyIdx];
- * for (int i = startOfTxnOffset; i < endOfTxnOffset; i++)
- * {
- * TxnId id = txnIds[keyToTxnId[i]]
- * ...
- * }
- * }
- */
- final int[] keyToTxnId; // Key -> [TxnId]
- // Lazy loaded in ensureTxnIdToKey()
- int[] txnIdToKey; // TxnId -> [Key]
-
- Deps(Keys keys, TxnId[] txnIds, int[] keyToTxnId)
+ public Deps(KeyDeps keyDeps, RangeDeps rangeDeps)
{
- this.keys = keys;
- this.txnIds = txnIds;
- this.keyToTxnId = keyToTxnId;
- if (!(keys.isEmpty() || keyToTxnId[keys.size() - 1] == keyToTxnId.length))
- throw new IllegalArgumentException(String.format("Last key (%s) in keyToTxnId does not point (%d) to the end of the array (%d);\nkeyToTxnId=%s", keys.get(keys.size() - 1), keyToTxnId[keys.size() - 1], keyToTxnId.length, Arrays.toString(keyToTxnId)));
- if (DEBUG_CHECKS)
- checkValid();
+ this.keyDeps = keyDeps;
+ this.rangeDeps = rangeDeps;
}
- public PartialDeps slice(Ranges ranges)
- {
- if (isEmpty())
- return new PartialDeps(ranges, keys, txnIds, keyToTxnId);
-
- Keys select = keys.slice(ranges);
-
- if (select.isEmpty())
- return new PartialDeps(ranges, Keys.EMPTY, NO_TXNIDS, NO_INTS);
-
- if (select.size() == keys.size())
- return new PartialDeps(ranges, keys, txnIds, keyToTxnId);
-
- int i = 0;
- int offset = select.size();
- for (int j = 0 ; j < select.size() ; ++j)
- {
- int findi = keys.findNext(i, select.get(j), FAST);
- if (findi < 0)
- continue;
-
- i = findi;
- offset += keyToTxnId[i] - (i == 0 ? keys.size() : keyToTxnId[i - 1]);
- }
-
- int[] src = keyToTxnId;
- int[] trg = new int[offset];
-
- i = 0;
- offset = select.size();
- for (int j = 0 ; j < select.size() ; ++j)
- {
- int findi = keys.findNext(i, select.get(j), FAST);
- if (findi >= 0)
- {
- i = findi;
- int start = i == 0 ? keys.size() : src[i - 1];
- int count = src[i] - start;
- System.arraycopy(src, start, trg, offset, count);
- offset += count;
- }
- trg[j] = offset;
- }
-
- TxnId[] txnIds = trimUnusedTxnId(select, this.txnIds, trg);
- return new PartialDeps(ranges, select, txnIds, trg);
- }
-
- /**
- * Returns the set of {@link TxnId}s that are referenced by {@code keysToTxnId}, and <strong>updates</strong>
- * {@code keysToTxnId} to point to the new offsets in the returned set.
- * @param keys object referenced by {@code keysToTxnId} index
- * @param txnIds to trim to the seen {@link TxnId}s
- * @param keysToTxnId to use as reference for trimming, this index will be updated to reflect the trimmed offsets.
- * @return smallest set of {@link TxnId} seen in {@code keysToTxnId}
- */
- private static TxnId[] trimUnusedTxnId(Keys keys, TxnId[] txnIds, int[] keysToTxnId)
+ public boolean contains(TxnId txnId)
{
- IntBuffers cache = ArrayBuffers.cachedInts();
- // we use remapTxnId twice:
- // - first we use the end to store a bitmap of those TxnId we are actually using
- // - then we use it to store the remap index (incrementally replacing the bitmap)
- int bitMapOffset = txnIds.length + 1 - (txnIds.length+31)/32;
- int[] remapTxnId = cache.getInts(txnIds.length + 1);
- try
- {
- Arrays.fill(remapTxnId, bitMapOffset, txnIds.length + 1, 0);
- for (int i = keys.size() ; i < keysToTxnId.length ; ++i)
- setBit(remapTxnId, bitMapOffset, keysToTxnId[i]);
-
- int offset = 0;
- for (int i = 0 ; i < txnIds.length ; ++i)
- {
- if (hasSetBit(remapTxnId, bitMapOffset, i)) remapTxnId[i] = offset++;
- else remapTxnId[i] = -1;
- }
-
- TxnId[] result = txnIds;
- if (offset < txnIds.length)
- {
- result = new TxnId[offset];
- for (int i = 0 ; i < txnIds.length ; ++i)
- {
- if (remapTxnId[i] >= 0)
- result[remapTxnId[i]] = txnIds[i];
- }
- // Update keysToTxnId to point to the new remapped TxnId offsets
- for (int i = keys.size() ; i < keysToTxnId.length ; ++i)
- keysToTxnId[i] = remapTxnId[keysToTxnId[i]];
- }
-
- return result;
- }
- finally
- {
- cache.forceDiscard(remapTxnId, txnIds.length);
- }
+ return keyDeps.contains(txnId) || rangeDeps.contains(txnId);
}
public Deps with(Deps that)
{
- if (isEmpty() || that.isEmpty())
- return isEmpty() ? that : this;
-
- return linearUnion(
- this.keys.keys, this.keys.keys.length, this.txnIds, this.txnIds.length, this.keyToTxnId, this.keyToTxnId.length,
- that.keys.keys, that.keys.keys.length, that.txnIds, that.txnIds.length, that.keyToTxnId, that.keyToTxnId.length,
- cachedKeys(), cachedTxnIds(), cachedInts(),
- (keys, keysLength, txnIds, txnIdsLength, out, outLength) ->
- new Deps(Keys.ofSortedUnchecked(cachedKeys().complete(keys, keysLength)),
- cachedTxnIds().complete(txnIds, txnIdsLength),
- cachedInts().complete(out, outLength))
- );
- }
-
- /**
- * Turn a set of key, value and mapping buffers into a merge result;
- * K and V are either Key and TxnId, or vice versa, depending on which mapping direction was present
- */
- interface DepsConstructor<K, V, T>
- {
- T construct(K[] keys, int keysLength, V[] values, int valuesLength, int[] out, int outLength);
- }
-
- private static boolean arraysEqual(int[] left, int[] right, int length)
- {
- if (left.length < length || right.length < length)
- return false;
-
- for (int i=0; i<length; i++)
- if (left[i] !=right[i])
- return false;
-
- return true;
- }
-
- private static <T> boolean arraysEqual(T[] left, T[] right, int length)
- {
- if (left.length < length || right.length < length)
- return false;
-
- for (int i=0; i<length; i++)
- if (!Objects.equals(left[i], right[i]))
- return false;
-
- return true;
- }
-
- // TODO (low priority, efficiency): this method supports merging keyToTxnId OR txnIdToKey; we can perhaps save time
- // and effort when constructing Deps on remote hosts by only producing txnIdToKey with OrderedCollector and serializing
- // only this, and merging on the recipient before inverting, so that we only have to invert the final assembled deps
- private static <K extends Comparable<? super K>, V extends Comparable<? super V>, T>
- T linearUnion(K[] leftKeys, int leftKeysLength, V[] leftValues, int leftValuesLength, int[] left, int leftLength,
- K[] rightKeys, int rightKeysLength, V[] rightValues, int rightValuesLength, int[] right, int rightLength,
- ObjectBuffers<K> keyBuffers, ObjectBuffers<V> valueBuffers, IntBuffers intBuffers, DepsConstructor<K, V, T> constructor)
- {
- K[] outKeys = null;
- V[] outValues = null;
- int[] remapLeft = null, remapRight = null, out = null;
- int outLength = 0, outKeysLength = 0, outTxnIdsLength = 0;
-
- try
- {
- outKeys = SortedArrays.linearUnion(leftKeys, leftKeysLength, rightKeys, rightKeysLength, keyBuffers);
- outKeysLength = keyBuffers.lengthOfLast(outKeys);
- outValues = SortedArrays.linearUnion(leftValues, leftValuesLength, rightValues, rightValuesLength, valueBuffers);
- outTxnIdsLength = valueBuffers.lengthOfLast(outValues);
-
- remapLeft = remapToSuperset(leftValues, leftValuesLength, outValues, outTxnIdsLength, intBuffers);
- remapRight = remapToSuperset(rightValues, rightValuesLength, outValues, outTxnIdsLength, intBuffers);
-
- if (remapLeft == null && remapRight == null && leftLength == rightLength && leftKeysLength == rightKeysLength
- && arraysEqual(left, right, rightLength)
- && arraysEqual(leftKeys, rightKeys, rightKeysLength)
- )
- {
- return constructor.construct(leftKeys, leftKeysLength, leftValues, leftValuesLength, left, leftLength);
- }
-
- int lk = 0, rk = 0, ok = 0, l = leftKeysLength, r = rightKeysLength;
- outLength = outKeysLength;
-
- if (remapLeft == null && outKeys == leftKeys)
- {
- // "this" knows all the TxnId and Keys already, but do both agree on what Keys map to TxnIds?
- noOp: while (lk < leftKeysLength && rk < rightKeysLength)
- {
- int ck = leftKeys[lk].compareTo(rightKeys[rk]);
- if (ck < 0)
- {
- // "this" knows of a key not present in "that"
- outLength += left[lk] - l; // logically append the key's TxnIds to the size
- l = left[lk];
- assert outLength == l && ok == lk && left[ok] == outLength;
- ok++;
- lk++;
- }
- else if (ck > 0)
- {
- // if this happened there is a bug with keys.union or keys are not actually sorted
- throwUnexpectedMissingKeyException(leftKeys, lk, leftKeysLength, rightKeys, rk, rightKeysLength, true);
- }
- else
- {
- // both "this" and "that" know of the key
- while (l < left[lk] && r < right[rk])
- {
- int nextLeft = left[l];
- int nextRight = remap(right[r], remapRight);
-
- if (nextLeft < nextRight)
- {
- // "this" knows of the txn that "that" didn't
- outLength++;
- l++;
- }
- else if (nextRight < nextLeft)
- {
- out = copy(left, outLength, leftLength + rightLength - r, intBuffers);
- break noOp;
- }
- else
- {
- outLength++;
- l++;
- r++;
- }
- }
-
- if (l < left[lk])
- {
- outLength += left[lk] - l;
- l = left[lk];
- }
- else if (r < right[rk])
- {
- // "that" thinks a key includes a TxnId as a dependency but "this" doesn't, need to include this knowledge
- out = copy(left, outLength, leftLength + rightLength - r, intBuffers);
- break;
- }
-
- assert outLength == l && ok == lk && left[ok] == outLength;
- ok++;
- rk++;
- lk++;
- }
- }
-
- if (out == null)
- return constructor.construct(leftKeys, leftKeysLength, leftValues, leftValuesLength, left, leftLength);
- }
- else if (remapRight == null && outKeys == rightKeys)
- {
- // "that" knows all the TxnId and keys already, but "this" does not
- noOp: while (lk < leftKeysLength && rk < rightKeysLength)
- {
- int ck = leftKeys[lk].compareTo(rightKeys[rk]);
- if (ck < 0)
- {
- // if this happened there is a bug with keys.union or keys are not actually sorted
- throwUnexpectedMissingKeyException(leftKeys, lk, leftKeysLength, rightKeys, rk, rightKeysLength, false);
- }
- else if (ck > 0)
- {
- outLength += right[rk] - r;
- r = right[rk];
- assert outLength == r && ok == rk && right[ok] == outLength;
- ok++;
- rk++;
- }
- else
- {
- // both "this" and "that" know of the key
- while (l < left[lk] && r < right[rk])
- {
- int nextLeft = remap(left[l], remapLeft);
- int nextRight = right[r];
-
- if (nextLeft < nextRight)
- {
- // "this" thinks a TxnID depends on Key but "that" doesn't, need to include this knowledge
- out = copy(right, outLength, rightLength + leftLength - l, intBuffers);
- break noOp;
- }
- else if (nextRight < nextLeft)
- {
- // "that" knows of the txn that "this" didn't
- outLength++;
- r++;
- }
- else
- {
- outLength++;
- l++;
- r++;
- }
- }
-
- if (l < left[lk])
- {
- out = copy(right, outLength, rightLength + leftLength - l, intBuffers);
- break;
- }
- else if (r < right[rk])
- {
- outLength += right[rk] - r;
- r = right[rk];
- }
-
- assert outLength == r && ok == rk && right[ok] == outLength;
- ok++;
- rk++;
- lk++;
- }
- }
-
- if (out == null)
- return constructor.construct(rightKeys, rightKeysLength, rightValues, rightValuesLength, right, rightLength);
- }
- else
- {
- out = intBuffers.getInts(leftLength + rightLength);
- }
-
- while (lk < leftKeysLength && rk < rightKeysLength)
- {
- int ck = leftKeys[lk].compareTo(rightKeys[rk]);
- if (ck < 0)
- {
- while (l < left[lk])
- out[outLength++] = remap(left[l++], remapLeft);
- out[ok++] = outLength;
- lk++;
- }
- else if (ck > 0)
- {
- while (r < right[rk])
- out[outLength++] = remap(right[r++], remapRight);
- out[ok++] = outLength;
- rk++;
- }
- else
- {
- while (l < left[lk] && r < right[rk])
- {
- int nextLeft = remap(left[l], remapLeft);
- int nextRight = remap(right[r], remapRight);
-
- if (nextLeft <= nextRight)
- {
- out[outLength++] = nextLeft;
- l += 1;
- r += nextLeft == nextRight ? 1 : 0;
- }
- else
- {
- out[outLength++] = nextRight;
- ++r;
- }
- }
-
- while (l < left[lk])
- out[outLength++] = remap(left[l++], remapLeft);
-
- while (r < right[rk])
- out[outLength++] = remap(right[r++], remapRight);
-
- out[ok++] = outLength;
- rk++;
- lk++;
- }
- }
-
- while (lk < leftKeysLength)
- {
- while (l < left[lk])
- out[outLength++] = remap(left[l++], remapLeft);
- out[ok++] = outLength;
- lk++;
- }
-
- while (rk < rightKeysLength)
- {
- while (r < right[rk])
- out[outLength++] = remap(right[r++], remapRight);
- out[ok++] = outLength;
- rk++;
- }
-
- return constructor.construct(outKeys, outKeysLength, outValues, outTxnIdsLength, out, outLength);
- }
- finally
- {
- if (outKeys != null)
- keyBuffers.discard(outKeys, outKeysLength);
- if (outValues != null)
- valueBuffers.discard(outValues, outTxnIdsLength);
- if (out != null)
- intBuffers.discard(out, outLength);
- if (remapLeft != null)
- intBuffers.forceDiscard(remapLeft, leftValuesLength);
- if (remapRight != null)
- intBuffers.forceDiscard(remapRight, rightValuesLength);
- }
- }
-
- private static <A> void throwUnexpectedMissingKeyException(A[] leftKeys, int leftKeyIndex, int leftKeyLength, A[] rightKeys, int rightKeyIndex, int rightKeyLength, boolean isMissingLeft)
- {
- StringBuilder sb = new StringBuilder();
- String missing = isMissingLeft ? "left" : "right";
- String extra = isMissingLeft ? "right" : "left";
- sb.append(missing).append(" knows all keys, yet ").append(extra).append(" knew of an extra key at indexes left[")
- .append(leftKeyIndex).append("] = ").append(leftKeys[leftKeyIndex])
- .append(", right[").append(rightKeyIndex).append("] = ").append(rightKeys[rightKeyIndex]).append("\n");
- sb.append("leftKeys = ").append(Arrays.stream(leftKeys, 0, leftKeyLength).map(Object::toString).collect(Collectors.joining())).append('\n');
- sb.append("rightKeys = ").append(Arrays.stream(rightKeys, 0, rightKeyLength).map(Object::toString).collect(Collectors.joining())).append('\n');
- throw new IllegalStateException(sb.toString());
- }
-
- private static int[] copy(int[] src, int to, int length, IntBuffers bufferManager)
- {
- if (length == 0)
- return NO_INTS;
-
- int[] result = bufferManager.getInts(length);
- if (result.length < length)
- throw new IllegalStateException();
- System.arraycopy(src, 0, result, 0, to);
- return result;
+ return new Deps(this.keyDeps.with(that.keyDeps), this.rangeDeps.with(that.rangeDeps));
}
public Deps without(Predicate<TxnId> remove)
{
- if (isEmpty())
- return this;
-
- IntBuffers cache = ArrayBuffers.cachedInts();
- TxnId[] oldTxnIds = txnIds;
- int[] oldKeyToTxnId = keyToTxnId;
- int[] remapTxnIds = cache.getInts(oldTxnIds.length);
- int[] newKeyToTxnId = null;
- TxnId[] newTxnIds;
- int o = 0;
- try
- {
- int count = 0;
- for (int i = 0 ; i < oldTxnIds.length ; ++i)
- {
- if (remove.test(oldTxnIds[i])) remapTxnIds[i] = -1;
- else remapTxnIds[i] = count++;
- }
-
- if (count == oldTxnIds.length)
- return this;
-
- if (count == 0)
- return NONE;
-
- newTxnIds = new TxnId[count];
- for (int i = 0 ; i < oldTxnIds.length ; ++i)
- {
- if (remapTxnIds[i] >= 0)
- newTxnIds[remapTxnIds[i]] = oldTxnIds[i];
- }
-
- newKeyToTxnId = cache.getInts(oldKeyToTxnId.length);
- int k = 0, i = keys.size();
- o = i;
- while (i < oldKeyToTxnId.length)
- {
- while (oldKeyToTxnId[k] == i)
- newKeyToTxnId[k++] = o;
-
- int remapped = remapTxnIds[oldKeyToTxnId[i]];
- if (remapped >= 0)
- newKeyToTxnId[o++] = remapped;
- ++i;
- }
-
- while (k < keys.size())
- newKeyToTxnId[k++] = o;
- }
- catch (Throwable t)
- {
- cache.forceDiscard(newKeyToTxnId, o);
- throw t;
- }
- finally
- {
- cache.forceDiscard(remapTxnIds, oldTxnIds.length);
- }
-
- newKeyToTxnId = cache.completeAndDiscard(newKeyToTxnId, o);
- return new Deps(keys, newTxnIds, newKeyToTxnId);
+ return new Deps(keyDeps.without(remove), rangeDeps.without(remove));
}
- public boolean contains(TxnId txnId)
+ public PartialDeps slice(Ranges covering)
{
- return Arrays.binarySearch(txnIds, txnId) >= 0;
+ return new PartialDeps(covering, keyDeps.slice(covering), rangeDeps.slice(covering));
}
- // return true iff we map any keys to any txnId
- // if the mapping is empty we return false, whether or not we have any keys or txnId by themselves
public boolean isEmpty()
{
- return keyToTxnId.length == keys.size();
- }
-
- public Keys someKeys(TxnId txnId)
- {
- int txnIdIndex = Arrays.binarySearch(txnIds, txnId);
- if (txnIdIndex < 0)
- return Keys.EMPTY;
-
- ensureTxnIdToKey();
-
- int start = txnIdIndex == 0 ? txnIds.length : txnIdToKey[txnIdIndex - 1];
- int end = txnIdToKey[txnIdIndex];
- if (start == end)
- return Keys.EMPTY;
-
- Key[] result = new Key[end - start];
- for (int i = start ; i < end ; ++i)
- result[i - start] = keys.get(txnIdToKey[i]);
- return Keys.of(result);
- }
-
- public Unseekables<RoutingKey, ?> someRoutables(TxnId txnId)
- {
- return toUnseekables(txnId, array -> {
- if (array.length == 0)
- throw new IllegalStateException("Cannot create a RouteFragment without any keys");
- return new RoutingKeys(array);
- });
- }
-
- private <R> R toUnseekables(TxnId txnId, Function<RoutingKey[], R> constructor)
- {
- int txnIdIndex = Arrays.binarySearch(txnIds, txnId);
- if (txnIdIndex < 0)
- constructor.apply(RoutingKeys.EMPTY.keys);
-
- ensureTxnIdToKey();
-
- int start = txnIdIndex == 0 ? txnIds.length : txnIdToKey[txnIdIndex - 1];
- int end = txnIdToKey[txnIdIndex];
- RoutingKey[] result = new RoutingKey[end - start];
- if (start == end)
- constructor.apply(RoutingKeys.EMPTY.keys);
-
- result[0] = keys.get(txnIdToKey[start]).toUnseekable();
- int resultCount = 1;
- for (int i = start + 1 ; i < end ; ++i)
- {
- RoutingKey next = keys.get(txnIdToKey[i]).toUnseekable();
- if (!next.equals(result[resultCount - 1]))
- result[resultCount++] = next;
- }
-
- if (resultCount < result.length)
- result = Arrays.copyOf(result, resultCount);
- return constructor.apply(result);
- }
-
- void ensureTxnIdToKey()
- {
- if (txnIdToKey != null)
- return;
-
- txnIdToKey = invert(keyToTxnId, keyToTxnId.length, keys.size(), txnIds.length);
- }
-
- private static int[] invert(int[] src, int srcLength, int srcKeyCount, int trgKeyCount)
- {
- int[] trg = new int[trgKeyCount + srcLength - srcKeyCount];
-
- // first pass, count number of txnId per key
- for (int i = srcKeyCount ; i < srcLength ; ++i)
- trg[src[i]]++;
-
- // turn into offsets (i.e. add txnIds.size() and then sum them)
- trg[0] += trgKeyCount;
- for (int i = 1; i < trgKeyCount ; ++i)
- trg[i] += trg[i - 1];
-
- // shuffle forwards one, so we have the start index rather than end
- System.arraycopy(trg, 0, trg, 1, trgKeyCount - 1);
- trg[0] = trgKeyCount;
-
- // convert the offsets to end, and set the key at the target positions
- int k = 0;
- for (int i = srcKeyCount ; i < srcLength ; ++i)
- {
- // if at the end offset, switch to the next key
- while (i == src[k])
- ++k;
-
- // find the next key offset for the TxnId and set the offset to this key
- trg[trg[src[i]]++] = k;
- }
-
- return trg;
- }
-
- public void forEachOn(Ranges ranges, Predicate<Key> include, BiConsumer<Key, TxnId> forEach)
- {
- Routables.foldl(keys, ranges, (key, value, index) -> {
- if (!include.test(key))
- return null;
-
- for (int t = startOffset(index), end = endOffset(index); t < end ; ++t)
- {
- TxnId txnId = txnIds[keyToTxnId[t]];
- forEach.accept(key, txnId);
- }
- return null;
- }, null);
- }
-
- /**
- * For each {@link TxnId} that references a key within the {@link Ranges}; the {@link TxnId} will be seen exactly once.
- * @param ranges to match on
- * @param forEach function to call on each unique {@link TxnId}
- */
- public void forEachOn(Ranges ranges, Consumer<TxnId> forEach)
- {
- // Find all keys within the ranges, but record existence within an int64 bitset. Since the bitset is limited
- // to 64, this search must be called multiple times searching for different TxnIds in txnIds; this also has
- // the property that forEach is called in TxnId order.
- //TODO (expected, efficiency): reconsider this, probably not worth trying to save allocations at cost of multiple loop
- // use BitSet, or perhaps extend so we can have no nested allocations when few bits
- for (int offset = 0 ; offset < txnIds.length ; offset += 64)
- {
- long bitset = Routables.foldl(keys, ranges, (key, off, value, keyIndex) -> {
- int index = startOffset(keyIndex);
- int end = endOffset(keyIndex);
- if (off > 0)
- {
- // TODO (low priority, efficiency): interpolation search probably great here
- index = Arrays.binarySearch(keyToTxnId, index, end, (int)off);
- if (index < 0)
- index = -1 - index;
- }
-
- while (index < end)
- {
- long next = keyToTxnId[index++] - off;
- if (next >= 64)
- break;
- value |= 1L << next;
- }
-
- return value;
- }, offset, 0, -1L);
-
- while (bitset != 0)
- {
- int i = Long.numberOfTrailingZeros(bitset);
- TxnId txnId = txnIds[offset + i];
- forEach.accept(txnId);
- bitset ^= Long.lowestOneBit(bitset);
- }
- }
- }
-
- public void forEach(Key key, Consumer<TxnId> forEach)
- {
- int keyIndex = keys.indexOf(key);
- if (keyIndex < 0)
- return;
-
- int index = startOffset(keyIndex);
- int end = endOffset(keyIndex);
- while (index < end)
- forEach.accept(txnIds[keyToTxnId[index++]]);
- }
-
- public Keys keys()
- {
- return keys;
+ return keyDeps.isEmpty() && rangeDeps.isEmpty();
}
public int txnIdCount()
{
- return txnIds.length;
- }
-
- public int totalCount()
- {
- return keyToTxnId.length - keys.size();
+ return keyDeps.txnIdCount() + rangeDeps.txnIdCount();
}
public TxnId txnId(int i)
{
- return txnIds[i];
+ return i < keyDeps.txnIdCount()
+ ? keyDeps.txnId(i)
+ : rangeDeps.txnId(i - keyDeps.txnIdCount());
}
- public Collection<TxnId> txnIds()
+ public List<TxnId> txnIds()
{
- return Arrays.asList(txnIds);
- }
-
- public List<TxnId> txnIds(Key key)
- {
- int keyIndex = keys.indexOf(key);
- if (keyIndex < 0)
- return Collections.emptyList();
-
- int start = startOffset(keyIndex);
- int end = endOffset(keyIndex);
- int size = end - start;
-
- return new AbstractList<TxnId>()
- {
+ final int txnIdCount = txnIdCount();
+ final int keyDepsCount = keyDeps.txnIdCount();
+ return new AbstractList<TxnId>() {
@Override
public TxnId get(int index)
{
- if (index > end)
- throw new IndexOutOfBoundsException();
- return txnIds[keyToTxnId[start + index]];
+ return index < keyDepsCount
+ ? keyDeps.txnId(index)
+ : rangeDeps.txnId(index - keyDepsCount);
}
@Override
- public int size()
- {
- return size;
+ public int size() {
+ return txnIdCount;
}
};
}
- private int startOffset(int keyIndex)
- {
- return keyIndex == 0 ? keys.size() : keyToTxnId[keyIndex - 1];
- }
-
- private int endOffset(int keyIndex)
+ public List<TxnId> txnIds(Seekable keyOrRange)
{
- return keyToTxnId[keyIndex];
- }
-
- @Override
- public Iterator<Map.Entry<Key, TxnId>> iterator()
- {
- return new Iterator<Map.Entry<Key, TxnId>>()
+ List<TxnId> keyIds, rangeIds;
+ switch (keyOrRange.domain())
{
- int i = keys.size(), k = 0;
-
- @Override
- public boolean hasNext()
+ default: throw new AssertionError();
+ case Key:
{
- return i < keyToTxnId.length;
+ Key key = keyOrRange.asKey();
+ keyIds = keyDeps.txnIds(key);
+ rangeIds = rangeDeps.txnIds(key);
+ break;
}
-
- @Override
- public Map.Entry<Key, TxnId> next()
+ case Range:
{
- Entry result = new Entry(keys.get(k), txnIds[keyToTxnId[i++]]);
- if (i == keyToTxnId[k])
- ++k;
- return result;
+ Range range = keyOrRange.asRange();
+ keyIds = keyDeps.txnIds(range);
+ rangeIds = rangeDeps.txnIds(range);
}
- };
- }
-
- @Override
- public String toString()
- {
- return toSimpleString();
- }
+ }
- public String toSimpleString()
- {
- if (keys.isEmpty())
- return "{}";
+ if (rangeIds.isEmpty()) return keyIds;
+ if (keyIds.isEmpty()) return rangeIds;
- StringBuilder builder = new StringBuilder("{");
- for (int k = 0, t = keys.size(); k < keys.size() ; ++k)
+ List<TxnId> output = new ArrayList<>();
+ int ki = 0, ri = 0;
+ while (ki < keyIds.size() && ri < rangeIds.size())
{
- if (builder.length() > 1)
- builder.append(", ");
-
- builder.append(keys.get(k));
- builder.append(":[");
- boolean first = true;
- while (t < keyToTxnId[k])
- {
- if (first) first = false;
- else builder.append(", ");
- builder.append(txnIds[keyToTxnId[t++]]);
- }
- builder.append("]");
+ int c = keyIds.get(ki).compareTo(rangeIds.get(ri));
+ Invariants.checkState(c != 0);
+ if (c < 0) output.add(keyIds.get(ki++));
+ else output.add(rangeIds.get(ri++));
}
- builder.append("}");
- return builder.toString();
+ while (ki < keyIds.size())
+ output.add(keyIds.get(ki++));
+ while (ri < rangeIds.size())
+ output.add(rangeIds.get(ri++));
+ return output;
}
- @Override
- public boolean equals(Object o)
+ public Unseekables<?, ?> someUnseekables(TxnId txnId)
{
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- return equals((Deps) o);
+ if (keyDeps.contains(txnId))
+ return keyDeps.someUnseekables(txnId);
+ else
+ return rangeDeps.someUnseekables(txnId);
}
- public boolean equals(Deps that)
+ // NOTE: filter only applied to keyDeps
+ public void forEach(Ranges ranges, Consumer<TxnId> forEach)
{
- return this.txnIds.length == that.txnIds.length
- && this.keys.size() == that.keys.size()
- && Arrays.equals(this.keyToTxnId, that.keyToTxnId)
- && Arrays.equals(this.txnIds, that.txnIds)
- && this.keys.equals(that.keys);
+ keyDeps.forEachUniqueTxnId(ranges, forEach);
+ rangeDeps.forEachUniqueTxnId(ranges, forEach);
}
- public static class Entry implements Map.Entry<Key, TxnId>
+ public static <T> Deps merge(List<T> list, Function<T, Deps> getter)
{
- final Key key;
- final TxnId txnId;
-
- public Entry(Key key, TxnId txnId)
- {
- this.key = key;
- this.txnId = txnId;
- }
-
- @Override
- public Key getKey()
- {
- return key;
- }
-
- @Override
- public TxnId getValue()
- {
- return txnId;
- }
-
- @Override
- public TxnId setValue(TxnId value)
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public String toString()
- {
- return key + "->" + txnId;
- }
+ return new Deps(KeyDeps.merge(list, getter, deps -> deps.keyDeps),
+ RangeDeps.merge(list, getter, deps -> deps.rangeDeps));
}
- private void checkValid()
+ @Override
+ public String toString()
{
- int k = 0;
- for (int i = keys.size() ; i < keyToTxnId.length ; ++i)
- {
- boolean first = true;
- while (i < keyToTxnId[k])
- {
- if (first) first = false;
- else if (keyToTxnId[i - 1] == keyToTxnId[i])
- {
- Key key = keys.get(i);
- TxnId txnId = txnIds[keyToTxnId[i]];
- throw new IllegalStateException(String.format("Duplicate TxnId (%s) found for key %s", txnId, key));
- }
- i++;
- }
- ++k;
- }
+ return keyDeps.toString() + ", " + rangeDeps.toString();
}
- private static void setBit(int[] array, int offset, int index)
+ public boolean equals(Object that)
{
- array[offset + index / 32] |= (1 << (index & 31));
+ return this == that || (that instanceof Deps && equals((Deps)that));
}
- private static boolean hasSetBit(int[] array, int offset, int index)
+ public boolean equals(Deps that)
{
- return (array[offset + index / 32] & (1 << (index & 31))) != 0;
+ return this.keyDeps.equals(that.keyDeps) && this.rangeDeps.equals(that.rangeDeps);
Review Comment:
Add hash code throwing unsupported operation exception or the actual hash code?
##########
accord-core/src/main/java/accord/primitives/Deps.java:
##########
@@ -18,1311 +18,220 @@
package accord.primitives;
-import java.util.*;
-import java.util.function.BiConsumer;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Predicate;
-
-import java.util.stream.Collectors;
-
import accord.api.Key;
-import accord.utils.ArrayBuffers;
-import accord.api.RoutingKey;
-import accord.utils.SortedArrays;
import accord.utils.Invariants;
-import static accord.utils.ArrayBuffers.*;
-import static accord.utils.SortedArrays.*;
-import static accord.utils.SortedArrays.Search.FAST;
+import java.util.AbstractList;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
/**
- * A collection of dependencies for a transaction, organised by the key the dependency is adopted via.
- * An inverse map from TxnId to Key may also be constructed and stored in this collection.
+ * A collection of transaction dependencies, keyed by the key or range on which they were adopted
*/
-// TODO (desired, consider): switch to RoutingKey? Would mean adopting execution dependencies less precisely, but saving ser/deser of large keys
-public class Deps implements Iterable<Map.Entry<Key, TxnId>>
+public class Deps
{
- private static final boolean DEBUG_CHECKS = true;
-
- private static final TxnId[] NO_TXNIDS = new TxnId[0];
- private static final int[] NO_INTS = new int[0];
- public static final Deps NONE = new Deps(Keys.EMPTY, NO_TXNIDS, NO_INTS);
+ public static final Deps NONE = new Deps(KeyDeps.NONE, RangeDeps.NONE);
- public static class SerializerSupport
+ public static Builder builder()
{
- private SerializerSupport() {}
-
- public static int keyToTxnIdCount(Deps deps)
- {
- return deps.keyToTxnId.length;
- }
-
- public static int keyToTxnId(Deps deps, int idx)
- {
- return deps.keyToTxnId[idx];
- }
-
- public static Deps create(Keys keys, TxnId[] txnIds, int[] keyToTxnId)
- {
- return new Deps(keys, txnIds, keyToTxnId);
- }
- }
-
- public static Deps none(Keys keys)
- {
- int[] keysToTxnId = new int[keys.size()];
- Arrays.fill(keysToTxnId, keys.size());
- return new Deps(keys, NO_TXNIDS, keysToTxnId);
- }
-
- /**
- * Expects Command to be provided in TxnId order
- */
- public static OrderedBuilder orderedBuilder(boolean hasOrderedTxnId)
- {
- return new OrderedBuilder(hasOrderedTxnId);
+ return new Builder();
}
// TODO (expected, efficiency): cache this object per thread
- public static abstract class AbstractOrderedBuilder<T extends Deps> implements AutoCloseable
+ public static abstract class AbstractBuilder<T extends Deps> implements AutoCloseable
{
- final ObjectBuffers<TxnId> cachedTxnIds = cachedTxnIds();
- final ObjectBuffers<Key> cachedKeys = cachedKeys();
- final IntBuffers cachedInts = cachedInts();
-
- final boolean hasOrderedTxnId;
- Key[] keys;
- int[] keyLimits;
- // txnId -> Offset
- TxnId[] keyToTxnId;
- int keyCount;
- int keyOffset;
- int totalCount;
-
- public AbstractOrderedBuilder(boolean hasOrderedTxnId)
- {
- this.keys = cachedKeys.get(16);
- this.keyLimits = cachedInts.getInts(keys.length);
- this.hasOrderedTxnId = hasOrderedTxnId;
- this.keyToTxnId = cachedTxnIds.get(16);
- }
-
- public boolean isEmpty()
- {
- return totalCount() == 0;
- }
-
- private int totalCount()
- {
- return totalCount;
- }
+ final KeyDeps.Builder keyBuilder;
+ RangeDeps.Builder rangeBuilder;
- public void nextKey(Key key)
+ AbstractBuilder()
{
- if (keyCount > 0 && keys[keyCount - 1].compareTo(key) >= 0)
- {
- throw new IllegalArgumentException("Key " + key + " has already been visited or was provided out of order ("
- + Arrays.toString(Arrays.copyOf(keys, keyCount)) + ")");
- }
-
- finishKey();
-
- if (keyCount == keys.length)
- {
- Key[] newKeys = cachedKeys.get(keyCount * 2);
- System.arraycopy(keys, 0, newKeys, 0, keyCount);
- cachedKeys.forceDiscard(keys, keyCount);
- keys = newKeys;
- int[] newKeyLimits = cachedInts.getInts(keyCount * 2);
- System.arraycopy(keyLimits, 0, newKeyLimits, 0, keyCount);
- cachedInts.forceDiscard(keyLimits, keyCount);
- keyLimits = newKeyLimits;
- }
- keys[keyCount++] = key;
+ this.keyBuilder = KeyDeps.builder();
}
- private void finishKey()
+ public AbstractBuilder<T> add(Seekable keyOrRange, TxnId txnId)
{
- if (totalCount == keyOffset && keyCount > 0)
- --keyCount; // remove this key; no data
-
- if (keyCount == 0)
- return;
-
- if (totalCount != keyOffset && !hasOrderedTxnId)
+ switch (keyOrRange.domain())
{
- // TODO (low priority, efficiency): this allocates a significant amount of memory: would be preferable to be able to sort using a pre-defined scratch buffer
- Arrays.sort(keyToTxnId, keyOffset, totalCount);
- for (int i = keyOffset + 1 ; i < totalCount ; ++i)
- {
- if (keyToTxnId[i - 1].equals(keyToTxnId[i]))
- throw new IllegalArgumentException("TxnId for " + keys[keyCount - 1] + " are not unique: " + Arrays.asList(keyToTxnId).subList(keyOffset, totalCount));
- }
+ default: throw new AssertionError();
+ case Key:
+ keyBuilder.add(keyOrRange.asKey(), txnId);
+ break;
+ case Range:
+ if (rangeBuilder == null)
+ rangeBuilder = RangeDeps.builder();
+ rangeBuilder.add(keyOrRange.asRange(), txnId);
+ break;
}
-
- keyLimits[keyCount - 1] = totalCount;
- keyOffset = totalCount;
- }
-
- public void add(Key key, TxnId txnId)
- {
- if (keyCount == 0 || !keys[keyCount - 1].equals(key))
- nextKey(key);
- add(txnId);
- }
-
- /**
- * Add this command as a dependency for each intersecting key
- */
- public void add(TxnId txnId)
- {
- if (hasOrderedTxnId && totalCount > keyOffset && keyToTxnId[totalCount - 1].compareTo(txnId) >= 0)
- throw new IllegalArgumentException("TxnId provided out of order");
-
- if (totalCount >= keyToTxnId.length)
- {
- TxnId[] newTxnIds = cachedTxnIds.get(keyToTxnId.length * 2);
- System.arraycopy(keyToTxnId, 0, newTxnIds, 0, totalCount);
- cachedTxnIds.forceDiscard(keyToTxnId, totalCount);
- keyToTxnId = newTxnIds;
- }
-
- keyToTxnId[totalCount++] = txnId;
- }
-
- public T build()
- {
- if (totalCount == 0)
- return build(Keys.EMPTY, NO_TXNIDS, NO_INTS);
-
- finishKey();
-
- TxnId[] uniqueTxnId = cachedTxnIds.get(totalCount);
- System.arraycopy(keyToTxnId, 0, uniqueTxnId, 0, totalCount);
- Arrays.sort(uniqueTxnId, 0, totalCount);
- int txnIdCount = 1;
- for (int i = 1 ; i < totalCount ; ++i)
- {
- if (!uniqueTxnId[txnIdCount - 1].equals(uniqueTxnId[i]))
- uniqueTxnId[txnIdCount++] = uniqueTxnId[i];
- }
-
- TxnId[] txnIds = cachedTxnIds.complete(uniqueTxnId, txnIdCount);
- cachedTxnIds.discard(uniqueTxnId, totalCount);
-
- int[] result = new int[keyCount + totalCount];
- int offset = keyCount;
- for (int k = 0 ; k < keyCount ; ++k)
- {
- result[k] = keyCount + keyLimits[k];
- int from = k == 0 ? 0 : keyLimits[k - 1];
- int to = keyLimits[k];
- offset = (int)SortedArrays.foldlIntersection(txnIds, 0, txnIdCount, keyToTxnId, from, to, (key, p, v, li, ri) -> {
- result[(int)v] = li;
- return v + 1;
- }, keyCount, offset, -1);
- }
-
- return build(Keys.ofSortedUnchecked(cachedKeys.complete(keys, keyCount)), txnIds, result);
+ return this;
}
- abstract T build(Keys keys, TxnId[] txnIds, int[] keyToTxnId);
+ public abstract T build();
@Override
public void close()
{
- cachedKeys.discard(keys, keyCount);
- cachedInts.forceDiscard(keyLimits, keyCount);
- cachedTxnIds.forceDiscard(keyToTxnId, totalCount);
- }
- }
-
- public static class OrderedBuilder extends AbstractOrderedBuilder<Deps>
- {
- public OrderedBuilder(boolean hasOrderedTxnId)
- {
- super(hasOrderedTxnId);
- }
-
- @Override
- Deps build(Keys keys, TxnId[] txnIds, int[] keysToTxnIds)
- {
- return new Deps(keys, txnIds, keysToTxnIds);
+ keyBuilder.close();
+ if (rangeBuilder != null)
+ rangeBuilder.close();
}
}
- /**
- * An object for managing a sequence of efficient linear merges Deps objects.
- * Its primary purpose is to manage input and output buffers, so that we reuse output buffers
- * as input to the next merge, and if any input is a superset of the other inputs that this input
- * is returned unmodified.
- *
- * This is achieved by using PassThroughXBuffers so that the result buffers (and their sizes) are returned
- * unmodified, and the buffers are cached as far as possible. In general, the buffers should be taken
- * out of pre-existing caches, but if the buffers are too large then we cache any additional buffers we
- * allocate for the duration of the merge.
- */
- private static class LinearMerger extends PassThroughObjectAndIntBuffers<TxnId> implements DepsConstructor<Key, TxnId, Object>
+ public static class Builder extends AbstractBuilder<Deps>
{
- final PassThroughObjectBuffers<Key> keyBuffers;
- Key[] bufKeys;
- TxnId[] bufTxnIds;
- int[] buf = null;
- int bufKeysLength, bufTxnIdsLength = 0, bufLength = 0;
- Deps from = null;
-
- LinearMerger()
- {
- super(cachedTxnIds(), cachedInts());
- keyBuffers = new PassThroughObjectBuffers<>(cachedKeys());
- }
-
- @Override
- public Object construct(Key[] keys, int keysLength, TxnId[] txnIds, int txnIdsLength, int[] out, int outLength)
- {
- if (from == null)
- {
- // if our input buffers were themselves buffers, we want to discard them unless they have been returned back to us
- discard(keys, txnIds, out);
- }
- else if (buf != out)
- {
- // the output is not equal to a prior input
- from = null;
- }
-
- if (from == null)
- {
- bufKeys = keys;
- bufKeysLength = keysLength;
- bufTxnIds = txnIds;
- bufTxnIdsLength = txnIdsLength;
- buf = out;
- bufLength = outLength;
- }
- else
- {
- Invariants.checkState(keys == bufKeys && keysLength == bufKeysLength);
- Invariants.checkState(txnIds == bufTxnIds && txnIdsLength == bufTxnIdsLength);
- Invariants.checkState(outLength == bufLength);
- }
- return null;
- }
-
- void update(Deps deps)
+ public Builder()
{
- if (buf == null)
- {
- bufKeys = deps.keys.keys;
- bufKeysLength = deps.keys.keys.length;
- bufTxnIds = deps.txnIds;
- bufTxnIdsLength = deps.txnIds.length;
- buf = deps.keyToTxnId;
- bufLength = deps.keyToTxnId.length;
- from = deps;
- return;
- }
-
- linearUnion(
- bufKeys, bufKeysLength, bufTxnIds, bufTxnIdsLength, buf, bufLength,
- deps.keys.keys, deps.keys.keys.length, deps.txnIds, deps.txnIds.length, deps.keyToTxnId, deps.keyToTxnId.length,
- keyBuffers, this, this, this
- );
- if (buf == deps.keyToTxnId)
- {
- Invariants.checkState(deps.keys.keys == bufKeys && deps.keys.keys.length == bufKeysLength);
- Invariants.checkState(deps.txnIds == bufTxnIds && deps.txnIds.length == bufTxnIdsLength);
- Invariants.checkState(deps.keyToTxnId.length == bufLength);
- from = deps;
- }
+ super();
}
- Deps get()
+ public Deps build()
{
- if (buf == null)
- return NONE;
-
- if (from != null)
- return from;
-
- return new Deps(
- Keys.ofSortedUnchecked(keyBuffers.realComplete(bufKeys, bufKeysLength)),
- realComplete(bufTxnIds, bufTxnIdsLength),
- realComplete(buf, bufLength));
- }
-
- /**
- * Free any buffers we no longer need
- */
- void discard()
- {
- if (from == null)
- discard(null, null, null);
- }
-
- /**
- * Free buffers unless they are equal to the corresponding parameter
- */
- void discard(Key[] freeKeysIfNot, TxnId[] freeTxnIdsIfNot, int[] freeBufIfNot)
- {
- if (from != null)
- return;
-
- if (bufKeys != freeKeysIfNot)
- {
- keyBuffers.realDiscard(bufKeys, bufKeysLength);
- bufKeys = null;
- }
- if (bufTxnIds != freeTxnIdsIfNot)
- {
- realDiscard(bufTxnIds, bufTxnIdsLength);
- bufTxnIds = null;
- }
- if (buf != freeBufIfNot)
- {
- realDiscard(buf, bufLength);
- buf = null;
- }
+ return new Deps(keyBuilder.build(), rangeBuilder == null ? RangeDeps.NONE : rangeBuilder.build());
}
}
- public static <T> Deps merge(List<T> merge, Function<T, Deps> getter)
- {
- LinearMerger linearMerger = new LinearMerger();
- try
- {
- int mergeIndex = 0, mergeSize = merge.size();
- while (mergeIndex < mergeSize)
- {
- Deps deps = getter.apply(merge.get(mergeIndex++));
- if (deps == null || deps.isEmpty())
- continue;
-
- linearMerger.update(deps);
- }
-
- return linearMerger.get();
- }
- finally
- {
- linearMerger.discard();
- }
- }
-
- final Keys keys; // unique Keys
- final TxnId[] txnIds; // unique TxnId TODO (low priority, efficiency): this could be a BTree?
+ public final KeyDeps keyDeps;
+ public final RangeDeps rangeDeps;
- /**
- * This represents a map of {@code Key -> [TxnId] } where each TxnId is actually a pointer into the txnIds array.
- * The beginning of the array (the first keys.size() entries) are offsets into this array.
- * <p/>
- * Example:
- * <p/>
- * {@code
- * int keyIdx = keys.indexOf(key);
- * int startOfTxnOffset = keyIdx == 0 ? keys.size() : keyToTxnId[keyIdx - 1];
- * int endOfTxnOffset = keyToTxnId[keyIdx];
- * for (int i = startOfTxnOffset; i < endOfTxnOffset; i++)
- * {
- * TxnId id = txnIds[keyToTxnId[i]]
- * ...
- * }
- * }
- */
- final int[] keyToTxnId; // Key -> [TxnId]
- // Lazy loaded in ensureTxnIdToKey()
- int[] txnIdToKey; // TxnId -> [Key]
-
- Deps(Keys keys, TxnId[] txnIds, int[] keyToTxnId)
+ public Deps(KeyDeps keyDeps, RangeDeps rangeDeps)
{
- this.keys = keys;
- this.txnIds = txnIds;
- this.keyToTxnId = keyToTxnId;
- if (!(keys.isEmpty() || keyToTxnId[keys.size() - 1] == keyToTxnId.length))
- throw new IllegalArgumentException(String.format("Last key (%s) in keyToTxnId does not point (%d) to the end of the array (%d);\nkeyToTxnId=%s", keys.get(keys.size() - 1), keyToTxnId[keys.size() - 1], keyToTxnId.length, Arrays.toString(keyToTxnId)));
- if (DEBUG_CHECKS)
- checkValid();
+ this.keyDeps = keyDeps;
+ this.rangeDeps = rangeDeps;
}
- public PartialDeps slice(Ranges ranges)
- {
- if (isEmpty())
- return new PartialDeps(ranges, keys, txnIds, keyToTxnId);
-
- Keys select = keys.slice(ranges);
-
- if (select.isEmpty())
- return new PartialDeps(ranges, Keys.EMPTY, NO_TXNIDS, NO_INTS);
-
- if (select.size() == keys.size())
- return new PartialDeps(ranges, keys, txnIds, keyToTxnId);
-
- int i = 0;
- int offset = select.size();
- for (int j = 0 ; j < select.size() ; ++j)
- {
- int findi = keys.findNext(i, select.get(j), FAST);
- if (findi < 0)
- continue;
-
- i = findi;
- offset += keyToTxnId[i] - (i == 0 ? keys.size() : keyToTxnId[i - 1]);
- }
-
- int[] src = keyToTxnId;
- int[] trg = new int[offset];
-
- i = 0;
- offset = select.size();
- for (int j = 0 ; j < select.size() ; ++j)
- {
- int findi = keys.findNext(i, select.get(j), FAST);
- if (findi >= 0)
- {
- i = findi;
- int start = i == 0 ? keys.size() : src[i - 1];
- int count = src[i] - start;
- System.arraycopy(src, start, trg, offset, count);
- offset += count;
- }
- trg[j] = offset;
- }
-
- TxnId[] txnIds = trimUnusedTxnId(select, this.txnIds, trg);
- return new PartialDeps(ranges, select, txnIds, trg);
- }
-
- /**
- * Returns the set of {@link TxnId}s that are referenced by {@code keysToTxnId}, and <strong>updates</strong>
- * {@code keysToTxnId} to point to the new offsets in the returned set.
- * @param keys object referenced by {@code keysToTxnId} index
- * @param txnIds to trim to the seen {@link TxnId}s
- * @param keysToTxnId to use as reference for trimming, this index will be updated to reflect the trimmed offsets.
- * @return smallest set of {@link TxnId} seen in {@code keysToTxnId}
- */
- private static TxnId[] trimUnusedTxnId(Keys keys, TxnId[] txnIds, int[] keysToTxnId)
+ public boolean contains(TxnId txnId)
{
- IntBuffers cache = ArrayBuffers.cachedInts();
- // we use remapTxnId twice:
- // - first we use the end to store a bitmap of those TxnId we are actually using
- // - then we use it to store the remap index (incrementally replacing the bitmap)
- int bitMapOffset = txnIds.length + 1 - (txnIds.length+31)/32;
- int[] remapTxnId = cache.getInts(txnIds.length + 1);
- try
- {
- Arrays.fill(remapTxnId, bitMapOffset, txnIds.length + 1, 0);
- for (int i = keys.size() ; i < keysToTxnId.length ; ++i)
- setBit(remapTxnId, bitMapOffset, keysToTxnId[i]);
-
- int offset = 0;
- for (int i = 0 ; i < txnIds.length ; ++i)
- {
- if (hasSetBit(remapTxnId, bitMapOffset, i)) remapTxnId[i] = offset++;
- else remapTxnId[i] = -1;
- }
-
- TxnId[] result = txnIds;
- if (offset < txnIds.length)
- {
- result = new TxnId[offset];
- for (int i = 0 ; i < txnIds.length ; ++i)
- {
- if (remapTxnId[i] >= 0)
- result[remapTxnId[i]] = txnIds[i];
- }
- // Update keysToTxnId to point to the new remapped TxnId offsets
- for (int i = keys.size() ; i < keysToTxnId.length ; ++i)
- keysToTxnId[i] = remapTxnId[keysToTxnId[i]];
- }
-
- return result;
- }
- finally
- {
- cache.forceDiscard(remapTxnId, txnIds.length);
- }
+ return keyDeps.contains(txnId) || rangeDeps.contains(txnId);
}
public Deps with(Deps that)
{
- if (isEmpty() || that.isEmpty())
- return isEmpty() ? that : this;
-
- return linearUnion(
- this.keys.keys, this.keys.keys.length, this.txnIds, this.txnIds.length, this.keyToTxnId, this.keyToTxnId.length,
- that.keys.keys, that.keys.keys.length, that.txnIds, that.txnIds.length, that.keyToTxnId, that.keyToTxnId.length,
- cachedKeys(), cachedTxnIds(), cachedInts(),
- (keys, keysLength, txnIds, txnIdsLength, out, outLength) ->
- new Deps(Keys.ofSortedUnchecked(cachedKeys().complete(keys, keysLength)),
- cachedTxnIds().complete(txnIds, txnIdsLength),
- cachedInts().complete(out, outLength))
- );
- }
-
- /**
- * Turn a set of key, value and mapping buffers into a merge result;
- * K and V are either Key and TxnId, or vice versa, depending on which mapping direction was present
- */
- interface DepsConstructor<K, V, T>
- {
- T construct(K[] keys, int keysLength, V[] values, int valuesLength, int[] out, int outLength);
- }
-
- private static boolean arraysEqual(int[] left, int[] right, int length)
- {
- if (left.length < length || right.length < length)
- return false;
-
- for (int i=0; i<length; i++)
- if (left[i] !=right[i])
- return false;
-
- return true;
- }
-
- private static <T> boolean arraysEqual(T[] left, T[] right, int length)
- {
- if (left.length < length || right.length < length)
- return false;
-
- for (int i=0; i<length; i++)
- if (!Objects.equals(left[i], right[i]))
- return false;
-
- return true;
- }
-
- // TODO (low priority, efficiency): this method supports merging keyToTxnId OR txnIdToKey; we can perhaps save time
- // and effort when constructing Deps on remote hosts by only producing txnIdToKey with OrderedCollector and serializing
- // only this, and merging on the recipient before inverting, so that we only have to invert the final assembled deps
- private static <K extends Comparable<? super K>, V extends Comparable<? super V>, T>
- T linearUnion(K[] leftKeys, int leftKeysLength, V[] leftValues, int leftValuesLength, int[] left, int leftLength,
- K[] rightKeys, int rightKeysLength, V[] rightValues, int rightValuesLength, int[] right, int rightLength,
- ObjectBuffers<K> keyBuffers, ObjectBuffers<V> valueBuffers, IntBuffers intBuffers, DepsConstructor<K, V, T> constructor)
- {
- K[] outKeys = null;
- V[] outValues = null;
- int[] remapLeft = null, remapRight = null, out = null;
- int outLength = 0, outKeysLength = 0, outTxnIdsLength = 0;
-
- try
- {
- outKeys = SortedArrays.linearUnion(leftKeys, leftKeysLength, rightKeys, rightKeysLength, keyBuffers);
- outKeysLength = keyBuffers.lengthOfLast(outKeys);
- outValues = SortedArrays.linearUnion(leftValues, leftValuesLength, rightValues, rightValuesLength, valueBuffers);
- outTxnIdsLength = valueBuffers.lengthOfLast(outValues);
-
- remapLeft = remapToSuperset(leftValues, leftValuesLength, outValues, outTxnIdsLength, intBuffers);
- remapRight = remapToSuperset(rightValues, rightValuesLength, outValues, outTxnIdsLength, intBuffers);
-
- if (remapLeft == null && remapRight == null && leftLength == rightLength && leftKeysLength == rightKeysLength
- && arraysEqual(left, right, rightLength)
- && arraysEqual(leftKeys, rightKeys, rightKeysLength)
- )
- {
- return constructor.construct(leftKeys, leftKeysLength, leftValues, leftValuesLength, left, leftLength);
- }
-
- int lk = 0, rk = 0, ok = 0, l = leftKeysLength, r = rightKeysLength;
- outLength = outKeysLength;
-
- if (remapLeft == null && outKeys == leftKeys)
- {
- // "this" knows all the TxnId and Keys already, but do both agree on what Keys map to TxnIds?
- noOp: while (lk < leftKeysLength && rk < rightKeysLength)
- {
- int ck = leftKeys[lk].compareTo(rightKeys[rk]);
- if (ck < 0)
- {
- // "this" knows of a key not present in "that"
- outLength += left[lk] - l; // logically append the key's TxnIds to the size
- l = left[lk];
- assert outLength == l && ok == lk && left[ok] == outLength;
- ok++;
- lk++;
- }
- else if (ck > 0)
- {
- // if this happened there is a bug with keys.union or keys are not actually sorted
- throwUnexpectedMissingKeyException(leftKeys, lk, leftKeysLength, rightKeys, rk, rightKeysLength, true);
- }
- else
- {
- // both "this" and "that" know of the key
- while (l < left[lk] && r < right[rk])
- {
- int nextLeft = left[l];
- int nextRight = remap(right[r], remapRight);
-
- if (nextLeft < nextRight)
- {
- // "this" knows of the txn that "that" didn't
- outLength++;
- l++;
- }
- else if (nextRight < nextLeft)
- {
- out = copy(left, outLength, leftLength + rightLength - r, intBuffers);
- break noOp;
- }
- else
- {
- outLength++;
- l++;
- r++;
- }
- }
-
- if (l < left[lk])
- {
- outLength += left[lk] - l;
- l = left[lk];
- }
- else if (r < right[rk])
- {
- // "that" thinks a key includes a TxnId as a dependency but "this" doesn't, need to include this knowledge
- out = copy(left, outLength, leftLength + rightLength - r, intBuffers);
- break;
- }
-
- assert outLength == l && ok == lk && left[ok] == outLength;
- ok++;
- rk++;
- lk++;
- }
- }
-
- if (out == null)
- return constructor.construct(leftKeys, leftKeysLength, leftValues, leftValuesLength, left, leftLength);
- }
- else if (remapRight == null && outKeys == rightKeys)
- {
- // "that" knows all the TxnId and keys already, but "this" does not
- noOp: while (lk < leftKeysLength && rk < rightKeysLength)
- {
- int ck = leftKeys[lk].compareTo(rightKeys[rk]);
- if (ck < 0)
- {
- // if this happened there is a bug with keys.union or keys are not actually sorted
- throwUnexpectedMissingKeyException(leftKeys, lk, leftKeysLength, rightKeys, rk, rightKeysLength, false);
- }
- else if (ck > 0)
- {
- outLength += right[rk] - r;
- r = right[rk];
- assert outLength == r && ok == rk && right[ok] == outLength;
- ok++;
- rk++;
- }
- else
- {
- // both "this" and "that" know of the key
- while (l < left[lk] && r < right[rk])
- {
- int nextLeft = remap(left[l], remapLeft);
- int nextRight = right[r];
-
- if (nextLeft < nextRight)
- {
- // "this" thinks a TxnID depends on Key but "that" doesn't, need to include this knowledge
- out = copy(right, outLength, rightLength + leftLength - l, intBuffers);
- break noOp;
- }
- else if (nextRight < nextLeft)
- {
- // "that" knows of the txn that "this" didn't
- outLength++;
- r++;
- }
- else
- {
- outLength++;
- l++;
- r++;
- }
- }
-
- if (l < left[lk])
- {
- out = copy(right, outLength, rightLength + leftLength - l, intBuffers);
- break;
- }
- else if (r < right[rk])
- {
- outLength += right[rk] - r;
- r = right[rk];
- }
-
- assert outLength == r && ok == rk && right[ok] == outLength;
- ok++;
- rk++;
- lk++;
- }
- }
-
- if (out == null)
- return constructor.construct(rightKeys, rightKeysLength, rightValues, rightValuesLength, right, rightLength);
- }
- else
- {
- out = intBuffers.getInts(leftLength + rightLength);
- }
-
- while (lk < leftKeysLength && rk < rightKeysLength)
- {
- int ck = leftKeys[lk].compareTo(rightKeys[rk]);
- if (ck < 0)
- {
- while (l < left[lk])
- out[outLength++] = remap(left[l++], remapLeft);
- out[ok++] = outLength;
- lk++;
- }
- else if (ck > 0)
- {
- while (r < right[rk])
- out[outLength++] = remap(right[r++], remapRight);
- out[ok++] = outLength;
- rk++;
- }
- else
- {
- while (l < left[lk] && r < right[rk])
- {
- int nextLeft = remap(left[l], remapLeft);
- int nextRight = remap(right[r], remapRight);
-
- if (nextLeft <= nextRight)
- {
- out[outLength++] = nextLeft;
- l += 1;
- r += nextLeft == nextRight ? 1 : 0;
- }
- else
- {
- out[outLength++] = nextRight;
- ++r;
- }
- }
-
- while (l < left[lk])
- out[outLength++] = remap(left[l++], remapLeft);
-
- while (r < right[rk])
- out[outLength++] = remap(right[r++], remapRight);
-
- out[ok++] = outLength;
- rk++;
- lk++;
- }
- }
-
- while (lk < leftKeysLength)
- {
- while (l < left[lk])
- out[outLength++] = remap(left[l++], remapLeft);
- out[ok++] = outLength;
- lk++;
- }
-
- while (rk < rightKeysLength)
- {
- while (r < right[rk])
- out[outLength++] = remap(right[r++], remapRight);
- out[ok++] = outLength;
- rk++;
- }
-
- return constructor.construct(outKeys, outKeysLength, outValues, outTxnIdsLength, out, outLength);
- }
- finally
- {
- if (outKeys != null)
- keyBuffers.discard(outKeys, outKeysLength);
- if (outValues != null)
- valueBuffers.discard(outValues, outTxnIdsLength);
- if (out != null)
- intBuffers.discard(out, outLength);
- if (remapLeft != null)
- intBuffers.forceDiscard(remapLeft, leftValuesLength);
- if (remapRight != null)
- intBuffers.forceDiscard(remapRight, rightValuesLength);
- }
- }
-
- private static <A> void throwUnexpectedMissingKeyException(A[] leftKeys, int leftKeyIndex, int leftKeyLength, A[] rightKeys, int rightKeyIndex, int rightKeyLength, boolean isMissingLeft)
- {
- StringBuilder sb = new StringBuilder();
- String missing = isMissingLeft ? "left" : "right";
- String extra = isMissingLeft ? "right" : "left";
- sb.append(missing).append(" knows all keys, yet ").append(extra).append(" knew of an extra key at indexes left[")
- .append(leftKeyIndex).append("] = ").append(leftKeys[leftKeyIndex])
- .append(", right[").append(rightKeyIndex).append("] = ").append(rightKeys[rightKeyIndex]).append("\n");
- sb.append("leftKeys = ").append(Arrays.stream(leftKeys, 0, leftKeyLength).map(Object::toString).collect(Collectors.joining())).append('\n');
- sb.append("rightKeys = ").append(Arrays.stream(rightKeys, 0, rightKeyLength).map(Object::toString).collect(Collectors.joining())).append('\n');
- throw new IllegalStateException(sb.toString());
- }
-
- private static int[] copy(int[] src, int to, int length, IntBuffers bufferManager)
- {
- if (length == 0)
- return NO_INTS;
-
- int[] result = bufferManager.getInts(length);
- if (result.length < length)
- throw new IllegalStateException();
- System.arraycopy(src, 0, result, 0, to);
- return result;
+ return new Deps(this.keyDeps.with(that.keyDeps), this.rangeDeps.with(that.rangeDeps));
}
public Deps without(Predicate<TxnId> remove)
{
- if (isEmpty())
- return this;
-
- IntBuffers cache = ArrayBuffers.cachedInts();
- TxnId[] oldTxnIds = txnIds;
- int[] oldKeyToTxnId = keyToTxnId;
- int[] remapTxnIds = cache.getInts(oldTxnIds.length);
- int[] newKeyToTxnId = null;
- TxnId[] newTxnIds;
- int o = 0;
- try
- {
- int count = 0;
- for (int i = 0 ; i < oldTxnIds.length ; ++i)
- {
- if (remove.test(oldTxnIds[i])) remapTxnIds[i] = -1;
- else remapTxnIds[i] = count++;
- }
-
- if (count == oldTxnIds.length)
- return this;
-
- if (count == 0)
- return NONE;
-
- newTxnIds = new TxnId[count];
- for (int i = 0 ; i < oldTxnIds.length ; ++i)
- {
- if (remapTxnIds[i] >= 0)
- newTxnIds[remapTxnIds[i]] = oldTxnIds[i];
- }
-
- newKeyToTxnId = cache.getInts(oldKeyToTxnId.length);
- int k = 0, i = keys.size();
- o = i;
- while (i < oldKeyToTxnId.length)
- {
- while (oldKeyToTxnId[k] == i)
- newKeyToTxnId[k++] = o;
-
- int remapped = remapTxnIds[oldKeyToTxnId[i]];
- if (remapped >= 0)
- newKeyToTxnId[o++] = remapped;
- ++i;
- }
-
- while (k < keys.size())
- newKeyToTxnId[k++] = o;
- }
- catch (Throwable t)
- {
- cache.forceDiscard(newKeyToTxnId, o);
- throw t;
- }
- finally
- {
- cache.forceDiscard(remapTxnIds, oldTxnIds.length);
- }
-
- newKeyToTxnId = cache.completeAndDiscard(newKeyToTxnId, o);
- return new Deps(keys, newTxnIds, newKeyToTxnId);
+ return new Deps(keyDeps.without(remove), rangeDeps.without(remove));
}
- public boolean contains(TxnId txnId)
+ public PartialDeps slice(Ranges covering)
{
- return Arrays.binarySearch(txnIds, txnId) >= 0;
+ return new PartialDeps(covering, keyDeps.slice(covering), rangeDeps.slice(covering));
}
- // return true iff we map any keys to any txnId
- // if the mapping is empty we return false, whether or not we have any keys or txnId by themselves
public boolean isEmpty()
{
- return keyToTxnId.length == keys.size();
- }
-
- public Keys someKeys(TxnId txnId)
- {
- int txnIdIndex = Arrays.binarySearch(txnIds, txnId);
- if (txnIdIndex < 0)
- return Keys.EMPTY;
-
- ensureTxnIdToKey();
-
- int start = txnIdIndex == 0 ? txnIds.length : txnIdToKey[txnIdIndex - 1];
- int end = txnIdToKey[txnIdIndex];
- if (start == end)
- return Keys.EMPTY;
-
- Key[] result = new Key[end - start];
- for (int i = start ; i < end ; ++i)
- result[i - start] = keys.get(txnIdToKey[i]);
- return Keys.of(result);
- }
-
- public Unseekables<RoutingKey, ?> someRoutables(TxnId txnId)
- {
- return toUnseekables(txnId, array -> {
- if (array.length == 0)
- throw new IllegalStateException("Cannot create a RouteFragment without any keys");
- return new RoutingKeys(array);
- });
- }
-
- private <R> R toUnseekables(TxnId txnId, Function<RoutingKey[], R> constructor)
- {
- int txnIdIndex = Arrays.binarySearch(txnIds, txnId);
- if (txnIdIndex < 0)
- constructor.apply(RoutingKeys.EMPTY.keys);
-
- ensureTxnIdToKey();
-
- int start = txnIdIndex == 0 ? txnIds.length : txnIdToKey[txnIdIndex - 1];
- int end = txnIdToKey[txnIdIndex];
- RoutingKey[] result = new RoutingKey[end - start];
- if (start == end)
- constructor.apply(RoutingKeys.EMPTY.keys);
-
- result[0] = keys.get(txnIdToKey[start]).toUnseekable();
- int resultCount = 1;
- for (int i = start + 1 ; i < end ; ++i)
- {
- RoutingKey next = keys.get(txnIdToKey[i]).toUnseekable();
- if (!next.equals(result[resultCount - 1]))
- result[resultCount++] = next;
- }
-
- if (resultCount < result.length)
- result = Arrays.copyOf(result, resultCount);
- return constructor.apply(result);
- }
-
- void ensureTxnIdToKey()
- {
- if (txnIdToKey != null)
- return;
-
- txnIdToKey = invert(keyToTxnId, keyToTxnId.length, keys.size(), txnIds.length);
- }
-
- private static int[] invert(int[] src, int srcLength, int srcKeyCount, int trgKeyCount)
- {
- int[] trg = new int[trgKeyCount + srcLength - srcKeyCount];
-
- // first pass, count number of txnId per key
- for (int i = srcKeyCount ; i < srcLength ; ++i)
- trg[src[i]]++;
-
- // turn into offsets (i.e. add txnIds.size() and then sum them)
- trg[0] += trgKeyCount;
- for (int i = 1; i < trgKeyCount ; ++i)
- trg[i] += trg[i - 1];
-
- // shuffle forwards one, so we have the start index rather than end
- System.arraycopy(trg, 0, trg, 1, trgKeyCount - 1);
- trg[0] = trgKeyCount;
-
- // convert the offsets to end, and set the key at the target positions
- int k = 0;
- for (int i = srcKeyCount ; i < srcLength ; ++i)
- {
- // if at the end offset, switch to the next key
- while (i == src[k])
- ++k;
-
- // find the next key offset for the TxnId and set the offset to this key
- trg[trg[src[i]]++] = k;
- }
-
- return trg;
- }
-
- public void forEachOn(Ranges ranges, Predicate<Key> include, BiConsumer<Key, TxnId> forEach)
- {
- Routables.foldl(keys, ranges, (key, value, index) -> {
- if (!include.test(key))
- return null;
-
- for (int t = startOffset(index), end = endOffset(index); t < end ; ++t)
- {
- TxnId txnId = txnIds[keyToTxnId[t]];
- forEach.accept(key, txnId);
- }
- return null;
- }, null);
- }
-
- /**
- * For each {@link TxnId} that references a key within the {@link Ranges}; the {@link TxnId} will be seen exactly once.
- * @param ranges to match on
- * @param forEach function to call on each unique {@link TxnId}
- */
- public void forEachOn(Ranges ranges, Consumer<TxnId> forEach)
- {
- // Find all keys within the ranges, but record existence within an int64 bitset. Since the bitset is limited
- // to 64, this search must be called multiple times searching for different TxnIds in txnIds; this also has
- // the property that forEach is called in TxnId order.
- //TODO (expected, efficiency): reconsider this, probably not worth trying to save allocations at cost of multiple loop
- // use BitSet, or perhaps extend so we can have no nested allocations when few bits
- for (int offset = 0 ; offset < txnIds.length ; offset += 64)
- {
- long bitset = Routables.foldl(keys, ranges, (key, off, value, keyIndex) -> {
- int index = startOffset(keyIndex);
- int end = endOffset(keyIndex);
- if (off > 0)
- {
- // TODO (low priority, efficiency): interpolation search probably great here
- index = Arrays.binarySearch(keyToTxnId, index, end, (int)off);
- if (index < 0)
- index = -1 - index;
- }
-
- while (index < end)
- {
- long next = keyToTxnId[index++] - off;
- if (next >= 64)
- break;
- value |= 1L << next;
- }
-
- return value;
- }, offset, 0, -1L);
-
- while (bitset != 0)
- {
- int i = Long.numberOfTrailingZeros(bitset);
- TxnId txnId = txnIds[offset + i];
- forEach.accept(txnId);
- bitset ^= Long.lowestOneBit(bitset);
- }
- }
- }
-
- public void forEach(Key key, Consumer<TxnId> forEach)
- {
- int keyIndex = keys.indexOf(key);
- if (keyIndex < 0)
- return;
-
- int index = startOffset(keyIndex);
- int end = endOffset(keyIndex);
- while (index < end)
- forEach.accept(txnIds[keyToTxnId[index++]]);
- }
-
- public Keys keys()
- {
- return keys;
+ return keyDeps.isEmpty() && rangeDeps.isEmpty();
}
public int txnIdCount()
{
- return txnIds.length;
- }
-
- public int totalCount()
- {
- return keyToTxnId.length - keys.size();
+ return keyDeps.txnIdCount() + rangeDeps.txnIdCount();
}
public TxnId txnId(int i)
{
- return txnIds[i];
+ return i < keyDeps.txnIdCount()
+ ? keyDeps.txnId(i)
+ : rangeDeps.txnId(i - keyDeps.txnIdCount());
}
- public Collection<TxnId> txnIds()
+ public List<TxnId> txnIds()
{
- return Arrays.asList(txnIds);
- }
-
- public List<TxnId> txnIds(Key key)
- {
- int keyIndex = keys.indexOf(key);
- if (keyIndex < 0)
- return Collections.emptyList();
-
- int start = startOffset(keyIndex);
- int end = endOffset(keyIndex);
- int size = end - start;
-
- return new AbstractList<TxnId>()
- {
+ final int txnIdCount = txnIdCount();
+ final int keyDepsCount = keyDeps.txnIdCount();
+ return new AbstractList<TxnId>() {
@Override
public TxnId get(int index)
{
- if (index > end)
- throw new IndexOutOfBoundsException();
- return txnIds[keyToTxnId[start + index]];
+ return index < keyDepsCount
+ ? keyDeps.txnId(index)
+ : rangeDeps.txnId(index - keyDepsCount);
}
@Override
- public int size()
- {
- return size;
+ public int size() {
+ return txnIdCount;
}
};
}
- private int startOffset(int keyIndex)
- {
- return keyIndex == 0 ? keys.size() : keyToTxnId[keyIndex - 1];
- }
-
- private int endOffset(int keyIndex)
+ public List<TxnId> txnIds(Seekable keyOrRange)
Review Comment:
Could use Iterable instead of List and then not have to materialize
##########
accord-core/src/main/java/accord/messages/Commit.java:
##########
@@ -36,6 +36,7 @@
import static accord.local.Status.Committed;
import static accord.local.Status.Known.DefinitionOnly;
+import static accord.primitives.Routables.Slice.Overlapping;
Review Comment:
Unused
##########
accord-core/src/main/java/accord/messages/ReadData.java:
##########
@@ -256,7 +257,7 @@ public Iterable<TxnId> txnIds()
@Override
public Seekables<?, ?> keys()
{
- return readScope;
+ return Keys.EMPTY;
Review Comment:
Nothing to preload? Will keep reading
##########
accord-core/src/main/java/accord/messages/ReadData.java:
##########
@@ -39,6 +39,7 @@
import static accord.messages.ReadData.ReadNack.NotCommitted;
import static accord.messages.ReadData.ReadNack.Redundant;
import static accord.messages.TxnRequest.*;
+import static accord.primitives.Routables.Slice.Overlapping;
Review Comment:
Unused
##########
accord-core/src/main/java/accord/messages/PreAccept.java:
##########
@@ -93,13 +92,21 @@ public PreAcceptReply apply(SafeCommandStore safeStore)
// note: this diverges from the paper, in that instead of waiting for JoinShard,
// we PreAccept to both old and new topologies and require quorums in both.
// This necessitates sending to ALL replicas of old topology, not only electorate (as fast path may be unreachable).
+ if (minEpoch < txnId.epoch() && !safeStore.ranges().at(txnId.epoch()).intersects(scope))
+ {
+ // we only preaccept in the coordination epoch, but we might contact other epochs for dependencies
Review Comment:
confused how there can be multiple epochs at preaccept since the `executeAt` epoch hasn't been determined yet? Will keep reading.
##########
accord-core/src/main/java/accord/messages/TxnRequest.java:
##########
@@ -36,6 +36,7 @@
import accord.topology.Topologies;
import accord.topology.Topology;
+import static accord.primitives.Routables.Slice.Overlapping;
Review Comment:
Unused
##########
accord-core/src/main/java/accord/messages/GetDeps.java:
##########
@@ -11,6 +11,7 @@
import java.util.Collections;
import static accord.messages.PreAccept.calculatePartialDeps;
+import static accord.primitives.Routables.Slice.Overlapping;
Review Comment:
Unused
##########
accord-core/src/main/java/accord/primitives/Routables.java:
##########
@@ -55,22 +81,61 @@
/**
* Perform {@link SortedArrays#exponentialSearch} from {@code thisIndex} looking for {@code find} with behaviour of {@code search}
*/
- int findNext(int thisIndex, K find, SortedArrays.Search search);
+ int findNext(int thisIndex, RoutableKey find, SortedArrays.Search search);
- Routable.Domain kindOfContents();
+ Domain domain();
+
+ @Inline
+ static <Param, Output> Output switchOnDomain(Routables<?, ?> inputs, Param param, BiFunction<AbstractKeys<?, ?>, Param, Output> ifKeys, BiFunction<AbstractRanges<?>, Param, Output> ifRanges)
+ {
+ switch (inputs.domain())
+ {
+ default: throw new AssertionError();
+ case Key: return ifKeys.apply((AbstractKeys<?, ?>) inputs, param);
+ case Range: return ifRanges.apply((AbstractRanges<?>) inputs, param);
+ }
+ }
+
+ @Inline
+ static <Param, Output> Output switchOnDomain(Domain domain, Param param, Function<Param, Output> ifKeys, Function<Param, Output> ifRanges)
+ {
+ switch (domain)
+ {
+ default: throw new AssertionError();
+ case Key: return ifKeys.apply(param);
+ case Range: return ifRanges.apply(param);
+ }
+ }
@Inline
static <Input extends Routable, T> T foldl(Routables<Input, ?> inputs, AbstractRanges<?> matching, IndexedFold<? super Input, T> fold, T initialValue)
{
return Helper.foldl(Routables::findNextIntersection, Helper::findLimit, inputs, matching, fold, initialValue);
}
+ @Inline
+ static <T> T foldlMinimal(Seekables<?, ?> inputs, AbstractRanges<?> matching, IndexedFold<? super Seekable, T> fold, T initialValue)
+ {
+ return Helper.foldlMinimal(inputs, matching, fold, initialValue);
Review Comment:
Could really use an explanation for what minimal means.
There are 8 different wrappers and 4 different implementations in Helper. At this point they need documentation (implementation at least) just to speed up knowing what they do.
##########
accord-core/src/main/java/accord/primitives/Ranges.java:
##########
@@ -122,6 +150,14 @@ public Ranges union(UnionMode mode, Ranges that)
});
}
+ public Ranges union(UnionMode mode, AbstractRanges<?> that)
+ {
+ return union(mode, this, that, this, that, (left, right, ranges) -> {
+ if (ranges == left.ranges) return left;
Review Comment:
This doesn't have `if (ranges == right.ranges) return right`
I assume this is some case where left and right are already the same contents?
Or is it because we only know `left` is a `Ranges`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org