Merge branch 'cassandra-6.0' into trunk

Author: belliottsmith

.build/checkstyle.xml

@@ -63,6 +63,12 @@
        <property name="influenceFormat" value="0"/>
     </module>
 
+    <module name="SuppressWithNearbyCommentFilter">
+      <property name="commentFormat" value="checkstyle: permit this invocation"/>
+      <property name="idFormat" value="blockInstantNow"/>
+      <property name="influenceFormat" value="0"/>
+    </module>
+
     <module name="RegexpSinglelineJava">
       <!-- block system time -->
       <property name="id" value="blockSystemClock"/>

modules/accord

@@ -66,11 +66,15 @@ 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.
+         *
+         * This is incompatible with QueueSubmissionModel.SIGNAL
          */
         THREAD_PER_SHARD,
 
         /**
          * Same number of threads as shards, and the shard lock is held for the duration of serving requests.
+         *
+         * This is incompatible with QueueSubmissionModel.SIGNAL
          */
         THREAD_PER_SHARD_SYNC_QUEUE,
 
@@ -103,6 +107,14 @@ public enum QueueSubmissionModel
          */
         ASYNC,
 
+        /**
+         * Queue workers try to avoid competing for the lock, with the lock owner distributing work to any waiting threads
+         * and signalling them without them taking the lock
+         *
+         * NOTE: EXPERIMENTAL
+         */
+        SIGNAL,
+
         /**
          * The queue is backed by submission to a single-threaded plain executor.
          * This implementation does not honor the sharding model option.
@@ -151,8 +163,30 @@ public enum QueuePriorityModel
      */
     public volatile OptionaldPositiveInt queue_shard_count = OptionaldPositiveInt.UNDEFINED;
 
+    /**
+     * The total number of threads to share between queue shards
+     */
+    public volatile OptionaldPositiveInt queue_thread_count = OptionaldPositiveInt.UNDEFINED;
+
     public QueuePriorityModel queue_priority_model = HLC_FIFO;
 
+    /**
+     * If set, the signal loop does not match park/unpark pairs, but instead consumers perform timed-park spin waits
+     */
+    public DurationSpec.LongMicrosecondsBound queue_spin_interval;
+
+    /**
+     * If set, the signal loop reduces the number of threads it is using when the time spent parked exceeds real-time
+     * by this interval.
+     */
+    public DurationSpec.LongMicrosecondsBound queue_stop_check_interval;
+
+    /**
+     * If set, the signal loop reduces the number of threads it is using when the time spent parked exceeds real-time
+     * by this interval.
+     */
+    public DurationSpec.LongMicrosecondsBound queue_signal_stop_check_interval_credit;
+
     // yield to other executor threads after executing this many tasks in a row, if there are waiting threads and tasks
     public int queue_yield_interval = 100;
 
@@ -456,4 +490,9 @@ public int userVersion()
             return version.version;
         }
     }
+
+    public int commandStoreShardCount()
+    {
+        return command_store_shard_count.or(DatabaseDescriptor::getAvailableProcessors);
+    }
 }

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

@@ -203,7 +203,6 @@ public static Set<String> splitCommaDelimited(String src)
 
     public int concurrent_reads = 32;
     public int concurrent_writes = 32;
-    public int concurrent_accord_operations = 32;
     public int concurrent_counter_writes = 32;
     public int concurrent_materialized_view_writes = 32;
     public OptionaldPositiveInt available_processors = new OptionaldPositiveInt(CASSANDRA_AVAILABLE_PROCESSORS.getInt(OptionaldPositiveInt.UNDEFINED_VALUE));

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

@@ -700,9 +700,6 @@ else if (DiskAccessMode.direct == conf.compaction_read_disk_access_mode)
         if (conf.concurrent_counter_writes < 2)
             throw new ConfigurationException("concurrent_counter_writes must be at least 2, but was " + conf.concurrent_counter_writes, false);
 
-        if (conf.concurrent_accord_operations < 1)
-            throw new ConfigurationException("concurrent_accord_operations must be at least 1, but was " + conf.concurrent_accord_operations, false);
-
         if (conf.networking_cache_size == null)
             conf.networking_cache_size = new DataStorageSpec.IntMebibytesBound(Math.min(128, (int) (Runtime.getRuntime().maxMemory() / (16 * 1048576))));
 
@@ -2900,7 +2897,7 @@ public static void setConcurrentViewWriters(int concurrent_materialized_view_wri
 
     public static int getAccordConcurrentOps()
     {
-        return conf.concurrent_accord_operations;
+        return conf.accord.queue_thread_count.or(2 * FBUtilities.getAvailableProcessors());
     }
 
     public static void setConcurrentAccordOps(int concurrent_operations)
@@ -2909,7 +2906,7 @@ public static void setConcurrentAccordOps(int concurrent_operations)
         {
             throw new IllegalArgumentException("Concurrent accord operations must be non-negative");
         }
-        conf.concurrent_accord_operations = concurrent_operations;
+        conf.accord.queue_thread_count = new OptionaldPositiveInt(concurrent_operations);
     }
 
     public static int getFlushWriters()
@@ -5635,6 +5632,7 @@ public static AccordConfig getAccord()
         return conf.accord;
     }
 
+    // TODO (expected): move all getAccordX into AccordConfig
     public static AccordConfig.TransactionalRangeMigration getTransactionalRangeMigration()
     {
         return conf.accord.range_migration;
@@ -5665,44 +5663,6 @@ public static void setAccordTransactionsEnabled(boolean b)
         conf.accord.enabled = b;
     }
 
-    public static AccordConfig.QueueShardModel getAccordQueueShardModel()
-    {
-        return conf.accord.queue_shard_model;
-    }
-
-    public static AccordConfig.QueueSubmissionModel getAccordQueueSubmissionModel()
-    {
-        return conf.accord.queue_submission_model;
-    }
-
-    public static int getAccordQueueShardCount()
-    {
-        switch (getAccordQueueShardModel())
-        {
-            default: throw new AssertionError("Unhandled queue_shard_model: " + conf.accord.queue_shard_model);
-            case THREAD_PER_SHARD:
-            case THREAD_PER_SHARD_SYNC_QUEUE:
-                return conf.accord.queue_shard_count.or(DatabaseDescriptor::getAvailableProcessors);
-            case THREAD_POOL_PER_SHARD:
-                return conf.accord.queue_shard_count.or(DatabaseDescriptor.getAvailableProcessors()/4);
-        }
-    }
-
-    public static int getAccordCommandStoreShardCount()
-    {
-        return conf.accord.command_store_shard_count.or(DatabaseDescriptor::getAvailableProcessors);
-    }
-
-    public static int getAccordMaxQueuedLoadCount()
-    {
-        return conf.accord.max_queued_loads.or(getAccordConcurrentOps());
-    }
-
-    public static int getAccordMaxQueuedRangeLoadCount()
-    {
-        return conf.accord.max_queued_range_loads.or(Math.max(4, getAccordConcurrentOps() / 4));
-    }
-
     public static DefaultProgressLog.Config getAccordProgressLogConfig()
     {
         return accordProgressLogConfig;

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

@@ -245,26 +245,6 @@ public AccordCommandStore(int id,
         if (this.progressLog instanceof DefaultProgressLog)
             ((DefaultProgressLog)this.progressLog).unsafeSetConfig(DatabaseDescriptor.getAccordProgressLogConfig());
 
-        final AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands;
-        final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKey;
-        try (AccordExecutor.ExclusiveGlobalCaches exclusive = sharedExecutor.lockCaches())
-        {
-            commands = exclusive.commands.newInstance(this);
-            commandsForKey = exclusive.commandsForKey.newInstance(this);
-            this.caches = new ExclusiveCaches(sharedExecutor.unsafeLock(), exclusive.global, commands, commandsForKey);
-        }
-
-        this.exclusiveExecutor = sharedExecutor.executor(id);
-        {
-            AccordConfig.RangeIndexMode mode = getAccord().range_index_mode;
-            switch (mode)
-            {
-                default: throw new UnhandledEnum(mode);
-                case journal_sai: rangeIndex = new JournalRangeIndex(this); break;
-                case in_memory: rangeIndex = new InMemoryRangeIndex(this); break;
-            }
-        }
-
         maybeLoadRedundantBefore(journal.loadRedundantBefore(id()));
         maybeLoadBootstrapBeganAt(journal.loadBootstrapBeganAt(id()));
         maybeLoadSafeToRead(journal.loadSafeToRead(id()));
@@ -284,6 +264,26 @@ public AccordCommandStore(int id,
             return a;
         }).orElseThrow(() -> Invariants.illegalState("CommandStore %d created with no ranges", id));
 
+        final AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands;
+        final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKey;
+        try (AccordExecutor.ExclusiveGlobalCaches exclusive = sharedExecutor.lockCaches())
+        {
+            commands = exclusive.commands.newInstance(this);
+            commandsForKey = exclusive.commandsForKey.newInstance(this);
+            this.caches = new ExclusiveCaches(sharedExecutor.unsafeLock(), exclusive.global, commands, commandsForKey);
+        }
+        this.exclusiveExecutor = sharedExecutor.executor(id);
+
+        {
+            AccordConfig.RangeIndexMode mode = getAccord().range_index_mode;
+            switch (mode)
+            {
+                default: throw new UnhandledEnum(mode);
+                case journal_sai: rangeIndex = new JournalRangeIndex(this); break;
+                case in_memory: rangeIndex = new InMemoryRangeIndex(this); break;
+            }
+        }
+
         if (AccordService.isStarted())
             progressLog.unsafeStart();
     }

src/java/org/apache/cassandra/service/accord/AccordCommandStore.java

@@ -48,16 +48,17 @@
 import org.apache.cassandra.cache.CacheSize;
 import org.apache.cassandra.concurrent.ScheduledExecutors;
 import org.apache.cassandra.concurrent.Shutdownable;
+import org.apache.cassandra.config.AccordConfig;
 import org.apache.cassandra.config.AccordConfig.QueueShardModel;
 import org.apache.cassandra.config.DatabaseDescriptor;
 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.getAccordQueueShardCount;
-import static org.apache.cassandra.config.DatabaseDescriptor.getAccordQueueSubmissionModel;
+import static org.apache.cassandra.config.DatabaseDescriptor.getAccord;
 import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITHOUT_LOCK;
 import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITH_LOCK;
 import static org.apache.cassandra.service.accord.AccordExecutor.constant;
@@ -83,10 +84,12 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
               AccordCommandStore.factory(id -> executors[id % executors.length]));
         this.executors = executors;
         this.mask = Integer.highestOneBit(executors.length) - 1;
+
         cacheSize = DatabaseDescriptor.getAccordCacheSizeInMiB() << 20;
         workingSetSize = DatabaseDescriptor.getAccordWorkingSetSizeInMiB() << 20;
-        maxQueuedLoads = DatabaseDescriptor.getAccordMaxQueuedLoadCount();
-        maxQueuedRangeLoads = DatabaseDescriptor.getAccordMaxQueuedRangeLoadCount();
+        AccordConfig config = DatabaseDescriptor.getAccord();
+        maxQueuedLoads = maxQueuedLoads(config);
+        maxQueuedRangeLoads = maxQueuedRangeLoads(config);
         shrinkingOn = DatabaseDescriptor.getAccordCacheShrinkingOn();
         refreshCapacities();
         ScheduledExecutors.scheduledFastTasks.scheduleWithFixedDelay(() -> {
@@ -102,15 +105,21 @@ public class AccordCommandStores extends CommandStores implements CacheSize, Shu
     static Factory factory()
     {
         return (NodeCommandStoreService time, Agent agent, DataStore store, RandomSource random, Journal journal, ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, LocalListeners.Factory listenersFactory) -> {
-            AccordExecutor[] executors = new AccordExecutor[getAccordQueueShardCount()];
+            AccordConfig config = getAccord();
+            AccordExecutor[] executors = new AccordExecutor[executorShards(config)];
             AccordExecutorFactory factory;
             int maxThreads = Integer.MAX_VALUE;
-            switch (getAccordQueueSubmissionModel())
+            switch (config.queue_submission_model)
             {
-                default: throw new AssertionError("Unhandled QueueSubmissionModel: " + getAccordQueueSubmissionModel());
+                default: throw new AssertionError("Unhandled QueueSubmissionModel: " + config.queue_submission_model);
                 case SYNC: factory = AccordExecutorSyncSubmit::new; break;
                 case SEMI_SYNC: factory = AccordExecutorSemiSyncSubmit::new; break;
                 case ASYNC: factory = AccordExecutorAsyncSubmit::new; break;
+                case SIGNAL:
+                    long spinIntervalNanos = config.queue_spin_interval == null ? 0 : config.queue_spin_interval.to(TimeUnit.NANOSECONDS);
+                    long stopCheckIntervalNanos = config.queue_stop_check_interval == null ? 0 : config.queue_stop_check_interval.to(TimeUnit.NANOSECONDS);
+                    factory = (executorId, mode, threads, name, agent0) -> new AccordExecutorSignalLoop(executorId, mode, threads, spinIntervalNanos, stopCheckIntervalNanos, TimeUnit.NANOSECONDS, name, agent0);
+                    break;
                 case EXEC_ST:
                     factory = AccordExecutorSimple::new;
                     maxThreads = 1;
@@ -119,16 +128,16 @@ static Factory factory()
 
             for (int id = 0; id < executors.length; id++)
             {
-                QueueShardModel shardModel = DatabaseDescriptor.getAccordQueueShardModel();
+                QueueShardModel shardModel = config.queue_shard_model;
                 String baseName = AccordExecutor.class.getSimpleName() + '[' + id;
-                int threads = Math.min(maxThreads, Math.max(DatabaseDescriptor.getAccordConcurrentOps() / getAccordQueueShardCount(), 1));
                 switch (shardModel)
                 {
                     case THREAD_PER_SHARD:
                     case THREAD_PER_SHARD_SYNC_QUEUE:
                         executors[id] = factory.get(id, shardModel == THREAD_PER_SHARD ? RUN_WITHOUT_LOCK : RUN_WITH_LOCK, 1, constant(baseName + ']'), agent);
                         break;
                     case THREAD_POOL_PER_SHARD:
+                        int threads = Math.min(maxThreads, Math.max(DatabaseDescriptor.getAccordConcurrentOps() / executors.length, 1));
                         executors[id] = factory.get(id, RUN_WITHOUT_LOCK, threads, num -> baseName + ',' + num + ']', agent);
                         break;
                 }
@@ -204,8 +213,8 @@ synchronized void refreshCapacities()
     {
         long capacityPerExecutor = cacheSize / executors.length;
         long workingSetPerExecutor = workingSetSize < 0 ? Long.MAX_VALUE : workingSetSize / executors.length;
-        int maxLoadsPerExecutor = (maxQueuedLoads + executors.length - 1) / executors.length;
-        int maxRangeLoadsPerExecutor = (maxQueuedRangeLoads + executors.length - 1) / 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(() -> {
@@ -319,4 +328,34 @@ public AsyncChain<List<Map.Entry<Integer, Long>>> restoreState()
         }
         return AsyncChains.allOf(chains);
     }
+
+    private static int executorShards(AccordConfig config)
+    {
+        switch (config.queue_shard_model)
+        {
+            default: throw new AssertionError("Unhandled queue_shard_model: " + config.queue_shard_model);
+            case THREAD_PER_SHARD:
+            case THREAD_PER_SHARD_SYNC_QUEUE:
+                return config.queue_shard_count.or(DatabaseDescriptor::getAvailableProcessors);
+            case THREAD_POOL_PER_SHARD:
+                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);
+    }
+
+
 }