You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "aweisberg (via GitHub)" <gi...@apache.org> on 2023/09/28 21:20:49 UTC
[GitHub] [cassandra-accord] aweisberg commented on a diff in pull request #58: Improvements to chaos and various bug fixes
aweisberg commented on code in PR #58:
URL: https://github.com/apache/cassandra-accord/pull/58#discussion_r1331936281
##########
accord-core/src/main/java/accord/coordinate/CoordinateNoOp.java:
##########
@@ -76,13 +76,15 @@ void onPreAccepted(Topologies topologies, Timestamp executeAt, List<PreAcceptOk>
}
else
{
- new Propose<Timestamp>(node, topologies, Ballot.ZERO, txnId, txn, route, executeAt, Deps.NONE, this)
+ Deps deps = Deps.merge(successes, ok -> ok.deps);
+ new Propose<Timestamp>(node, topologies, Ballot.ZERO, txnId, txn, route, executeAt, deps, this)
{
@Override
void onAccepted()
{
Writes writes = txn.execute(txnId, txnId, null);
Result result = txn.result(txnId, executeAt, null);
+ Deps deps = Deps.merge(acceptOks, ok -> ok.deps);
Apply.sendMaximal(node, txnId, route, txn, executeAt, deps, writes, result);
Review Comment:
```suggestion
Deps preacceptDeps = Deps.merge(successes, ok -> ok.deps);
new Propose<Timestamp>(node, topologies, Ballot.ZERO, txnId, txn, route, executeAt, preacceptDeps, this)
{
@Override
void onAccepted()
{
Writes writes = txn.execute(txnId, txnId, null);
Result result = txn.result(txnId, executeAt, null);
Deps acceptDeps = Deps.merge(this.acceptOks, ok -> ok.deps);
Apply.sendMaximal(node, txnId, route, txn, executeAt, acceptDeps, writes, result);
```
Just a little more explicit about there being two different sets of deps in scope and that acceptOks comes from the `Propose`.
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +442,83 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // TODO (required): erasing implies that the whole command has been fully applied globally, when in fact it
+ // might only be a single shard that has Truncated its state. The simplest thing for us here to ensure consistency
+ // is to permit partial application where possible, instead of erasing the whole state; we could also consider
+ // more clearly reifying the difference between Erased and Truncated(WithoutAnyKnowledge).
+ // In practice this may not be important, because a coordinator cannot have relied on this replica's vote for durability
+ // and so our Erased vote should not adversely impact any inferences from a quorum of responses. But for consistency
+ // and ease of logic this is best to retain.
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+
+ Known required = PreApplied.minKnown;
+ Known requireExtra = required.subtract(command.known());
+ Ranges achieveRanges = full.sufficientFor(requireExtra, executeRanges);
+ Participants<?> participants = route.participants().slice(executeRanges, Minimal);
+
+ Ranges staleRanges = executeRanges.subtract(achieveRanges);
+ Participants<?> staleParticipants = participants.slice(staleRanges, Minimal);
+ staleRanges = staleParticipants.toRanges();
+
+ if (staleRanges.isEmpty())
+ {
+ Invariants.checkState(achieveRanges.containsAll(participants));
+ return required;
+ }
+
+ // TODO (expected): try to partially apply the transaction locally to limit the ranges we have to mark dead
+ safeStore.commandStore().markShardStale(safeStore, executeAt, staleRanges, true);
+ if (!staleRanges.containsAll(participants))
+ return required;
+
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ private Void updateDurability(SafeCommandStore safeStore, SafeCommand safeCommand)
+ {
+ // TODO (expected): Infer durability status from cleanup/truncation
RoutingKey homeKey = full.homeKey;
if (!full.durability.isDurable() || homeKey == null)
Review Comment:
What does the home key being null mean here? Why would we only `setDurability` if this is part of the coordinating shard?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ // TODO (required): we could in principle be stale for future epochs we haven't witnessed yet. Ensure up to date epochs before finalising this application.
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+ executeRanges = safeStore.commandStore().redundantBefore().expectToExecute(txnId, executeAt, executeRanges);
+
+ if (executeRanges.isEmpty())
+ {
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Known required = PreApplied.minKnown;
+ Known requireExtra = required.subtract(command.known());
+ Ranges achieveRanges = full.sufficientFor(requireExtra, executeRanges);
+ Participants<?> participants = route.participants().slice(executeRanges, Minimal);
+
+ if (participants.isEmpty())
+ {
+ // we only coordinate this transaction, so being unable to retrieve its state does not imply any staleness
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Ranges staleRanges = executeRanges.subtract(achieveRanges);
+ Participants<?> staleParticipants = participants.slice(staleRanges, Minimal);
+ staleRanges = staleParticipants.toRanges();
+
+ if (staleRanges.isEmpty())
+ {
+ Invariants.checkState(achieveRanges.containsAll(participants));
+ return required;
+ }
+
+ safeStore.commandStore().markShardStale(safeStore, executeAt, staleRanges, true);
+ if (!staleRanges.containsAll(participants))
+ return required;
+
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ private Void updateDurability(SafeCommandStore safeStore, SafeCommand safeCommand)
+ {
+ // TODO (expected): Infer durability status from cleanup/truncation
RoutingKey homeKey = full.homeKey;
if (!full.durability.isDurable() || homeKey == null)
return null;
if (!safeStore.ranges().coordinates(txnId).contains(homeKey))
return null;
- Timestamp executeAt = full.saveStatus.known.executeAt.hasDecidedExecuteAt() ? full.executeAt : null;
+ Timestamp executeAt = full.executeAtIfKnown();
Commands.setDurability(safeStore, safeCommand, full.durability, route, executeAt);
Review Comment:
Why do we only use durability information collected for the coordination shard (the `!safeStore.ranges().coordinates(txnId).contains(homeKey)` check earlier)?
Don't all shards want to find out about durability and cleanup?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ // TODO (required): we could in principle be stale for future epochs we haven't witnessed yet. Ensure up to date epochs before finalising this application.
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+ executeRanges = safeStore.commandStore().redundantBefore().expectToExecute(txnId, executeAt, executeRanges);
+
+ if (executeRanges.isEmpty())
+ {
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Known required = PreApplied.minKnown;
Review Comment:
Why `PreApplied.minKnown` is required and required for what?
##########
accord-core/src/main/java/accord/coordinate/Infer.java:
##########
@@ -168,18 +170,24 @@ public static Status invalidIfNotAtLeast(SafeCommandStore safeStore, TxnId txnId
return Status.NotDefined;
}
- public static boolean inferInvalidated(CheckStatus.WithQuorum withQuorum, Status invalidIfNotAtLeast, SaveStatus saveStatus, SaveStatus maxSaveStatus)
+ public static boolean inferInvalidated(CheckStatus.WithQuorum withQuorum, Status minInvalidIfNotAtLeast, Status maxInvalidIfNotAtLeast, SaveStatus saveStatus, SaveStatus maxSaveStatus, Ballot minPromised)
Review Comment:
What is `inferInvalidated` about? Does this exist when we can't discover anyone who says it was invalidated, due to truncation or non-participation, but we need to know it was invalidated?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +419,80 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
Review Comment:
the?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -314,11 +317,14 @@ public static void propagate(Node node, TxnId txnId, long sourceEpoch, WithQuoru
else
{
Invariants.checkState(sourceEpoch == txnId.epoch(), "%d != %d", sourceEpoch, txnId.epoch());
- achieved = new Known(achieved.definition, achieved.executeAt, knownForExecution.deps, knownForExecution.outcome);
+ achieved = new Known(achieved.route, achieved.definition, achieved.executeAt, knownForExecution.deps, knownForExecution.outcome);
}
}
}
+ // TODO (now): move to Infer
+ boolean isTruncated = withQuorum == HasQuorum && (achieved.outcome.isTruncated() || (achieved.outcome == Status.Outcome.Apply && full.truncated.intersects(covering)));
Review Comment:
Hard to determine when/if this is going to be set since we won't get a quorum if we end up hitting the sufficent for path, and that depends on what the target `Known` is and the current state of the transaction. Why this is strange is truncation is a terminal state and wouldn't it be captured by `Status`?
So what is this boolean `isTruncated` tracking? It's kind of a mix of several things.
##########
accord-core/src/main/java/accord/messages/CheckStatus.java:
##########
@@ -145,14 +151,20 @@ public CheckStatusReply apply(SafeCommandStore safeStore)
Ranges truncated = command.hasBeen(Truncated) ? safeStore.ranges().allBetween(command.txnId().epoch(), command.executeAtOrTxnId().epoch()) : Ranges.EMPTY;
Status invalidIfNotAtLeast = invalidIfNotAtLeast(safeStore);
boolean isCoordinating = isCoordinating(node, command);
+ Durability durability = command.durability();
+ Route<?> route = command.route();
+ if (Route.isFullRoute(route))
+ durability = Durability.merge(durability, safeStore.commandStore().durableBefore().min(txnId, route));
Review Comment:
Why does the durability information depend on the route being full?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -347,40 +353,61 @@ public Void apply(SafeCommandStore safeStore)
{
SafeCommand safeCommand = safeStore.get(txnId, this, route);
Command command = safeCommand.current();
- if (command.saveStatus().phase.compareTo(Phase.Persist) >= 0)
- return null;
+ PartialTxn partialTxn = this.partialTxn;
+ PartialDeps partialDeps = this.partialDeps;
+ switch (command.saveStatus().phase)
+ {
+ case Persist: return updateDurability(safeStore, safeCommand);
+ case Cleanup: return null;
+ }
+
+ Known achieved = this.achieved;
+ if (isTruncated)
+ {
+ achieved = applyOrUpgradeTruncated(safeStore, safeCommand, command);
+ if (achieved == null)
+ return null;
+
+ if (achieved.executeAt.isDecided())
+ {
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAt);
+ if (partialTxn == null && this.full.saveStatus.known.definition.isKnown())
+ {
+ Ranges needed = safeStore.ranges().allBetween(txnId.epoch(), executeAt.epoch());
+ PartialTxn existing = command.partialTxn();
+ if (existing != null)
+ needed = needed.subtract(existing.covering());
+ partialTxn = full.partialTxn.slice(needed, true).reconstitutePartial(needed);
Review Comment:
Is `reconstitutePartial` an implicit way of making sure all the needed ranges were present when slicing? Will `full` include what we already know?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -347,61 +353,84 @@ public Void apply(SafeCommandStore safeStore)
{
SafeCommand safeCommand = safeStore.get(txnId, this, route);
Command command = safeCommand.current();
- if (command.saveStatus().phase.compareTo(Phase.Persist) >= 0)
- return null;
- Status propagate = achieved.propagate();
+ PartialTxn partialTxn = this.partialTxn;
+ PartialDeps partialDeps = this.partialDeps;
+ switch (command.saveStatus().phase)
+ {
+ case Persist: return updateDurability(safeStore, safeCommand);
Review Comment:
```suggestion
// Already know the outcome, waiting on durability so maybe update with new durability information which can also trigger cleanup
case Persist: return updateDurability(safeStore, safeCommand);
```
##########
accord-core/src/main/java/accord/impl/SafeCommandsForKey.java:
##########
@@ -165,22 +167,30 @@ public <D> CommandsForKey listenerUpdate(Command command)
}
@VisibleForImplementation
- public <D> CommandsForKey updateLastExecutionTimestamps(Timestamp executeAt, boolean isForWriteTxn)
+ public <D> CommandsForKey updateLastExecutionTimestamps(SafeCommandStore safeStore, Timestamp executeAt, boolean isForWriteTxn)
{
CommandsForKey current = current();
Timestamp lastWrite = current.lastWriteTimestamp();
if (executeAt.compareTo(lastWrite) < 0)
+ {
+ if (!safeStore.commandStore().safeToReadAt(executeAt).contains(key.toUnseekable()))
Review Comment:
What is this case where it is OK to write with an older timestamp than a prior committed transaction?
Safe to read at is about whether it's bootstrapped right? Referring to bootstrap instead of safe to read would be clearer.
What I would infer from this is that when a range is bootstrapping we violate this constraint on `executeAt`, possibly because the way coordination works we don't enforce the invariant properly?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +442,83 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // TODO (required): erasing implies that the whole command has been fully applied globally, when in fact it
+ // might only be a single shard that has Truncated its state. The simplest thing for us here to ensure consistency
+ // is to permit partial application where possible, instead of erasing the whole state; we could also consider
+ // more clearly reifying the difference between Erased and Truncated(WithoutAnyKnowledge).
+ // In practice this may not be important, because a coordinator cannot have relied on this replica's vote for durability
+ // and so our Erased vote should not adversely impact any inferences from a quorum of responses. But for consistency
+ // and ease of logic this is best to retain.
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
Review Comment:
In what scenarios are nodes now allowed to become stale? Who is discarding state while a node is missing it?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +442,83 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // TODO (required): erasing implies that the whole command has been fully applied globally, when in fact it
+ // might only be a single shard that has Truncated its state. The simplest thing for us here to ensure consistency
+ // is to permit partial application where possible, instead of erasing the whole state; we could also consider
+ // more clearly reifying the difference between Erased and Truncated(WithoutAnyKnowledge).
+ // In practice this may not be important, because a coordinator cannot have relied on this replica's vote for durability
+ // and so our Erased vote should not adversely impact any inferences from a quorum of responses. But for consistency
+ // and ease of logic this is best to retain.
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
Review Comment:
Need to come back and review this
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -243,68 +245,69 @@ static class OnDone implements MapReduceConsume<SafeCommandStore, Void>, EpochSu
final CheckStatusOkFull full;
// this is a WHOLE NODE measure, so if commit epoch has more ranges we do not count as committed if we can only commit in coordination epoch
final Known achieved;
+ final boolean isTruncated;
+ final WithQuorum withQuorum;
final PartialTxn partialTxn;
final PartialDeps partialDeps;
final long toEpoch;
final BiConsumer<Known, Throwable> callback;
- OnDone(Node node, TxnId txnId, Route<?> route, RoutingKey progressKey, CheckStatusOkFull full, Known achieved, PartialTxn partialTxn, PartialDeps partialDeps, long toEpoch, BiConsumer<Known, Throwable> callback)
+ OnDone(Node node, TxnId txnId, Route<?> route, RoutingKey progressKey, CheckStatusOkFull full, Known achieved, boolean isTruncated, WithQuorum withQuorum, PartialTxn partialTxn, PartialDeps partialDeps, long toEpoch, BiConsumer<Known, Throwable> callback)
{
this.node = node;
this.txnId = txnId;
this.route = route;
this.progressKey = progressKey;
this.full = full;
this.achieved = achieved;
+ this.isTruncated = isTruncated;
+ this.withQuorum = withQuorum;
this.partialTxn = partialTxn;
this.partialDeps = partialDeps;
this.toEpoch = toEpoch;
this.callback = callback;
}
- @SuppressWarnings({"rawtypes", "unchecked"})
+ @SuppressWarnings({"rawtypes"})
public static void propagate(Node node, TxnId txnId, long sourceEpoch, WithQuorum withQuorum, Route route, @Nullable Known target, CheckStatusOkFull full, BiConsumer<Known, Throwable> callback)
Review Comment:
At a high level everything happening in propagate and beyond is pretty hard to figure out. The purpose, to propagate what we fetched to local command stores is obvious enough. But there are early returns throughout, repeated delegation to `updateDurability`.
This looks like a critical part of the protocol and recovery and I hope that we don't duplicate any of the logic here elsewhere, and that we can thoroughly explain what is happening and why at each step.
##########
accord-core/src/main/java/accord/local/Bootstrap.java:
##########
@@ -363,7 +363,7 @@ private synchronized void maybeComplete(SafeToRead state)
}
}
- store.markSafeToRead(state.safeToReadAt, newDone);
+ store.markSafeToRead(globalSyncId, state.safeToReadAt, newDone);
Review Comment:
Is the issue here that the bootstrap and the sync occur at different times so the ranges it is valid for are invalidated by any staleness introduced after the global sync that preceded bootstrapping?
I'm not even clear on why there is a `safeToReadAt`, is that when streaming occurred and streaming flushed memtables and ensured it wouldn't contain transactions later than `safeToReadAt`?
##########
accord-core/src/main/java/accord/local/Status.java:
##########
@@ -457,16 +535,24 @@ public enum LogicalEpoch
*/
public enum Durability
{
- NotDurable, Local, Majority, Universal, DurableOrInvalidated;
+ NotDurable, Local, MajorityOrInvalidated, Majority, UniversalOrInvalidated, Universal;
Review Comment:
Can you update the comment above?
When do we have a durability of `OrInvalidated`? When everything we can see is truncated and we don't know the outcome?
##########
accord-core/src/main/java/accord/impl/InMemoryCommandStore.java:
##########
@@ -774,7 +784,8 @@ public <T> T mapReduce(Seekables<?, ?> keysOrRanges, Ranges slice, TestKind test
if (!command.known().deps.hasProposedOrDecidedDeps())
return;
- if ((testDep == WITH) == !command.partialDeps().contains(depId))
+ // TODO (required): ensure C* matches this behaviour
+ if ((testDep == WITH) == !command.partialDeps().intersects(depId, rangeCommand.ranges))
Review Comment:
This intersection is checking that at least one key (or range) for `depId` transaction's dependency in `rangeCommand` intersect the ranges for the `rangeCommand`?
How is it possible to have a key that doesn't intersect if both the deps and the command ranges are from the same slice of the txn? Isn't that the definition of having a dependencies and slicing them?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
Review Comment:
What is this case in `safeToCleanup`?
```
if (command.is(Status.NotDefined))
return !command.saveStatus().isUninitialised();
```
When `safeToCleanup` returns `false` why do we know we are partially stale? What doesn't partially state mean? Part of what?
##########
accord-core/src/main/java/accord/local/CommandStore.java:
##########
@@ -534,30 +553,77 @@ public final void removeRedundantDependencies(Unseekables<?> participants, Waiti
}
return b;
}, builder, keyDeps, null, ignore -> false);
- RangeDeps rangeDeps = builder.deps.rangeDeps;
- // TODO (now): slice to only those ranges we own, maybe don't even construct rangeDeps.covering()
- redundantBefore().foldl(participants, (e, b, d, rs, pi, pj) -> {
- // TODO (now): foldlInt so we can track the lower rangeidx bound and not revisit unnecessarily
- boolean useBootstrap = e.isLocalRedundancyViaBootstrap();
+
+ class RangeState
+ {
+ final WaitingOn.Update builder;
int txnIdx;
+ Range range;
+ Map<Integer, Ranges> partiallyBootstrapping;
+
+ RangeState(WaitingOn.Update builder)
+ {
+ this.builder = builder;
+ }
+
+ boolean isFullyBootstrapping(int rangeTxnIdx)
+ {
+ if (partiallyBootstrapping == null)
+ partiallyBootstrapping = new HashMap<>();
+ Ranges prev = partiallyBootstrapping.get(rangeTxnIdx);
+ Ranges remaining = prev;
+ if (remaining == null) remaining = builder.deps.rangeDeps.ranges(rangeTxnIdx);
+ else Invariants.checkState(!remaining.isEmpty());
+ remaining = remaining.subtract(Ranges.of(range));
+ if (prev == null) Invariants.checkState(!remaining.isEmpty());
+ partiallyBootstrapping.put(txnIdx, remaining);
+ return remaining.isEmpty();
+ }
+ }
+ RangeDeps rangeDeps = builder.deps.rangeDeps;
+ // TODO (required, consider): slice to only those ranges we own, maybe don't even construct rangeDeps.covering()
+ redundantBefore().foldl(participants, (e, s, d, ps, pi, pj) -> {
Review Comment:
This `foldl` is pretty inscrutable. Like how do `pi` and `pj` end up being indexes into `d`? What `deps` even are `d`? Ok sure it's `rangeDeps`, but then `pi` and `pj` seem to come from indexes in the reducing range map.
##########
accord-core/src/main/java/accord/local/CommandStore.java:
##########
@@ -534,30 +553,77 @@ public final void removeRedundantDependencies(Unseekables<?> participants, Waiti
}
return b;
}, builder, keyDeps, null, ignore -> false);
- RangeDeps rangeDeps = builder.deps.rangeDeps;
- // TODO (now): slice to only those ranges we own, maybe don't even construct rangeDeps.covering()
- redundantBefore().foldl(participants, (e, b, d, rs, pi, pj) -> {
- // TODO (now): foldlInt so we can track the lower rangeidx bound and not revisit unnecessarily
- boolean useBootstrap = e.isLocalRedundancyViaBootstrap();
+
+ class RangeState
Review Comment:
At a high level what is the algorithm for relating dependencies to bootstrapping ranges that is supposed to be implemented here?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ // TODO (required): we could in principle be stale for future epochs we haven't witnessed yet. Ensure up to date epochs before finalising this application.
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+ executeRanges = safeStore.commandStore().redundantBefore().expectToExecute(txnId, executeAt, executeRanges);
+
+ if (executeRanges.isEmpty())
+ {
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Known required = PreApplied.minKnown;
+ Known requireExtra = required.subtract(command.known());
+ Ranges achieveRanges = full.sufficientFor(requireExtra, executeRanges);
+ Participants<?> participants = route.participants().slice(executeRanges, Minimal);
+
+ if (participants.isEmpty())
+ {
+ // we only coordinate this transaction, so being unable to retrieve its state does not imply any staleness
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Ranges staleRanges = executeRanges.subtract(achieveRanges);
+ Participants<?> staleParticipants = participants.slice(staleRanges, Minimal);
+ staleRanges = staleParticipants.toRanges();
+
+ if (staleRanges.isEmpty())
+ {
+ Invariants.checkState(achieveRanges.containsAll(participants));
+ return required;
+ }
+
+ safeStore.commandStore().markShardStale(safeStore, executeAt, staleRanges, true);
+ if (!staleRanges.containsAll(participants))
+ return required;
+
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ private Void updateDurability(SafeCommandStore safeStore, SafeCommand safeCommand)
Review Comment:
```suggestion
/*
* If there is new information about the command being durable and we are in the coordination shard in the coordination epoch then update the durability information and possibly cleanup
*/
private Void updateDurability(SafeCommandStore safeStore, SafeCommand safeCommand)
```
##########
accord-core/src/main/java/accord/local/CommandStore.java:
##########
@@ -534,30 +553,77 @@ public final void removeRedundantDependencies(Unseekables<?> participants, Waiti
}
return b;
}, builder, keyDeps, null, ignore -> false);
- RangeDeps rangeDeps = builder.deps.rangeDeps;
- // TODO (now): slice to only those ranges we own, maybe don't even construct rangeDeps.covering()
- redundantBefore().foldl(participants, (e, b, d, rs, pi, pj) -> {
- // TODO (now): foldlInt so we can track the lower rangeidx bound and not revisit unnecessarily
- boolean useBootstrap = e.isLocalRedundancyViaBootstrap();
+
+ class RangeState
+ {
+ final WaitingOn.Update builder;
int txnIdx;
+ Range range;
+ Map<Integer, Ranges> partiallyBootstrapping;
+
+ RangeState(WaitingOn.Update builder)
+ {
+ this.builder = builder;
+ }
+
+ boolean isFullyBootstrapping(int rangeTxnIdx)
+ {
+ if (partiallyBootstrapping == null)
+ partiallyBootstrapping = new HashMap<>();
+ Ranges prev = partiallyBootstrapping.get(rangeTxnIdx);
+ Ranges remaining = prev;
+ if (remaining == null) remaining = builder.deps.rangeDeps.ranges(rangeTxnIdx);
+ else Invariants.checkState(!remaining.isEmpty());
+ remaining = remaining.subtract(Ranges.of(range));
+ if (prev == null) Invariants.checkState(!remaining.isEmpty());
+ partiallyBootstrapping.put(txnIdx, remaining);
+ return remaining.isEmpty();
+ }
+ }
+ RangeDeps rangeDeps = builder.deps.rangeDeps;
+ // TODO (required, consider): slice to only those ranges we own, maybe don't even construct rangeDeps.covering()
+ redundantBefore().foldl(participants, (e, s, d, ps, pi, pj) -> {
+ // TODO (desired, efficiency): foldlInt so we can track the lower rangeidx bound and not revisit unnecessarily
+ WaitingOn.Update b = s.builder;
{
- int tmp = d.txnIds().find(useBootstrap ? e.bootstrappedAt : e.redundantBefore);
- txnIdx = tmp < 0 ? -1 - tmp : tmp;
+ int tmp = d.txnIds().find(e.redundantBefore);
+ s.txnIdx = tmp < 0 ? -1 - tmp : tmp;
}
- rangeDeps.forEach(participants, e.range, pi, pj, d, (d0, txnIdx0) -> {
- if (txnIdx0 < txnIdx)
- b.setAppliedOrInvalidatedRangeIdx(txnIdx0);
+ d.forEach(ps, e.range, pi, pj, b, s, (b0, s0, txnIdx) -> {
+ if (txnIdx < s0.txnIdx)
+ b0.setAppliedOrInvalidatedRangeIdx(txnIdx);
});
- return b;
- }, builder, rangeDeps, participants, ignore -> false);
+ if (e.bootstrappedAt.compareTo(e.redundantBefore) > 0)
+ {
+ {
+ int tmp = d.txnIds().find(e.bootstrappedAt);
+ s.txnIdx = tmp < 0 ? -1 - tmp : tmp;
+ }
+ s.range = e.range;
+ d.forEach(ps, e.range, pi, pj, b, s, (b0, s0, txnIdx) -> {
+ if (txnIdx < s0.txnIdx && b0.isWaitingOnRange(txnIdx))
+ {
+ if (b0.deps.rangeDeps.foldEachRange(txnIdx, s0.range, true, (r1, r2, p) -> p && r1.contains(r2)))
Review Comment:
Can the relationship between the two ranges not be contains here and just intersection? Does this still work?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -347,61 +353,84 @@ public Void apply(SafeCommandStore safeStore)
{
SafeCommand safeCommand = safeStore.get(txnId, this, route);
Command command = safeCommand.current();
- if (command.saveStatus().phase.compareTo(Phase.Persist) >= 0)
- return null;
- Status propagate = achieved.propagate();
+ PartialTxn partialTxn = this.partialTxn;
+ PartialDeps partialDeps = this.partialDeps;
+ switch (command.saveStatus().phase)
+ {
+ case Persist: return updateDurability(safeStore, safeCommand);
+ case Cleanup: return null;
Review Comment:
```suggestion
// Already in cleanup nothing to do
case Cleanup: return null;
```
##########
accord-core/src/main/java/accord/local/RedundantBefore.java:
##########
@@ -47,19 +52,23 @@ public static RedundantBefore create(boolean inclusiveEnds, RoutingKey[] ends, E
}
}
+ public enum PreBootstrapOrStale { NOT_OWNED, FULLY, PARTIALLY, POST_BOOTSTRAP }
Review Comment:
Document these enum values
##########
accord-core/src/main/java/accord/local/Commands.java:
##########
@@ -1006,13 +1026,14 @@ public void accept(SafeCommandStore safeStore)
{
default: throw new AssertionError("Unexpected redundant status: " + redundantStatus);
case NOT_OWNED: throw new AssertionError("Invalid state: waiting for execution of command that is not owned at the execution time");
- case LOCALLY_REDUNDANT:
- case PRE_BOOTSTRAP:
- removeRedundantDependencies(safeStore, prevSafe, txnIds[depth], redundantStatus == PRE_BOOTSTRAP);
+ case SHARD_REDUNDANT:
+ case PARTIALLY_REDUNDANT_PRE_BOOTSTRAP_OR_STALE:
+ case PRE_BOOTSTRAP_OR_STALE:
+ removeRedundantDependencies(safeStore, prevSafe, txnIds[depth], redundantStatus == PRE_BOOTSTRAP_OR_STALE);
Review Comment:
Rename the parameter in `removeRedundantDependencies` to indicate it also covers staleness now?
I am having a really hard time figuring out what is going on inside `removeRedundantDependencies`. It seems like we go to the command store and get the redundant status for `prevSafe` and then go into the command store again inside `removeRedundantDependencies`, and then again remove a waiting on in the form of whatever `txnIds[depth]` is pointing to. It would help to pull that out to named variable to explain what it is.
How does `removeInvalidatedAppliedOrTruncated` work? Seems like it can return `false` without setting `appliedOrInvalidated` and it seems like `appliedOrInvalidated` is just there to prevent it from happening multiple times? Confusing as to why.
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
Review Comment:
If we know the command has been `PreApplied` why do we get to take in the durability info and bail out here? If it's truncated what if we were supposed to apply but didn't?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ // TODO (required): we could in principle be stale for future epochs we haven't witnessed yet. Ensure up to date epochs before finalising this application.
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+ executeRanges = safeStore.commandStore().redundantBefore().expectToExecute(txnId, executeAt, executeRanges);
Review Comment:
We established we didn't need to execute this transaction, does execute mean we weren't in the coordination or execution epoch? So we are aware of it (somehow), but can truncate it?
##########
accord-core/src/main/java/accord/local/RedundantBefore.java:
##########
@@ -107,15 +126,54 @@ static RedundantStatus get(Entry entry, TxnId txnId, EpochSupplier executeAt)
return entry.get(txnId);
}
+ static PreBootstrapOrStale getAndMerge(Entry entry, @Nonnull PreBootstrapOrStale prev, TxnId txnId, EpochSupplier executeAt)
+ {
+ if (prev == PARTIALLY || entry == null || entry.outOfBounds(txnId, executeAt))
+ return prev;
+
+ boolean isPreBootstrapOrStale = entry.staleUntilAtLeast != null || entry.bootstrappedAt.compareTo(txnId) > 0;
+ return isPreBootstrapOrStale ? prev == POST_BOOTSTRAP ? PARTIALLY : FULLY
Review Comment:
E_TOO_MUCH_TERNARY
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ // TODO (required): we could in principle be stale for future epochs we haven't witnessed yet. Ensure up to date epochs before finalising this application.
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+ executeRanges = safeStore.commandStore().redundantBefore().expectToExecute(txnId, executeAt, executeRanges);
+
+ if (executeRanges.isEmpty())
+ {
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Known required = PreApplied.minKnown;
+ Known requireExtra = required.subtract(command.known());
+ Ranges achieveRanges = full.sufficientFor(requireExtra, executeRanges);
+ Participants<?> participants = route.participants().slice(executeRanges, Minimal);
+
+ if (participants.isEmpty())
+ {
+ // we only coordinate this transaction, so being unable to retrieve its state does not imply any staleness
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Ranges staleRanges = executeRanges.subtract(achieveRanges);
Review Comment:
Is it going to be problematic marking stale one key at a time if it's the domain of the transaction is keys not ranges?
To make sure I understand (and this would probably be good as a comment in some form):
The ranges which we were supposed to execute on this shard, but we neither know locally or from fetching data what we need to know to get to `PreApplied`, basically whatever was in `requireExtra`?
Therefore it we will never (or likely never?) be able to apply this transaction and need to mark stale and bootstrap again.
##########
accord-core/src/main/java/accord/local/Commands.java:
##########
@@ -908,32 +926,34 @@ public static Cleanup shouldCleanup(TxnId txnId, Status status, Durability durab
// TODO (expected): we can impose additional validations here IF we receive an epoch upper bound
// TODO (expected): we should be more robust to the presence/absence of executeAt
if (route.isParticipatingHomeKey() || redundantBefore.get(txnId, toEpoch, route.homeKey()) == NOT_OWNED)
- throw new IllegalStateException("Command " + txnId + " that is being loaded is not owned by this shard on route " + route);
+ illegalState("Command " + txnId + " that is being loaded is not owned by this shard on route " + route);
case LIVE:
- case PARTIALLY_PRE_BOOTSTRAP:
- case PRE_BOOTSTRAP:
+ case PARTIALLY_PRE_BOOTSTRAP_OR_STALE:
+ case PRE_BOOTSTRAP_OR_STALE:
+ case PARTIALLY_REDUNDANT_PRE_BOOTSTRAP_OR_STALE:
return NO;
- case LOCALLY_REDUNDANT:
- if (status.hasBeen(PreCommitted) && !status.hasBeen(Applied)) // TODO (required): may be stale
- throw new IllegalStateException("Loading redundant command that has been PreCommitted but not Applied");
+ case SHARD_REDUNDANT:
+ if (status.hasBeen(PreCommitted) && !status.hasBeen(Applied) && redundantBefore.preBootstrapOrStale(txnId, toEpoch, route.participants()) != FULLY)
Review Comment:
All not `FULLY` makes this an error? Does `PreCommitted` but not `Applied` mean if we were in `PARTIALLY` or `POST_BOOTSTRAP` we should have also made it to `Applied`?
`POST_BOOTSTRAP` I get I just wasn't sure about `PARTIALLY`.
I'll probably understand this better once the `PreBootstrapOrStale` enum is documented.
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -347,61 +353,84 @@ public Void apply(SafeCommandStore safeStore)
{
SafeCommand safeCommand = safeStore.get(txnId, this, route);
Command command = safeCommand.current();
- if (command.saveStatus().phase.compareTo(Phase.Persist) >= 0)
- return null;
- Status propagate = achieved.propagate();
+ PartialTxn partialTxn = this.partialTxn;
+ PartialDeps partialDeps = this.partialDeps;
+ switch (command.saveStatus().phase)
+ {
+ case Persist: return updateDurability(safeStore, safeCommand);
+ case Cleanup: return null;
+ }
+
+ Known achieved = this.achieved;
+ if (isTruncated)
+ {
+ achieved = applyOrUpgradeTruncated(safeStore, safeCommand, command);
+ if (achieved == null)
+ return null;
+
+ if (achieved.executeAt.isDecided())
Review Comment:
`applyUpgradeTruncated` returns `required` which is set to `PreApplied.minKnown` which isn't really conditional here at the moment.
So if we hit `isTruncated` and there is any staleness detected then we continue on with achieved == `PreApplied.minKnown`. That's a lot to suss out from `applyUpgradeTruncated`.
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -347,61 +353,84 @@ public Void apply(SafeCommandStore safeStore)
{
SafeCommand safeCommand = safeStore.get(txnId, this, route);
Command command = safeCommand.current();
- if (command.saveStatus().phase.compareTo(Phase.Persist) >= 0)
- return null;
- Status propagate = achieved.propagate();
+ PartialTxn partialTxn = this.partialTxn;
+ PartialDeps partialDeps = this.partialDeps;
+ switch (command.saveStatus().phase)
+ {
+ case Persist: return updateDurability(safeStore, safeCommand);
+ case Cleanup: return null;
+ }
+
+ Known achieved = this.achieved;
+ if (isTruncated)
+ {
+ achieved = applyOrUpgradeTruncated(safeStore, safeCommand, command);
+ if (achieved == null)
+ return null;
+
+ if (achieved.executeAt.isDecided())
+ {
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAt);
+ if (partialTxn == null && this.full.saveStatus.known.definition.isKnown())
+ {
+ Ranges needed = safeStore.ranges().allBetween(txnId.epoch(), executeAt.epoch());
+ PartialTxn existing = command.partialTxn();
+ if (existing != null)
+ needed = needed.subtract(existing.covering());
+ partialTxn = full.partialTxn.slice(needed, true).reconstitutePartial(needed);
+ }
+
+ if (partialDeps == null && !safeCommand.current().known().deps.hasDecidedDeps())
Review Comment:
If `achieved` is `PreApplied.minKnown` aren't `DepsKnown` at this point? Is it that it's just the local state that is not `DepsKnown` so we want to include the `CommittedDeps`?
And if `full` is `DepsKnown` it covers all the ranges?
I know in `CheckStatus.sufficientFor` we are checking that the deps actually cover. Not sure if that happens here. Doesn't look like slicing actually checks.
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ // TODO (required): we could in principle be stale for future epochs we haven't witnessed yet. Ensure up to date epochs before finalising this application.
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+ executeRanges = safeStore.commandStore().redundantBefore().expectToExecute(txnId, executeAt, executeRanges);
+
+ if (executeRanges.isEmpty())
+ {
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Known required = PreApplied.minKnown;
+ Known requireExtra = required.subtract(command.known());
+ Ranges achieveRanges = full.sufficientFor(requireExtra, executeRanges);
+ Participants<?> participants = route.participants().slice(executeRanges, Minimal);
+
+ if (participants.isEmpty())
+ {
+ // we only coordinate this transaction, so being unable to retrieve its state does not imply any staleness
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
Review Comment:
`executeRanges` is our `executeRanges`, and it doesn't participate, ok, so it doesn't imply staleness because we didn't need to apply the side effects so we can't be missing anything we need going forward?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ // TODO (required): we could in principle be stale for future epochs we haven't witnessed yet. Ensure up to date epochs before finalising this application.
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+ executeRanges = safeStore.commandStore().redundantBefore().expectToExecute(txnId, executeAt, executeRanges);
+
+ if (executeRanges.isEmpty())
+ {
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Known required = PreApplied.minKnown;
+ Known requireExtra = required.subtract(command.known());
+ Ranges achieveRanges = full.sufficientFor(requireExtra, executeRanges);
+ Participants<?> participants = route.participants().slice(executeRanges, Minimal);
+
+ if (participants.isEmpty())
+ {
+ // we only coordinate this transaction, so being unable to retrieve its state does not imply any staleness
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Ranges staleRanges = executeRanges.subtract(achieveRanges);
+ Participants<?> staleParticipants = participants.slice(staleRanges, Minimal);
+ staleRanges = staleParticipants.toRanges();
+
+ if (staleRanges.isEmpty())
+ {
+ Invariants.checkState(achieveRanges.containsAll(participants));
+ return required;
+ }
+
+ safeStore.commandStore().markShardStale(safeStore, executeAt, staleRanges, true);
+ if (!staleRanges.containsAll(participants))
Review Comment:
So we return a non-null value if at least some of the ranges are not stale, meaning we need followup actions for whatever those ranges were?
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -347,61 +353,84 @@ public Void apply(SafeCommandStore safeStore)
{
SafeCommand safeCommand = safeStore.get(txnId, this, route);
Command command = safeCommand.current();
- if (command.saveStatus().phase.compareTo(Phase.Persist) >= 0)
- return null;
- Status propagate = achieved.propagate();
+ PartialTxn partialTxn = this.partialTxn;
+ PartialDeps partialDeps = this.partialDeps;
+ switch (command.saveStatus().phase)
+ {
+ case Persist: return updateDurability(safeStore, safeCommand);
+ case Cleanup: return null;
+ }
+
+ Known achieved = this.achieved;
+ if (isTruncated)
+ {
+ achieved = applyOrUpgradeTruncated(safeStore, safeCommand, command);
+ if (achieved == null)
+ return null;
+
+ if (achieved.executeAt.isDecided())
+ {
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAt);
+ if (partialTxn == null && this.full.saveStatus.known.definition.isKnown())
Review Comment:
`propagate` also does something similar across all local shards, why is `partialTxn` null here and how is what we do here different from what `propagate` does creating a `partialTxn` based on ` Known achieved = full.sufficientFor(participatingKeys, withQuorum);`?
##########
accord-core/src/main/java/accord/coordinate/ReadCoordinator.java:
##########
@@ -218,7 +212,18 @@ protected void finishOnExhaustion()
{
Invariants.checkState(!isDone);
if (failure == null)
- failure = new Exhausted(txnId, null);
+ {
Review Comment:
If failure is not null shouldn't it be added as a suppressed exception? Why drop it entirely?
##########
accord-core/src/main/java/accord/coordinate/Infer.java:
##########
@@ -168,18 +170,24 @@ public static Status invalidIfNotAtLeast(SafeCommandStore safeStore, TxnId txnId
return Status.NotDefined;
}
- public static boolean inferInvalidated(CheckStatus.WithQuorum withQuorum, Status invalidIfNotAtLeast, SaveStatus saveStatus, SaveStatus maxSaveStatus)
+ public static boolean inferInvalidated(CheckStatus.WithQuorum withQuorum, Status minInvalidIfNotAtLeast, Status maxInvalidIfNotAtLeast, SaveStatus saveStatus, SaveStatus maxSaveStatus, Ballot minPromised)
{
if (saveStatus == SaveStatus.Invalidated)
return true;
if (withQuorum != HasQuorum)
return false;
- // should not be possible to reach a quorum without finding the definition unless cleanup is in progress
- Invariants.checkState(saveStatus != SaveStatus.Accepted || maxSaveStatus.phase == Status.Phase.Cleanup);
+ // if everyone has witnessed a recovery, or everyone had invalidIfNotAtLeast, then we can rely on any inference;
+ // otherwise we might have raced with the original coordinator, and only witnessed part of their effects
+ Status invalidIfNotAtLeast = minPromised.compareTo(Ballot.ZERO) > 0 ? maxInvalidIfNotAtLeast : minInvalidIfNotAtLeast;
Review Comment:
Why does `minPromised.compareTo(Ballot.ZERO)` imply everyone witnessed a recovery? What does picking max vs min do here?
##########
accord-core/src/main/java/accord/impl/SimpleProgressLog.java:
##########
@@ -646,6 +649,8 @@ public void waiting(SafeCommand blockedBy, LocalExecution blockedUntil, Route<?>
// ensure we have a record to work with later; otherwise may think has been truncated
blockedBy.initialise();
+ if (blockedBy.current().has(blockedUntil.requires))
Review Comment:
Is it blocked by this or not? Why do `waiting` at all? To make sure we know we aren't waiting?
##########
accord-core/src/main/java/accord/impl/SimpleProgressLog.java:
##########
@@ -355,12 +355,7 @@ void run(SafeCommandStore safeStore, SafeCommand safeCommand)
return;
setProgress(Expected);
- if (fail == null && blockedUntil.isSatisfiedBy(success))
- {
- Command test = safeStore0.ifInitialised(txnId).current();
- Invariants.checkState(test.has(success), "Command %s was expected to have known %s, but had %s", test, success, test.known());
- record(success);
- }
+ Invariants.checkState(fail != null || !blockedUntil.isSatisfiedBy(success.propagates()));
Review Comment:
What changed that makes it so `FetchData` is guaranteed to not find that the transaction is `blockedUntil` is not satisfied (or `progress() != Investigating`)?
##########
accord-core/src/main/java/accord/coordinate/Infer.java:
##########
@@ -168,18 +170,24 @@ public static Status invalidIfNotAtLeast(SafeCommandStore safeStore, TxnId txnId
return Status.NotDefined;
}
- public static boolean inferInvalidated(CheckStatus.WithQuorum withQuorum, Status invalidIfNotAtLeast, SaveStatus saveStatus, SaveStatus maxSaveStatus)
+ public static boolean inferInvalidated(CheckStatus.WithQuorum withQuorum, Status minInvalidIfNotAtLeast, Status maxInvalidIfNotAtLeast, SaveStatus saveStatus, SaveStatus maxSaveStatus, Ballot minPromised)
{
if (saveStatus == SaveStatus.Invalidated)
return true;
if (withQuorum != HasQuorum)
return false;
- // should not be possible to reach a quorum without finding the definition unless cleanup is in progress
- Invariants.checkState(saveStatus != SaveStatus.Accepted || maxSaveStatus.phase == Status.Phase.Cleanup);
+ // if everyone has witnessed a recovery, or everyone had invalidIfNotAtLeast, then we can rely on any inference;
+ // otherwise we might have raced with the original coordinator, and only witnessed part of their effects
+ Status invalidIfNotAtLeast = minPromised.compareTo(Ballot.ZERO) > 0 ? maxInvalidIfNotAtLeast : minInvalidIfNotAtLeast;
+ if (invalidIfNotAtLeast == NotDefined)
+ return false;
- if (saveStatus.status.compareTo(invalidIfNotAtLeast) < 0)
+ // we could reach a quorum without finding the definition without having to have entered the cleanup phase if
Review Comment:
What is the cleanup phase? Is the switch to `maxSaveStatus` because it was a mistake to use just `saveStatus` earlier? Trying to puzzle through `mergeStatus` vs `mergeMaxStatus` in `CheckStatus` and `preferKnowledge` and what that might mean.
##########
accord-core/src/main/java/accord/local/Command.java:
##########
@@ -820,16 +916,16 @@ public static class Committed extends Accepted
private Committed(CommonAttributes common, SaveStatus status, Timestamp executeAt, Ballot promised, Ballot accepted, WaitingOn waitingOn)
{
- super(common, status, executeAt, promised, accepted);
+ super(common, status, promised, executeAt, accepted);
this.waitingOn = waitingOn;
Invariants.checkState(common.route().kind().isFullRoute(), "Expected a full route but given %s", common.route().kind());
- Invariants.checkState(waitingOn.deps.equals(common.partialDeps()), "Deps do not match; expected %s == %s", waitingOn.deps, common.partialDeps());
+ Invariants.checkState(waitingOn == WaitingOn.EMPTY || waitingOn.deps.equals(common.partialDeps()), "Deps do not match; expected %s == %s", waitingOn.deps, common.partialDeps());
Review Comment:
Is `WaitingOn.EMPTY` used instead of a `WaitingOn` with the correct deps? Should we stop using it that way?
##########
accord-core/src/main/java/accord/local/RedundantBefore.java:
##########
@@ -107,15 +126,54 @@ static RedundantStatus get(Entry entry, TxnId txnId, EpochSupplier executeAt)
return entry.get(txnId);
}
+ static PreBootstrapOrStale getAndMerge(Entry entry, @Nonnull PreBootstrapOrStale prev, TxnId txnId, EpochSupplier executeAt)
+ {
+ if (prev == PARTIALLY || entry == null || entry.outOfBounds(txnId, executeAt))
+ return prev;
+
+ boolean isPreBootstrapOrStale = entry.staleUntilAtLeast != null || entry.bootstrappedAt.compareTo(txnId) > 0;
+ return isPreBootstrapOrStale ? prev == POST_BOOTSTRAP ? PARTIALLY : FULLY
+ : prev == FULLY ? PARTIALLY : POST_BOOTSTRAP;
+ }
+
+ static Ranges validateSafeToRead(Entry entry, @Nonnull Ranges safeToRead, Timestamp bootstrapAt, Object ignore)
+ {
+ if (entry == null)
+ return safeToRead;
+
+ if (bootstrapAt.compareTo(entry.bootstrappedAt) < 0 || (entry.staleUntilAtLeast != null && bootstrapAt.compareTo(entry.staleUntilAtLeast) < 0))
Review Comment:
It really needs to be `staleUntilAtMost` for this to actually mean that this `bootstrapAt` is useful.
##########
accord-core/src/main/java/accord/local/Bootstrap.java:
##########
@@ -400,6 +400,9 @@ void maybeComplete()
if (!fetchedAndSafeToRead.containsAll(fetchCompleted ? fetched : valid))
return;
+ // normalise fetched and fetchedAndSafeToRead against remaining valid ranges before completion
+ fetched = fetched.slice(valid, Minimal);
+ fetchedAndSafeToRead = fetchedAndSafeToRead.slice(valid, Minimal);
Review Comment:
`fetchedAndSafeToRead` is already sliced by `valid` and `valid` has been sliced by the ranges that are invalidated so the ranges we need to retry are gone?
##########
accord-core/src/main/java/accord/local/Command.java:
##########
@@ -1322,8 +1421,9 @@ static Command.PreAccepted preaccept(Command command, CommonAttributes attrs, Ti
}
else if (command.status() == Status.AcceptedInvalidate && command.executeAt() == null)
{
+ // TODO (now): reconsider this special-casing
Command.Accepted accepted = command.asAccepted();
- return Command.Accepted.accepted(attrs, accepted.saveStatus(), executeAt, ballot, accepted.accepted());
+ return Command.Accepted.accepted(attrs, SaveStatus.enrich(accepted.saveStatus(), SaveStatus.PreAccepted.known), executeAt, ballot, accepted.accepted());
Review Comment:
These cases are all basically dueling coordinators trying to collect enough votes to pick an outcome (invalidated, `executeAt` of the invalidation)?
Why enrich here?
##########
accord-core/src/main/java/accord/local/CommandStore.java:
##########
@@ -472,14 +468,37 @@ final void markBootstrapping(SafeCommandStore safeStore, TxnId globalSyncId, Ran
// TODO (expected): we can immediately truncate dependencies locally once an exclusiveSyncPoint applies, we don't need to wait for the whole shard
public void markShardDurable(SafeCommandStore safeStore, TxnId globalSyncId, Ranges ranges)
{
- // TODO (now): IMPORTANT we should insert additional entries for each epoch our ownership differs
ranges = ranges.slice(safeStore.ranges().allUntil(globalSyncId.epoch()), Minimal);
RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, globalSyncId, TxnId.NONE);
setRedundantBefore(RedundantBefore.merge(redundantBefore, addRedundantBefore));
DurableBefore addDurableBefore = DurableBefore.create(ranges, globalSyncId, globalSyncId);
setDurableBefore(DurableBefore.merge(durableBefore, addDurableBefore));
}
+ // TODO (expected): we can immediately truncate dependencies locally once an exclusiveSyncPoint applies, we don't need to wait for the whole shard
+ public void markShardStale(SafeCommandStore safeStore, Timestamp staleSince, Ranges ranges, boolean isSincePrecise)
+ {
+ agent.onStale(staleSince, ranges);
+
+ Timestamp staleUntilAtLeast = staleSince;
+ if (isSincePrecise)
+ {
+ ranges = ranges.slice(safeStore.ranges().allAt(staleSince.epoch()), Minimal);
+ }
+ else
+ {
+ ranges = ranges.slice(safeStore.ranges().allSince(staleSince.epoch()), Minimal);
+ // make sure no in-progress bootstrap attempts will override the stale since for commands whose staleness bounds are unknown
Review Comment:
It would be interesting to know where commands with unbounded staleness come from. Seems heavyweight to invalidate all in progress bootstraps.
##########
accord-core/src/main/java/accord/local/CommandStore.java:
##########
@@ -534,30 +553,77 @@ public final void removeRedundantDependencies(Unseekables<?> participants, Waiti
}
return b;
}, builder, keyDeps, null, ignore -> false);
- RangeDeps rangeDeps = builder.deps.rangeDeps;
- // TODO (now): slice to only those ranges we own, maybe don't even construct rangeDeps.covering()
- redundantBefore().foldl(participants, (e, b, d, rs, pi, pj) -> {
- // TODO (now): foldlInt so we can track the lower rangeidx bound and not revisit unnecessarily
- boolean useBootstrap = e.isLocalRedundancyViaBootstrap();
+
+ class RangeState
+ {
+ final WaitingOn.Update builder;
int txnIdx;
+ Range range;
+ Map<Integer, Ranges> partiallyBootstrapping;
+
+ RangeState(WaitingOn.Update builder)
+ {
+ this.builder = builder;
+ }
+
+ boolean isFullyBootstrapping(int rangeTxnIdx)
+ {
+ if (partiallyBootstrapping == null)
+ partiallyBootstrapping = new HashMap<>();
+ Ranges prev = partiallyBootstrapping.get(rangeTxnIdx);
+ Ranges remaining = prev;
+ if (remaining == null) remaining = builder.deps.rangeDeps.ranges(rangeTxnIdx);
+ else Invariants.checkState(!remaining.isEmpty());
+ remaining = remaining.subtract(Ranges.of(range));
+ if (prev == null) Invariants.checkState(!remaining.isEmpty());
+ partiallyBootstrapping.put(txnIdx, remaining);
+ return remaining.isEmpty();
+ }
+ }
+ RangeDeps rangeDeps = builder.deps.rangeDeps;
+ // TODO (required, consider): slice to only those ranges we own, maybe don't even construct rangeDeps.covering()
+ redundantBefore().foldl(participants, (e, s, d, ps, pi, pj) -> {
+ // TODO (desired, efficiency): foldlInt so we can track the lower rangeidx bound and not revisit unnecessarily
+ WaitingOn.Update b = s.builder;
{
- int tmp = d.txnIds().find(useBootstrap ? e.bootstrappedAt : e.redundantBefore);
- txnIdx = tmp < 0 ? -1 - tmp : tmp;
+ int tmp = d.txnIds().find(e.redundantBefore);
+ s.txnIdx = tmp < 0 ? -1 - tmp : tmp;
}
- rangeDeps.forEach(participants, e.range, pi, pj, d, (d0, txnIdx0) -> {
- if (txnIdx0 < txnIdx)
- b.setAppliedOrInvalidatedRangeIdx(txnIdx0);
+ d.forEach(ps, e.range, pi, pj, b, s, (b0, s0, txnIdx) -> {
+ if (txnIdx < s0.txnIdx)
+ b0.setAppliedOrInvalidatedRangeIdx(txnIdx);
});
- return b;
- }, builder, rangeDeps, participants, ignore -> false);
+ if (e.bootstrappedAt.compareTo(e.redundantBefore) > 0)
Review Comment:
Are we checking bootstrap here because we intend to fulfill the dependency via bootstrap instead of via applying transactions? I am probably not understanding the differences between bootstrapping and redundancy which seem to overlap since bootstrap does set redundant before state.
And the redundant before checking is basically saying remove dependencies because they are redundant and have already been made durable?
##########
accord-core/src/main/java/accord/local/CommandStore.java:
##########
@@ -472,14 +468,37 @@ final void markBootstrapping(SafeCommandStore safeStore, TxnId globalSyncId, Ran
// TODO (expected): we can immediately truncate dependencies locally once an exclusiveSyncPoint applies, we don't need to wait for the whole shard
public void markShardDurable(SafeCommandStore safeStore, TxnId globalSyncId, Ranges ranges)
{
- // TODO (now): IMPORTANT we should insert additional entries for each epoch our ownership differs
ranges = ranges.slice(safeStore.ranges().allUntil(globalSyncId.epoch()), Minimal);
RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, globalSyncId, TxnId.NONE);
setRedundantBefore(RedundantBefore.merge(redundantBefore, addRedundantBefore));
DurableBefore addDurableBefore = DurableBefore.create(ranges, globalSyncId, globalSyncId);
setDurableBefore(DurableBefore.merge(durableBefore, addDurableBefore));
}
+ // TODO (expected): we can immediately truncate dependencies locally once an exclusiveSyncPoint applies, we don't need to wait for the whole shard
+ public void markShardStale(SafeCommandStore safeStore, Timestamp staleSince, Ranges ranges, boolean isSincePrecise)
Review Comment:
Worth explaining what it means to mark a shard stale, why we would do it, how `staleSince` is supposed to be derived, and what it means for `isSincePrecise` to be true/false.
##########
accord-core/src/main/java/accord/local/CommandStore.java:
##########
@@ -534,30 +553,77 @@ public final void removeRedundantDependencies(Unseekables<?> participants, Waiti
}
return b;
}, builder, keyDeps, null, ignore -> false);
- RangeDeps rangeDeps = builder.deps.rangeDeps;
- // TODO (now): slice to only those ranges we own, maybe don't even construct rangeDeps.covering()
- redundantBefore().foldl(participants, (e, b, d, rs, pi, pj) -> {
- // TODO (now): foldlInt so we can track the lower rangeidx bound and not revisit unnecessarily
- boolean useBootstrap = e.isLocalRedundancyViaBootstrap();
+
+ class RangeState
+ {
+ final WaitingOn.Update builder;
int txnIdx;
+ Range range;
+ Map<Integer, Ranges> partiallyBootstrapping;
+
+ RangeState(WaitingOn.Update builder)
+ {
+ this.builder = builder;
+ }
+
+ boolean isFullyBootstrapping(int rangeTxnIdx)
Review Comment:
Comment explaining what `isFullyBootstrapping` is?
##########
accord-core/src/main/java/accord/local/Status.java:
##########
@@ -457,16 +535,24 @@ public enum LogicalEpoch
*/
public enum Durability
{
- NotDurable, Local, Majority, Universal, DurableOrInvalidated;
+ NotDurable, Local, MajorityOrInvalidated, Majority, UniversalOrInvalidated, Universal;
public boolean isDurable()
{
- return compareTo(Majority) >= 0 && compareTo(DurableOrInvalidated) < 0;
+ return this == Majority || this == Universal;
}
public boolean isDurableOrInvalidated()
{
- return compareTo(Majority) >= 0;
+ return compareTo(MajorityOrInvalidated) >= 0;
+ }
+
+ public static Durability merge(Durability a, Durability b)
+ {
+ int c = a.compareTo(b);
+ if (c < 0) { Durability tmp = a; a = b; b = tmp; }
+ if (a == UniversalOrInvalidated && b == Majority) a = Universal;
Review Comment:
So if we learned the outcome we want to remove the `OrInvalidated` from `Universal`? Would be a good comment just to confirm that.
##########
accord-core/src/main/java/accord/local/Commands.java:
##########
@@ -1190,7 +1217,8 @@ static Command removeRedundantDependencies(SafeCommandStore safeStore, SafeComma
if (minWaitingOnTxnId != null && commandStore.hasLocallyRedundantDependencies(update.minWaitingOnTxnId(), current.executeAt(), current.route().participants()))
safeStore.commandStore().removeRedundantDependencies(current.route().participants(), update);
- if (isPreBootstrap) update.removeWaitingOn(redundant); // above operation not guaranteed to remove all redundant dependencies; transaction may
+ // if we are a range transaction, being redundant for this transaction does not imply we are redundant for all transactions
+ if (isPreBootstrap || redundant.domain().isRange()) update.removeWaitingOn(redundant);
Review Comment:
Really need help understanding `removeInvalidatedAppliedOrTruncated` vs `removeRedundantDependencies`
##########
accord-core/src/main/java/accord/local/RedundantBefore.java:
##########
@@ -47,19 +52,23 @@ public static RedundantBefore create(boolean inclusiveEnds, RoutingKey[] ends, E
}
}
+ public enum PreBootstrapOrStale { NOT_OWNED, FULLY, PARTIALLY, POST_BOOTSTRAP }
+
public static class Entry
{
public final Range range;
public final long startEpoch, endEpoch;
public final @Nonnull TxnId redundantBefore, bootstrappedAt;
+ public final @Nullable Timestamp staleUntilAtLeast;
Review Comment:
`redundantBefore`, `boostrappedAt` and `staleUntilAtLeast` should all have a brief description of what they mean.
In particular I don't think we document very often the concept of `redundancy` which is specifically global durability given how it is used in `RedundantBefore.create`.
`staleUntilAtLeast` is also a pretty thorny concept, so it's worth describing how we leverage knowledge of staleness during transaction scheduling such as deciding not to execute transactions that are for ranges we are stale and will have to bootstrap from and also we end up having staleness in the first place.
##########
accord-core/src/test/java/accord/topology/TopologyRandomizer.java:
##########
@@ -365,4 +367,17 @@ private boolean everyShardHasOverlaps(long prevEpoch, Shard[] in, Shard[] out)
Invariants.checkState (i == in.length && o == out.length);
return true;
}
+
+ public void onStale(Node.Id id, Timestamp sinceAtLeast, Ranges ranges)
Review Comment:
So for test purposes we don't expect there to be any stale ranges in the protocol? Which makes sense of course since we haven't added anything to allow nodes to fall behind and re-bootstrap yet.
##########
accord-maelstrom/src/main/java/accord/maelstrom/Cluster.java:
##########
@@ -344,4 +349,28 @@ public static void run(Id[] nodes, QueueSupplier queueSupplier, Consumer<Packet>
lookup.values().forEach(Node::shutdown);
}
}
+//
+// public List<TxnId> waitingOn(String txnIdString)
Review Comment:
Remove these?
##########
accord-core/src/main/java/accord/primitives/Range.java:
##########
@@ -155,6 +183,22 @@ public int compareTo(RoutableKey key)
}
return 0;
}
+
+ @Override
+ public int compareStartTo(RoutableKey key)
+ {
+ int c = start().compareTo(key);
+ if (c == 0 && !startInclusive) c = 1;
+ return c;
+ }
+
+ @Override
+ public int compareEndTo(RoutableKey key)
Review Comment:
Might be slightly DRYer to use a static function to share the start inclusive and end inclusive behaviors and maybe have a 3rd class for both start and end inclusive at the same time. Also should it be an error to request neither start nor end be inclusive?
##########
accord-core/src/test/java/accord/impl/list/ListRead.java:
##########
@@ -62,25 +63,25 @@ public ListRead(Function<? super CommandStore, AsyncExecutor> executor, Seekable
}
@Override
- public AsyncChain<Data> read(Seekable key, Txn.Kind kind, SafeCommandStore commandStore, Timestamp executeAt, DataStore store)
+ public AsyncChain<Data> read(Seekable key, Txn.Kind kind, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
{
+ // read synchronously, logically taking a snapshot, so we can impose our invariant of not reading the future
ListStore s = (ListStore)store;
- return executor.apply(commandStore.commandStore()).submit(() -> {
- ListData result = new ListData();
- switch (key.domain())
- {
- default: throw new AssertionError();
- case Key:
- Timestamped<int[]> data = s.get((Key)key);
- logger.trace("READ on {} at {} key:{} -> {}", s.node, executeAt, key, data);
- result.put((Key)key, data);
- break;
- case Range:
- for (Map.Entry<Key, Timestamped<int[]>> e : s.get((Range)key))
- result.put(e.getKey(), e.getValue());
- }
- return result;
- });
+ ListData result = new ListData();
Review Comment:
I guess I am a little confused either this is already running in the command store in which case we can just read, or it isn't and it needs to run in the command store for `ListStore` to be read from the right thread?
##########
accord-core/src/main/java/accord/utils/random/Suppliers.java:
##########
@@ -16,24 +16,40 @@
* limitations under the License.
*/
-package accord.local;
+package accord.utils.random;
-import accord.primitives.TxnId;
+import java.util.function.BooleanSupplier;
+import java.util.function.IntSupplier;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
-public interface CommandListener
+public class Suppliers
Review Comment:
Weird that this is a rename from `CommandListener.java` and also seems to be unused now?
##########
accord-core/src/main/java/accord/local/RedundantBefore.java:
##########
@@ -69,18 +78,26 @@ public static Entry reduce(Entry a, Entry b)
private static Entry merge(Range range, Entry a, Entry b)
{
+ if (a.startEpoch > b.endEpoch)
Review Comment:
In `RedundantBefore` it seems like we are discarding information about older epochs, but also we use information about those to drive txn execution and have the out of bounds check we use to return `NOT_OWNED`. Wondering if that is always safe.
##########
accord-core/src/main/java/accord/utils/SearchableRangeList.java:
##########
@@ -200,18 +201,18 @@ public <P1, P2, P3> int forEach(RoutableKey key, IndexedTriConsumer<P1, P2, P3>
start = floor = -2 - floor;
if (start < 0)
start = floor = 0;
- else if (ranges[start].end().compareTo(key) <= 0)
+ else if (ranges[start].compareEndTo(key) < 0)
++start;
}
- // Key cannot equal RoutingKey, so can ignore inclusivity/exclusivity of end bound
- return forEach(start, end, floor, key, forEachScanOrCheckpoint, forEachRange, p1, p2, p3, minIndex);
+ int bound = ranges[0].endInclusive() ? -1 : 0;
Review Comment:
It would take a thorough second pass to understand these changes (and the search as a whole), but if the test added to `RangeDepsTest` covers this issue then maybe it is less important. What exactly was the problem here with ignoring inclusivity/exclusivity of bounds?
##########
accord-core/src/main/java/accord/local/SaveStatus.java:
##########
@@ -188,24 +201,25 @@ public boolean isComplete()
}
}
+ // TODO (expected): merge Known only, and ensure 1:1 mapping so can reconstruct composite
// TODO (expected, testing): exhaustive testing, particularly around PreCommitted
public static SaveStatus get(Status status, Known known)
{
switch (status)
{
default: throw new AssertionError("Unexpected status: " + status);
- case NotDefined: return NotDefined;
- case PreAccepted: return PreAccepted;
+ case NotDefined: return known.executeAt.isDecided() ? PreCommitted : NotDefined;
+ case PreAccepted: return known.executeAt.isDecided() ? PreCommittedWithDefinition : PreAccepted;
Review Comment:
It's a bit unclear why these should be true based on just `executeAt` being decided. That means you have at some point communicated with someone who would have given you the definition?
Why are we upgrading the `Status` we know based on `Known`? If we got `Known` from somewhere doesn't that also include the `Status` associated with that `Known`?
##########
accord-core/src/test/java/accord/burn/BurnTest.java:
##########
@@ -196,24 +196,25 @@ static void burn(RandomSource random, TopologyFactory topologyFactory, List<Id>
RandomDelayQueue delayQueue = new Factory(random).get();
PropagatingPendingQueue queue = new PropagatingPendingQueue(failures, delayQueue);
RandomSource retryRandom = random.fork();
- ListAgent agent = new ListAgent(1000L, failures::add, retry -> {
+ Function<BiConsumer<Timestamp, Ranges>, ListAgent> agentSupplier = onStale -> new ListAgent(1000L, failures::add, retry -> {
Review Comment:
Why was it necessary to do all this just so `TopologyRandomizer` could get called for `onStale`? Why a `ListAgent` per node now and a `globalExecutor` copy per node?
##########
accord-core/src/main/java/accord/primitives/TxnId.java:
##########
@@ -155,4 +159,13 @@ public static TxnId minForEpoch(long epoch)
{
return new TxnId(epochMsb(epoch), 0, Id.NONE);
}
+
+ private static final Pattern PARSE = Pattern.compile("\\[(?<epoch>[0-9]+),(?<hlc>[0-9]+),(?<flags>[0-9]+),(?<node>[0-9]+)]");
+ public static TxnId parse(String txnIdString)
Review Comment:
There is a `toString` and `fromString` in `Timestamp` already, should this go there?
##########
accord-core/src/main/java/accord/messages/CheckStatus.java:
##########
@@ -193,39 +208,53 @@ public interface CheckStatusReply extends Reply
public static class CheckStatusOk implements CheckStatusReply
{
+ // TODO (required): tighten up constraints here to ensure we only report truncated when the range is Durable
public final Ranges truncated;
- public final Status invalidIfNotAtLeast;
+ public final Status minInvalidIfNotAtLeast, maxInvalidIfNotAtLeast;
Review Comment:
It's really worth explaining when/why we use the min and the max, and also the definition of invalid if not at least.
Also how/why that works in `Infer.inferInvalidated` and how it relates to also now tracking `minPromised` and `maxPromised`.
##########
accord-core/src/test/java/accord/impl/list/ListWrite.java:
##########
@@ -50,12 +53,15 @@ public ListWrite(Function<? super CommandStore, AsyncExecutor> executor)
}
@Override
- public AsyncChain<Void> apply(Seekable key, SafeCommandStore commandStore, Timestamp executeAt, DataStore store)
+ public AsyncChain<Void> apply(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
{
ListStore s = (ListStore) store;
if (!containsKey(key))
return Writes.SUCCESS;
- return executor.apply(commandStore.commandStore()).submit(() -> {
+ InMemorySafeCommandsForKey cfk = ((InMemoryCommandStore.InMemorySafeStore) safeStore).commandsForKey((Key) key);
Review Comment:
If this was missed in an implementation and is necessary it says this needs to be factored out into a common base method.
##########
accord-core/src/test/java/accord/impl/basic/SimulatedDelayedExecutorService.java:
##########
@@ -183,8 +193,13 @@ public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDela
@Override
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
{
- ScheduledTask<?> task = new ScheduledTask<>(sequenceNumber++, initialDelay, -delay, unit, Executors.callable(command));
+ ScheduledTask<?> task = new ScheduledTask<>(sequenceNumber.incrementAndGet(), initialDelay, -delay, unit, Executors.callable(command));
schedule(task, initialDelay, unit);
return task;
}
+
+ public AgentExecutor withAgent(Agent agent)
+ {
+ return new SimulatedDelayedExecutorService(pending, agent, sequenceNumber);
Review Comment:
This is kind of a weird way to clone with a shared pending and sequence number, but different agents. Would expect to have a factory that given an agent creates a new instance.
##########
accord-core/src/main/java/accord/messages/ReadTxnData.java:
##########
@@ -175,9 +175,12 @@ protected synchronized ReadNack apply(SafeCommandStore safeStore, SafeCommand sa
++waitingOnCount;
safeCommand.addListener(this);
- safeStore.progressLog().waiting(safeCommand, WaitingToExecute, null, readScope);
if (status == Committed) return null;
- else return NotCommitted;
+ else
+ {
+ safeStore.progressLog().waiting(safeCommand, ReadyToExecute, null, readScope);
Review Comment:
Hmm this code has been copy pasted a bit in other branches as well as existing places like `WaitOnCommit` and `WaitUntilApplied`. They all seem to have difderent versions of initialize `waitingOn`, add a listener, and maybe notify the progress log.
None of them do exactly the same thing.
##########
accord-core/src/main/java/accord/local/SaveStatus.java:
##########
@@ -259,6 +275,7 @@ public static SaveStatus enrich(SaveStatus status, Known known)
return status;
}
+ // TODO (expected): tighten up distinction of "preferKnowledge" and its interaction with CheckStatus
Review Comment:
Would it simplify to have `SaveStatus` contain enough field to simultaneously represent both `mergeStatus` and `mergeMaxStatus`?
##########
accord-core/src/main/java/accord/local/Status.java:
##########
@@ -233,20 +273,63 @@ public boolean canProposeInvalidation()
return deps.canProposeInvalidation() && executeAt.canProposeInvalidation() && outcome.canProposeInvalidation();
}
+ public Known subtract(Known subtract)
+ {
+ if (!subtract.isSatisfiedBy(this))
+ return Known.Nothing;
+
+ Definition newDefinition = subtract.definition.compareTo(definition) >= 0 ? DefinitionUnknown : definition;
+ KnownExecuteAt newExecuteAt = subtract.executeAt.compareTo(executeAt) >= 0 ? ExecuteAtUnknown : executeAt;
+ KnownDeps newDeps = subtract.deps.compareTo(deps) >= 0 ? DepsUnknown : deps;
+ Outcome newOutcome = subtract.outcome.compareTo(outcome) >= 0 ? Unknown : outcome;
+ return new Known(route, newDefinition, newExecuteAt, newDeps, newOutcome);
+ }
+
+
public String toString()
{
return Stream.of(definition.isKnown() ? "Definition" : null,
- executeAt.hasDecidedExecuteAt() ? "ExecuteAt" : null,
+ executeAt.isDecided() ? "ExecuteAt" : null,
deps.hasDecidedDeps() ? "Deps" : null,
outcome.isOrWasApply() ? "Outcome" : null
).filter(Objects::nonNull).collect(Collectors.joining(",", "[", "]"));
}
}
+ public enum KnownRoute
+ {
+ /**
+ * A route may or may not be known, but it may not cover (or even intersect) this shard.
+ * The route should be relied upon only if it is a FullRoute.
+ */
+ Maybe,
Review Comment:
If it's a full route then wouldn't this be `Full` and not `Maybe`?
##########
accord-core/src/main/java/accord/local/CommandStores.java:
##########
@@ -260,6 +292,21 @@ public String toString()
return IntStream.range(0, ranges.length).mapToObj(i -> epochs[i] + ": " + ranges[i])
.collect(Collectors.joining(", "));
}
+
+ public @Nullable EpochSupplier latestEpochWithNewParticipants(long sinceEpoch, Unseekables<?> keysOrRanges)
+ {
+ int i = floorIndex(sinceEpoch);
+ long latest = -1;
+ Ranges existing = i < 0 ? Ranges.EMPTY : ranges[i];
+ while (++i < ranges.length)
+ {
+ if (ranges[i].subtract(existing).intersects(keysOrRanges))
+ latest = epochs[i];
+ existing = existing.with(ranges[i]);
+ }
+
+ return latest == -1 ? null : constant(latest);
Review Comment:
This doesn't get called often, but the epoch supplier could be preallocated in an array like epochs.
##########
accord-core/src/test/java/accord/impl/basic/NodeSink.java:
##########
@@ -100,30 +95,22 @@ public void reply(Id replyToNode, ReplyContext replyContext, Reply reply)
private boolean maybeEnqueue(Node.Id to, long id, Message message, SafeCallback callback)
{
- Runnable task = () -> {
- Packet packet;
- if (message instanceof Reply) packet = new Packet(self, to, id, (Reply) message);
- else packet = new Packet(self, to, id, (Request) message);
- parent.add(packet, networkJitterNanos(to), TimeUnit.NANOSECONDS);
- };
+ Link link = parent.links.apply(self, to);
if (to.equals(self) || lookup.apply(to) == null /* client */)
{
- parent.messageListener.onMessage(Action.DELIVER, self, to, id, message);
- task.run();
+ deliver(to, id, message, link);
Review Comment:
Conversation earlier mentioned making it possible to have chaos with messages to yourself, but this delviers before checking the link action?
##########
accord-core/src/main/java/accord/messages/CheckStatus.java:
##########
@@ -409,22 +471,57 @@ public Known sufficientFor(Participants<?> participants, WithQuorum withQuorum)
return sufficientFor(participants, saveStatus, route, partialTxn, committedDeps, writes, result);
}
+ // it is assumed that we are invoking this for a transaction that will execute;
+ // the result may be erroneous if the transaction is invalidated, as logically this can apply to all ranges
+ public Ranges sufficientFor(Known required, Ranges expect)
Review Comment:
Maybe rename to `subRangesSufficientFor` or `rangesSufficientFor`?
##########
accord-core/src/main/java/accord/local/RedundantBefore.java:
##########
@@ -47,19 +52,23 @@ public static RedundantBefore create(boolean inclusiveEnds, RoutingKey[] ends, E
}
}
+ public enum PreBootstrapOrStale { NOT_OWNED, FULLY, PARTIALLY, POST_BOOTSTRAP }
+
public static class Entry
{
public final Range range;
public final long startEpoch, endEpoch;
public final @Nonnull TxnId redundantBefore, bootstrappedAt;
+ public final @Nullable Timestamp staleUntilAtLeast;
- public Entry(Range range, long startEpoch, long endEpoch, @Nonnull TxnId redundantBefore, @Nonnull TxnId bootstrappedAt)
+ public Entry(Range range, long startEpoch, long endEpoch, @Nonnull TxnId redundantBefore, @Nonnull TxnId bootstrappedAt, @Nullable Timestamp staleUntilAtLeast)
Review Comment:
We have stale until at least, but not when the staleness starts. Maybe it doesn't matter and by the time we discover the staleness we will have executed everything preceding the thing that triggered discovering the staleness.
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
+ {
+ // if our peers have truncated this command, then either:
+ // 1) we have already applied it locally; 2) the command doesn't apply locally; 3) we are stale; or 4) the command is invalidated
+ if (command.saveStatus().isUninitialised())
+ return null; // TODO (expected): maybe stale?
+
+ if (command.hasBeen(PreApplied))
+ {
+ updateDurability(safeStore, safeCommand);
+ return null;
+ }
+
+ if (Infer.safeToCleanup(safeStore, command, route, full.executeAt))
+ {
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ // we're now at least partially stale, but let's first see if we can progress this shard, or we can do so in part
+ Timestamp executeAt = command.executeAtIfKnown(full.executeAtIfKnown());
+ if (executeAt == null)
+ {
+ // we don't even know the execution time, so we cannot possibly proceed besides erasing the command state and marking ourselves stale
+ // TODO (required): we could in principle be stale for future epochs we haven't witnessed yet. Ensure up to date epochs before finalising this application.
+ safeStore.commandStore().markShardStale(safeStore, txnId, route.participants().toRanges(), false);
+ Commands.setErased(safeStore, safeCommand);
+ return null;
+ }
+
+ Ranges executeRanges = safeStore.ranges().allBetween(txnId, executeAt);
+ executeRanges = safeStore.commandStore().redundantBefore().expectToExecute(txnId, executeAt, executeRanges);
+
+ if (executeRanges.isEmpty())
+ {
+ // TODO (expected): we might prefer to adopt Redundant status, and permit ourselves to later accept the result of the execution and/or definition
+ Commands.setTruncatedApply(safeStore, safeCommand, route);
+ return null;
+ }
+
+ Known required = PreApplied.minKnown;
+ Known requireExtra = required.subtract(command.known());
+ Ranges achieveRanges = full.sufficientFor(requireExtra, executeRanges);
Review Comment:
`sufficientFor` is commented to say it only returns the correct answer if the transaction is not invalidated? We end up using this if `isTruncated` is true where it can be truncated but that doesn't establish whether it was invalidated.
Seems like `isTruncated` can be true, but it can also be true that the outcome is known to be invalidated.
##########
accord-core/src/main/java/accord/messages/ReadData.java:
##########
@@ -146,8 +146,8 @@ void read(SafeCommandStore safeStore, Timestamp executeAt, PartialTxn txn)
{
CommandStore unsafeStore = safeStore.commandStore();
Ranges unavailable = safeStore.ranges().unsafeToReadAt(executeAt);
-
- txn.read(safeStore, executeAt).begin((next, throwable) -> {
+ // TODO (required): do we need to check unavailable again on completion, or throughout execution?
Review Comment:
Shouldn't be an issue because you have it skip the unavailable ranges when reading now, and things shouldn't become unavailable while this read is occurring?
##########
accord-core/src/main/java/accord/local/Commands.java:
##########
@@ -1074,18 +1095,23 @@ else if (cur.isAtLeast(until))
}
else if ((directlyBlockedOnCommit = waitingOn.nextWaitingOnCommit()) != null)
{
- push(directlyBlockedOnCommit, LocalExecution.ReadyToExclude);
+ push(directlyBlockedOnCommit, WaitingToExecute);
prevSafe = curSafe;
}
else
{
- if (cur.hasBeen(Committed) && !cur.is(ReadyToExecute) && cur.saveStatus() != Applying && !cur.asCommitted().isWaitingOnDependency())
+ if (cur.hasBeen(Committed))
{
- if (!maybeExecute(safeStore, curSafe, false, false))
- throw new AssertionError("Is able to Apply, but has not done so");
- // loop and re-test the command's status; we may still want to notify blocking, esp. if not homeShard
- continue;
+ if (!cur.is(ReadyToExecute) && cur.saveStatus() != Applying && !cur.asCommitted().isWaitingOnDependency())
+ {
+ if (!maybeExecute(safeStore, curSafe, false, false))
+ throw new AssertionError("Is able to Apply, but has not done so");
+ // loop and re-test the command's status; we may still want to notify blocking, esp. if not homeShard
+ continue;
+ }
}
+ else if (!cur.hasBeen(PreCommitted))
+ until = ReadyToExclude;
Review Comment:
`LocalExecution` enum needs explanations for each of the values. I was trying to figure out how `ReadyToExclude` is used. Is it something that comes along with the various `PreCommitted` states?
When they are ready to exclude does that mean the plumbing might have enough information to remove them as a dependency and this is to give us something to block until? A comment here would go a long way towards explaining why we want to change the `until`
##########
accord-core/src/main/java/accord/messages/CheckStatus.java:
##########
@@ -397,10 +443,26 @@ public CheckStatusOk merge(CheckStatusOk that)
Writes writes = (fullMax.writes != null ? fullMax : fullMin).writes;
Result result = (fullMax.result != null ? fullMax : fullMin).result;
- return new CheckStatusOkFull(max.truncated, max.invalidIfNotAtLeast, max.saveStatus, max.maxSaveStatus, max.promised, max.accepted, max.executeAt, max.isCoordinating, max.durability, max.route,
+ return new CheckStatusOkFull(max.truncated, max.minInvalidIfNotAtLeast, max.maxInvalidIfNotAtLeast, max.saveStatus, max.maxSaveStatus, max.minPromised, max.maxPromised, max.accepted, max.executeAt, max.isCoordinating, max.durability, max.route,
max.homeKey, partialTxn, committedDeps, writes, result);
}
+ public CheckStatusOkFull merge(@Nonnull Route<?> route)
+ {
+ Route<?> mergedRoute = Route.merge((Route)this.route, route);
+ if (mergedRoute == this.route)
+ return this;
+ return new CheckStatusOkFull(truncated, minInvalidIfNotAtLeast, maxInvalidIfNotAtLeast, saveStatus, maxSaveStatus, minPromised, maxPromised, accepted, executeAt,
+ isCoordinating, durability, mergedRoute, homeKey, partialTxn, committedDeps, writes, result);
+ }
+
+ public CheckStatusOkFull merge(@Nonnull Durability durability)
+ {
+ durability = Durability.merge(durability, this.durability);
+ return new CheckStatusOkFull(truncated, minInvalidIfNotAtLeast, maxInvalidIfNotAtLeast, saveStatus, maxSaveStatus, minPromised, maxPromised, accepted, executeAt,
+ isCoordinating, durability, route, homeKey, partialTxn, committedDeps, writes, result);
+ }
+
public Known sufficientFor(Participants<?> participants, WithQuorum withQuorum)
Review Comment:
Maybe also rename to `knownFor`
##########
accord-core/src/main/java/accord/local/Status.java:
##########
@@ -233,20 +273,63 @@ public boolean canProposeInvalidation()
return deps.canProposeInvalidation() && executeAt.canProposeInvalidation() && outcome.canProposeInvalidation();
}
+ public Known subtract(Known subtract)
+ {
+ if (!subtract.isSatisfiedBy(this))
+ return Known.Nothing;
+
+ Definition newDefinition = subtract.definition.compareTo(definition) >= 0 ? DefinitionUnknown : definition;
+ KnownExecuteAt newExecuteAt = subtract.executeAt.compareTo(executeAt) >= 0 ? ExecuteAtUnknown : executeAt;
+ KnownDeps newDeps = subtract.deps.compareTo(deps) >= 0 ? DepsUnknown : deps;
+ Outcome newOutcome = subtract.outcome.compareTo(outcome) >= 0 ? Unknown : outcome;
+ return new Known(route, newDefinition, newExecuteAt, newDeps, newOutcome);
+ }
+
+
public String toString()
{
return Stream.of(definition.isKnown() ? "Definition" : null,
- executeAt.hasDecidedExecuteAt() ? "ExecuteAt" : null,
+ executeAt.isDecided() ? "ExecuteAt" : null,
deps.hasDecidedDeps() ? "Deps" : null,
outcome.isOrWasApply() ? "Outcome" : null
).filter(Objects::nonNull).collect(Collectors.joining(",", "[", "]"));
}
}
+ public enum KnownRoute
+ {
+ /**
+ * A route may or may not be known, but it may not cover (or even intersect) this shard.
+ * The route should be relied upon only if it is a FullRoute.
+ */
+ Maybe,
+
+ /**
+ * A route is known that covers the ranges this shard participates in.
+ * Note that if the status is less than Committed, this may not be the final set of owned ranges,
+ * and the route may not cover whatever this is decided as.
+ *
+ * This status primarily exists to communicate semantically to the reader.
+ */
+ Covering,
Review Comment:
If `Covering` doesn't actually establish it is `Covering` it's just a random partial route, then should there just be `Maybe`?
##########
accord-core/src/main/java/accord/messages/CheckStatus.java:
##########
@@ -409,22 +471,57 @@ public Known sufficientFor(Participants<?> participants, WithQuorum withQuorum)
return sufficientFor(participants, saveStatus, route, partialTxn, committedDeps, writes, result);
}
+ // it is assumed that we are invoking this for a transaction that will execute;
+ // the result may be erroneous if the transaction is invalidated, as logically this can apply to all ranges
+ public Ranges sufficientFor(Known required, Ranges expect)
+ {
+ Invariants.checkState(saveStatus != SaveStatus.Invalidated);
+ Known have = saveStatus.known;
+ if (!required.isSatisfiedBy(have))
+ return Ranges.EMPTY;
+
+ Ranges result = expect;
+ if (required.deps.hasDecidedDeps())
+ result = result.slice(committedDeps.covering, Minimal);
+
+ if (required.definition.isKnown())
+ result = result.slice(partialTxn.covering(), Minimal);
+
+ return result;
+ }
+
private static Known sufficientFor(Participants<?> participants, SaveStatus saveStatus, Route<?> route, PartialTxn partialTxn, PartialDeps committedDeps, Writes writes, Result result)
{
- Status.Definition definition = saveStatus.known.definition;
+ Known known = saveStatus.known;
+ Status.KnownRoute knownRoute;
Review Comment:
I think at one point we are going to always use the full route, are back to sending partial routes?
##########
accord-core/src/main/java/accord/local/RedundantBefore.java:
##########
@@ -224,14 +347,28 @@ protected boolean equals(Entry a, Entry b)
return a.equalsIgnoreRange(b);
}
+ @Override
+ public void append(RoutingKey start, @Nullable Entry value, BiFunction<Entry, Entry, Entry> reduce)
Review Comment:
Comment explaining what the override is doing? I can sort of tell it's modifying after the append if the append actually ended up increasing the size of `values` and modifying the existing tail entry to stop before the newly added entry, but it's also not super clear what `value` is and why it's `Nullable`.
It seems like the key is `start` and value has a begin and end range, but we use null values to create some kind of end entry? It's not super clear how that is supposed to work or is used and whether it's specific to `DurableBefore` or `ReducingIntervalMap`.
##########
accord-core/src/main/java/accord/coordinate/FetchData.java:
##########
@@ -415,14 +444,92 @@ public Void apply(SafeCommandStore safeStore)
break;
}
+ return updateDurability(safeStore, safeCommand);
+ }
+
+ // if can only propagate Truncated, we might be stale; try to upgrade the
+ private Known applyOrUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, Command command)
Review Comment:
The `apply` in this name means we apply the truncation locally? Upgrade to what exactly?
This also seems to try to cleanup as well.
##########
accord-core/src/main/java/accord/primitives/Participants.java:
##########
@@ -31,6 +31,42 @@ public interface Participants<K extends Unseekable> extends Unseekables<K>
@Override
Participants<K> slice(Ranges ranges, Slice slice);
Participants<K> with(Participants<K> with);
+ Participants<K> subtract(Ranges ranges);
+
+ Ranges toRanges();
+
+ static boolean isParticipants(Unseekables<?> unseekables)
+ {
+ switch (unseekables.kind())
+ {
+ default: throw new AssertionError("Unhandled Unseekables.Kind: " + unseekables.kind());
+ case FullKeyRoute:
+ case FullRangeRoute:
+ case PartialKeyRoute:
+ case PartialRangeRoute:
+ return Route.castToRoute(unseekables).isParticipatingHomeKey();
Review Comment:
Hmm I think there is some confusion around what `isParticipatingHomeKey` means. The comment in `Route` says it means the home key is not participating and is not included in the array, but I have made changes to `AbstractKeys.toRoutingKeysArray` that I think sets `isParticipatingHomeKey` to true if the key is already in the keys.
Here `isParticipants` seems to be conditional on whether the home key participates or not and if it's true then it is the participants which makes sense because the home key is one of the participants. It's placed in the collection either way.
So is the comment on `Route` what is wrong?
##########
accord-core/src/main/java/accord/local/Status.java:
##########
@@ -507,7 +593,8 @@ public static <T> T max(List<T> list, Function<T, Status> getStatus, Function<T,
Ballot accepted = getAccepted.apply(item);
boolean update = max == null
|| maxStatus.phase.compareTo(status.phase) < 0
- || (status.phase.equals(Phase.Accept) && maxAccepted.compareTo(accepted) < 0);
+ || (status.phase == Accept && maxAccepted.compareTo(accepted) < 0)
+ || status.phase != Accept && maxStatus.phase == status.phase && maxStatus.compareTo(status) > 0;
Review Comment:
Could/should this reuse the other static max function below that seems to do a lot of the same comparisons?
##########
accord-core/src/main/java/accord/local/Status.java:
##########
@@ -172,17 +181,48 @@ public Known with(Outcome newOutcome)
{
if (outcome == newOutcome)
return this;
- return new Known(definition, executeAt, deps, newOutcome);
+ return new Known(route, definition, executeAt, deps, newOutcome);
}
public Known with(KnownDeps newDeps)
{
if (deps == newDeps)
return this;
- return new Known(definition, executeAt, newDeps, outcome);
+ return new Known(route, definition, executeAt, newDeps, outcome);
}
- public Status propagate()
+ // TODO (expected): merge propagates and propagatesStatus
+ public Known propagates()
Review Comment:
Why would `Known` not just return itself? With `isSatisfiedBy` shouldn't that work?
##########
accord-core/src/main/java/accord/primitives/Participants.java:
##########
@@ -31,6 +31,42 @@ public interface Participants<K extends Unseekable> extends Unseekables<K>
@Override
Participants<K> slice(Ranges ranges, Slice slice);
Participants<K> with(Participants<K> with);
+ Participants<K> subtract(Ranges ranges);
+
+ Ranges toRanges();
+
+ static boolean isParticipants(Unseekables<?> unseekables)
+ {
+ switch (unseekables.kind())
+ {
+ default: throw new AssertionError("Unhandled Unseekables.Kind: " + unseekables.kind());
+ case FullKeyRoute:
+ case FullRangeRoute:
+ case PartialKeyRoute:
+ case PartialRangeRoute:
+ return Route.castToRoute(unseekables).isParticipatingHomeKey();
+ case RoutingRanges:
+ case RoutingKeys:
+ return true;
Review Comment:
This seems like it's just taken on faith that nothing added the home key?
##########
accord-core/src/main/java/accord/local/Commands.java:
##########
@@ -1006,13 +1026,14 @@ public void accept(SafeCommandStore safeStore)
{
Review Comment:
#Above I don't get how status works with `txnIds[depth]` referring to the top of the stack, but `prevSafe` refers to one below. We pass an `executeAt` epoch supplier from `prevSafe` but the `txnId` and `participants` come from the deps of `prevSafe`. Aren't we querying about the status of of `curSafe`, is it more of a `toEpoch` or something than an `executeAt`? Would be clearer to call out what `executeAt` epoch is expected here.
I guess for `prevSafe` this is the intersecting set of `participants` and not the full set? Should probably be called intersecting participants or maybe not even be a `Participants`? Are there partial vs full participants?
Also `curSafe` and `prevSafe` are not good names. `Safe` could be dropped and instead of `cur` and `prev` it would be clearer to be something like depender and dependee or something.
##########
accord-core/src/main/java/accord/messages/CheckStatus.java:
##########
@@ -409,22 +471,57 @@ public Known sufficientFor(Participants<?> participants, WithQuorum withQuorum)
return sufficientFor(participants, saveStatus, route, partialTxn, committedDeps, writes, result);
}
+ // it is assumed that we are invoking this for a transaction that will execute;
+ // the result may be erroneous if the transaction is invalidated, as logically this can apply to all ranges
+ public Ranges sufficientFor(Known required, Ranges expect)
+ {
+ Invariants.checkState(saveStatus != SaveStatus.Invalidated);
+ Known have = saveStatus.known;
+ if (!required.isSatisfiedBy(have))
+ return Ranges.EMPTY;
+
+ Ranges result = expect;
+ if (required.deps.hasDecidedDeps())
+ result = result.slice(committedDeps.covering, Minimal);
+
+ if (required.definition.isKnown())
+ result = result.slice(partialTxn.covering(), Minimal);
+
+ return result;
+ }
+
private static Known sufficientFor(Participants<?> participants, SaveStatus saveStatus, Route<?> route, PartialTxn partialTxn, PartialDeps committedDeps, Writes writes, Result result)
Review Comment:
Can we renamed this to `knownFor`? It's not blanket sufficiency it's constructing a `Known` which is then be checked for things
##########
accord-core/src/main/java/accord/local/Status.java:
##########
@@ -530,6 +617,11 @@ public static <T> T max(T a, Status statusA, Ballot acceptedA, T b, Status statu
return b;
}
+ public static Status simpleMin(Status a, Status b)
Review Comment:
Simple doesn't really tell you why you would use this and what the advantages/disadvantages are compared to the more complex `max` versions.
--
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