CNDB-13565: allow to set memtable shard lock fairness via JMX and a system property (#1785)

### What is the issue

Memtable shard lock (required for `put`) is non-fair. We suspect this
leads to elevated latencies in case of bursty load, as in #13565

### What does this PR fix and why was it fixed

This change introduces `cassandra.trie.memtable.shard.lock.fairness
system` property and `LockFairness` property
of `org.apache.cassandra.db:type=TrieMemtableConfig` JMX object to
configure it persistently or on-line.
The on-line change is effective once a new memtable is created (i.e.
after flush). If forcing a flush is not desired, one can watch
`BytesFlushed` metric for the table

Author: jakubzytka

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

@@ -21,6 +21,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.memtable.TrieMemtable;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.compress.AdaptiveCompressor;
 import org.apache.cassandra.io.compress.LZ4Compressor;
@@ -276,7 +277,7 @@ public enum CassandraRelevantProperties
 
     /** Represents the maximum size (in bytes) of a serialized mutation that can be cached **/
     CACHEABLE_MUTATION_SIZE_LIMIT("cassandra.cacheable_mutation_size_limit_bytes", Long.toString(1_000_000)),
-    
+
     MIGRATION_DELAY("cassandra.migration_delay_ms", "60000"),
     /** Defines how often schema definitions are pulled from the other nodes */
     SCHEMA_PULL_INTERVAL_MS("cassandra.schema_pull_interval_ms", "60000"),
@@ -316,7 +317,7 @@ public enum CassandraRelevantProperties
      * Number of polls without gossip state change to consider gossip as settled.
      */
     GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED("cassandra.gossip_settle_poll_success_required", "3"),
-    
+
     /** Which class to use for token metadata provider */
     CUSTOM_TMD_PROVIDER_PROPERTY("cassandra.custom_token_metadata_provider_class"),
 
@@ -641,7 +642,17 @@ public enum CassandraRelevantProperties
      * Allows custom implementation of {@link OperationContext.Factory} to optionally create and configure custom
      * {@link OperationContext} instances.
      */
-    OPERATION_CONTEXT_FACTORY("cassandra.operation_context_factory_class");
+    OPERATION_CONTEXT_FACTORY("cassandra.operation_context_factory_class"),
+
+    /**
+     * Number of shards for TrieMemtable. If not specified, defaults to {@link TrieMemtable#autoShardCount}
+     */
+    TRIE_MEMTABLE_SHARD_COUNT("cassandra.trie.memtable.shard.count"),
+
+    /**
+     * Whether to use fair locking for TrieMemtable shard locks. Defaults to false.
+     */
+    TRIE_MEMTABLE_SHARD_LOCK_FAIRNESS("cassandra.trie.memtable.shard.lock.fairness", "false");
 
     CassandraRelevantProperties(String key, String defaultVal)
     {

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

@@ -30,6 +30,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.BufferDecoratedKey;
 import org.apache.cassandra.db.ColumnFamilyStore;
@@ -49,7 +50,6 @@
 import org.apache.cassandra.db.partitions.TriePartitionUpdate;
 import org.apache.cassandra.db.partitions.TriePartitionUpdater;
 import org.apache.cassandra.db.rows.EncodingStats;
-import org.apache.cassandra.db.rows.Unfiltered;
 import org.apache.cassandra.db.rows.UnfilteredRowIterator;
 import org.apache.cassandra.db.tries.Direction;
 import org.apache.cassandra.db.tries.InMemoryTrie;
@@ -150,10 +150,8 @@ public class TrieMemtable extends AbstractAllocatorMemtable
      */
     private volatile MemtableAverageRowSize estimatedAverageRowSize;
 
-    @VisibleForTesting
-    public static final String SHARD_COUNT_PROPERTY = "cassandra.trie.memtable.shard.count";
-
-    public static volatile int SHARD_COUNT = Integer.getInteger(SHARD_COUNT_PROPERTY, autoShardCount());
+    public static volatile int SHARD_COUNT = CassandraRelevantProperties.TRIE_MEMTABLE_SHARD_COUNT.getInt(autoShardCount());
+    public static volatile boolean SHARD_LOCK_FAIRNESS = CassandraRelevantProperties.TRIE_MEMTABLE_SHARD_LOCK_FAIRNESS.getBoolean();
 
     private static int autoShardCount()
     {
@@ -595,7 +593,7 @@ public static class MemtableShard
         private volatile int partitionCount = 0;
 
         @Unmetered
-        private ReentrantLock writeLock = new ReentrantLock();
+        private ReentrantLock writeLock = new ReentrantLock(SHARD_LOCK_FAIRNESS);
 
         // Content map for the given shard. This is implemented as a memtable trie which uses the prefix-free
         // byte-comparable ByteSource representations of the keys to address the partitions.
@@ -886,12 +884,14 @@ public void setShardCount(String shardCount)
             if ("auto".equalsIgnoreCase(shardCount))
             {
                 SHARD_COUNT = autoShardCount();
+                CassandraRelevantProperties.TRIE_MEMTABLE_SHARD_COUNT.setInt(SHARD_COUNT);
             }
             else
             {
                 try
                 {
                     SHARD_COUNT = Integer.valueOf(shardCount);
+                    CassandraRelevantProperties.TRIE_MEMTABLE_SHARD_COUNT.setInt(SHARD_COUNT);
                 }
                 catch (NumberFormatException ex)
                 {
@@ -908,6 +908,20 @@ public String getShardCount()
         {
             return "" + SHARD_COUNT;
         }
+
+        @Override
+        public void setLockFairness(String fairness)
+        {
+            SHARD_LOCK_FAIRNESS = Boolean.parseBoolean(fairness);
+            CassandraRelevantProperties.TRIE_MEMTABLE_SHARD_LOCK_FAIRNESS.setBoolean(SHARD_LOCK_FAIRNESS);
+            logger.info("Requested setting shard lock fairness to {}; set to: {}", fairness, SHARD_LOCK_FAIRNESS);
+        }
+
+        @Override
+        public String getLockFairness()
+        {
+            return "" + SHARD_LOCK_FAIRNESS;
+        }
     }
 
 }

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

@@ -23,4 +23,8 @@ public interface TrieMemtableConfigMXBean
     public void setShardCount(String numShards);
 
     public String getShardCount();
+
+    public void setLockFairness(String fairness);
+
+    public String getLockFairness();
 }

test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java

@@ -54,7 +54,7 @@ public static void setUpClass()
     {
         // AlterTest uses Murmur3 partitioner, but injects OrderPreservingPartitioner.StringToken
         // into TokenMetadata; expect trouble
-        System.setProperty(TrieMemtable.SHARD_COUNT_PROPERTY, "1");
+        System.setProperty(CassandraRelevantProperties.TRIE_MEMTABLE_SHARD_COUNT.getKey(), "1");
         CQLTester.setUpClass();
     }
 

test/unit/org/apache/cassandra/db/memtable/TrieMemtableConfigTest.java

@@ -59,4 +59,12 @@ public void testAutoShardCount() throws MalformedObjectNameException, Reflection
         jmxConnection.setAttribute(new ObjectName(TRIE_MEMTABLE_CONFIG_OBJECT_NAME), new Attribute("ShardCount", "auto"));
         assertEquals(4 * FBUtilities.getAvailableProcessors(), Integer.parseInt(new TrieMemtable.TrieMemtableConfig().getShardCount()));
     }
+
+    @Test
+    public void testShardLockFairnessSetByJMX() throws MalformedObjectNameException, ReflectionException, AttributeNotFoundException, InstanceNotFoundException, MBeanException, IOException, InvalidAttributeValueException
+    {
+        assertEquals(false, Boolean.parseBoolean(new TrieMemtable.TrieMemtableConfig().getLockFairness()));
+        jmxConnection.setAttribute(new ObjectName(TRIE_MEMTABLE_CONFIG_OBJECT_NAME), new Attribute("LockFairness", "TrUe"));
+        assertEquals(true, Boolean.parseBoolean(new TrieMemtable.TrieMemtableConfig().getLockFairness()));
+    }
 }