This is an automated email from the ASF dual-hosted git repository. ifesdjeen pushed a commit to branch CASSANDRA-20112 in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit 81adb7832974e36ea71bc0c1cba0de01d5de69cb Author: Benedict Elliott Smith <[email protected]> AuthorDate: Fri Nov 22 15:24:08 2024 +0000 chance to have full cache, and meta-randomise chance of presence in cache --- accord-core/src/main/java/accord/local/Commands.java | 6 ++++-- accord-core/src/main/java/accord/primitives/Txn.java | 5 +++++ .../src/main/java/accord/primitives/TxnId.java | 5 +++++ .../src/test/java/accord/impl/basic/Cluster.java | 13 ++++++++++--- .../java/accord/impl/basic/DelayedCommandStores.java | 20 +++++++++++--------- 5 files changed, 35 insertions(+), 14 deletions(-) diff --git a/accord-core/src/main/java/accord/local/Commands.java b/accord-core/src/main/java/accord/local/Commands.java index 357fa013..13b1f8d6 100644 --- a/accord-core/src/main/java/accord/local/Commands.java +++ b/accord-core/src/main/java/accord/local/Commands.java @@ -80,6 +80,7 @@ import static accord.primitives.Status.Stable; import static accord.primitives.Status.Truncated; import static accord.primitives.Route.isFullRoute; import static accord.primitives.Txn.Kind.EphemeralRead; +import static accord.primitives.Txn.Kind.ExclusiveSyncPoint; import static accord.primitives.Txn.Kind.Read; import static accord.utils.Invariants.illegalState; @@ -968,13 +969,14 @@ public class Commands case LOCALLY_REDUNDANT: case REDUNDANT_PRE_BOOTSTRAP_OR_STALE: case PRE_BOOTSTRAP_OR_STALE: + case WAS_OWNED_RETIRED: removeRedundantDependencies(safeStore, waitingSafe, loadDepId); break; case WAS_OWNED: case WAS_OWNED_CLOSED: - case WAS_OWNED_RETIRED: - removeNoLongerOwnedDependency(safeStore, waitingSafe, loadDepId); + if (!waitingId.awaitsPreviouslyOwned()) + removeNoLongerOwnedDependency(safeStore, waitingSafe, loadDepId); break; case LIVE: diff --git a/accord-core/src/main/java/accord/primitives/Txn.java b/accord-core/src/main/java/accord/primitives/Txn.java index 55645602..3771357f 100644 --- a/accord-core/src/main/java/accord/primitives/Txn.java +++ b/accord-core/src/main/java/accord/primitives/Txn.java @@ -246,6 +246,11 @@ public interface Txn return 0 != (ENCODED_ORDINAL_INFO & (1L << (AWAITS_ONLY_DEPS_ORDINAL_INFO_OFFSET + ordinal))); } + public static boolean awaitsPreviouslyOwned(int ordinal) + { + return ordinal == ExclusiveSyncPoint.ordinal(); + } + public Kinds witnesses() { switch (this) diff --git a/accord-core/src/main/java/accord/primitives/TxnId.java b/accord-core/src/main/java/accord/primitives/TxnId.java index 3f5fa5cb..186e8252 100644 --- a/accord-core/src/main/java/accord/primitives/TxnId.java +++ b/accord-core/src/main/java/accord/primitives/TxnId.java @@ -127,6 +127,11 @@ public class TxnId extends Timestamp return Kind.awaitsOnlyDeps(kindOrdinal(flags())); } + public boolean awaitsPreviouslyOwned() + { + return Kind.awaitsPreviouslyOwned(kindOrdinal(flags())); + } + public boolean is(Domain domain) { return domainOrdinal(flags()) == domain.ordinal(); diff --git a/accord-core/src/test/java/accord/impl/basic/Cluster.java b/accord-core/src/test/java/accord/impl/basic/Cluster.java index d712dce3..8223817c 100644 --- a/accord-core/src/test/java/accord/impl/basic/Cluster.java +++ b/accord-core/src/test/java/accord/impl/basic/Cluster.java @@ -528,25 +528,32 @@ public class Cluster BurnTestConfigurationService configService = new BurnTestConfigurationService(id, nodeExecutor, randomSupplier, topology, nodeMap::get, topologyUpdates); DelayedCommandStores.CacheLoadingChance isLoadedCheck = new DelayedCommandStores.CacheLoadingChance() { + final float presentChance = random.nextBoolean() ? 1.0f : random.nextFloat(); private final BooleanSupplier cacheEmptyChance = Gens.supplier(Gens.bools().mixedDistribution().next(random), random); public boolean cacheEmpty() { return cacheEmptyChance.getAsBoolean(); } - private final BooleanSupplier commandLoadedChance = random.biasedUniformBools(0.1f); + private final BooleanSupplier cacheFullChance = Gens.supplier(Gens.bools().mixedDistribution().next(random), random); + public boolean cacheFull() + { + return cacheFullChance.getAsBoolean(); + } + + private final BooleanSupplier commandLoadedChance = random.biasedUniformBools(presentChance); public boolean commandLoaded() { return commandLoadedChance.getAsBoolean(); } - private final BooleanSupplier cfkLoadedChance = random.biasedUniformBools(0.1f); + private final BooleanSupplier cfkLoadedChance = random.biasedUniformBools(presentChance); public boolean cfkLoaded() { return cfkLoadedChance.getAsBoolean(); } - private final BooleanSupplier tfkLoadedChance = random.biasedUniformBools(0.1f); + private final BooleanSupplier tfkLoadedChance = random.biasedUniformBools(presentChance); public boolean tfkLoaded() { return tfkLoadedChance.getAsBoolean(); diff --git a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java index 7a41fe72..e3f7d8be 100644 --- a/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java +++ b/accord-core/src/test/java/accord/impl/basic/DelayedCommandStores.java @@ -131,14 +131,14 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread private final SimulatedDelayedExecutorService executor; private final Queue<Task<?>> pending = new LinkedList<>(); - private final CacheLoadingChance isLoadedCheck; + private final CacheLoadingChance cacheLoadingChance; private final Journal journal; - public DelayedCommandStore(int id, NodeCommandStoreService time, Agent agent, DataStore store, ProgressLog.Factory progressLogFactory, LocalListeners.Factory listenersFactory, EpochUpdateHolder epochUpdateHolder, SimulatedDelayedExecutorService executor, CacheLoadingChance isLoadedCheck, Journal journal) + public DelayedCommandStore(int id, NodeCommandStoreService time, Agent agent, DataStore store, ProgressLog.Factory progressLogFactory, LocalListeners.Factory listenersFactory, EpochUpdateHolder epochUpdateHolder, SimulatedDelayedExecutorService executor, CacheLoadingChance cacheLoadingChance, Journal journal) { super(id, time, agent, store, progressLogFactory, listenersFactory, epochUpdateHolder); this.executor = executor; - this.isLoadedCheck = isLoadedCheck; + this.cacheLoadingChance = cacheLoadingChance; this.journal = journal; } @@ -161,7 +161,7 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread @Override protected boolean canExposeUnloaded() { - return !isLoadedCheck.cacheEmpty(); + return !cacheLoadingChance.cacheEmpty(); } private static CommandStore.Factory factory(SimulatedDelayedExecutorService executor, CacheLoadingChance isLoadedCheck, Journal journal) @@ -247,7 +247,7 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread @Override protected InMemorySafeStore createSafeStore(PreLoadContext context, RangesForEpoch ranges, Map<TxnId, InMemorySafeCommand> commands, Map<RoutableKey, InMemorySafeTimestampsForKey> timestampsForKey, Map<RoutableKey, InMemorySafeCommandsForKey> commandsForKeys) { - return new DelayedSafeStore(this, ranges, context, commands, timestampsForKey, commandsForKeys, isLoadedCheck); + return new DelayedSafeStore(this, ranges, context, commands, timestampsForKey, commandsForKeys, cacheLoadingChance); } @Override @@ -292,13 +292,14 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread @Override protected InMemoryCommandStore.InMemoryCommandStoreCaches tryGetCaches() { - if (commandStore.canExposeUnloaded()) + if (!cacheLoadingChance.cacheEmpty()) { + boolean cacheFull = cacheLoadingChance.cacheFull(); return commandStore.new InMemoryCommandStoreCaches() { @Override public InMemorySafeCommand acquireIfLoaded(TxnId txnId) { - if (cacheLoadingChance.commandLoaded()) + if (cacheFull || cacheLoadingChance.commandLoaded()) return super.acquireIfLoaded(txnId); return null; } @@ -306,7 +307,7 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread @Override public InMemorySafeTimestampsForKey acquireTfkIfLoaded(RoutingKey key) { - if (cacheLoadingChance.tfkLoaded()) + if (cacheFull || cacheLoadingChance.tfkLoaded()) return super.acquireTfkIfLoaded(key); return null; } @@ -314,7 +315,7 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread @Override public InMemorySafeCommandsForKey acquireIfLoaded(RoutingKey key) { - if (cacheLoadingChance.cfkLoaded()) + if (cacheFull || cacheLoadingChance.cfkLoaded()) return super.acquireIfLoaded(key); return null; } @@ -336,6 +337,7 @@ public class DelayedCommandStores extends InMemoryCommandStores.SingleThread public interface CacheLoadingChance { boolean cacheEmpty(); + boolean cacheFull(); boolean commandLoaded(); boolean cfkLoaded(); boolean tfkLoaded(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
