Improve Topology Management

Merge ConfigurationService with TopologyManager to remove cyclic dependency and duplicated work.
Also:
 - Improve visibility of work blocking topology processing
 - Ensure we cannot double-count peers when deciding an epoch's distributed readiness
 - Remove the possibility of distributed stalls, by processing new topologies as soon as a
   contiguous sequence is known locally, regardless of whether anyprior local epoch is ready to
   coordinate or has recorded this fact to peers. The notification of local readiness continues
   to be processed only once all prior epochs are ready, but we can begin using and readying
   later epochs immediately.

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

Author: belliottsmith

modules/accord

@@ -56,6 +56,8 @@
 import accord.primitives.Ranges;
 import accord.primitives.Routable;
 import accord.primitives.RoutingKeys;
+import accord.topology.ActiveEpoch;
+import accord.topology.ActiveEpochs;
 import accord.topology.Shard;
 import accord.topology.Topology;
 import accord.utils.SortedListMap;
@@ -2166,7 +2168,7 @@ private ShardEpochsTable()
         public void collect(PartitionsCollector collector)
         {
             IAccordService service = AccordService.unsafeInstance();
-            List<Topology> snapshot = service.node().topology().topologySnapshot();
+            ActiveEpochs snapshot = service.node().topology().active();
             Map<TableId, Map<TokenKey, List<ShardAndEpochs>>> tableIdLookup = new HashMap<>();
 
             {
@@ -2177,8 +2179,9 @@ public void collect(PartitionsCollector collector)
 
                 TableId prevTableId = null;
                 Map<TokenKey, List<ShardAndEpochs>> startLookup = null;
-                for (Topology topology : snapshot)
+                for (ActiveEpoch epoch : snapshot)
                 {
+                    Topology topology = epoch.global();
                     for (Shard shard : topology.shards())
                     {
                         Range range = shard.range;

src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java

@@ -32,9 +32,9 @@
 
 import accord.primitives.Range;
 import accord.primitives.Ranges;
-import accord.topology.TopologyManager.EpochsSnapshot;
-import accord.topology.TopologyManager.EpochsSnapshot.Epoch;
-import accord.topology.TopologyManager.EpochsSnapshot.EpochReady;
+import accord.topology.ActiveEpoch;
+import accord.topology.ActiveEpochs;
+import accord.topology.EpochReady;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
 import org.apache.cassandra.db.marshal.LongType;
@@ -46,8 +46,6 @@
 import org.apache.cassandra.service.accord.AccordService;
 import org.apache.cassandra.service.accord.TokenRange;
 
-import static accord.topology.TopologyManager.EpochsSnapshot.ResultStatus.SUCCESS;
-
 public class AccordVirtualTables
 {
     public static final String EPOCHS = "accord_epochs";
@@ -79,7 +77,7 @@ public EpochReadyTable(String keyspace)
         {
             super(parse(keyspace, "CREATE TABLE " + EPOCHS + " (\n" +
                                   "  epoch bigint PRIMARY KEY,\n" +
-                                  "  ready_metadata text,\n" +
+                                  "  ready_active text,\n" +
                                   "  ready_coordinate text,\n" +
                                   "  ready_data text,\n" +
                                   "  ready_reads text,\n" +
@@ -94,16 +92,16 @@ public EpochReadyTable(String keyspace)
         public DataSet data()
         {
             SimpleDataSet ds = new SimpleDataSet(metadata());
-            EpochsSnapshot snapshot = epochsSnapshot();
-            for (Epoch epoch : snapshot)
+            ActiveEpochs snapshot = AccordService.instance().topology().active();
+            for (ActiveEpoch epoch : snapshot)
             {
-                ds.row(epoch.epoch);
-                EpochReady ready = epoch.ready;
-                ds.column("ready_metadata", ready.metadata.value);
-                ds.column("ready_coordinate", ready.coordinate.value);
-                ds.column("ready_data", ready.data.value);
-                ds.column("ready_reads", ready.reads.value);
-                ds.column("ready", ready.reads == SUCCESS);
+                ds.row(epoch.epoch());
+                EpochReady ready = epoch.epochReady();
+                ds.column("ready_active", ready.active.toString());
+                ds.column("ready_coordinate", ready.coordinate.toString());
+                ds.column("ready_data", ready.data.toString());
+                ds.column("ready_reads", ready.reads.toString());
+                ds.column("ready", ready.active.isDone() && ready.coordinate.isDone() && ready.data.isDone() && ready.reads.isDone());
             }
             return ds;
         }
@@ -133,21 +131,21 @@ protected EpochSyncRanges(String keyspace)
         public DataSet data()
         {
             SimpleDataSet ds = new SimpleDataSet(metadata());
-            EpochsSnapshot snapshot = epochsSnapshot();
-            for (Epoch state : snapshot)
+            ActiveEpochs snapshot = AccordService.instance().topology().active();
+            for (ActiveEpoch state : snapshot)
             {
                 Map<TableId, List<TokenRange>> addedRanges = groupByTable(state.addedRanges);
                 Map<TableId, List<TokenRange>> removedRanges = groupByTable(state.removedRanges);
-                Map<TableId, List<TokenRange>> synced = groupByTable(state.synced);
-                Map<TableId, List<TokenRange>> closed = groupByTable(state.closed);
-                Map<TableId, List<TokenRange>> retired = groupByTable(state.retired);
+                Map<TableId, List<TokenRange>> synced = groupByTable(state.quorumReady());
+                Map<TableId, List<TokenRange>> closed = groupByTable(state.closed());
+                Map<TableId, List<TokenRange>> retired = groupByTable(state.retired());
 
                 Set<TableId> allTables = union(addedRanges.keySet(), removedRanges.keySet(), synced.keySet(), closed.keySet(), retired.keySet());
                 for (TableId table : allTables)
                 {
                     TableMetadata metadata = Schema.instance.getTableMetadata(table);
                     if (metadata == null) continue; // table dropped, ignore
-                    ds.row(state.epoch, metadata.keyspace, metadata.name);
+                    ds.row(state.epoch(), metadata.keyspace, metadata.name);
 
                     ds.column("added", format(addedRanges.get(table)));
                     ds.column("removed", format(removedRanges.get(table)));
@@ -179,11 +177,6 @@ private static List<String> format(@Nullable List<TokenRange> list)
         }
     }
 
-    private static EpochsSnapshot epochsSnapshot()
-    {
-        return AccordService.instance().topology().epochsSnapshot();
-    }
-
     private static String toStringNoTable(TokenRange tr)
     {
         // TokenRange extends Range.EndInclusive

src/java/org/apache/cassandra/db/virtual/AccordVirtualTables.java

@@ -33,7 +33,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import accord.api.ConfigurationService.EpochReady;
+import accord.topology.EpochReady;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.dht.Range;