Limit replay to those records not durably persisted to both command store and data store

Also fix:
 - Limit truncation to TruncatedApplyWithOutcome until data is persisted durably to local store
 - IntervalUpdater not invoking super.close()
 - Do not invoke preRunExclusive without holding lock
 - IntervalUpdater not correctly initialise BranchBuilder.inUse
 - AccordExecutor should notify if more work on unlock of caches
 - Relax paranoid CFK validation during restart
Also improve:
 - Flush logs before System.exit
 - Start/stop progress log explicitly
 - Limit progress log concurrency
 - Clear heavy fields in some messages once processed

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20780

Author: belliottsmith

modules/accord

@@ -57,14 +57,11 @@ public enum QueueShardModel
         /**
          * Same number of threads as queue shards, but the shard lock is held only while managing the queue,
          * so that submitting threads may queue load/save work.
-         *
-         * The global READ and WRITE stages are used for IO.
          */
         THREAD_PER_SHARD,
 
         /**
          * Same number of threads as shards, and the shard lock is held for the duration of serving requests.
-         * The global READ and WRITE stages are used for IO.
          */
         THREAD_PER_SHARD_SYNC_QUEUE,
 
@@ -73,12 +70,6 @@ public enum QueueShardModel
          * Fewer shards is generally better, until queue-contention is encountered.
          */
         THREAD_POOL_PER_SHARD,
-
-        /**
-         * More threads than shards. Threads update transaction state only, relying on READ and WRITE stages for IO.
-         * Fewer shards is generally better, until queue-contention is encountered.
-         */
-        THREAD_POOL_PER_SHARD_EXCLUDES_IO,
     }
 
     public enum QueueSubmissionModel
@@ -130,6 +121,7 @@ public enum QueueSubmissionModel
 
     public volatile OptionaldPositiveInt max_queued_loads = OptionaldPositiveInt.UNDEFINED;
     public volatile OptionaldPositiveInt max_queued_range_loads = OptionaldPositiveInt.UNDEFINED;
+    public volatile OptionaldPositiveInt max_progress_log_concurrency = OptionaldPositiveInt.UNDEFINED;
 
     public DataStorageSpec.LongMebibytesBound cache_size = null;
     public DataStorageSpec.LongMebibytesBound working_set_size = null;
@@ -158,8 +150,8 @@ public enum QueueSubmissionModel
 
     public volatile DurationSpec.IntSecondsBound fast_path_update_delay = null;
 
-    public volatile DurationSpec.IntSecondsBound gc_delay = new DurationSpec.IntSecondsBound("5m");
     public volatile int shard_durability_target_splits = 16;
+    public volatile int shard_durability_max_splits = 128;
     public volatile DurationSpec.IntSecondsBound durability_txnid_lag = new DurationSpec.IntSecondsBound(5);
     public volatile DurationSpec.IntSecondsBound shard_durability_cycle = new DurationSpec.IntSecondsBound(5, TimeUnit.MINUTES);
     public volatile DurationSpec.IntSecondsBound global_durability_cycle = new DurationSpec.IntSecondsBound(5, TimeUnit.MINUTES);

src/java/org/apache/cassandra/config/AccordSpec.java

@@ -5372,7 +5372,6 @@ public static int getAccordQueueShardCount()
             case THREAD_PER_SHARD_SYNC_QUEUE:
                 return conf.accord.queue_shard_count.or(DatabaseDescriptor::getAvailableProcessors);
             case THREAD_POOL_PER_SHARD:
-            case THREAD_POOL_PER_SHARD_EXCLUDES_IO:
                 int defaultMax = getAccordQueueSubmissionModel() == AccordSpec.QueueSubmissionModel.SYNC ? 8 : 4;
                 return conf.accord.queue_shard_count.or(Math.min(defaultMax, DatabaseDescriptor.getAvailableProcessors()));
         }
@@ -5393,6 +5392,11 @@ public static int getAccordMaxQueuedRangeLoadCount()
         return conf.accord.max_queued_range_loads.or(Math.max(4, getAccordConcurrentOps() / 4));
     }
 
+    public static int getAccordProgressLogMaxConcurrency()
+    {
+        return conf.accord.max_progress_log_concurrency.or(64);
+    }
+
     public static boolean getAccordCacheShrinkingOn()
     {
         return conf.accord.shrink_cache_entries_before_eviction;
@@ -5426,14 +5430,14 @@ public static long getAccordFastPathUpdateDelayMillis()
         return bound == null ? -1 : bound.to(TimeUnit.MILLISECONDS);
     }
 
-    public static long getAccordGCDelay(TimeUnit unit)
+    public static int getAccordShardDurabilityTargetSplits()
     {
-        return conf.accord.gc_delay.to(unit);
+        return conf.accord.shard_durability_target_splits;
     }
 
-    public static int getAccordShardDurabilityTargetSplits()
+    public static int getAccordShardDurabilityMaxSplits()
     {
-        return conf.accord.shard_durability_target_splits;
+        return conf.accord.shard_durability_max_splits;
     }
 
     public static long getAccordScheduleDurabilityTxnIdLag(TimeUnit unit)

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -1374,6 +1374,7 @@ public Collection<SSTableReader> flushMemtable(ColumnFamilyStore cfs, Memtable m
                 }
             }
             cfs.replaceFlushed(memtable, sstables);
+            memtable.notifyFlushed();
             reclaim(memtable);
             cfs.compactionStrategyManager.compactionLogger.flush(sstables);
             logger.debug("Flushed to {} ({} sstables, {}), biggest {}, smallest {}",

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -18,15 +18,19 @@
 
 package org.apache.cassandra.db.memtable;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentSkipListSet;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
 
 import org.apache.cassandra.db.RegularAndStaticColumns;
 import org.apache.cassandra.db.commitlog.CommitLogPosition;
@@ -51,6 +55,7 @@ public abstract class AbstractMemtable implements Memtable
     // The smallest local deletion time for all partitions in this memtable
     protected AtomicLong minLocalDeletionTime = new AtomicLong(Long.MAX_VALUE);
     private final long id = nextId.incrementAndGet();
+    private Map<Object, Consumer<TableMetadata>> onFlush = ImmutableMap.of();
     // Note: statsCollector has corresponding statistics to the two above, but starts with an epoch value which is not
     // correct for their usage.
 
@@ -147,6 +152,29 @@ public LifecycleTransaction setFlushTransaction(LifecycleTransaction flushTransa
         return this.flushTransaction.getAndSet(flushTransaction);
     }
 
+    @Override
+    public synchronized <T extends Consumer<TableMetadata>> T ensureFlushListener(Object key, Supplier<T> factory)
+    {
+        if (onFlush == null)
+            return null;
+
+        T listener;
+        if (null == (listener = (T)onFlush.get(key)))
+            onFlush = ImmutableMap.<Object, Consumer<TableMetadata>>builder().putAll(onFlush).put(key, listener = factory.get()).build();
+        return listener;
+    }
+
+    public void notifyFlushed()
+    {
+        Collection<Consumer<TableMetadata>> run;
+        synchronized (this)
+        {
+            run = onFlush.values();
+            onFlush = null;
+        }
+        run.forEach(c -> c.accept(metadata()));
+    }
+
     protected static class ColumnsCollector
     {
         private final HashMap<ColumnMetadata, AtomicBoolean> predefined = new HashMap<>();

src/java/org/apache/cassandra/db/memtable/AbstractMemtable.java

@@ -19,6 +19,8 @@
 package org.apache.cassandra.db.memtable;
 
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
 import javax.annotation.concurrent.NotThreadSafe;
 
 import org.apache.cassandra.db.ColumnFamilyStore;
@@ -420,6 +422,10 @@ default boolean shouldSwitch(ColumnFamilyStore.FlushReason reason)
         return shouldSwitch(reason, metadata());
     }
 
+    // returns null if already flushed
+    <T extends Consumer<TableMetadata>> T ensureFlushListener(Object key, Supplier<T> factory);
+    void notifyFlushed();
+
     /**
      * Called when the table's metadata is updated. The memtable's metadata reference now points to the new version.
      * This will not be called if {@link #shouldSwitch } (SCHEMA_CHANGE) returns true, as the memtable will be swapped out

src/java/org/apache/cassandra/db/memtable/Memtable.java

@@ -52,14 +52,14 @@
 import accord.utils.Invariants;
 import accord.utils.QuadFunction;
 import accord.utils.TriFunction;
-import accord.utils.async.Cancellable;
 import org.agrona.collections.Object2ObjectHashMap;
 import org.apache.cassandra.cache.CacheSize;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.exceptions.UnknownTableException;
 import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.metrics.AccordCacheMetrics;
 import org.apache.cassandra.metrics.CacheAccessMetrics;
+import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor;
 import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
 import org.apache.cassandra.service.accord.events.CacheEvents;
 import org.apache.cassandra.service.accord.serializers.Version;
@@ -72,6 +72,8 @@
 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;
+import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.SAVING;
+import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.WAITING_TO_SAVE;
 
 /**
  * Cache for AccordCommand and AccordCommandsForKey, available memory is shared between the two object types.
@@ -100,6 +102,7 @@ public interface Adapter<K, V, S>
     {
         @Nullable V load(AccordCommandStore commandStore, K key);
         @Nullable Runnable save(AccordCommandStore commandStore, K key, @Nullable V value, @Nullable Object shrunk);
+        default boolean canSave(@Nullable V value, @Nullable Object shrunk) { return true; }
         // a result of null means we can immediately evict, without saving
         @Nullable V quickShrink(V value);
         // a result of null means we cannot shrink, and should save/evict as appropriate
@@ -138,9 +141,7 @@ public ImmutableStats(Stats stats)
     }
 
     private final List<Type<?, ?, ?>> types = new CopyOnWriteArrayList<>();
-    private final Function<Runnable, Cancellable> saveExecutor;
-    private final AccordCacheEntry.OnSaved onSaved;
-    // TODO (required): monitor this queue and periodically clean up entries, or implement an eviction deadline system
+    final AccordCacheEntry.SaveExecutor saveExecutor;
     private final IntrusiveLinkedList<AccordCacheEntry<?,?>> evictQueue = new IntrusiveLinkedList<>();
     private final IntrusiveLinkedList<AccordCacheEntry<?,?>> noEvictQueue = new IntrusiveLinkedList<>();
 
@@ -155,10 +156,9 @@ public ImmutableStats(Stats stats)
     final AccordCacheMetrics metrics;
     final Stats stats = new Stats();
 
-    public AccordCache(Function<Runnable, Cancellable> saveExecutor, AccordCacheEntry.OnSaved onSaved, long maxSizeInBytes, AccordCacheMetrics metrics)
+    public AccordCache(AccordCacheEntry.SaveExecutor saveExecutor, long maxSizeInBytes, AccordCacheMetrics metrics)
     {
         this.saveExecutor = saveExecutor;
-        this.onSaved = onSaved;
         this.maxSizeInBytes = maxSizeInBytes;
         this.metrics = metrics;
     }
@@ -265,14 +265,21 @@ public <K, V> void tryEvict(AccordCacheEntry<K, V> node)
                 evict(node, true);
                 break;
             case MODIFIED:
-                Type<K, V, ?> parent = node.owner.parent();
-                node.save(saveExecutor, parent.adapter, onSaved);
+                node.save();
                 boolean evict = node.status() == LOADED;
                 node.unlink();
                 if (evict) evict(node, true);
         }
     }
 
+    public void saveWhenReadyExclusive(AccordCacheEntry<?, ?> entry, Runnable onSuccess)
+    {
+        if (!entry.isSavingOrWaiting() && !entry.saveWhenReady())
+            onSuccess.run();
+        else
+            entry.savingOrWaitingToSave().identity.onSuccess(onSuccess);
+    }
+
     private void evict(AccordCacheEntry<?, ?> node, boolean updateUnreferenced)
     {
         if (logger.isTraceEnabled())
@@ -302,10 +309,10 @@ private void evict(AccordCacheEntry<?, ?> node, boolean updateUnreferenced)
         node.evicted();
     }
 
-    <P, K, V> Collection<AccordTask<?>> load(BiFunction<P, Runnable, Cancellable> loadExecutor, P param, AccordCacheEntry<K, V> node, AccordCacheEntry.OnLoaded onLoaded)
+    <P1, P2, K, V> Collection<AccordTask<?>> load(LoadExecutor<P1, P2> loadExecutor, P1 p1, P2 p2, AccordCacheEntry<K, V> node)
     {
         Type<K, V, ?> parent = node.owner.parent();
-        return node.load(loadExecutor, param, parent.adapter, onLoaded).waiters();
+        return node.load(loadExecutor, p1, p2).waiters();
     }
 
     <K, V> void loaded(AccordCacheEntry<K, V> node, V value)
@@ -329,7 +336,7 @@ <K, V> void failedToLoad(AccordCacheEntry<K, V> node)
 
     <K, V> void saved(AccordCacheEntry<K, V> node, Object identity, Throwable fail)
     {
-        if (node.saved(identity, fail) && node.references() == 0)
+        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
     }
 
@@ -539,6 +546,7 @@ else if (node.isLoadingOrWaiting())
                     switch (status)
                     {
                         default: throw new IllegalStateException("Unhandled status " + status);
+                        case WAITING_TO_SAVE:
                         case WAITING_TO_LOAD:
                         case LOADING:
                         case LOADED:
@@ -560,15 +568,15 @@ public Stream<AccordCacheEntry<K, V>> stream()
                 return cache.values().stream();
             }
 
-            Type<K, V, S> parent()
+            final Type<K, V, S> parent()
             {
                 return Type.this;
             }
 
             @Override
             public Iterator<AccordCacheEntry<K, V>> iterator()
             {
-                return stream().iterator();
+                return cache.values().iterator();
             }
 
             void validateLoadEvicted(AccordCacheEntry<?, ?> node)
@@ -672,7 +680,6 @@ public boolean tryUnregister(Listener<K, V> l)
                     listeners = null;
                 return true;
             }
-
         }
 
         private final Class<K> keyClass;
@@ -733,7 +740,7 @@ private void incrementCacheMisses()
             AccordCache.this.stats.misses++;
         }
 
-        AccordCache parent()
+        final AccordCache parent()
         {
             return AccordCache.this;
         }
@@ -768,7 +775,7 @@ public BiFunction<AccordCommandStore, K, V> unsafeGetLoadFunction()
             return ((SettableWrapper<K, V, S>)adapter).load;
         }
 
-        Adapter<K, V, S> adapter()
+        final Adapter<K, V, S> adapter()
         {
             return adapter;
         }
@@ -1120,6 +1127,12 @@ public Runnable save(AccordCommandStore commandStore, RoutingKey key, @Nullable
             return commandStore.saveCommandsForKey(key, value, serialized);
         }
 
+        @Override
+        public boolean canSave(@Nullable CommandsForKey value, @Nullable Object serialized)
+        {
+            return value == null || !value.isLoadingPruned();
+        }
+
         @Override
         public CommandsForKey quickShrink(CommandsForKey value)
         {