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/08/03 14:08:15 UTC
(cassandra-accord) branch trunk updated: Bootstrap must first wait for a quorum to apply the sync point to ensure the DurableBefore Majority condition holds transitively, since later quorums may include the bootstrapping node which does not participate in the durability of preceding transactions Also Improve: - Apply MaxDecidedRX filtering to CommandsForKey RX dependencies - Optimise AbstractRanges.sliceMinimal (burn test hotspot) - Don't start shard schedulers on topology changes if not already started - Only registerTransit [...]
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 d5de79b5 Bootstrap must first wait for a quorum to apply the sync point to ensure the DurableBefore Majority condition holds transitively, since later quorums may include the bootstrapping node which does not participate in the durability of preceding transactions Also Improve: - Apply MaxDecidedRX filtering to CommandsForKey RX dependencies - Optimise AbstractRanges.sliceMinimal (burn test hotspot) - Don't start shard schedulers on topology changes if not already started - O [...]
d5de79b5 is described below
commit d5de79b56f792d3b6d868ca46845b830e8906828
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Thu Jul 31 11:41:21 2025 +0100
Bootstrap must first wait for a quorum to apply the sync point to ensure the DurableBefore Majority condition holds transitively,
since later quorums may include the bootstrapping node which does not participate in the durability of preceding transactions
Also Improve:
- Apply MaxDecidedRX filtering to CommandsForKey RX dependencies
- Optimise AbstractRanges.sliceMinimal (burn test hotspot)
- Don't start shard schedulers on topology changes if not already started
- Only registerTransitive if waitingOnSync
- Reserve DurabilityRequest until ShardDurability is started
- Don't notify progress log if stopped
- Remove duplicate unmanaged filtering
- Include only the first higher RX for each key on recovery
Also fix:
- Use normal machinery for register transitive to ensure durability
- Edge case in notification across bootstrap boundary by resubmitting Unmanaged to be recomputed
- Serialization of CommandsForKey.TxnInfo.missing() collection when non-identity flag bits are present
patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20811
---
.../java/accord/coordinate/ExecuteSyncPoint.java | 27 +-
.../main/java/accord/coordinate/MaybeRecover.java | 2 +
.../coordinate/tracking/AbstractTracker.java | 7 +-
.../coordinate/tracking/DurabilityTracker.java | 12 +
.../java/accord/impl/InMemoryCommandStore.java | 140 +++---
.../impl/progresslog/DefaultProgressLog.java | 3 +
.../src/main/java/accord/local/Bootstrap.java | 44 +-
.../src/main/java/accord/local/CommandStore.java | 87 +++-
.../main/java/accord/local/CommandSummaries.java | 475 ++++++++++++++-------
.../src/main/java/accord/local/Commands.java | 5 +-
.../src/main/java/accord/local/DepsCalculator.java | 56 ++-
.../src/main/java/accord/local/LoadKeysFor.java | 5 +-
.../src/main/java/accord/local/MaxDecidedRX.java | 5 +-
.../src/main/java/accord/local/PreLoadContext.java | 6 +-
.../src/main/java/accord/local/SafeCommand.java | 4 +-
.../main/java/accord/local/SafeCommandStore.java | 66 ++-
.../main/java/accord/local/cfk/CommandsForKey.java | 23 +-
.../main/java/accord/local/cfk/PostProcess.java | 27 +-
.../src/main/java/accord/local/cfk/Pruning.java | 4 +-
.../java/accord/local/cfk/SafeCommandsForKey.java | 2 +-
.../src/main/java/accord/local/cfk/Serialize.java | 92 +++-
.../accord/local/durability/DurabilityService.java | 15 +-
.../accord/local/durability/ShardDurability.java | 48 ++-
.../main/java/accord/primitives/AbstractKeys.java | 47 +-
.../java/accord/primitives/AbstractRanges.java | 92 +++-
.../accord/primitives/AbstractUnseekableKeys.java | 4 +-
.../src/main/java/accord/primitives/KeyRoute.java | 3 +-
.../src/main/java/accord/primitives/Routables.java | 7 +
.../src/main/java/accord/primitives/Timestamp.java | 1 +
.../src/main/java/accord/primitives/TxnId.java | 18 +
.../main/java/accord/primitives/Unseekables.java | 3 +-
.../src/main/java/accord/topology/Shard.java | 5 +
.../main/java/accord/utils/async/AsyncChains.java | 4 +
.../test/java/accord/impl/RemoteListenersTest.java | 2 +-
.../java/accord/local/cfk/CommandsForKeyTest.java | 4 +-
.../accord/utils/BTreeReducingRangeMapTest.java | 3 +-
.../java/accord/utils/ReducingRangeMapTest.java | 4 +-
accord-core/src/test/resources/burn-logback.xml | 1 +
38 files changed, 961 insertions(+), 392 deletions(-)
diff --git a/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java b/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
index a37ff476..357987da 100644
--- a/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
+++ b/accord-core/src/main/java/accord/coordinate/ExecuteSyncPoint.java
@@ -76,7 +76,7 @@ public class ExecuteSyncPoint extends SettableResult<DurabilityResult> implement
final SettableResult<DurabilityResult> onQuorum = new SettableResult<>();
final int attempt;
private Throwable failures;
- boolean reportedQuorum;
+ boolean reportedQuorum, reportedMinorityQuorum;
long retryInFutureEpoch;
protected ExecuteSyncPoint(Node node, SyncPoint<Range> syncPoint, Set<Node.Id> excludeSuccess, AsyncExecutor executor, int attempt)
@@ -185,16 +185,31 @@ public class ExecuteSyncPoint extends SettableResult<DurabilityResult> implement
update(tracker.recordFailure(from));
}
- private void update(RequestStatus status)
+ private void maybeReportQuorums()
{
- if (status == RequestStatus.NoChange)
+ if (!reportedQuorum)
{
- if (tracker.hasQuorumSuccess() && !reportedQuorum)
+ if (tracker.hasQuorumSuccess())
{
DurabilityResult current = current();
onQuorum.trySuccess(current);
node.durability().report(current);
+ reportedQuorum = true;
+ }
+ else if (!reportedMinorityQuorum && tracker.hasMinorityQuorumSuccess())
+ {
+ DurabilityResult current = current();
+ node.durability().report(current);
+ reportedMinorityQuorum = true;
}
+ }
+ }
+
+ private void update(RequestStatus status)
+ {
+ if (status == RequestStatus.NoChange)
+ {
+ maybeReportQuorums();
return;
}
@@ -225,6 +240,10 @@ public class ExecuteSyncPoint extends SettableResult<DurabilityResult> implement
{
node.send(tracker.nodes(), new SetShardDurable(syncPoint, Majority));
}
+ else
+ {
+ maybeReportQuorums();
+ }
node.durability().report(result);
trySuccess(result);
}
diff --git a/accord-core/src/main/java/accord/coordinate/MaybeRecover.java b/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
index 0a04519a..cfe1d182 100644
--- a/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
+++ b/accord-core/src/main/java/accord/coordinate/MaybeRecover.java
@@ -56,6 +56,7 @@ public class MaybeRecover extends CheckShards<Route<?>>
this.prevProgress = prevProgress;
this.callback = callback;
this.reportTo = reportTo;
+ Invariants.expect(!prevProgress.durability.isDurableOrInvalidated());
}
public static Object maybeRecover(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Route<?> someRoute, ProgressToken prevProgress, LatentStoreSelector reportTo, BiConsumer<Outcome, Throwable> callback)
@@ -143,6 +144,7 @@ public class MaybeRecover extends CheckShards<Route<?>>
}
else
{
+ Invariants.expect(!full.durability.isDurableOrInvalidated());
if (tracing != null)
tracing.trace(null, "MaybeRecover invoking RecoverWithRoute");
node.recover(txnId, full.invalidIf, Route.castToFullRoute(someRoute), reportTo, tracing).begin(callback);
diff --git a/accord-core/src/main/java/accord/coordinate/tracking/AbstractTracker.java b/accord-core/src/main/java/accord/coordinate/tracking/AbstractTracker.java
index 1d3727dc..75040782 100644
--- a/accord-core/src/main/java/accord/coordinate/tracking/AbstractTracker.java
+++ b/accord-core/src/main/java/accord/coordinate/tracking/AbstractTracker.java
@@ -53,11 +53,6 @@ public abstract class AbstractTracker<ST extends ShardTracker>
this.result = result;
}
- private boolean isTerminalState()
- {
- return compareTo(Success) <= 0;
- }
-
private static ShardOutcomes min(ShardOutcomes a, ShardOutcomes b)
{
return a.compareTo(b) <= 0 ? a : b;
@@ -147,7 +142,7 @@ public abstract class AbstractTracker<ST extends ShardTracker>
Invariants.require(self == this); // we just accept self as parameter for type safety
ShardOutcomes status = NoChange;
int maxShards = maxShardsPerEpoch();
- for (int i = 0; i < topologyLimit && !status.isTerminalState() ; ++i)
+ for (int i = 0; i < topologyLimit ; ++i)
{
status = topologies.get(i).mapReduceOn(node, i * maxShards, AbstractTracker::apply, self, function, param, ShardOutcomes::min, status);
}
diff --git a/accord-core/src/main/java/accord/coordinate/tracking/DurabilityTracker.java b/accord-core/src/main/java/accord/coordinate/tracking/DurabilityTracker.java
index b21d334a..79893abf 100644
--- a/accord-core/src/main/java/accord/coordinate/tracking/DurabilityTracker.java
+++ b/accord-core/src/main/java/accord/coordinate/tracking/DurabilityTracker.java
@@ -113,6 +113,11 @@ public class DurabilityTracker extends SimpleTracker<DurabilityTracker.Durabilit
return successes.size() >= shard.slowQuorumSize;
}
+ public boolean hasMinorityQuorumSuccess()
+ {
+ return successes.size() >= shard.minorityQuorumSize();
+ }
+
boolean hasInFlight()
{
return waitingOn > 0;
@@ -173,6 +178,8 @@ public class DurabilityTracker extends SimpleTracker<DurabilityTracker.Durabilit
return SyncRemote.All;
if (hasQuorumSuccess())
return SyncRemote.Quorum;
+ if (hasMinorityQuorumSuccess())
+ return SyncRemote.MinorityQuorum;
return SyncRemote.NoRemote;
}
@@ -191,4 +198,9 @@ public class DurabilityTracker extends SimpleTracker<DurabilityTracker.Durabilit
Invariants.require((waitingOnQuorum == 0) == all(DurabilityShardTracker::hasQuorumSuccess));
return waitingOnQuorum == 0;
}
+
+ public boolean hasMinorityQuorumSuccess()
+ {
+ return all(DurabilityShardTracker::hasMinorityQuorumSuccess);
+ }
}
diff --git a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
index 9e72f6dd..30f1ab12 100644
--- a/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
+++ b/accord-core/src/main/java/accord/impl/InMemoryCommandStore.java
@@ -58,6 +58,7 @@ import accord.local.CommandStore;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.CommandSummaries;
import accord.local.Commands;
+import accord.local.MaxDecidedRX;
import accord.local.NodeCommandStoreService;
import accord.local.PreLoadContext;
import accord.local.PreLoadContext.Empty;
@@ -73,7 +74,6 @@ import accord.primitives.AbstractRanges;
import accord.primitives.AbstractUnseekableKeys;
import accord.primitives.PartialDeps;
import accord.primitives.Participants;
-import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
import accord.primitives.Routable.Domain;
import accord.primitives.RoutableKey;
@@ -83,7 +83,6 @@ import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.Txn.Kind.Kinds;
import accord.primitives.TxnId;
-import accord.primitives.Unseekable;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
@@ -98,7 +97,6 @@ import static accord.local.LoadKeys.NONE;
import static accord.local.LoadKeysFor.WRITE;
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;
import static accord.primitives.Routables.Slice.Minimal;
@@ -109,12 +107,12 @@ import static accord.primitives.SaveStatus.ReadyToExecute;
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;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-import static accord.primitives.Txn.Kind.Read;
import static accord.utils.Invariants.illegalState;
import static java.lang.String.format;
@@ -146,13 +144,25 @@ public abstract class InMemoryCommandStore extends CommandStore
}
}
- void save(CommandsForKey cfk)
+ void saveCallback(CommandsForKey cfk)
{
- commandsForKey.put(cfk.key(), Serialize.toBytesWithoutKey(cfk.maximalPrune()));
+ save(cfk);
if (--waitingForCfk == 0)
trySuccess(this);
}
+ private void save(CommandsForKey cfk)
+ {
+ cfk = cfk.maximalPrune();
+ ByteBuffer serialized = Serialize.toBytesWithoutKey(cfk);
+ CommandsForKey roundTrip = Serialize.fromBytes(cfk.key(), serialized);
+ if (roundTrip != null)
+ {
+ Invariants.require(cfk.equalContents(roundTrip));
+ commandsForKey.put(cfk.key(), serialized);
+ }
+ }
+
public static AsyncResult<Snapshot> snapshot(InMemoryCommandStore commandStore)
{
Snapshot snapshot = new Snapshot();
@@ -170,7 +180,7 @@ public abstract class InMemoryCommandStore extends CommandStore
}
else
{
- snapshot.commandsForKey.put(cfk.key(), Serialize.toBytesWithoutKey(cfk.maximalPrune()));
+ snapshot.save(cfk);
}
}
if (snapshot.waitingForCfk == 0)
@@ -420,30 +430,32 @@ public abstract class InMemoryCommandStore extends CommandStore
Map<RoutableKey, InMemorySafeCommandsForKey> commandsForKey = new HashMap<>();
context.forEachId(txnId -> commands.put(txnId, lazyReference(txnId)));
-
- for (Unseekable unseekable : context.keys())
+ if (context.loadKeys() != NONE)
{
- switch (unseekable.domain())
+ Unseekables unseekables = context.keys();
+ if (unseekables.domain() == Key)
{
- case Key:
- RoutableKey key = (RoutableKey) unseekable;
- switch (context.loadKeys())
- {
- case NONE:
- continue;
- case INCR:
- case SYNC:
- case ASYNC:
- commandsForKey.put(key, commandsForKey((RoutingKey) key).createSafeReference());
- break;
- default: throw new UnsupportedOperationException("Unknown key history: " + context.loadKeys());
- }
- break;
- case Range:
- // load range cfks here
- break;
+ for (RoutingKey key : (AbstractUnseekableKeys)unseekables)
+ commandsForKey.put(key, commandsForKey(key).createSafeReference());
+ }
+ else
+ {
+ CommandSummaries.SummaryLoader loader = CommandSummaries.SummaryLoader.loader(unsafeGetRedundantBefore(), unsafeGetMaxDecidedRX(), context);
+ for (GlobalCommandsForKey global : this.commandsForKey.values())
+ {
+ if (!unseekables.contains(global.key))
+ continue;
+
+ if (global.value() == null || !loader.isRelevant(global.value()))
+ continue;
+
+ InMemorySafeCommandsForKey safeCfk = commandsForKey.get(global.key);
+ if (safeCfk == null)
+ commandsForKey.put(global.key, global.createSafeReference());
+ }
}
}
+
return createSafeStore(context, ranges, commands, commandsForKey);
}
@@ -593,7 +605,7 @@ public abstract class InMemoryCommandStore extends CommandStore
public GlobalState<CommandsForKey> value(CommandsForKey value)
{
if (!pendingSnapshots.isEmpty() && !value.isLoadingPruned())
- pendingSnapshots.forEach(snapshot -> snapshot.save(value));
+ pendingSnapshots.forEach(snapshot -> snapshot.saveCallback(value));
return super.value(value);
}
@@ -814,60 +826,63 @@ public abstract class InMemoryCommandStore extends CommandStore
return commandsForRanges;
Invariants.require(context.loadKeysFor() != WRITE);
- Summary.Loader loader = Summary.Loader.loader(redundantBefore(), context.primaryTxnId(), context.loadKeysFor(), context.keys());
+ MaxDecidedRX maxDecidedRX = commandStore().unsafeGetMaxDecidedRX();
+ SummaryLoader loader = SummaryLoader.loader(redundantBefore(), maxDecidedRX, context);
TreeMap<Timestamp, Summary> summaries = new TreeMap<>();
for (RangeCommand rangeCommand : commandStore().rangeCommands.values())
{
GlobalCommand global = commandStore().commands.get(rangeCommand.txnId);
Command command = global == null ? null : global.value();
+ if (!loader.isMaybeRelevant(rangeCommand.txnId))
+ continue;
+
Summary summary;
- if (command == null)
- {
- summary = loader.ifRelevant(rangeCommand.txnId, rangeCommand.txnId, NotDefined, rangeCommand.ranges, null);
- }
- else
+ if (command == null) summary = loader.ifRelevant(rangeCommand.txnId, rangeCommand.txnId, NotDefined, NotDurable, rangeCommand.ranges, null);
+ else summary = loader.ifRelevant(command);
+ if (summary != null)
{
- summary = loader.ifRelevant(command);
+ summaries.put(summary.plainTxnId(), summary);
+ loader.maybeRecordFutureRx(summary);
}
- if (summary != null)
- summaries.put(summary.txnId, summary);
}
- final Kinds kinds = new Kinds(Read, ExclusiveSyncPoint);
- return commandsForRanges = new ByTxnIdSnapshot()
- {
- @Override public NavigableMap<Timestamp, Summary> byTxnId() { return summaries; }
- };
+ return commandsForRanges = () -> summaries;
}
private boolean visitForKey(Unseekables<?> keysOrRanges, Predicate<CommandsForKey> forEach)
{
- for (GlobalCommandsForKey global : commandStore().commandsForKey.values())
+ for (SafeCommandsForKey safeCfk : commandsForKey.values())
{
- if (!keysOrRanges.contains(global.key))
+ if (!keysOrRanges.contains(safeCfk.key()))
continue;
- InMemorySafeCommandsForKey safeCfk = commandsForKey.get(global.key);
- if (safeCfk == null)
- commandsForKey.put(global.key, safeCfk = global.createSafeReference());
-
if (!forEach.test(safeCfk.current()))
return false;
}
return true;
}
+ private <P1, P2> void visitForKey(Unseekables<?> keysOrRanges, Timestamp startedBefore, Kinds testKind, ActiveCommandVisitor<P1, P2> visitor, P1 p1, P2 p2)
+ {
+ visitForKey(keysOrRanges, cfk -> { cfk.visit(startedBefore, testKind, visitor, p1, p2); return true; });
+ }
+
+ public boolean visitForKey(Unseekables<?> keysOrRanges, TxnId testTxnId, Kinds testKind, TestStartedAt testStartedAt, Timestamp testStartedAtTimestamp, ComputeIsDep computeIsDep, AllCommandVisitor visit)
+ {
+ return visitForKey(keysOrRanges, cfk -> cfk.visit(testTxnId, testKind, testStartedAt, testStartedAtTimestamp, computeIsDep, null, visit));
+ }
+
@Override
public <P1, P2> void visit(Unseekables<?> keysOrRanges, Timestamp startedBefore, Kinds testKind, ActiveCommandVisitor<P1, P2> visitor, P1 p1, P2 p2)
{
- visitForKey(keysOrRanges, cfk -> { cfk.visit(startedBefore, testKind, visitor, p1, p2); return true; });
+ visitForKey(keysOrRanges, startedBefore, testKind, visitor, p1, p2);
commandsForRanges().visit(keysOrRanges, startedBefore, testKind, visitor, p1, p2);
}
@Override
public boolean visit(Unseekables<?> keysOrRanges, TxnId testTxnId, Kinds testKind, TestStartedAt testStartedAt, Timestamp testStartedAtTimestamp, ComputeIsDep computeIsDep, AllCommandVisitor visit)
{
- return visitForKey(keysOrRanges, cfk -> cfk.visit(testTxnId, testKind, testStartedAt, testStartedAtTimestamp, computeIsDep, null, visit))
+ return visitForKey(keysOrRanges, testTxnId, testKind, testStartedAt, testStartedAtTimestamp, computeIsDep, visit)
&& commandsForRanges().visit(keysOrRanges, testTxnId, testKind, testStartedAt, testStartedAtTimestamp, computeIsDep, visit);
}
@@ -1277,29 +1292,4 @@ public abstract class InMemoryCommandStore extends CommandStore
return apply(command);
}
}
-
- @Override
- protected void registerTransitive(SafeCommandStore safeStore, RangeDeps rangeDeps)
- {
- RangesForEpoch rangesForEpoch = this.rangesForEpoch;
- Ranges allRanges = rangesForEpoch.all();
-
- TreeMap<TxnId, RangeCommand> rangeCommands = this.rangeCommands;
- rangeDeps.forEachUniqueTxnId(allRanges, null, (ignore, txnId) -> {
- GlobalCommand global = commands.get(txnId);
- if (global != null && global.value().known().has(MaybeRoute))
- return;
-
- Ranges ranges = rangeDeps.ranges(txnId);
- ranges = ranges.without(rangesForEpoch.coordinates(txnId)); // already coordinates, no need to replicate
- if (ranges.isEmpty())
- return;
-
- ranges = ranges.slice(rangesForEpoch.allSince(txnId.epoch()), Minimal); // never coordinated, no need to replicate for dependency or recovery calculations
- if (ranges.isEmpty())
- return;
-
- rangeCommands.computeIfAbsent(txnId, RangeCommand::new).add(ranges);
- });
- }
}
diff --git a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
index 518a283f..38af0f2b 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
@@ -871,6 +871,9 @@ public class DefaultProgressLog implements ProgressLog, Consumer<SafeCommandStor
public void maybeNotify()
{
+ if (stopped)
+ return;
+
if (commandStore.inStore())
{
accept(null);
diff --git a/accord-core/src/main/java/accord/local/Bootstrap.java b/accord-core/src/main/java/accord/local/Bootstrap.java
index bf335167..4adcf210 100644
--- a/accord-core/src/main/java/accord/local/Bootstrap.java
+++ b/accord-core/src/main/java/accord/local/Bootstrap.java
@@ -40,6 +40,9 @@ import accord.utils.ReducingRangeMap;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
+import static accord.local.durability.DurabilityService.SyncLocal.NoLocal;
+import static accord.local.durability.DurabilityService.SyncRemote.MinorityQuorum;
+import static accord.local.durability.DurabilityService.SyncRemote.Quorum;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.utils.Invariants.illegalState;
@@ -135,20 +138,31 @@ class Bootstrap
// of these ranges as part of this attempt
Ranges commitRanges = valid;
safeStore = safeStore;
- // we submit a separate execution so that we know markBootstrapping is durable before we initiate the fetch
- safeStore.commandStore()
- .build((PreLoadContext.Empty) () -> "Start Bootstrap RX", safeStore0 -> {
- store.markBootstrapping(safeStore0, globalSyncId, commitRanges);
- return CoordinateSyncPoint.exclusive(node, globalSyncId, commitRanges);
- })
- .flatMap(i -> i)
- .flatMap(syncPoint -> node.withEpochAtLeast(epoch, null, () -> store.build((PreLoadContext.Empty) () -> "Start Bootstrap Fetch", safeStore1 -> {
- if (valid.isEmpty()) // we've lost ownership of the range
- return AsyncResults.success(Ranges.EMPTY);
- return fetch = safeStore1.dataStore().fetch(node, safeStore1, valid, syncPoint, this);
- })))
- .flatMap(i -> i)
- .begin(this);
+ CommandStore commandStore = safeStore.commandStore();
+ CoordinateSyncPoint.exclusive(node, globalSyncId, commitRanges)
+ .flatMap(syncPoint -> {
+ // before using the sync point we first make sure it is durable to a majority, since any later durability conditions
+ // this node participates in will not guarantee a quorum for preceding transactions
+ // we must do this before we mark ourselves bootstrapping, else we may participate in a durability quorum
+ // without actually waiting for the durability condition to be reached locally
+ // TODO (required): introduce a more robust mechanism when evaluating Durability quorums, esp. since this does not handle markStale
+ // TODO (required): configurable timeout?
+ // TODO (required): can we do better than minority quorum? still not sufficient for bootstrap replace at rf=2
+ return node.durability().sync("Bootstrap " + commitRanges + " for " + commandStore, syncPoint, NoLocal, MinorityQuorum, 1L, TimeUnit.HOURS)
+ .map(i -> syncPoint);
+ })
+ .flatMap(success -> commandStore.build((PreLoadContext.Empty) () -> "Mark Bootstrapping", safeStore0 -> {
+ // we submit a separate execution so that we know markBootstrapping is durable before we initiate the fetch
+ store.markBootstrapping(safeStore0, globalSyncId, commitRanges);
+ return success;
+ }))
+ .flatMap(syncPoint -> node.withEpochAtLeast(epoch, null, () -> store.build((PreLoadContext.Empty) () -> "Start Bootstrap Fetch", safeStore1 -> {
+ if (valid.isEmpty()) // we've lost ownership of the range
+ return AsyncResults.success(Ranges.EMPTY);
+ return fetch = safeStore1.dataStore().fetch(node, safeStore1, valid, syncPoint, this);
+ })))
+ .flatMap(i -> i)
+ .begin(this);
}
// we no longer want to fetch these ranges (perhaps we no longer own them)
@@ -273,7 +287,7 @@ class Bootstrap
store.agent().onFailedBootstrap(attempt, "PartialFetch", newFailures, () -> {
node.scheduler().selfRecurring(() -> {
- store.execute((PreLoadContext.Empty) () -> "Restart Bootstrap", safeStore -> restart(safeStore, newFailures.slice(allValid), attempt + 1), store.agent());
+ store.execute((PreLoadContext.Empty) () -> "Restart Bootstrap", safeStore -> restart(safeStore, newFailures.slice(allValid, Minimal), attempt + 1), store.agent());
}, 0L, TimeUnit.NANOSECONDS);
}, failure);
Invariants.require(!newFailures.intersects(fetchedAndSafeToRead));
diff --git a/accord-core/src/main/java/accord/local/CommandStore.java b/accord-core/src/main/java/accord/local/CommandStore.java
index 30d6018a..14bf6677 100644
--- a/accord-core/src/main/java/accord/local/CommandStore.java
+++ b/accord-core/src/main/java/accord/local/CommandStore.java
@@ -46,7 +46,6 @@ import accord.api.ProgressLog;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.RedundantBefore.Bounds;
import accord.local.RedundantStatus.SomeStatus;
-import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
import accord.primitives.Routables;
import accord.primitives.Status.Durability;
@@ -170,12 +169,12 @@ public abstract class CommandStore implements SequentialAsyncExecutor
{
final AsyncResults.SettableResult<Void> whenDone;
final Ranges allRanges;
- Ranges ranges;
+ Ranges waitingOn, waitingOnDurable;
WaitingOnSync(AsyncResults.SettableResult<Void> whenDone, Ranges ranges)
{
this.whenDone = whenDone;
- this.allRanges = this.ranges = ranges;
+ this.allRanges = this.waitingOn = this.waitingOnDurable = ranges;
}
}
private final TreeMap<Long, WaitingOnSync> waitingOnSync = new TreeMap<>();
@@ -313,8 +312,6 @@ public abstract class CommandStore implements SequentialAsyncExecutor
public abstract void shutdown();
- protected abstract void registerTransitive(SafeCommandStore safeStore, RangeDeps deps);
-
protected void unsafeSetMaxDecidedRX(MaxDecidedRX newMaxDecidedRX)
{
this.maxDecidedRX = newMaxDecidedRX;
@@ -666,7 +663,62 @@ public abstract class CommandStore implements SequentialAsyncExecutor
listeners.clearBefore(this, clearWaitingBefore);
}
- protected void markSynced(SafeCommandStore safeStore, TxnId syncId, Ranges ranges)
+ protected final Ranges isWaitingOnSync(TxnId syncId, Ranges ranges)
+ {
+ if (waitingOnSync.isEmpty())
+ return Ranges.EMPTY;
+
+ Ranges waitingOn = Ranges.EMPTY;
+ for (Map.Entry<Long, WaitingOnSync> e : waitingOnSync.entrySet())
+ {
+ if (e.getKey() > syncId.epoch())
+ break;
+
+ Ranges remaining = e.getValue().waitingOn;
+ Ranges intersecting = remaining.slice(ranges, Minimal);
+ if (!intersecting.isEmpty())
+ {
+ ranges = ranges.without(intersecting);
+ waitingOn = waitingOn.with(intersecting);
+ }
+ }
+
+ return waitingOn;
+ }
+
+ protected final void markSyncing(TxnId syncId, Ranges ranges)
+ {
+ if (waitingOnSync.isEmpty())
+ return;
+
+ for (Map.Entry<Long, WaitingOnSync> e : waitingOnSync.entrySet())
+ {
+ if (e.getKey() > syncId.epoch())
+ break;
+
+ Ranges remaining = e.getValue().waitingOn.without(ranges);
+ if (e.getValue().waitingOn != remaining)
+ e.getValue().waitingOn = remaining;
+ }
+ }
+
+ protected final void unmarkSyncing(TxnId syncId, Ranges ranges)
+ {
+ if (waitingOnSync.isEmpty())
+ return;
+
+ for (Map.Entry<Long, WaitingOnSync> e : waitingOnSync.entrySet())
+ {
+ if (e.getKey() > syncId.epoch())
+ break;
+
+ Ranges unmark = e.getValue().waitingOnDurable.slice(ranges, Minimal);
+ if (!unmark.isEmpty())
+ e.getValue().waitingOn = e.getValue().waitingOn.with(unmark);
+ }
+ }
+
+ protected final void markSynced(SafeCommandStore safeStore, TxnId syncId, Ranges ranges)
{
RedundantBefore addRedundantBefore = RedundantBefore.create(ranges, syncId, LOCALLY_WITNESSED_ONLY);
safeStore.upsertRedundantBefore(addRedundantBefore);
@@ -680,13 +732,15 @@ public abstract class CommandStore implements SequentialAsyncExecutor
if (e.getKey() > syncId.epoch())
break;
- Ranges remaining = e.getValue().ranges;
- Ranges synced = remaining.slice(ranges, Minimal);
- boolean intersects = remaining.intersects(ranges);
+ Ranges waitingOn = e.getValue().waitingOn;
+ Ranges waitingOnDurable = e.getValue().waitingOnDurable;
+ Ranges synced = waitingOnDurable.slice(ranges, Minimal);
+ boolean intersects = waitingOnDurable.intersects(ranges);
if (intersects)
{
- e.getValue().ranges = remaining = remaining.without(ranges);
- if (e.getValue().ranges.isEmpty())
+ e.getValue().waitingOn = waitingOn = waitingOn.without(ranges);
+ e.getValue().waitingOnDurable = waitingOnDurable = waitingOnDurable.without(ranges);
+ if (waitingOnDurable.isEmpty())
{
logger.debug("Completed full sync for {} on epoch {} using {}", e.getValue().allRanges, e.getKey(), syncId);
e.getValue().whenDone.trySuccess(null);
@@ -696,7 +750,7 @@ public abstract class CommandStore implements SequentialAsyncExecutor
}
else
{
- logger.debug("Completed partial sync for {} on epoch {} using {}; {} still to sync", synced, e.getKey(), syncId, remaining);
+ logger.debug("Completed partial sync for {} on epoch {} using {}; {} still to sync and {} to sync durably", synced, e.getKey(), syncId, waitingOn, waitingOnDurable);
}
}
}
@@ -816,11 +870,16 @@ public abstract class CommandStore implements SequentialAsyncExecutor
public static ImmutableSortedMap<TxnId, Ranges> bootstrap(TxnId at, Ranges ranges, NavigableMap<TxnId, Ranges> bootstrappedAt)
{
- Invariants.requireArgument(bootstrappedAt.lastKey().compareTo(at) < 0 || at == TxnId.NONE);
+ Invariants.requireArgument(!ranges.isEmpty());
if (at == TxnId.NONE)
+ {
for (Ranges rs : bootstrappedAt.values())
Invariants.require(!ranges.intersects(rs));
- Invariants.requireArgument(!ranges.isEmpty());
+ }
+ else
+ {
+ bootstrappedAt.floorEntry(at).getValue().containsAll(ranges);
+ }
// if we're bootstrapping these ranges, then any period we previously owned the ranges for is effectively invalidated
return purgeAndInsert(bootstrappedAt, at, ranges);
}
diff --git a/accord-core/src/main/java/accord/local/CommandSummaries.java b/accord-core/src/main/java/accord/local/CommandSummaries.java
index 2295ae49..0fefbbdb 100644
--- a/accord-core/src/main/java/accord/local/CommandSummaries.java
+++ b/accord-core/src/main/java/accord/local/CommandSummaries.java
@@ -25,24 +25,28 @@ import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import accord.local.RedundantBefore.Bounds;
+import accord.local.cfk.CommandsForKey;
import accord.primitives.PartialDeps;
import accord.primitives.Participants;
import accord.primitives.Ranges;
import accord.primitives.SaveStatus;
-import accord.primitives.Status;
+import accord.primitives.Status.Durability;
import accord.primitives.Timestamp;
+import accord.primitives.Txn;
import accord.primitives.Txn.Kind.Kinds;
import accord.primitives.TxnId;
import accord.primitives.Unseekable;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
+import accord.utils.ReducingRangeMap;
import accord.utils.UnhandledEnum;
import static accord.local.CommandSummaries.SummaryStatus.ACCEPTED;
+import static accord.local.CommandSummaries.SummaryStatus.APPLIED;
import static accord.local.LoadKeysFor.RECOVERY;
+import static accord.local.MaxDecidedRX.minDecidedDependencyId;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Durability.NotDurable;
-import static accord.primitives.Txn.Kind.AnyGloballyVisible;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.primitives.Txn.Kind.Nothing;
@@ -60,185 +64,343 @@ public interface CommandSummaries
INVALIDATED;
public static final SummaryStatus NONE = null;
+
+ private static final SummaryStatus[] SUMMARY_STATUSES = values();
}
- enum IsDep { IS_COORD_DEP, IS_NOT_COORD_DEP, NOT_ELIGIBLE, IS_STABLE_DEP, IS_NOT_STABLE_DEP }
+ enum IsDep
+ {
+ IS_COORD_DEP, IS_NOT_COORD_DEP, NOT_ELIGIBLE, IS_STABLE_DEP, IS_NOT_STABLE_DEP;
+ private static final IsDep[] IS_DEPS = values();
+ }
- class Summary
+ class Summary extends TxnId
{
- public final @Nonnull TxnId txnId;
- public final @Nonnull Timestamp executeAt;
- public final @Nonnull SummaryStatus status;
- public final @Nonnull Unseekables<?> participants;
+ private static final int SUMMARY_STATUS_MASK = 0x7;
+ private static final int IS_DEP_SHIFT = 3;
+ private static final int IS_DEP_MASK = 0x7;
+ private static final int DURABILITY_SHIFT = 6;
+ private static final int DURABILITY_MASK = 0x7;
+
+ final @Nonnull Timestamp executeAt;
+ final int encoded;
+ final Unseekables<?> participants;
- public final IsDep dep;
- public final TxnId findAsDep;
+ public Summary slice(Ranges ranges)
+ {
+ return new Summary(this, this.executeAt, encoded, participants.slice(ranges, Minimal));
+ }
@VisibleForTesting
- public Summary(@Nonnull TxnId txnId, @Nonnull Timestamp executeAt, @Nonnull SummaryStatus status, @Nonnull Unseekables<?> participants, IsDep dep, TxnId findAsDep)
+ public Summary(@Nonnull TxnId txnId, @Nonnull Timestamp executeAt, @Nonnull SummaryStatus status, Durability durability, IsDep dep, Unseekables<?> participants)
{
- this.txnId = txnId;
- this.executeAt = executeAt;
- this.status = status;
+ super(txnId);
this.participants = participants;
- this.findAsDep = findAsDep;
- this.dep = dep;
+ this.executeAt = executeAt.equals(txnId) ? this : executeAt;
+ this.encoded = status.ordinal() | (dep == null ? Integer.MIN_VALUE : (dep.ordinal() << IS_DEP_SHIFT)) | (durability.ordinal() << DURABILITY_SHIFT);
+ }
+
+ private Summary(@Nonnull TxnId txnId, @Nonnull Timestamp executeAt, int encoded, Unseekables<?> participants)
+ {
+ super(txnId);
+ this.participants = participants;
+ this.executeAt = executeAt == txnId || executeAt.equals(txnId) ? this : executeAt;
+ this.encoded = encoded;
+ }
+
+ public boolean is(IsDep isDep)
+ {
+ return (encoded >> IS_DEP_SHIFT) == isDep.ordinal();
+ }
+
+ public IsDep isDep()
+ {
+ if (encoded < 0)
+ return null;
+ return IsDep.IS_DEPS[(encoded >>> IS_DEP_SHIFT) & IS_DEP_MASK];
+ }
+
+ public Durability durability()
+ {
+ return Durability.forOrdinal((encoded >>> DURABILITY_SHIFT) & DURABILITY_MASK);
+ }
+
+ public boolean is(SummaryStatus summaryStatus)
+ {
+ return (encoded & SUMMARY_STATUS_MASK) == summaryStatus.ordinal();
+ }
+
+ public SummaryStatus status()
+ {
+ int ordinal = encoded & SUMMARY_STATUS_MASK;
+ return SummaryStatus.SUMMARY_STATUSES[ordinal];
+ }
+
+ public TxnId plainTxnId()
+ {
+ return new TxnId(this);
}
- public Summary slice(Ranges slice)
+ public Timestamp plainExecuteAt()
{
- return new Summary(txnId, executeAt, status, participants.slice(slice, Minimal), dep, findAsDep);
+ return executeAt == this ? new Timestamp(this) : executeAt;
}
@Override
public String toString()
{
return "Summary{" +
- "txnId=" + txnId +
- ", executeAt=" + executeAt +
- ", saveStatus=" + status +
- ", participants=" + participants +
- ", maybeDep=" + dep +
- ", findAsDep=" + findAsDep +
+ "txnId=" + plainTxnId() +
+ ", executeAt=" + plainExecuteAt() +
+ ", saveStatus=" + status() +
+ ", isDep=" + isDep() +
'}';
}
+ }
- public static class Loader
+ class SummaryLoader
+ {
+ public interface Factory<L extends SummaryLoader>
{
- public interface Factory<L extends Loader>
- {
- L create(@Nullable TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, Kinds testKind, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep);
- }
+ L create(RedundantBefore redundantBefore, @Nullable MaxDecidedRX maxDecidedRX, TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, Kinds testKind, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep);
+ }
- protected final Unseekables<?> searchKeysOrRanges;
- protected final RedundantBefore redundantBefore;
- // TODO (expected): separate out Kinds we need before/after primaryTxnId/executeAt
- protected final Kinds testKind;
- protected final TxnId minTxnId;
- protected final Timestamp maxTxnId;
- @Nullable protected final TxnId findAsDep;
+ private static ReducingRangeMap<TxnId> NO_FUTURE_RX = new ReducingRangeMap<>();
+ protected final RedundantBefore redundantBefore;
+ protected final MaxDecidedRX maxDecidedRX;
+ protected final Unseekables<?> searchKeysOrRanges;
+ // TODO (expected): separate out Kinds we need before/after primaryTxnId/executeAt
+ protected final Kinds testKind;
+ protected final TxnId primaryTxnId, findAsDep, minTxnId, minDecidedId;
+ protected final Timestamp maxTxnId;
- // TODO (expected): provide executeAt to PreLoadContext so we can more aggressively filter what we load, esp. by Kind
- public static Loader loader(RedundantBefore redundantBefore, PreLoadContext context)
- {
- return loader(redundantBefore, context.primaryTxnId(), context.loadKeysFor(), context.keys());
- }
+ private ReducingRangeMap<TxnId> minVisitedFutureRX = NO_FUTURE_RX;
+ private TxnId maxRx = TxnId.MAX;
- public static Loader loader(RedundantBefore redundantBefore, @Nullable TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges)
- {
- return loader(redundantBefore, primaryTxnId, loadKeysFor, keysOrRanges, Loader::new);
- }
+ // TODO (expected): provide executeAt to PreLoadContext so we can more aggressively filter what we load, esp. by Kind
+ public static SummaryLoader loader(RedundantBefore redundantBefore, MaxDecidedRX maxDecidedRX, PreLoadContext context)
+ {
+ return loader(redundantBefore, maxDecidedRX, context.primaryTxnId(), context.loadKeysFor(), context.keys());
+ }
- public static <L extends Loader> L loader(RedundantBefore redundantBefore, @Nullable TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges, Factory<L> factory)
- {
- TxnId minTxnId = redundantBefore.min(keysOrRanges, Bounds::gcBefore);
- Timestamp maxTxnId = primaryTxnId == null || loadKeysFor == RECOVERY || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
- TxnId findAsDep = primaryTxnId != null && loadKeysFor == RECOVERY ? primaryTxnId : null;
- Kinds kinds = primaryTxnId == null ? AnyGloballyVisible : primaryTxnId.witnesses().or(loadKeysFor == RECOVERY ? primaryTxnId.witnessedBy() : Nothing);
- return factory.create(primaryTxnId, keysOrRanges, redundantBefore, kinds, minTxnId, maxTxnId, findAsDep);
- }
+ public static SummaryLoader loader(RedundantBefore redundantBefore, MaxDecidedRX maxDecidedRX, TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges)
+ {
+ return loader(redundantBefore, maxDecidedRX, primaryTxnId, loadKeysFor, keysOrRanges, SummaryLoader::new);
+ }
- public Loader(@Nullable TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, Kinds testKind, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
- {
- this.searchKeysOrRanges = searchKeysOrRanges;
- this.redundantBefore = redundantBefore;
- this.testKind = testKind;
- this.minTxnId = minTxnId;
- this.maxTxnId = maxTxnId;
- this.findAsDep = findAsDep;
- }
+ public static <L extends SummaryLoader> L loader(RedundantBefore redundantBefore, MaxDecidedRX maxDecidedRX, PreLoadContext context, Factory<L> factory)
+ {
+ return loader(redundantBefore, maxDecidedRX, context.primaryTxnId(), context.loadKeysFor(), context.keys(), factory);
+ }
- public Summary ifRelevant(Command cmd)
- {
- return ifRelevant(cmd.txnId(), cmd.executeAtOrTxnId(), cmd.saveStatus(), cmd.participants(), cmd.partialDeps());
- }
+ public static <L extends SummaryLoader> L loader(RedundantBefore redundantBefore, MaxDecidedRX maxDecidedRX, TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges, Factory<L> factory)
+ {
+ Invariants.require(primaryTxnId != null);
+ TxnId minTxnId = redundantBefore.min(keysOrRanges, Bounds::gcBefore);
+ Timestamp maxTxnId = loadKeysFor == RECOVERY || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
+ TxnId findAsDep = loadKeysFor == RECOVERY ? primaryTxnId : null;
+ Kinds kinds = primaryTxnId.witnesses().or(loadKeysFor == RECOVERY ? primaryTxnId.witnessedBy() : Nothing);
+ if (!primaryTxnId.is(Txn.Kind.ExclusiveSyncPoint))
+ maxDecidedRX = null;
+ return factory.create(redundantBefore, maxDecidedRX, primaryTxnId, keysOrRanges, kinds, minTxnId, maxTxnId, findAsDep);
+ }
+
+ public SummaryLoader(RedundantBefore redundantBefore, MaxDecidedRX maxDecidedRX, TxnId primaryTxnId, Unseekables<?> searchKeysOrRanges, Kinds testKind, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
+ {
+ this.redundantBefore = redundantBefore;
+ this.maxDecidedRX = maxDecidedRX;
+ this.primaryTxnId = primaryTxnId;
+ this.searchKeysOrRanges = searchKeysOrRanges;
+ this.testKind = testKind;
+ this.minTxnId = minTxnId;
+ this.maxTxnId = maxTxnId;
+ this.findAsDep = findAsDep;
+ this.minDecidedId = minDecidedDependencyId(maxDecidedRX, searchKeysOrRanges, primaryTxnId);
+ }
+
+ public boolean isRelevant(CommandsForKey cfk)
+ {
+ //noinspection SizeReplaceableByIsEmpty (not equivalent)
+ if (cfk == null || cfk.size() == 0)
+ return false;
+
+ // NOTE: we CANNOT safely filter on first element, as we may have pruned dependencies we need to witness
+ // and that will be populated on the receiving replicas as necessary - that is,
+ // we must permit adopting future dependencies
+ CommandsForKey.TxnInfo last = cfk.get(cfk.size() - 1);
+ if (last.compareTo(minTxnId) < 0)
+ return false;
+
+ if (maxDecidedRX == null)
+ return true;
+
+ CommandsForKey.TxnInfo minUndecided = cfk.minUndecided();
+ if (minUndecided != null)
+ return true;
+
+ if (minDecidedId != null && last.compareTo(minDecidedId) < 0)
+ return false;
+
+ TxnId minDecidedId = maxDecidedRX.minDecidedDependencyId(cfk.key(), primaryTxnId);
+ return minDecidedId == null || last.compareTo(minDecidedId) >= 0;
+ }
- private boolean isEligibleDep(SummaryStatus status, TxnId findAsDep, TxnId txnId, Timestamp executeAt)
+ // the caller must manage mutual exclusion for this method, but not to any others
+ public void maybeRecordFutureRx(Summary summary)
+ {
+ if (summary.is(ExclusiveSyncPoint) && summary.compareTo(primaryTxnId) > 0 && (summary.is(SummaryStatus.STABLE) || summary.is(APPLIED)))
{
- switch (status)
- {
- default: throw new UnhandledEnum(status);
- case NOT_DIRECTLY_WITNESSED:
- case INVALIDATED:
- return false;
- case NOTACCEPTED:
- case PREACCEPTED:
- if (!txnId.is(TxnId.FastPath.PrivilegedCoordinatorWithDeps))
- return false;
- case ACCEPTED:
- return txnId.compareTo(findAsDep) > 0;
- case COMMITTED:
- case APPLIED:
- case STABLE:
- return executeAt.compareTo(findAsDep) > 0;
- }
+ minVisitedFutureRX = ReducingRangeMap.merge(minVisitedFutureRX, ReducingRangeMap.create(summary.participants.toRanges(), summary.plainTxnId()), TxnId::min);
+ maxRx = minVisitedFutureRX.foldlWithDefault(searchKeysOrRanges, TxnId::max, TxnId.MAX, TxnId.NONE);
}
+ }
+
+ public final Summary ifRelevant(Command cmd)
+ {
+ return ifRelevant(cmd.txnId(), cmd.executeAtOrTxnId(), cmd.saveStatus(), cmd.durability(), cmd.participants(), cmd.partialDeps());
+ }
+
+ public final boolean isMaybeRelevant(TxnId txnId)
+ {
+ return isMaybeRelevant(txnId, null, null);
+ }
+
+ // durability is used as a proxy for durably *decided*
+ public final boolean isMaybeRelevant(TxnId txnId, @Nullable Durability durability, @Nullable Unseekables<?> participants)
+ {
+ if (!txnId.is(testKind))
+ return false;
+
+ if (txnId.compareTo(minTxnId) < 0 || txnId.compareTo(maxTxnId) > 0)
+ return false;
- public Summary ifRelevant(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, StoreParticipants participants, @Nullable PartialDeps partialDeps)
+ if (txnId.is(ExclusiveSyncPoint))
{
- if (participants == null)
- return null;
+ if (txnId.compareTo(maxRx) >= 0)
+ return false;
- return ifRelevant(txnId, executeAt, saveStatus, participants.touches(), partialDeps);
+ if (participants != null && txnId.compareTo(minVisitedFutureRX.foldlWithDefault(participants, TxnId::max, TxnId.MAX, TxnId.NONE)) >= 0)
+ return false;
}
- public Summary ifRelevant(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, Participants<?> touches, @Nullable PartialDeps partialDeps)
+ boolean mayFilterAsDecided = maxDecidedRX != null && (txnId.is(ExclusiveSyncPoint) || (durability != null && durability.isDurableOrInvalidated()));
+ if (!mayFilterAsDecided)
+ return true;
+
+ if (minDecidedId != null && txnId.compareTo(minDecidedId) < 0)
+ return false;
+
+ if (participants == null)
+ return true;
+
+ TxnId moreSpecificMinDecidedId = minDecidedDependencyId(maxDecidedRX, participants, primaryTxnId);
+ return moreSpecificMinDecidedId == null || txnId.compareTo(moreSpecificMinDecidedId) >= 0;
+ }
+
+ public final Summary ifRelevant(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, Durability durability, StoreParticipants participants, @Nullable PartialDeps partialDeps)
+ {
+ if (participants == null)
+ return null;
+
+ return ifRelevant(txnId, executeAt, saveStatus, durability, participants.touches(), partialDeps);
+ }
+
+ public final Summary ifRelevant(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, Durability durability, Participants<?> touches, @Nullable PartialDeps partialDeps)
+ {
+ SummaryStatus summaryStatus = saveStatus.summary;
+ if (summaryStatus == null)
+ return null;
+
+ if (!txnId.is(testKind))
+ return null;
+
+ if (txnId.compareTo(minTxnId) < 0 || txnId.compareTo(maxTxnId) > 0)
+ return null;
+
+ boolean mayFilterAsDecided = maxDecidedRX != null && (durability.isDurableOrInvalidated() || txnId.is(ExclusiveSyncPoint));
+ if (mayFilterAsDecided && minDecidedId != null && txnId.compareTo(minDecidedId) < 0)
+ return null;
+
+ // start in search key domain, since this is what we consult to decide if can be recovered
+ Unseekables<?> intersecting = searchKeysOrRanges.intersecting(touches, Minimal);
+ if (intersecting.isEmpty())
+ return null;
+
+ if (redundantBefore != null)
{
- SummaryStatus summaryStatus = saveStatus.summary;
- if (summaryStatus == null)
+ // TODO (expected): consider whether this is necessary (and document it).
+ Unseekables<?> newIntersecting = redundantBefore.foldlWithBounds(intersecting, (e, accum, start, end) -> {
+ if (e.gcBefore.compareTo(txnId) <= 0)
+ return accum;
+ return accum.without(Ranges.of(start.rangeFactory().newRange(start, end)));
+ }, intersecting, ignore -> false);
+
+ if (newIntersecting.isEmpty())
return null;
- if (!txnId.is(testKind))
- return null;
+ intersecting = newIntersecting;
+ }
- // start in search key domain, since this is what we consult to decide if can be recovered
- Unseekables<?> intersecting = searchKeysOrRanges.intersecting(touches, Minimal);
- if (intersecting.isEmpty())
+ if (mayFilterAsDecided)
+ {
+ TxnId minDecidedId = minDecidedDependencyId(maxDecidedRX, intersecting, primaryTxnId);
+ if (minDecidedId != null && txnId.compareTo(minDecidedId) < 0)
return null;
+ }
- if (redundantBefore != null)
+ Invariants.require(partialDeps != null || findAsDep == null || !saveStatus.known.deps().hasProposedOrDecidedDeps());
+ IsDep isDep = null;
+ if (findAsDep != null)
+ {
+ if (partialDeps == null || !isEligibleDep(summaryStatus, findAsDep, txnId, executeAt))
{
- // TODO (expected): consider whether this is necessary (and document it).
- Unseekables<?> newIntersecting = redundantBefore.foldlWithBounds(intersecting, (e, accum, start, end) -> {
- if (e.gcBefore.compareTo(txnId) <= 0)
- return accum;
- return accum.without(Ranges.of(start.rangeFactory().newRange(start, end)));
- }, intersecting, ignore -> false);
-
- if (newIntersecting.isEmpty())
- return null;
-
- intersecting = newIntersecting;
+ isDep = IsDep.NOT_ELIGIBLE;
}
-
- Invariants.require(partialDeps != null || findAsDep == null || !saveStatus.known.deps().hasProposedOrDecidedDeps());
- IsDep isDep = null;
- if (findAsDep != null)
+ else
{
- if (partialDeps == null || !isEligibleDep(summaryStatus, findAsDep, txnId, executeAt))
- {
- isDep = IsDep.NOT_ELIGIBLE;
- }
- else
- {
- boolean isCoordDeps = summaryStatus.compareTo(ACCEPTED) < 0;
- int index = partialDeps.indexOf(findAsDep);
- // TODO (desired): don't construct participants, pass intersecting as parameter
- boolean isAnyDep = index >= 0 && partialDeps.isStable(index)
- && partialDeps.participants(index).containsAll(intersecting);
-
- isDep = isAnyDep ? (isCoordDeps ? IsDep.IS_COORD_DEP : IsDep.IS_STABLE_DEP)
- : (isCoordDeps ? IsDep.IS_NOT_COORD_DEP : IsDep.IS_NOT_STABLE_DEP);
- }
+ boolean isCoordDeps = summaryStatus.compareTo(ACCEPTED) < 0;
+ int index = partialDeps.indexOf(findAsDep);
+ // TODO (desired): don't construct participants, pass intersecting as parameter
+ boolean isAnyDep = index >= 0 && partialDeps.isStable(index)
+ && partialDeps.participants(index).containsAll(intersecting);
+
+ isDep = isAnyDep ? (isCoordDeps ? IsDep.IS_COORD_DEP : IsDep.IS_STABLE_DEP)
+ : (isCoordDeps ? IsDep.IS_NOT_COORD_DEP : IsDep.IS_NOT_STABLE_DEP);
}
+ }
+
+ // convert to the domain of the command we're loading
+ intersecting = touches.intersecting(intersecting, Minimal);
+ return construct(txnId, executeAt, summaryStatus, durability, isDep, intersecting);
+ }
- // convert to the domain of the command we're loading
- intersecting = touches.intersecting(intersecting, Minimal);
- return new Summary(txnId, executeAt, summaryStatus, intersecting, isDep, findAsDep);
+ final boolean isEligibleDep(SummaryStatus status, TxnId findAsDep, TxnId txnId, Timestamp executeAt)
+ {
+ switch (status)
+ {
+ default: throw new UnhandledEnum(status);
+ case NOT_DIRECTLY_WITNESSED:
+ case INVALIDATED:
+ return false;
+ case NOTACCEPTED:
+ case PREACCEPTED:
+ if (!txnId.is(TxnId.FastPath.PrivilegedCoordinatorWithDeps))
+ return false;
+ case ACCEPTED:
+ return txnId.compareTo(findAsDep) > 0;
+ case COMMITTED:
+ case APPLIED:
+ case STABLE:
+ return executeAt.compareTo(findAsDep) > 0;
}
}
- }
+ protected Summary construct(TxnId txnId, Timestamp executeAt, SummaryStatus summaryStatus, Durability durability, IsDep isDep, Unseekables<?> participants)
+ {
+ return new Summary(txnId, executeAt, summaryStatus, durability, isDep, participants);
+ }
+ }
+
enum TestStartedAt { STARTED_BEFORE, STARTED_AFTER, ANY }
enum ComputeIsDep
{
@@ -251,7 +413,7 @@ public interface CommandSummaries
interface ActiveCommandVisitor<P1, P2>
{
- void visit(P1 p1, P2 p2, SummaryStatus status, Unseekable keyOrRange, TxnId txnId);
+ void visit(P1 p1, P2 p2, SummaryStatus status, Durability durability, Unseekable keyOrRange, TxnId txnId);
default void visitMaxAppliedHlc(long maxAppliedHlc) {}
}
@@ -260,7 +422,7 @@ public interface CommandSummaries
/**
* Note: Durability is not guaranteed to return anything besides NotDurable; implementation is free to return more information if easily available.
*/
- boolean visit(Unseekable keyOrRange, TxnId txnId, Timestamp executeAt, SummaryStatus status, @Nullable IsDep dep, Status.Durability minDurability);
+ boolean visit(Unseekable keyOrRange, TxnId txnId, Timestamp executeAt, SummaryStatus status, @Nullable IsDep dep, Durability minDurability);
}
boolean visit(Unseekables<?> keysOrRanges, TxnId testTxnId, Kinds testKind, TestStartedAt testStartedAt, Timestamp testStartAtTimestamp, ComputeIsDep computeIsDep, AllCommandVisitor visit);
@@ -271,9 +433,23 @@ public interface CommandSummaries
*/
<P1, P2> void visit(Unseekables<?> keysOrRanges, Timestamp startedBefore, Kinds testKind, ActiveCommandVisitor<P1, P2> visit, P1 p1, P2 p2);
+ // TODO (expected): ByRangeSnapshot based on IntervalBTree so we can elide superseded dependencies as we do with CommandsForKey
interface ByTxnIdSnapshot extends CommandSummaries
{
NavigableMap<Timestamp, Summary> byTxnId();
+ class Helper
+ {
+ static <S extends Summary> NavigableMap<Timestamp, S> slice(TestStartedAt testStartedAt, Timestamp testStartedAtTimestamp, NavigableMap<Timestamp, S> map)
+ {
+ switch (testStartedAt)
+ {
+ default: throw new UnhandledEnum(testStartedAt);
+ case STARTED_AFTER: return map.tailMap(testStartedAtTimestamp, false);
+ case STARTED_BEFORE: return map.headMap(testStartedAtTimestamp, false);
+ case ANY: return map;
+ }
+ }
+ }
default boolean visit(Unseekables<?> keysOrRanges,
TxnId testTxnId,
@@ -283,36 +459,23 @@ public interface CommandSummaries
ComputeIsDep computeIsDep,
AllCommandVisitor visit)
{
- NavigableMap<Timestamp, Summary> map = byTxnId();
- switch (testStartedAt)
- {
- default: throw new AssertionError("Unknown started at: " + testStartedAt);
- case STARTED_AFTER:
- map = map.tailMap(testStartedAtTimestamp, false);
- break;
- case STARTED_BEFORE:
- map = map.headMap(testStartedAtTimestamp, false);
- break;
- case ANY:
- break;
- }
+ NavigableMap<Timestamp, Summary> map = Helper.slice(testStartedAt, testStartedAtTimestamp, byTxnId());
for (Summary value : map.values())
{
- TxnId txnId = value.txnId;
- if (!testKind.test(txnId))
+ if (!testKind.test(value))
continue;
Unseekables<?> participants = value.participants;
Unseekables<?> intersecting = participants.intersecting(keysOrRanges);
if (!intersecting.isEmpty())
{
- Timestamp executeAt = value.executeAt;
- SummaryStatus status = value.status;
- IsDep dep = value.dep;
+ Timestamp executeAt = value.plainExecuteAt();
+ SummaryStatus status = value.status();
+ IsDep dep = value.isDep();
for (Unseekable participant : intersecting)
{
- if (!visit.visit(participant, txnId, executeAt, status, dep, NotDurable))
+ if (!visit.visit(participant, value.plainTxnId(), executeAt, status, dep, NotDurable))
return false;
}
}
@@ -327,16 +490,14 @@ public interface CommandSummaries
NavigableMap<Timestamp, Summary> map = byTxnId();
for (Summary value : map.headMap(startedBefore, false).values())
{
- TxnId txnId = value.txnId;
- if (!testKind.test(txnId))
+ if (!testKind.test(value))
continue;
- SummaryStatus status = value.status;
- if (status == SummaryStatus.INVALIDATED)
+ if (value.is(SummaryStatus.INVALIDATED))
continue;
for (Unseekable keyOrRange : value.participants.intersecting(keysOrRanges, Minimal))
- visit.visit(p1, p2, status, keyOrRange, txnId);
+ visit.visit(p1, p2, value.status(), value.durability(), keyOrRange, value.plainTxnId());
}
}
}
diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java
index 8b621aa9..2de2fc4e 100644
--- a/accord-core/src/main/java/accord/local/Commands.java
+++ b/accord-core/src/main/java/accord/local/Commands.java
@@ -1111,7 +1111,8 @@ public class Commands
TxnId txnId = command.txnId();
if (dependencyElision() == IF_DURABLE && CommandsForKey.manages(txnId))
{
- PreLoadContext context = PreLoadContext.contextFor(updated.participants().touches(), INCR, WRITE, "Set Durable");
+ AbstractUnseekableKeys keys = (AbstractUnseekableKeys)updated.participants().touches();
+ PreLoadContext context = PreLoadContext.contextFor(keys, INCR, WRITE, "Set Durable");
PreLoadContext execute = safeStore.canExecute(context);
if (execute != null)
{
@@ -1120,7 +1121,7 @@ public class Commands
if (execute != context)
{
if (execute != null)
- context = contextFor(context.keys().without(execute.keys()), INCR, WRITE, "Set Durable");
+ context = contextFor(keys.without(execute.keys()), INCR, WRITE, "Set Durable");
Invariants.require(!context.keys().isEmpty());
safeStore = safeStore; // prevent accidental usage inside lambda
diff --git a/accord-core/src/main/java/accord/local/DepsCalculator.java b/accord-core/src/main/java/accord/local/DepsCalculator.java
index c67b4459..e4edaae8 100644
--- a/accord-core/src/main/java/accord/local/DepsCalculator.java
+++ b/accord-core/src/main/java/accord/local/DepsCalculator.java
@@ -18,25 +18,66 @@
package accord.local;
+import javax.annotation.Nullable;
+
import accord.coordinate.ExecuteFlag.ExecuteFlags;
import accord.local.CommandSummaries.SummaryStatus;
import accord.primitives.Deps;
import accord.primitives.EpochSupplier;
import accord.primitives.Participants;
import accord.primitives.RangeDeps;
+import accord.primitives.Status;
+import accord.primitives.Status.Durability;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.primitives.Unseekable;
+import accord.primitives.Unseekables;
import accord.utils.Invariants;
import static accord.coordinate.ExecuteFlag.HAS_UNIQUE_HLC;
import static accord.coordinate.ExecuteFlag.READY_TO_EXECUTE;
import static accord.local.CommandSummaries.SummaryStatus.APPLIED;
+import static accord.local.CommandSummaries.SummaryStatus.COMMITTED;
import static accord.primitives.Txn.Kind.EphemeralRead;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-public class DepsCalculator extends Deps.Builder implements CommandSummaries.ActiveCommandVisitor<TxnId, Object>
+public class DepsCalculator extends Deps.Builder implements CommandSummaries.ActiveCommandVisitor<TxnId, DepsCalculator.MinDependencyCalculator>
{
+ public static class MinDependencyCalculator
+ {
+ final MaxDecidedRX maxDecidedRX;
+ final TxnId absoluteMinDecidedId;
+ final TxnId txnId;
+ Unseekable prevKeyOrRange;
+ TxnId prevMinDecidedId;
+
+ MinDependencyCalculator(MaxDecidedRX maxDecidedRX, Unseekables<?> keysOrRanges, TxnId txnId)
+ {
+ this.maxDecidedRX = maxDecidedRX;
+ this.absoluteMinDecidedId = maxDecidedRX.minDecidedDependencyId(keysOrRanges, txnId);
+ this.txnId = txnId;
+ }
+
+ boolean include(Durability durability, Unseekable keyOrRange, TxnId depId)
+ {
+ if (durability.isDurable() || depId.is(ExclusiveSyncPoint))
+ {
+ if (absoluteMinDecidedId != null && depId.compareTo(absoluteMinDecidedId) < 0)
+ return false;
+
+ if (!keyOrRange.equals(prevKeyOrRange))
+ {
+ prevKeyOrRange = keyOrRange;
+ prevMinDecidedId = maxDecidedRX.minDecidedDependencyId(keyOrRange, txnId);
+ }
+
+ if (prevMinDecidedId != null && depId.compareTo(prevMinDecidedId) < 0)
+ return false;
+ }
+ return true;
+ }
+ }
+
private boolean hasUnappliedDependency;
private long maxAppliedHlc;
@@ -46,10 +87,13 @@ public class DepsCalculator extends Deps.Builder implements CommandSummaries.Act
}
@Override
- public void visit(TxnId self, Object o, SummaryStatus status, Unseekable keyOrRange, TxnId txnId)
+ public void visit(TxnId self, @Nullable MinDependencyCalculator minDepCalc, SummaryStatus status, Durability durability, Unseekable keyOrRange, TxnId depId)
{
- if (self == null || !self.equals(txnId))
- add(keyOrRange, txnId);
+ if (minDepCalc != null && !minDepCalc.include(durability, keyOrRange, depId))
+ return;
+
+ if (self == null || !self.equals(depId))
+ add(keyOrRange, depId);
if (status.compareTo(APPLIED) < 0)
hasUnappliedDependency = true;
}
@@ -99,7 +143,9 @@ public class DepsCalculator extends Deps.Builder implements CommandSummaries.Act
}
// NOTE: ExclusiveSyncPoint *relies* on STARTED_BEFORE to ensure it reports a dependency on *every* earlier TxnId that may execute (before or after it).
- safeStore.visit(touches, executeAt, txnId.witnesses(), this, executeAt.equals(txnId) ? null : txnId, null);
+ MinDependencyCalculator minDepCalc = null;
+ if (txnId.is(ExclusiveSyncPoint)) minDepCalc = new MinDependencyCalculator(safeStore.maxDecidedRX(), touches, txnId);
+ safeStore.visit(touches, executeAt, txnId.witnesses(), this, executeAt.equals(txnId) ? null : txnId, minDepCalc);
Deps result = super.build();
result = new Deps(result.keyDeps, result.rangeDeps.with(redundant));
Invariants.require(!txnId.isVisible() || !result.contains(txnId));
diff --git a/accord-core/src/main/java/accord/local/LoadKeysFor.java b/accord-core/src/main/java/accord/local/LoadKeysFor.java
index 1e4ba7f0..dcf14f12 100644
--- a/accord-core/src/main/java/accord/local/LoadKeysFor.java
+++ b/accord-core/src/main/java/accord/local/LoadKeysFor.java
@@ -34,9 +34,8 @@ public enum LoadKeysFor
WRITE,
/**
- * READ covers all intersecting summaries of relevant key or range transactions that might be
- * witnessed by the primaryTxnId, for any commands that should be witnessed by primaryTxnId.
- *
+ * READ covers all intersecting summaries of relevant key or range transactions,
+ * including any commands that should be witnessed by primaryTxnId.
* This means range transactions MUST be able to consult all intersecting key summaries.
*/
READ_WRITE,
diff --git a/accord-core/src/main/java/accord/local/MaxDecidedRX.java b/accord-core/src/main/java/accord/local/MaxDecidedRX.java
index 680db265..5b73c236 100644
--- a/accord-core/src/main/java/accord/local/MaxDecidedRX.java
+++ b/accord-core/src/main/java/accord/local/MaxDecidedRX.java
@@ -22,7 +22,6 @@ import javax.annotation.Nullable;
import accord.api.RoutingKey;
import accord.api.VisibleForImplementation;
import accord.primitives.Range;
-import accord.primitives.Ranges;
import accord.primitives.Routable.Domain;
import accord.primitives.Routables;
import accord.primitives.Txn;
@@ -30,11 +29,13 @@ import accord.primitives.TxnId;
import accord.primitives.Unseekable;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
-import accord.utils.ReducingRangeMap;
import static accord.primitives.TxnId.maxIfNull;
import static accord.primitives.TxnId.noneIfNull;
+import accord.primitives.Ranges;
+import accord.utils.ReducingRangeMap;
+
public class MaxDecidedRX extends ReducingRangeMap<TxnId>
{
public static final MaxDecidedRX EMPTY = new MaxDecidedRX();
diff --git a/accord-core/src/main/java/accord/local/PreLoadContext.java b/accord-core/src/main/java/accord/local/PreLoadContext.java
index e4da78fc..ba2b7170 100644
--- a/accord-core/src/main/java/accord/local/PreLoadContext.java
+++ b/accord-core/src/main/java/accord/local/PreLoadContext.java
@@ -20,6 +20,8 @@ package accord.local;
import accord.api.RoutingKey;
import accord.local.cfk.CommandsForKey;
+import accord.primitives.AbstractUnseekableKeys;
+import accord.primitives.Routable;
import accord.primitives.RoutingKeys;
import accord.primitives.TxnId;
@@ -211,8 +213,10 @@ public interface PreLoadContext
return contextFor(RoutingKeys.of(key), loadKeys, loadKeysFor, describe);
}
- static PreLoadContext contextFor(Unseekables<?> keys, LoadKeys loadKeys, LoadKeysFor loadKeysFor, String reason)
+ // we don't currently permit range queries without an associated TxnId
+ static PreLoadContext contextFor(AbstractUnseekableKeys keys, LoadKeys loadKeys, LoadKeysFor loadKeysFor, String reason)
{
+ Invariants.require(keys.domain() == Routable.Domain.Key);
return new PreLoadContext()
{
@Override public @Nullable TxnId primaryTxnId() { return null; }
diff --git a/accord-core/src/main/java/accord/local/SafeCommand.java b/accord-core/src/main/java/accord/local/SafeCommand.java
index aea21803..d060c9d1 100644
--- a/accord-core/src/main/java/accord/local/SafeCommand.java
+++ b/accord-core/src/main/java/accord/local/SafeCommand.java
@@ -98,9 +98,7 @@ public abstract class SafeCommand
return prev;
participants = participants.filter(LOAD, safeStore, prev.txnId(), prev.executeAtIfKnown());
- Command update = incidentalUpdate(prev.updateParticipants(participants));
- safeStore.progressLog().update(safeStore, txnId, prev, update, false);
- return update;
+ return update(safeStore, prev.updateParticipants(participants));
}
public Command.PreAccepted preaccept(SafeCommandStore safeStore, SaveStatus saveStatus, StoreParticipants participants, Ballot promised, Timestamp executeAt, PartialTxn partialTxn, PartialDeps partialDeps)
diff --git a/accord-core/src/main/java/accord/local/SafeCommandStore.java b/accord-core/src/main/java/accord/local/SafeCommandStore.java
index 626cb918..4900bc13 100644
--- a/accord-core/src/main/java/accord/local/SafeCommandStore.java
+++ b/accord-core/src/main/java/accord/local/SafeCommandStore.java
@@ -18,6 +18,8 @@
package accord.local;
+import java.util.ArrayList;
+import java.util.List;
import java.util.NavigableMap;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -27,6 +29,7 @@ import accord.api.DataStore;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
import accord.api.RoutingKey;
+import accord.impl.InMemoryCommandStore;
import accord.local.CommandStores.RangesForEpochSupplier;
import accord.local.RedundantBefore.RedundantBeforeSupplier;
import accord.local.cfk.CommandsForKey;
@@ -45,9 +48,11 @@ import accord.primitives.Txn.Kind;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
+import accord.utils.Reduce;
import accord.utils.SimpleBitSet;
import accord.utils.SortedList;
import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
import static accord.local.LoadKeys.INCR;
import static accord.local.LoadKeys.NONE;
@@ -56,6 +61,7 @@ import static accord.local.RedundantStatus.SomeStatus.LOCALLY_WITNESSED_ONLY;
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
import static accord.local.cfk.UpdateUnmanagedMode.REGISTER;
+import static accord.primitives.Known.KnownRoute.MaybeRoute;
import static accord.primitives.Routable.Domain.Range;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.SaveStatus.Applied;
@@ -469,7 +475,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
if (execute == context)
{
if (next.txnId().is(Range))
- registerTransitive(safeStore, txnId, next);
+ registerTransitiveRangeDeps(safeStore, txnId, next);
}
else
{
@@ -480,7 +486,7 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
CommandStore unsafeStore = safeStore.commandStore();
AsyncChain<Void> submit = unsafeStore.build(context, safeStore0 -> { updateUnmanagedCommandsForKey(safeStore0, safeStore0.context().keys() , txnId, mode); });
if (next.txnId().is(Range))
- submit = submit.flatMap(success -> unsafeStore.build((PreLoadContext.Empty) () -> "Register Transitive Dependencies", safeStore0 -> { registerTransitive(safeStore0, txnId, next); }));
+ submit = submit.flatMap(success -> unsafeStore.build((PreLoadContext.Empty) () -> "Register Transitive Dependencies", safeStore0 -> { registerTransitiveRangeDeps(safeStore0, txnId, next); }));
submit.begin(safeStore.commandStore().agent);
}
}
@@ -503,13 +509,54 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
}
}
- private static void registerTransitive(SafeCommandStore safeStore, TxnId txnId, Command next)
+ private static void registerTransitiveRangeDeps(SafeCommandStore safeStore, TxnId syncId, Command syncCommand)
{
+ if (!syncId.is(Kind.ExclusiveSyncPoint))
+ return;
+
CommandStore commandStore = safeStore.commandStore();
- Ranges ranges = next.participants().touches().toRanges();
- commandStore.registerTransitive(safeStore, next.partialDeps().rangeDeps);
- if (txnId.is(Kind.ExclusiveSyncPoint))
- commandStore.markSynced(safeStore, txnId, ranges);
+ Ranges touches = syncCommand.participants().touches().toRanges();
+ Ranges waitingOn = commandStore.isWaitingOnSync(syncId, touches);
+ if (waitingOn.isEmpty())
+ return;
+
+ List<AsyncChain<Void>> async = new ArrayList<>();
+ RangeDeps rangeDeps = syncCommand.partialDeps().rangeDeps;
+ rangeDeps.forEachUniqueTxnId(waitingOn, null, (ignore, txnIdWithFlags) -> {
+ TxnId txnId = txnIdWithFlags.withoutNonIdentityFlags();
+ PreLoadContext context = PreLoadContext.contextFor(txnId, "Register Transitive Range Deps");
+ Ranges ranges = rangeDeps.ranges(txnId);
+ if (safeStore.canExecuteWith(context)) registerTransitive(safeStore, txnId, ranges);
+ else async.add(safeStore.commandStore().build(context, safeStore0 -> {
+ registerTransitive(safeStore0, txnId, ranges);
+ }));
+ });
+
+ AsyncChains.ofRunnable(Runnable::run, () -> commandStore.markSyncing(syncId, waitingOn))
+ .flatMap(ignore -> AsyncChains.reduce(async, Reduce.toNull(), null))
+ .begin((success, fail) -> {
+ if (fail == null) commandStore.execute((PreLoadContext.Empty)() -> "Mark Synced", safeStore0 -> commandStore.markSynced(safeStore0, syncId, waitingOn));
+ else commandStore.execute((PreLoadContext.Empty)() -> "Unmark Syncing", safeStore0 -> commandStore.unmarkSyncing(syncId, waitingOn));
+ });
+ }
+
+ private static void registerTransitive(SafeCommandStore safeStore, TxnId txnId, Ranges witnessedBy)
+ {
+ SafeCommand safeCommand = safeStore.unsafeGet(txnId);
+ if (safeCommand != null && safeCommand.current().known().route() != MaybeRoute)
+ return;
+
+ CommandStores.RangesForEpoch rangesForEpoch = safeStore.ranges();
+ // TODO (required): this is incompatible with rebootstrap - we need to use some additional condition
+ witnessedBy = witnessedBy.without(rangesForEpoch.coordinates(txnId)); // already coordinates, no need to replicate
+ if (witnessedBy.isEmpty())
+ return;
+
+ witnessedBy = witnessedBy.slice(rangesForEpoch.allSince(txnId.epoch()), Minimal); // never coordinated, no need to replicate for dependency or recovery calculations
+ if (witnessedBy.isEmpty())
+ return;
+
+ safeCommand.updateParticipants(safeStore, safeCommand.current().participants().supplement(null, witnessedBy));
}
public abstract CommandStore commandStore();
@@ -534,6 +581,11 @@ public abstract class SafeCommandStore implements RangesForEpochSupplier, Redund
return commandStore().unsafeGetRedundantBefore();
}
+ public MaxDecidedRX maxDecidedRX()
+ {
+ return commandStore().unsafeGetMaxDecidedRX();
+ }
+
public DurableBefore durableBefore()
{
return commandStore().node.durableBefore();
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 552a22aa..d6e26bbb 100644
--- a/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
+++ b/accord-core/src/main/java/accord/local/cfk/CommandsForKey.java
@@ -591,9 +591,10 @@ public class CommandsForKey extends CommandsForKeyUpdate
TxnInfo info = (TxnInfo) o;
return status() == info.status()
&& (executeAt == this ? info.executeAt == info : Objects.equals(executeAt, info.executeAt))
- && Arrays.equals(missing(), info.missing());
+ && TxnId.equalsStrict(missing(), info.missing());
}
+
public TxnId plainTxnId()
{
return new TxnId(this);
@@ -1429,7 +1430,7 @@ public class CommandsForKey extends CommandsForKeyUpdate
while (nextPruned != null && nextPruned.compareTo(txn) < 0)
{
if (nextPruned.isVisible && nextPruned.is(testKind))
- visitor.visit(p1, p2, NOT_DIRECTLY_WITNESSED, key, nextPruned.plainTxnId());
+ visitor.visit(p1, p2, NOT_DIRECTLY_WITNESSED, NotDurable, key, nextPruned.plainTxnId());
if (loadingPruned.hasNext()) nextPruned = loadingPruned.next();
else nextPruned = null;
@@ -1497,7 +1498,7 @@ public class CommandsForKey extends CommandsForKeyUpdate
}
}
- visitor.visit(p1, p2, txn.summaryStatus(), key, txn.plainTxnId());
+ visitor.visit(p1, p2, txn.summaryStatus(), txn.isDurable() ? Majority : NotDurable, key, txn.plainTxnId());
}
if (end <= prunedBeforeById)
@@ -1519,7 +1520,7 @@ public class CommandsForKey extends CommandsForKeyUpdate
break;
}
if (best.executeAt.epoch() <= startedBefore.epoch() && !best.equals(startedBefore))
- visitor.visit(p1, p2, best.summaryStatus(), key, best.plainTxnId());
+ visitor.visit(p1, p2, best.summaryStatus(), best.isDurable() ? Majority : NotDurable, key, best.plainTxnId());
}
}
@@ -2037,7 +2038,7 @@ public class CommandsForKey extends CommandsForKeyUpdate
return (int) (unappliedCounters >>> ((WRITE_COUNTERS_DELTA_SHIFT >>> (kindOrdinal * 8)) & 63));
}
- public CommandsForKeyUpdate withRedundantBeforeAtLeast(QuickBounds newBounds)
+ public CommandsForKeyUpdate withRedundantBeforeAtLeast(QuickBounds newBounds, boolean expectUpToDate)
{
// we can't let HLC epoch go backwards as this breaks assumptions around maxUniqueHlc tracking
if (newBounds.gcBefore.hlc() < bounds.gcBefore.hlc())
@@ -2075,9 +2076,9 @@ public class CommandsForKey extends CommandsForKeyUpdate
}
Object[] newLoadingPruned = Pruning.removeRedundantLoadingPruned(loadingPruned, redundantBefore(newBounds));
- TxnInfo[] newById = removeRedundantById(byId, newLoadingPruned != loadingPruned, bounds, newBounds);
+ TxnInfo[] newById = removeRedundantById(byId, newLoadingPruned != loadingPruned, bounds, newBounds, expectUpToDate);
int newPrunedBeforeById = prunedBeforeId(newById, prunedBefore(), redundantBefore(newBounds));
- Invariants.paranoid(newPrunedBeforeById < 0 ? prunedBeforeById < 0 || byId[prunedBeforeById].compareTo(newBounds.gcBefore) < 0 : newById[newPrunedBeforeById].equals(byId[prunedBeforeById]));
+ Invariants.paranoid(!expectUpToDate || (newPrunedBeforeById < 0 ? prunedBeforeById < 0 || byId[prunedBeforeById].compareTo(newBounds.gcBefore) < 0 : newById[newPrunedBeforeById].equals(byId[prunedBeforeById])));
return notifyManagedPreBootstrap(this, newBounds, reconstructAndUpdateUnmanaged(key, newBounds, true, newById, maxUniqueHlc, newLoadingPruned, newPrunedBeforeById, unmanageds));
}
@@ -2088,14 +2089,14 @@ public class CommandsForKey extends CommandsForKeyUpdate
* on load and not no-op due to e.g. newSafelyPrunedBefore or newBootstrappedAt being non-null.
*/
@VisibleForImplementation
- public CommandsForKey withRedundantBeforeAtLeast(TxnId newRedundantBefore)
+ public CommandsForKey withRedundantBeforeAtLeast(TxnId newRedundantBefore, boolean expectUpToDate)
{
QuickBounds newBoundsInfo = bounds.withGcBeforeBeforeAtLeast(newRedundantBefore);
Object[] newLoadingPruned = Pruning.removeRedundantLoadingPruned(loadingPruned, newRedundantBefore);
- TxnInfo[] newById = removeRedundantById(byId, newLoadingPruned != loadingPruned, bounds, newBoundsInfo);
+ TxnInfo[] newById = removeRedundantById(byId, newLoadingPruned != loadingPruned, bounds, newBoundsInfo, expectUpToDate);
int newPrunedBeforeById = prunedBeforeId(newById, prunedBefore(), newRedundantBefore);
- Invariants.paranoid(newPrunedBeforeById < 0 ? prunedBeforeById < 0 : newById[newPrunedBeforeById].equals(byId[prunedBeforeById]));
+ Invariants.paranoid(!expectUpToDate || (newPrunedBeforeById < 0 ? prunedBeforeById < 0 || byId[prunedBeforeById].compareTo(newRedundantBefore) < 0 : newById[newPrunedBeforeById].equals(byId[prunedBeforeById])));
return reconstruct(key, newBoundsInfo, true, newById, maxUniqueHlc, newLoadingPruned, newPrunedBeforeById, unmanageds);
}
@@ -2371,7 +2372,7 @@ public class CommandsForKey extends CommandsForKeyUpdate
return false;
if (!a.ballot().equals(b.ballot()))
return false;
- if (!Arrays.equals(a.missing(), b.missing()))
+ if (!TxnId.equalsStrict(a.missing(), b.missing()))
return false;
}
return true;
diff --git a/accord-core/src/main/java/accord/local/cfk/PostProcess.java b/accord-core/src/main/java/accord/local/cfk/PostProcess.java
index b37ffb56..87a88990 100644
--- a/accord-core/src/main/java/accord/local/cfk/PostProcess.java
+++ b/accord-core/src/main/java/accord/local/cfk/PostProcess.java
@@ -175,20 +175,7 @@ abstract class PostProcess
notify = cachedTxnIds().completeAndDiscard(notify, notifyCount);
PostProcess newPostProcess = new NotifyNotWaiting(update.postProcess(), notify);
-
- CommandsForKey cfk = update.cfk();
- if (maxApplied != null)
- {
- int start = findFirstApply(cfk.unmanageds);
- int end = findApply(cfk.unmanageds, start, maxApplied);
- if (start != end)
- {
- TxnId[] notifyNotWaiting = selectUnmanaged(cfk.unmanageds, start, end);
- cfk = cfk.update(removeUnmanaged(cfk.unmanageds, start, end));
- newPostProcess = new PostProcess.NotifyNotWaiting(newPostProcess, notifyNotWaiting);
- }
- }
- return new CommandsForKeyUpdateWithPostProcess(cfk, newPostProcess);
+ return new CommandsForKeyUpdateWithPostProcess(update.cfk(), newPostProcess);
}
static class NotifyUnmanagedOfCommit extends PostProcess
@@ -337,17 +324,20 @@ abstract class PostProcess
int start = firstApply;
int end = start;
int j = 1 + maxContiguousManagedAppliedIndex(committedByExecuteAt, maxAppliedWriteByExecuteAt, bootstrappedAt);
+ boolean hasUnapplied = false;
while (end < unmanageds.length && j < committedByExecuteAt.length)
{
+ hasUnapplied |= committedByExecuteAt[j].compareTo(APPLIED) < 0;
int c = committedByExecuteAt[j].executeAt.compareTo(unmanageds[end].waitingUntil);
if (c == 0)
{
if (start != end)
{
- TxnId[] notifyNotWaiting = selectUnmanaged(unmanageds, start, end);
+ TxnId[] notify = selectUnmanaged(unmanageds, start, end);
unmanageds = removeUnmanaged(unmanageds, start, end);
end = start;
- notifier = new PostProcess.NotifyNotWaiting(notifier, notifyNotWaiting);
+ notifier = hasUnapplied ? new PostProcess.NotifyUnmanagedOfCommit(notifier, notify)
+ : new PostProcess.NotifyNotWaiting(notifier, notify);
}
start = ++end;
}
@@ -362,9 +352,10 @@ abstract class PostProcess
}
if (start != unmanageds.length)
{
- TxnId[] notifyNotWaiting = selectUnmanaged(unmanageds, start, unmanageds.length);
+ TxnId[] notify = selectUnmanaged(unmanageds, start, unmanageds.length);
unmanageds = removeUnmanaged(unmanageds, start, unmanageds.length);
- notifier = new PostProcess.NotifyNotWaiting(notifier, notifyNotWaiting);
+ notifier = hasUnapplied ? new PostProcess.NotifyUnmanagedOfCommit(notifier, notify)
+ : new PostProcess.NotifyNotWaiting(notifier, notify);
}
}
}
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 68b4ad1d..31a6fff5 100644
--- a/accord-core/src/main/java/accord/local/cfk/Pruning.java
+++ b/accord-core/src/main/java/accord/local/cfk/Pruning.java
@@ -530,7 +530,7 @@ public class Pruning
return epochPrunedBefores;
}
- static TxnInfo[] removeRedundantById(TxnInfo[] byId, boolean hasRedundantLoadingPruned, QuickBounds prevBounds, QuickBounds newBounds)
+ static TxnInfo[] removeRedundantById(TxnInfo[] byId, boolean hasRedundantLoadingPruned, QuickBounds prevBounds, QuickBounds newBounds, boolean expectUpToDate)
{
TxnId newAppliedBefore = appliedBefore(newBounds);
TxnId newRedundantBefore = redundantBefore(newBounds);
@@ -573,7 +573,7 @@ public class Pruning
{
if (byId[i].mayExecute())
{
- Invariants.require((byId[i].isNot(COMMITTED) && byId[i].isNot(STABLE)) || !reportLinearizabilityViolations(), "%s redundant; expected to be applied, undecided or to execute in a future epoch", byId[i]);
+ Invariants.require(!expectUpToDate || (byId[i].isNot(COMMITTED) && byId[i].isNot(STABLE)) || !reportLinearizabilityViolations(), "%s redundant; expected to be applied, undecided or to execute in a future epoch", byId[i]);
// we only filter those that would apply locally
removeUnappliedCount++;
}
diff --git a/accord-core/src/main/java/accord/local/cfk/SafeCommandsForKey.java b/accord-core/src/main/java/accord/local/cfk/SafeCommandsForKey.java
index 9c7158a5..9b716fce 100644
--- a/accord-core/src/main/java/accord/local/cfk/SafeCommandsForKey.java
+++ b/accord-core/src/main/java/accord/local/cfk/SafeCommandsForKey.java
@@ -98,7 +98,7 @@ public abstract class SafeCommandsForKey implements SafeState<CommandsForKey>
public void updateRedundantBefore(SafeCommandStore safeStore, RedundantBefore.Bounds redundantBefore)
{
CommandsForKey prevCfk = current();
- update(safeStore, null, prevCfk, prevCfk.withRedundantBeforeAtLeast(redundantBefore), false);
+ update(safeStore, null, prevCfk, prevCfk.withRedundantBeforeAtLeast(redundantBefore, true), false);
}
public void initialize()
diff --git a/accord-core/src/main/java/accord/local/cfk/Serialize.java b/accord-core/src/main/java/accord/local/cfk/Serialize.java
index 9162c71a..a204b42f 100644
--- a/accord-core/src/main/java/accord/local/cfk/Serialize.java
+++ b/accord-core/src/main/java/accord/local/cfk/Serialize.java
@@ -77,6 +77,7 @@ public class Serialize
private static final int HAS_BALLOT_HEADER_BIT_SHIFT = 2;
private static final int HAS_STATUS_OVERRIDES_HEADER_BIT = 0x8;
private static final int HAS_STATUS_OVERRIDES_HEADER_BIT_SHIFT = 3;
+ private static final int HAS_MISSING_DEPS_FLAGS_HEADER_BIT = 0x10;
private static final int COMMAND_HEADER_BIT_FLAGS_MASK = 0x1f;
private static final int HAS_BOOTSTRAPPED_AT_HEADER_BIT = 0x20;
private static final int HAS_BOUNDS_FLAGS_HEADER_BIT = 0x40;
@@ -96,6 +97,7 @@ public class Serialize
* bit 1 is set if there are any executeAt specified
* bit 2 is set if there are any ballots specified
* bit 3 is set if there are any queries with override flags
+ * bit 4 is set if there are any queries with missing ids that have non-identity flags set
* bits 6-7 number of header bytes to read for each command
* bits 8-9: level 0 extra hlc bytes to read
* bits 10-11: level 1 extra hlc bytes to read (+ 1 + level 0)
@@ -159,8 +161,8 @@ public class Serialize
// first compute the unique Node Ids and some basic characteristics of the data, such as
// whether we have any missing transactions to encode, any executeAt that are not equal to their TxnId
// and whether there are any non-standard flag bits to encode
- int nodeIdCount, missingIdCount = 0, executeAtCount = 0, ballotCount = 0, overrideCount = 0;
- int bitsPerExecuteAtEpoch = 0, bitsPerExecuteAtFlags = 0, bitsPerExecuteAtHlc = 1; // to permit us to use full 64 bits and encode in 5 bits we force at least one hlc bit
+ int nodeIdCount, missingIdCount = 0, missingIdWithFlagsCount = 0, executeAtCount = 0, ballotCount = 0, overrideCount = 0;
+ int bitsPerExecuteAtEpoch = 0, bitsPerExecuteAtFlags = 0, bitsPerExecuteAtHlc = 1, bitsPerMissingIdFlags = 0; // to permit us to use full 64 bits and encode in 5 bits we force at least one hlc bit
{
nodeIds[0] = cfk.redundantBefore().node.id;
nodeIdCount = 1;
@@ -196,7 +198,21 @@ public class Serialize
if (txn.getClass() == TxnInfoExtra.class)
{
TxnInfoExtra extra = (TxnInfoExtra) txn;
- missingIdCount += extra.missing.length;
+ TxnId[] missing = extra.missing;
+ if (missing.length > 0)
+ {
+ Invariants.require(txn.hasDeps());
+ missingIdCount += missing.length;
+ for (TxnId txnId : missing)
+ {
+ int nonIdentityFlags = txnId.nonIdentityFlags();
+ if (nonIdentityFlags != 0)
+ {
+ ++missingIdWithFlagsCount;
+ bitsPerMissingIdFlags = Math.max(bitsPerMissingIdFlags, numberOfBitsToRepresent(nonIdentityFlags));
+ }
+ }
+ }
Invariants.require(extra.missing.length == 0 || txn.hasDeps());
if (extra.ballot != Ballot.ZERO)
{
@@ -228,9 +244,9 @@ public class Serialize
// of additional space we'll need to store the TxnId and its basic info
int bitsPerNodeId = numberOfBitsToRepresent(nodeIdCount);
int minHeaderBits = 10 + bitsPerNodeId + (overrideCount > 0 ? 1 : 0);
- int headerFlags = (executeAtCount > 0 ? 1 : 0)
- | (missingIdCount > 0 ? 2 : 0)
- | (ballotCount > 0 ? 4 : 0);
+ int headerFlags = (executeAtCount > 0 ? HAS_EXECUTE_AT_HEADER_BIT : 0)
+ | (missingIdCount > 0 ? HAS_MISSING_DEPS_HEADER_BIT : 0)
+ | (ballotCount > 0 ? HAS_BALLOT_HEADER_BIT : 0);
int maxHeaderBits = minHeaderBits;
int totalBytes = 0;
@@ -281,11 +297,12 @@ public class Serialize
totalBytes += 1 + encodedFlagBits;
if (txn.hasExecuteAt())
- headerBits += headerFlags & 0x1;
+ headerBits += (headerFlags >>> HAS_EXECUTE_AT_HEADER_BIT_SHIFT) & 1;
if (txn.hasDeps())
- headerBits += (headerFlags >>> 1) & 0x1;
+ headerBits += (headerFlags >>> HAS_MISSING_DEPS_HEADER_BIT_SHIFT) & 1;
if (txn.hasBallot())
- headerBits += (headerFlags >>> 2);
+ headerBits += (headerFlags >>> HAS_BALLOT_HEADER_BIT_SHIFT) & 1;
+
maxHeaderBits = Math.max(headerBits, maxHeaderBits);
int basicBytes = (headerBits + payloadBits + 7)/8;
bytesHistogram[basicBytes]++;
@@ -302,6 +319,7 @@ public class Serialize
bytesHistogram[i] += bytesHistogram[i-1];
int globalFlags = (missingIdCount > 0 ? HAS_MISSING_DEPS_HEADER_BIT : 0)
+ | (missingIdWithFlagsCount > 0 ? HAS_MISSING_DEPS_FLAGS_HEADER_BIT : 0)
| (executeAtCount > 0 ? HAS_EXECUTE_AT_HEADER_BIT : 0)
| (ballotCount > 0 ? HAS_BALLOT_HEADER_BIT : 0)
| (overrideCount > 0 ? HAS_STATUS_OVERRIDES_HEADER_BIT : 0)
@@ -404,7 +422,9 @@ public class Serialize
totalBytes += 2; // encode bit widths
// account for encoding missing id stream
- int missingIdBits = 1 + numberOfBitsToRepresent(commandCount);
+ int missingIdBits = (missingIdWithFlagsCount > 0 ? 1 : 0) // bit to encode whether there are extra flags
+ + 1 // last element bit
+ + numberOfBitsToRepresent(commandCount);
int executeAtBits = bitsPerNodeId
+ bitsPerExecuteAtEpoch
+ bitsPerExecuteAtHlc
@@ -414,6 +434,7 @@ public class Serialize
+ bitsPerBallotHlc
+ bitsPerBallotFlags;
totalBytes += (missingIdBits * missingIdCount
+ + bitsPerMissingIdFlags * missingIdWithFlagsCount + (missingIdWithFlagsCount > 0 ? 4 : 0)
+ executeAtBits * executeAtCount
+ (ballotCount > 0 ? ballotBits * (ballotCount - 1) + bitsPerNodeId + 128 : 0)
+ 7)/8;
@@ -607,6 +628,14 @@ public class Serialize
long buffer = 0L;
int bufferCount = 0;
+ if (0 != (globalFlags & HAS_MISSING_DEPS_FLAGS_HEADER_BIT))
+ {
+ Invariants.require(bitsPerMissingIdFlags < 16);
+ buffer = flushBits(buffer, bufferCount, bitsPerMissingIdFlags, 4, out);
+ bufferCount = (bufferCount + 4) & 63;
+ ++bitsPerMissingId;
+ }
+
Ballot prevBallot = null;
for (int i = 0 ; i < commandCount ; ++i)
{
@@ -648,17 +677,21 @@ public class Serialize
TxnId[] missing = extra.missing;
if (missing.length > 0)
{
- int j = 0;
- while (j < missing.length - 1)
+ for (int j = 0 ; j < missing.length ; ++j)
{
- int missingId = cfk.indexOf(missing[j++]);
+ TxnId txnId = missing[j];
+ int missingId = cfk.indexOf(txnId);
+ int nonIdentityFlags = txnId.nonIdentityFlags();
+ missingId |= (nonIdentityFlags != 0 ? 1 : 0) << (bitsPerCommandId + 1);
+ missingId |= (j == missing.length - 1 ? 1 : 0) << bitsPerCommandId;
buffer = flushBits(buffer, bufferCount, missingId, bitsPerMissingId, out);
bufferCount = (bufferCount + bitsPerMissingId) & 63;
+ if (nonIdentityFlags != 0)
+ {
+ buffer = flushBits(buffer, bufferCount, nonIdentityFlags, bitsPerMissingIdFlags, out);
+ bufferCount = (bufferCount + bitsPerMissingIdFlags) & 63;
+ }
}
- int missingId = cfk.indexOf(missing[missing.length - 1]);
- missingId |= 1 << bitsPerCommandId;
- buffer = flushBits(buffer, bufferCount, missingId, bitsPerMissingId, out);
- bufferCount = (bufferCount + bitsPerMissingId) & 63;
}
Ballot ballot = extra.ballot;
@@ -927,6 +960,13 @@ public class Serialize
Ballot prevBallot = null;
final BitUtils.BitReader reader = new BitUtils.BitReader();
+ int bitsPerMissingIdFlags = 0;
+ if (0 != (globalFlags & HAS_MISSING_DEPS_FLAGS_HEADER_BIT))
+ {
+ bitsPerMissingIdFlags = (int) reader.read(4, in);
+ ++bitsPerMissingId;
+ }
+
for (int i = 0 ; i < commandCount ; ++i)
{
TxnId txnId = txnIds[i];
@@ -969,9 +1009,21 @@ public class Serialize
int next = (int) reader.read(bitsPerMissingId, in);
Invariants.require(next > prev);
- missingIdBuffer[missingIdCount++] = txnIds[next & txnIdMask];
- if (next >= commandCount)
- break; // finished this array
+ int controlBits = next >>> Integer.bitCount(txnIdMask);
+ next &= txnIdMask;
+ missingIdBuffer[missingIdCount++] = txnIds[next];
+ if (controlBits != 0)
+ {
+ // either finished this array or have non-standard flags
+ if (controlBits > 1)
+ {
+ int nonIdentityFlags = (int) reader.read(bitsPerMissingIdFlags, in);
+ missingIdBuffer[missingIdCount - 1] = missingIdBuffer[missingIdCount - 1].addFlags(nonIdentityFlags << TxnId.NON_IDENTITY_FLAGS_SHIFT);
+ controlBits ^= 2;
+ }
+ if (controlBits != 0)
+ break; // finished this array
+ }
prev = next;
}
diff --git a/accord-core/src/main/java/accord/local/durability/DurabilityService.java b/accord-core/src/main/java/accord/local/durability/DurabilityService.java
index ee83530d..2df4976b 100644
--- a/accord-core/src/main/java/accord/local/durability/DurabilityService.java
+++ b/accord-core/src/main/java/accord/local/durability/DurabilityService.java
@@ -33,7 +33,9 @@ import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService;
import accord.api.Timeouts.Timeout;
import accord.local.Node;
+import accord.primitives.Range;
import accord.primitives.Ranges;
+import accord.primitives.SyncPoint;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.topology.Topology;
@@ -48,7 +50,7 @@ public class DurabilityService implements ConfigurationService.Listener
private static final Logger logger = LoggerFactory.getLogger(DurabilityService.class);
public enum SyncLocal { NoLocal, Self }
- public enum SyncRemote { NoRemote, Quorum, All }
+ public enum SyncRemote { NoRemote, MinorityQuorum, Quorum, All }
private boolean started;
private final Node node;
@@ -130,6 +132,17 @@ public class DurabilityService implements ConfigurationService.Listener
return submit(new DurabilityRequest(requestedBy, minBound, ranges, local, remote, include, startedAt, timeoutAt)).result;
}
+ public AsyncResult<Void> sync(Object requestedBy, SyncPoint<Range> syncPoint, SyncLocal local, SyncRemote remote, long timeoutDelay, TimeUnit timeoutUnits)
+ {
+ long startedAt = node.elapsed(MICROSECONDS);
+ long timeoutAt = startedAt + timeoutUnits.toMicros(timeoutDelay);
+ DurabilityRequest request = new DurabilityRequest(requestedBy, syncPoint.syncId, syncPoint.route.toRanges(), local, remote, null, startedAt, timeoutAt);
+ logger.info("Requesting durability {}", request);
+ register(request);
+ shards.queue().submit(syncPoint, request);
+ return request.result;
+ }
+
private DurabilityRequest submit(DurabilityRequest request)
{
register(request);
diff --git a/accord-core/src/main/java/accord/local/durability/ShardDurability.java b/accord-core/src/main/java/accord/local/durability/ShardDurability.java
index d49fb129..82db5db2 100644
--- a/accord-core/src/main/java/accord/local/durability/ShardDurability.java
+++ b/accord-core/src/main/java/accord/local/durability/ShardDurability.java
@@ -346,6 +346,9 @@ public class ShardDurability
synchronized void start()
{
+ if (!ShardDurability.this.isStarted())
+ return;
+
Invariants.require(scheduled == null);
Invariants.require(active != null);
ShardDistributor distributor = node.commandStores().shardDistributor();
@@ -451,6 +454,7 @@ public class ShardDurability
private final ConcurrencyControl syncPointControl = new ConcurrencyControl(8);
private final DurabilityQueue durabilityQueue;
private long latestEpoch;
+ private List<DurabilityRequest> waitingOnStartup;
volatile boolean stop;
volatile Scheduled scheduled;
@@ -469,17 +473,13 @@ public class ShardDurability
public synchronized void setTargetShardSplits(int targetShardSplits)
{
this.targetShardSplits = BitUtil.findNextPositivePowerOfTwo(targetShardSplits);
- if (scheduled != null)
- scheduled.cancel();
- scheduled = node.scheduler().recurring(this::tick, shardCycleTimeMicros / targetShardSplits, MICROSECONDS);
+ reschedule();
}
public synchronized void setShardCycleTime(long newShardCycleTime, TimeUnit units)
{
shardCycleTimeMicros = units.toMicros(newShardCycleTime);
- if (scheduled != null)
- scheduled.cancel();
- scheduled = node.scheduler().recurring(this::tick, shardCycleTimeMicros / targetShardSplits, MICROSECONDS);
+ reschedule();
}
public synchronized void reconfigure(int targetShardSplits, int maxShardSplits, long newShardCycleTime, TimeUnit units)
@@ -487,7 +487,16 @@ public class ShardDurability
this.maxShardSplits = maxShardSplits;
this.targetShardSplits = BitUtil.findNextPositivePowerOfTwo(targetShardSplits);
this.shardCycleTimeMicros = units.toMicros(newShardCycleTime);
- Invariants.require(scheduled == null);
+ reschedule();
+ }
+
+ private void reschedule()
+ {
+ if (scheduled != null)
+ {
+ scheduled.cancel();
+ scheduled = node.scheduler().recurring(this::tick, shardCycleTimeMicros / targetShardSplits, MICROSECONDS);
+ }
}
/**
@@ -497,6 +506,17 @@ public class ShardDurability
{
Invariants.require(!stop); // cannot currently restart safely
scheduled = node.scheduler().recurring(this::tick, shardCycleTimeMicros / targetShardSplits, MICROSECONDS);
+ if (waitingOnStartup != null)
+ {
+ List<DurabilityRequest> submit = waitingOnStartup;
+ waitingOnStartup = null;
+ submit.forEach(this::request);
+ }
+ }
+
+ private boolean isStarted()
+ {
+ return scheduled != null;
}
private synchronized void tick()
@@ -520,6 +540,14 @@ public class ShardDurability
public synchronized void request(DurabilityRequest request)
{
+ if (!isStarted())
+ {
+ if (waitingOnStartup == null)
+ waitingOnStartup = new ArrayList<>();
+ waitingOnStartup.add(request);
+ return;
+ }
+
request(request, request.ranges);
}
@@ -531,6 +559,7 @@ public class ShardDurability
private void request(DurabilityRequest request, Range range)
{
+ Invariants.require(isStarted());
{
Map.Entry<Range, ShardScheduler> e = shardSchedulers.floorEntry(range);
// we look up ceiling entry as well, because our comparison on both start and end
@@ -595,6 +624,11 @@ public class ShardDurability
}
}
+ DurabilityQueue queue()
+ {
+ return durabilityQueue;
+ }
+
synchronized void updateTopology(Topology latestGlobal)
{
if (latestGlobal.epoch() <= latestEpoch)
diff --git a/accord-core/src/main/java/accord/primitives/AbstractKeys.java b/accord-core/src/main/java/accord/primitives/AbstractKeys.java
index 40ad6799..4b6af0b9 100644
--- a/accord-core/src/main/java/accord/primitives/AbstractKeys.java
+++ b/accord-core/src/main/java/accord/primitives/AbstractKeys.java
@@ -174,12 +174,12 @@ public abstract class AbstractKeys<K extends RoutableKey> implements Iterable<K>
@Override
public String toString()
{
- return toString("", null);
+ return toString(null);
}
- protected String toString(String suffix, @Nullable Function<? super K, String> keySuffixes)
+ protected String toString(@Nullable RoutingKey printHomeKey)
{
- if (isEmpty() && suffix.isEmpty())
+ if (isEmpty() && printHomeKey == null)
return "[]";
StringBuilder sb = new StringBuilder();
@@ -189,6 +189,12 @@ public abstract class AbstractKeys<K extends RoutableKey> implements Iterable<K>
{
if (i > 0) sb.append(", ");
Object prefix = keys[i].prefix();
+ RoutingKey homeKey = null;
+ if (printHomeKey != null && Objects.equals(prefix, printHomeKey.prefix()))
+ {
+ homeKey = printHomeKey;
+ printHomeKey = null;
+ }
int j = i + 1;
while (j < keys.length && Objects.equals(prefix, keys[j].prefix()))
++j;
@@ -202,15 +208,42 @@ public abstract class AbstractKeys<K extends RoutableKey> implements Iterable<K>
{
K key = keys[i++];
sb.append(key.printableSuffix());
- if (keySuffixes != null)
- sb.append(keySuffixes.apply(key));
+ if (homeKey != null && Objects.equals(homeKey.suffix(), key.suffix()))
+ {
+ sb.append('*');
+ homeKey = null;
+ }
if (i < j) sb.append(',');
}
+ if (prefix != null)
+ {
+ sb.append(']');
+ }
+ if (homeKey != null)
+ {
+ sb.append('{');
+ sb.append(homeKey.printableSuffix());
+ sb.append("*}");
+ }
+ }
+ if (printHomeKey == null) sb.append(']');
+ else
+ {
+ if (keys.length > 0)
+ sb.append(", ");
+
+ Object prefix = printHomeKey.prefix();
+ if (prefix != null)
+ {
+ sb.append(printHomeKey.prefix());
+ sb.append(":[");
+ }
+ sb.append("]{");
+ sb.append(printHomeKey.printableSuffix());
+ sb.append("*}");
if (prefix != null)
sb.append(']');
}
- sb.append(suffix);
- sb.append(']');
return sb.toString();
}
diff --git a/accord-core/src/main/java/accord/primitives/AbstractRanges.java b/accord-core/src/main/java/accord/primitives/AbstractRanges.java
index 62f9e8d6..db97ffb1 100644
--- a/accord-core/src/main/java/accord/primitives/AbstractRanges.java
+++ b/accord-core/src/main/java/accord/primitives/AbstractRanges.java
@@ -28,6 +28,8 @@ import com.google.common.collect.Iterators;
import accord.api.Key;
import accord.api.RoutingKey;
+import accord.utils.ArrayBuffers;
+import accord.utils.ArrayBuffers.ObjectBufferCache;
import accord.utils.ArrayBuffers.ObjectBuffers;
import accord.utils.IndexedFoldToLong;
import accord.utils.IndexedTriFold;
@@ -205,6 +207,36 @@ public abstract class AbstractRanges implements Iterable<Range>, Routables<Range
return SortedArrays.findNextIntersectionWithMultipleMatches(ranges, thisi, that.ranges, thati, Range::compareIntersecting, Range::compareIntersecting);
}
+ @Override
+ public final long findFirstIntersection(AbstractRanges that)
+ {
+ if (this.ranges.length == 0 || that.ranges.length == 0)
+ return -1;
+
+ boolean found = true;
+ int thisi = SortedArrays.binarySearch(ranges, 0, ranges.length, that.ranges[0], Range::compareIntersecting, CEIL);
+ if (thisi < 0)
+ {
+ found = false;
+ thisi = -1 - thisi;
+ }
+ if (thisi == ranges.length)
+ return -1;
+ int thati = SortedArrays.binarySearch(that.ranges, 0, that.ranges.length, ranges[thisi], Range::compareIntersecting, CEIL);
+ if (thati < 0)
+ {
+ thati = -1 - thati;
+ found = false;
+ }
+ if (thati == that.ranges.length)
+ return -1;
+
+ if (found)
+ return thisi | ((long)thati << 32);
+
+ return findNextIntersection(thisi, that, thati);
+ }
+
// returns ki in bottom 32 bits, ri in top, or -1 if no match found
public final long findNextExactIntersection(int thisi, AbstractRanges that, int thati)
{
@@ -376,42 +408,53 @@ public abstract class AbstractRanges implements Iterable<Range>, Routables<Range
static <C extends AbstractRanges, P, O> O sliceMinimal(C covering, AbstractRanges input, P param, SliceConstructor<C, P, O> constructor)
{
ObjectBuffers<Range> cachedRanges = cachedRanges();
-
- Range[] buffer = cachedRanges.get(covering.ranges.length + input.ranges.length);
+ Range[] buffer = null;
int bufferCount = 0;
try
{
- int li = 0, ri = 0;
- while (true)
+ long lri = covering.findFirstIntersection(input);
+ while (lri >= 0)
{
- long lri = covering.findNextIntersection(li, input, ri);
- if (lri < 0)
- break;
-
- if (bufferCount == buffer.length)
- buffer = cachedRanges.resize(buffer, bufferCount, bufferCount + 1 + (bufferCount/2));
-
- li = (int) (lri);
- ri = (int) (lri >>> 32);
+ int li = (int) (lri);
+ int ri = (int) (lri >>> 32);
Range l = covering.ranges[li], r = input.ranges[ri];
RoutingKey ls = l.start(), rs = r.start(), le = l.end(), re = r.end();
int cs = rs.compareTo(ls), ce = re.compareTo(le);
if (cs >= 0 && ce <= 0)
{
- buffer[bufferCount++] = r;
+ if (buffer != null || ri != bufferCount)
+ {
+ buffer = initOrResize(buffer, bufferCount, input.ranges, cachedRanges);
+ buffer[bufferCount] = r;
+ }
+ ++bufferCount;
++ri;
}
else
{
+ buffer = initOrResize(buffer, bufferCount, input.ranges, cachedRanges);
buffer[bufferCount++] = r.newRange(cs >= 0 ? rs : ls, ce <= 0 ? re : le);
if (ce <= 0) ++ri;
}
if (ce >= 0) li++; // le <= re
+
+ lri = covering.findNextIntersection(li, input, ri);
+ }
+
+ if (buffer == null)
+ {
+ Range[] ranges = input.ranges;
+ if (bufferCount != input.ranges.length)
+ ranges = Arrays.copyOf(ranges, bufferCount);
+ return constructor.construct(covering, param, ranges);
+ }
+ else
+ {
+ Range[] result = cachedRanges.complete(buffer, bufferCount);
+ cachedRanges.discard(buffer, bufferCount);
+ return constructor.construct(covering, param, result);
}
- Range[] result = cachedRanges.complete(buffer, bufferCount);
- cachedRanges.discard(buffer, bufferCount);
- return constructor.construct(covering, param, result);
}
catch (Throwable t)
{
@@ -420,6 +463,21 @@ public abstract class AbstractRanges implements Iterable<Range>, Routables<Range
}
}
+ private static Range[] initOrResize(Range[] buffer, int bufferCount, Range[] input, ObjectBuffers<Range> cachedRanges)
+ {
+ if (buffer == null)
+ {
+ buffer = cachedRanges.get(input.length);
+ if (bufferCount > 0)
+ System.arraycopy(input, 0, buffer, 0, bufferCount);
+ }
+ else if (bufferCount == buffer.length)
+ {
+ buffer = cachedRanges.resize(buffer, bufferCount, bufferCount + 1 + (bufferCount/2));
+ }
+ return buffer;
+ }
+
static <C extends AbstractRanges, P, O> O sliceMaximal(C covering, AbstractRanges input, P param, SliceConstructor<C, P, O> constructor)
{
ObjectBuffers<Range> cachedRanges = cachedRanges();
diff --git a/accord-core/src/main/java/accord/primitives/AbstractUnseekableKeys.java b/accord-core/src/main/java/accord/primitives/AbstractUnseekableKeys.java
index 0f68707b..5489fc84 100644
--- a/accord-core/src/main/java/accord/primitives/AbstractUnseekableKeys.java
+++ b/accord-core/src/main/java/accord/primitives/AbstractUnseekableKeys.java
@@ -142,7 +142,7 @@ implements Iterable<RoutingKey>, Unseekables<RoutingKey>, Participants<RoutingKe
}
@Override
- public Participants<RoutingKey> without(Unseekables<?> keysOrRanges)
+ public AbstractUnseekableKeys without(Unseekables<?> keysOrRanges)
{
switch (keysOrRanges.domain())
{
@@ -165,7 +165,7 @@ implements Iterable<RoutingKey>, Unseekables<RoutingKey>, Participants<RoutingKe
return without((AbstractRanges) ranges);
}
- private Participants<RoutingKey> without(AbstractRanges ranges)
+ private AbstractUnseekableKeys without(AbstractRanges ranges)
{
RoutingKey[] output = subtract(ranges, keys, RoutingKey[]::new);
return output == keys ? this : new RoutingKeys(output);
diff --git a/accord-core/src/main/java/accord/primitives/KeyRoute.java b/accord-core/src/main/java/accord/primitives/KeyRoute.java
index 6f2c08e2..0d0c0429 100644
--- a/accord-core/src/main/java/accord/primitives/KeyRoute.java
+++ b/accord-core/src/main/java/accord/primitives/KeyRoute.java
@@ -168,7 +168,6 @@ public abstract class KeyRoute extends AbstractUnseekableKeys implements Route<R
@Override
public String toString()
{
- boolean containsHomeKey = containsHomeKey();
- return toString(containsHomeKey ? "" : "(homeKey:" + homeKey + ')', containsHomeKey ? test -> test.equals(homeKey) ? "*" : "" : null);
+ return toString(homeKey);
}
}
diff --git a/accord-core/src/main/java/accord/primitives/Routables.java b/accord-core/src/main/java/accord/primitives/Routables.java
index 09c66a17..b24024ba 100644
--- a/accord-core/src/main/java/accord/primitives/Routables.java
+++ b/accord-core/src/main/java/accord/primitives/Routables.java
@@ -111,6 +111,13 @@ public interface Routables<K extends Routable> extends Iterable<K>
FullRoute<?> toRoute(RoutingKey homeKey);
+ /**
+ * Search forwards from the beginning of both collections to find the first entries in each collection
+ * that intersect with each other. Return their position packed in a long, with low bits representing
+ * the resultant {@code thisIndex} and high bits {@code withIndex}.
+ */
+ default long findFirstIntersection(AbstractRanges with) { return findNextIntersection(0, with, 0); }
+
/**
* Search forwards from {code thisIndex} and {@code withIndex} to find the first entries in each collection
* that intersect with each other. Return their position packed in a long, with low bits representing
diff --git a/accord-core/src/main/java/accord/primitives/Timestamp.java b/accord-core/src/main/java/accord/primitives/Timestamp.java
index c1e0ceb4..751fe414 100644
--- a/accord-core/src/main/java/accord/primitives/Timestamp.java
+++ b/accord-core/src/main/java/accord/primitives/Timestamp.java
@@ -97,6 +97,7 @@ public class Timestamp implements Comparable<Timestamp>, EpochSupplier
static final int MERGE_FLAGS = REJECTED.bit | UNSTABLE.bit | HLC_BOUND.bit | SHARD_BOUND.bit;
public static final long IDENTITY_LSB = 0xFFFFFFFF_FFFF00FFL;
public static final int IDENTITY_FLAGS = 0x00000000_000000FF;
+ public static final int NON_IDENTITY_FLAGS_SHIFT = Integer.bitCount(IDENTITY_FLAGS);
public static final int KIND_AND_DOMAIN_FLAGS = 0x00000000_0000000F;
public static final long MAX_EPOCH = (1L << 48) - 1;
private static final long HLC_INCR = 1L << 16;
diff --git a/accord-core/src/main/java/accord/primitives/TxnId.java b/accord-core/src/main/java/accord/primitives/TxnId.java
index 48d477ae..5c5c223f 100644
--- a/accord-core/src/main/java/accord/primitives/TxnId.java
+++ b/accord-core/src/main/java/accord/primitives/TxnId.java
@@ -298,6 +298,11 @@ public class TxnId extends Timestamp
return (flags() & ~IDENTITY_FLAGS) == 0 ? this : new TxnId(msb, lsb & IDENTITY_LSB, node);
}
+ public final int nonIdentityFlags()
+ {
+ return flags() >>> NON_IDENTITY_FLAGS_SHIFT;
+ }
+
public final boolean hasPrivilegedCoordinator()
{
return FastPath.hasPrivilegedCoordinator(flagsUnmasked());
@@ -520,4 +525,17 @@ public class TxnId extends Timestamp
return null;
return fromValues(Long.parseLong(m.group("epoch")), Long.parseLong(m.group("hlc")), Integer.parseInt(m.group("flags")), new Id(Integer.parseInt(m.group("node"))));
}
+
+ public static boolean equalsStrict(TxnId[] a, TxnId[] b)
+ {
+ if (a.length != b.length)
+ return false;
+
+ for (int i = 0 ; i < a.length ; ++i)
+ {
+ if (!a[i].equalsStrict(b[i]))
+ return false;
+ }
+ return true;
+ }
}
diff --git a/accord-core/src/main/java/accord/primitives/Unseekables.java b/accord-core/src/main/java/accord/primitives/Unseekables.java
index 832440ca..4d64e6fc 100644
--- a/accord-core/src/main/java/accord/primitives/Unseekables.java
+++ b/accord-core/src/main/java/accord/primitives/Unseekables.java
@@ -60,7 +60,6 @@ public interface Unseekables<K extends Unseekable> extends Iterable<K>, Routable
}
}
-
@Override
Unseekables<K> slice(int from, int to);
@Override
@@ -137,4 +136,6 @@ public interface Unseekables<K extends Unseekable> extends Iterable<K>, Routable
return left.with(right);
}
+
+ Ranges toRanges();
}
diff --git a/accord-core/src/main/java/accord/topology/Shard.java b/accord-core/src/main/java/accord/topology/Shard.java
index 28d55da3..73a354ce 100644
--- a/accord-core/src/main/java/accord/topology/Shard.java
+++ b/accord-core/src/main/java/accord/topology/Shard.java
@@ -113,6 +113,11 @@ public class Shard
flags);
}
+ public final int minorityQuorumSize()
+ {
+ return (rf + 1) / 2;
+ }
+
public boolean electorateIsSubset()
{
return !notInFastPath.isEmpty();
diff --git a/accord-core/src/main/java/accord/utils/async/AsyncChains.java b/accord-core/src/main/java/accord/utils/async/AsyncChains.java
index 5fd001ce..c1b15c66 100644
--- a/accord-core/src/main/java/accord/utils/async/AsyncChains.java
+++ b/accord-core/src/main/java/accord/utils/async/AsyncChains.java
@@ -115,6 +115,10 @@ public abstract class AsyncChains<V> implements AsyncChain<V>
{
cause.addSuppressed(t);
}
+ catch (Error rethrow)
+ {
+ throw rethrow;
+ }
catch (Throwable ignore)
{
// can't add as suppressed...
diff --git a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
index 4e0e9f80..f00ce10b 100644
--- a/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
+++ b/accord-core/src/test/java/accord/impl/RemoteListenersTest.java
@@ -68,6 +68,7 @@ import accord.utils.AccordGens;
import accord.utils.RandomSource;
import accord.utils.RandomTestRunner;
import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
import org.agrona.collections.IntHashSet;
import org.agrona.collections.ObjectHashSet;
@@ -415,7 +416,6 @@ public class RemoteListenersTest
@Override public AsyncChain<Void> build(PreLoadContext context, Consumer<? super SafeCommandStore> consumer) { return null; }
@Override public <T> AsyncChain<T> build(PreLoadContext context, Function<? super SafeCommandStore, T> apply) { return null; }
@Override public void shutdown() {}
- @Override protected void registerTransitive(SafeCommandStore safeStore, RangeDeps deps) { }
@Override public <T> AsyncChain<T> build(Callable<T> task) { return null; }
}
diff --git a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
index e90c205f..4e7261f2 100644
--- a/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
+++ b/accord-core/src/test/java/accord/local/cfk/CommandsForKeyTest.java
@@ -89,6 +89,7 @@ import accord.utils.DefaultRandom;
import accord.utils.Invariants;
import accord.utils.RandomSource;
import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResults;
import static accord.local.Command.Executed.executed;
@@ -1017,9 +1018,6 @@ public class CommandsForKeyTest
Invariants.require(queue.isEmpty());
}
- @Override
- protected void registerTransitive(SafeCommandStore safeStore, RangeDeps deps){ }
-
@Override
public <T> AsyncChain<T> build(Callable<T> task)
{
diff --git a/accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java b/accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java
index 45a2be5a..34c16a8e 100644
--- a/accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java
+++ b/accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java
@@ -223,8 +223,7 @@ public class BTreeReducingRangeMapTest
}
catch (Throwable t)
{
- if (!(t instanceof AssertionFailedError))
- throw new RuntimeException(id, t);
+ throw new AssertionFailedError(id, t);
}
}
diff --git a/accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java b/accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java
index 8bd97f99..a1a84a7b 100644
--- a/accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java
+++ b/accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java
@@ -229,8 +229,7 @@ public class ReducingRangeMapTest
}
catch (Throwable t)
{
- if (!(t instanceof AssertionFailedError))
- throw new RuntimeException(id, t);
+ throw new AssertionFailedError(id, t);
}
}
@@ -279,7 +278,6 @@ public class ReducingRangeMapTest
addOneRandom(random, maxNumberOfRangesPerAddition, maxCoveragePerAddition, minRoutingKeyChance);
}
-
static ReducingRangeMap<Timestamp> build(Random random, int count, int maxNumberOfRangesPerAddition, float maxCoveragePerRange, float chanceOfMinRoutingKey)
{
RandomMap result = new RandomMap();
diff --git a/accord-core/src/test/resources/burn-logback.xml b/accord-core/src/test/resources/burn-logback.xml
index 4d3f729c..61cf7055 100644
--- a/accord-core/src/test/resources/burn-logback.xml
+++ b/accord-core/src/test/resources/burn-logback.xml
@@ -37,6 +37,7 @@
</appender>
<logger name="accord.local.durability.DurabilityService" level="ERROR"/>
+ <logger name="accord.impl.progresslog.DefaultProgressLog" level="ERROR"/>
<logger name="accord.local.durability.DurabilityQueue" level="ERROR"/>
<logger name="accord.local.durability.ShardDurability" level="ERROR"/>
<logger name="accord.local.durability.GlobalDurability" level="ERROR"/>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org