- Detect and avoid taking two AccordExecutor locks simultaneously

Author: belliottsmith

modules/accord

@@ -30,7 +30,7 @@
 
 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.QueueSubmissionModel.SEMI_SYNC;
 import static org.apache.cassandra.config.AccordSpec.RangeIndexMode.in_memory;
 
 // TODO (expected): rename to AccordConf?
@@ -93,6 +93,8 @@ public enum QueueSubmissionModel
 
         /**
          * The queue workers only require ownership of the lock, submissions happens fully asynchronously.
+         *
+         * NOTE: EXPERIMENTAL
          */
         ASYNC,
 
@@ -106,7 +108,7 @@ public enum QueueSubmissionModel
     }
 
     public QueueShardModel queue_shard_model = THREAD_POOL_PER_SHARD;
-    public QueueSubmissionModel queue_submission_model = SYNC;
+    public QueueSubmissionModel queue_submission_model = SEMI_SYNC;
 
     /**
      * The number of queue (and cache) shards.
@@ -137,16 +139,16 @@ public enum QueueSubmissionModel
     public DurationSpec.IntMillisecondsBound repair_timeout = new DurationSpec.IntMillisecondsBound("10m");
     public String recover_txn = "5s*attempts <= 60s";
     public StringRetryStrategy recover_syncpoint = new StringRetryStrategy("60s <= 30s*attempts...60s*attempts <= 600s");
-    public String fetch_txn = "1s*attempts";
-    public String fetch_syncpoint = "5s*attempts";
-    public String expire_txn = "5s*attempts";
+    public String fetch_txn = "2s*attempts <= 60s";
+    public String fetch_syncpoint = "5s*attempts <= 60s";
+    public String expire_txn = "5s*attempts <= 60s";
     public String expire_syncpoint = "60s*attempts<=300s";
     public String expire_epoch_wait = "10s";
     // we don't want to wait ages for durability as it blocks other durability progress; even this might be too long, as we can always retry
     public String expire_durability = "10s*attempts <= 30s";
     public String slow_syncpoint_preaccept = "10s";
-    public String slow_txn_preaccept = "30ms <= p50*2 <= 100ms";
-    public String slow_read = "30ms <= p50*2 <= 100ms";
+    public String slow_txn_preaccept = "30ms <= p50*2 <= 1000ms";
+    public String slow_read = "30ms <= p50*2 <= 1000ms";
     public StringRetryStrategy retry_syncpoint = new StringRetryStrategy("10s*attempt <= 600s");
     public StringRetryStrategy retry_durability = new StringRetryStrategy("10s*attempt <= 600s");
     public StringRetryStrategy retry_bootstrap = new StringRetryStrategy("10s*attempt <= 600s");
@@ -189,8 +191,6 @@ public enum CatchupMode
         HARD
     }
 
-    public RebootstrapMode rebootstrap_mode = RebootstrapMode.full_repair;
-
     /**
      * default transactional mode for tables created by this node when no transactional mode has been specified in the DDL
      */
@@ -199,7 +199,8 @@ public enum CatchupMode
     public boolean state_cache_listener_jfr_enabled = false;
 
     public float hard_reject_ratio = 0.5f;
-    public int soft_reject_count = 100;
+    public int min_soft_reject_count = 10;
+    public int max_soft_reject_count = 100;
     public DurationSpec.LongMicrosecondsBound soft_reject_age = new DurationSpec.LongMicrosecondsBound("10s");
     public DurationSpec.LongMicrosecondsBound soft_reject_cumulative_age = new DurationSpec.LongMicrosecondsBound("60s");
 

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

@@ -5577,8 +5577,7 @@ public static int getAccordQueueShardCount()
             case THREAD_PER_SHARD_SYNC_QUEUE:
                 return conf.accord.queue_shard_count.or(DatabaseDescriptor::getAvailableProcessors);
             case THREAD_POOL_PER_SHARD:
-                int defaultMax = getAccordQueueSubmissionModel() == AccordSpec.QueueSubmissionModel.SYNC ? 8 : 4;
-                return conf.accord.queue_shard_count.or(Math.min(defaultMax, DatabaseDescriptor.getAvailableProcessors()));
+                return conf.accord.queue_shard_count.or(DatabaseDescriptor.getAvailableProcessors()/4);
         }
     }
 

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

@@ -156,6 +156,9 @@ public static boolean isSupported(AbstractCompactionStrategy.ScannerList scanner
 
     public static boolean unsupportedMetadata(TableMetadata metadata)
     {
+        if (metadata.keyspace.equals(SchemaConstants.ACCORD_KEYSPACE_NAME))
+            return false;
+
         if (!metadata.partitioner.supportsReusableKeys())
         {
             if (LOGGER.isDebugEnabled()) logDebugReason(metadata, "Incompatible partitioner, does not support reusable keys:" + metadata.partitioner.getClass().getSimpleName());

src/java/org/apache/cassandra/db/compaction/CursorCompactor.java

@@ -534,7 +534,7 @@ public void release(AccordSafeState<K, V> safeRef, AccordTask<?> owner)
 
                 AccordCacheEntry<K, V> node = cache.get(key);
 
-                require(!safeRef.invalidated());
+                require(!safeRef.isUnsafe());
                 require(safeRef.global() != null, "safeRef node is null for %s", key);
                 require(safeRef.global() == node, "safeRef node not in map: %s != %s", safeRef.global(), node);
                 require(node.references() > 0, "references (%d) are zero for %s (%s)", node.references(), key, node);
@@ -563,7 +563,7 @@ else if (node.isLoadingOrWaiting())
                 {
                     evict = node.is(LOADED) && node.isNull();
                 }
-                safeRef.invalidate();
+                safeRef.markUnsafe();
 
                 if (node.decrement() == 0)
                 {

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

@@ -250,7 +250,7 @@ public AccordCommandStore(int id,
         {
             commands = exclusive.commands.newInstance(this);
             commandsForKey = exclusive.commandsForKey.newInstance(this);
-            this.caches = new ExclusiveCaches(sharedExecutor.lock, exclusive.global, commands, commandsForKey);
+            this.caches = new ExclusiveCaches(sharedExecutor.unsafeLock(), exclusive.global, commands, commandsForKey);
         }
 
         this.exclusiveExecutor = sharedExecutor.executor(id);

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

@@ -87,6 +87,8 @@
 import org.apache.cassandra.utils.concurrent.Condition;
 import org.apache.cassandra.utils.concurrent.Future;
 
+import io.netty.util.concurrent.FastThreadLocal;
+
 import static accord.utils.Invariants.createIllegalState;
 import static org.apache.cassandra.service.accord.AccordCache.CommandAdapter.COMMAND_ADAPTER;
 import static org.apache.cassandra.service.accord.AccordCache.CommandsForKeyAdapter.CFK_ADAPTER;
@@ -117,22 +119,20 @@ public enum Mode { RUN_WITH_LOCK, RUN_WITHOUT_LOCK }
     // WARNING: this is a shared object, so close is NOT idempotent
     public static final class ExclusiveGlobalCaches extends GlobalCaches implements AutoCloseable
     {
-        final Lock lock;
         final AccordExecutor executor;
 
         public ExclusiveGlobalCaches(AccordExecutor executor, AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand> commands, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKey)
         {
             super(global, commands, commandsForKey);
-            this.lock = executor.lock;
             this.executor = executor;
         }
 
         @Override
         public void close()
         {
             executor.beforeUnlock();
-            global.tryShrinkOrEvict(lock);
-            lock.unlock();
+            global.tryShrinkOrEvict(executor.lock);
+            executor.unlock();
         }
     }
 
@@ -150,7 +150,7 @@ public GlobalCaches(AccordCache global, AccordCache.Type<TxnId, Command, AccordS
         }
     }
 
-    final Lock lock;
+    private final Lock lock;
     final Agent agent;
     final int executorId;
     private final AccordCache cache;
@@ -246,9 +246,57 @@ public int executorId()
 
     public ExclusiveGlobalCaches lockCaches()
     {
+        lock();
+        return caches;
+    }
+
+    private static final FastThreadLocal<Lock> paranoidPriorityInversionCheck = new FastThreadLocal<>();
+
+    final Lock unsafeLock()
+    {
+        return lock;
+    }
+
+    final void lock()
+    {
+        if (Invariants.isParanoid())
+        {
+            Lock locked = paranoidPriorityInversionCheck.getAndSet(lock);
+            Invariants.require(locked == null || locked == lock, "Tried to take multiple AccordExecutor locks on same thread - this is dangerous for progress");
+        }
         //noinspection LockAcquiredButNotSafelyReleased
         lock.lock();
-        return caches;
+    }
+
+    final void unlock()
+    {
+        lock.unlock();
+        paranoidPriorityInversionCheck.set(null);
+    }
+
+    final boolean tryLock()
+    {
+        boolean result = lock.tryLock();
+        if (Invariants.isParanoid())
+        {
+            if (result)
+            {
+                Lock locked = paranoidPriorityInversionCheck.getAndSet(lock);
+                if (locked != null && locked != lock)
+                {
+                    lock.unlock();
+                    paranoidPriorityInversionCheck.set(locked);
+                    Invariants.require(false, "Tried to take multiple AccordExecutor locks on same thread - this is dangerous for progress");
+                    return false;
+                }
+            }
+            else
+            {
+                Lock locked = paranoidPriorityInversionCheck.get();
+                Invariants.require(locked == null || locked == lock, "Tried to take multiple AccordExecutor locks on same thread - this is dangerous for progress");
+            }
+        }
+        return result;
     }
 
     public AccordCache cacheExclusive()
@@ -290,7 +338,7 @@ public Stream<? extends DebuggableTaskRunner> active()
     public void waitForQuiescence()
     {
         Condition condition;
-        lock.lock();
+        lock();
         try
         {
             if (tasks == 0 && runningThreads == 0)
@@ -303,7 +351,7 @@ public void waitForQuiescence()
         }
         finally
         {
-            lock.unlock();
+            unlock();
         }
         condition.awaitThrowUncheckedOnInterrupt();
     }
@@ -325,7 +373,7 @@ protected void notifyQuiescentExclusive()
 
     public void afterSubmittedAndConsequences(Runnable run)
     {
-        lock.lock();
+        lock();
         try
         {
             if (tasks == 0 && runningThreads == 0)
@@ -344,7 +392,7 @@ public void afterSubmittedAndConsequences(Runnable run)
         }
         finally
         {
-            lock.unlock();
+            unlock();
         }
     }
 
@@ -814,15 +862,15 @@ private Cancellable submit(Plain task)
 
     public void executeDirectlyWithLock(Runnable command)
     {
-        lock.lock();
+        lock();
         try
         {
             command.run();
         }
         finally
         {
             beforeUnlock();
-            lock.unlock();
+            unlock();
         }
     }
 
@@ -900,6 +948,7 @@ void clearRunning()
     public static abstract class Task extends IntrusivePriorityHeap.Node
     {
         int queuePosition;
+        Thread assigned;
 
         protected Task()
         {
@@ -999,8 +1048,8 @@ public class SequentialExecutor extends TaskQueue<Task> implements SequentialAsy
         final int commandStoreId;
         final SequentialQueueTask selfTask;
         private Task task;
+        private Thread assigned;
         private volatile Thread owner, waiting;
-        private boolean running;
         private boolean stopped;
         private volatile boolean visibleStopped;
         private boolean terminated;
@@ -1020,20 +1069,24 @@ public class SequentialExecutor extends TaskQueue<Task> implements SequentialAsy
         void preRunTask()
         {
             Invariants.require(task != null);
+            assigned = Thread.currentThread();
             task.preRunExclusive();
-            running = true;
         }
 
         void runTask()
         {
-            Thread self = Thread.currentThread();
-            while (!ownerUpdater.compareAndSet(this, null, self))
+            outer: while (!ownerUpdater.compareAndSet(this, null, assigned))
             {
-                waiting = self;
-                while (owner != null)
+                waiting = assigned;
+                while (true)
+                {
+                    Thread owner = this.owner;
+                    if (owner == assigned) break outer;
+                    if (owner == null) continue outer;
                     LockSupport.park();
-                waiting = null;
+                }
             }
+            waiting = null;
 
             try
             {
@@ -1068,7 +1121,7 @@ void cleanupTask()
             try { task.cleanupExclusive(); }
             finally
             {
-                running = false;
+                assigned = null;
                 task = super.poll();
 
                 // it should only be possible for this method to be invoked once we're on the running queue
@@ -1087,12 +1140,12 @@ protected void append(Task newTask)
         {
             if (task != null)
             {
-                Invariants.require(running || waitingToRun.contains(selfTask));
+                Invariants.require(assigned != null || waitingToRun.contains(selfTask));
                 super.append(newTask);
             }
             else
             {
-                Invariants.require(!running && isEmpty());
+                Invariants.require(assigned == null && isEmpty());
                 task = newTask;
                 selfTask.queuePosition = newTask.queuePosition;
                 waitingToRun.append(selfTask);
@@ -1115,7 +1168,7 @@ protected boolean removeIfContains(Task remove)
 
         private boolean removeCurrentTask(Node remove)
         {
-            if (running)
+            if (assigned != null)
                 return false;
 
             Invariants.require(remove == task);
@@ -1183,7 +1236,7 @@ protected Task peek()
         @Override
         protected boolean contains(Task contains)
         {
-            return super.contains(contains) || (task == contains && !running);
+            return super.contains(contains) || (task == contains && assigned == null);
         }
 
         @Override
@@ -1268,7 +1321,10 @@ public boolean tryExecuteImmediately(Runnable run)
                 {
                     this.owner = null;
                     if (waiting != null)
+                    {
                         LockSupport.unpark(waiting);
+                        ownerUpdater.compareAndSet(this, null, waiting);
+                    }
                 }
             }
             return true;
@@ -1810,7 +1866,7 @@ public int compareTo(TaskInfo that)
     public List<TaskInfo> taskSnapshot()
     {
         List<TaskInfo> result = new ArrayList<>();
-        lock.lock();
+        lock();
         try
         {
             addToSnapshot(result, waitingToLoad, TaskInfo.Status.WAITING_TO_LOAD, TaskInfo.Status.WAITING_TO_LOAD);
@@ -1822,7 +1878,7 @@ public List<TaskInfo> taskSnapshot()
         }
         finally
         {
-            lock.unlock();
+            unlock();
         }
         result.sort(TaskInfo::compareTo);
         return result;