This is an automated email from the ASF dual-hosted git repository.
belliottsmith 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 ce855de1 Accond: Torn Single Partition Reads
ce855de1 is described below
commit ce855de13902e7e3ebfa4356d265f0c49be1bb8e
Author: Benedict Elliott Smith <[email protected]>
AuthorDate: Fri Oct 9 10:39:52 2026 +0100
Accond: Torn Single Partition Reads
To support linearizable ephemeral reads, writes to a single partition must
be atomic, and the ephemeral read must itself be performed atomically against
the partition state. The former we achieve by merging updates to the same
partition before submitting them to be written; the latter we achieve by simply
(for now) disabling single key ephemeral reads if there is more than one read
to perform. Ephemeral reads for range read compatibility is unaffected, as
these have weaker isolation g [...]
patch by Benedict; reviewed by Aleksey Yeschenko for CASSANDRA-21741
---
.../impl/progresslog/DefaultProgressLog.java | 39 ++++++++++++++++------
.../java/accord/impl/progresslog/TxnState.java | 6 ++++
2 files changed, 34 insertions(+), 11 deletions(-)
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 affba896..c4c55717 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/DefaultProgressLog.java
@@ -136,7 +136,21 @@ public class DefaultProgressLog implements ProgressLog,
Consumer<SafeCommandStor
private final Object2ObjectHashMap<TxnId, PendingTask> pendingHome = new
Object2ObjectHashMap<>();
private final Long2ObjectHashMap<Object> active = new
Long2ObjectHashMap<>();
- private final Map<TxnId, TxnState> debugDeleted = Invariants.debug() &&
Invariants.isParanoid() && Invariants.isTesting() ? new
Object2ObjectHashMap<>() : null;
+ public static final int DEBUG_LEVEL;
+ private static final TxnState DEBUG_SENTINEL = new TxnState(TxnId.NONE);
+ static
+ {
+ int debugLevel = Invariants.debug() ? 1 : 0;
+ String str = System.getProperty("accord.debug.progresslog");
+ if (str != null)
+ {
+ try { debugLevel = Integer.parseInt(str); }
+ catch (Throwable t) { logger.error("Failed to parse system
property accord.debug.progresslog=" + str); }
+ }
+ DEBUG_LEVEL = debugLevel;
+ }
+
+ private final Map<TxnId, TxnState> debugDeleted = DEBUG_LEVEL > 0 ? new
Object2ObjectHashMap<>() : null;
// peek into the future by this many micros when picking what to run.
// this is primarily to handle the fact that we do not support 0 delay
when scheduling, so tasks that should be run immediately are schedule with 1us
delay
@@ -200,7 +214,7 @@ public class DefaultProgressLog implements ProgressLog,
Consumer<SafeCommandStor
TxnState result = BTree.<TxnId, TxnState>find(stateMap, (id, state) ->
id.compareTo(state.txnId), txnId);
if (result == null)
{
- Invariants.require(debugDeleted == null ||
!debugDeleted.containsKey(txnId));
+ Invariants.require(DEBUG_LEVEL == 0 ||
!debugDeleted.containsKey(txnId));
stateMap = BTree.update(stateMap, BTree.singleton(result = new
TxnState(txnId)), TxnState::compareTo);
}
return result;
@@ -208,7 +222,7 @@ public class DefaultProgressLog implements ProgressLog,
Consumer<SafeCommandStor
private TxnState insert(TxnId txnId)
{
- Invariants.require(debugDeleted == null ||
!debugDeleted.containsKey(txnId));
+ Invariants.require(DEBUG_LEVEL == 0 ||
!debugDeleted.containsKey(txnId));
TxnState result = new TxnState(txnId);
stateMap = BTree.update(stateMap, BTree.singleton(result),
TxnState::compareTo);
return result;
@@ -447,7 +461,7 @@ public class DefaultProgressLog implements ProgressLog,
Consumer<SafeCommandStor
pendingWaiting.clear();
pendingHome.clear();
- if (debugDeleted != null)
+ if (DEBUG_LEVEL > 0)
debugDeleted.clear();
runBuffer = EMPTY_RUN_BUFFER;
@@ -465,10 +479,13 @@ public class DefaultProgressLog implements ProgressLog,
Consumer<SafeCommandStor
void remove(TxnState state)
{
stateMap = BTreeRemoval.<TxnId, TxnState>remove(stateMap, (id, s) ->
id.compareTo(s.txnId), state.txnId);
- if (debugDeleted != null)
+ if (DEBUG_LEVEL > 0)
{
- DeletedTxnState copy = new DeletedTxnState(state);
- debugDeleted.put(copy.txnId, copy);
+ TxnState save;
+ if (DEBUG_LEVEL == 1) save = DEBUG_SENTINEL;
+ else if (debugDeletion == null) save = new TxnState(state);
+ else save = new DeletedTxnState(state);
+ debugDeleted.put(save.txnId, save);
}
}
@@ -1202,13 +1219,13 @@ public class DefaultProgressLog implements ProgressLog,
Consumer<SafeCommandStor
final Object debug;
protected DeletedTxnState(TxnState copy)
{
- super(copy.txnId);
- this.encodedState = copy.encodedState;
- this.debug = debugDeletion.apply(copy.txnId);
+ super(copy);
+ Function<TxnId, ?> f = debugDeletion;
+ this.debug = f == null ? null : debugDeletion.apply(copy.txnId);
}
}
- private static volatile Function<TxnId, ?> debugDeletion = id -> null;
+ private static volatile Function<TxnId, ?> debugDeletion = null;
public static void setDebugDeletion(Function<TxnId, ?> newDebugDeletion)
{
debugDeletion = newDebugDeletion;
diff --git a/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
b/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
index acc8e85b..6456e8a5 100644
--- a/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
+++ b/accord-core/src/main/java/accord/impl/progresslog/TxnState.java
@@ -49,6 +49,12 @@ public class TxnState extends WaitingState implements
ExecutionContext
super(txnId);
}
+ protected TxnState(TxnState copy)
+ {
+ super(copy.txnId);
+ this.encodedState = copy.encodedState;
+ }
+
final void updateScheduling(SafeCommandStore safeStore, DefaultProgressLog
owner, TxnStateKind updated, @Nullable BlockedUntil blockedUntil, Progress
newProgress)
{
long newDelay;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
