You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2025/03/12 18:12:36 UTC
(cassandra-accord) branch trunk updated: Fix: - Fast path stable commits must not override recovery propose/commit decisions regarding visibility of a transaction - RejectBefore must mergeMax, not merge, to ensure we maintain epoch and hlc increasing independently - Bad commitInvalidate decision - consistent filtering for touches and stillTouches - ensure TRUNCATE_BEFORE implies SHARD_APPLIED - TopologyManager.unsyncedOnly off-by-one error - DurabilityQueue should not retry SyncPointErased - handle rare case of no deps bu [...]
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
The following commit(s) were added to refs/heads/trunk by this push:
new 62b6d688 Fix: - Fast path stable commits must not override recovery propose/commit decisions regarding visibility of a transaction - RejectBefore must mergeMax, not merge, to ensure we maintain epoch and hlc increasing independently - Bad commitInvalidate decision - consistent filtering for touches and stillTouches - ensure TRUNCATE_BEFORE implies SHARD_APPLIED - TopologyManager.unsyncedOnly off-by-one error - DurabilityQueue should not retry SyncPointErased - handle rare [...]
62b6d688 is described below
commit 62b6d6883b689236efb375b6efa7f2350d8d771d
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Wed Mar 5 14:09:34 2025 +0000
Fix:
- Fast path stable commits must not override recovery propose/commit decisions regarding visibility of a transaction
- RejectBefore must mergeMax, not merge, to ensure we maintain epoch and hlc increasing independently
- Bad commitInvalidate decision
- consistent filtering for touches and stillTouches
- ensure TRUNCATE_BEFORE implies SHARD_APPLIED
- TopologyManager.unsyncedOnly off-by-one error
- DurabilityQueue should not retry SyncPointErased
- handle rare case of no deps but none needed
- not updating CFK synchronously on recovery, which can lead to erroneous recovery decisions for other transactions
- Don't return partial read response when one commandStore rejects the commit
- Filter touches/stillTouches consistently
- WaitingState computeLowEpoch must use hasTouched to handle historic key with no route
- Propagate TopologyRetiredException
- InMemoryJournal partial compaction incorrectly appending new diffs to end, rather than position of input
- Incorrect logic ensuring we save applyAt with participants+saveStatus to ensure expunging can be done correctly
- CommandChange should not pre-filter cleanup by participants presence
- Cleanup.filter should not modify EXPUNGE; we obviously still want to expunge Invalidated
- Remove invariant that is incompatible with partial compaction
Improve:
- Use format parameters to defer building Invariants.requireArgument string
- streamline RedundantStatus/RedundantBefore
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20427
---
.../java/accord/coordinate/ExecuteSyncPoint.java | 23 +-
.../main/java/accord/coordinate/ExecuteTxn.java | 6 +-
.../main/java/accord/coordinate/Invalidate.java | 34 ++-
.../src/main/java/accord/impl/AbstractLoader.java | 1 -
.../src/main/java/accord/impl/CommandChange.java | 37 ++-
.../java/accord/impl/InMemoryCommandStore.java | 33 ++-
.../java/accord/impl/progresslog/WaitingState.java | 6 +-
.../src/main/java/accord/local/Cleanup.java | 76 +++----
.../src/main/java/accord/local/Command.java | 22 --
.../src/main/java/accord/local/CommandStore.java | 30 ++-
.../src/main/java/accord/local/Commands.java | 84 ++++---
.../src/main/java/accord/local/PreLoadContext.java | 6 +-
.../main/java/accord/local/RedundantBefore.java | 219 ++++++++++--------
.../main/java/accord/local/RedundantStatus.java | 252 +++++++++++++--------
.../src/main/java/accord/local/RejectBefore.java | 20 +-
.../main/java/accord/local/SafeCommandStore.java | 6 +-
.../main/java/accord/local/StoreParticipants.java | 38 +++-
.../main/java/accord/local/cfk/CommandsForKey.java | 1 -
.../src/main/java/accord/local/cfk/Pruning.java | 2 +-
.../accord/local/durability/DurabilityQueue.java | 5 +-
.../src/main/java/accord/messages/Propagate.java | 15 +-
.../src/main/java/accord/messages/ReadData.java | 13 +-
.../main/java/accord/messages/SetShardDurable.java | 7 +-
.../main/java/accord/messages/StableThenRead.java | 6 +-
.../src/main/java/accord/primitives/Deps.java | 24 +-
.../src/main/java/accord/primitives/KeyDeps.java | 9 +
.../main/java/accord/primitives/PartialDeps.java | 6 +
.../src/main/java/accord/primitives/RangeDeps.java | 8 +
.../main/java/accord/primitives/SaveStatus.java | 5 -
.../src/main/java/accord/primitives/TxnId.java | 6 +
.../main/java/accord/topology/TopologyManager.java | 11 +-
.../src/main/java/accord/utils/ArrayBuffers.java | 107 +++++++++
.../src/main/java/accord/utils/SortedArrays.java | 80 +++++++
.../src/test/java/accord/impl/basic/Cluster.java | 4 +-
.../accord/impl/basic/DelayedCommandStores.java | 2 +
.../java/accord/impl/basic/InMemoryJournal.java | 56 +++--
.../java/accord/local/RedundantBeforeTest.java | 72 +++---
accord-core/src/test/resources/burn-logback.xml | 1 +
38 files changed, 832 insertions(+), 501 deletions(-)
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java b/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
index 398f34e6..0f37728c 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
@@ -47,14 +47,16 @@ import accord.utils.SortedListMap;
import accord.utils.UnhandledEnum;
import accord.utils.WrappableException;
import accord.utils.async.AsyncResult;
+import accord.utils.async.AsyncResults;
import accord.utils.async.AsyncResults.SettableResult;
import static accord.coordinate.CoordinationAdapter.Adapters.exclusiveSyncPoint;
+import static accord.primitives.Status.Durability.Majority;
+import static accord.primitives.Status.Durability.UniversalOrInvalidated;
import static accord.topology.Topologies.SelectNodeOwnership.SHARE;
public class ExecuteSyncPoint extends SettableResult<DurabilityResult> implements Callback<ReadReply>
{
-
public static class SyncPointErased extends Throwable implements WrappableException<SyncPointErased>
{
public SyncPointErased() {}
@@ -184,7 +186,11 @@ public class ExecuteSyncPoint extends SettableResult<DurabilityResult> implement
if (result.achievedRemote == SyncRemote.All)
{
node.configService().reportEpochRetired(syncPoint.route.toRanges(), syncPoint.syncId.epoch() - 1);
- node.send(tracker.nodes(), new SetShardDurable(syncPoint));
+ node.send(tracker.nodes(), new SetShardDurable(syncPoint, UniversalOrInvalidated));
+ }
+ else if (result.achievedRemote == SyncRemote.Quorum)
+ {
+ node.send(tracker.nodes(), new SetShardDurable(syncPoint, Majority));
}
node.durability().report(result);
trySuccess(result);
@@ -227,9 +233,16 @@ public class ExecuteSyncPoint extends SettableResult<DurabilityResult> implement
public static AsyncResult<DurabilityResult> coordinate(Node node, Function<Topologies, Set<Node.Id>> excludeSuccess, SyncPoint<Range> exclusiveSyncPoint, AgentExecutor executor, int attempt)
{
- ExecuteSyncPoint coordinate = new ExecuteSyncPoint(node, exclusiveSyncPoint, excludeSuccess, executor, attempt);
- coordinate.start();
- return coordinate;
+ try
+ {
+ ExecuteSyncPoint coordinate = new ExecuteSyncPoint(node, exclusiveSyncPoint, excludeSuccess, executor, attempt);
+ coordinate.start();
+ return coordinate;
+ }
+ catch (Throwable t)
+ {
+ return AsyncResults.failure(t);
+ }
}
protected void start()
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java b/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
index 02a3d451..ac0eb391 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteTxn.java
@@ -25,6 +25,7 @@ import accord.api.Result;
import accord.api.Timeouts;
import accord.local.CommandStore;
import accord.local.Commands;
+import accord.local.Commands.CommitOutcome;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.SafeCommand;
@@ -70,7 +71,6 @@ import static accord.messages.Commit.Kind.StableWithTxnAndDeps;
import static accord.messages.ReadData.CommitOrReadNack.Waiting;
import static accord.primitives.SaveStatus.Stable;
import static accord.primitives.Status.Phase.Execute;
-import static accord.primitives.Status.PreAccepted;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
// TODO (expected): return Waiting from ReadData if not ready to execute, and do not submit more than one speculative retry in this case
@@ -295,9 +295,9 @@ public class ExecuteTxn extends ReadCoordinator<ReadReply> implements Timeouts.T
@Override
protected CommitOrReadNack apply(SafeCommandStore safeStore, SafeCommand safeCommand, StoreParticipants participants)
{
- if (txnId.hasPrivilegedCoordinator() && path == FAST && (!safeCommand.current().promised().equals(Ballot.ZERO) || safeCommand.current().status() != PreAccepted))
+ if (CommitOutcome.Rejected == Commands.commit(safeStore, safeCommand, participants, Stable, Ballot.ZERO, txnId, route, txn, executeAt, stableDeps, commitKind()))
return CommitOrReadNack.Rejected;
- Commands.commit(safeStore, safeCommand, participants, Stable, Ballot.ZERO, txnId, route, txn, executeAt, stableDeps, commitKind());
+
return super.apply(safeStore, safeCommand, participants);
}
diff --git a/accord-core/src/main/java/accord/coordinate/Invalidate.java b/accord-core/src/main/java/accord/coordinate/Invalidate.java
index fb74ffa1..47cb19b7 100644
--- a/accord-core/src/main/java/accord/coordinate/Invalidate.java
+++ b/accord-core/src/main/java/accord/coordinate/Invalidate.java
@@ -229,28 +229,22 @@ public class Invalidate implements Callback<InvalidateReply>
if (max != maxNotTruncated || max.maxStatus != max.maxKnowledgeStatus)
{
+ // note: this may all be out of date given other evolutions (and it was previously buggy)
+ // TODO (required): revisit this logic some more
+ // this exists because if we have truncated a range we cannot seek votes to advance the state machine, so we may not be able to reach a quorum
+ // but, equally, we may not propagate the GC point to all replicas.
+ // the current logic has pitfalls when *some* replica has a GC point that others cannot reach.
+ // we should reconsider and at least
+ // 1) ensure GC points can be propagated between nodes;
+ // 2) broaden cases where we can derive that the command has not been decided and send invalidated/erased to everyone
+ // TODO (required): we have another edge cases to consider here only when we don't have a FullRoute and
+ // we report that the outcome is durable - which may not be the case for all of the shards.
+ // The home shard will stop attempting to recover the transaction in this case.
+ // This is perhaps not even a problem, and requires that no healthy home shard even has the FullRoute.
+ // Other shards would be expected to coordinate the invalidation of this transaction themselves.
Invariants.require(maxNotTruncated == null || !maxNotTruncated.maxKnowledgeStatus.hasBeen(Status.PreCommitted));
isDone = true;
- if (!tracker.isAnyNotTruncated()) // TODO (required): this check may be insufficient unless we do not permit shards to truncate when e.g. pre-bootstrap. If we require this, document it!
- {
- commitInvalidate();
- }
- else
- {
- // TODO (required): revisit this logic some more
- // this exists because if we have truncated a range we cannot seek votes to advance the state machine, so we may not be able to reach a quorum
- // but, equally, we may not propagate the GC point to all replicas.
- // the current logic has pitfalls when *some* replica has a GC point that others cannot reach.
- // we should reconsider and at least
- // 1) ensure GC points can be propagated between nodes;
- // 2) broaden cases where we can derive that the command has not been decided and send invalidated/erased to everyone
- // TODO (required): we have another edge cases to consider here only when we don't have a FullRoute and
- // we report that the outcome is durable - which may not be the case for all of the shards.
- // The home shard will stop attempting to recover the transaction in this case.
- // This is perhaps not even a problem, and requires that no healthy home shard even has the FullRoute.
- // Other shards would be expected to coordinate the invalidation of this transaction themselves.
- callback.accept(TRUNCATED_DURABLE_OR_INVALIDATED, null);
- }
+ callback.accept(TRUNCATED_DURABLE_OR_INVALIDATED, null);
return;
}
diff --git a/accord-core/src/main/java/accord/impl/AbstractLoader.java b/accord-core/src/main/java/accord/impl/AbstractLoader.java
index 566b07d1..c55ac6c8 100644
--- a/accord-core/src/main/java/accord/impl/AbstractLoader.java
+++ b/accord-core/src/main/java/accord/impl/AbstractLoader.java
@@ -49,7 +49,6 @@ public abstract class AbstractLoader implements Journal.Loader
case NO:
break;
case INVALIDATE:
- case TRUNCATE_WITH_OUTCOME_AND_DEPS:
case TRUNCATE_WITH_OUTCOME:
case TRUNCATE:
case ERASE:
diff --git a/accord-core/src/main/java/accord/impl/CommandChange.java b/accord-core/src/main/java/accord/impl/CommandChange.java
index 7600540f..51c56332 100644
--- a/accord-core/src/main/java/accord/impl/CommandChange.java
+++ b/accord-core/src/main/java/accord/impl/CommandChange.java
@@ -61,6 +61,7 @@ import static accord.impl.CommandChange.Field.RESULT;
import static accord.impl.CommandChange.Field.SAVE_STATUS;
import static accord.impl.CommandChange.Field.WAITING_ON;
import static accord.impl.CommandChange.Field.WRITES;
+import static accord.local.Cleanup.EXPUNGE;
import static accord.local.Cleanup.NO;
import static accord.local.Command.Accepted.accepted;
import static accord.local.Command.Committed.committed;
@@ -299,9 +300,6 @@ public class CommandChange
if (!nextCalled)
return NO;
- if (saveStatus == null || participants == null)
- return Cleanup.NO;
-
Durability durability = this.durability;
if (durability == null) durability = NotDurable;
Cleanup cleanup = Cleanup.shouldCleanup(input, agent, txnId, executeAt, saveStatus, durability, participants, redundantBefore, durableBefore);
@@ -310,30 +308,31 @@ public class CommandChange
return cleanup;
}
- public boolean maybeCleanup(Input input, Agent agent, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ public Cleanup maybeCleanup(Input input, Agent agent, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
Cleanup cleanup = shouldCleanup(input, agent, redundantBefore, durableBefore);
return maybeCleanup(input, cleanup);
}
- public boolean maybeCleanup(Input input, Cleanup cleanup)
+ public Cleanup maybeCleanup(Input input, Cleanup cleanup)
{
- if (saveStatus == null || cleanup == NO)
- return false;
+ if (cleanup == NO || cleanup == EXPUNGE)
+ return cleanup;
SaveStatus newSaveStatus = cleanup.appliesIfNot;
- if (saveStatus.compareTo(newSaveStatus) >= 0)
- return false;
-
- forceSetNulls(eraseKnownFieldsMask[newSaveStatus.ordinal()]);
- if (input == Input.FULL)
+ if (saveStatus == null || saveStatus.compareTo(newSaveStatus) < 0)
{
- // TODO (expected): this special-casing shouldn't be necessary
- if (newSaveStatus == SaveStatus.TruncatedApply && !saveStatus.known.is(ApplyAtKnown))
- newSaveStatus = SaveStatus.TruncatedUnapplied;
- saveStatus = newSaveStatus;
+ if (input == Input.FULL)
+ {
+ // TODO (expected): this special-casing should be declared in Cleanup
+ if (newSaveStatus == SaveStatus.TruncatedApply && (saveStatus == null || !saveStatus.known.is(ApplyAtKnown)))
+ newSaveStatus = SaveStatus.TruncatedUnapplied;
+ saveStatus = newSaveStatus;
+ }
+ forceSetNulls(eraseKnownFieldsMask[newSaveStatus.ordinal()]);
}
- return true;
+
+ return cleanup;
}
protected void setNulls(int mask)
@@ -436,7 +435,6 @@ public class CommandChange
switch (status)
{
default: throw new UnhandledEnum(status);
- case TruncatedApplyWithOutcomeAndDeps:
case TruncatedApplyWithOutcome:
case TruncatedApply:
case TruncatedUnapplied:
@@ -542,8 +540,9 @@ public class CommandChange
flags = collectFlags(before, after, Command::result, false, RESULT, flags);
// make sure we have enough information to decide whether to expunge timestamps (for unique ApplyAt HLC guarantees)
- if (isChanged(EXECUTE_AT, flags) && after.saveStatus().known.is(ApplyAtKnown))
+ if (after.saveStatus().known.is(ApplyAtKnown) && (before == null || !before.saveStatus().known.is(ApplyAtKnown)))
{
+ flags = setChanged(EXECUTE_AT, flags);
flags = setChanged(PARTICIPANTS, flags);
flags = setChanged(SAVE_STATUS, flags);
}
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index 33c792bc..9d03b7de 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -91,6 +91,7 @@ import static accord.local.Cleanup.Input.FULL;
import static accord.local.KeyHistory.ASYNC;
import static accord.local.KeyHistory.SYNC;
import static accord.local.RedundantStatus.Coverage.ALL;
+import static accord.local.StoreParticipants.Filter.LOAD;
import static accord.primitives.Known.KnownRoute.MaybeRoute;
import static accord.primitives.Routable.Domain.Key;
import static accord.primitives.Routable.Domain.Range;
@@ -103,6 +104,7 @@ import static accord.primitives.SaveStatus.Vestigial;
import static accord.primitives.Status.Applied;
import static accord.primitives.Status.Committed;
import static accord.primitives.Status.Durability.NotDurable;
+import static accord.primitives.Status.Durability.UniversalOrInvalidated;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
import static accord.primitives.Txn.Kind.EphemeralRead;
@@ -279,14 +281,15 @@ public abstract class InMemoryCommandStore extends CommandStore
}
});
});
- TxnId clearProgressLogBefore = unsafeGetRedundantBefore().minShardRedundantBefore();
+ TxnId clearProgressLogBefore = unsafeGetRedundantBefore().minShardAndLocallyAppliedBefore();
List<TxnId> clearing = ((DefaultProgressLog) progressLog).activeBefore(clearProgressLogBefore);
for (TxnId txnId : clearing)
{
GlobalCommand globalCommand = commands.get(txnId);
Invariants.require(globalCommand != null && !globalCommand.isEmpty());
Command command = globalCommand.value();
- Cleanup cleanup = Cleanup.shouldCleanup(FULL, agent, txnId, command.executeAt(), command.saveStatus(), command.durability(), command.participants(), unsafeGetRedundantBefore(), durableBefore());
+ StoreParticipants participants = command.participants().filter(LOAD, safeStore, txnId, command.executeAtIfKnown());
+ Cleanup cleanup = Cleanup.shouldCleanup(FULL, agent, txnId, command.executeAtIfKnown(), command.saveStatus(), command.durability(), participants, unsafeGetRedundantBefore(), durableBefore());
Invariants.require(command.hasBeen(Applied)
|| cleanup.compareTo(Cleanup.TRUNCATE) >= 0
|| (durableBefore().min(txnId) == NotDurable &&
@@ -297,10 +300,11 @@ public abstract class InMemoryCommandStore extends CommandStore
}
@Override
- public void markShardDurable(SafeCommandStore safeStore, TxnId syncId, Ranges ranges)
+ public void markShardDurable(SafeCommandStore safeStore, TxnId syncId, Ranges ranges, Status.Durability durability)
{
- super.markShardDurable(safeStore, syncId, ranges);
- markShardDurable(syncId, ranges);
+ super.markShardDurable(safeStore, syncId, ranges, durability);
+ if (durability.compareTo(UniversalOrInvalidated) >= 0)
+ markShardDurable(syncId, ranges);
}
private void markShardDurable(TxnId syncId, Ranges ranges)
@@ -396,34 +400,25 @@ public abstract class InMemoryCommandStore extends CommandStore
}
}
- private <T> T executeInContext(InMemoryCommandStore commandStore, PreLoadContext preLoadContext, Function<? super SafeCommandStore, T> function, boolean isDirectCall)
+ protected <T> T executeInContext(InMemoryCommandStore commandStore, PreLoadContext preLoadContext, Function<? super SafeCommandStore, T> function)
{
SafeCommandStore safeStore = commandStore.beginOperation(preLoadContext);
try
{
return function.apply(safeStore);
}
- catch (Throwable t)
- {
- if (isDirectCall) logger.error("Uncaught exception", t);
- throw new RuntimeException("Caught exception in command store " + this, t);
- }
+ // for history: we don't want to wrap exceptions here as we can't then handle the specific exception effectively
finally
{
commandStore.completeOperation(safeStore);
}
}
- protected <T> T executeInContext(InMemoryCommandStore commandStore, PreLoadContext context, Function<? super SafeCommandStore, T> function)
- {
- return executeInContext(commandStore, context, function, true);
- }
-
protected <T> void executeInContext(InMemoryCommandStore commandStore, PreLoadContext context, Function<? super SafeCommandStore, T> function, BiConsumer<? super T, Throwable> callback)
{
try
{
- T result = executeInContext(commandStore, context, function, false);
+ T result = executeInContext(commandStore, context, function);
callback.accept(result, null);
}
catch (Throwable t)
@@ -802,7 +797,7 @@ public abstract class InMemoryCommandStore extends CommandStore
if (txnId.is(EphemeralRead)) continue;
Participants<?> intersecting = (txnId.is(ExclusiveSyncPoint) ? command.participants().owns(): command.participants().stillWaitsOn()).intersecting(covering, Minimal);
if (intersecting.isEmpty()) continue;
- if (commandStore().unsafeGetRedundantBefore().preBootstrapOrStale(command.txnId(), intersecting) == ALL) continue;
+ if (commandStore().unsafeGetRedundantBefore().locallyDefunct(command.txnId(), intersecting) == ALL) continue;
if (txnId.is(Key))
{
// TODO (required): document our invariants around this scenario, where a transaction with a higher txnId
@@ -1209,6 +1204,8 @@ public abstract class InMemoryCommandStore extends CommandStore
return apply(commandStore.command(txnId).value());
Command command = commandStore.journal.loadCommand(commandStore.id, txnId, commandStore.unsafeGetRedundantBefore(), commandStore.durableBefore());
+ if (command == null)
+ return AsyncChains.success(null);
return load(command).flatMap(this::apply);
}
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
index 9dea6210..31405a4f 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/WaitingState.java
@@ -261,10 +261,8 @@ abstract class WaitingState extends BaseTxnState
if (command.known().deps().hasPreAcceptedOrProposedOrDecidedDeps() && participants.touchesOnlyOwned())
return txnIdEpoch;
- if (participants.route() == null)
- return txnIdEpoch;
-
- Participants<?> unsynced = safeStore.node().topology().unsyncedOnly(participants.route(), txnIdEpoch);
+ // note: we use hasTouched rather than route here, to account for cases route is null and we have already "touched" some key in an earlier epoch
+ Participants<?> unsynced = safeStore.node().topology().unsyncedOnly(participants.hasTouched(), txnIdEpoch);
if (unsynced == null || unsynced.isEmpty())
return txnIdEpoch;
diff --git a/accord-core/src/main/java/accord/local/Cleanup.java b/accord-core/src/main/java/accord/local/Cleanup.java
index a5821286..75454675 100644
--- a/accord-core/src/main/java/accord/local/Cleanup.java
+++ b/accord-core/src/main/java/accord/local/Cleanup.java
@@ -19,11 +19,11 @@
package accord.local;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import accord.api.Agent;
import accord.primitives.FullRoute;
import accord.primitives.Participants;
-import accord.primitives.Ranges;
import accord.primitives.Route;
import accord.primitives.SaveStatus;
import accord.primitives.Status.Durability;
@@ -33,28 +33,21 @@ import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import static accord.api.ProtocolModifiers.Toggles.requiresUniqueHlcs;
-import static accord.local.Cleanup.Input.FULL;
import static accord.local.Cleanup.Input.PARTIAL;
-import static accord.local.RedundantStatus.Coverage.ALL;
import static accord.local.RedundantStatus.Property.GC_BEFORE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
import static accord.local.RedundantStatus.Property.LOCALLY_DEFUNCT;
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.NOT_OWNED;
-import static accord.local.RedundantStatus.Property.SHARD_AND_LOCALLY_APPLIED;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_REDUNDANT;
-import static accord.local.RedundantStatus.Property.SHARD_ONLY_APPLIED;
+import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
import static accord.local.RedundantStatus.Property.TRUNCATE_BEFORE;
import static accord.primitives.Known.KnownExecuteAt.ApplyAtKnown;
-import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.SaveStatus.Erased;
import static accord.primitives.SaveStatus.Invalidated;
import static accord.primitives.SaveStatus.TruncatedApply;
import static accord.primitives.SaveStatus.TruncatedApplyWithOutcome;
-import static accord.primitives.SaveStatus.TruncatedApplyWithOutcomeAndDeps;
import static accord.primitives.SaveStatus.Uninitialised;
import static accord.primitives.SaveStatus.Vestigial;
-import static accord.primitives.Status.Applied;
import static accord.primitives.Status.Durability.UniversalOrInvalidated;
import static accord.primitives.Status.PreCommitted;
import static accord.primitives.Status.Truncated;
@@ -70,7 +63,6 @@ import static accord.primitives.TxnId.Cardinality.Any;
public enum Cleanup
{
NO(Uninitialised),
- TRUNCATE_WITH_OUTCOME_AND_DEPS(TruncatedApplyWithOutcomeAndDeps),
TRUNCATE_WITH_OUTCOME(TruncatedApplyWithOutcome),
TRUNCATE(TruncatedApply),
VESTIGIAL(Vestigial),
@@ -90,7 +82,7 @@ public enum Cleanup
public final Cleanup filter(SaveStatus saveStatus)
{
- return saveStatus.compareTo(appliesIfNot) >= 0 ? NO : this;
+ return saveStatus != null && saveStatus.compareTo(appliesIfNot) >= 0 && this != EXPUNGE ? NO : this;
}
public enum Input { PARTIAL, FULL }
@@ -145,7 +137,7 @@ public enum Cleanup
* we pessimistically assume the whole cluster may need to see its outcome
* TODO (expected): we should be able to rely on replicas to infer Invalidated from an Erased record
*/
- private static Cleanup shouldCleanupInternal(Input input, Agent agent, TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, Durability durability, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ private static Cleanup shouldCleanupInternal(Input input, Agent agent, TxnId txnId, @Nullable Timestamp executeAt, @Nullable SaveStatus saveStatus, Durability durability, @Nullable StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
if (txnId.kind() == EphemeralRead)
return NO;
@@ -153,7 +145,7 @@ public enum Cleanup
if (expunge(txnId, executeAt, saveStatus, participants, redundantBefore, durableBefore))
return expunge();
- if (participants == null)
+ if (saveStatus == null || participants == null)
return NO;
if (participants.hasFullRoute())
@@ -166,34 +158,25 @@ public enum Cleanup
// We first check if the command is redundant locally, i.e. whether it has been applied to all non-faulty replicas of the local shard
// If not, we don't want to truncate its state else we may make catching up for these other replicas much harder
FullRoute<?> route = Route.castToFullRoute(participants.route());
- if (!saveStatus.known.is(ApplyAtKnown)) executeAt = null;
- RedundantStatus redundant = redundantBefore.status(txnId, executeAt, route);
+ RedundantStatus redundant = redundantBefore.status(txnId, saveStatus.known.is(ApplyAtKnown) ? executeAt : null, route);
Invariants.require(redundant.none(NOT_OWNED),"Command " + txnId + " that is being loaded is not owned by this shard on route " + route);
if (redundant.none(LOCALLY_REDUNDANT))
return NO;
- Cleanup ifUndecided = cleanupIfUndecidedOrDefunctWithFullRoute(input, txnId, saveStatus, redundant, redundant.all(TRUNCATE_BEFORE) ? null : NO);
+ Cleanup ifUndecided = cleanupIfUndecidedWithFullRoute(input, txnId, saveStatus, redundant);
if (ifUndecided != null)
return ifUndecided;
- if (!redundant.all(SHARD_ONLY_APPLIED))
- return truncateWithOutcomeAndDeps(txnId);
-
- if (input == FULL)
+ if (!redundant.all(TRUNCATE_BEFORE))
{
- Participants<?> waitsOn = participants.waitsOn();
- if (!saveStatus.hasBeen(Applied) && (waitsOn == null || (!waitsOn.isEmpty() && redundantBefore.preBootstrapOrStale(txnId, waitsOn) != ALL)))
- {
- // if we should execute this transaction locally, and we have not done so by the time we reach a GC point, something has gone wrong
- TxnId supersededBy = redundantBefore.max(participants.route(), e -> e.maxBound(SHARD_AND_LOCALLY_APPLIED));
- Participants<?> on = waitsOn.slice(redundantBefore.foldl(participants.route(), (e, r, s) -> s.equals(e.maxBound(SHARD_AND_LOCALLY_APPLIED)) ? r.with(Ranges.of(e.range)) : r, Ranges.EMPTY, supersededBy), Minimal);
- String message = "Loading " + redundant + " command " + txnId + " with status " + saveStatus + " (that should have been Applied). Expected to be witnessed and executed by " + supersededBy + ".";
- agent.onViolation(message, on, txnId, executeAt, supersededBy, supersededBy);
+ // TODO (expected): see if we can improve our invariants so we can remove this special-case
+ if (input != PARTIAL && redundant.all(SHARD_APPLIED) && redundant.all(LOCALLY_DEFUNCT) && !redundant.any(LOCALLY_APPLIED))
return truncate(txnId);
- }
+ return NO;
}
+ Invariants.paranoid(redundant.all(SHARD_APPLIED));
Durability test = Durability.max(durability, durableBefore.min(txnId, participants.route()));
switch (test)
{
@@ -201,7 +184,10 @@ public enum Cleanup
case Local:
case NotDurable:
case ShardUniversal:
- return truncateWithOutcome(txnId);
+ // TODO (required): consider how we guarantee not to break recovery of other shards if a majority on this shard are PRE_BOOTSTRAP
+ // (if the condition is false and we fall through to removing Outcome)
+ if (participants.doesStillExecute())
+ return truncateWithOutcome(txnId);
case MajorityOrInvalidated:
case Majority:
@@ -209,7 +195,7 @@ public enum Cleanup
case UniversalOrInvalidated:
case Universal:
- if (redundant.get(GC_BEFORE) == ALL)
+ if (redundant.all(GC_BEFORE))
return erase();
return truncate(txnId);
}
@@ -228,7 +214,7 @@ public enum Cleanup
if (saveStatus == Uninitialised)
return NO;
- if (txnId.compareTo(redundantBefore.minShardRedundantBefore()) >= 0)
+ if (txnId.compareTo(redundantBefore.minShardAndLocallyAppliedBefore()) >= 0)
return NO;
return vestigial(txnId);
@@ -238,17 +224,10 @@ public enum Cleanup
return cleanupUndecided(txnId, redundant);
}
- private static Cleanup cleanupIfUndecidedOrDefunctWithFullRoute(Input input, TxnId txnId, SaveStatus saveStatus, RedundantStatus redundantStatus, Cleanup ifDecided)
+ private static Cleanup cleanupIfUndecidedWithFullRoute(Input input, TxnId txnId, SaveStatus saveStatus, RedundantStatus redundantStatus)
{
if (saveStatus.hasBeen(PreCommitted))
- {
- // TODO (required): consider more the invariants we're guaranteeing here, particularly with respect to other shards
- // also consider whether we interfere with stronger cleanup that would run after in cleanupWithFullRoute
- // This branch may primarily exist for complying with invariants that could also be weakened
- if (input != PARTIAL && redundantStatus.all(SHARD_ONLY_APPLIED) && !redundantStatus.any(LOCALLY_APPLIED) && redundantStatus.all(LOCALLY_DEFUNCT))
- return truncate(txnId);
- return ifDecided;
- }
+ return null;
if (input == PARTIAL)
return NO;
@@ -261,13 +240,13 @@ public enum Cleanup
if (redundantStatus.any(LOCALLY_APPLIED))
return invalidate(txnId);
- if (redundantStatus.all(SHARD_APPLIED_AND_LOCALLY_REDUNDANT))
+ if (redundantStatus.all(SHARD_APPLIED, LOCALLY_REDUNDANT))
return vestigial(txnId);
return NO;
}
- private static boolean expunge(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ private static boolean expunge(TxnId txnId, @Nullable Timestamp executeAt, @Nullable SaveStatus saveStatus, @Nullable StoreParticipants participants, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
if (txnId.is(Any) && durableBefore.min(txnId) != UniversalOrInvalidated)
return false;
@@ -278,10 +257,13 @@ public enum Cleanup
return false;
if (!requiresUniqueHlcs() || !txnId.is(Write)) return true;
- if (!saveStatus.known.is(ApplyAtKnown)) return true;
+ if (saveStatus == null || !saveStatus.known.is(ApplyAtKnown)) return true;
if (executeAt == null) return true;
if (minGcBefore.is(HLC_BOUND) && executeAt.uniqueHlc() < minGcBefore.hlc()) return true;
- return participants.waitsOn().isEmpty();
+ if (participants == null)
+ return true;
+ Participants<?> waitsOn = participants.waitsOn();
+ return waitsOn == null || waitsOn.isEmpty();
}
public static Cleanup forOrdinal(int ordinal)
@@ -294,10 +276,6 @@ public enum Cleanup
{
return INVALIDATE;
}
- private static Cleanup truncateWithOutcomeAndDeps(TxnId txnId)
- {
- return TRUNCATE_WITH_OUTCOME_AND_DEPS;
- }
private static Cleanup truncateWithOutcome(TxnId txnId)
{
return TRUNCATE_WITH_OUTCOME;
diff --git a/accord-core/src/main/java/accord/local/Command.java b/accord-core/src/main/java/accord/local/Command.java
index 148b4dee..fd31213c 100644
--- a/accord-core/src/main/java/accord/local/Command.java
+++ b/accord-core/src/main/java/accord/local/Command.java
@@ -69,7 +69,6 @@ import static accord.primitives.Routables.Slice;
import static accord.primitives.SaveStatus.AcceptedInvalidate;
import static accord.primitives.SaveStatus.TruncatedApply;
import static accord.primitives.SaveStatus.TruncatedApplyWithOutcome;
-import static accord.primitives.SaveStatus.TruncatedApplyWithOutcomeAndDeps;
import static accord.primitives.SaveStatus.TruncatedUnapplied;
import static accord.primitives.SaveStatus.Vestigial;
import static accord.primitives.SaveStatus.ReadyToExecute;
@@ -446,26 +445,6 @@ public abstract class Command implements ICommand
return truncated(command, participants, newSaveStatus);
}
- public static Truncated truncatedApplyWithOutcomeAndDeps(Executed command)
- {
- return truncatedApplyWithOutcomeAndDeps(command, command.participants());
- }
-
- public static Truncated truncatedApplyWithOutcomeAndDeps(Executed command, StoreParticipants participants)
- {
- return truncated(command, participants, TruncatedApplyWithOutcomeAndDeps);
- }
-
- public static Truncated truncatedApplyWithOutcome(Executed command)
- {
- return truncatedApplyWithOutcome(command, command.participants());
- }
-
- public static Truncated truncatedApplyWithOutcome(Executed command, StoreParticipants participants)
- {
- return truncated(command, participants, TruncatedApplyWithOutcome);
- }
-
public static Truncated truncated(TxnId txnId, SaveStatus saveStatus, Durability durability, StoreParticipants participants, @Nullable Timestamp executeAt, @Nullable PartialDeps partialDeps, @Nullable Writes writes, Result result)
{
Invariants.requireArgument(!txnId.awaitsOnlyDeps());
@@ -1589,7 +1568,6 @@ public abstract class Command implements ICommand
case TruncatedUnapplied:
case TruncatedApply:
case TruncatedApplyWithOutcome:
- case TruncatedApplyWithOutcomeAndDeps:
if (txnId.awaitsOnlyDeps())
return validateCommandClass(status, TruncatedAwaitsOnlyDeps.class, klass);
case Erased:
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index 1fbd3e02..f5ee9835 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -28,9 +28,11 @@ import accord.api.Agent;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.RedundantBefore.Bounds;
+import accord.local.RedundantStatus.SomeStatus;
import accord.primitives.RangeDeps;
import accord.primitives.Routables;
import accord.primitives.Route;
+import accord.primitives.Status.Durability;
import accord.primitives.Unseekables;
import accord.utils.async.AsyncChain;
@@ -66,11 +68,12 @@ import org.agrona.collections.LongHashSet;
import static accord.api.ConfigurationService.EpochReady.DONE;
import static accord.api.ProtocolModifiers.Toggles.requiresUniqueHlcs;
import static accord.local.PreLoadContext.empty;
-import static accord.local.RedundantStatus.GC_BEFORE_AND_LOCALLY_APPLIED;
-import static accord.local.RedundantStatus.LOCALLY_APPLIED_ONLY;
-import static accord.local.RedundantStatus.LOCALLY_WITNESSED_ONLY;
-import static accord.local.RedundantStatus.PRE_BOOTSTRAP_ONLY;
-import static accord.local.RedundantStatus.SHARD_ONLY_APPLIED_ONLY;
+import static accord.local.RedundantStatus.SomeStatus.GC_BEFORE_AND_LOCALLY_APPLIED;
+import static accord.local.RedundantStatus.SomeStatus.LOCALLY_APPLIED_ONLY;
+import static accord.local.RedundantStatus.SomeStatus.LOCALLY_WITNESSED_ONLY;
+import static accord.local.RedundantStatus.SomeStatus.MAJORITY_APPLIED_ONLY;
+import static accord.local.RedundantStatus.SomeStatus.PRE_BOOTSTRAP_ONLY;
+import static accord.local.RedundantStatus.SomeStatus.SHARD_APPLIED_ONLY;
import static accord.primitives.AbstractRanges.UnionMode.MERGE_ADJACENT;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Timestamp.Flag.HLC_BOUND;
@@ -108,7 +111,7 @@ public abstract class CommandStore implements AgentExecutor
public void remove(long epoch, RangesForEpoch newRangesForEpoch, Ranges removeRanges)
{
- RedundantBefore addRedundantBefore = RedundantBefore.create(removeRanges, Long.MIN_VALUE, epoch, TxnId.NONE, RedundantStatus.NONE);
+ RedundantBefore addRedundantBefore = RedundantBefore.create(removeRanges, Long.MIN_VALUE, epoch, TxnId.NONE, SomeStatus.NONE);
update(newRangesForEpoch, addRedundantBefore);
}
@@ -650,14 +653,20 @@ public abstract class CommandStore implements AgentExecutor
}
// 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 durableRanges)
+ public void markShardDurable(SafeCommandStore safeStore, TxnId globalSyncId, Ranges durableRanges, Durability durability)
{
+ if (durability.compareTo(Durability.MajorityOrInvalidated) < 0)
+ return;
+
+ SomeStatus status = durability.compareTo(Durability.UniversalOrInvalidated) >= 0 ? SHARD_APPLIED_ONLY : MAJORITY_APPLIED_ONLY;
final Ranges slicedRanges = durableRanges.slice(safeStore.ranges().allUntil(globalSyncId.epoch()), Minimal);
TxnId locallyRedundantBefore = safeStore.redundantBefore().min(slicedRanges, Bounds::maxLocallyAppliedBefore);
- RedundantBefore addShardRedundant = RedundantBefore.create(slicedRanges, globalSyncId, SHARD_ONLY_APPLIED_ONLY);
+ RedundantBefore addShardRedundant = RedundantBefore.create(slicedRanges, globalSyncId, status);
safeStore.upsertRedundantBefore(addShardRedundant);
updatedRedundantBefore(safeStore, globalSyncId, slicedRanges);
- safeStore = safeStore; // make unusable in lambda
+
+ if (status != SHARD_APPLIED_ONLY)
+ return;
if (locallyRedundantBefore.compareTo(globalSyncId) < 0)
{
@@ -673,6 +682,7 @@ public abstract class CommandStore implements AgentExecutor
// TODO (desired): not all systems care about HLC_BOUND for GC, make configurable
if (globalSyncId.is(HLC_BOUND) || !requiresUniqueHlcs())
{
+ safeStore = safeStore; // make unusable in lambda
safeStore.dataStore().snapshot(slicedRanges, globalSyncId).begin((success, fail) -> {
if (fail != null)
{
@@ -690,7 +700,7 @@ public abstract class CommandStore implements AgentExecutor
protected void updatedRedundantBefore(SafeCommandStore safeStore, TxnId syncId, Ranges ranges)
{
- TxnId clearWaitingBefore = redundantBefore.minShardRedundantBefore();
+ TxnId clearWaitingBefore = redundantBefore.minShardAndLocallyAppliedBefore();
TxnId clearAnyBefore = durableBefore().min.majorityBefore;
progressLog.clearBefore(safeStore, clearWaitingBefore, clearAnyBefore);
listeners.clearBefore(this, clearWaitingBefore);
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index 76216ec6..8817a787 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -63,6 +63,7 @@ import static accord.api.ProgressLog.BlockedUntil.HasDecidedExecuteAt;
import static accord.api.ProtocolModifiers.Toggles.DependencyElision.IF_DURABLE;
import static accord.api.ProtocolModifiers.Toggles.dependencyElision;
import static accord.api.ProtocolModifiers.Toggles.markStaleIfCannotExecute;
+import static accord.local.Cleanup.EXPUNGE;
import static accord.local.Cleanup.Input.FULL;
import static accord.local.Cleanup.NO;
import static accord.local.Cleanup.shouldCleanup;
@@ -71,6 +72,7 @@ import static accord.local.Command.Truncated.invalidated;
import static accord.local.Command.Truncated.truncated;
import static accord.local.Command.Truncated.vestigial;
import static accord.local.Commands.Validated.INSUFFICIENT;
+import static accord.local.Commands.Validated.UPDATE_TXN_AND_DEPS_INTERSECT_STABLE;
import static accord.local.Commands.Validated.UPDATE_TXN_IGNORE_DEPS;
import static accord.local.Commands.Validated.UPDATE_TXN_KEEP_DEPS;
import static accord.local.Commands.Validated.UPDATE_TXN_AND_DEPS;
@@ -78,19 +80,18 @@ import static accord.local.Commands.Validated.UPDATE_TXN_MERGE_DEPS;
import static accord.local.KeyHistory.INCR;
import static accord.local.KeyHistory.SYNC;
import static accord.local.PreLoadContext.contextFor;
-import static accord.local.RedundantStatus.Coverage.ALL;
-import static accord.local.RedundantStatus.Coverage.NONE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
import static accord.local.RedundantStatus.Property.LOCALLY_DEFUNCT;
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP_OR_STALE;
-import static accord.local.RedundantStatus.Property.SHARD_AND_LOCALLY_APPLIED;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_REDUNDANT;
+import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
import static accord.local.RedundantStatus.Property.WAS_OWNED;
import static accord.local.StoreParticipants.Filter.LOAD;
import static accord.local.StoreParticipants.Filter.UPDATE;
+import static accord.messages.Commit.Kind.StableFastPath;
import static accord.messages.Commit.Kind.StableMediumPath;
+import static accord.primitives.Known.KnownDeps.DepsFromCoordinator;
import static accord.primitives.Known.KnownDeps.DepsKnown;
import static accord.primitives.Known.KnownDeps.DepsProposedFixed;
import static accord.primitives.Known.KnownExecuteAt.ApplyAtKnown;
@@ -116,6 +117,7 @@ import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
import static accord.primitives.Route.isFullRoute;
import static accord.primitives.Txn.Kind.EphemeralRead;
+import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.primitives.Txn.Kind.Write;
import static accord.primitives.TxnId.FastPath.PrivilegedCoordinatorWithDeps;
import static accord.utils.Invariants.illegalState;
@@ -310,8 +312,10 @@ public class Commands
public static CommitOutcome commit(SafeCommandStore safeStore, SafeCommand safeCommand, StoreParticipants participants, SaveStatus newSaveStatus, Ballot ballot, TxnId txnId, Route<?> route, @Nullable Txn txn, Timestamp executeAt, Deps deps, @Nullable Commit.Kind kind)
{
final Command command = safeCommand.current();
- SaveStatus curStatus = command.saveStatus();
+ if (kind == StableFastPath && !command.promised().equals(Ballot.ZERO))
+ return CommitOutcome.Rejected;
+ SaveStatus curStatus = command.saveStatus();
Invariants.requireArgument(newSaveStatus == SaveStatus.Committed || newSaveStatus == SaveStatus.Stable);
if (newSaveStatus == SaveStatus.Committed && ballot.compareTo(command.promised()) < 0)
return curStatus.is(Truncated) || participants.owns().isEmpty()
@@ -341,8 +345,7 @@ public class Commands
}
participants = participants.filter(UPDATE, safeStore, txnId, executeAt);
- Known known = curStatus.known;
- Validated validated = validate(ballot, known, newSaveStatus, command, participants, route, txn, deps, kind);
+ Validated validated = validate(ballot, newSaveStatus, command, participants, route, txn, deps, kind, executeAt);
if (validated == INSUFFICIENT)
return CommitOutcome.Insufficient;
@@ -495,7 +498,7 @@ public class Commands
}
participants = participants.filter(UPDATE, safeStore, txnId, executeAt);
- Validated validated = validate(Ballot.ZERO, SaveStatus.PreApplied, command, participants, route, txn, deps);
+ Validated validated = validate(Ballot.ZERO, SaveStatus.PreApplied, command, participants, route, txn, deps, null, executeAt);
if (validated == INSUFFICIENT)
return ApplyOutcome.Insufficient;
@@ -722,7 +725,6 @@ public class Commands
switch (dependency.saveStatus())
{
default: throw new AssertionError("Unhandled saveStatus: " + dependency.saveStatus());
- case TruncatedApplyWithOutcomeAndDeps:
case TruncatedApplyWithOutcome:
case TruncatedApply:
case TruncatedUnapplied:
@@ -765,7 +767,7 @@ public class Commands
Participants<?> waitsOn = participants.intersecting(waiting.participants().stillWaitsOn(), Minimal);
RedundantStatus status = safeStore.redundantBefore().status(dependencyId, waitingExecuteAt, waitsOn);
- if (status.get(LOCALLY_DEFUNCT) == ALL)
+ if (status.all(LOCALLY_DEFUNCT))
return false;
throw illegalState("We have a dependency (" + dependency + ") to wait on, but have already finished waiting (" + waiting + ")");
@@ -920,7 +922,6 @@ public class Commands
result = invalidated(command, newParticipants);
break;
- case TRUNCATE_WITH_OUTCOME_AND_DEPS:
case TRUNCATE_WITH_OUTCOME:
Invariants.requireArgument(command.known().is(Apply));
Invariants.requireArgument(command.known().is(ApplyAtKnown));
@@ -937,8 +938,8 @@ public class Commands
break;
case ERASE:
- case EXPUNGE:
Invariants.require(command.saveStatus().compareTo(Erased) < 0);
+ case EXPUNGE:
result = erased(command, newParticipants);
break;
}
@@ -953,14 +954,14 @@ public class Commands
TxnId txnId = command.txnId();
RedundantStatus status = redundantBefore.status(txnId, null, participants.route());
- if (status.any(SHARD_AND_LOCALLY_APPLIED))
+ if (status.any(LOCALLY_APPLIED))
{
Invariants.paranoid(participants.stillTouches().isEmpty() ||
(participants.stillWaitsOn() != null && participants.stillWaitsOn().isEmpty()));
return true;
}
- if (status.all(SHARD_APPLIED_AND_LOCALLY_REDUNDANT) || status.all(PRE_BOOTSTRAP_OR_STALE))
+ if (status.all(SHARD_APPLIED, LOCALLY_REDUNDANT) || status.all(LOCALLY_DEFUNCT))
return true;
if (force && participants.waitsOn() != null && participants.stillWaitsOn().isEmpty())
@@ -982,7 +983,7 @@ public class Commands
return true;
}
- Invariants.require(command.saveStatus().compareTo(cleanup.appliesIfNot) < 0);
+ Invariants.require(cleanup == EXPUNGE || command.saveStatus().compareTo(cleanup.appliesIfNot) < 0);
purge(safeStore, safeCommand, command, cleanupParticipants, cleanup, true);
return true;
}
@@ -1065,7 +1066,7 @@ public class Commands
if (loadDepId != null)
{
depSafe = safeStore.ifInitialised(loadDepId);
- if (depSafe == null)
+ if (depSafe == null) // TODO (required): slice to waiting.participants().waitsOn? can simplify method
depSafe = initialiseOrRemoveDependency(safeStore, waitingSafe, loadDepId, waiting.partialDeps().participants(loadDepId));
}
}
@@ -1195,10 +1196,10 @@ public class Commands
static <R> R maybeCleanupRedundantDependency(SafeCommandStore safeStore, SafeCommand waitingSafe, TxnId depId, Function<R, SaveStatus> saveStatusGetter, Participants<?> executes, R ifNotRedundant)
{
RedundantStatus status = safeStore.redundantBefore().status(depId, null, executes);
- if (status.get(LOCALLY_REDUNDANT) == NONE)
+ if (status.none(LOCALLY_REDUNDANT))
return ifNotRedundant;
- if (status.get(LOCALLY_APPLIED) != NONE)
+ if (status.any(LOCALLY_APPLIED))
{
// we've been applied or invalidated
SaveStatus saveStatus = saveStatusGetter.apply(ifNotRedundant);
@@ -1207,14 +1208,15 @@ public class Commands
return null;
}
- boolean remove = status.get(LOCALLY_REDUNDANT) == ALL;
- if (remove && waitingSafe.txnId().awaitsPreviouslyOwned())
- remove = status.get(LOCALLY_SYNCED) == ALL || status.get(PRE_BOOTSTRAP_OR_STALE) == ALL;
+ boolean remove = status.all(LOCALLY_REDUNDANT);
+ // TODO (required): consider this logic again, incl. whether it is even needed
+ if (remove && waitingSafe.txnId().is(ExclusiveSyncPoint) && depId.is(ExclusiveSyncPoint))
+ remove = status.all(LOCALLY_SYNCED) || status.all(PRE_BOOTSTRAP_OR_STALE);
if (!remove)
return ifNotRedundant;
- if (status.get(WAS_OWNED) == ALL) removeNoLongerOwnedDependency(safeStore, waitingSafe, depId);
+ if (status.all(WAS_OWNED)) removeNoLongerOwnedDependency(safeStore, waitingSafe, depId);
else removeRedundantDependencies(safeStore, waitingSafe, depId);
return null;
}
@@ -1336,23 +1338,36 @@ public class Commands
case UPDATE_TXN_MERGE_DEPS:
return upd.partialDeps().with(newDeps.intersecting(participants.stillTouches()));
case UPDATE_TXN_AND_DEPS:
- return newDeps.intersecting(participants.stillTouches());
+ return prepareNewDeps(participants, newDeps);
+ case UPDATE_TXN_AND_DEPS_INTERSECT_STABLE:
+ return prepareNewDeps(participants, newDeps).intersectStable(upd.partialDeps(), upd.txnId());
case UPDATE_TXN_IGNORE_DEPS:
return null;
}
}
- enum Validated { INSUFFICIENT, UPDATE_TXN_IGNORE_DEPS, UPDATE_TXN_KEEP_DEPS, UPDATE_TXN_MERGE_DEPS, UPDATE_TXN_AND_DEPS }
+ private static PartialDeps prepareNewDeps(StoreParticipants participants, Deps newDeps)
+ {
+ if (newDeps != null)
+ return newDeps.intersecting(participants.stillTouches());
+
+ Invariants.require(participants.stillTouches().isEmpty());
+ return PartialDeps.NONE;
+ }
+
+ enum Validated { INSUFFICIENT, UPDATE_TXN_IGNORE_DEPS, UPDATE_TXN_KEEP_DEPS, UPDATE_TXN_MERGE_DEPS, UPDATE_TXN_AND_DEPS_INTERSECT_STABLE, UPDATE_TXN_AND_DEPS }
private static Validated validate(@Nullable Ballot ballot, SaveStatus newStatus, Command cur, StoreParticipants participants,
Route<?> addRoute, @Nullable Txn addPartialTxn, @Nullable Deps partialDeps)
{
- return validate(ballot, cur.known(), newStatus, cur, participants, addRoute, addPartialTxn, partialDeps, null);
+ return validate(ballot, newStatus, cur, participants, addRoute, addPartialTxn, partialDeps, null, null);
}
- private static Validated validate(@Nullable Ballot ballot, Known haveKnown, SaveStatus newStatus, Command cur, StoreParticipants participants,
- Route<?> addRoute, @Nullable Txn addPartialTxn, @Nullable Deps partialDeps, @Nullable Commit.Kind commitKind)
+ private static Validated validate(@Nullable Ballot ballot, SaveStatus newStatus, Command cur, StoreParticipants participants,
+ Route<?> addRoute, @Nullable Txn addPartialTxn, @Nullable Deps partialDeps,
+ @Nullable Commit.Kind commitKind, @Nullable Timestamp executeAt)
{
+ Known haveKnown = cur.known();
Known expectKnown = newStatus.known;
Invariants.require(addRoute == participants.route());
@@ -1399,22 +1414,19 @@ public class Commands
if (!containsAll(partialDeps, participants.stillTouches()))
return INSUFFICIENT;
+ if (executeAt != null && expectKnown.is(DepsKnown) && haveKnown.compareTo(DepsFromCoordinator) > 0 && executeAt.equals(cur.txnId()) && !cur.acceptedOrCommitted().equals(Ballot.ZERO))
+ return UPDATE_TXN_AND_DEPS_INTERSECT_STABLE;
+
return UPDATE_TXN_AND_DEPS;
}
- private static <V> boolean containsAll(Txn adding, Participants<?> required)
+ private static boolean containsAll(Txn adding, Participants<?> required)
{
- if (adding == null ? required.isEmpty() : adding.covers(required))
- return true;
-
- return false;
+ return adding == null ? required.isEmpty() : adding.covers(required);
}
private static <V> boolean containsAll(Deps adding, Participants<?> required)
{
- if (adding == null ? required.isEmpty() : adding.covers(required))
- return true;
-
- return false;
+ return adding == null ? required.isEmpty() : adding.covers(required);
}
}
diff --git a/accord-core/src/main/java/accord/local/PreLoadContext.java b/accord-core/src/main/java/accord/local/PreLoadContext.java
index e322470e..5b9b29f8 100644
--- a/accord-core/src/main/java/accord/local/PreLoadContext.java
+++ b/accord-core/src/main/java/accord/local/PreLoadContext.java
@@ -32,8 +32,6 @@ import java.util.List;
import java.util.function.Consumer;
import javax.annotation.Nullable;
-import static accord.local.KeyHistory.ASYNC;
-import static accord.local.KeyHistory.INCR;
import static accord.local.KeyHistory.NONE;
import static accord.local.KeyHistory.SYNC;
@@ -116,9 +114,9 @@ public interface PreLoadContext
KeyHistory requiredHistory = keyHistory();
if (requiredHistory != NONE)
{
- if (requiredHistory == INCR || requiredHistory == ASYNC)
+ if (requiredHistory.compareTo(SYNC) < 0)
requiredHistory = SYNC;
- if (requiredHistory != superset.keyHistory())
+ if (requiredHistory.compareTo(superset.keyHistory()) < 0)
return false;
}
diff --git a/accord-core/src/main/java/accord/local/RedundantBefore.java b/accord-core/src/main/java/accord/local/RedundantBefore.java
index 7d109222..492dddc2 100644
--- a/accord-core/src/main/java/accord/local/RedundantBefore.java
+++ b/accord-core/src/main/java/accord/local/RedundantBefore.java
@@ -38,6 +38,7 @@ import javax.annotation.Nullable;
import accord.api.RoutingKey;
import accord.api.VisibleForImplementation;
import accord.local.RedundantStatus.Coverage;
+import accord.local.RedundantStatus.SomeStatus;
import accord.local.RedundantStatus.Property;
import accord.primitives.AbstractRanges;
import accord.primitives.Deps;
@@ -55,33 +56,31 @@ import accord.utils.ReducingIntervalMap;
import accord.utils.ReducingRangeMap;
import static accord.api.ProtocolModifiers.Toggles.requiresUniqueHlcs;
-import static accord.local.RedundantStatus.Coverage.ALL;
import static accord.local.RedundantStatus.Coverage.SOME;
import static accord.local.RedundantStatus.ONLY_LE_MASK;
import static accord.local.RedundantStatus.NOT_OWNED_ONLY;
import static accord.local.RedundantStatus.PRE_BOOTSTRAP_OR_STALE_ONLY;
import static accord.local.RedundantStatus.Property.GC_BEFORE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
+import static accord.local.RedundantStatus.Property.LOCALLY_DEFUNCT;
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
import static accord.local.RedundantStatus.Property.LOCALLY_WITNESSED;
import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP;
import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP_OR_STALE;
-import static accord.local.RedundantStatus.Property.SHARD_AND_LOCALLY_APPLIED;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_REDUNDANT;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_SYNCED;
-import static accord.local.RedundantStatus.Property.SHARD_ONLY_APPLIED;
-import static accord.local.RedundantStatus.WAS_OWNED_LOCALLY_RETIRED;
+import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
+import static accord.local.RedundantStatus.WAS_OWNED_SYNCED;
import static accord.local.RedundantStatus.WAS_OWNED_ONLY;
-import static accord.local.RedundantStatus.WAS_OWNED_SHARD_RETIRED;
+import static accord.local.RedundantStatus.WAS_OWNED_RETIRED;
import static accord.local.RedundantStatus.addHistory;
import static accord.local.RedundantStatus.any;
import static accord.local.RedundantStatus.mask;
import static accord.local.RedundantStatus.matchesMask;
+import static accord.local.RedundantStatus.toAll;
import static accord.primitives.Timestamp.Flag.SHARD_BOUND;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.utils.ArrayBuffers.cachedAny;
-import static accord.utils.ArrayBuffers.cachedInts;
+import static accord.utils.ArrayBuffers.cachedShorts;
import static accord.utils.Invariants.illegalState;
import static accord.utils.Invariants.require;
import static accord.utils.Invariants.requireStrictlyOrdered;
@@ -142,7 +141,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
// OR we may be able to safely overwrite them with some better invariants and adequate testing
public final TxnId[] bounds;
// two entries per bound, first for equality (LE) matches, second for inequality (LT) matches
- public final int[] statuses;
+ public final short[] statuses;
private transient final long maxBoundEpoch, maxBoundHlc;
public transient final TxnId depBound;
@@ -157,7 +156,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
private transient final RedundantStatus noMatch;
private transient RedundantStatus last = RedundantStatus.NONE;
- public Bounds(Range range, long startEpoch, long endEpoch, TxnId[] bounds, int[] statuses, @Nullable Timestamp staleUntilAtLeast)
+ public Bounds(Range range, long startEpoch, long endEpoch, TxnId[] bounds, short[] statuses, @Nullable Timestamp staleUntilAtLeast)
{
super(startEpoch, endEpoch, maxBound(bounds, statuses, PRE_BOOTSTRAP), maxBound(bounds, statuses, GC_BEFORE));
this.range = range;
@@ -170,22 +169,22 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
this.depBound = depBound(bounds, statuses);
checkMinBoundOrRX(bounds);
requireStrictlyOrdered(Comparator.reverseOrder(), bounds);
- require(isShardBound(gcBefore) || gcBefore.equals(TxnId.NONE));
+ require(isShardBound(gcBefore) || isMinBound(gcBefore));
}
- public static Bounds create(Range range, TxnId bound, RedundantStatus status, @Nullable Timestamp staleUntilAtLeast)
+ public static Bounds create(Range range, TxnId bound, SomeStatus status, @Nullable Timestamp staleUntilAtLeast)
{
return create(range, Long.MIN_VALUE, Long.MAX_VALUE, bound, status, staleUntilAtLeast);
}
- public static Bounds create(Range range, long startEpoch, long endEpoch, TxnId bound, RedundantStatus status, @Nullable Timestamp staleUntilAtLeast)
+ public static Bounds create(Range range, long startEpoch, long endEpoch, TxnId bound, SomeStatus status, @Nullable Timestamp staleUntilAtLeast)
{
- return new Bounds(range, startEpoch, endEpoch, new TxnId[] { bound }, new int[] { status.encoded & ONLY_LE_MASK, status.encoded }, staleUntilAtLeast);
+ return new Bounds(range, startEpoch, endEpoch, new TxnId[] { bound }, new short[] { (short) (status.encoded & ONLY_LE_MASK), status.encoded }, staleUntilAtLeast);
}
- private static TxnId depBound(TxnId[] bounds, int[] statuses)
+ private static TxnId depBound(TxnId[] bounds, short[] statuses)
{
- TxnId depBound = maxBound(bounds, statuses, SHARD_AND_LOCALLY_APPLIED);
+ TxnId depBound = maxBound(bounds, statuses, mask(SHARD_APPLIED, SOME) | mask(LOCALLY_APPLIED, SOME));
if (depBound.equals(TxnId.NONE))
return null;
return depBound.addFlag(SHARD_BOUND);
@@ -234,19 +233,19 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
long startEpoch = Long.max(cur.startEpoch, add.startEpoch);
long endEpoch = Long.min(cur.endEpoch, add.endEpoch);
TxnId[] mergedBounds;
- int[] mergedStatuses;
+ short[] mergedStatuses;
{
Object[] boundBuf = null;
- int[] statusBuf = null;
+ short[] statusBuf = null;
int mergedCount = 0;
- int prevLtStatus = staleUntilAtLeast == null ? 0 : PRE_BOOTSTRAP_OR_STALE_ONLY.encoded;
- int prevExistingLtStatus = prevLtStatus; // we don't apply PRE_BOOTSTRAP_MERGE_MASK as this already applied
+ short prevLtStatus = staleUntilAtLeast == null ? 0 : PRE_BOOTSTRAP_OR_STALE_ONLY.encodedPart(SOME);
+ short prevExistingLtStatus = prevLtStatus; // we don't apply PRE_BOOTSTRAP_MERGE_MASK as this already applied
int i = 0, j = 0;
while (i < cur.bounds.length || j < add.bounds.length)
{
int c = i == cur.bounds.length ? -1 : j == add.bounds.length ? 1 : cur.bounds[i].compareTo(add.bounds[j]);
TxnId nextBound;
- int leStatus, ltStatus;
+ short leStatus, ltStatus;
if (c > 0)
{
nextBound = cur.bounds[i];
@@ -257,15 +256,15 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
else if (c < 0)
{
nextBound = add.bounds[j];
- leStatus = addHistory(prevLtStatus | add.statuses[j*2], prevExistingLtStatus);
- ltStatus = addHistory(prevLtStatus | add.statuses[j*2+1], prevExistingLtStatus);
+ leStatus = addHistory((short) (prevLtStatus | add.statuses[j * 2]), prevExistingLtStatus);
+ ltStatus = addHistory((short) (prevLtStatus | add.statuses[j * 2 + 1]), prevExistingLtStatus);
++j;
}
else
{
nextBound = cur.bounds[i].addFlags(add.bounds[j]);
- leStatus = addHistory(prevLtStatus | add.statuses[j*2], cur.statuses[i*2]);
- ltStatus = addHistory(prevLtStatus | add.statuses[j*2+1], prevExistingLtStatus = cur.statuses[i*2+1]);
+ leStatus = addHistory((short) (prevLtStatus | add.statuses[j * 2]), cur.statuses[i * 2]);
+ ltStatus = addHistory((short) (prevLtStatus | add.statuses[j * 2 + 1]), prevExistingLtStatus = cur.statuses[i * 2 + 1]);
++i;
++j;
}
@@ -278,9 +277,9 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
if (mergedCount >= 2)
{
- int[] prev = statusBuf != null ? statusBuf : cur.statuses;
- int prev2LtStatus = prev[mergedCount*2 - 3];
- int prevLeStatus = prev[mergedCount*2 - 2];
+ short[] prev = statusBuf != null ? statusBuf : cur.statuses;
+ short prev2LtStatus = prev[mergedCount*2 - 3];
+ short prevLeStatus = prev[mergedCount*2 - 2];
Invariants.require(prevLtStatus == prev[mergedCount*2 - 1]);
if (prevLtStatus == prev2LtStatus && prevLeStatus == prev2LtStatus)
--mergedCount;
@@ -298,7 +297,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
}
boundBuf = cachedAny().get(cur.bounds.length + add.bounds.length);
- statusBuf = cachedInts().getInts((cur.bounds.length + add.bounds.length)*2);
+ statusBuf = cachedShorts().getShorts((cur.bounds.length + add.bounds.length) * 2);
System.arraycopy(cur.bounds, 0, boundBuf, 0, mergedCount);
System.arraycopy(cur.statuses, 0, statusBuf, 0, mergedCount*2);
}
@@ -317,18 +316,18 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
else
{
mergedBounds = new TxnId[mergedCount];
- mergedStatuses = new int[mergedCount*2];
+ mergedStatuses = new short[mergedCount*2];
System.arraycopy(boundBuf, 0, mergedBounds, 0, mergedCount);
System.arraycopy(statusBuf, 0, mergedStatuses, 0, mergedCount*2);
cachedAny().forceDiscard(boundBuf, mergedCount);
- cachedInts().forceDiscard(statusBuf);
+ cachedShorts().forceDiscard(statusBuf);
}
}
return new Bounds(range, startEpoch, endEpoch, mergedBounds, mergedStatuses, staleUntilAtLeast);
}
- private static Timestamp maybeClearStaleUntilAtLeast(Timestamp staleUntilAtLeast, TxnId[] bounds, int[] statuses)
+ private static Timestamp maybeClearStaleUntilAtLeast(Timestamp staleUntilAtLeast, TxnId[] bounds, short[] statuses)
{
for (int i = 0 ; staleUntilAtLeast != null && i < bounds.length && bounds[i].compareTo(staleUntilAtLeast) > 0 ; ++i)
{
@@ -338,10 +337,10 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return staleUntilAtLeast;
}
- public Bounds with(TxnId newBound, RedundantStatus addStatus)
+ public Bounds with(TxnId newBound, SomeStatus addStatus)
{
// TODO (desired): introduce special-cased faster merge for adding a single value
- return merge(range, this, new Bounds(range, Long.MIN_VALUE, Long.MAX_VALUE, new TxnId[] { newBound }, new int[] { addStatus.encoded }, null));
+ return merge(range, this, new Bounds(range, Long.MIN_VALUE, Long.MAX_VALUE, new TxnId[] { newBound }, new short[] { addStatus.encoded }, null));
}
@VisibleForImplementation
@@ -352,7 +351,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
static @Nonnull Boolean isShardOnlyApplied(Bounds bounds, @Nonnull Boolean prev, TxnId txnId)
{
- return is(bounds, prev, txnId, SHARD_ONLY_APPLIED);
+ return is(bounds, prev, txnId, SHARD_APPLIED);
}
static @Nonnull Boolean is(Bounds bounds, @Nonnull Boolean prev, TxnId txnId, Property property)
@@ -397,6 +396,15 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return maxBound(property);
}
+ TxnId bound(Property a, Property b)
+ {
+ Invariants.require(a != GC_BEFORE);
+ Invariants.require(a.mergeWithPreBootstrapOrStale);
+ Invariants.require(b != GC_BEFORE);
+ Invariants.require(b.mergeWithPreBootstrapOrStale);
+ return maxBound(a, b);
+ }
+
boolean noBoundMatches(TxnId txnId)
{
long epoch = txnId.epoch();
@@ -428,12 +436,17 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return maxBound(bounds, statuses, property);
}
- static TxnId maxBound(TxnId[] bounds, int[] statuses, Property property)
+ public TxnId maxBound(Property a, Property b)
{
- return maxBound(bounds, statuses, mask(property, ALL));
+ return maxBound(bounds, statuses, mask(a, SOME) | mask(b, SOME));
}
- static TxnId maxBound(TxnId[] bounds, int[] statuses, int propertyMask)
+ static TxnId maxBound(TxnId[] bounds, short[] statuses, Property property)
+ {
+ return maxBound(bounds, statuses, mask(property, SOME));
+ }
+
+ static TxnId maxBound(TxnId[] bounds, short[] statuses, int propertyMask)
{
// we ignore LT/LE, as this should be applied by the caller as part of comparing maxBound
for (int i = 1; i < statuses.length ; i += 2)
@@ -531,7 +544,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
boolean outOfBounds = executeAt == null ? bounds.outOfBounds(txnId) : bounds.outOfBounds(txnId, executeAt);
Invariants.expect(!outOfBounds, "Trying to apply withoutRedundantAnd_StaleOrPreBootstrap to %s for a range we don't own (%s), suggesting we computed ownership without an up-to-date epoch", txnId, bounds);
- if (outOfBounds || bounds.is(txnId, SHARD_ONLY_APPLIED, PRE_BOOTSTRAP_OR_STALE))
+ if (outOfBounds || bounds.is(txnId, SHARD_APPLIED, PRE_BOOTSTRAP_OR_STALE))
return execute.without(Ranges.of(bounds.range));
return execute;
@@ -542,7 +555,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
if (bounds == null)
return execute;
- if (bounds.is(txnId, SHARD_ONLY_APPLIED)
+ // TODO (required): audit each of these methods
+ if (bounds.is(txnId, SHARD_APPLIED)
&& ((bounds.endEpoch <= txnId.epoch() && (!txnId.awaitsPreviouslyOwned() || bounds.isLocallyRetired()))
|| (executeAtIfKnown != null && executeAtIfKnown.epoch() < bounds.startEpoch)
|| bounds.is(txnId, PRE_BOOTSTRAP_OR_STALE)))
@@ -556,7 +570,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
if (bounds == null)
return ownedOrNotRedundant;
- if (txnId.compareTo(bounds.maxBound(SHARD_ONLY_APPLIED)) <= 0
+ // TODO (required): audit each of these methods and their call-sites
+ if (txnId.compareTo(bounds.maxBound(SHARD_APPLIED)) <= 0
&& (bounds.endEpoch <= txnId.epoch()
|| (executeAtIfKnown != null && executeAtIfKnown.epoch() < bounds.startEpoch)))
return ownedOrNotRedundant.without(Ranges.of(bounds.range));
@@ -564,6 +579,17 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return ownedOrNotRedundant;
}
+ static Participants<?> withoutShardAppliedBefore(Bounds bounds, @Nonnull Participants<?> notShardApplied, TxnId txnId)
+ {
+ if (bounds == null)
+ return notShardApplied;
+
+ if (txnId.compareTo(bounds.maxBound(SHARD_APPLIED)) <= 0)
+ return notShardApplied.without(Ranges.of(bounds.range));
+
+ return notShardApplied;
+ }
+
static Ranges withoutBeforeGc(Bounds entry, @Nonnull Ranges notGarbage, TxnId txnId, @Nullable Timestamp executeAt)
{
if (entry == null || (executeAt == null ? entry.outOfBounds(txnId) : entry.outOfBounds(txnId, executeAt)))
@@ -588,7 +614,7 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
static Ranges withoutAnyRetired(Bounds bounds, @Nonnull Ranges notRetired)
{
- if (bounds == null || bounds.endEpoch > bounds.maxBound(SHARD_ONLY_APPLIED).epoch())
+ if (bounds == null || bounds.endEpoch > bounds.maxBound(SHARD_APPLIED).epoch())
return notRetired;
return notRetired.without(Ranges.of(bounds.range));
@@ -605,40 +631,49 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return notPreBootstrap;
}
- RedundantStatus get(TxnId txnId, @Nullable Timestamp executeAtIfKnown)
+ RedundantStatus get(TxnId txnId, @Nullable Timestamp applyAtIfKnown)
{
if (wasOwned(txnId))
{
- if (isShardRetired()) return WAS_OWNED_SHARD_RETIRED;
- if (isLocallyRetired()) return WAS_OWNED_LOCALLY_RETIRED;
+ if (isRetired()) return WAS_OWNED_RETIRED;
+ if (isLocallyRetired()) return WAS_OWNED_SYNCED;
return WAS_OWNED_ONLY;
}
- return getIgnoringOwnership(txnId, executeAtIfKnown);
+ return getIgnoringOwnership(txnId, applyAtIfKnown);
}
- RedundantStatus getIgnoringOwnership(TxnId txnId, @Nullable Timestamp executeAtIfKnown)
+ RedundantStatus getIgnoringOwnership(TxnId txnId, @Nullable Timestamp applyAtIfKnown)
{
int i = findStatusIndex(txnId);
if (i < 0)
return noMatch;
- int status = statuses[i];
+ short status = statuses[i];
if (any(status, GC_BEFORE))
{
- if (requiresUniqueHlcs() && executeAtIfKnown != null && bounds[i/2].hlc() <= executeAtIfKnown.uniqueHlc())
+ if (requiresUniqueHlcs() && txnId.isWrite())
{
- long uniqueHlc = executeAtIfKnown.uniqueHlc();
- i/= 2;
- while (--i >= 0 && bounds[i].hlc() <= uniqueHlc) {}
- if (i < 0 || !any(statuses[i*2+1], GC_BEFORE))
- status &= ~mask(GC_BEFORE, ALL);
+ if (applyAtIfKnown == null)
+ {
+ // if this is partial input (e.g. compaction) we don't want to infer ERASE
+ // we anyway only EXPUNGE on summary information before this point
+ status &= (short) ~(1 << GC_BEFORE.shift());
+ }
+ else if (bounds[i/2].hlc() <= applyAtIfKnown.uniqueHlc())
+ {
+ long uniqueHlc = applyAtIfKnown.uniqueHlc();
+ i/= 2;
+ while (--i >= 0 && bounds[i].hlc() <= uniqueHlc) {}
+ if (i < 0 || !any(statuses[i*2+1], GC_BEFORE))
+ status &= (short) ~(1 << GC_BEFORE.shift());
+ }
}
}
RedundantStatus reuse = last;
if (status == reuse.encoded)
return reuse;
- return last = new RedundantStatus(status);
+ return last = new RedundantStatus(toAll(status));
}
private static int compareStaleUntilAtLeast(@Nullable Timestamp a, @Nullable Timestamp b)
@@ -653,9 +688,14 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return gcBefore;
}
- public final TxnId shardRedundantBefore()
+ public final TxnId shardAndLocallyRedundantBefore()
{
- return bound(SHARD_APPLIED_AND_LOCALLY_REDUNDANT);
+ return bound(SHARD_APPLIED, LOCALLY_REDUNDANT);
+ }
+
+ public final TxnId shardAppliedBefore()
+ {
+ return bound(SHARD_APPLIED);
}
public final TxnId locallyWitnessedBefore()
@@ -686,14 +726,16 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return ub.epoch() < startEpoch || lb.epoch() >= endEpoch;
}
+ // TODO (expected): should we consider executeAt here?
+ // Anyone that didn't own txnId will have to bootstrap to include this transaction anyway, so it's safe
private boolean wasOwned(EpochSupplier lb)
{
return lb.epoch() >= endEpoch;
}
- private boolean isShardRetired()
+ private boolean isRetired()
{
- return endEpoch <= maxBound(SHARD_APPLIED_AND_LOCALLY_SYNCED).epoch();
+ return endEpoch <= maxBound(SHARD_APPLIED, LOCALLY_SYNCED).epoch();
}
private boolean isLocallyRetired()
@@ -745,15 +787,15 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
public static RedundantBefore EMPTY = new RedundantBefore();
private final Ranges staleRanges, locallyRetiredRanges;
- private final TxnId maxBootstrap, maxGcBefore;
- private final TxnId minShardRedundantBefore, minGcBefore;
+ private final TxnId maxBootstrap, maxShardAppliedBefore, maxGcBefore;
+ private final TxnId minShardAndLocallyAppliedBefore, minGcBefore;
private final long maxStartEpoch, minLocallyRetiredEpoch;
private RedundantBefore()
{
staleRanges = locallyRetiredRanges = Ranges.EMPTY;
- maxBootstrap = maxGcBefore = TxnId.NONE;
- minShardRedundantBefore = minGcBefore = TxnId.MAX;
+ maxBootstrap = maxShardAppliedBefore = maxGcBefore = TxnId.NONE;
+ minShardAndLocallyAppliedBefore = minGcBefore = TxnId.MAX;
maxStartEpoch = 0;
minLocallyRetiredEpoch = Long.MAX_VALUE;
}
@@ -763,7 +805,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
super(inclusiveEnds, starts, values);
staleRanges = extractRanges(values, b -> b.staleUntilAtLeast != null);
locallyRetiredRanges = extractRanges(values, Bounds::isLocallyRetired);
- TxnId maxBootstrap = TxnId.NONE, maxGcBefore = TxnId.NONE, minShardRedundantBefore = TxnId.MAX, minGcBefore = TxnId.MAX;
+ TxnId maxBootstrap = TxnId.NONE, maxGcBefore = TxnId.NONE, maxShardAppliedBefore = TxnId.NONE;
+ TxnId minShardAndLocallyRedundantBefore = TxnId.MAX, minGcBefore = TxnId.MAX;
long minLocallyRetiredEpoch = Long.MAX_VALUE, maxStartEpoch = 0;
boolean hasLocallyRetired = !locallyRetiredRanges.isEmpty();
for (Bounds bounds : values)
@@ -784,9 +827,12 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
minGcBefore = gcBefore;
}
{
- TxnId shardRedundantBefore = bounds.shardRedundantBefore();
- if (shardRedundantBefore.compareTo(minShardRedundantBefore) < 0)
- minShardRedundantBefore = shardRedundantBefore;
+ TxnId shardAndLocallyRedundantBefore = bounds.shardAndLocallyRedundantBefore();
+ TxnId shardAppliedBefore = bounds.shardAppliedBefore();
+ if (shardAndLocallyRedundantBefore.compareTo(minShardAndLocallyRedundantBefore) < 0)
+ minShardAndLocallyRedundantBefore = shardAndLocallyRedundantBefore;
+ if (shardAppliedBefore.compareTo(maxShardAppliedBefore) > 0)
+ maxShardAppliedBefore = shardAppliedBefore;
}
if (bounds.startEpoch > maxStartEpoch)
maxStartEpoch = bounds.startEpoch;
@@ -794,8 +840,9 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
minLocallyRetiredEpoch = bounds.endEpoch;
}
this.maxBootstrap = maxBootstrap;
+ this.maxShardAppliedBefore = maxShardAppliedBefore;
this.maxGcBefore = maxGcBefore;
- this.minShardRedundantBefore = minShardRedundantBefore;
+ this.minShardAndLocallyAppliedBefore = minShardAndLocallyRedundantBefore;
this.minGcBefore = minGcBefore;
this.maxStartEpoch = maxStartEpoch;
this.minLocallyRetiredEpoch = minLocallyRetiredEpoch;
@@ -824,27 +871,27 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return Ranges.ofSortedAndDeoverlapped(result).mergeTouching();
}
- public static RedundantBefore create(AbstractRanges ranges, TxnId bound, RedundantStatus status)
+ public static RedundantBefore create(AbstractRanges ranges, TxnId bound, SomeStatus status)
{
return create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, bound, status);
}
public static RedundantBefore createStale(AbstractRanges ranges, Timestamp staleUntilAtLeast)
{
- return create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, TxnId.NONE, RedundantStatus.NONE, staleUntilAtLeast);
+ return create(ranges, Long.MIN_VALUE, Long.MAX_VALUE, TxnId.NONE, SomeStatus.NONE, staleUntilAtLeast);
}
- public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, TxnId bound, RedundantStatus status)
+ public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, TxnId bound, SomeStatus status)
{
return create(ranges, startEpoch, endEpoch, bound, status, null);
}
- public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, TxnId bound, RedundantStatus status, @Nullable Timestamp staleUntilAtLeast)
+ public static RedundantBefore create(AbstractRanges ranges, long startEpoch, long endEpoch, TxnId bound, SomeStatus status, @Nullable Timestamp staleUntilAtLeast)
{
if (ranges.isEmpty())
return new RedundantBefore();
- Bounds bounds = new Bounds(null, startEpoch, endEpoch, new TxnId[] { bound }, new int[] { status.encoded & ONLY_LE_MASK, status.encoded }, staleUntilAtLeast);
+ Bounds bounds = new Bounds(null, startEpoch, endEpoch, new TxnId[] { bound }, new short[] { (short) (status.encoded & ONLY_LE_MASK), status.encoded }, staleUntilAtLeast);
Builder builder = new Builder(ranges.get(0).endInclusive(), ranges.size() * 2);
for (int i = 0 ; i < ranges.size() ; ++i)
{
@@ -880,9 +927,9 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
* RedundantStatus.REDUNDANT overrides PRE_BOOTSTRAP; to avoid complicating that state machine,
* for cases where we care independently about the overall pre-bootstrap state we have a separate mechanism
*/
- public Coverage preBootstrapOrStale(TxnId txnId, Participants<?> participants)
+ public Coverage locallyDefunct(TxnId txnId, Participants<?> participants)
{
- return status(txnId, null, participants).get(PRE_BOOTSTRAP_OR_STALE);
+ return status(txnId, null, participants).get(LOCALLY_DEFUNCT);
}
public <T extends Deps> RangeDeps.BuilderByRange collectDeps(Routables<?> participants, RangeDeps.BuilderByRange builder, EpochSupplier minEpoch, EpochSupplier executeAt)
@@ -934,9 +981,9 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
return foldl(ranges, Bounds::withoutAnyRetired, ranges, r -> false);
}
- public TxnId minShardRedundantBefore()
+ public TxnId minShardAndLocallyAppliedBefore()
{
- return minShardRedundantBefore;
+ return minShardAndLocallyAppliedBefore;
}
public TxnId minGcBefore()
@@ -962,6 +1009,8 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
/**
* Subtract anything we don't need to coordinate (because they are known to be shard durable),
* and we don't execute locally, i.e. are pre-bootstrap or stale (or for RX are on ranges that are already retired)
+ *
+ * TODO (required): document and justify all of this, in terms of invariants we're maintaining
*/
public Participants<?> expectToOwn(TxnId txnId, @Nullable Timestamp executeAt, Participants<?> participants)
{
@@ -1024,21 +1073,11 @@ public class RedundantBefore extends ReducingRangeMap<RedundantBefore.Bounds>
/**
* Subtract any ranges we consider stale, pre-bootstrap, or that were previously owned and have been retired
*/
- public Participants<?> expectToCalculateDependenciesOrConsultOnRecovery(TxnId txnId, @Nullable Timestamp executeAtIfKnown, Participants<?> participants)
- {
- if (!mayFilterStaleOrPreBootstrapOrRetiredOrNotOwned(txnId, executeAtIfKnown, participants))
- return participants;
- return foldl(participants, Bounds::withoutNotOwnedShardOnlyRedundant, participants, txnId, executeAtIfKnown);
- }
-
- /**
- * Subtract any ranges we consider stale, pre-bootstrap, or that were previously owned and have been retired
- */
- public Participants<?> expectToOwnOrExecuteOrConsultOnRecovery(TxnId txnId, @Nullable Timestamp executeAtIfKnown, Participants<?> participants)
+ public Participants<?> withoutShardApplied(TxnId txnId, Participants<?> participants)
{
- if (!mayFilterStaleOrPreBootstrapOrRetiredOrNotOwned(txnId, executeAtIfKnown, participants))
+ if (participants.isEmpty() || maxShardAppliedBefore.compareTo(txnId) < 0)
return participants;
- return foldl(participants, Bounds::withoutRedundantAnd_StaleOrPreBootstrapOrRetiredOrNotOwned, participants, txnId, executeAtIfKnown);
+ return foldl(participants, Bounds::withoutShardAppliedBefore, participants, txnId);
}
public static class Builder extends AbstractIntervalBuilder<RoutingKey, Bounds, RedundantBefore>
diff --git a/accord-core/src/main/java/accord/local/RedundantStatus.java b/accord-core/src/main/java/accord/local/RedundantStatus.java
index 06548214..080eff86 100644
--- a/accord-core/src/main/java/accord/local/RedundantStatus.java
+++ b/accord-core/src/main/java/accord/local/RedundantStatus.java
@@ -20,6 +20,7 @@ package accord.local;
import accord.utils.Invariants;
import accord.utils.TinyEnumSet;
+import accord.utils.UnhandledEnum;
import static accord.local.RedundantStatus.Cmp.LT;
import static accord.local.RedundantStatus.Cmp.LE;
@@ -27,20 +28,15 @@ import static accord.local.RedundantStatus.Coverage.ALL;
import static accord.local.RedundantStatus.Coverage.SOME;
import static accord.local.RedundantStatus.Property.GC_BEFORE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
-import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
import static accord.local.RedundantStatus.Property.LOCALLY_WITNESSED;
+import static accord.local.RedundantStatus.Property.MAJORITY_APPLIED;
import static accord.local.RedundantStatus.Property.NOT_OWNED;
import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP;
import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP_OR_STALE;
import static accord.local.RedundantStatus.Property.REVERSE_PROPERTIES;
-import static accord.local.RedundantStatus.Property.SHARD_AND_LOCALLY_APPLIED;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_REDUNDANT;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_SYNCED;
-import static accord.local.RedundantStatus.Property.SHARD_ONLY_APPLIED;
+import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
import static accord.local.RedundantStatus.Property.WAS_OWNED;
-import static java.lang.Integer.numberOfLeadingZeros;
-import static java.lang.Integer.numberOfTrailingZeros;
// TODO (testing): validate that we never lose a status previously held,
// i.e. once any particular property holds for a TxnId, it should continue to hold in perpetuity
@@ -48,9 +44,16 @@ public class RedundantStatus
{
public enum Coverage
{
- NONE,
- SOME,
- ALL;
+ NONE(0),
+ SOME(1),
+ ALL(0x10001);
+
+ final int mask;
+
+ Coverage(int mask)
+ {
+ this.mask = mask;
+ }
public boolean atLeast(Coverage coverage)
{
@@ -60,18 +63,18 @@ public class RedundantStatus
enum Cmp
{
- LE, LT;
+ LE, LT
}
public enum Property
{
- NOT_OWNED (false, false, LE),
// applied or pre-bootstrap or stale or was owned
- LOCALLY_REDUNDANT (true, true, LE),
+ // (DEFUNCT | APPLIED)
+ LOCALLY_REDUNDANT (false, true, LT),
- // was owned OR pre-bootstrap or stale
- LOCALLY_DEFUNCT (true, true, LE, LOCALLY_REDUNDANT),
- WAS_OWNED (true, false, LE, LOCALLY_DEFUNCT),
+ // pre-bootstrap or stale or was owned
+ // (WAS_OWNED | PRE_BOOTSTRAP_OR_STALE)
+ LOCALLY_DEFUNCT (false, true, LT, LOCALLY_REDUNDANT),
/**
* We can bootstrap ranges at different times, and have a transaction that participates in both ranges -
@@ -81,34 +84,34 @@ public class RedundantStatus
* which then execute in an unordered fashion.
*
* See also {@link SafeCommandStore#safeToReadAt()}.
+ * TODO (expected): do we need to distinguish this case from DEFUNCT?
*/
- PRE_BOOTSTRAP_OR_STALE (false, true, LT, LOCALLY_DEFUNCT),
- PRE_BOOTSTRAP (false, true, LT, PRE_BOOTSTRAP_OR_STALE),
+ PRE_BOOTSTRAP_OR_STALE (true, true, LT, LOCALLY_DEFUNCT),
+ PRE_BOOTSTRAP (true, true, LT, PRE_BOOTSTRAP_OR_STALE),
- LOCALLY_WITNESSED (true, true, LE),
+ LOCALLY_WITNESSED (false, true, LE),
// we've applied a sync point locally covering the transaction, but the transaction itself may not have applied
- LOCALLY_SYNCED (true, true, LE, LOCALLY_REDUNDANT),
- LOCALLY_APPLIED (false, false, LE, LOCALLY_SYNCED),
+ LOCALLY_SYNCED (false, true, LE, LOCALLY_REDUNDANT),
+ LOCALLY_APPLIED (true, false, LE, LOCALLY_SYNCED),
/**
- * We have fully executed until across all healthy non-bootstrapping replicas for the range in question,
+ * We have fully executed until across all a majority of replicas for the range in question,
* but not necessarily ourselves.
*/
- SHARD_ONLY_APPLIED (true, true, LE),
- SHARD_APPLIED_AND_LOCALLY_REDUNDANT(true, true, LE, SHARD_ONLY_APPLIED, LOCALLY_REDUNDANT),
- SHARD_APPLIED_AND_LOCALLY_SYNCED (true, true, LE, SHARD_APPLIED_AND_LOCALLY_REDUNDANT, LOCALLY_SYNCED),
+ MAJORITY_APPLIED (false, true, LE),
/**
- * We have fully executed across all healthy non-bootstrapping replicas, including ourselves.
- *
- * Note that in some cases we can safely use this property in place of gcBefore for cleaning up or inferring
- * invalidations, but remember that if we are erasing data we may report to peers then we must provide an RX
- * in place of that data to prevent a stale peer thinking they have enough information.
+ * We have fully executed until across all healthy non-bootstrapping replicas for the range in question,
+ * but not necessarily ourselves.
*/
- SHARD_AND_LOCALLY_APPLIED (false, false, LE, SHARD_APPLIED_AND_LOCALLY_SYNCED, LOCALLY_APPLIED),
- TRUNCATE_BEFORE (false, true, LT, SHARD_APPLIED_AND_LOCALLY_SYNCED),
+ SHARD_APPLIED (false, true, LE, MAJORITY_APPLIED),
+
+ TRUNCATE_BEFORE (false, true, LT, SHARD_APPLIED, LOCALLY_SYNCED),
+ GC_BEFORE (false, true, LT, TRUNCATE_BEFORE),
- GC_BEFORE (false, true, LT, TRUNCATE_BEFORE),
+ // not persisted
+ WAS_OWNED (false, false, LT, LOCALLY_DEFUNCT),
+ NOT_OWNED (false, false, LT),
;
@@ -125,15 +128,15 @@ public class RedundantStatus
}
}
- final boolean mergeWithWasOwned;
+ final boolean overrideWasOwned;
final boolean mergeWithPreBootstrapOrStale;
final Cmp cmp;
final int compareLessEqual;
final Property[] implies;
- Property(boolean mergeWithWasOwned, boolean mergeWithPreBootstrapOrStale, Cmp cmp, Property ... implies)
+ Property(boolean overrideWasOwned, boolean mergeWithPreBootstrapOrStale, Cmp cmp, Property ... implies)
{
- this.mergeWithWasOwned = mergeWithWasOwned;
+ this.overrideWasOwned = overrideWasOwned;
this.mergeWithPreBootstrapOrStale = mergeWithPreBootstrapOrStale;
this.cmp = cmp;
this.compareLessEqual = cmp == Cmp.LT ? -1 : 0;
@@ -142,24 +145,41 @@ public class RedundantStatus
final int shift()
{
- return ordinal() * 2;
+ return ordinal();
}
}
- public static final RedundantStatus NONE = new RedundantStatus(0);
- public static final RedundantStatus NOT_OWNED_ONLY = oneSlow(NOT_OWNED);
+ public static class SomeStatus
+ {
+ public static final SomeStatus NONE = new SomeStatus((short)0);
+
+ public static final SomeStatus PRE_BOOTSTRAP_ONLY = oneSlow(PRE_BOOTSTRAP);
+ public static final SomeStatus LOCALLY_WITNESSED_ONLY = oneSlow(LOCALLY_WITNESSED);
+ public static final SomeStatus LOCALLY_APPLIED_ONLY = oneSlow(LOCALLY_APPLIED);
+ public static final SomeStatus MAJORITY_APPLIED_ONLY = oneSlow(MAJORITY_APPLIED);
+ public static final SomeStatus SHARD_APPLIED_ONLY = oneSlow(SHARD_APPLIED);
+ public static final SomeStatus GC_BEFORE_AND_LOCALLY_APPLIED = multi(GC_BEFORE, LOCALLY_APPLIED);
+
+ final short encoded;
+ public SomeStatus(short encoded)
+ {
+ this.encoded = encoded;
+ }
- public static final RedundantStatus WAS_OWNED_ONLY = oneSlow(WAS_OWNED);
- public static final RedundantStatus WAS_OWNED_LOCALLY_RETIRED = multi(WAS_OWNED, LOCALLY_SYNCED);
- public static final RedundantStatus WAS_OWNED_SHARD_RETIRED = multi(WAS_OWNED, SHARD_APPLIED_AND_LOCALLY_SYNCED);
+ public boolean is(Property property)
+ {
+ return 0 != (1 & (encoded >>> property.shift()));
+ }
+ }
- public static final RedundantStatus PRE_BOOTSTRAP_ONLY = oneSlow(PRE_BOOTSTRAP);
- public static final RedundantStatus PRE_BOOTSTRAP_OR_STALE_ONLY = oneSlow(PRE_BOOTSTRAP_OR_STALE);
+ private static final Coverage[] COVERAGE_TO_STRING = new Coverage[] { ALL, SOME };
+ public static final RedundantStatus NONE = new RedundantStatus(0);
+ public static final RedundantStatus PRE_BOOTSTRAP_OR_STALE_ONLY = toAll(oneSlow(PRE_BOOTSTRAP_OR_STALE));
+ public static final RedundantStatus NOT_OWNED_ONLY = toAll(oneSlow(NOT_OWNED));
- public static final RedundantStatus LOCALLY_WITNESSED_ONLY = oneSlow(LOCALLY_WITNESSED);
- public static final RedundantStatus LOCALLY_APPLIED_ONLY = oneSlow(LOCALLY_APPLIED);
- public static final RedundantStatus SHARD_ONLY_APPLIED_ONLY = oneSlow(SHARD_ONLY_APPLIED);
- public static final RedundantStatus GC_BEFORE_AND_LOCALLY_APPLIED = multi(GC_BEFORE, LOCALLY_APPLIED);
+ public static final RedundantStatus WAS_OWNED_ONLY = toAll(oneSlow(WAS_OWNED));
+ public static final RedundantStatus WAS_OWNED_SYNCED = toAll(multi(WAS_OWNED, LOCALLY_SYNCED));
+ public static final RedundantStatus WAS_OWNED_RETIRED = toAll(multi(WAS_OWNED, GC_BEFORE));
final int encoded;
RedundantStatus(int encoded)
@@ -179,6 +199,12 @@ public class RedundantStatus
return selectOrCreate(bits, this, that);
}
+ public RedundantStatus subtract(RedundantStatus that)
+ {
+ int bits = this.encoded & ~that.encoded;
+ return selectOrCreate(bits, this, that);
+ }
+
static RedundantStatus addHistory(RedundantStatus add, RedundantStatus history)
{
int addEncoded = add.encoded;
@@ -188,16 +214,11 @@ public class RedundantStatus
return selectOrCreate(encoded, history, add);
}
- static int addHistory(int add, int history)
+ static short addHistory(short add, short history)
{
if (decode(history, PRE_BOOTSTRAP_OR_STALE) != Coverage.NONE)
add &= PRE_BOOTSTRAP_MERGE_MASK;
- int merged = add | history;
- // upgrade any combination of LOCAL and SHARD to SHARD_AND_LOCAL
- int upgradeMask = -((merged >>> SHARD_ONLY_APPLIED.shift()) & 1);
- merged |= (upgradeMask & merged & UPGRADE_SHARD_LOCAL_APPLIED_SYNCED_MASK) << UPGRADE_SHARD_LOCAL_APPLIED_SYNCED_SHIFT;
- merged |= (upgradeMask & merged & UPGRADE_SHARD_LOCAL_REDUNDANT_MASK) << UPGRADE_SHARD_LOCAL_REDUNDANT_SHIFT;
- return merged;
+ return (short) (add | history);
}
static RedundantStatus selectOrCreate(int encoded, RedundantStatus a, RedundantStatus b)
@@ -207,6 +228,24 @@ public class RedundantStatus
return new RedundantStatus(encoded);
}
+ static SomeStatus selectOrCreate(short encoded, SomeStatus a, SomeStatus b)
+ {
+ if (encoded == a.encoded) return a;
+ if (encoded == b.encoded) return b;
+ return new SomeStatus(encoded);
+ }
+
+ public short encodedPart(Coverage coverage)
+ {
+ switch (coverage)
+ {
+ default: throw new UnhandledEnum(coverage);
+ case NONE: return (short)0;
+ case SOME: return (short)(encoded & 0xff);
+ case ALL: return (short)((encoded >>> 16) & 0xff);
+ }
+ }
+
public Coverage get(Property property)
{
return get(encoded, property);
@@ -222,11 +261,21 @@ public class RedundantStatus
return all(encoded, property);
}
+ public boolean all(Property a, Property b)
+ {
+ return all(encoded, a, b);
+ }
+
public static boolean all(int encoded, Property property)
{
return get(encoded, property) == ALL;
}
+ public static boolean all(int encoded, Property a, Property b)
+ {
+ return all(encoded, a) && all(encoded, b);
+ }
+
public boolean none(Property property)
{
return none(encoded, property);
@@ -254,61 +303,45 @@ public class RedundantStatus
public static int mask(Property property, Coverage coverage)
{
- int mask = coverage.ordinal();
- mask |= mask >>> 1;
+ int mask = coverage.mask;
return mask << property.shift();
}
public static Coverage decode(long encoded, Property property)
{
- int coverage = (int)((encoded >>> property.shift()) & 0x3);
+ int coverage = (int)((encoded >>> property.shift()) & ALL.mask);
switch (coverage)
{
default: throw new IllegalStateException("Invalid Coverage value encoded for " + property + ": " + coverage);
case 0: return Coverage.NONE;
case 1: return SOME;
- case 3: return ALL;
+ case 0x10001: return ALL;
}
}
- private static final int ALL_BITS = 0xAAAAAAAA;
- private static final int ANY_BITS = 0x55555555;
- private static final int WAS_OWNED_NO_MERGE_MASK;
- static final int PRE_BOOTSTRAP_MERGE_MASK;
- static final int ONLY_LE_MASK;
- private static final int WAS_OWNED_MASK = 0x3 << WAS_OWNED.shift();
- private static final int NOT_OWNED_MASK = 0x3 << NOT_OWNED.shift();
- private static final int UPGRADE_SHARD_LOCAL_APPLIED_SYNCED_MASK, UPGRADE_SHARD_LOCAL_REDUNDANT_MASK;
- private static final int UPGRADE_SHARD_LOCAL_APPLIED_SYNCED_SHIFT, UPGRADE_SHARD_LOCAL_REDUNDANT_SHIFT;
+ private static final int ALL_BITS = 0xFFFF0000;
+ private static final int ANY_BITS = 0x0000FFFF;
+ private static final int WAS_OWNED_OVERRIDE_MASK;
+ static final short PRE_BOOTSTRAP_MERGE_MASK;
+ static final short ONLY_LE_MASK;
+ private static final int WAS_ALL_OWNED_MASK = ALL.mask << WAS_OWNED.shift();
+ private static final int NOT_OWNED_MASK = ALL.mask << NOT_OWNED.shift();
static
{
- int wasOwnedMask = 0, preBootstrapMask = 0, leMask = 0;
+ int wasOwnedMask = 0;
+ short preBootstrapMask = 0, leMask = 0;
for (Property property : Property.values())
{
- if (!property.mergeWithWasOwned)
- wasOwnedMask |= encode(property);
+ if (!property.overrideWasOwned)
+ wasOwnedMask |= encodeAll(property);
if (property.mergeWithPreBootstrapOrStale)
- preBootstrapMask |= encode(property);
+ preBootstrapMask |= encodeAny(property);
if (property.cmp == LE)
- leMask |= encode(property);
+ leMask |= encodeAny(property);
}
- WAS_OWNED_NO_MERGE_MASK = wasOwnedMask;
+ WAS_OWNED_OVERRIDE_MASK = wasOwnedMask;
PRE_BOOTSTRAP_MERGE_MASK = preBootstrapMask;
ONLY_LE_MASK = leMask;
-
- int localAppliedSyncedMask = encode(LOCALLY_APPLIED) | encode(LOCALLY_SYNCED);
- int localRedundantMask = encode(LOCALLY_REDUNDANT);
- int shardAndLocalAppliedSyncedMask = encode(SHARD_AND_LOCALLY_APPLIED) | encode(SHARD_APPLIED_AND_LOCALLY_SYNCED);
- int shardAndLocalRedundantMask = encode(SHARD_APPLIED_AND_LOCALLY_REDUNDANT);
- Invariants.require(numberOfLeadingZeros(localAppliedSyncedMask) + numberOfTrailingZeros(localAppliedSyncedMask) == 28);
- Invariants.require(numberOfLeadingZeros(shardAndLocalAppliedSyncedMask) + numberOfTrailingZeros(shardAndLocalAppliedSyncedMask) == 28);
- UPGRADE_SHARD_LOCAL_APPLIED_SYNCED_MASK = localAppliedSyncedMask;
- UPGRADE_SHARD_LOCAL_APPLIED_SYNCED_SHIFT = numberOfTrailingZeros(shardAndLocalAppliedSyncedMask) - numberOfTrailingZeros(localAppliedSyncedMask);
- UPGRADE_SHARD_LOCAL_REDUNDANT_MASK = localRedundantMask;
- UPGRADE_SHARD_LOCAL_REDUNDANT_SHIFT = numberOfTrailingZeros(shardAndLocalRedundantMask) - numberOfTrailingZeros(localRedundantMask);
- Invariants.require(encode(LOCALLY_APPLIED) << UPGRADE_SHARD_LOCAL_APPLIED_SYNCED_SHIFT == encode(SHARD_AND_LOCALLY_APPLIED));
- Invariants.require(encode(LOCALLY_SYNCED) << UPGRADE_SHARD_LOCAL_APPLIED_SYNCED_SHIFT == encode(SHARD_APPLIED_AND_LOCALLY_SYNCED));
- Invariants.require(encode(LOCALLY_REDUNDANT) << UPGRADE_SHARD_LOCAL_REDUNDANT_SHIFT == encode(SHARD_APPLIED_AND_LOCALLY_REDUNDANT));
}
public static int mergeShards(int a, int b)
@@ -318,38 +351,57 @@ public class RedundantStatus
int all = (a & b) & ALL_BITS;
int any = either & ANY_BITS;
int result = all | any;
- if ((either & WAS_OWNED_MASK) == WAS_OWNED_MASK)
- result |= either & WAS_OWNED_NO_MERGE_MASK;
+ if ((either & WAS_ALL_OWNED_MASK) == WAS_ALL_OWNED_MASK)
+ {
+ if ((a & WAS_ALL_OWNED_MASK) != WAS_ALL_OWNED_MASK)
+ result |= a & WAS_OWNED_OVERRIDE_MASK;
+ if ((b & WAS_ALL_OWNED_MASK) != WAS_ALL_OWNED_MASK)
+ result |= b & WAS_OWNED_OVERRIDE_MASK;
+ }
return result;
}
- public static RedundantStatus oneSlow(Property property)
+ public static SomeStatus oneSlow(Property property)
{
- return new RedundantStatus(transitiveClosure(property));
+ return new SomeStatus(transitiveClosure(property));
}
- private static RedundantStatus multi(Property ... properties)
+ private static SomeStatus multi(Property ... properties)
{
- int encoded = 0;
+ short encoded = 0;
for (Property property : properties)
encoded |= transitiveClosure(property);
- return new RedundantStatus(encoded);
+ return new SomeStatus(encoded);
}
- private static int transitiveClosure(Property property)
+ private static short transitiveClosure(Property property)
{
- int encoded = encode(property);
+ short encoded = encodeAny(property);
for (Property implied : property.implies)
encoded |= transitiveClosure(implied);
return encoded;
}
- static int encode(Property property)
+ static short encodeAny(Property property)
{
- return 0x3 << (property.shift());
+ return (short) (1 << property.shift());
+ }
+
+ static int encodeAll(Property property)
+ {
+ return ALL.mask << property.shift();
+ }
+
+ static int toAll(short encodedKeyStatus)
+ {
+ return encodedKeyStatus | (((int)encodedKeyStatus)<<16);
+ }
+
+ static RedundantStatus toAll(SomeStatus key)
+ {
+ return new RedundantStatus(toAll(key.encoded));
}
- private static final Coverage[] COVERAGE_TO_STRING = new Coverage[] { ALL, SOME };
@Override
public String toString()
{
diff --git a/accord-core/src/main/java/accord/local/RejectBefore.java b/accord-core/src/main/java/accord/local/RejectBefore.java
index d3d12c28..67b072da 100644
--- a/accord-core/src/main/java/accord/local/RejectBefore.java
+++ b/accord-core/src/main/java/accord/local/RejectBefore.java
@@ -22,16 +22,18 @@ import java.util.Objects;
import accord.api.RoutingKey;
import accord.primitives.AbstractRanges;
+import accord.primitives.Ranges;
import accord.primitives.Routables;
+import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.ReducingIntervalMap;
import accord.utils.ReducingRangeMap;
-public class RejectBefore extends ReducingRangeMap<TxnId>
+public class RejectBefore extends ReducingRangeMap<Timestamp>
{
public static class SerializerSupport
{
- public static RejectBefore create(boolean inclusiveEnds, RoutingKey[] ends, TxnId[] values)
+ public static RejectBefore create(boolean inclusiveEnds, RoutingKey[] ends, Timestamp[] values)
{
return new RejectBefore(inclusiveEnds, ends, values);
}
@@ -42,12 +44,12 @@ public class RejectBefore extends ReducingRangeMap<TxnId>
super();
}
- protected RejectBefore(boolean inclusiveEnds, RoutingKey[] starts, TxnId[] values)
+ protected RejectBefore(boolean inclusiveEnds, RoutingKey[] starts, Timestamp[] values)
{
super(inclusiveEnds, starts, values);
}
- public static RejectBefore add(RejectBefore existing, AbstractRanges ranges, TxnId value)
+ public static RejectBefore add(RejectBefore existing, Ranges ranges, TxnId value)
{
RejectBefore add = create(ranges, value);
return merge(existing, add);
@@ -58,17 +60,17 @@ public class RejectBefore extends ReducingRangeMap<TxnId>
return null == foldl(participants, (rejectIfBefore, test) -> rejects(rejectIfBefore, test) ? null : test, txnId, Objects::isNull);
}
- private static boolean rejects(TxnId rejectIfBefore, TxnId test)
+ private static boolean rejects(Timestamp rejectIfBefore, TxnId test)
{
return rejectIfBefore.compareSimultaneousEpochAndHlc(test) >= 0;
}
public static RejectBefore merge(RejectBefore historyLeft, RejectBefore historyRight)
{
- return ReducingIntervalMap.merge(historyLeft, historyRight, TxnId::max, Builder::new);
+ return ReducingIntervalMap.merge(historyLeft, historyRight, Timestamp::mergeMax, Builder::new);
}
- public static RejectBefore create(AbstractRanges ranges, TxnId value)
+ public static RejectBefore create(AbstractRanges ranges, Timestamp value)
{
if (value == null)
throw new IllegalArgumentException("value is null");
@@ -79,7 +81,7 @@ public class RejectBefore extends ReducingRangeMap<TxnId>
return create(ranges, value, Builder::new);
}
- static class Builder extends AbstractBoundariesBuilder<RoutingKey, TxnId, RejectBefore>
+ static class Builder extends AbstractBoundariesBuilder<RoutingKey, Timestamp, RejectBefore>
{
protected Builder(boolean inclusiveEnds, int capacity)
{
@@ -89,7 +91,7 @@ public class RejectBefore extends ReducingRangeMap<TxnId>
@Override
protected RejectBefore buildInternal()
{
- return new RejectBefore(inclusiveEnds, starts.toArray(new RoutingKey[0]), values.toArray(new TxnId[0]));
+ return new RejectBefore(inclusiveEnds, starts.toArray(new RoutingKey[0]), values.toArray(new Timestamp[0]));
}
}
}
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 3a0d14ef..47012d48 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -51,9 +51,9 @@ import accord.utils.async.AsyncChain;
import static accord.local.KeyHistory.INCR;
import static accord.local.KeyHistory.NONE;
-import static accord.local.RedundantStatus.LOCALLY_WITNESSED_ONLY;
+import static accord.local.RedundantStatus.SomeStatus.LOCALLY_WITNESSED_ONLY;
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
-import static accord.local.RedundantStatus.Property.SHARD_ONLY_APPLIED;
+import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
import static accord.local.cfk.UpdateUnmanagedMode.REGISTER;
import static accord.primitives.Routable.Domain.Range;
import static accord.primitives.Routables.Slice.Minimal;
@@ -418,7 +418,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
TxnId maxTxnId = txnIdsForKey.get(txnIdsForKey.size() - 1);
// TODO (desired): convert to O(n) merge
RedundantStatus status = redundantBefore.status(maxTxnId, null, key);
- if (!status.all(SHARD_ONLY_APPLIED) || !status.all(LOCALLY_REDUNDANT))
+ if (!status.all(SHARD_APPLIED) || !status.all(LOCALLY_REDUNDANT))
select.set(i);
}
if (select.getSetBitCount() != keys.size())
diff --git a/accord-core/src/main/java/accord/local/StoreParticipants.java b/accord-core/src/main/java/accord/local/StoreParticipants.java
index 05d86d14..67373793 100644
--- a/accord-core/src/main/java/accord/local/StoreParticipants.java
+++ b/accord-core/src/main/java/accord/local/StoreParticipants.java
@@ -114,9 +114,12 @@ public class StoreParticipants
this.stillTouches = stillTouches;
if (Invariants.isParanoid())
{
- Invariants.require(stillWaitsOn == null || stillTouches.containsAll(stillWaitsOn));
- Invariants.require(stillExecutes == null || stillOwns.containsAll(stillExecutes));
- Invariants.require(stillTouches.containsAll(stillOwns));
+ Invariants.requireArgument(stillWaitsOn == null || stillTouches.containsAll(stillWaitsOn));
+ Invariants.requireArgument(stillExecutes == null || stillOwns.containsAll(stillExecutes));
+ Invariants.requireArgument(executes == null || executes.containsAll(stillExecutes));
+ Invariants.requireArgument(touches.containsAll(stillTouches));
+ Invariants.requireArgument(owns.containsAll(stillOwns));
+ Invariants.requireArgument(stillTouches.containsAll(stillOwns));
}
}
@@ -267,6 +270,12 @@ public class StoreParticipants
return executesIsNull ? null : owns;
}
+ public final boolean doesStillExecute()
+ {
+ Participants<?> executes = stillExecutes();
+ return executes != null && !executes.isEmpty();
+ }
+
/**
* If set, the keys we are known to still execute (i.e. excluding any that are pre-bootstrap or stale)
* @return
@@ -344,12 +353,10 @@ public class StoreParticipants
Participants<?> curTouches = touches();
Participants<?> touches = curTouches;
if (filter == UPDATE)
- touches = redundantBefore.expectToCalculateDependenciesOrConsultOnRecovery(txnId, executeAtIfKnown, curTouches);
+ touches = filterTouches(txnId, redundantBefore, owns, touches);
Participants<?> curStillTouches = stillTouches();
- Participants<?> stillTouches = redundantBefore.expectToOwnOrExecuteOrConsultOnRecovery(txnId, executeAtIfKnown, curStillTouches);
- if (stillTouches != curStillTouches && owns() == curStillTouches && stillTouches.equals(owns))
- stillTouches = owns;
+ Participants<?> stillTouches = filterTouches(txnId, redundantBefore, stillOwns, curStillTouches);
if (curStillOwns != stillOwns || curStillTouches != stillTouches || curStillExecutes != stillExecutes || curStillWaitsOn != stillWaitsOn)
return new FilteredStoreParticipants(route, owns, executes(), waitsOn(), touches, hasTouched(), stillOwns, stillExecutes, stillWaitsOn, stillTouches);
@@ -360,6 +367,22 @@ public class StoreParticipants
return this;
}
+ private static Participants<?> filterTouches(TxnId txnId, RedundantBefore redundantBefore, Participants<?> owns, Participants<?> touches)
+ {
+ if (touches != owns)
+ {
+ Participants<?> extraTouches = touches.without(owns);
+ if (!extraTouches.isEmpty())
+ {
+ Participants<?> filteredExtra = redundantBefore.withoutShardApplied(txnId, extraTouches);
+ if (extraTouches != filteredExtra)
+ touches = owns.with((Participants)filteredExtra);
+ }
+ }
+
+ return touches;
+ }
+
public final boolean touches(RoutingKey key)
{
return touches().contains(key);
@@ -496,6 +519,7 @@ public class StoreParticipants
return Objects.equals(route(), that.route())
&& owns().equals(that.owns())
&& Objects.equals(executes(), that.executes())
+ && Objects.equals(waitsOn(), that.waitsOn())
&& touches().equals(that.touches())
&& hasTouched().equals(that.hasTouched());
}
diff --git a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
index 724c4dfb..d6b45dfb 100644
--- a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
+++ b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
@@ -885,7 +885,6 @@ public class CommandsForKey extends CommandsForKeyUpdate
FROM_SAVE_STATUS.put(SaveStatus.PreApplied, STABLE);
FROM_SAVE_STATUS.put(SaveStatus.Applying, STABLE);
FROM_SAVE_STATUS.put(SaveStatus.Applied, APPLIED_NOT_DURABLE);
- FROM_SAVE_STATUS.put(SaveStatus.TruncatedApplyWithOutcomeAndDeps, APPLIED_DURABLE);
// We don't map TruncatedApplyX or Erased as we want to retain them as APPLIED
// esp. to support pruning where we expect the prunedBefore entr*ies* to be APPLIED
// Note importantly that we have multiple logical pruned befores - the last APPLIED
diff --git a/accord-core/src/main/java/accord/local/cfk/Pruning.java b/accord-core/src/main/java/accord/local/cfk/Pruning.java
index b5ed632c..5cb9476b 100644
--- a/accord-core/src/main/java/accord/local/cfk/Pruning.java
+++ b/accord-core/src/main/java/accord/local/cfk/Pruning.java
@@ -544,7 +544,7 @@ public class Pruning
{
int startPos = prevBootstrappedAt == null ? 0 : insertPos(byId, prevBootstrappedAt);
for (int i = startPos ; i < pos ; ++i)
- Invariants.require(byId[i].isNot(COMMITTED), "%s expected to be applied or undecided, as marked redundant", byId[i]);
+ Invariants.require(byId[i].isNot(COMMITTED) || !byId[i].mayExecute(), "%s redundant; expected to be applied, undecided or to execute in a future epoch", byId[i]);
}
newById = Arrays.copyOfRange(byId, pos, byId.length);
diff --git a/accord-core/src/main/java/accord/local/durability/DurabilityQueue.java b/accord-core/src/main/java/accord/local/durability/DurabilityQueue.java
index 773e486a..431a396c 100644
--- a/accord-core/src/main/java/accord/local/durability/DurabilityQueue.java
+++ b/accord-core/src/main/java/accord/local/durability/DurabilityQueue.java
@@ -48,6 +48,7 @@ import accord.primitives.Ranges;
import accord.primitives.Route;
import accord.primitives.SyncPoint;
import accord.primitives.TxnId;
+import accord.topology.TopologyManager;
import accord.utils.Invariants;
import accord.utils.async.AsyncResult;
import org.agrona.collections.ObjectHashSet;
@@ -257,12 +258,14 @@ public class DurabilityQueue
if (fail != null)
{
if (logger.isTraceEnabled()) logger.trace("{}: failed awaiting durability for {}{}.", txnId, ranges, requestor, fail);
- if (fail.getClass() == SyncPointErased.class)
+ if (fail instanceof SyncPointErased || fail instanceof TopologyManager.TopologyRetiredException)
{
// we can't succeed. if this was requested, and the request is still waiting, submit another coordination request
// TODO (required): add back-off and expand this to all unknown exception outcomes
if (request != null)
node.durability().shards().request(request, request.stillWaiting(exclusiveSyncPoint.route));
+ success();
+ return;
}
}
if (success == null || (success.achievedRemote.compareTo(request == null ? All : request.remote) < 0))
diff --git a/accord-core/src/main/java/accord/messages/Propagate.java b/accord-core/src/main/java/accord/messages/Propagate.java
index 7a24b028..3597b390 100644
--- a/accord-core/src/main/java/accord/messages/Propagate.java
+++ b/accord-core/src/main/java/accord/messages/Propagate.java
@@ -30,6 +30,7 @@ import accord.local.PreLoadContext;
import accord.local.RedundantStatus;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
+import accord.primitives.Known.Outcome;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.Known;
@@ -59,7 +60,8 @@ import static accord.local.Cleanup.ERASE;
import static accord.local.Cleanup.VESTIGIAL;
import static accord.local.Commands.purge;
import static accord.local.RedundantStatus.Property.LOCALLY_DEFUNCT;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_SYNCED;
+import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
+import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
import static accord.local.StoreParticipants.Filter.UPDATE;
import static accord.primitives.Known.KnownDeps.DepsKnown;
import static accord.primitives.Known.KnownDeps.DepsUnknown;
@@ -225,7 +227,7 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
boolean isShardTruncated = withQuorum == HasQuorum && known.hasAnyFullyTruncated(participants.stillTouches());
if (isShardTruncated)
{
- found = tryUpgradeTruncated(safeStore, safeCommand, participants, command, executeAtIfKnown);
+ found = tryUpgradeTruncated(safeStore, safeCommand, participants, command, executeAtIfKnown, found);
if (found == null)
{
// TODO (expected): should be ownsOrExecutes()?
@@ -353,7 +355,7 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
}
// if can only propagate Truncated, we might be stale; try to upgrade for this command store only, even partially if necessary
- private Known tryUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, StoreParticipants participants, Command command, Timestamp executeAtIfKnown)
+ private Known tryUpgradeTruncated(SafeCommandStore safeStore, SafeCommand safeCommand, StoreParticipants participants, Command command, Timestamp executeAtIfKnown, Known found)
{
// 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
@@ -382,18 +384,19 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
Participants<?> staleOwnsOrMayExecute = stillOwnsOrMayExecute.without(notStaleOwnsOrMayExecutes);
if (staleOwnsOrMayExecute.isEmpty())
{
- if (staleTouches.isEmpty())
+ if (staleTouches.isEmpty() && found.is(Outcome.Apply))
{
Invariants.require(notStaleTouches.containsAll(stillTouches));
Invariants.require(notStaleOwnsOrMayExecutes.containsAll(stillOwnsOrMayExecute));
return required;
}
- if (stillOwnsOrMayExecute.isEmpty() && known.hasFullyTruncated(staleTouches))
+ if (stillOwnsOrMayExecute.isEmpty() && (!found.is(Outcome.Apply) || known.hasFullyTruncated(staleTouches)))
{
Commands.setTruncatedOrVestigial(safeStore, safeCommand, participants);
return null;
}
+ Invariants.require(!staleTouches.isEmpty());
}
Participants<?> stale = staleTouches.with((Participants) staleOwnsOrMayExecute);
@@ -422,7 +425,7 @@ public class Propagate implements PreLoadContext, MapReduceConsume<SafeCommandSt
private boolean tryPurge(SafeCommandStore safeStore, SafeCommand safeCommand, RedundantStatus status)
{
- if (!status.all(SHARD_APPLIED_AND_LOCALLY_SYNCED))
+ if (!status.all(SHARD_APPLIED, LOCALLY_SYNCED))
return false;
Cleanup cleanup = status.all(LOCALLY_DEFUNCT) ? VESTIGIAL : ERASE;
diff --git a/accord-core/src/main/java/accord/messages/ReadData.java b/accord-core/src/main/java/accord/messages/ReadData.java
index 6304f9a4..2e7f4cf9 100644
--- a/accord-core/src/main/java/accord/messages/ReadData.java
+++ b/accord-core/src/main/java/accord/messages/ReadData.java
@@ -581,11 +581,6 @@ public abstract class ReadData implements PreLoadContext, Request, MapReduceCons
public enum CommitOrReadNack implements ReadData.ReadReply
{
- /**
- * The commit has been rejected due to stale ballot.
- */
- Rejected("CommitRejected", true),
-
/**
* Either not committed, or not stable
*/
@@ -596,7 +591,13 @@ public abstract class ReadData implements PreLoadContext, Request, MapReduceCons
*/
Waiting("ReadOrApplyWaiting", false),
- Redundant("CommitOrReadRedundant", true);
+ Redundant("CommitOrReadRedundant", true),
+
+ /**
+ * The commit has been rejected due to stale ballot.
+ */
+ Rejected("CommitRejected", true),
+ ;
final String fullname;
final boolean isFinal;
diff --git a/accord-core/src/main/java/accord/messages/SetShardDurable.java b/accord-core/src/main/java/accord/messages/SetShardDurable.java
index 3123c63a..eea0bb66 100644
--- a/accord-core/src/main/java/accord/messages/SetShardDurable.java
+++ b/accord-core/src/main/java/accord/messages/SetShardDurable.java
@@ -21,6 +21,7 @@ package accord.messages;
import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
import accord.primitives.AbstractRanges;
+import accord.primitives.Status.Durability;
import accord.primitives.SyncPoint;
import accord.primitives.TxnId;
import accord.utils.MapReduceConsume;
@@ -34,11 +35,13 @@ public class SetShardDurable extends AbstractRequest<SimpleReply>
implements Request, PreLoadContext, MapReduceConsume<SafeCommandStore, SimpleReply>
{
public final SyncPoint exclusiveSyncPoint;
+ public final Durability durability;
- public SetShardDurable(SyncPoint exclusiveSyncPoint)
+ public SetShardDurable(SyncPoint exclusiveSyncPoint, Durability durability)
{
super(exclusiveSyncPoint.syncId);
this.exclusiveSyncPoint = exclusiveSyncPoint;
+ this.durability = durability;
}
private TxnId syncIdWithFlags()
@@ -61,7 +64,7 @@ public class SetShardDurable extends AbstractRequest<SimpleReply>
@Override
public SimpleReply apply(SafeCommandStore safeStore)
{
- safeStore.commandStore().markShardDurable(safeStore, syncIdWithFlags(), ((AbstractRanges) exclusiveSyncPoint.route).toRanges());
+ safeStore.commandStore().markShardDurable(safeStore, syncIdWithFlags(), ((AbstractRanges) exclusiveSyncPoint.route).toRanges(), durability);
return Ok;
}
diff --git a/accord-core/src/main/java/accord/messages/StableThenRead.java b/accord-core/src/main/java/accord/messages/StableThenRead.java
index d887c714..1fcd18fe 100644
--- a/accord-core/src/main/java/accord/messages/StableThenRead.java
+++ b/accord-core/src/main/java/accord/messages/StableThenRead.java
@@ -91,7 +91,11 @@ public class StableThenRead extends ReadData
Route<?> route = this.route == null ? (Route)scope : this.route;
StoreParticipants participants = StoreParticipants.execute(safeStore, route, txnId, minEpoch(), executeAtEpoch);
SafeCommand safeCommand = safeStore.get(txnId, participants);
- Commands.commit(safeStore, safeCommand, participants, kind.saveStatus, Ballot.ZERO, txnId, route, partialTxn, executeAt, partialDeps, kind);
+ if (Commands.CommitOutcome.Rejected == Commands.commit(safeStore, safeCommand, participants, kind.saveStatus, Ballot.ZERO, txnId, route, partialTxn, executeAt, partialDeps, kind))
+ {
+ cancel();
+ return CommitOrReadNack.Rejected;
+ }
if (participants.executes().isEmpty())
return null;
return super.apply(safeStore, safeCommand, participants);
diff --git a/accord-core/src/main/java/accord/primitives/Deps.java b/accord-core/src/main/java/accord/primitives/Deps.java
index 0cf9d036..ae62b949 100644
--- a/accord-core/src/main/java/accord/primitives/Deps.java
+++ b/accord-core/src/main/java/accord/primitives/Deps.java
@@ -25,6 +25,7 @@ import accord.utils.IndexedFunction;
import accord.utils.Invariants;
import accord.utils.MergeFewDisjointSortedListsCursor;
import accord.utils.RelationMultiMap;
+import accord.utils.SortedArrays;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.SortedList;
import accord.utils.SortedList.MergeCursor;
@@ -32,6 +33,7 @@ import accord.utils.TriFunction;
import accord.utils.UnhandledEnum;
import java.util.AbstractList;
+import java.util.Arrays;
import java.util.List;
import java.util.function.BiFunction;
import java.util.function.Consumer;
@@ -113,7 +115,7 @@ public class Deps
public AbstractBuilder<T> add(Unseekable keyOrRange, TxnId txnId)
{
- Invariants.requireArgument(keyOrRange.domain() == txnId.domain(), keyOrRange + " is not same domain as " + txnId);
+ Invariants.requireArgument(keyOrRange.domain() == txnId.domain(), "%s is not same domain as %s", keyOrRange, txnId);
switch (txnId.domain())
{
default: throw new AssertionError();
@@ -282,6 +284,26 @@ public class Deps
};
}
+ public Deps intersectStable(Deps that, TxnId until)
+ {
+ return new Deps(keyDeps.withTxnIds(intersectStable(keyDeps.txnIds, that.keyDeps.txnIds, until)),
+ rangeDeps.withTxnIds(intersectStable(rangeDeps.txnIds, that.rangeDeps.txnIds, until)));
+ }
+
+ static TxnId[] intersectStable(TxnId[] keep, TxnId[] intersectStable, TxnId until)
+ {
+ if (keep.length == 0)
+ return keep;
+
+ int updateEnd = keep.length;
+ if (keep[updateEnd - 1].compareTo(until) >= 0)
+ {
+ updateEnd = Arrays.binarySearch(keep, until);
+ if (updateEnd < 0) updateEnd = -1 - updateEnd;
+ }
+ return SortedArrays.linearUpdate(keep, 0, updateEnd, keep.length, intersectStable, 0, intersectStable.length, TxnId::compareTo, TxnId::propagateUnstable, TxnId[]::new);
+ }
+
public int indexOf(TxnId txnId)
{
switch (txnId.domain())
diff --git a/accord-core/src/main/java/accord/primitives/KeyDeps.java b/accord-core/src/main/java/accord/primitives/KeyDeps.java
index fea955cd..4aea2a14 100644
--- a/accord-core/src/main/java/accord/primitives/KeyDeps.java
+++ b/accord-core/src/main/java/accord/primitives/KeyDeps.java
@@ -25,6 +25,7 @@ import accord.utils.IndexedBiConsumer;
import accord.utils.IndexedConsumer;
import accord.utils.IndexedFunction;
import accord.utils.IndexedTriConsumer;
+import accord.utils.Invariants;
import accord.utils.RelationMultiMap;
import accord.utils.SimpleBitSet;
import accord.utils.SortedArrays.SortedArrayList;
@@ -210,6 +211,14 @@ public class KeyDeps implements Iterable<Map.Entry<RoutingKey, TxnId>>, KeyOrRan
return select(toRoutingKeys(select));
}
+ KeyDeps withTxnIds(TxnId[] txnIds)
+ {
+ if (txnIds == this.txnIds)
+ return this;
+ Invariants.require(txnIds.length == this.txnIds.length);
+ return new KeyDeps(keys, txnIds, keysToTxnIds, txnIdsToKeys);
+ }
+
public @Nullable TxnId minTxnId(Unseekables<?> participants)
{
return Routables.foldl(keys, participants, (rk, min, index) -> {
diff --git a/accord-core/src/main/java/accord/primitives/PartialDeps.java b/accord-core/src/main/java/accord/primitives/PartialDeps.java
index 72d98a9a..75c1ce37 100644
--- a/accord-core/src/main/java/accord/primitives/PartialDeps.java
+++ b/accord-core/src/main/java/accord/primitives/PartialDeps.java
@@ -107,6 +107,12 @@ public class PartialDeps extends Deps
return new PartialDeps(this.covering.intersecting(participants, Minimal), keyDeps.intersecting(participants), rangeDeps.intersecting(participants));
}
+ public PartialDeps intersectStable(Deps that, TxnId until)
+ {
+ return new PartialDeps(covering, keyDeps.withTxnIds(intersectStable(keyDeps.txnIds, that.keyDeps.txnIds, until)),
+ rangeDeps.withTxnIds(intersectStable(rangeDeps.txnIds, that.rangeDeps.txnIds, until)));
+ }
+
public Deps asFullUnsafe()
{
return new Deps(keyDeps, rangeDeps);
diff --git a/accord-core/src/main/java/accord/primitives/RangeDeps.java b/accord-core/src/main/java/accord/primitives/RangeDeps.java
index 5ab90c16..67e2bf31 100644
--- a/accord-core/src/main/java/accord/primitives/RangeDeps.java
+++ b/accord-core/src/main/java/accord/primitives/RangeDeps.java
@@ -665,6 +665,14 @@ public class RangeDeps implements Iterable<Map.Entry<Range, TxnId>>, KeyOrRangeD
}
}
+ public RangeDeps withTxnIds(TxnId[] txnIds)
+ {
+ if (txnIds == this.txnIds)
+ return this;
+ Invariants.require(txnIds.length == this.txnIds.length);
+ return new RangeDeps(ranges, txnIds, rangesToTxnIds, txnIdsToRanges);
+ }
+
private RangeDeps build(RangeAndMapCollector collector)
{
if (collector.rangesCount == 0)
diff --git a/accord-core/src/main/java/accord/primitives/SaveStatus.java b/accord-core/src/main/java/accord/primitives/SaveStatus.java
index fb6091f8..3fa71119 100644
--- a/accord-core/src/main/java/accord/primitives/SaveStatus.java
+++ b/accord-core/src/main/java/accord/primitives/SaveStatus.java
@@ -106,7 +106,6 @@ public enum SaveStatus
Applied (Status.Applied, LocalExecution.Applied),
// TruncatedApplyWithOutcomeAndDeps exists to support re-populating CommandsForKey on replay with any dependencies needed for computing recovery superseding-rejects decisions
// TODO (expected): test replay
- TruncatedApplyWithOutcomeAndDeps(Status.Truncated, FullRoute, DefinitionErased, ApplyAtKnown, DepsKnown, Apply, CleaningUp),
TruncatedApplyWithOutcome (Status.Truncated, FullRoute, DefinitionErased, ApplyAtKnown, DepsErased, Apply, CleaningUp),
TruncatedApply (Status.Truncated, MaybeRoute, DefinitionErased, ApplyAtKnown, DepsErased, WasApply,CleaningUp),
TruncatedUnapplied (Status.Truncated, MaybeRoute, DefinitionErased, ExecuteAtKnown, DepsErased, WasApply,CleaningUp),
@@ -386,10 +385,6 @@ public enum SaveStatus
if (!known.is(DepsKnown))
return TruncatedApplyWithOutcome;
- case TruncatedApplyWithOutcomeAndDeps:
- if (!known.is(DefinitionKnown))
- return TruncatedApplyWithOutcome;
-
return Applied;
}
}
diff --git a/accord-core/src/main/java/accord/primitives/TxnId.java b/accord-core/src/main/java/accord/primitives/TxnId.java
index 053e4801..0f611592 100644
--- a/accord-core/src/main/java/accord/primitives/TxnId.java
+++ b/accord-core/src/main/java/accord/primitives/TxnId.java
@@ -31,6 +31,7 @@ import accord.primitives.Txn.Kind.Kinds;
import accord.utils.Invariants;
import accord.utils.TinyEnumSet;
+import static accord.primitives.Timestamp.Flag.UNSTABLE;
import static accord.primitives.Txn.Kind.EphemeralRead;
import static accord.primitives.Txn.Kind.Read;
import static accord.primitives.Txn.Kind.Write;
@@ -394,6 +395,11 @@ public class TxnId extends Timestamp implements PreLoadContext
return addFlags(flag.bit);
}
+ public TxnId propagateUnstable(@Nullable TxnId ifNullOrUnstable)
+ {
+ return ifNullOrUnstable == null || ifNullOrUnstable.is(UNSTABLE) ? addFlag(UNSTABLE) : this;
+ }
+
public final TxnId addFlags(TxnId merge)
{
return addFlags(this, merge, TxnId::new);
diff --git a/accord-core/src/main/java/accord/topology/TopologyManager.java b/accord-core/src/main/java/accord/topology/TopologyManager.java
index 4c22806a..39943296 100644
--- a/accord-core/src/main/java/accord/topology/TopologyManager.java
+++ b/accord-core/src/main/java/accord/topology/TopologyManager.java
@@ -999,7 +999,7 @@ public class TopologyManager
public <U extends Participants<?>> @Nullable U unsyncedOnly(U select, long beforeEpoch)
{
- return extra(select, 0, beforeEpoch - 1, cur -> cur.synced, (UnsyncedSelector<U>)UnsyncedSelector.INSTANCE);
+ return extra(select, 0, beforeEpoch, cur -> cur.synced, (UnsyncedSelector<U>)UnsyncedSelector.INSTANCE);
}
public Topologies withUnsyncedEpochs(Unseekables<?> select, long minEpoch, long maxEpoch)
@@ -1044,6 +1044,8 @@ public class TopologyManager
{
Invariants.requireArgument(minEpoch <= maxEpoch);
Epochs snapshot = epochs;
+ if (maxEpoch < snapshot.minEpoch())
+ throw new TopologyRetiredException(maxEpoch, snapshot.minEpoch());
if (maxEpoch == Long.MAX_VALUE) maxEpoch = snapshot.currentEpoch;
else Invariants.require(snapshot.currentEpoch >= maxEpoch, "current epoch %d < max %d", snapshot.currentEpoch, maxEpoch);
@@ -1160,18 +1162,15 @@ public class TopologyManager
Epochs snapshot = epochs;
minEpoch = Math.max(snapshot.minEpoch(), minEpoch);
- if (maxEpoch < minEpoch)
- return collectors.none();
maxEpoch = validateMax(maxEpoch, snapshot);
+ if (maxEpoch == minEpoch)
+ return collectors.none();
EpochState cur = nonNull(snapshot.get(maxEpoch));
select = (K) select.without(remove.apply(cur));
if (select.isEmpty())
return collectors.none();
- if (minEpoch == maxEpoch)
- return collectors.one(cur, select, true);
-
int i = (int)(snapshot.currentEpoch - maxEpoch);
int maxi = (int)(Math.min(1 + snapshot.currentEpoch - minEpoch, snapshot.epochs.length));
C collector = collectors.allocate(maxi - i);
diff --git a/accord-core/src/main/java/accord/utils/ArrayBuffers.java b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
index 92446978..3b5527ee 100644
--- a/accord-core/src/main/java/accord/utils/ArrayBuffers.java
+++ b/accord-core/src/main/java/accord/utils/ArrayBuffers.java
@@ -57,6 +57,7 @@ public class ArrayBuffers
private static final boolean FULLY_UNCACHED = true;
// TODO (low priority, efficiency): we should periodically clear the thread locals to ensure we aren't slowly accumulating unnecessarily large objects on every thread
+ private static final ThreadLocal<ShortBufferCache> SHORTS = ThreadLocal.withInitial(() -> new ShortBufferCache(4, 1 << 10));
private static final ThreadLocal<IntBufferCache> INTS = ThreadLocal.withInitial(() -> new IntBufferCache(4, 1 << 14));
private static final ThreadLocal<LongBufferCache> LONGS = ThreadLocal.withInitial(() -> new LongBufferCache(4, 1 << 14));
private static final ThreadLocal<ObjectBufferCache<Key>> KEYS = ThreadLocal.withInitial(() -> new ObjectBufferCache<>(3, 1 << 9, Key[]::new));
@@ -65,6 +66,10 @@ public class ArrayBuffers
private static final ThreadLocal<ObjectBufferCache<TxnId>> TXN_IDS = ThreadLocal.withInitial(() -> new ObjectBufferCache<>(3, 1 << 12, TxnId[]::new));
private static final ThreadLocal<ObjectBufferCache<Object>> OBJECTS = ThreadLocal.withInitial(() -> new ObjectBufferCache<>(3, 1 << 12, Object[]::new));
+ public static ShortBuffers cachedShorts()
+ {
+ return SHORTS.get();
+ }
public static IntBuffers cachedInts()
{
return INTS.get();
@@ -109,6 +114,73 @@ public class ArrayBuffers
public static IntBuffers uncachedInts() { return UncachedIntBuffers.INSTANCE; }
+ public interface ShortBufferAllocator
+ {
+ /**
+ * Return an {@code short[]} of size at least {@code minSize}, possibly from a pool.
+ * This array may not be zero initialized, and its contents should be treated as random.
+ */
+ short[] getShorts(int minSize);
+ }
+
+ public interface ShortBuffers extends ShortBufferAllocator
+ {
+ /**
+ * Return an {@code int[]} of size at least {@code minSize}, possibly from a pool,
+ * and copy the contents of {@code copyAndDiscard} into it.
+ *
+ * The remainder of the array may not be zero-initialized, and should be assumed to contain random data.
+ *
+ * The parameter will be returned to the pool, if eligible.
+ */
+ default short[] resize(short[] copyAndDiscard, int usedSize, int minSize)
+ {
+ short[] newBuf = getShorts(minSize);
+ System.arraycopy(copyAndDiscard, 0, newBuf, 0, usedSize);
+ forceDiscard(copyAndDiscard);
+ return newBuf;
+ }
+
+ /**
+ * To be invoked on the result buffer with the number of elements contained;
+ * either the buffer will be returned and the size optionally captured, or else the result may be
+ * shrunk to the size of the contents, depending on implementation.
+ */
+ short[] complete(short[] buffer, int usedSize);
+
+ /**
+ * The buffer is no longer needed by the caller, which is discarding the array;
+ * if {@link #complete(short[], int)} returned the buffer as its result this buffer should NOT be
+ * returned to any pool.
+ *
+ * Note that this method assumes {@link #complete(short[], int)} was invoked on this buffer previously.
+ * However, it is guaranteed that a failure to do so does not leak memory or pool space, only produces some
+ * additional garbage.
+ *
+ * @return true if the buffer is discarded (and discard-able), false if it was retained or is believed to be in use
+ */
+ boolean discard(short[] buffer, int usedSize);
+
+ /**
+ * Equivalent to
+ * int[] result = complete(buffer, usedSize);
+ * discard(buffer, usedSize);
+ * return result;
+ */
+ default short[] completeAndDiscard(short[] buffer, int usedSize)
+ {
+ short[] result = complete(buffer, usedSize);
+ discard(buffer, usedSize);
+ return result;
+ }
+
+ /**
+ * Indicate this buffer is definitely unused, and return it to a pool if possible
+ * @return true if the buffer is discarded (and discard-able), false if it was retained
+ */
+ boolean forceDiscard(short[] buffer);
+ }
+
public interface IntBufferAllocator
{
/**
@@ -502,6 +574,41 @@ public class ArrayBuffers
}
}
+ public static class ShortBufferCache extends AbstractBufferCache<short[]> implements ShortBuffers
+ {
+ ShortBufferCache(int maxCount, int maxSize)
+ {
+ super(short[]::new, (array, size) -> {}, maxCount, maxSize);
+ }
+
+ @Override
+ public short[] complete(short[] buffer, int usedSize)
+ {
+ if (usedSize == buffer.length)
+ return buffer;
+
+ return Arrays.copyOf(buffer, usedSize);
+ }
+
+ @Override
+ public boolean discard(short[] buffer, int usedSize)
+ {
+ return discardInternal(buffer, buffer.length, usedSize, false);
+ }
+
+ @Override
+ public boolean forceDiscard(short[] buffer)
+ {
+ return discardInternal(buffer, buffer.length, -1, true);
+ }
+
+ @Override
+ public short[] getShorts(int minSize)
+ {
+ return getInternal(minSize);
+ }
+ }
+
public static class IntBufferCache extends AbstractBufferCache<int[]> implements IntBuffers
{
IntBufferCache(int maxCount, int maxSize)
diff --git a/accord-core/src/main/java/accord/utils/SortedArrays.java b/accord-core/src/main/java/accord/utils/SortedArrays.java
index d63159a9..70c60bab 100644
--- a/accord-core/src/main/java/accord/utils/SortedArrays.java
+++ b/accord-core/src/main/java/accord/utils/SortedArrays.java
@@ -447,6 +447,86 @@ public class SortedArrays
}
}
+ // applies the update function to all entries in left, and any matching entries in right
+ public static <O, T extends O> O[] linearUpdate(T[] left, int leftStart, int leftUpdateEnd, int leftEnd, T[] right, int rightStart, int rightEnd, AsymmetricComparator<? super T, ? super T> comparator, BiFunction<? super T, ? super T, ? extends T> update, IntFunction<T[]> allocate)
+ {
+ int leftIdx = leftStart;
+ int rightIdx = rightStart;
+
+ T nextKey = null;
+ while (leftIdx < leftUpdateEnd && rightIdx < rightEnd)
+ {
+ T leftKey = left[leftIdx];
+ T rightKey = right[rightIdx];
+ int cmp = leftKey == rightKey ? 0 : comparator.compare(leftKey, rightKey);
+ if (cmp > 0)
+ {
+ ++rightIdx;
+ continue;
+ }
+
+ if (cmp < 0) rightKey = null;
+ else ++rightIdx;
+
+ T testKey = update.apply(leftKey, rightKey);
+ if (testKey != leftKey)
+ {
+ nextKey = testKey;
+ break;
+ }
+ ++leftIdx;
+ }
+
+ if (nextKey == null)
+ {
+ while (leftIdx < leftUpdateEnd)
+ {
+ T leftKey = left[leftIdx];
+ T testKey = update.apply(left[leftIdx], null);
+ if (testKey != leftKey)
+ {
+ nextKey = testKey;
+ break;
+ }
+ ++leftIdx;
+ }
+
+ if (nextKey == null)
+ return left;
+ }
+
+ O[] result = allocate.apply(leftEnd - leftStart);
+ int resultSize = leftIdx - leftStart;
+ System.arraycopy(left, leftStart, result, 0, resultSize);
+ result[resultSize++] = nextKey;
+ ++leftIdx;
+
+ while (leftIdx < leftUpdateEnd && rightIdx < rightEnd)
+ {
+ T leftKey = left[leftIdx];
+ T rightKey = right[rightIdx];
+ int cmp = leftKey == rightKey ? 0 : comparator.compare(leftKey, rightKey);
+
+ if (cmp > 0)
+ {
+ ++rightIdx;
+ continue;
+ }
+
+ if (cmp < 0) rightKey = null;
+ else ++rightIdx;
+ ++leftIdx;
+
+ result[resultSize++] = update.apply(leftKey, rightKey);
+ }
+
+ while (leftIdx < leftUpdateEnd)
+ result[resultSize++] = update.apply(left[leftIdx++], null);
+
+ System.arraycopy(left, leftUpdateEnd, result, resultSize, leftEnd - leftUpdateEnd);
+ return result;
+ }
+
/**
* {@link #linearIntersection(Comparable[], int, Comparable[], int, ObjectBuffers)}
*/
diff --git a/accord-core/src/test/java/accord/impl/basic/Cluster.java b/accord-core/src/test/java/accord/impl/basic/Cluster.java
index e34f7e98..1cd03eeb 100644
--- a/accord-core/src/test/java/accord/impl/basic/Cluster.java
+++ b/accord-core/src/test/java/accord/impl/basic/Cluster.java
@@ -581,7 +581,7 @@ public class Cluster
command = command.updateParticipants(command.participants().filter(LOAD, commandStore.unsafeGetRedundantBefore(), command.txnId(), command.executeAtIfKnown()));
// Journal will not have result persisted. This part is here for test purposes and ensuring that we have strict object equality.
Command reconstructed = journal.loadCommand(commandStore.id(), command.txnId(), commandStore.unsafeGetRedundantBefore(), commandStore.durableBefore());
- if (reconstructed.saveStatus().hasBeen(Status.Truncated))
+ if (reconstructed == null || reconstructed.saveStatus().hasBeen(Status.Truncated))
return;
List<ReflectionUtils.Difference<?>> diff = ReflectionUtils.recursiveEquals(command, reconstructed);
@@ -957,7 +957,7 @@ public class Cluster
if (Cleanup.shouldCleanup(FULL, s.agent(), beforeCommand, store.unsafeGetRedundantBefore(), store.durableBefore()) == EXPUNGE)
continue;
- if (store.unsafeGetRedundantBefore().min(beforeCommand.participants().owns(), RedundantBefore.Bounds::shardRedundantBefore).compareTo(txnId) > 0)
+ if (store.unsafeGetRedundantBefore().min(beforeCommand.participants().owns(), RedundantBefore.Bounds::shardAndLocallyRedundantBefore).compareTo(txnId) > 0)
continue;
if (beforeCommand.participants().owns().isEmpty() && store.durableBefore().min(txnId).compareTo(Status.Durability.MajorityOrInvalidated) >= 0)
diff --git a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
index 0d8ebd34..975cc727 100644
--- a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
+++ b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java
@@ -390,6 +390,7 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread
}
}
+ static int counter = 0;
public static class DelayedSafeStore extends InMemoryCommandStore.InMemorySafeStore
{
private final DelayedCommandStore commandStore;
@@ -405,6 +406,7 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread
super(commandStore, ranges, context, commands, commandsForKey);
this.commandStore = commandStore;
this.cacheLoading = cacheLoading;
+ ++counter;
}
@Override
diff --git a/accord-core/src/test/java/accord/impl/basic/InMemoryJournal.java b/accord-core/src/test/java/accord/impl/basic/InMemoryJournal.java
index ade5402c..00f00094 100644
--- a/accord-core/src/test/java/accord/impl/basic/InMemoryJournal.java
+++ b/accord-core/src/test/java/accord/impl/basic/InMemoryJournal.java
@@ -100,7 +100,6 @@ import static accord.local.Cleanup.Input.PARTIAL;
import static accord.local.Cleanup.NO;
import static accord.local.Cleanup.TRUNCATE;
import static accord.local.Cleanup.TRUNCATE_WITH_OUTCOME;
-import static accord.local.Cleanup.TRUNCATE_WITH_OUTCOME_AND_DEPS;
import static accord.primitives.SaveStatus.Erased;
import static accord.primitives.Status.Invalidated;
import static accord.primitives.Status.Truncated;
@@ -137,7 +136,12 @@ public class InMemoryJournal implements Journal
return null;
Builder builder = reconstruct(saved, ALL);
- builder.maybeCleanup(FULL, agent, redundantBefore, durableBefore);
+ if (builder == null)
+ return null;
+
+ Cleanup cleanup = builder.maybeCleanup(FULL, agent, redundantBefore, durableBefore);
+ if (cleanup == EXPUNGE)
+ return null;
return builder.construct(redundantBefore);
}
@@ -310,6 +314,7 @@ public class InMemoryJournal implements Journal
onFlush.run();
}
+ static int counter = 0;
@Override
public void purge(CommandStores commandStores, EpochSupplier minEpoch)
{
@@ -326,37 +331,38 @@ public class InMemoryJournal implements Journal
for (Map.Entry<TxnId, List<Diff>> e2 : localJournal.entrySet())
{
List<Diff> diffs = e2.getValue();
- int from = 0;
+ int from;
if (diffs.isEmpty()) continue;
+ List<Diff> subset = diffs;
if (isPartialCompaction)
{
- List<Diff> copy = new ArrayList<>();
+ subset = new ArrayList<>();
int tmp1 = random.nextInt(diffs.size());
int tmp2 = random.nextInt(diffs.size());
from = Math.min(tmp1, tmp2);
int max = Math.max(tmp1, tmp2);
for (int i = from; i < max; i++)
- copy.add(diffs.get(i));
-
- if (copy.isEmpty())
- continue;
- diffs = copy;
+ subset.add(diffs.get(i));
}
- InMemoryJournal.Builder builder = reconstruct(diffs, ALL);
- if (builder.saveStatus() == null)
+
+ InMemoryJournal.Builder builder = reconstruct(subset, ALL);
+ if (builder == null || builder.saveStatus() == null)
continue; // Partial compaction, no save status present
if (builder.saveStatus().status == Truncated || builder.saveStatus().status == Invalidated)
continue; // Already truncated
Input input = isPartialCompaction ? PARTIAL : FULL;
+ ++counter;
Cleanup cleanup = builder.shouldCleanup(input, store.agent(),store.unsafeGetRedundantBefore(), store.durableBefore());
-
- if (builder.maybeCleanup(input, cleanup))
+ cleanup = builder.maybeCleanup(input, cleanup);
+ if (cleanup != NO)
{
- Invariants.require(cleanup != NO);
if (cleanup == EXPUNGE)
- e2.setValue(new PurgedList());
+ {
+ if (input == FULL) e2.setValue(new PurgedList());
+ else diffs.removeAll(subset);
+ }
else
{
int flags = CommandChange.eraseKnownFieldsMask(cleanup.appliesIfNot);
@@ -375,7 +381,6 @@ public class InMemoryJournal implements Journal
}
values.put(SAVE_STATUS, cleanup.appliesIfNot);
values.put(CLEANUP, cleanup);
-
flags = setChanged(SAVE_STATUS, flags);
flags = setChanged(CLEANUP, flags);
@@ -383,9 +388,17 @@ public class InMemoryJournal implements Journal
// During partial compaction, we can only append_to the existing list, removing items that got compacted away
if (isPartialCompaction)
{
- List<Diff> newDiffs = new ArrayList<>(e2.getValue());
- newDiffs.removeIf(diffs::contains);
+ int i = 0, j = 0;
+ while (i < diffs.size() && j < subset.size())
+ {
+ if (subset.get(j) == diffs.get(i))
+ ++j;
+ ++i;
+ }
+
+ List<Diff> newDiffs = new ArrayList<>(diffs.subList(0, i));
newDiffs.add(diff);
+ newDiffs.addAll(diffs.subList(i, diffs.size()));
e2.setValue(newDiffs);
}
// During full compaction, we erase all previous records, replacing them with new image
@@ -452,7 +465,8 @@ public class InMemoryJournal implements Journal
@Override
public boolean add(Diff diff)
{
- if (diff.changes.get(SAVE_STATUS) == Erased)
+ // TODO (expected): we shouldn't really be saving updates (such as durability updates) to Erased commands
+ if (diff.changes.get(SAVE_STATUS) == Erased || diff.changes.get(SAVE_STATUS) == null)
return false;
throw illegalState();
}
@@ -552,10 +566,6 @@ public class InMemoryJournal implements Journal
{
switch (after.saveStatus())
{
- case TruncatedApplyWithOutcomeAndDeps:
- changes.put(CLEANUP, TRUNCATE_WITH_OUTCOME_AND_DEPS);
- flags = setChanged(CLEANUP, unsetFieldIsNull(CLEANUP, flags));
- break;
case TruncatedApplyWithOutcome:
changes.put(CLEANUP, TRUNCATE_WITH_OUTCOME);
flags = setChanged(CLEANUP, unsetFieldIsNull(CLEANUP, flags));
diff --git a/accord-core/src/test/java/accord/local/RedundantBeforeTest.java b/accord-core/src/test/java/accord/local/RedundantBeforeTest.java
index 4e8f7bf7..99eb4c9f 100644
--- a/accord-core/src/test/java/accord/local/RedundantBeforeTest.java
+++ b/accord-core/src/test/java/accord/local/RedundantBeforeTest.java
@@ -30,6 +30,7 @@ import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
import accord.api.RoutingKey;
+import accord.local.RedundantStatus.SomeStatus;
import accord.local.RedundantStatus.Property;
import accord.primitives.Ranges;
import accord.primitives.Routable;
@@ -44,16 +45,9 @@ import org.assertj.core.api.Assertions;
import static accord.impl.IntKey.routing;
import static accord.local.RedundantStatus.ONLY_LE_MASK;
import static accord.local.RedundantStatus.PRE_BOOTSTRAP_MERGE_MASK;
-import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
-import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
-import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP_OR_STALE;
-import static accord.local.RedundantStatus.Property.SHARD_AND_LOCALLY_APPLIED;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_REDUNDANT;
-import static accord.local.RedundantStatus.Property.SHARD_APPLIED_AND_LOCALLY_SYNCED;
-import static accord.local.RedundantStatus.Property.SHARD_ONLY_APPLIED;
-import static accord.local.RedundantStatus.any;
import static accord.local.RedundantStatus.selectOrCreate;
+import static accord.local.RedundantStatus.toAll;
public class RedundantBeforeTest
{
@@ -87,9 +81,9 @@ public class RedundantBeforeTest
static class CanonStatus
{
final List<Unmerged> unmerged;
- RedundantStatus mergedLt, mergedLe;
+ SomeStatus mergedLt, mergedLe;
- CanonStatus(List<Unmerged> unmerged, RedundantStatus mergedLt, RedundantStatus mergedLe)
+ CanonStatus(List<Unmerged> unmerged, SomeStatus mergedLt, SomeStatus mergedLe)
{
this.unmerged = unmerged;
this.mergedLt = mergedLt;
@@ -101,8 +95,8 @@ public class RedundantBeforeTest
List<Unmerged> unmerged = new ArrayList<>();
unmerged.addAll(this.unmerged);
unmerged.addAll(add.unmerged);
- RedundantStatus lt = merge(filter(add.mergedLt, mergedLt), mergedLt);
- RedundantStatus le = merge(filter(add.mergedLe, mergedLe), mergedLe);
+ SomeStatus lt = merge(filter(add.mergedLt, mergedLt), mergedLt);
+ SomeStatus le = merge(filter(add.mergedLe, mergedLe), mergedLe);
return new CanonStatus(unmerged, lt, le);
}
@@ -142,9 +136,9 @@ public class RedundantBeforeTest
int ignore = 0;
for (int j = 0 ; j < innerLoop ; ++j)
{
- RedundantStatus status = RedundantStatus.oneSlow(picker.get());
+ SomeStatus status = RedundantStatus.oneSlow(picker.get());
TxnId txnId = txnIdSupplier.get();
- if (status.any(Property.GC_BEFORE))
+ if (status.is(Property.GC_BEFORE))
txnId = txnId.addFlag(Timestamp.Flag.SHARD_BOUND);
RoutingKey key = keys[rs.nextInt(keys.length)];
@@ -159,7 +153,7 @@ public class RedundantBeforeTest
TxnId txnId = txnIdSupplier.get();
RoutingKey key = keys[rs.nextInt(keys.length)];
- RedundantStatus canonStatus = getCanon(txnId, canon.get(key));
+ RedundantStatus canonStatus = toAll(getCanon(txnId, canon.get(key)));
RedundantBefore.Bounds bounds = redundantBefore.get(key);
RedundantStatus status = bounds == null ? RedundantStatus.NONE : bounds.getIgnoringOwnership(txnId, null);
Assertions.assertThat(status).isEqualTo(canonStatus);
@@ -167,17 +161,17 @@ public class RedundantBeforeTest
}
}
- static void addCanon(TxnId txnId, RedundantStatus status, TreeMap<TxnId, CanonStatus> map)
+ static void addCanon(TxnId txnId, SomeStatus status, TreeMap<TxnId, CanonStatus> map)
{
- RedundantStatus add;
+ SomeStatus add;
{
Map.Entry<TxnId, CanonStatus> e = map.ceilingEntry(txnId);
- RedundantStatus mergele, mergelt;
- if (e == null) mergele = mergelt = RedundantStatus.NONE;
+ SomeStatus mergele, mergelt;
+ if (e == null) mergele = mergelt = SomeStatus.NONE;
else if (e.getKey().equals(txnId)) { mergele = e.getValue().mergedLe; mergelt = e.getValue().mergedLt; }
else mergele = mergelt = e.getValue().mergedLt;
- RedundantStatus le = merge(filter(le(status), mergele), mergele);
- RedundantStatus lt = merge(filter(status, mergelt), mergelt);
+ SomeStatus le = merge(filter(le(status), mergele), mergele);
+ SomeStatus lt = merge(filter(status, mergelt), mergelt);
add = map.merge(txnId, new CanonStatus(Collections.singletonList(new Unmerged(status.encoded, map.size())), lt, le), CanonStatus::add).mergedLt;
}
@@ -193,48 +187,34 @@ public class RedundantBeforeTest
}
}
- private static RedundantStatus getCanon(TxnId txnId, TreeMap<TxnId, CanonStatus> canon)
+ private static SomeStatus getCanon(TxnId txnId, TreeMap<TxnId, CanonStatus> canon)
{
Map.Entry<TxnId, CanonStatus> e = canon.ceilingEntry(txnId);
- if (e == null) return RedundantStatus.NONE;
+ if (e == null) return SomeStatus.NONE;
else if (e.getKey().equals(txnId)) return e.getValue().mergedLe;
else return e.getValue().mergedLt;
}
- private static int enrich(int encoded)
+ private static short merge(short a, short b)
{
- if (!any(encoded, SHARD_ONLY_APPLIED))
- return encoded;
-
- if (any(encoded, LOCALLY_REDUNDANT))
- encoded |= RedundantStatus.encode(SHARD_APPLIED_AND_LOCALLY_REDUNDANT);
- if (any(encoded, LOCALLY_SYNCED))
- encoded |= RedundantStatus.encode(SHARD_APPLIED_AND_LOCALLY_SYNCED);
- if (any(encoded, LOCALLY_APPLIED))
- encoded |= RedundantStatus.encode(SHARD_AND_LOCALLY_APPLIED);
- return encoded;
+ return (short) (a | b);
}
- private static int merge(int a, int b)
- {
- return enrich(a | b);
- }
-
- private static RedundantStatus merge(RedundantStatus a, RedundantStatus b)
+ private static SomeStatus merge(SomeStatus a, SomeStatus b)
{
return RedundantStatus.selectOrCreate(merge(a.encoded, b.encoded), a, b);
}
- private static RedundantStatus filter(RedundantStatus a, RedundantStatus history)
+ private static SomeStatus filter(SomeStatus a, SomeStatus history)
{
- if (history.any(PRE_BOOTSTRAP_OR_STALE))
- return selectOrCreate(a.encoded & PRE_BOOTSTRAP_MERGE_MASK, a, history);
+ if (history.is(PRE_BOOTSTRAP_OR_STALE))
+ return selectOrCreate((short) (a.encoded & PRE_BOOTSTRAP_MERGE_MASK), a, history);
return a;
}
- private static RedundantStatus le(RedundantStatus le)
+ private static SomeStatus le(SomeStatus le)
{
- int masked = le.encoded & ONLY_LE_MASK;
- return masked == le.encoded ? le : new RedundantStatus(masked);
+ short masked = (short) (le.encoded & ONLY_LE_MASK);
+ return masked == le.encoded ? le : new SomeStatus(masked);
}
}
diff --git a/accord-core/src/test/resources/burn-logback.xml b/accord-core/src/test/resources/burn-logback.xml
index 577d5465..4d3f729c 100644
--- a/accord-core/src/test/resources/burn-logback.xml
+++ b/accord-core/src/test/resources/burn-logback.xml
@@ -40,6 +40,7 @@
<logger name="accord.local.durability.DurabilityQueue" level="ERROR"/>
<logger name="accord.local.durability.ShardDurability" level="ERROR"/>
<logger name="accord.local.durability.GlobalDurability" level="ERROR"/>
+ <logger name="accord.local.CommandStore" level="ERROR"/>
<root level="INFO">
<appender-ref ref="FILE"/>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org