Improve DurabilityQueue:

 - All retries are appended to a delay queue so overlaps can be pruned
 - Quorum successes are not retried if there are superseding sync points covering the ranges
 - User-initiated requests are not purged unless the request has timed out or otherwise completed
 - Overlapping requests are queued up against the next to run
Alsp (C*):
 - Catch-up with quorums on restart
 - Manage an ordered set of keys in cache for faster range searches
Also (Accord):
 - Update copy of BTree and import IntervalBTree
 - Fix RedundantStatus WAS_OWNED_OVERRIDE_MASK
 - Add Catchup mechanism to reach parity with a quorum on restart

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-21013

Author: belliottsmith

modules/accord

@@ -178,6 +178,13 @@ public enum TransactionalRangeMigration
     public TransactionalMode default_transactional_mode = TransactionalMode.off;
     public boolean ephemeralReadEnabled = true;
     public boolean state_cache_listener_jfr_enabled = true;
+
+    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;
+    public boolean catchup_on_start_exit_on_failure = true;
+    public boolean catchup_on_start = true;
+
     public final JournalSpec journal = new JournalSpec();
 
     public enum MixedTimeSourceHandling

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

@@ -21,19 +21,17 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Comparator;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.ToLongFunction;
-import java.util.stream.Stream;
-
 import javax.annotation.Nullable;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -68,6 +66,7 @@
 import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement;
 import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.btree.BTree;
 
 import static accord.utils.Invariants.illegalState;
 import static accord.utils.Invariants.require;
@@ -112,6 +111,7 @@ public interface Adapter<K, V, S>
         long estimateShrunkHeapSize(Object shrunk);
         boolean validate(AccordCommandStore commandStore, K key, V value);
         S safeRef(AccordCacheEntry<K, V> node);
+        default Comparator<K> keyComparator() { return null; }
 
         default AccordCacheEntry<K, V> newEntry(K key, AccordCache.Type<K, V, ?>.Instance owner)
         {
@@ -274,7 +274,7 @@ public void saveWhenReadyExclusive(AccordCacheEntry<?, ?> entry, Runnable onSucc
             entry.savingOrWaitingToSave().identity.onSuccess(onSuccess);
     }
 
-    private void evict(AccordCacheEntry<?, ?> node, boolean updateUnreferenced)
+    private <K> void evict(AccordCacheEntry<K, ?> node, boolean updateUnreferenced)
     {
         if (logger.isTraceEnabled())
             logger.trace("Evicting {}", node);
@@ -297,7 +297,7 @@ private void evict(AccordCacheEntry<?, ?> node, boolean updateUnreferenced)
         if (node.status() == LOADED && VALIDATE_LOAD_ON_EVICT)
             owner.validateLoadEvicted(node);
 
-        AccordCacheEntry<?, ?> self = node.owner.cache.remove(node.key());
+        AccordCacheEntry<K, ?> self = node.owner.remove(node.key());
         Invariants.require(self.references() == 0);
         require(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self);
         node.notifyListeners(Listener::onEvict);
@@ -400,6 +400,8 @@ public class Instance implements Iterable<AccordCacheEntry<K, V>>
             // TODO (desired): don't need to store key separately as stored in node; ideally use a hash set that allows us to get the current entry
             private final Map<K, AccordCacheEntry<K, V>> cache = new Object2ObjectHashMap<>();
             private List<Listener<K, V>> listeners = null;
+            // TODO (expected): update this after releasing the lock
+            private OrderedKeys<K> orderedKeys;
 
             public Instance(AccordCommandStore commandStore)
             {
@@ -435,7 +437,6 @@ public void recordPreAcquired(AccordSafeState<K, V> ref)
 
             private AccordCacheEntry<K, V> acquire(K key, boolean onlyIfLoaded)
             {
-                @SuppressWarnings("unchecked")
                 AccordCacheEntry<K, V> node = cache.get(key);
                 return node == null
                        ? acquireAbsent(key, onlyIfLoaded)
@@ -454,8 +455,10 @@ private AccordCacheEntry<K, V> acquireAbsent(K key, boolean onlyIfLoaded)
                 node.increment();
 
                 Object prev = cache.put(key, node);
-                node.initSize(parent());
                 Invariants.require(prev == null, "%s not absent from cache: %s already present", key, node);
+                if (orderedKeys != null)
+                    orderedKeys.add(key);
+                node.initSize(parent());
                 ++size;
                 node.notifyListeners(Listener::onAdd);
                 maybeShrinkOrEvictSomeNodes();
@@ -555,16 +558,27 @@ else if (node.isLoadingOrWaiting())
                 maybeShrinkOrEvictSomeNodes();
             }
 
-            public Stream<AccordCacheEntry<K, V>> stream()
+            AccordCacheEntry<K, ?> remove(K key)
             {
-                return cache.values().stream();
+                AccordCacheEntry<K, ?> result = cache.remove(key);
+                if (orderedKeys != null && result != null)
+                    orderedKeys.remove(key);
+                return result;
             }
 
             final Type<K, V, S> parent()
             {
                 return Type.this;
             }
 
+            public Iterable<K> keysBetween(K start, boolean startInclusive, K end, boolean endInclusive)
+            {
+                if (orderedKeys == null)
+                    orderedKeys = new OrderedKeys<>(adapter.keyComparator(), cache.keySet());
+
+                return orderedKeys.between(start, startInclusive, end, endInclusive);
+            }
+
             @Override
             public Iterator<AccordCacheEntry<K, V>> iterator()
             {
@@ -602,11 +616,6 @@ public AccordCacheEntry<K, V> getUnsafe(K key)
                 return cache.get(key);
             }
 
-            public Set<K> keySet()
-            {
-                return cache.keySet();
-            }
-
             @VisibleForTesting
             public boolean isReferenced(K key)
             {
@@ -1039,6 +1048,12 @@ public AccordCacheEntry<K, V> newEntry(K key, Type<K, V, ?>.Instance owner)
         {
             return newNode.apply(key, owner);
         }
+
+        @Override
+        public Comparator<K> keyComparator()
+        {
+            return Comparator.comparing(a -> ((Comparable) a));
+        }
     }
 
     static class SettableWrapper<K, V, S> extends FunctionalAdapter<K, V, S>
@@ -1148,6 +1163,12 @@ public AccordSafeCommandsForKey safeRef(AccordCacheEntry<RoutingKey, CommandsFor
         {
             return new AccordSafeCommandsForKey(node);
         }
+
+        @Override
+        public Comparator<RoutingKey> keyComparator()
+        {
+            return RoutingKey::compareAsRoutingKey;
+        }
     }
 
     public static class CommandAdapter implements Adapter<TxnId, Command, AccordSafeCommand>

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

@@ -55,6 +55,7 @@
 import accord.local.SafeCommandStore;
 import accord.local.cfk.CommandsForKey;
 import accord.primitives.PartialTxn;
+import accord.primitives.Range;
 import accord.primitives.Ranges;
 import accord.primitives.RoutableKey;
 import accord.primitives.Route;
@@ -490,12 +491,16 @@ class Ready extends CountingResult implements Runnable
             Ready ready = new Ready();
             try (ExclusiveCaches caches = lockCaches())
             {
-                for (AccordCacheEntry<RoutingKey, CommandsForKey> e : caches.commandsForKeys())
+                for (Range range : ranges)
                 {
-                    if (ranges.contains(e.key()) && e.isModified())
+                    for (RoutingKey k : caches.commandsForKeys().keysBetween(range.start(), range.startInclusive(), range.end(), range.endInclusive()))
                     {
-                        ready.increment();
-                        caches.global().saveWhenReadyExclusive(e, ready);
+                        AccordCacheEntry<RoutingKey, CommandsForKey> e = caches.commandsForKeys().getUnsafe(k);
+                        if (e.isModified())
+                        {
+                            ready.increment();
+                            caches.global().saveWhenReadyExclusive(e, ready);
+                        }
                     }
                 }
             }