Replay nondurable

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

@@ -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/DatabaseDescriptor.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/db/ColumnFamilyStore.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/compaction/CompactionTask.java

@@ -30,6 +30,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.RateLimiter;
+
 import org.apache.cassandra.db.compaction.unified.UnifiedCompactionTask;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;

src/java/org/apache/cassandra/db/memtable/AbstractMemtable.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,35 @@ 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 = (T)onFlush.get(key);
+        if (null == listener)
+        {
+            listener = factory.get();
+            onFlush = ImmutableMap.<Object, Consumer<TableMetadata>>builder()
+                                  .putAll(onFlush)
+                                  .put(key, listener)
+                                  .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/Memtable.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