natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Accord Executor QoS 

Fairness: tasks are grouped by kind of work, and for each group we track work arrival and service via decaying counters. If work for some group(s) begin to be served at a much lower rate than some other group (e.g. because that group has a large backlog, that by simple priority comes first), then we begin processing some fraction of the work by service-ratio rather than priority. This means that we cannot starve request processing because, e.g., a sync point has produced thousands of state save tasks.

Incremental processing: tasks that process many keys may be declared INCR or ASYNC. ASYNC indicates the keys are needed (so should be loaded), but may be used in follow-up work so the task may be executed if the data is not yet loaded. An ASYNC task may be followed by an INCR task that processes keys as they are loaded, with limits to the batch sizes. This limits the latency impact of processing larger transactions such as sync points. INCR tasks may be declared to be processed in either an arbitrary order with no isolation, or “atomically” with its parent task. In the latter case, the complete unit of work appears to execute as a single execution to external observers (blocking progress on unprocessed keys).

Key-level ordering: AccordCacheEntry can inflate a special Queue object that can be used to impose ordering constraints: FIFO for atomic work (once it has been part processed, or submitted as part of a parent task); prioritised for tasks that have a key-level priority to impose (i.e. PreAccept/Accept/Commit/Stable/Apply want to be ordered by TxnId); and unsequenced tasks for those that have no sequencing restrictions.

Additional improvements:
 - AccordCacheEntry are now explicitly LOCKED for the duration of their usage, which may span multiple executions for long running INCR tasks.
 - ExclusiveExecutor releases its ownership lock immediately, since AccordCacheEntry locks guarantee safety (as they will not be released until the task is cleaned up)
 - CassandraThread tracks active and locked AccordExecutor, so that we may safely and more aggressively use executeMaybeImmediately, safe in the knowledge it will never permit cross-executor executions or multiple executor locks to be held.

Also:
 - Rename PreLoadContext -> ExecutionContext
 - Break out AccordExecutor/AccordTask into separate package
This commit is contained in:
Benedict Elliott Smith 2026-07-15 22:50:50 +01:00
parent 1cffe9a8a0
commit 37b4fbf08f
39 changed files with 3317 additions and 1973 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit fff32de2e915772fbc70d16b1c32346313877838
Subproject commit 84ac4db03251c3b842c98a29868e072792662f51

14
src/java/org/apache/cassandra/concurrent/CassandraThread.java
View File

@ -31,6 +31,7 @@ public class CassandraThread extends FastThreadLocalThread implements AccordExec
private ExecutorLocals executorLocals;
private AccordExecutor accordActiveExecutor;
private AccordExecutor accordLockedExecutor;
private int accordLockedExecutorDepth;
private volatile AccordExecutor.Task accordActiveTask;
private static final AtomicReferenceFieldUpdater<CassandraThread, AccordExecutor.Task> accordActiveTaskUpdater = AtomicReferenceFieldUpdater.newUpdater(CassandraThread.class, AccordExecutor.Task.class, "accordActiveTask");
@ -113,18 +114,19 @@ public class CassandraThread extends FastThreadLocalThread implements AccordExec
}
@Override
public final boolean trySetAccordLockedExecutor(AccordExecutor newLockedExecutor)
public final boolean tryEnterAccordLockedExecutor(AccordExecutor newLockedExecutor)
{
if (accordLockedExecutor != null)
return false;
accordLockedExecutor = newLockedExecutor;
if (accordLockedExecutor == null) accordLockedExecutor = newLockedExecutor;
else if (accordLockedExecutor != newLockedExecutor) return false;
++accordLockedExecutorDepth;
return true;
}
@Override
public final void clearAccordLockedExecutor()
public final void exitAccordLockedExecutor()
{
accordLockedExecutor = null;
if (--accordLockedExecutorDepth == 0)
accordLockedExecutor = null;
}
public final AccordExecutor.Task accordActiveTask()

47
src/java/org/apache/cassandra/config/AccordConfig.java
View File

@ -156,31 +156,12 @@ public class AccordConfig
*/
PHASE_ONLY,
/**
* Always pick the task by priority.
*/
PRIORITY_BUDGET,
/**
* Pick by phase first, so the highest phase with budget ALWAYS runs.
* Within a phase, pick by priority.
*/
PHASE_BUDGET,
/**
* Pick by phase first, selecting the phase that has processed the least work recently relative to arrivals.
* Within a phase, pick by priority.
*/
PHASE_FAIR,
/**
* Pick by phase first, selecting the phase that has processed the least work recently relative to arrivals.
* However, each phase has a budget that is consumed on dispatch, and we pick only from those phases with budget.
* If there is no phase with budget, the budget resets.
* Within a phase, pick by priority.
*/
PHASE_BUDGET_FAIR,
/**
* While phases are within a threshold of imbalance, pick tasks by priority.
* Once the threshold is crossed, over-processed phases have a small penalty applied
@ -190,8 +171,6 @@ public class AccordConfig
* Within a phase, pick by priority.
*/
BLENDED_PRIORITY_PHASE_FAIR,
BLENDED_PRIORITY_PHASE_BUDGET_FAIR,
}
public QueueShardModel queue_shard_model = THREAD_POOL_PER_SHARD;
@ -212,8 +191,29 @@ public class AccordConfig
public Integer queue_flow_imbalance_onset = null;
public Integer queue_flow_imbalance_width_shift = null;
public String queue_active_limits;
public String queue_budgets;
public Boolean queue_nonsync_enabled;
/**
* Size at which we will begin processing a task that is ASYNC, INCR OR INCR_ATOMIC.
* Note that a size of zero will effectively give implicit priority to INCR_ATOMIC tasks, as they may immediately
* take a FIFO queue slot (which is processed preferentially).
*/
public Integer queue_nonsync_min_batch_size;
/**
* If there are more than min_batch_size keys ready for an ASYNC, INCR or INCR_ATOMIC task,
* process up to this many keys at once.
*/
public Integer queue_nonsync_max_batch_size;
/**
* An ASYNC, INCR or INCR_ATOMIC task that is ready to run but waiting for batch_size work will proceed
* once this number of tasks are blocked behind it, regardless of batch_size.
*/
public Integer queue_nonsync_blocked_limit;
/**
* If set, the signal loop does not match park/unpark pairs, but instead consumers perform timed-park spin waits
@ -240,9 +240,6 @@ public class AccordConfig
*/
public volatile OptionaldPositiveInt command_store_shard_count = OptionaldPositiveInt.UNDEFINED;
public volatile OptionaldPositiveInt max_queued_loads = OptionaldPositiveInt.UNDEFINED;
public volatile OptionaldPositiveInt max_queued_range_loads = OptionaldPositiveInt.UNDEFINED;
public volatile OptionaldPositiveInt progress_log_concurrency = OptionaldPositiveInt.UNDEFINED;
public DurationSpec.IntMillisecondsBound progress_log_query_fallback_timeout = new DurationSpec.IntMillisecondsBound("1m");

4
src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
View File

@ -1650,7 +1650,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
{
try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches())
{
AccordCacheEntry<TxnId, Command> entry = caches.commands().getUnsafe(txnId);
AccordCacheEntry<TxnId, Command, ?> entry = caches.commands().getUnsafe(txnId);
return entry == null ? null : entry.getExclusive();
}
}
@ -1895,7 +1895,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
AccordService.getBlocking(accord.node()
.commandStores()
.forId(commandStoreId)
.chain(ExecutionContext.contextFor(txnId, TXN_OPS), apply)
.chain(ExecutionContext.unsequenced(txnId, TXN_OPS), apply)
.flatMap(i -> i));
}

240
src/java/org/apache/cassandra/service/accord/AccordCache.java
View File

@ -44,6 +44,7 @@ import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.SafeState;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.Serialize;
import accord.primitives.Routable;
@ -66,6 +67,7 @@ import org.apache.cassandra.metrics.LogLinearHistogram;
import org.apache.cassandra.metrics.ShardedHitRate;
import org.apache.cassandra.service.accord.AccordCache.Adapter.Shrink;
import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor;
import org.apache.cassandra.service.accord.AccordCacheEntry.Loading;
import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
import org.apache.cassandra.service.accord.AccordSafeCommandsForKey.CommandsForKeyCacheEntry;
import org.apache.cassandra.service.accord.events.CacheEvents;
@ -79,6 +81,8 @@ import org.apache.cassandra.utils.ObjectSizes;
import static accord.utils.Invariants.illegalState;
import static accord.utils.Invariants.require;
import static org.apache.cassandra.service.accord.AccordCacheEntry.AGE_MASK;
import static org.apache.cassandra.service.accord.AccordCacheEntry.GENERATION_MASK;
import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.EVICTED;
import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.LOADED;
import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.MODIFIED;
@ -106,7 +110,7 @@ public class AccordCache implements CacheSize
VALIDATE_LOAD_ON_EVICT = value;
}
public interface Adapter<K, V, S>
public interface Adapter<K, V, S extends SafeState<V> & AccordSafeState<K, V, S>>
{
enum Shrink { EVICT, DONE, PERFORM_WITHOUT_LOCK }
@ -122,10 +126,10 @@ public class AccordCache implements CacheSize
long estimateHeapSize(V value);
long estimateShrunkHeapSize(Object shrunk);
boolean validate(AccordCommandStore commandStore, K key, V value);
S safeRef(AccordCacheEntry<K, V> node);
S safeRef(AccordCacheEntry<K, V, S> node);
default Comparator<K> keyComparator() { return null; }
default AccordCacheEntry<K, V> newEntry(K key, AccordCache.Type<K, V, ?>.Instance owner)
default AccordCacheEntry<K, V, S> newEntry(K key, AccordCache.Type<K, V, S>.Instance owner)
{
return AccordCacheEntry.createReadyToLoad(key, owner);
}
@ -151,8 +155,8 @@ public class AccordCache implements CacheSize
private final List<Type<?, ?, ?>> types = new CopyOnWriteArrayList<>();
final AccordCacheEntry.SaveExecutor saveExecutor;
private final IntrusiveLinkedList<AccordCacheEntry<?,?>> evictQueue = new IntrusiveLinkedList<>();
private final IntrusiveLinkedList<AccordCacheEntry<?,?>> noEvictQueue = new IntrusiveLinkedList<>();
private final IntrusiveLinkedList<AccordCacheEntry<?,?, ?>> evictQueue = new IntrusiveLinkedList<>();
private final IntrusiveLinkedList<AccordCacheEntry<?,?, ?>> noEvictQueue = new IntrusiveLinkedList<>();
private long unreferencedBytes;
private int unreferenced;
@ -192,16 +196,16 @@ public class AccordCache implements CacheSize
*/
void processNoEvictQueue()
{
noEvictGeneration = (noEvictGeneration + 1) & 0xffff;
noEvictGeneration = (noEvictGeneration + 1) & GENERATION_MASK;
if (noEvictQueue.isEmpty())
return;
Iterator<AccordCacheEntry<?, ?>> iter = noEvictQueue.iterator();
Iterator<AccordCacheEntry<?, ?, ?>> iter = noEvictQueue.iterator();
int skipCount = 3;
while (skipCount > 0 && iter.hasNext())
{
AccordCacheEntry<?, ?> entry = iter.next();
int age = (noEvictGeneration - entry.noEvictGeneration()) & 0xffff;
AccordCacheEntry<?, ?, ?> entry = iter.next();
int age = (noEvictGeneration - entry.noEvictGeneration()) & GENERATION_MASK;
if (age >= entry.noEvictMaxAge())
{
evictNoEvict.warn(entry, age, entry.noEvictMaxAge());
@ -226,14 +230,14 @@ public class AccordCache implements CacheSize
while (bytesCached > maxSizeInBytes && !evictQueue.isEmpty())
{
AccordCacheEntry<?, ?> node = evictQueue.peek();
AccordCacheEntry<?, ?, ?> node = evictQueue.peek();
shrinkOrEvict(lock, node);
}
tryShrinkOrEvict = false;
}
@VisibleForTesting
private <K, V> void shrinkOrEvict(Lock lock, AccordCacheEntry<K, V> node)
private <K, V> void shrinkOrEvict(Lock lock, AccordCacheEntry<K, V, ?> node)
{
require(node.references() == 0);
@ -244,7 +248,7 @@ public class AccordCache implements CacheSize
}
else
{
IntrusiveLinkedList<AccordCacheEntry<?,?>> queue;
IntrusiveLinkedList<AccordCacheEntry<?,?, ?>> queue;
queue = node.isNoEvict() ? noEvictQueue : evictQueue;
node.unlink();
if (shrink == Shrink.DONE)
@ -272,7 +276,7 @@ public class AccordCache implements CacheSize
}
@VisibleForTesting
public <K, V> void tryEvict(AccordCacheEntry<K, V> node)
public <K, V> void tryEvict(AccordCacheEntry<K, V, ?> node)
{
require(node.references() == 0);
@ -290,7 +294,6 @@ public class AccordCache implements CacheSize
case LOADING:
node.loading().loading.cancel();
case WAITING_TO_LOAD:
Invariants.paranoid(node.loadingOrWaiting().waiters == null);
case LOADED:
node.unlink();
evict(node, true);
@ -305,7 +308,7 @@ public class AccordCache implements CacheSize
}
}
public void saveWhenReadyExclusive(AccordCacheEntry<?, ?> entry, Runnable onSuccess)
public void saveWhenReadyExclusive(AccordCacheEntry<?, ?, ?> entry, Runnable onSuccess)
{
if (!entry.isSavingOrWaiting() && !entry.saveWhenReady())
onSuccess.run();
@ -313,7 +316,7 @@ public class AccordCache implements CacheSize
entry.savingOrWaitingToSave().identity.onSuccess(onSuccess);
}
private <K> void evict(AccordCacheEntry<K, ?> node, boolean updateUnreferenced)
private <K> void evict(AccordCacheEntry<K, ?, ?> node, boolean updateUnreferenced)
{
if (logger.isTraceEnabled())
logger.trace("Evicting {}", node);
@ -336,25 +339,25 @@ public class AccordCache implements CacheSize
if (node.status() == LOADED && VALIDATE_LOAD_ON_EVICT)
owner.validateLoadEvicted(node);
AccordCacheEntry<K, ?> self = node.owner.remove(node.key());
AccordCacheEntry<K, ?, ?> self = node.owner.remove(node.key());
Invariants.require(self.references() == 0);
require(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
node.notifyListeners(Listener::onEvict);
node.evicted();
}
<P1, P2, K, V> Collection<AccordTask<?>> load(LoadExecutor<P1, P2> loadExecutor, P1 p1, P2 p2, AccordCacheEntry<K, V> node)
<P1, P2, K, V> Loading load(LoadExecutor<P1, P2> loadExecutor, P1 p1, P2 p2, AccordCacheEntry<K, V, ?> node)
{
return node.load(loadExecutor, p1, p2).waiters();
return node.load(loadExecutor, p1, p2);
}
<K, V> void loaded(AccordCacheEntry<K, V> node, V value)
<K, V> void loaded(AccordCacheEntry<K, V, ?> node, V value)
{
node.loaded(value);
node.notifyListeners(Listener::onUpdate);
}
<K, V> void failedToLoad(AccordCacheEntry<K, V> node)
<K, V> void failedToLoad(AccordCacheEntry<K, V, ?> node)
{
Invariants.require(node.references() == 0);
if (node.isUnqueued())
@ -367,38 +370,38 @@ public class AccordCache implements CacheSize
evict(node, true);
}
<K, V> void saved(AccordCacheEntry<K, V> node, Object identity, Throwable fail)
<K, V> void saved(AccordCacheEntry<K, V, ?> node, Object identity, Throwable fail)
{
if (node.saved(identity, fail) && node.references() == 0 && node.isUnqueued())
evictQueue.addFirst(node); // add to front since we have just saved, so we were eligible for eviction
}
public <K, V, S extends AccordSafeState<K, V>> void release(S safeRef, AccordTask<?> owner)
public <K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> void release(S safeRef, AccordTask<?> owner)
{
safeRef.global().owner.release(safeRef, owner);
}
public <K, V, S extends AccordSafeState<K, V>> Type<K, V, S> newType(Class<K> keyClass, Adapter<K, V, S> adapter, AccordCacheMetrics.Shard metrics)
public <K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> Type<K, V, S> newType(Class<K> keyClass, Adapter<K, V, S> adapter, AccordCacheMetrics.Shard metrics)
{
Type<K, V, S> instance = new Type<>(keyClass, adapter, metrics);
types.add(instance);
return instance;
}
public <K, V, S extends AccordSafeState<K, V>> Type<K, V, S> newType(
public <K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> Type<K, V, S> newType(
Class<K> keyClass,
BiFunction<AccordCommandStore, K, V> loadFunction,
QuadFunction<AccordCommandStore, K, V, Object, Runnable> saveFunction,
Function<V, V> quickShrink,
TriFunction<AccordCommandStore, K, V, Boolean> validateFunction,
ToLongFunction<V> heapEstimator,
Function<AccordCacheEntry<K, V>, S> safeRefFactory,
Function<AccordCacheEntry<K, V, S>, S> safeRefFactory,
AccordCacheMetrics.Shard metrics)
{
return newType(keyClass, loadFunction, saveFunction, quickShrink, (i, j) -> j, (c, i, j) -> (V)j, validateFunction, heapEstimator, i -> 0, safeRefFactory, metrics);
}
public <K, V, S extends AccordSafeState<K, V>> Type<K, V, S> newType(
public <K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> Type<K, V, S> newType(
Class<K> keyClass,
BiFunction<AccordCommandStore, K, V> loadFunction,
QuadFunction<AccordCommandStore, K, V, Object, Runnable> saveFunction,
@ -408,7 +411,7 @@ public class AccordCache implements CacheSize
TriFunction<AccordCommandStore, K, V, Boolean> validateFunction,
ToLongFunction<V> heapEstimator,
ToLongFunction<Object> shrunkHeapEstimator,
Function<AccordCacheEntry<K, V>, S> safeRefFactory,
Function<AccordCacheEntry<K, V, S>, S> safeRefFactory,
AccordCacheMetrics.Shard metrics)
{
return newType(keyClass, new FunctionalAdapter<>(loadFunction, saveFunction, quickShrink,
@ -425,18 +428,18 @@ public class AccordCache implements CacheSize
public interface Listener<K, V>
{
default void onAdd(AccordCacheEntry<K, V> state) {}
default void onUpdate(AccordCacheEntry<K, V> state) {}
default void onEvict(AccordCacheEntry<K, V> state) {}
default void onAdd(AccordCacheEntry<K, V, ?> state) {}
default void onUpdate(AccordCacheEntry<K, V, ?> state) {}
default void onEvict(AccordCacheEntry<K, V, ?> state) {}
}
public class Type<K, V, S extends AccordSafeState<K, V>> implements CacheSize
public class Type<K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> implements CacheSize
{
public class Instance implements Iterable<AccordCacheEntry<K, V>>
public class Instance implements Iterable<AccordCacheEntry<K, V, S>>
{
final AccordCommandStore commandStore;
// TODO (desired): don't need to store key separately as stored in node; ideally use a hash set that allows us to get the current entry
private final Map<K, AccordCacheEntry<K, V>> cache = new Object2ObjectHashMap<>();
private final Map<K, AccordCacheEntry<K, V, S>> cache = new Object2ObjectHashMap<>();
private List<Listener<K, V>> listeners = null;
// TODO (expected): update this after releasing the lock
private OrderedKeys<K> orderedKeys;
@ -446,50 +449,51 @@ public class AccordCache implements CacheSize
this.commandStore = commandStore;
}
public S acquire(K key)
public final S acquire(K key)
{
AccordCacheEntry<K, V> node = acquire(key, false);
AccordCacheEntry<K, V, S> node = acquire(key, false);
return adapter.safeRef(node);
}
public S acquireIfLoaded(K key)
public final S acquireIfLoadedAndPermitted(K key)
{
AccordCacheEntry<K, V> node = acquire(key, true);
AccordCacheEntry<K, V, S> node = acquire(key, true);
if (node == null)
return null;
return adapter.safeRef(node);
}
public S acquire(AccordCacheEntry<K, V> node)
public final S acquire(AccordCacheEntry<K, V, S> node)
{
Invariants.require(node.owner == this);
acquireExisting(node, false);
return adapter.safeRef(node);
}
public void recordPreAcquired(AccordSafeState<K, V> ref)
public final void recordPreAcquired(AccordCacheEntry<K, V, S> entry)
{
Invariants.require(ref.global().owner == this);
incrementCacheHits();
Invariants.require(entry.owner == this);
if (entry.isLoaded()) incrementCacheHits();
else incrementCacheMisses();
}
private AccordCacheEntry<K, V> acquire(K key, boolean onlyIfLoaded)
private AccordCacheEntry<K, V, S> acquire(K key, boolean onlyIfLoadedAndPermitted)
{
AccordCacheEntry<K, V> node = cache.get(key);
AccordCacheEntry<K, V, S> node = cache.get(key);
return node == null
? acquireAbsent(key, onlyIfLoaded)
: acquireExisting(node, onlyIfLoaded);
? acquireAbsent(key, onlyIfLoadedAndPermitted)
: acquireExisting(node, onlyIfLoadedAndPermitted);
}
/*
* Can only return a LOADING Node (or null)
*/
private AccordCacheEntry<K, V> acquireAbsent(K key, boolean onlyIfLoaded)
private AccordCacheEntry<K, V, S> acquireAbsent(K key, boolean onlyIfLoaded)
{
incrementCacheMisses();
if (onlyIfLoaded)
return null;
AccordCacheEntry<K, V> node = adapter.newEntry(key, this);
AccordCacheEntry<K, V, S> node = adapter.newEntry(key, this);
node.increment();
Object prev = cache.put(key, node);
@ -506,7 +510,7 @@ public class AccordCache implements CacheSize
/*
* Can't return EVICTED or INITIALIZED
*/
private AccordCacheEntry<K, V> acquireExisting(AccordCacheEntry<K, V> node, boolean onlyIfLoaded)
private AccordCacheEntry<K, V, S> acquireExisting(AccordCacheEntry<K, V, S> node, boolean onlyIfLoadedAndPermitted)
{
boolean isLoaded = node.isLoaded();
if (isLoaded)
@ -514,8 +518,11 @@ public class AccordCache implements CacheSize
else
incrementCacheMisses();
if (onlyIfLoaded && !isLoaded)
return null;
if (onlyIfLoadedAndPermitted)
{
if (!isLoaded || node.hasFifoOrLocked())
return null;
}
if (node.increment() == 1)
{
@ -527,21 +534,21 @@ public class AccordCache implements CacheSize
return node;
}
public void release(AccordSafeState<K, V> safeRef, AccordTask<?> owner)
public final void release(S safeRef, AccordTask<?> owner)
{
K key = safeRef.global().key();
logger.trace("Releasing resources for {}: {}", key, safeRef);
AccordCacheEntry<K, V> node = cache.get(key);
AccordCacheEntry<K, V, ?> node = cache.get(key);
require(!safeRef.isUnsafe());
require(!safeRef.isReleased());
require(safeRef.global() != null, "safeRef node is null for %s", key);
require(safeRef.global() == node, "safeRef node not in map: %s != %s", safeRef.global(), node);
require(node.references() > 0, "references (%d) are zero for %s (%s)", node.references(), key, node);
require(node.isUnqueued());
boolean evict = false;
if (safeRef.hasUpdate())
if (safeRef.isModified())
{
V update = safeRef.current();
if (update != null)
@ -555,15 +562,12 @@ public class AccordCache implements CacheSize
}
node.notifyListeners(Listener::onUpdate);
}
else if (node.isLoadingOrWaiting())
{
node.loadingOrWaiting().remove(owner);
}
else
{
evict = node.is(LOADED) && node.isNull();
}
safeRef.markUnsafe();
node.remove(owner, safeRef.isSafe());
safeRef.setReleased();
if (node.decrement() == 0)
{
@ -596,9 +600,9 @@ public class AccordCache implements CacheSize
tryShrinkOrEvict = true;
}
AccordCacheEntry<K, ?> remove(K key)
final AccordCacheEntry<K, ?, ?> remove(K key)
{
AccordCacheEntry<K, ?> result = cache.remove(key);
AccordCacheEntry<K, ?, ?> result = cache.remove(key);
if (orderedKeys != null && result != null)
orderedKeys.remove(key);
return result;
@ -609,7 +613,7 @@ public class AccordCache implements CacheSize
return Type.this;
}
public Iterable<K> keysBetween(K start, boolean startInclusive, K end, boolean endInclusive)
public final Iterable<K> keysBetween(K start, boolean startInclusive, K end, boolean endInclusive)
{
if (orderedKeys == null)
orderedKeys = new OrderedKeys<>(adapter.keyComparator(), cache.keySet());
@ -618,15 +622,15 @@ public class AccordCache implements CacheSize
}
@Override
public Iterator<AccordCacheEntry<K, V>> iterator()
public final Iterator<AccordCacheEntry<K, V, S>> iterator()
{
return cache.values().iterator();
}
void validateLoadEvicted(AccordCacheEntry<?, ?> node)
final void validateLoadEvicted(AccordCacheEntry<?, ?, ?> node)
{
@SuppressWarnings("unchecked")
AccordCacheEntry<K, V> state = (AccordCacheEntry<K, V>) node;
AccordCacheEntry<K, V, ?> state = (AccordCacheEntry<K, V, ?>) node;
K key = state.key();
V evicted = state.tryGetFull();
if (evicted == null)
@ -649,46 +653,46 @@ public class AccordCache implements CacheSize
}
@VisibleForTesting
public AccordCacheEntry<K, V> getUnsafe(K key)
public final AccordCacheEntry<K, V, ?> getUnsafe(K key)
{
return cache.get(key);
}
@VisibleForTesting
public boolean isReferenced(K key)
public final boolean isReferenced(K key)
{
AccordCacheEntry<K, V> node = cache.get(key);
AccordCacheEntry<K, V, ?> node = cache.get(key);
return node != null && node.references() > 0;
}
@VisibleForTesting
boolean keyIsReferenced(Object key, Class<? extends AccordSafeState<?, ?>> valClass)
final boolean keyIsReferenced(Object key, Class<? extends AccordSafeState<?, ?, S>> valClass)
{
AccordCacheEntry<?, ?> node = cache.get(key);
AccordCacheEntry<?, ?, ?> node = cache.get(key);
return node != null && node.references() > 0;
}
@VisibleForTesting
boolean keyIsCached(Object key, Class<? extends AccordSafeState<?, ?>> valClass)
final boolean keyIsCached(Object key, Class<? extends AccordSafeState<?, ?, S>> valClass)
{
AccordCacheEntry<?, ?> node = cache.get(key);
AccordCacheEntry<?, ?, ?> node = cache.get(key);
return node != null;
}
@VisibleForTesting
int references(Object key, Class<? extends AccordSafeState<?, ?>> valClass)
final int references(Object key, Class<? extends AccordSafeState<?, ?, S>> valClass)
{
AccordCacheEntry<?, ?> node = cache.get(key);
AccordCacheEntry<?, ?, ?> node = cache.get(key);
return node != null ? node.references() : 0;
}
void notifyListeners(BiConsumer<Listener<K, V>, AccordCacheEntry<K, V>> notify, AccordCacheEntry<K, V> node)
final void notifyListeners(BiConsumer<Listener<K, V>, AccordCacheEntry<K, V, ?>> notify, AccordCacheEntry<K, V, ?> node)
{
notifyListeners(listeners, notify, node);
notifyListeners(typeListeners, notify, node);
}
void notifyListeners(List<Listener<K, V>> listeners, BiConsumer<Listener<K, V>, AccordCacheEntry<K, V>> notify, AccordCacheEntry<K, V> node)
final void notifyListeners(List<Listener<K, V>> listeners, BiConsumer<Listener<K, V>, AccordCacheEntry<K, V, ?>> notify, AccordCacheEntry<K, V, ?> node)
{
if (listeners != null)
{
@ -698,20 +702,20 @@ public class AccordCache implements CacheSize
}
}
public void register(Listener<K, V> l)
public final void register(Listener<K, V> l)
{
if (listeners == null)
listeners = new ArrayList<>();
listeners.add(l);
}
public void unregister(Listener<K, V> l)
public final void unregister(Listener<K, V> l)
{
if (!tryUnregister(l))
throw illegalState("Listener was not registered");
}
public boolean tryUnregister(Listener<K, V> l)
public final boolean tryUnregister(Listener<K, V> l)
{
if (listeners == null || !listeners.remove(l))
return false;
@ -719,9 +723,23 @@ public class AccordCache implements CacheSize
listeners = null;
return true;
}
final boolean isCommandsForKey()
{
return getClass() == KeyInstance.class;
}
}
private final Class<K> keyClass;
// KeyInstance exists to provide us slightly easier discrimination about the Type an AccordCacheEntry is associated with
public final class KeyInstance extends Instance
{
public KeyInstance(AccordCommandStore commandStore)
{
super(commandStore);
}
}
private final Class<K> keyClass; // type of key, useful primarily for toString(), but also piggyback for deciding Instance type
private Adapter<K, V, S> adapter;
private long bytesCached;
private int size;
@ -734,6 +752,8 @@ public class AccordCache implements CacheSize
public Type(Class<K> keyClass, Adapter<K, V, S> adapter, AccordCacheMetrics.Shard metrics)
{
// Integer and String permitted for testing, but the Invariant exists only to enforce that we construct the right kind of Instance
Invariants.require(keyClass == RoutingKey.class || keyClass == TxnId.class || keyClass == String.class || keyClass == Integer.class);
this.keyClass = keyClass;
this.adapter = adapter;
this.objectSize = metrics.objectSize;
@ -751,7 +771,7 @@ public class AccordCache implements CacheSize
// can be safely garbage collected if empty
Instance newInstance(AccordCommandStore commandStore)
{
return new Instance(commandStore);
return keyClass == RoutingKey.class ? new KeyInstance(commandStore) : new Instance(commandStore);
}
private void incrementCacheHits()
@ -867,17 +887,17 @@ public class AccordCache implements CacheSize
}
@VisibleForTesting
AccordCacheEntry<?, ?> head()
AccordCacheEntry<?, ?, ?> head()
{
Iterator<AccordCacheEntry<?, ?>> iter = evictQueue.iterator();
Iterator<AccordCacheEntry<?, ?, ?>> iter = evictQueue.iterator();
return iter.hasNext() ? iter.next() : null;
}
@VisibleForTesting
AccordCacheEntry<?, ?> tail()
AccordCacheEntry<?, ?, ?> tail()
{
AccordCacheEntry<?,?> last = null;
Iterator<AccordCacheEntry<?, ?>> iter = evictQueue.iterator();
AccordCacheEntry<?,?, ?> last = null;
Iterator<AccordCacheEntry<?, ?, ?>> iter = evictQueue.iterator();
while (iter.hasNext())
last = iter.next();
return last;
@ -888,7 +908,7 @@ public class AccordCache implements CacheSize
return size() == 0;
}
Iterable<AccordCacheEntry<?, ?>> evictionQueue()
Iterable<AccordCacheEntry<?, ?, ?>> evictionQueue()
{
return evictQueue::iterator;
}
@ -937,10 +957,10 @@ public class AccordCache implements CacheSize
return;
type.register(new AccordCache.Listener<>() {
private final IdentityHashMap<AccordCacheEntry<?, ?>, CacheEvents.Evict> pendingEvicts = new IdentityHashMap<>();
private final IdentityHashMap<AccordCacheEntry<?, ?, ?>, CacheEvents.Evict> pendingEvicts = new IdentityHashMap<>();
@Override
public void onAdd(AccordCacheEntry<K, V> state)
public void onAdd(AccordCacheEntry<K, V, ?> state)
{
CacheEvents.Add add = new CacheEvents.Add();
CacheEvents.Evict evict = new CacheEvents.Evict();
@ -957,7 +977,7 @@ public class AccordCache implements CacheSize
}
@Override
public void onEvict(AccordCacheEntry<K, V> state)
public void onEvict(AccordCacheEntry<K, V, ?> state)
{
CacheEvents.Evict event = pendingEvicts.remove(state);
if (event == null) return;
@ -967,7 +987,7 @@ public class AccordCache implements CacheSize
});
}
private static void updateMutable(AccordCache.Type<?, ?, ?> type, AccordCacheEntry<?, ?> state, CacheEvents event)
private static void updateMutable(AccordCache.Type<?, ?, ?> type, AccordCacheEntry<?, ?, ?> state, CacheEvents event)
{
event.status = state.status().name();
@ -987,7 +1007,7 @@ public class AccordCache implements CacheSize
event.update();
}
static class FunctionalAdapter<K, V, S> implements Adapter<K, V, S>
static class FunctionalAdapter<K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> implements Adapter<K, V, S>
{
final BiFunction<AccordCommandStore, K, V> load;
final QuadFunction<AccordCommandStore, K, V, Object, Runnable> save;
@ -997,8 +1017,8 @@ public class AccordCache implements CacheSize
final TriFunction<AccordCommandStore, K, V, Boolean> validate;
final ToLongFunction<V> estimateHeapSize;
final ToLongFunction<Object> estimateShrunkHeapSize;
final Function<AccordCacheEntry<K, V>, S> newSafeRef;
final BiFunction<K, AccordCache.Type<K, V, ?>.Instance, AccordCacheEntry<K, V>> newNode;
final Function<AccordCacheEntry<K, V, S>, S> newSafeRef;
final BiFunction<K, AccordCache.Type<K, V, S>.Instance, AccordCacheEntry<K, V, S>> newNode;
FunctionalAdapter(BiFunction<AccordCommandStore, K, V> load,
QuadFunction<AccordCommandStore, K, V, Object, Runnable> save,
@ -1007,8 +1027,8 @@ public class AccordCache implements CacheSize
TriFunction<AccordCommandStore, K, V, Boolean> validate,
ToLongFunction<V> estimateHeapSize,
ToLongFunction<Object> estimateShrunkHeapSize,
Function<AccordCacheEntry<K, V>, S> newSafeRef,
BiFunction<K, Type<K, V, ?>.Instance, AccordCacheEntry<K, V>> newNode)
Function<AccordCacheEntry<K, V, S>, S> newSafeRef,
BiFunction<K, Type<K, V, S>.Instance, AccordCacheEntry<K, V, S>> newNode)
{
this.load = load;
this.save = save;
@ -1082,13 +1102,13 @@ public class AccordCache implements CacheSize
}
@Override
public S safeRef(AccordCacheEntry<K, V> node)
public S safeRef(AccordCacheEntry<K, V, S> node)
{
return newSafeRef.apply(node);
}
@Override
public AccordCacheEntry<K, V> newEntry(K key, Type<K, V, ?>.Instance owner)
public AccordCacheEntry<K, V, S> newEntry(K key, Type<K, V, S>.Instance owner)
{
return newNode.apply(key, owner);
}
@ -1100,7 +1120,7 @@ public class AccordCache implements CacheSize
}
}
static class SettableWrapper<K, V, S> extends FunctionalAdapter<K, V, S>
static class SettableWrapper<K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> extends FunctionalAdapter<K, V, S>
{
volatile BiFunction<AccordCommandStore, K, V> load;
@ -1110,9 +1130,9 @@ public class AccordCache implements CacheSize
this.load = super.load;
}
public static <K, V> Adapter<K, V, ?> loadOnly(BiFunction<AccordCommandStore, K, V> load)
public static <K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> Adapter<K, V, S> loadOnly(BiFunction<AccordCommandStore, K, V> load)
{
SettableWrapper<K, V, ?> result = new SettableWrapper<>(new NoOpAdapter<>());
SettableWrapper<K, V, S> result = new SettableWrapper<>(new NoOpAdapter<K, V, S>());
result.load = load;
return result;
}
@ -1124,7 +1144,7 @@ public class AccordCache implements CacheSize
}
}
static class NoOpAdapter<K, V, S> implements Adapter<K, V, S>
static class NoOpAdapter<K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> implements Adapter<K, V, S>
{
@Override public V load(AccordCommandStore commandStore, K key) { return null; }
@Override public Runnable save(AccordCommandStore commandStore, K key, @Nullable V value, @Nullable Object shrunk) { return null; }
@ -1135,7 +1155,7 @@ public class AccordCache implements CacheSize
@Override public long estimateHeapSize(V value) { return 0; }
@Override public long estimateShrunkHeapSize(Object shrunk) { return 0; }
@Override public boolean validate(AccordCommandStore commandStore, K key, V value) { return false; }
@Override public S safeRef(AccordCacheEntry<K, V> node) { return null; }
@Override public S safeRef(AccordCacheEntry<K, V, S> node) { return null; }
}
public static class CommandsForKeyAdapter implements Adapter<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>
@ -1239,7 +1259,7 @@ public class AccordCache implements CacheSize
}
@Override
public AccordSafeCommandsForKey safeRef(AccordCacheEntry<RoutingKey, CommandsForKey> node)
public AccordSafeCommandsForKey safeRef(AccordCacheEntry<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> node)
{
return new AccordSafeCommandsForKey(node);
}
@ -1251,7 +1271,7 @@ public class AccordCache implements CacheSize
}
@Override
public AccordCacheEntry<RoutingKey, CommandsForKey> newEntry(RoutingKey key, Type<RoutingKey, CommandsForKey, ?>.Instance owner)
public AccordCacheEntry<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> newEntry(RoutingKey key, Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance owner)
{
CommandsForKeyCacheEntry entry = new CommandsForKeyCacheEntry(key, owner);
entry.readyToLoad();
@ -1376,19 +1396,19 @@ public class AccordCache implements CacheSize
}
@Override
public AccordSafeCommand safeRef(AccordCacheEntry<TxnId, Command> node)
public AccordSafeCommand safeRef(AccordCacheEntry<TxnId, Command, AccordSafeCommand> node)
{
return new AccordSafeCommand(node);
}
@Override
public AccordCacheEntry<TxnId, Command> newEntry(TxnId txnId, Type<TxnId, Command, ?>.Instance owner)
public AccordCacheEntry<TxnId, Command, AccordSafeCommand> newEntry(TxnId txnId, Type<TxnId, Command, AccordSafeCommand>.Instance owner)
{
AccordCacheEntry<TxnId, Command> node = new AccordCacheEntry<>(txnId, owner);
AccordCacheEntry<TxnId, Command, AccordSafeCommand> node = new AccordCacheEntry<>(txnId, owner);
if (txnId.is(Txn.Kind.EphemeralRead))
{
node.initialize(null);
int maxAge = (int)Math.min(0xff, 1 + DatabaseDescriptor.getReadRpcTimeout(TimeUnit.SECONDS));
int maxAge = (int)Math.min(AGE_MASK, 1 + DatabaseDescriptor.getReadRpcTimeout(TimeUnit.SECONDS));
node.markNoEvict(owner.parent().parent().noEvictGeneration, maxAge);
}
else

1149
src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java
View File

File diff suppressed because it is too large Load Diff

39
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -56,7 +56,6 @@ import accord.impl.progresslog.TxnState;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.CommandSummaries;
import accord.local.ExecutionContext;
import accord.local.ExecutionContext.Empty;
import accord.local.MaxConflicts;
@ -178,13 +177,16 @@ public class AccordCommandStore extends CommandStore
@Override
public AccordSafeCommand acquireIfLoaded(TxnId txnId)
{
return commands().acquireIfLoaded(txnId);
// note: we must return false if the entry is locked to enforce ordering.
// note importantly that this is also coupled to the safety of synchronously releasing ExclusiveExecutor.owner,
// rather than waiting until the (potentially asynchronous) cleanup of the task completes
return commands().acquireIfLoadedAndPermitted(txnId);
}
@Override
public AccordSafeCommandsForKey acquireIfLoaded(RoutingKey key)
{
return commandsForKeys().acquireIfLoaded(key);
return commandsForKeys().acquireIfLoadedAndPermitted(key);
}
@Override
@ -299,7 +301,7 @@ public class AccordCommandStore extends CommandStore
void tryPreSetup(AccordTask<?> task)
{
if (inStore() && current != null)
task.presetup(current.task);
task.preSetup(current.task);
}
public final TableId tableId()
@ -432,13 +434,19 @@ public class AccordCommandStore extends CommandStore
return taskExecutor().tryExecuteImmediately(run);
}
public AccordSafeCommandStore begin(AccordTask<?> operation, @Nullable CommandSummaries commandsForRanges)
public AccordSafeCommandStore begin(AccordSafeCommandStore safeStore)
{
require(current == null);
current = AccordSafeCommandStore.create(operation, commandsForRanges, this);
current = safeStore;
return current;
}
public void complete(AccordSafeCommandStore store)
{
require(current == store);
current = null;
}
public boolean hasSafeStore()
{
return current != null;
@ -454,19 +462,6 @@ public class AccordCommandStore extends CommandStore
return progressLog;
}
public void complete(AccordSafeCommandStore store)
{
require(current == store);
current.postExecute();
current = null;
}
public void abort(AccordSafeCommandStore store)
{
Invariants.require(store == current);
current = null;
}
@Override
public void shutdown()
{
@ -650,7 +645,7 @@ public class AccordCommandStore extends CommandStore
public Ready() { super(1); }
@Override public void run() { decrement(); }
void maybeFlush(ExclusiveCaches caches, AccordCacheEntry<RoutingKey, CommandsForKey> e)
void maybeFlush(ExclusiveCaches caches, AccordCacheEntry<RoutingKey, CommandsForKey, ?> e)
{
if (e.isModified())
{
@ -665,7 +660,7 @@ public class AccordCommandStore extends CommandStore
{
if (ranges == null)
{
for (AccordCacheEntry<RoutingKey, CommandsForKey> e : caches.commandsForKeys())
for (AccordCacheEntry<RoutingKey, CommandsForKey, ?> e : caches.commandsForKeys())
ready.maybeFlush(caches, e);
}
else
@ -717,7 +712,7 @@ public class AccordCommandStore extends CommandStore
if (!maybeShouldReplay(txnId))
return AsyncChains.success(null);
return commandStore.chain(ExecutionContext.contextFor(txnId, "Replay"), safeStore -> {
return commandStore.chain(ExecutionContext.unsequenced(txnId, "Replay"), safeStore -> {
Replay replay = shouldReplay(txnId, safeStore.unsafeGet(txnId).current().participants());
if (replay == Replay.NONE)
return null;

34
src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
View File

@ -55,7 +55,6 @@ import org.apache.cassandra.journal.Descriptor;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.AccordCommandStore.DurablyAppliedTo;
import org.apache.cassandra.service.accord.AccordExecutor.AccordExecutorFactory;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.AccordConfig.QueueShardModel.THREAD_PER_SHARD;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccord;
@ -73,7 +72,6 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
private final int mask;
private long cacheSize, workingSetSize;
private int maxQueuedLoads, maxQueuedRangeLoads;
private boolean shrinkingOn;
AccordCommandStores(NodeCommandStoreService node, Agent agent, DataStore store, RandomSource random,
@ -87,12 +85,9 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
cacheSize = DatabaseDescriptor.getAccordCacheSizeInMiB() << 20;
workingSetSize = DatabaseDescriptor.getAccordWorkingSetSizeInMiB() << 20;
AccordConfig config = DatabaseDescriptor.getAccord();
maxQueuedLoads = maxQueuedLoads(config);
maxQueuedRangeLoads = maxQueuedRangeLoads(config);
shrinkingOn = DatabaseDescriptor.getAccordCacheShrinkingOn();
refreshCapacities();
ScheduledExecutors.scheduledFastTasks.scheduleWithFixedDelay(() -> {
ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(() -> {
for (AccordExecutor executor : executors)
{
executor.executeDirectlyWithLock(() -> {
@ -173,13 +168,6 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
refreshCapacities();
}
public synchronized void setMaxQueuedLoads(int total, int range)
{
maxQueuedLoads = total;
maxQueuedRangeLoads = range;
refreshCapacities();
}
public synchronized void setShrinking(boolean on)
{
shrinkingOn = on;
@ -213,14 +201,11 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
{
long capacityPerExecutor = cacheSize / executors.length;
long workingSetPerExecutor = workingSetSize < 0 ? Long.MAX_VALUE : workingSetSize / executors.length;
int maxLoadsPerExecutor = Math.max(1, (maxQueuedLoads + executors.length - 1) / executors.length);
int maxRangeLoadsPerExecutor = Math.max(1, (maxQueuedRangeLoads + executors.length - 1) / executors.length);
for (AccordExecutor executor : executors)
{
executor.executeDirectlyWithLock(() -> {
executor.setCapacity(capacityPerExecutor);
executor.setWorkingSetSize(workingSetPerExecutor);
executor.setMaxQueuedLoads(maxLoadsPerExecutor, maxRangeLoadsPerExecutor);
executor.cacheExclusive().setShrinkingOn(shrinkingOn);
});
}
@ -341,21 +326,4 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
return Math.max(1, config.queue_shard_count.or(DatabaseDescriptor.getAvailableProcessors() / 8));
}
}
private static int threads(AccordConfig config)
{
return config.queue_thread_count.or(2 * FBUtilities.getAvailableProcessors());
}
public static int maxQueuedLoads(AccordConfig config)
{
return config.max_queued_loads.or(FBUtilities.getAvailableProcessors());
}
public static int maxQueuedRangeLoads(AccordConfig config)
{
return config.max_queued_range_loads.or(maxQueuedLoads(config) / 4);
}
}

1207
src/java/org/apache/cassandra/service/accord/AccordExecutor.java
View File

File diff suppressed because it is too large Load Diff

19
src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLockLoop.java
View File

@ -26,7 +26,6 @@ import accord.utils.QuintConsumer;
import org.apache.cassandra.concurrent.CassandraThread;
import org.apache.cassandra.service.accord.AccordExecutorLoops.LoopTask;
import org.apache.cassandra.service.accord.debug.DebugExecution.DebugExecutorLoop;
import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITH_LOCK;
import static org.apache.cassandra.service.accord.debug.DebugExecution.DEBUG_EXECUTION;
@ -214,18 +213,21 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
if (task != null)
{
self.setAccordActiveTask(task);
boolean executed = false;
try
{
task.preRunExclusive();
executed = true;
task.run();
}
catch (Throwable t)
{
task.fail(t);
executed = false;
task.failExecution(t);
}
finally
{
completeTaskExclusive(task);
cleanupTaskExclusive(task, executed);
self.setAccordActiveTask(null);
}
}
@ -265,7 +267,6 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
{
return new LoopTask(name)
{
final DebugExecutorLoop debug = DEBUG_EXECUTION ? new DebugExecutorLoop(AccordExecutorAbstractLockLoop.this.debug) : null;
@Override
public void run()
{
@ -276,17 +277,15 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
Task task = null;
while (true)
{
if (DEBUG_EXECUTION) debug.onLock();
lock(self);
try
{
if (DEBUG_EXECUTION) debug.onEnterLock();
enterLockLoop();
if (task != null)
{
Task tmp = task;
task = null;
completeTaskExclusive(tmp);
cleanupTaskExclusive(tmp, true);
self.setAccordActiveTask(null);
}
@ -322,9 +321,9 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
{
if (task != null)
{
try { task.fail(t); }
try { task.failExecution(t); }
catch (Throwable t2) { t.addSuppressed(t2); }
try { completeTaskExclusive(task); }
try { cleanupTaskExclusive(task, false); }
catch (Throwable t2) { t.addSuppressed(t2); }
try { agent.onException(t); }
catch (Throwable t2) { /* nothing we can sensibly do after already reporting */ }
@ -350,7 +349,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutorAbstractLoop
}
catch (Throwable t)
{
try { task.fail(t); }
try { task.failExecution(t); }
catch (Throwable t2)
{
try

21
src/java/org/apache/cassandra/service/accord/AccordExecutorAbstractLoop.java
View File

@ -44,11 +44,6 @@ abstract class AccordExecutorAbstractLoop extends AccordExecutor
return unqueued != null;
}
Task unqueued()
{
return unqueued;
}
final Task push(Task submit)
{
Invariants.require(submit.next == null);
@ -89,26 +84,16 @@ abstract class AccordExecutorAbstractLoop extends AccordExecutor
Invariants.require(cur != null);
Task next = cur.next;
cur.next = null;
if (cur.isReadyToCleanup()) completeTaskExclusive(cur);
else cur.submitExclusive(this);
if (cur.is(Task.State.UNINITIALIZED)) cur.submitExclusive(this);
else cleanupTaskExclusive(cur, true);
return next;
}
final Task enqueueOneCleanup(Task cur)
final Task destructiveNext(Task cur)
{
Invariants.require(cur != null);
Task next = cur.next;
cur.next = null;
completeTaskExclusive(cur);
return next;
}
final Task enqueueOneSubmit(Task cur)
{
Invariants.require(cur != null);
Task next = cur.next;
cur.next = null;
cur.submitExclusive(this);
return next;
}

119
src/java/org/apache/cassandra/service/accord/AccordExecutorSignalLoop.java
View File

@ -33,12 +33,14 @@ import org.apache.cassandra.service.accord.debug.DebugExecution.DebugTask;
import org.apache.cassandra.utils.concurrent.SignalLock;
import static accord.utils.Invariants.nonNull;
import static org.apache.cassandra.service.accord.AccordExecutor.Task.State.UNINITIALIZED;
import static org.apache.cassandra.service.accord.debug.DebugExecution.DEBUG_EXECUTION;
public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
{
private static class ShutdownException extends RuntimeException {}
private static final int MAX_LOOPS = 1000; // limit the amount of time we hold the lock for
private final SignalLock lock;
private final AccordExecutorLoops loops;
private int readyToRunTarget = 1;
@ -46,8 +48,7 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
// TODO (desired): intrusive queue using Task.next, but a little challenging because we reuse SequentialQueueTask so have ABA problem
private final ConcurrentLinkedQueue<Task> readyToRun = new ConcurrentLinkedQueue<>();
private Task pendingSequentialHead, pendingSequentialTail;
private Task pendingCleanupHead, pendingCleanupTail;
private Task pendingExecutedHead, pendingExecutedTail;
private Task pendingNewHead, pendingNewTail;
private int pendingCount;
@ -105,14 +106,17 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
{
updatePendingUnqueued();
Task requeue = null, requeueTail = null;
Task submit = null, submitTail = null;
Task cur = pendingNewHead;
while (cur != null)
{
Task next = cur.next;
cur.next = null;
if (cur.isNewWork())
{
cur.next = null;
cur.submitExclusive(this);
if (submit == null) submit = cur;
else submitTail.next = cur;
submitTail = cur;
--pendingCount;
}
else
@ -126,6 +130,13 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
pendingNewHead = requeue;
pendingNewTail = requeueTail;
while (submit != null)
{
Task next = submit.next;
submit.next = null;
submit.submitExclusive(this);
submit = next;
}
}
private boolean enqueueOnePending()
@ -133,25 +144,23 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
if (pendingCount == 0)
return false;
if (pendingSequentialHead != null)
--pendingCount;
if (pendingExecutedHead != null)
{
pendingSequentialHead = enqueueOneCleanup(pendingSequentialHead);
if (pendingSequentialHead == null)
pendingSequentialTail = null;
}
else if (pendingCleanupHead != null)
{
pendingCleanupHead = enqueueOneCleanup(pendingCleanupHead);
if (pendingCleanupHead == null)
pendingCleanupTail = null;
Task executed = pendingExecutedHead;
pendingExecutedHead = destructiveNext(executed);
if (pendingExecutedHead == null)
pendingExecutedTail = null;
cleanupTaskExclusive(executed, true);
}
else
{
pendingNewHead = enqueueOneSubmit(pendingNewHead);
Task submit = pendingNewHead;
pendingNewHead = destructiveNext(submit);
if (pendingNewHead == null)
pendingNewTail = null;
submit.submitExclusive(this);
}
--pendingCount;
return true;
}
@ -178,22 +187,29 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
}
boolean hasDrainedSignal = false;
int loops = 0;
while (true)
{
long state = lock.state();
int signals = SignalLock.asyncSignalCount(state);
int waiters = SignalLock.waitingEnabledThreadCount(state);
if (signals >= readyToRunTarget)
if (signals > 0)
{
if (enqueueOnePending() || (updatePendingUnqueued() && enqueueOnePending())) continue;
else if (hasDrainedSignal)
lock.signalLockWorkExclusive();
return;
}
else if (waiters > 0 && signals > 1 && SignalLock.activeEnabledThreadCount(state) == 1)
{
// ensure at least one other thread is running if there's enough work for it; it will spin up other threads if necessary
lock.propagateAsyncWorkSignals(1);
if (++loops > MAX_LOOPS)
return;
if (signals >= readyToRunTarget)
{
if (enqueueOnePending() || (updatePendingUnqueued() && enqueueOnePending())) continue;
else if (hasDrainedSignal)
lock.signalLockWorkExclusive();
return;
}
else if (waiters > 0 && signals > 1 && SignalLock.activeEnabledThreadCount(state) == 1)
{
// ensure at least one other thread is running if there's enough work for it; it will spin up other threads if necessary
lock.propagateAsyncWorkSignals(1);
}
}
Task task = pollAlreadyWaitingToRunExclusive();
@ -216,9 +232,9 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
try { task.preRunExclusive(); }
catch (Throwable t)
{
try { task.fail(t); }
try { task.failExclusive(t, Task.State.FAILED_OTHER); }
catch (Throwable t2) { try { t.addSuppressed(t2); } catch (Throwable t3) {} }
try { completeTaskExclusive(task); }
try { cleanupTaskExclusive(task, false); }
catch (Throwable t2) { try { t.addSuppressed(t2); } catch (Throwable t3) {} }
continue;
}
@ -237,28 +253,22 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
return false;
int count = 0;
Task addSequentialHead = null, addSequentialTail = null;
Task addCleanupHead = null, addCleanupTail = null;
Task addExecutedHead = null, addExecutedTail = null;
Task addNewHead = null, addNewTail = null;
{
Task cur = Task.reverse(acquireUnqueuedExclusive());
while (cur != null)
{
Task next = cur.next;
if (!cur.isReadyToCleanup())
if (cur.is(UNINITIALIZED))
{
if (addNewHead == null) addNewHead = addNewTail = setNextNull(cur);
else addNewHead = reverseOne(addNewHead, cur);
}
else if (cur instanceof ExclusiveExecutorTask)
{
if (addSequentialHead == null) addSequentialHead = addSequentialTail = setNextNull(cur);
else addSequentialHead = reverseOne(addSequentialHead, cur);
}
else
{
if (addCleanupHead == null) addCleanupHead = addCleanupTail = setNextNull(cur);
else addCleanupHead = reverseOne(addCleanupHead, cur);
if (addExecutedHead == null) addExecutedHead = addExecutedTail = setNextNull(cur);
else addExecutedHead = reverseOne(addExecutedHead, cur);
}
++count;
cur = next;
@ -266,17 +276,11 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
}
pendingCount += count;
if (addSequentialHead != null)
if (addExecutedHead != null)
{
if (pendingSequentialHead == null) pendingSequentialHead = addSequentialHead;
else pendingSequentialTail.next = addSequentialHead;
pendingSequentialTail = addSequentialTail;
}
if (addCleanupHead != null)
{
if (pendingCleanupHead == null) pendingCleanupHead = addCleanupHead;
else pendingCleanupTail.next = addCleanupHead;
pendingCleanupTail = addCleanupTail;
if (pendingExecutedHead == null) pendingExecutedHead = addExecutedHead;
else pendingExecutedTail.next = addExecutedHead;
pendingExecutedTail = addExecutedTail;
}
if (addNewHead != null)
{
@ -349,22 +353,22 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
}
}
private Task cleanupAndMaybeGetWork(AccordTaskRunner self, @Nullable Task cleanup)
private Task executedAndMaybeGetWork(AccordTaskRunner self, @Nullable Task executed)
{
if (lock.tryAcquireAsyncWork())
{
if (shutdown)
throw new ShutdownException();
return pushCleanupAndReturn(cleanup, nonNull(pollReadyToRun()));
return pushExecutedAndReturn(executed, nonNull(pollReadyToRun()));
}
if (!tryLock(self))
return pushCleanupAndReturn(cleanup, null);
return pushExecutedAndReturn(executed, null);
try
{
completeTaskExclusive(cleanup);
cleanupTaskExclusive(executed, true);
fetchWorkExclusive();
}
catch (Throwable t)
@ -383,10 +387,9 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
return null;
}
private Task pushCleanupAndReturn(Task cleanup, Task result)
private Task pushExecutedAndReturn(Task complete, Task result)
{
cleanup.setReadyToCleanup();
if (push(cleanup) == null)
if (push(complete) == null)
lock.signalLockWork();
return result;
}
@ -401,7 +404,7 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
final boolean unlockAndAcquire(AccordTaskRunner self)
{
self.clearAccordLockedExecutor();
self.exitAccordLockedExecutor();
if (DEBUG_EXECUTION) debug.onExitLock();
return lock.unlockAndAcquireAsyncWork();
}
@ -422,7 +425,7 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
{
if (task != null)
{
try { task = cleanupAndMaybeGetWork(self, task); }
try { task = executedAndMaybeGetWork(self, task); }
catch (Throwable t) { task = null; throw t; }
}
if (task == null)
@ -435,7 +438,7 @@ public class AccordExecutorSignalLoop extends AccordExecutorAbstractLoop
}
catch (Throwable t)
{
try { task.fail(t); }
try { task.failExecution(t); }
catch (Throwable t2)
{
try

22
src/java/org/apache/cassandra/service/accord/AccordExecutorSimple.java
View File

@ -82,7 +82,8 @@ class AccordExecutorSimple extends AccordExecutor
protected void run()
{
Thread self = Thread.currentThread();
AccordTaskRunner self = AccordTaskRunner.get();
self.setAccordActiveExecutor(AccordExecutorSimple.this);
lock.lock();
try
{
@ -95,11 +96,24 @@ class AccordExecutorSimple extends AccordExecutor
return;
}
try { task.preRunExclusive(); task.run(); }
catch (Throwable t) { task.fail(t); }
// TODO (expected): dedup with AbstractLockLoop, and cleanup (executed flag is a bit ugly)
self.setAccordActiveTask(task);
boolean executed = false;
try
{
task.preRunExclusive();
executed = true;
task.run();
}
catch (Throwable t)
{
executed = false;
task.failExecution(t);
}
finally
{
completeTaskExclusive(task);
cleanupTaskExclusive(task, executed);
self.setAccordActiveTask(null);
}
}
}

91
src/java/org/apache/cassandra/service/accord/AccordSafeCommand.java
View File

@ -20,51 +20,22 @@ package org.apache.cassandra.service.accord;
import java.util.Objects;
import com.google.common.annotations.VisibleForTesting;
import accord.api.Journal;
import accord.local.Command;
import accord.local.SafeCommand;
import accord.primitives.TxnId;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.service.accord.AccordCacheEntry.LockMode;
public class AccordSafeCommand extends SafeCommand implements AccordSafeState<TxnId, Command>
public class AccordSafeCommand extends SafeCommand implements AccordSafeState<TxnId, Command, AccordSafeCommand>
{
public static class DebugAccordSafeCommand extends AccordSafeCommand
{
final Ref<?> selfRef;
public DebugAccordSafeCommand(AccordCacheEntry<TxnId, Command> global)
{
super(global);
selfRef = new Ref<>(this, null);
selfRef.debug(global.key().toString());
}
@Override
public void markUnsafe()
{
super.markUnsafe();
selfRef.release();
}
public static void trace(AccordSafeCommand safeCommand, String message)
{
((DebugAccordSafeCommand)safeCommand).selfRef.debug(message);
}
}
private boolean unsafe;
private final AccordCacheEntry<TxnId, Command> global;
private final AccordCacheEntry<TxnId, Command, AccordSafeCommand> global;
private Command original;
private Command current;
public AccordSafeCommand(AccordCacheEntry<TxnId, Command> global)
public AccordSafeCommand(AccordCacheEntry<TxnId, Command, AccordSafeCommand> global)
{
super(global.key());
this.global = global;
this.original = null;
this.current = null;
}
@Override
@ -73,7 +44,7 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState<Tx
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
AccordSafeCommand that = (AccordSafeCommand) o;
return Objects.equals(original, that.original) && Objects.equals(current, that.current);
return Objects.equals(this.original, that.original) && Objects.equals(this.current(), that.current());
}
@Override
@ -86,66 +57,36 @@ public class AccordSafeCommand extends SafeCommand implements AccordSafeState<Tx
public String toString()
{
return "AccordSafeCommand{" +
"invalidated=" + unsafe +
"status=" + statusString() +
", global=" + global +
", original=" + original +
", current=" + current +
'}';
}
@Override
public AccordCacheEntry<TxnId, Command> global()
public AccordCacheEntry<TxnId, Command, AccordSafeCommand> global()
{
checkNotInvalidated();
return global;
}
@Override
public Command current()
public void postExecute(AccordTask<?> owner)
{
checkNotInvalidated();
return current;
}
@Override
@VisibleForTesting
public void set(Command command)
{
checkNotInvalidated();
this.current = command;
}
@Override
public Command original()
{
checkNotInvalidated();
return original;
global.releaseExclusive(this, owner);
}
public Journal.CommandUpdate update()
{
return new Journal.CommandUpdate(original, current);
return new Journal.CommandUpdate(original, current());
}
@Override
public void preExecute()
public void preExecute(AccordTask<?> owner, LockMode lockMode)
{
checkNotInvalidated();
original = global.getExclusive();
requireUninitialised();
original = global.lockExclusive(owner, lockMode);
current = original;
if (isUnset())
uninitialised();
}
@Override
public void markUnsafe()
{
unsafe = true;
}
@Override
public boolean isUnsafe()
{
return unsafe;
if (current == null)
initialise();
setSafe();
}
}

140
src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
View File

@ -18,13 +18,8 @@
package org.apache.cassandra.service.accord;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import accord.api.Agent;
@ -35,16 +30,20 @@ import accord.api.RoutingKey;
import accord.impl.AbstractSafeCommandStore;
import accord.local.Command;
import accord.local.CommandStores;
import accord.local.CommandSummaries;
import accord.local.ExecutionContext;
import accord.local.NodeCommandStoreService;
import accord.local.RedundantBefore;
import accord.local.SafeState;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.SafeCommandsForKey;
import accord.primitives.AbstractUnseekableKeys;
import accord.primitives.Routables;
import accord.primitives.Timestamp;
import accord.primitives.Txn.Kind.Kinds;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import org.apache.cassandra.metrics.LogLinearDecayingHistograms;
import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches;
@ -53,69 +52,53 @@ import org.apache.cassandra.service.accord.AccordDurableOnFlush.ReportDurable;
import org.apache.cassandra.service.paxos.PaxosState;
import static accord.utils.Invariants.illegalState;
import static org.apache.cassandra.service.accord.AccordCacheEntry.LockMode.UNQUEUED;
public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeCommand, AccordSafeCommandsForKey, AccordCommandStore.ExclusiveCaches>
public final class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeCommand, AccordSafeCommandsForKey, AccordCommandStore.ExclusiveCaches>
{
final AccordTask<?> task;
private final @Nullable CommandSummaries commandsForRanges;
private final AccordCommandStore commandStore;
private AccordSafeCommandStore(AccordTask<?> task,
@Nullable CommandSummaries commandsForRanges,
AccordCommandStore commandStore)
AccordSafeCommandStore(AccordTask<?> task, ExecutionContext context)
{
super(task.executionContext(), commandStore);
super(context);
this.task = task;
this.commandsForRanges = commandsForRanges;
this.commandStore = commandStore;
}
@Override
public CommandStores.RangesForEpoch ranges()
{
CommandStores.RangesForEpoch ranges = super.ranges();
if (ranges != null)
return ranges;
return commandStore.unsafeGetRangesForEpoch();
}
public static AccordSafeCommandStore create(AccordTask<?> task,
@Nullable CommandSummaries commandsForRanges,
AccordCommandStore commandStore)
{
return new AccordSafeCommandStore(task, commandsForRanges, commandStore);
}
@VisibleForTesting
public Set<RoutingKey> commandsForKeysKeys()
public Iterable<SafeCommandsForKey> safeCommandsForKeys()
{
if (task.commandsForKey() == null)
return Collections.emptySet();
return task.commandsForKey().keySet();
return task.refs.values().stream()
.filter(v -> v instanceof SafeCommandsForKey)
.map(v -> (SafeCommandsForKey)v)::iterator;
}
@Override
protected AccordSafeCommand getInternal(TxnId txnId)
{
Map<TxnId, AccordSafeCommand> commands = task.commands();
if (commands == null)
return null;
return commands.get(txnId);
return (AccordSafeCommand) task.refs.get(txnId);
}
@Override
protected ExclusiveCaches tryGetCaches()
{
return commandStore.tryLockCaches();
if (task.isIncremental())
{
// We could relax this, but requires careful consideration of semantics when:
// - touching keys we have scheduled to process later
// - touching txnIds we may have to logically lock until the whole processing completes to
// ensure persistent state machine updates are consistent with incremental updates to cache
return null;
}
return commandStore().tryLockCaches();
}
protected AccordSafeCommand add(AccordSafeCommand safeCommand, ExclusiveCaches caches)
{
Object check = task.ensureCommands().putIfAbsent(safeCommand.txnId(), safeCommand);
Object check = task.refs.putIfAbsent(safeCommand.txnId(), safeCommand);
if (check == null)
{
safeCommand.preExecute();
Invariants.require(!task.isIncremental()); // if isIncremental we'll have to supply holdQueue==true, but for now we forbid it
safeCommand.preExecute(task, UNQUEUED);
return safeCommand;
}
else
@ -131,7 +114,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
// Field persistence is handled by AccordTask
}
protected void persistFieldUpdatesInternal(Runnable onDone)
void persistFieldUpdatesInternal(Runnable onDone)
{
FieldUpdates updates = fieldUpdates();
if (updates == null)
@ -142,7 +125,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
long ticket = AccordCommandStore.nextSafeRedundantBeforeTicket.incrementAndGet();
SafeRedundantBefore update = new SafeRedundantBefore(ticket, updates.newRedundantBefore);
Runnable reportRedundantBefore = () -> {
AccordCommandStore.safeRedundantBeforeUpdater.accumulateAndGet(commandStore, update, SafeRedundantBefore::max);
AccordCommandStore.safeRedundantBeforeUpdater.accumulateAndGet(commandStore(), update, SafeRedundantBefore::max);
};
Runnable prevOnDone = onDone;
onDone = prevOnDone == null ? reportRedundantBefore : () -> {
@ -150,15 +133,15 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
finally { prevOnDone.run(); }
};
}
commandStore.persistFieldUpdates(updates, onDone);
commandStore().persistFieldUpdates(updates, onDone);
}
protected AccordSafeCommandsForKey add(AccordSafeCommandsForKey safeCfk, ExclusiveCaches caches)
{
Object check = task.ensureCommandsForKey().putIfAbsent(safeCfk.key(), safeCfk);
Object check = task.refs.putIfAbsent(safeCfk.key(), safeCfk);
if (check == null)
{
safeCfk.preExecute();
safeCfk.preExecute(task, UNQUEUED);
return safeCfk;
}
else
@ -171,17 +154,17 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
@Override
protected AccordSafeCommandsForKey getInternal(RoutingKey key)
{
Map<RoutingKey, AccordSafeCommandsForKey> commandsForKey = task.commandsForKey();
if (commandsForKey == null)
AccordSafeCommandsForKey safeCfk = (AccordSafeCommandsForKey) task.refs.get(key);
if (safeCfk == null || safeCfk.isUninitialised())
return null;
return commandsForKey.get(key);
return safeCfk;
}
@Override
public void setRangesForEpoch(CommandStores.RangesForEpoch rangesForEpoch)
{
super.setRangesForEpoch(rangesForEpoch);
commandStore.updateMinHlc(PaxosState.ballotTracker().getLowBound().unixMicros() + 1);
commandStore().updateMinHlc(PaxosState.ballotTracker().getLowBound().unixMicros() + 1);
}
@Override
@ -194,13 +177,13 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
public void reportDurable(RedundantBefore addRedundantBefore, int flags)
{
upsertRedundantBefore(addRedundantBefore);
commandStore.maybeTerminated(ReportDurable.isCommandStoreFlush(flags), ReportDurable.isDataStoreFlush(flags));
commandStore().maybeTerminated(ReportDurable.isCommandStoreFlush(flags), ReportDurable.isDataStoreFlush(flags));
}
@Override
public AccordCommandStore commandStore()
{
return commandStore;
return task.commandStore;
}
@Override
@ -212,7 +195,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
@Override
public Agent agent()
{
return commandStore.agent();
return commandStore().agent();
}
@Override
@ -224,16 +207,16 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
@Override
public NodeCommandStoreService node()
{
return commandStore.node();
return commandStore().node();
}
public LogLinearDecayingHistograms.Buffer histogramBuffer()
{
if (task.histogramBuffer == null)
{
task.histogramBuffer = commandStore.metricsBuffer;
task.histogramBuffer = commandStore().metricsBuffer;
if (task.histogramBuffer == null)
task.histogramBuffer = commandStore.metricsBuffer = new LogLinearDecayingHistograms.Buffer(commandStore.executor().histograms);
task.histogramBuffer = commandStore().metricsBuffer = new LogLinearDecayingHistograms.Buffer(commandStore().executor().histograms);
Invariants.require(task.histogramBuffer.isEmpty());
}
return task.histogramBuffer;
@ -241,35 +224,48 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
private boolean visitForKey(Unseekables<?> keysOrRanges, Predicate<CommandsForKey> forEach)
{
Map<RoutingKey, AccordSafeCommandsForKey> commandsForKey = task.commandsForKey;
if (commandsForKey == null)
return true;
Unseekables<?> skip = context.keys().without(keysOrRanges);
for (SafeCommandsForKey safeCfk : commandsForKey.values())
Unseekables<?> unseekables = context.keys();
switch (unseekables.domain())
{
if (skip.contains(safeCfk.key()))
continue;
default: throw new UnhandledEnum(unseekables.domain());
case Key:
AbstractUnseekableKeys keys = (AbstractUnseekableKeys) context.keys();
return Routables.foldl(keys, keysOrRanges, (self, f, key, v, index) -> {
SafeCommandsForKey safeCfk = (SafeCommandsForKey) self.task.refs.get(key);
return f.test(safeCfk.current());
}, this, forEach, Boolean.TRUE, cont -> !cont);
if (!forEach.test(safeCfk.current()))
return false;
case Range:
Unseekables<?> skip = context.keys().without(keysOrRanges);
for (SafeState<?> safeState : task.refs.values())
{
if (!(safeState instanceof AccordSafeCommandsForKey))
continue;
SafeCommandsForKey safeCfk = (SafeCommandsForKey) safeState;
if (skip.contains(safeCfk.key()))
continue;
if (!forEach.test(safeCfk.current()))
return false;
}
return true;
}
return true;
}
@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; });
if (commandsForRanges != null)
commandsForRanges.visit(keysOrRanges, startedBefore, testKind, visitor, p1, p2);
if (task.commandsForRanges != null)
task.commandsForRanges.visit(keysOrRanges, startedBefore, testKind, visitor, p1, p2);
}
@Override
public boolean visit(Unseekables<?> keysOrRanges, TxnId testTxnId, Kinds testKind, SupersedingCommandVisitor visit)
{
return visitForKey(keysOrRanges, cfk -> cfk.visit(testTxnId, testKind, visit))
&& (commandsForRanges == null || commandsForRanges.visit(keysOrRanges, testTxnId, testKind, visit));
&& (task.commandsForRanges == null || task.commandsForRanges.visit(keysOrRanges, testTxnId, testKind, visit));
}
@Override

86
src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForKey.java
View File

@ -20,38 +20,32 @@ package org.apache.cassandra.service.accord;
import java.util.Objects;
import com.google.common.annotations.VisibleForTesting;
import accord.api.RoutingKey;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.NotifySink;
import accord.local.cfk.SafeCommandsForKey;
import accord.utils.Invariants;
public class AccordSafeCommandsForKey extends SafeCommandsForKey implements AccordSafeState<RoutingKey, CommandsForKey>
import org.apache.cassandra.service.accord.AccordCacheEntry.LockMode;
public class AccordSafeCommandsForKey extends SafeCommandsForKey implements AccordSafeState<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>
{
public static class CommandsForKeyCacheEntry extends AccordCacheEntry<RoutingKey, CommandsForKey>
public static class CommandsForKeyCacheEntry extends AccordCacheEntry<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>
{
private NotifySink overrideSink;
CommandsForKeyCacheEntry(RoutingKey key, AccordCache.Type<RoutingKey, CommandsForKey, ?>.Instance owner)
CommandsForKeyCacheEntry(RoutingKey key, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance owner)
{
super(key, owner);
}
}
private boolean invalidated;
private final AccordCacheEntry<RoutingKey, CommandsForKey> global;
private CommandsForKey original;
private CommandsForKey current;
private final AccordCacheEntry<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> global;
public AccordSafeCommandsForKey(AccordCacheEntry<RoutingKey, CommandsForKey> global)
public AccordSafeCommandsForKey(AccordCacheEntry<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> global)
{
super(global.key());
this.global = global;
this.original = null;
this.current = null;
// if (overrideSink() == null)
// overrideSink(new RecordingNotifySink());
}
@Override
@ -60,7 +54,7 @@ public class AccordSafeCommandsForKey extends SafeCommandsForKey implements Acco
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
AccordSafeCommandsForKey that = (AccordSafeCommandsForKey) o;
return Objects.equals(original, that.original) && Objects.equals(current, that.current);
return Objects.equals(current, that.current);
}
@Override
@ -73,47 +67,21 @@ public class AccordSafeCommandsForKey extends SafeCommandsForKey implements Acco
public String toString()
{
return "AccordSafeCommandsForKey{" +
"invalidated=" + invalidated +
"state=" + statusString() +
", global=" + global +
", original=" + original +
", current=" + current +
'}';
}
@Override
public boolean hasUpdate()
public final AccordCacheEntry<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> global()
{
boolean hasUpdate = AccordSafeState.super.hasUpdate();
// cfk initialization is legal, but doesn't need to be propagated to the cache (and would
// cause an exception to be thrown if it were). Making an exception on the cache side could
// throw away applied cfk updates as well, so it's special cased here
if (hasUpdate && original == null && current != null && current.size() == 0)
return false;
return hasUpdate;
}
@Override
public AccordCacheEntry<RoutingKey, CommandsForKey> global()
{
checkNotInvalidated();
return global;
}
@Override
public CommandsForKey current()
public void postExecute(AccordTask<?> owner)
{
checkNotInvalidated();
return current;
}
@Override
@VisibleForTesting
public void set(CommandsForKey cfk)
{
checkNotInvalidated();
this.current = cfk;
global.releaseExclusive(this, owner);
}
@Override
@ -128,31 +96,13 @@ public class AccordSafeCommandsForKey extends SafeCommandsForKey implements Acco
return ((CommandsForKeyCacheEntry)global).overrideSink;
}
public CommandsForKey original()
public void preExecute(AccordTask<?> owner, LockMode lockMode)
{
checkNotInvalidated();
return original;
}
@Override
public void preExecute()
{
checkNotInvalidated();
original = global.getExclusive();
current = original;
if (isUnset())
requireUninitialised();
current = global.lockExclusive(owner, lockMode);
if (current == null)
initialize();
setSafe();
}
@Override
public void markUnsafe()
{
invalidated = true;
}
@Override
public boolean isUnsafe()
{
return invalidated;
}
}

42
src/java/org/apache/cassandra/service/accord/AccordSafeState.java
View File

@ -17,41 +17,31 @@
*/
package org.apache.cassandra.service.accord;
import accord.impl.SafeState;
import accord.local.SafeState;
public interface AccordSafeState<K, V> extends SafeState<V>
import org.apache.cassandra.service.accord.AccordCacheEntry.LockMode;
public interface AccordSafeState<K, V, S extends SafeState<V> & AccordSafeState<K, V, S>>
{
void set(V update);
V original();
void markUnsafe();
boolean isUnsafe();
void preExecute();
AccordCacheEntry<K, V, S> global();
void postExecute(AccordTask<?> owner);
void preExecute(AccordTask<?> owner, LockMode lockMode);
AccordCacheEntry<K, V> global();
default boolean hasUpdate()
static AccordCacheEntry<?, ?, ?> global(SafeState<?> safeState)
{
return original() != current();
return safeState.getClass() == AccordSafeCommand.class ? ((AccordSafeCommand) safeState).global()
: ((AccordSafeCommandsForKey) safeState).global();
}
default void revert()
static void postExecute(SafeState<?> safeState, AccordTask<?> owner)
{
set(original());
if (safeState.getClass() == AccordSafeCommand.class) ((AccordSafeCommand) safeState).postExecute(owner);
else ((AccordSafeCommandsForKey) safeState).postExecute(owner);
}
default K key()
static void preExecute(SafeState<?> safeState, AccordTask<?> owner, LockMode lockMode)
{
return global().key();
}
default Throwable failure()
{
return global().failure();
}
default void checkNotInvalidated()
{
if (isUnsafe())
throw new IllegalStateException("Cannot access invalidated " + this);
if (safeState.getClass() == AccordSafeCommand.class) ((AccordSafeCommand) safeState).preExecute(owner, lockMode);
else ((AccordSafeCommandsForKey) safeState).preExecute(owner, lockMode);
}
}

1
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -464,6 +464,7 @@ public class AccordService implements IAccordService, Shutdownable
agent.setup(localId);
AccordTimeService time = new AccordTimeService();
this.scheduler = new AccordScheduler();
// TODO (expected): can we pass ImmediateExecutor rather than Scheduler?
final RequestCallbacks callbacks = new RequestCallbacks(time, scheduler);
this.dataStore = new AccordDataStore();
this.journal = new AccordJournal(DatabaseDescriptor.getAccord().journal);

1348
src/java/org/apache/cassandra/service/accord/AccordTask.java
View File

File diff suppressed because it is too large Load Diff

3
src/java/org/apache/cassandra/service/accord/InMemoryRangeIndex.java
View File

@ -88,7 +88,7 @@ public class InMemoryRangeIndex extends InMemoryRangeSummaryIndex implements Ran
{
for (TxnId txnId : load)
{
AccordCacheEntry<TxnId, Command> entry = caches.commands().getUnsafe(txnId);
AccordCacheEntry<TxnId, Command, ?> entry = caches.commands().getUnsafe(txnId);
if (entry == null)
{
loadFromDisk.add(txnId);
@ -115,7 +115,6 @@ public class InMemoryRangeIndex extends InMemoryRangeSummaryIndex implements Ran
public void finish(Map<Timestamp, Summary> into)
{
cleanupExclusive(null);
owner.search(this, into::put, null);
}

2
src/java/org/apache/cassandra/service/accord/RangeIndex.java
View File

@ -82,7 +82,7 @@ public interface RangeIndex
return null;
}
public CommandSummaries.Summary ifRelevant(AccordCacheEntry<TxnId, Command> state)
public CommandSummaries.Summary ifRelevant(AccordCacheEntry<TxnId, Command, ?> state)
{
if (state.key().domain() != Routable.Domain.Range)
return null;

6
src/java/org/apache/cassandra/service/accord/debug/DebugBlockedTxns.java
View File

@ -178,7 +178,7 @@ public class DebugBlockedTxns
private AsyncChain<Txn> visitRootTxnAsync(CommandStore commandStore, TxnId txnId)
{
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_blocked_by"), safeStore -> {
return commandStore.chain(ExecutionContext.unsequenced(txnId, "Populate txn_blocked_by"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return null;
@ -188,7 +188,7 @@ public class DebugBlockedTxns
private AsyncChain<Txn> visitTxnAsync(CommandStore commandStore, TxnId txnId, Timestamp rootExecuteAt, @Nullable TokenKey byKey, int depth, boolean recurse)
{
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_blocked_by"), safeStore -> {
return commandStore.chain(ExecutionContext.unsequenced(txnId, "Populate txn_blocked_by"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return null;
@ -231,7 +231,7 @@ public class DebugBlockedTxns
private AsyncChain<Void> visitKeysAsync(CommandStore commandStore, TokenKey key, Timestamp rootExecuteAt, int depth)
{
return commandStore.chain(ExecutionContext.contextFor(RoutingKeys.of(key.toUnseekable()), SYNC, READ_WRITE, "Populate txn_blocked_by"), safeStore -> {
return commandStore.chain(ExecutionContext.unsequencedReadWrite(RoutingKeys.of(key.toUnseekable()), "Populate txn_blocked_by"), safeStore -> {
visitKeysSync(safeStore, key, rootExecuteAt, depth);
});
}

54
src/java/org/apache/cassandra/service/accord/debug/DebugExecution.java
View File

@ -40,10 +40,10 @@ public class DebugExecution
{
private static final Logger logger = LoggerFactory.getLogger(DebugExecution.class);
public static final boolean DEBUG_EXECUTION = CassandraRelevantProperties.ACCORD_DEBUG_EXECUTION.getBoolean(false);
private static final long REPORT_MIN_LATENCY_MICROS = 20_000;
private static final long REPORT_MIN_LATENCY_MICROS = 50_000;
private static final long REPORT_CPU_RATIO = 2;
private static final long REPORT_MAX_LATENCY_MICROS = 50_000;
private static final long REPORT_CPU_MICROS = 10_000;
private static final long REPORT_MAX_LATENCY_MICROS = 100_000;
private static final long REPORT_CPU_MICROS = 50_000;
// TODO (expected): use sharded histogram so we can report global stats
public static class DebugExecutor
@ -94,43 +94,16 @@ public class DebugExecution
long lockedForCpuMicros = (unlockedAtCpu - lockedAtCpu)/1000;
if (lockedForMicros >= REPORT_MAX_LATENCY_MICROS)
{
report("Held lock for {}us (cpu:{}us)\n", lockedForMicros, lockedForCpuMicros);
report("Held lock for {}us (cpu:{}us)", lockedForMicros, lockedForCpuMicros);
}
else if (lockedForMicros >= REPORT_MIN_LATENCY_MICROS && (lockedForMicros / lockedForCpuMicros) >= REPORT_CPU_RATIO)
{
report("Held lock for {}us with cpu time only {}us\n", lockedForMicros, lockedForCpuMicros);
report("Held lock for {}us with cpu time only {}us", lockedForMicros, lockedForCpuMicros);
}
locked.increment(lockedForMicros);
}
}
public static class DebugExecutorLoop
{
final DebugExecutor owner;
long lockAt;
public DebugExecutorLoop(DebugExecutor owner)
{
this.owner = owner;
}
public void onLock()
{
lockAt = nanoTime();
}
public void onEnterLock()
{
owner.onEnterLock(lockAt);
lockAt = 0;
}
public void onExitLock()
{
owner.onExitLock();
}
}
public static class DebugExclusiveExecutor
{
public static DebugExclusiveExecutor maybeDebug(DebugExecutor owner, int commandStoreId)
@ -200,7 +173,7 @@ public class DebugExecution
public List<Command> sanityCheck; // for AccordTask only
long polledAt, preRunAt, runCompleteAt, completedAt;
long releasedRangeScannerAt, releasedCommandsAt, releasedCommandsForKeyAt;
long releasedRangeScannerAt, releasedStateAt;
long runningAtCpu, runCompleteAtCpu;
Thread thread;
@ -231,14 +204,9 @@ public class DebugExecution
releasedRangeScannerAt = nanoTime();
}
public void onReleasedCommands()
public void onReleasedState()
{
releasedCommandsAt = nanoTime();
}
public void onReleasedCommandsForKeys()
{
releasedCommandsForKeyAt = nanoTime();
releasedStateAt = nanoTime();
}
public void onCompleted(DebugExecutor owner)
@ -254,9 +222,9 @@ public class DebugExecution
runningMicros = (runCompleteAt - task.runningAt) / 1000;
owner.running.increment(runningMicros);
}
long runToCleanMicros = (task.cleanupAt - runCompleteAt)/1000;
long runToCleanMicros = (task.completeAt - runCompleteAt) / 1000;
owner.runToCleanup.increment(runToCleanMicros);
long cleanupMicros = (completedAt - task.cleanupAt)/1000;
long cleanupMicros = (completedAt - task.completeAt) / 1000;
owner.cleanup.increment(cleanupMicros);
long totalMicros = (completedAt - polledAt)/1000;
owner.taskTotal.increment(totalMicros);
@ -266,7 +234,7 @@ public class DebugExecution
String reason = "";
if (totalMicros > REPORT_MAX_LATENCY_MICROS) reason += "LONG TIME ";
if (totalCpu > REPORT_CPU_MICROS) reason += "HIGH CPU ";
if ((totalMicros > REPORT_MIN_LATENCY_MICROS && (totalMicros/totalCpu) >= REPORT_CPU_RATIO)) reason += "LOW RATIO ";
if ((totalMicros > REPORT_MIN_LATENCY_MICROS && (totalCpu == 0 || (totalMicros/totalCpu) >= REPORT_CPU_RATIO))) reason += "LOW RATIO ";
report("{}{}: total {}us cpu:{}us ({}), pollToRun {}us, running {}us, runToClean {}us, cleanup {}us",
reason, task, totalMicros, totalCpu, thread, pollToRunMicros, runningMicros, runToCleanMicros, cleanupMicros);
}

6
src/java/org/apache/cassandra/service/accord/debug/DebugTxnGraph.java
View File

@ -221,7 +221,7 @@ public abstract class DebugTxnGraph<T, P>
private AsyncChain<TxnInfos<T>> submitRoot(CommandStore commandStore, TxnId txnId)
{
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
return commandStore.chain(ExecutionContext.unsequenced(txnId, "Populate txn_graph"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return AsyncChains.<TxnInfos<T>>success(null);
@ -232,7 +232,7 @@ public abstract class DebugTxnGraph<T, P>
private AsyncChain<TxnInfos<T>> submitParent(CommandStore commandStore, TxnId txnId, P param, Map<TxnId, SaveInfo> infos, Set<TxnId> visitedParent, int depth)
{
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
return commandStore.chain(ExecutionContext.unsequenced(txnId, "Populate txn_graph"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return AsyncChains.<TxnInfos<T>>success(null);
@ -344,7 +344,7 @@ public abstract class DebugTxnGraph<T, P>
private AsyncChain<Void> populateTxnAsync(CommandStore commandStore, TxnId txnId, Map<TxnId, SaveInfo> visited)
{
return commandStore.chain(ExecutionContext.contextFor(txnId, "Populate txn_graph"), safeStore -> {
return commandStore.chain(ExecutionContext.unsequenced(txnId, "Populate txn_graph"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
visited.putIfAbsent(txnId, command == null || command.saveStatus() == SaveStatus.Uninitialised ? SaveInfo.NONE : new SaveInfo(command.saveStatus(), command.executeAtIfKnown()));
});

8
src/java/org/apache/cassandra/service/accord/journal/JournalRangeIndex.java
View File

@ -107,7 +107,7 @@ public class JournalRangeIndex extends SemiSyncIntervalTree<Object[]> implements
}
@Override
public void onUpdate(AccordCacheEntry<TxnId, Command> state)
public void onUpdate(AccordCacheEntry<TxnId, Command, ?> state)
{
Summary summary = loader.ifRelevant(state);
if (summary != null)
@ -162,7 +162,7 @@ public class JournalRangeIndex extends SemiSyncIntervalTree<Object[]> implements
if (isMaybeRelevant(i))
{
TxnId txnId = i.txnId;
AccordCacheEntry<TxnId, Command> entry = c.getUnsafe(txnId);
AccordCacheEntry<TxnId, Command, ?> entry = c.getUnsafe(txnId);
Invariants.expect(entry != null, "%s found interval %s but no matching transaction in cache", owner.commandStore, i);
if (entry != null)
{
@ -267,7 +267,7 @@ public class JournalRangeIndex extends SemiSyncIntervalTree<Object[]> implements
}
@Override
public void onUpdate(AccordCacheEntry<TxnId, Command> state)
public void onUpdate(AccordCacheEntry<TxnId, Command, ?> state)
{
TxnId txnId = state.key();
if (txnId.is(Routable.Domain.Range))
@ -334,7 +334,7 @@ public class JournalRangeIndex extends SemiSyncIntervalTree<Object[]> implements
}
@Override
public void onEvict(AccordCacheEntry<TxnId, Command> state)
public void onEvict(AccordCacheEntry<TxnId, Command, ?> state)
{
TxnId txnId = state.key();
if (txnId.is(Routable.Domain.Range))

12
test/distributed/org/apache/cassandra/distributed/test/accord/AccordDropTableBase.java
View File

@ -22,12 +22,12 @@ import com.google.common.base.Throwables;
import org.assertj.core.api.Assertions;
import accord.api.RoutingKey;
import accord.local.CommandStores;
import accord.local.ExecutionContext;
import accord.local.LoadKeys;
import accord.local.Node;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.SafeCommandsForKey;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Txn;
@ -43,7 +43,6 @@ import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.AccordCommandStore;
import org.apache.cassandra.service.accord.AccordSafeCommandStore;
import org.apache.cassandra.service.accord.AccordSafeCommandsForKey;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.TokenRange;
@ -135,19 +134,16 @@ public class AccordDropTableBase extends TestBaseImpl
TableId tableId = TableId.fromString(s);
AccordService accord = (AccordService) AccordService.instance();
TxnId syntheticTxnId = new TxnId(TxnId.MAX_EPOCH, 0, Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range, new Node.Id(1));
ExecutionContext ctx = ExecutionContext.contextFor(syntheticTxnId, Ranges.single(TokenRange.fullRange(tableId, getPartitioner())), LoadKeys.SYNC, READ_WRITE, "Test");
ExecutionContext ctx = ExecutionContext.unsequencedReadWrite(syntheticTxnId, Ranges.single(TokenRange.fullRange(tableId, getPartitioner())), "Test");
CommandStores stores = accord.node().commandStores();
for (int storeId : stores.ids())
{
AccordCommandStore store = (AccordCommandStore) stores.forId(storeId);
getBlocking(store.chain(ctx, input -> {
AccordSafeCommandStore safe = (AccordSafeCommandStore) input;
for (RoutingKey key : safe.commandsForKeysKeys())
for (SafeCommandsForKey safeCfk : safe.safeCommandsForKeys())
{
AccordSafeCommandsForKey safeCFK = (AccordSafeCommandsForKey) safe.ifLoadedAndInitialised(key);
if (safeCFK == null) // we read and found a key, but its null at load time... so ignore it
continue;
CommandsForKey cfk = safeCFK.current();
CommandsForKey cfk = safeCfk.current();
CommandsForKey.TxnInfo minUndecided = cfk.minUndecidedManaged();
if (minUndecided != null)
throw new AssertionError("Undecided txn: " + minUndecided);

16
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
View File

@ -355,6 +355,12 @@ public class AccordLoadTest extends AccordTestBase
return builder.setArtificialLatencies(LATENCIES);
}
private static SettingsBuilder populate(SettingsBuilder builder, int keyCount)
{
return builder.setKeySelector(roundrobin(keyCount))
.setReadRatio(0f);
}
private static SettingsBuilder ycsbA(SettingsBuilder builder, int keyCount)
{
return builder.setKeySelector(ycsbZipfian(keyCount))
@ -832,7 +838,7 @@ public class AccordLoadTest extends AccordTestBase
if (storeId.get() >= 0)
{
CommandStore commandStore = service.node().commandStores().forId(storeId.get());
List<List<String>> result = AccordService.getBlocking(commandStore.submit(ExecutionContext.contextFor(candidate, "LoadTest"), safeStore -> {
List<List<String>> result = AccordService.getBlocking(commandStore.submit(ExecutionContext.unsequenced(candidate, "LoadTest"), safeStore -> {
SafeCommand safeCommand = safeStore.unsafeGet(candidate);
PartialDeps deps = safeCommand.current().partialDeps();
if (deps == null)
@ -855,7 +861,7 @@ public class AccordLoadTest extends AccordTestBase
for (List<String> info : result)
{
TxnId txnId = TxnId.parse(info.get(0));
AccordService.getBlocking(commandStore.execute(ExecutionContext.contextFor(txnId, "LoadTest"), safeStore -> {
AccordService.getBlocking(commandStore.execute(ExecutionContext.unsequenced(txnId, "LoadTest"), safeStore -> {
SafeCommand safeCommand = safeStore.unsafeGet(txnId);
if (safeCommand.current().executeAt != null)
info.add(safeCommand.current().executeAt.toString());
@ -915,7 +921,7 @@ public class AccordLoadTest extends AccordTestBase
cluster.forEach(() -> {
refresh(AccordExecutorMetrics.INSTANCE.elapsedRunning);
refresh(AccordExecutorMetrics.INSTANCE.elapsed);
System.out.printf("%tT.%tL (%d %d %d %d %d %d)ms (%d %d %d %d %d %d)ms (%d %d %d %d %.0f, %d %d %d)us %d %d %d\n", nowMillis, nowMillis,
System.out.printf("%tT.%tL (%d %d %d %d %d %d)ms (%d %d %d %d %d %d)ms (%d %d %d)us %d %.0f (%d %d %d)us %d %d %d\n", nowMillis, nowMillis,
getLatency(AccordCoordinatorMetrics.readMetrics.preacceptLatency, 0.5),
getLatency(AccordCoordinatorMetrics.readMetrics.executeLatency, 0.5),
getLatency(AccordCoordinatorMetrics.readMetrics.applyLatency, 0.5),
@ -1092,13 +1098,13 @@ public class AccordLoadTest extends AccordTestBase
try
{
test.setup();
test.testLoad(withArtificialLatencies(ycsbA(new SettingsBuilder(), 100_000)
test.testLoad(populate(new SettingsBuilder(), 1_000_000)
// .setRatePerSecond(400).setMinRatePerSecond(200)
// .setRatePerSecond(800).setMinRatePerSecond(200)
.setRatePerSecond(1600).setMinRatePerSecond(200)
.setIncreaseRatePerSecondInterval(5000)
// .setTraceLast(5000)
).build());
.build());
}
finally
{

3
test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalConsistentExpungeTest.java
View File

@ -38,6 +38,7 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.AccordCacheEntry;
import org.apache.cassandra.service.accord.AccordCommandStore;
import org.apache.cassandra.service.accord.AccordSafeCommand;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -86,7 +87,7 @@ public class AccordJournalConsistentExpungeTest extends TestBaseImpl
Node node = service.node();
AccordCommandStore commandStore = (AccordCommandStore) node.commandStores().unsafeForKey(key.toUnseekable());
Iterator<AccordCacheEntry<TxnId, Command>> iterator = commandStore.cachesUnsafe().commands().iterator();
Iterator<AccordCacheEntry<TxnId, Command, AccordSafeCommand>> iterator = commandStore.cachesUnsafe().commands().iterator();
TxnId txnId = TxnId.NONE;

329
test/simulator/test/org/apache/cassandra/simulator/test/AccordExecutorAndCacheTest.java Normal file
View File

@ -0,0 +1,329 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.simulator.test;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.LockSupport;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import javax.annotation.Nullable;
import org.junit.Test;
import accord.api.AsyncExecutor;
import accord.api.ExclusiveAsyncExecutor;
import accord.utils.async.Cancellable;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableSupplier;
import org.apache.cassandra.service.accord.AccordExecutor;
import org.apache.cassandra.service.accord.AccordExecutorAsyncSubmit;
import org.apache.cassandra.service.accord.AccordExecutorSignalLoop;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.SignalLock;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITHOUT_LOCK;
// TODO (required): have simulator intercept ReentantLock so we can test SyncSubmit and SemiSyncSubmit
public class AccordExecutorAndCacheTest extends SimulationTestBase
{
@Test
public void signalLoopTest()
{
executorTest(() -> new AccordExecutorSignalLoop(1, RUN_WITHOUT_LOCK, SignalLock.MAX_THREADS, -1, -1, TimeUnit.MICROSECONDS, i -> "Loop" + i, new AccordAgent()),
16);
}
static class Submitted
{
final AtomicInteger nextId = new AtomicInteger();
final AtomicInteger doneBefore = new AtomicInteger();
final ConcurrentHashMap<Integer, Collection<Future<?>>> consequences = new ConcurrentHashMap<>();
boolean isDone()
{
return isDoneBetween(0, nextId.get());
}
boolean isDoneBefore(int before)
{
if (!isDoneBetween(doneBefore.get(), before))
return false;
doneBefore.accumulateAndGet(before, Integer::max);
return true;
}
boolean isDoneBetween(int from, int before)
{
for (int id = from ; id < before ; ++id)
{
for (Future<?> future : consequences.get(id))
{
if (!future.isDone())
return false;
}
}
return true;
}
Collection<Future<?>> start()
{
ConcurrentLinkedQueue<Future<?>> result = new ConcurrentLinkedQueue<>();
while (true)
{
int id = nextId.get();
Object prev = consequences.putIfAbsent(id, result);
nextId.compareAndSet(id, id + 1);
if (prev == null)
return result;
}
}
}
static class Control extends ConcurrentLinkedQueue<Cancellable>
{
final Submitted submitted;
final AtomicInteger count = new AtomicInteger();
final float cancelChance;
float processChance;
Control(float cancelChance, Submitted submitted)
{
this(submitted, cancelChance, ThreadLocalRandom.current().nextFloat() * 0.5f);
}
Control(Submitted submitted, float cancelChance, float processChance)
{
this.submitted = submitted;
this.cancelChance = cancelChance;
this.processChance = processChance;
}
void submit(AsyncExecutor executor, Collection<Future<?>> consequences, Consumer<Collection<Future<?>>> run)
{
AsyncPromise<?> future = new AsyncPromise<>();
Cancellable cancel = executor.chain(() -> run.accept(consequences)).begin((success, fail) -> {
if (fail == null) future.trySuccess(null);
else
{
future.tryFailure(fail);
if (fail instanceof CancellationException)
run.accept(submitted.start());
}
});
consequences.add(future);
if (cancel != null && ThreadLocalRandom.current().nextFloat() <= cancelChance)
{
add(cancel);
count.incrementAndGet();
}
if (count.get() > 0 && ThreadLocalRandom.current().nextFloat() <= processChance)
{
int cancelCount = 0;
do
{
++cancelCount;
float delta = ThreadLocalRandom.current().nextFloat() - 0.5f;
if (delta < 0) processChance /= delta;
else processChance *= -delta;
if (processChance < 0.001f || processChance > 0.999f)
processChance = cancelChance;
} while (count.decrementAndGet() > 0 && ThreadLocalRandom.current().nextFloat() <= processChance);
// do outside of loop to avoid reentry
while (cancelCount-- > 0)
remove().cancel();
}
}
}
public void executorTest(SerializableSupplier<AccordExecutor> supplier, int submissionThreads)
{
simulate(arr(() -> {
try
{
DatabaseDescriptor.daemonInitialization();
ExecutorPlus submit = executorFactory().pooled("submit-test", submissionThreads);
AccordExecutor executor = supplier.get();
Lock lock = executor.unsafeLock();
ExclusiveAsyncExecutor sequentialExecutor = executor.newExclusiveExecutor();
Executor lockExecutor = executorFactory().sequential("lock");
for (float sleepChance : new float[] { 0f, 0.01f, 0.1f })
{
for (float lockChance : new float[] { 0f, 0.01f, 0.1f })
{
for (float cancelChance : new float[] { 0f, 0.01f, 0.1f })
{
System.out.println(String.format("sleepChance %.2f, lockChance %.2f, cancelChance %.2f", sleepChance, lockChance, cancelChance));
List<Future<?>> done = new ArrayList<>();
Submitted submitted = new Submitted();
for (int i = 0; i < submissionThreads; ++i)
{
int id = i;
done.add(submit.submit(() -> {
try
{
submitLoop(id, lock, executor, sequentialExecutor, lockExecutor, 20, 10, sleepChance, lockChance, new Control(cancelChance, submitted), submitted);
}
catch (ExecutionException | InterruptedException e)
{
throw new RuntimeException(e);
}
}));
}
for (Future<?> f : done)
f.get();
if (!submitted.isDone())
throw new AssertionError();
}
}
}
}
catch (Throwable t)
{
throw new RuntimeException(t);
}
}),
() -> {}, 1L);
}
private static void submitLoop(int id, Lock lock, AccordExecutor executor, ExclusiveAsyncExecutor sequentialExecutor, Executor lockExecutor, int outerLoop, int innerLoop, float sleepChance, float lockChance, Control control, Submitted submitted) throws ExecutionException, InterruptedException
{
ConcurrentLinkedQueue<Future<?>> awaitConsequences = new ConcurrentLinkedQueue<>();
while (outerLoop-- > 0)
{
List<Collection<Future<?>>> allAwaitSubmitted = new ArrayList<>();
for (int i = 0; i < innerLoop; ++i)
{
Collection<Future<?>> awaitSubmitted = submitted.start();
allAwaitSubmitted.add(awaitSubmitted);
submitRecursive(lock, executor, sequentialExecutor, 1 + i, awaitSubmitted, awaitConsequences, submitted, sleepChance, lockChance, control);
}
AtomicBoolean done = new AtomicBoolean();
submitUntil(lock, lockExecutor, sleepChance, done::get);
for (Collection<Future<?>> awaitSubmitted : allAwaitSubmitted)
await(awaitSubmitted, CancellationException.class);
await(awaitConsequences, null);
done.set(true);
System.out.println("Loop " + id + '.' + (1 + outerLoop));
}
}
private static void submitRecursive(Lock lock, AccordExecutor executor, ExclusiveAsyncExecutor sequentialExecutor, int count, Collection<Future<?>> consequences, Collection<Future<?>> awaitConsequences, Submitted submitted, float sleepChance, float lockChance, Control control)
{
AsyncExecutor submitTo = ThreadLocalRandom.current().nextBoolean() ? executor : sequentialExecutor;
control.submit(submitTo, consequences, nextConsequences -> {
ThreadLocalRandom rnd = ThreadLocalRandom.current();
boolean locked = false;
if (rnd.nextFloat() < lockChance)
{
if (rnd.nextBoolean()) locked = lock.tryLock();
else { locked = true; lock.lock(); }
}
if (ThreadLocalRandom.current().nextFloat() < 0.01f)
{
int expectDoneBefore = submitted.nextId.get();
AsyncPromise<Void> afterConsequences = new AsyncPromise<>();
executor.afterSubmittedAndConsequences(() -> {
if (!submitted.isDoneBefore(expectDoneBefore))
throw new AssertionError();
afterConsequences.setSuccess(null);
});
awaitConsequences.add(afterConsequences);
}
try
{
if (count > 1)
submitRecursive(lock, executor, sequentialExecutor, count -1, nextConsequences, awaitConsequences, submitted, sleepChance, lockChance, control);
if (rnd.nextFloat() < sleepChance)
LockSupport.parkNanos(rnd.nextInt(10000, 100000));
}
finally
{
if (locked)
lock.unlock();
}
});
}
private static void submitUntil(Lock lock, Executor executor, float sleepChance, BooleanSupplier done)
{
if (done.getAsBoolean())
return;
executor.execute(() -> {
ThreadLocalRandom rnd = ThreadLocalRandom.current();
boolean tryLock = rnd.nextBoolean();
boolean locked = !tryLock;
if (tryLock) locked = lock.tryLock();
else lock.lock();
try
{
if (rnd.nextFloat() < sleepChance)
LockSupport.parkNanos(rnd.nextInt(10000, 100000));
submitUntil(lock, executor, sleepChance, done);
}
finally
{
if (locked)
lock.unlock();
}
});
}
private static void await(Collection<Future<?>> await, @Nullable Class<? extends Throwable> ignore) throws InterruptedException, ExecutionException
{
for (Future<?> future : await)
{
try { future.get(); }
catch (ExecutionException e)
{
if (ignore == null || !(ignore.isInstance(e.getCause())))
throw e;
}
}
}
}

13
test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
View File

@ -91,6 +91,7 @@ import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.utils.FBUtilities;
import static accord.local.ExecutionContext.contextFor;
import static accord.local.ExecutionContext.unsequencedReadWrite;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.RedundantStatus.SomeStatus.GC_BEFORE_AND_LOCALLY_DURABLE;
@ -322,28 +323,28 @@ public class CompactionAccordIteratorsTest
PartialDeps partialDeps = Deps.NONE.intersecting(AccordTestUtils.fullRange(txn));
PartialTxn partialTxn = txn.slice(commandStore.unsafeGetRangesForEpoch().currentRanges(), true);
Route<?> partialRoute = route.overlapping(commandStore.unsafeGetRangesForEpoch().currentRanges());
getBlocking(commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
getBlocking(commandStore.execute(unsequencedReadWrite(txnId, route, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, (a, b) -> {});
}));
flush(commandStore);
getBlocking(commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
getBlocking(commandStore.execute(unsequencedReadWrite(txnId, route, "Test"), safe -> {
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, txnId, partialDeps, (a, b) -> {});
}));
flush(commandStore);
getBlocking(commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
getBlocking(commandStore.execute(unsequencedReadWrite(txnId, route, "Test"), safe -> {
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, txnId, partialDeps, (a, b) -> {});
}));
flush(commandStore);
getBlocking(commandStore.chain(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
getBlocking(commandStore.chain(unsequencedReadWrite(txnId, route, "Test"), safe -> {
return AccordTestUtils.processTxnResultDirect(safe, txnId, partialTxn, txnId);
}).flatMap(i -> i).flatMap(result -> commandStore.chain(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
}).flatMap(i -> i).flatMap(result -> commandStore.chain(unsequencedReadWrite(txnId, route, "Test"), safe -> {
CheckedCommands.apply(safe, txnId, route, txnId, partialDeps, partialTxn, result.left, result.right, (a, b) -> {});
})));
flush(commandStore);
// The apply chain is asychronous, so it is easiest to just spin until it is applied
// in order to have the updated state in the system table
spinAssertEquals(true, 5, () -> {
return getBlocking(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
return getBlocking(commandStore.submit(unsequencedReadWrite(txnId, route, "Test"), safe -> {
StoreParticipants participants = StoreParticipants.all(route);
Command command = safe.get(txnId, participants).current();
return command.hasBeen(Status.Applied);

14
test/unit/org/apache/cassandra/service/accord/AccordCacheEntryTest.java
View File

@ -20,14 +20,24 @@ package org.apache.cassandra.service.accord;
import org.junit.Assert;
import org.junit.Test;
import accord.local.SafeState;
import org.apache.cassandra.service.accord.AccordCache.Type;
import org.apache.cassandra.service.accord.AccordCacheEntry.LockMode;
import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
public class AccordCacheEntryTest
{
static class CacheEntry extends AccordCacheEntry<String, String>
static class TestSafeState extends SafeState<String> implements AccordSafeState<String, String, TestSafeState>
{
public CacheEntry(String key, Type<String, String, ?>.Instance instance)
@Override public AccordCacheEntry<String, String, TestSafeState> global() { return null; }
@Override public void preExecute(AccordTask<?> owner, LockMode lockMode) {}
@Override public void postExecute(AccordTask<?> owner) {}
}
static class CacheEntry extends AccordCacheEntry<String, String, TestSafeState>
{
public CacheEntry(String key, Type<String, String, TestSafeState>.Instance instance)
{
super(key, instance);
}

86
test/unit/org/apache/cassandra/service/accord/AccordCacheTest.java
View File

@ -23,13 +23,18 @@ import org.agrona.concurrent.NoOpLock;
import org.junit.Assert;
import org.junit.Test;
import accord.local.ExecutionContext;
import accord.local.SafeState;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.concurrent.ManualExecutor;
import org.apache.cassandra.metrics.AccordCacheMetrics;
import org.apache.cassandra.service.accord.AccordCacheEntry.LockMode;
import org.apache.cassandra.service.accord.AccordCacheEntry.SaveExecutor;
import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
import static org.apache.cassandra.service.accord.AccordCacheEntry.LockMode.RELEASE_QUEUE;
import static org.apache.cassandra.service.accord.AccordTestUtils.testLoad;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
@ -40,85 +45,56 @@ public class AccordCacheTest
{
private static final long DEFAULT_NODE_SIZE = nodeSize(0);
private static abstract class TestSafeState<T> implements AccordSafeState<T, T>
private static abstract class TestSafeState<T, S extends SafeState<T> & AccordSafeState<T, T, S>> extends SafeState<T> implements AccordSafeState<T, T, S>
{
protected boolean invalidated = false;
protected final AccordCacheEntry<T, T> global;
private T original = null;
protected final AccordCacheEntry<T, T, S> global;
public TestSafeState(AccordCacheEntry<T, T> global)
public TestSafeState(AccordCacheEntry<T, T, S> global)
{
this.global = global;
}
public AccordCacheEntry<T, T> global()
public AccordCacheEntry<T, T, S> global()
{
return global;
}
@Override
public T key()
{
return global.key();
}
public final T key() { return global.key(); }
@Override
public T current()
public void preExecute(AccordTask<?> owner, LockMode lockMode)
{
return global.getExclusive();
}
@Override
public void set(T update)
{
global.setExclusive(update);
}
@Override
public T original()
{
return original;
}
@Override
public void preExecute()
{
original = global.getExclusive();
}
@Override
public Throwable failure()
{
return global.failure();
}
@Override
public void markUnsafe()
{
invalidated = true;
}
@Override
public boolean isUnsafe()
{
return invalidated;
requireUninitialised();
current = global.lockExclusive(owner, lockMode);
setSafe();
}
}
private static class SafeString extends TestSafeState<String>
private static class SafeString extends TestSafeState<String, SafeString>
{
public SafeString(AccordCacheEntry<String, String> global)
public SafeString(AccordCacheEntry<String, String, SafeString> global)
{
super(global);
}
@Override
public void postExecute(AccordTask<?> owner)
{
global.releaseExclusive(this, owner);
}
}
private static class SafeInt extends TestSafeState<Integer>
private static class SafeInt extends TestSafeState<Integer, SafeInt>
{
public SafeInt(AccordCacheEntry<Integer, Integer> global)
public SafeInt(AccordCacheEntry<Integer, Integer, SafeInt> global)
{
super(global);
}
@Override
public void postExecute(AccordTask<?> owner)
{
global.releaseExclusive(this, owner);
}
}
private static long emptyNodeSize()
@ -264,6 +240,7 @@ public class AccordCacheTest
assertCacheMetrics(cacheMetrics, 0, 3, 3, 3);
SafeString safeString = instance.acquire("1");
safeString.preExecute(new AccordTask<>(null, (ExecutionContext.Empty)() -> "Test", null), RELEASE_QUEUE);
Assert.assertEquals(Status.LOADED, safeString.global.status());
assertCacheState(cache, 1, 3, nodeSize(1) * 3);
@ -392,6 +369,7 @@ public class AccordCacheTest
assertCacheState(cache, 1, 1, nodeSize(1));
SafeString safeString2 = instance.acquire("0");
safeString2.preExecute(new AccordTask<>(null, (ExecutionContext.Empty)() -> "Test", null), RELEASE_QUEUE);
Assert.assertEquals("0", safeString2.current());
Assert.assertEquals(Status.LOADED, safeString1.global.status());
Assert.assertEquals(2, instance.references("0", SafeString.class));

17
test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
View File

@ -18,8 +18,6 @@
package org.apache.cassandra.service.accord;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicLong;
import org.junit.Assert;
@ -73,6 +71,7 @@ import org.apache.cassandra.utils.Pair;
import static accord.primitives.Status.Durability.AllQuorums;
import static com.google.common.collect.Iterables.getOnlyElement;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
import static org.apache.cassandra.service.accord.AccordCacheEntry.LockMode.RELEASE_QUEUE;
import static org.apache.cassandra.service.accord.AccordService.getBlocking;
import static org.apache.cassandra.service.accord.AccordTestUtils.Commands.preaccepted;
import static org.apache.cassandra.service.accord.AccordTestUtils.ballot;
@ -140,6 +139,7 @@ public class AccordCommandStoreTest
promised, executeAt, txn, dependencies, accepted,
waitingOn, result.left, TxnDataResult.PERSISTABLE);
AccordSafeCommand safeCommand = new AccordSafeCommand(loaded(txnId, null));
safeCommand.preExecute(new AccordTask<>(null, ExecutionContext.unsequenced(txnId, "Test"), null), RELEASE_QUEUE);
safeCommand.set(expected);
// In practice we should never need to save it with the condition boolean set
// Not sure why this test does that
@ -168,10 +168,10 @@ public class AccordCommandStoreTest
Command command2 = preaccepted(txnId2, txn, timestamp(1, clock.incrementAndGet(), 1));
AccordSafeCommandsForKey cfk = new AccordSafeCommandsForKey(loaded(key, null));
cfk.initialize();
cfk.preExecute(new AccordTask<>(null, ExecutionContext.unsequenced(txnId1, "Test"), null), RELEASE_QUEUE);
cfk.set(cfk.current().update(new TestSafeCommandStore(ExecutionContext.contextFor(command1.txnId(), "Test")), command1).cfk());
cfk.set(cfk.current().update(new TestSafeCommandStore(ExecutionContext.contextFor(command1.txnId(), "Test")), command2).cfk());
cfk.set(cfk.current().update(new TestSafeCommandStore(ExecutionContext.unsequenced(command1.txnId(), "Test")), command1).cfk());
cfk.set(cfk.current().update(new TestSafeCommandStore(ExecutionContext.unsequenced(command1.txnId(), "Test")), command2).cfk());
CommandsForKeyAccessor.systemTableUpdater(commandStore.id(), (TokenKey)cfk.key(), cfk.current(), null, commandStore.nextSystemTimestampMicros()).run();
logger.info("E: {}", cfk);
@ -180,11 +180,4 @@ public class AccordCommandStoreTest
Assert.assertEquals(cfk.current(), actual);
}
private static <K, V extends AccordSafeState<K, ?>> NavigableMap<K, V> toNavigableMap(V safeState)
{
TreeMap<K, V> map = new TreeMap<>();
map.put(safeState.key(), safeState);
return map;
}
}

2
test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java
View File

@ -174,7 +174,7 @@ public class AccordCommandTest
Commit commit = Commit.SerializerSupport.create(txnId, route, 1, 1, Commit.Kind.StableWithTxnAndDeps, Ballot.ZERO, executeAt, partialTxn, deps, fullRoute);
getBlocking(commandStore.execute(commit, commit::apply));
getBlocking(commandStore.execute(ExecutionContext.contextFor(txnId, Keys.of(key).toParticipants(), LoadKeys.SYNC, READ_WRITE, "Test"), safeStore -> {
getBlocking(commandStore.execute(ExecutionContext.unsequencedReadWrite(txnId, Keys.of(key).toParticipants(), "Test"), safeStore -> {
Command before = safeStore.ifInitialised(txnId).current();
Assert.assertEquals(commit.executeAt, before.executeAt());
Assert.assertTrue(before.hasBeen(Status.Committed));

13
test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
View File

@ -86,6 +86,7 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.Condition;
import static accord.local.ExecutionContext.contextFor;
import static accord.local.ExecutionContext.unsequencedReadWrite;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.utils.Property.qt;
@ -127,7 +128,7 @@ public class AccordTaskTest
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
getBlocking(commandStore.execute(ExecutionContext.contextFor(txnId, "Test"), instance -> {
getBlocking(commandStore.execute(ExecutionContext.unsequenced(txnId, "Test"), instance -> {
// TODO review: This change to `ifInitialized` was done in a lot of places and it doesn't preserve this property
// I fixed this reference to point to `ifLoadedAndInitialised` and but didn't update other places
Assert.assertNull(instance.ifInitialised(txnId));
@ -141,7 +142,7 @@ public class AccordTaskTest
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
TxnId txnId = txnId(1, clock.incrementAndGet(), 1);
getBlocking(commandStore.execute(ExecutionContext.contextFor(txnId, "Test"), safe -> {
getBlocking(commandStore.execute(ExecutionContext.unsequenced(txnId, "Test"), safe -> {
StoreParticipants participants = StoreParticipants.empty(txnId);
SafeCommand command = safe.get(txnId, participants);
Assert.assertNotNull(command);
@ -198,7 +199,7 @@ public class AccordTaskTest
route.overlapping(ranges);
PartialDeps deps = PartialDeps.builder(ranges, true).build();
Command command = getBlocking(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
Command command = getBlocking(commandStore.submit(unsequencedReadWrite(txnId, route, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
return safe.ifInitialised(txnId).current();
@ -246,7 +247,7 @@ public class AccordTaskTest
Route<?> partialRoute = route.overlapping(ranges);
PartialDeps deps = PartialDeps.builder(ranges, true).build();
Command command = getBlocking(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
Command command = getBlocking(commandStore.submit(unsequencedReadWrite(txnId, route, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToLog(commandStore));
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, executeAt, deps, appendDiffToLog(commandStore));
CheckedCommands.commit(safe, SaveStatus.Committed, Ballot.ZERO, txnId, route, partialTxn, executeAt, deps, appendDiffToLog(commandStore));
@ -441,7 +442,7 @@ public class AccordTaskTest
AssertionError error = null;
for (T key : keys)
{
AccordCacheEntry<T, ?> node = cache.getUnsafe(key);
AccordCacheEntry<T, ?, ?> node = cache.getUnsafe(key);
if (node == null) continue;
try
{
@ -476,7 +477,7 @@ public class AccordTaskTest
{
for (T key : keys)
{
AccordCacheEntry<T, ?> node = cache.getUnsafe(key);
AccordCacheEntry<T, ?, ?> node = cache.getUnsafe(key);
if (node == null) continue;
Awaitility.await("For node " + node.key() + " to complete")
.atMost(Duration.ofMinutes(1))

21
test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
View File

@ -57,6 +57,7 @@ import accord.local.Node;
import accord.local.Node.Id;
import accord.local.NodeCommandStoreService;
import accord.local.SafeCommandStore;
import accord.local.SafeState;
import accord.local.StoreParticipants;
import accord.local.TimeService;
import accord.local.durability.DurabilityService;
@ -82,7 +83,6 @@ import accord.topology.TopologyManager;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.Cancellable;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.concurrent.ExecutorPlus;
@ -104,6 +104,7 @@ import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor;
import org.apache.cassandra.service.accord.AccordExecutor.IOTask;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.service.accord.journal.AccordJournal;
@ -114,7 +115,6 @@ import org.apache.cassandra.service.accord.txn.TxnQuery;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Condition;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static accord.primitives.Routable.Domain.Key;
@ -123,6 +123,7 @@ import static accord.primitives.SaveStatus.PreAccepted;
import static accord.primitives.Status.Durability.NotDurable;
import static accord.primitives.Txn.Kind.Write;
import static java.lang.String.format;
import static org.apache.cassandra.service.accord.AccordCacheEntry.LockMode.RELEASE_QUEUE;
import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITH_LOCK;
import static org.apache.cassandra.service.accord.AccordService.getBlocking;
@ -162,16 +163,16 @@ public class AccordTestUtils
}
}
public static <K, V> AccordCacheEntry<K, V> loaded(K key, V value)
public static <K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> AccordCacheEntry<K, V, S> loaded(K key, V value)
{
AccordCacheEntry<K, V> global = new AccordCacheEntry<>(key, null);
AccordCacheEntry<K, V, S> global = new AccordCacheEntry<>(key, null);
global.initialize(value);
return global;
}
public static AccordSafeCommand safeCommand(Command command)
{
AccordCacheEntry<TxnId, Command> global = loaded(command.txnId(), command);
AccordCacheEntry<TxnId, Command, AccordSafeCommand> global = loaded(command.txnId(), command);
return new AccordSafeCommand(global);
}
@ -188,9 +189,9 @@ public class AccordTestUtils
return new LoadExecutor<>()
{
@Override
public <K, V> Cancellable load(P1 p1, P2 p2, AccordCacheEntry<K, V> entry)
public <K, V> IOTask load(P1 p1, P2 p2, AccordCacheEntry<K, V, ?> entry)
{
Future<?> future = executor.submit(() -> {
executor.submit(() -> {
V v;
try { v = entry.owner.parent().adapter().load(entry.owner.commandStore, entry.key()); }
catch (Throwable t)
@ -200,19 +201,19 @@ public class AccordTestUtils
}
entry.loaded(v);
});
return () -> future.cancel(true);
return null;
}
};
}
public static <K, V> void testLoad(ManualExecutor executor, AccordSafeState<K, V> safeState, V val)
public static <K, V, S extends SafeState<V> & AccordSafeState<K, V, S>> void testLoad(ManualExecutor executor, S safeState, V val)
{
Assert.assertEquals(AccordCacheEntry.Status.WAITING_TO_LOAD, safeState.global().status());
safeState.global().load(loadExecutor(executor), null, null);
Assert.assertEquals(AccordCacheEntry.Status.LOADING, safeState.global().status());
executor.runOne();
Assert.assertEquals(AccordCacheEntry.Status.LOADED, safeState.global().status());
safeState.preExecute();
safeState.preExecute(new AccordTask<>(null, (ExecutionContext.Empty)() -> "Test", null), RELEASE_QUEUE);
Assert.assertEquals(val, safeState.current());
}

2
test/unit/org/apache/cassandra/service/accord/SimpleSimulatedAccordCommandStoreTest.java
View File

@ -41,7 +41,7 @@ public class SimpleSimulatedAccordCommandStoreTest extends SimulatedAccordComman
for (int i = 0, examples = 100; i < examples; i++)
{
TxnId id = AccordGens.txnIds().next(rs);
instance.process(ExecutionContext.contextFor(id, "Test"), (safe) -> {
instance.process(ExecutionContext.unsequenced(id, "Test"), (safe) -> {
var safeCommand = safe.get(id, StoreParticipants.empty(id));
var command = safeCommand.current();
Assertions.assertThat(command.saveStatus()).isEqualTo(SaveStatus.Uninitialised);

26
test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java
View File

@ -23,6 +23,7 @@ import java.util.concurrent.locks.LockSupport;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.BooleanSupplier;
import java.util.function.Function;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
@ -180,7 +181,10 @@ public class SimulatedAccordTaskTest extends SimulatedAccordCommandStoreTestBase
private static AccordTask<Void> operation(SimulatedAccordCommandStore instance, ExecutionContext ctx, Action action, BooleanSupplier delay)
{
return new SimulatedOperation(instance.commandStore, ctx, action == Action.FAILURE ? SimulatedOperation.Action.FAILURE : SimulatedOperation.Action.SUCCESS);
Function<SafeCommandStore, Void> function = action == Action.FAILURE ? safeStore -> { throw new SimulatedFault("Operation failed for keys " + ctx.keys()); }
: safeStore -> null;
return AccordTask.create(instance.commandStore, ctx, function);
}
private static class Counter implements BiConsumer<Object, Throwable>
@ -196,26 +200,6 @@ public class SimulatedAccordTaskTest extends SimulatedAccordCommandStoreTestBase
}
}
private static class SimulatedOperation extends AccordTask<Void>
{
enum Action { SUCCESS, FAILURE}
private final Action action;
public SimulatedOperation(AccordCommandStore commandStore, ExecutionContext executionContext, Action action)
{
super(commandStore, executionContext);
this.action = action;
}
@Override
public Void apply(SafeCommandStore safe)
{
if (action == Action.FAILURE)
throw new SimulatedFault("Operation failed for keys " + keys());
return null;
}
}
private static class SimulatedLoadFunctionWrapper implements FunctionWrapper
{
final Supplier<Action> actions;

32
test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
View File

@ -505,8 +505,8 @@ public class CommandsForKeySerializerTest
{
int next = source.nextInt(commands.size());
Command command = commands.get(next);
if (command.txnId.isSyncPoint()) cfk = cfk.registerUnmanaged(new TestSafeCommandStore(ExecutionContext.contextFor(command.txnId(), "Test")), new TestSafeCommand(command), REGISTER).cfk();
else cfk = cfk.update(new TestSafeCommandStore(ExecutionContext.contextFor(command.txnId(), "Test")), command).cfk();
if (command.txnId.isSyncPoint()) cfk = cfk.registerUnmanaged(new TestSafeCommandStore(ExecutionContext.unsequenced(command.txnId(), "Test")), new TestSafeCommand(command), REGISTER).cfk();
else cfk = cfk.update(new TestSafeCommandStore(ExecutionContext.unsequenced(command.txnId(), "Test")), command).cfk();
commands.set(next, commands.get(commands.size() - 1));
commands.remove(commands.size() - 1);
}
@ -547,33 +547,10 @@ public class CommandsForKeySerializerTest
static class TestSafeCommand extends SafeCommand
{
final Command command;
TestSafeCommand(Command command)
{
super(command.txnId);
this.command = command;
}
@Override
public Command current()
{
return command;
}
@Override
public void markUnsafe()
{
}
@Override
public boolean isUnsafe()
{
return false;
}
@Override
protected void set(Command command)
{
current = command;
}
}
@ -704,7 +681,7 @@ public class CommandsForKeySerializerTest
{
public TestSafeCommandStore(ExecutionContext context)
{
super(context, TestCommandStore.INSTANCE);
super(context);
}
@Override protected CommandStoreCaches tryGetCaches() { return null; }
@ -712,6 +689,7 @@ public class CommandsForKeySerializerTest
@Override protected SafeCommandsForKey add(SafeCommandsForKey safeCfk, CommandStoreCaches caches) { return null; }
@Override protected SafeCommand getInternal(TxnId txnId) { return null; }
@Override protected SafeCommandsForKey getInternal(RoutingKey key) { return null; }
@Override public CommandStore commandStore() { return TestCommandStore.INSTANCE; }
@Override public DataStore dataStore() { return null; }
@Override public Agent agent() { return null; }
@Override public ProgressLog progressLog() { return null; }