Improve Tail Latency

 - TransactionStatement token aware routing
 - Mitigate convoy effect:
   - ExecuteTxnBacklog v1 can execute single key transaction backlogs and disseminate the result
     - Direct local execution possible for single key transactions
   - Use shardAppliedBefore instead of gcBefore to cleanup CFK faster
Also Improve:
 - Introduce distributed tracing
 - Combine RejectBefore and MaxConflicts
 - Self-addressed messages are delivered directly
 - Disable AccordSyncPropagator by default, as incompatible with large clusters
 - FastPathStrategy.SIMPLE should not modify fast path contents; FastPathStrategy.UP introduced to adopt this behaviour
 - Configurable queue prioritisation
 - ProtocolModifiers mostly final
 - Remove unnecessary condition check when serializing TxnUpdate
Also Fix:
 - SequentialExecutor ownership bug
 - BTree.XUpdater.reset()

patch by Benedict; reviewed by Alex Petrov and Ariel Weisberg for CASSANDRA-21361

Author: belliottsmith

CHANGES.txt

@@ -1,4 +1,7 @@
 6.0-alpha2
+ * Accord: Tail Latency Improvements (CASSANDRA-21361)
+ * Artificial Latency Injection (CASSANDRA-17024)
+ * Accord: Clean Shutdown/Restart, Rebootstrap, et al (CASSANDRA-21355)
  * Reduce memory allocations in SelectStatement.getQuery (CASSANDRA-21351)
  * Avoid allocation by getFunctions in SelectStatement.authorize (CASSANDRA-21347)
  * Avoid unit conversion in DatabaseDescriptor.getMaxValueSize() for every deserializing Cell (CASSANDRA-21295)

modules/accord

@@ -22,23 +22,28 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 
+import accord.api.ProtocolModifiers.CleanCfkBefore;
+import accord.api.ProtocolModifiers.CoordinatorBacklogExecution;
+import accord.api.ProtocolModifiers.FastExecution;
+import accord.api.ProtocolModifiers.ReplicaExecution;
+import accord.api.ProtocolModifiers.SendStableMessages;
+import accord.primitives.TxnId;
 import accord.utils.Invariants;
 
 import org.apache.cassandra.journal.Params;
 import org.apache.cassandra.service.accord.serializers.Version;
 import org.apache.cassandra.service.consensus.TransactionalMode;
 
-import static org.apache.cassandra.config.AccordSpec.CatchupMode.NORMAL;
-import static org.apache.cassandra.config.AccordSpec.QueueShardModel.THREAD_POOL_PER_SHARD;
-import static org.apache.cassandra.config.AccordSpec.QueueSubmissionModel.SYNC;
-import static org.apache.cassandra.config.AccordSpec.RangeIndexMode.in_memory;
+import static org.apache.cassandra.config.AccordConfig.CatchupMode.NORMAL;
+import static org.apache.cassandra.config.AccordConfig.QueuePriorityModel.HLC_FIFO;
+import static org.apache.cassandra.config.AccordConfig.QueueShardModel.THREAD_POOL_PER_SHARD;
+import static org.apache.cassandra.config.AccordConfig.QueueSubmissionModel.SYNC;
+import static org.apache.cassandra.config.AccordConfig.RangeIndexMode.in_memory;
 
-// TODO (expected): rename to AccordConf?
-public class AccordSpec
+public class AccordConfig
 {
     public volatile boolean enabled = false;
 
-    // TODO (expected): move to JournalSpec
     public volatile String journal_directory;
 
     /**
@@ -107,6 +112,37 @@ public enum QueueSubmissionModel
         EXEC_ST
     }
 
+    public enum QueuePriorityModel
+    {
+        /**
+         * All work is queued on a first-come first-serve basis.
+         * Overload can lead to more rapid degradation, as later phases of the state machine are delayed
+         * by the arrival of new work.
+         */
+        FIFO,
+
+        /**
+         * If the work has an associated TxnId, prioritise by its HLC (and FIFO otherwise)
+         */
+        HLC_FIFO,
+
+        /**
+         * Prioritise Apply, Stable, Commit, Accept, and PreAccept messages in that order.
+         * Within a given message type, prioritise by HLC.
+         * Other messages will be mixed with PreAccept messages, but using the next counter rather than the HLC of the TxnId.
+         * Note: this can have some performance edge cases for contended keys, as we may process Stable messages for later commands before
+         *       we process earlier Accept/Commit, which may delay execution
+         */
+        PHASE_HLC_FIFO,
+
+        /**
+         * Prioritise Apply, Stable, Commit, Accept, and PreAccept messages from the original coordinator only, in that order.
+         * Within a given message type, prioritise by HLC.
+         * Other messages will be mixed with PreAccept messages, but using the next counter rather than the HLC of the TxnId.
+         */
+        ORIG_PHASE_HLC_FIFO
+    }
+
     public QueueShardModel queue_shard_model = THREAD_POOL_PER_SHARD;
     public QueueSubmissionModel queue_submission_model = SYNC;
 
@@ -115,6 +151,15 @@ public enum QueueSubmissionModel
      */
     public volatile OptionaldPositiveInt queue_shard_count = OptionaldPositiveInt.UNDEFINED;
 
+    public QueuePriorityModel queue_priority_model = HLC_FIFO;
+
+    // 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;
+
+    /**
+     * If the HLC is older than this, queue by FIFO instead
+     */
+    public DurationSpec.IntMillisecondsBound queue_priority_age_to_fifo = new DurationSpec.IntMillisecondsBound(500);
     /**
      * The target number of command stores to create per topology shard.
      * This determines the amount of execution parallelism possible for a given table/shard on the host.
@@ -165,6 +210,10 @@ public enum QueueSubmissionModel
     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);
 
+    public volatile DurationSpec.IntSecondsBound topology_watermark_interval = new DurationSpec.IntSecondsBound(60);
+    public volatile boolean topology_sync_propagator_enabled_pre_start = false;
+    public volatile boolean topology_sync_propagator_enabled_post_startup = false;
+
     public enum TransactionalRangeMigration
     {
         auto, explicit
@@ -178,11 +227,6 @@ public enum TransactionalRangeMigration
      */
     public volatile TransactionalRangeMigration range_migration = TransactionalRangeMigration.auto;
 
-    public enum RebootstrapMode
-    {
-        full_repair, truncate_and_stream
-    }
-
     public enum CatchupMode
     {
         DISABLED,
@@ -196,15 +240,43 @@ public enum CatchupMode
      */
     public TransactionalMode default_transactional_mode = TransactionalMode.off;
 
-    public boolean ephemeralReadEnabled = true;
+    // ******** PROTOCOL MODIFIERS ***********
+
+    public CoordinatorBacklogExecution coordinator_backlog_execution;
+    public Boolean permit_local_delivery;
+    public Boolean permit_coordinator_local_execution; // if disabled, the privileged coordinator optimisation will be counter-productive and should also be disabled
+    public TxnId.FastPath permit_fast_path;
+    public Boolean permit_track_stable_medium_path;
+    public Boolean permit_fast_quorum_medium_path;
+    public Boolean always_inform_durable_single_key;
+    public ReplicaExecution replica_execution;
+    public Float replica_execution_distributed_persist_chance;
+    public FastExecution fast_write_execution;
+    public FastExecution fast_read_execution;
+    public CleanCfkBefore clean_cfk_before;
+    public SendStableMessages send_stable;
+    /**
+     * include the least information expected to be necessary in messages -
+     * this is more efficient but may lead to some additional traffic and latency when earlier messages had not arrived
+     */
+    public Boolean send_minimal;
+    // note: simulator incompatible (for now)
+    public Boolean precise_micros;
+
+    public boolean ephemeral_reads = true;
     public boolean state_cache_listener_jfr_enabled = false;
 
     public float hard_reject_ratio = 0.5f;
-    public int min_soft_reject_count = 10;
-    public int max_soft_reject_count = 100;
+    public int min_soft_reject_count = 100;
+    public int max_soft_reject_count = 1000;
     public DurationSpec.LongMicrosecondsBound soft_reject_age = new DurationSpec.LongMicrosecondsBound("10s");
     public DurationSpec.LongMicrosecondsBound soft_reject_cumulative_age = new DurationSpec.LongMicrosecondsBound("60s");
 
+
+    public DurationSpec.IntSecondsBound commands_for_key_prune_delta = new DurationSpec.IntSecondsBound(1);
+    public int commands_for_key_prune_interval = 64;
+    public DurationSpec.IntSecondsBound max_conflicts_prune_delta = new DurationSpec.IntSecondsBound(1);
+
     public DurationSpec.IntSecondsBound catchup_on_start_success_latency = new DurationSpec.IntSecondsBound(60);
     public DurationSpec.IntSecondsBound catchup_on_start_fail_latency = new DurationSpec.IntSecondsBound(900);
     public int catchup_on_start_max_attempts = 5;
@@ -216,7 +288,7 @@ public enum CatchupMode
     public enum RangeIndexMode { in_memory, journal_sai }
     public RangeIndexMode range_index_mode = in_memory;
 
-    public final JournalSpec journal = new JournalSpec();
+    public final JournalConfig journal = new JournalConfig();
 
     public enum MixedTimeSourceHandling
     {
@@ -225,7 +297,7 @@ public enum MixedTimeSourceHandling
 
     public volatile MixedTimeSourceHandling mixedTimeSourceHandling = MixedTimeSourceHandling.reject;
 
-    public static class JournalSpec implements Params
+    public static class JournalConfig implements Params
     {
         public enum ReplayMode
         {
@@ -265,11 +337,17 @@ public enum StopMarkerFailurePolicy
              */
             EXIT,
 
+            /**
+             * @deprecated since alpha release, replaced by ALLOW_UNSAFE_STARTUP for consistency with FailurePolicy.ALLOW_UNSAFE_STARTUP
+             */
+            @Deprecated(since="6.0")
+            UNSAFE_STARTUP,
+
             /**
              * If the start marker exceeds the stop marker startup, assuming the consensus log has been determined complete externally.
              * Note this is VERY UNSAFE if you care about isolation guarantees.
              */
-            UNSAFE_STARTUP,
+            ALLOW_UNSAFE_STARTUP,
 
             REBOOTSTRAP
         }
@@ -290,7 +368,7 @@ public enum StopMarkerFailurePolicy
         public Version version = Version.DOWNGRADE_SAFE_VERSION;
         public boolean enable_compaction = true;
 
-        public JournalSpec setFlushPeriod(DurationSpec newFlushPeriod)
+        public JournalConfig setFlushPeriod(DurationSpec newFlushPeriod)
         {
             flushPeriod = newFlushPeriod;
             flushCombinedBlockPeriod = Long.MIN_VALUE;

…/apache/cassandra/config/AccordSpec.java …pache/cassandra/config/AccordConfig.javasrc/java/org/apache/cassandra/config/AccordSpec.java renamed to src/java/org/apache/cassandra/config/AccordConfig.java src/java/org/apache/cassandra/config/AccordSpec.java renamed to src/java/org/apache/cassandra/config/AccordConfig.java

@@ -44,6 +44,7 @@ public enum CassandraRelevantProperties
 {
     ACCORD_AGENT_CLASS("cassandra.test.accord.agent"),
     ACCORD_ALLOW_TEST_MODES("cassandra.test.accord.allow_test_modes", "false"),
+    ACCORD_DEBUG_EXECUTION("accord.debug_execution"),
     ACCORD_KEY_PARANOIA_COSTFACTOR(Invariants.KEY_PARANOIA_COSTFACTOR),
     ACCORD_KEY_PARANOIA_CPU(Invariants.KEY_PARANOIA_CPU),
     ACCORD_KEY_PARANOIA_MEMORY(Invariants.KEY_PARANOIA_MEMORY),

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

@@ -165,8 +165,6 @@ public static Set<String> splitCommaDelimited(String src)
     @Replaces(oldName = "cas_contention_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true)
     public volatile DurationSpec.LongMillisecondsBound cas_contention_timeout = new DurationSpec.LongMillisecondsBound("1800ms");
 
-    public volatile DurationSpec.LongMillisecondsBound accord_preaccept_timeout = new DurationSpec.LongMillisecondsBound("1s");
-
     @Replaces(oldName = "truncate_request_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true)
     public volatile DurationSpec.LongMillisecondsBound truncate_request_timeout = new DurationSpec.LongMillisecondsBound("60000ms");
 
@@ -1238,7 +1236,7 @@ public enum PaxosOnLinearizabilityViolation
      */
     public ParameterizedClass default_compaction = null;
 
-    public final AccordSpec accord = new AccordSpec();
+    public final AccordConfig accord = new AccordConfig();
 
     public static Supplier<Config> getOverrideLoadConfig()
     {

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

@@ -126,6 +126,7 @@
 import org.apache.cassandra.service.FileSystemOwnershipCheck;
 import org.apache.cassandra.service.StartupChecks;
 import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.accord.AccordService;
 import org.apache.cassandra.service.accord.api.AccordWaitStrategies;
 import org.apache.cassandra.service.consensus.TransactionalMode;
 import org.apache.cassandra.service.paxos.Paxos;
@@ -574,7 +575,7 @@ private static void applyAll() throws ConfigurationException
 
         applyGuardrails();
 
-        applyAccordProgressLog();
+        applyAccord();
 
         applyStartupChecks();
     }
@@ -1370,7 +1371,7 @@ private static void applyGuardrails()
         }
     }
 
-    private static void applyAccordProgressLog()
+    private static void applyAccord()
     {
         try
         {
@@ -1382,6 +1383,7 @@ private static void applyAccordProgressLog()
         {
             throw new ConfigurationException("Invalid accord progress log configuration: " + e.getMessage(), e);
         }
+        AccordService.applyProtocolModifiers(getAccord());
     }
 
     public static StartupChecksConfiguration getStartupChecksConfiguration()
@@ -1663,12 +1665,6 @@ static void checkForLowestAcceptedTimeouts(Config conf)
             logInfo("truncate_request_timeout", conf.truncate_request_timeout, LOWEST_ACCEPTED_TIMEOUT);
             conf.truncate_request_timeout = LOWEST_ACCEPTED_TIMEOUT;
         }
-
-        if (conf.accord_preaccept_timeout.toMilliseconds() < LOWEST_ACCEPTED_TIMEOUT.toMilliseconds())
-        {
-            logInfo("accord_preaccept_timeout", conf.accord_preaccept_timeout, LOWEST_ACCEPTED_TIMEOUT);
-            conf.accord_preaccept_timeout = LOWEST_ACCEPTED_TIMEOUT;
-        }
     }
 
     private static void logInfo(String property, DurationSpec.LongMillisecondsBound actualValue, DurationSpec.LongMillisecondsBound lowestAcceptedValue)
@@ -5634,18 +5630,17 @@ public static void setUseStatementsEnabled(boolean enabled)
         }
     }
 
-
-    public static AccordSpec getAccord()
+    public static AccordConfig getAccord()
     {
         return conf.accord;
     }
 
-    public static AccordSpec.TransactionalRangeMigration getTransactionalRangeMigration()
+    public static AccordConfig.TransactionalRangeMigration getTransactionalRangeMigration()
     {
         return conf.accord.range_migration;
     }
 
-    public static void setTransactionalRangeMigration(AccordSpec.TransactionalRangeMigration val)
+    public static void setTransactionalRangeMigration(AccordConfig.TransactionalRangeMigration val)
     {
         conf.accord.range_migration = Preconditions.checkNotNull(val);
     }
@@ -5670,12 +5665,12 @@ public static void setAccordTransactionsEnabled(boolean b)
         conf.accord.enabled = b;
     }
 
-    public static AccordSpec.QueueShardModel getAccordQueueShardModel()
+    public static AccordConfig.QueueShardModel getAccordQueueShardModel()
     {
         return conf.accord.queue_shard_model;
     }
 
-    public static AccordSpec.QueueSubmissionModel getAccordQueueSubmissionModel()
+    public static AccordConfig.QueueSubmissionModel getAccordQueueSubmissionModel()
     {
         return conf.accord.queue_submission_model;
     }
@@ -6278,7 +6273,7 @@ public static void setPaxosRepairRaceWait(boolean paxosRepairRaceWait)
 
     public static boolean getAccordEphemeralReadEnabledEnabled()
     {
-        return conf.accord.ephemeralReadEnabled;
+        return conf.accord.ephemeral_reads;
     }
 
     public static AutoRepairConfig getAutoRepairConfig()

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

@@ -153,9 +153,9 @@ public int getTimeToLive(QueryOptions options, TableMetadata metadata) throws In
     public void collectMarkerSpecification(VariableSpecifications boundNames)
     {
         if (timestamp != null)
-            timestamp.collectMarkerSpecification(boundNames);
+            timestamp.collectMarkerSpecification(boundNames, this);
         if (timeToLive != null)
-            timeToLive.collectMarkerSpecification(boundNames);
+            timeToLive.collectMarkerSpecification(boundNames, this);
     }
 
     public static class Raw

src/java/org/apache/cassandra/cql3/Attributes.java

@@ -386,10 +386,10 @@ public boolean isMapElementExpression()
      * @param boundNames the variables specification where to collect the
      * bind variables of the map key/collection element in.
      */
-    public void collectMarkerSpecification(VariableSpecifications boundNames)
+    public void collectMarkerSpecification(VariableSpecifications boundNames, Object owner)
     {
         if (element != null)
-            element.collectMarkerSpecification(boundNames);
+            element.collectMarkerSpecification(boundNames, owner);
     }
 
     /**

src/java/org/apache/cassandra/cql3/ColumnsExpression.java

@@ -140,9 +140,9 @@ public AbstractType<?> type()
      * @param boundNames the variables specification where to collect the
      * bind variables of the map key/collection element in.
      */
-    public void collectMarkerSpecification(VariableSpecifications boundNames)
+    public void collectMarkerSpecification(VariableSpecifications boundNames, Object owner)
     {
-        keyOrIndex.collectMarkerSpecification(boundNames);
+        keyOrIndex.collectMarkerSpecification(boundNames, owner);
     }
 
     /**

src/java/org/apache/cassandra/cql3/ElementExpression.java

@@ -86,7 +86,7 @@ public Marker(int bindIndex)
 
         public Prepared prepareAndCollectMarkers(TableMetadata metadata, Collection<ColumnMetadata> receivers, VariableSpecifications boundNames)
         {
-            boundNames.add(bindIndex, makeReceiver(metadata));
+            boundNames.add(bindIndex, makeReceiver(metadata), null);
             return new PreparedMarker(bindIndex, receivers);
         }
 
@@ -239,7 +239,7 @@ public DelayedColumnValue(PreparedMarker prepared, ColumnMetadata column, boolea
         }
 
         @Override
-        public void collectMarkerSpecification(VariableSpecifications boundNames)
+        public void collectMarkerSpecification(VariableSpecifications boundNames, Object owner)
         {
             // We've already collected what we should (and in practice this method is never called).
         }