Merge 'Speed up bootstrap with large number of tokens in the cluster …

…10X' from Asias He

=== Setup ===
1) start node1 with
```
scylla --num-tokens 20000 --smp 1
```
The large number of tokens per node is used to simulate large number of nodes in the cluster (large total number of tokens for the cluster).

2) start node2 with
```
scylla --num-tokens 20000 --smp 1
```
3) Measure the time to finish bootstrap

=== Result ===

1) With speed up patch:
```
node1 (16s)
INFO  2022-06-21 14:30:00,038 [shard 0] init - Scylla version 5.1.dev-0.20220621.a7b927bda764 with build-id d78b6233e8227975cc26259280ceabf2cf7817b9 starting ...
INFO  2022-06-21 14:30:16,019 [shard 0] init - Scylla version 5.1.dev-0.20220621.a7b927bda764 initialization completed.
node2 (bootstrap node,174s)
INFO  2022-06-21 14:30:40,954 [shard 0] init - Scylla version 5.1.dev-0.20220621.a7b927bda764 with build-id d78b6233e8227975cc26259280ceabf2cf7817b9 starting ...
INFO  2022-06-21 14:33:34,899 [shard 0] init - Scylla version 5.1.dev-0.20220621.a7b927bda764 initialization completed.
```
2) Without speed up patch:
```
node1 (171s)
INFO  2022-06-21 14:38:49,065 [shard 0] init - Scylla version 5.1.dev-0.20220621.6f4bfea99431 with build-id f22bfa5a75887258ab48ee092ec49b5299365168 starting ...
INFO  2022-06-21 14:41:40,601 [shard 0] init - Scylla version 5.1.dev-0.20220621.6f4bfea99431 initialization completed.
node2 (bootstrap node, 1181s)
INFO  2022-06-21 14:41:46,997 [shard 0] init - Scylla version 5.1.dev-0.20220621.6f4bfea99431 with build-id f22bfa5a75887258ab48ee092ec49b5299365168 starting ...
INFO  2022-06-21 15:01:27,507 [shard 0] init - Scylla version 5.1.dev-0.20220621.6f4bfea99431 initialization completed.
```

The improvements for bootstrap time:

node1: 171s  / 16s = 10.68X
node2: 1181s / 174s = 6.78X

Refs #10337
Refs #10817
Refs #10836
Refs #10837

Closes #10850

* github.com:scylladb/scylla:
  locator: Speed up abstract_replication_strategy::get_address_ranges
  locator: Speed up simple_strategy::calculate_natural_endpoint
  token_metadata: Speed up count_normal_token_owners

locator/abstract_replication_strategy.cc

@@ -234,7 +234,21 @@ abstract_replication_strategy::get_address_ranges(const token_metadata& tm) cons
 future<std::unordered_multimap<inet_address, dht::token_range>>
 abstract_replication_strategy::get_address_ranges(const token_metadata& tm, inet_address endpoint) const {
     std::unordered_multimap<inet_address, dht::token_range> ret;
+    if (!tm.is_member(endpoint)) {
+        co_return ret;
+    }
+    bool is_everywhere_topology = get_type() == replication_strategy_type::everywhere_topology;
     for (auto& t : tm.sorted_tokens()) {
+        // This is a fast path for everywhere_topology to avoid calculating
+        // natural endpoints, since any node that is part of the the ring
+        // will be responsible for all tokens.
+        if (is_everywhere_topology) {
+            dht::token_range_vector r = tm.get_primary_ranges_for(t);
+            for (auto&& rng : r) {
+                ret.emplace(endpoint, rng);
+            }
+            continue;
+        }
         auto eps = co_await calculate_natural_endpoints(t, tm);
         bool found = false;
         for (auto ep : eps) {

locator/simple_strategy.cc

@@ -45,7 +45,11 @@ future<inet_address_vector_replica_set> simple_strategy::calculate_natural_endpo
     endpoints.reserve(replicas);
 
     for (auto& token : tm.ring_range(t)) {
-        if (endpoints.size() == replicas) {
+        // If the number of nodes in the cluster is smaller than the desired
+        // replication factor we should return the loop when endpoints already
+        // contains all the nodes in the cluster because no more nodes could be
+        // added to endpoints lists.
+        if (endpoints.size() == replicas || endpoints.size() == tm.count_normal_token_owners()) {
            break;
         }
 

locator/token_metadata.cc

@@ -50,6 +50,9 @@ class token_metadata_impl final {
      */
     // FIXME: have to be BiMultiValMap
     std::unordered_map<token, inet_address> _token_to_endpoint_map;
+    // Track the number of nodes in _token_to_endpoint_map. Need to update
+    // _nr_normal_token_owners when _token_to_endpoint_map is updated.
+    size_t _nr_normal_token_owners;
 
     /** Maintains endpoint to host ID map of every node in the cluster */
     std::unordered_map<inet_address, utils::UUID> _endpoint_to_host_id_map;
@@ -276,6 +279,8 @@ class token_metadata_impl final {
     /* Returns the number of different endpoints that own tokens in the ring.
      * Bootstrapping tokens are not taken into account. */
     size_t count_normal_token_owners() const;
+private:
+    void update_normal_token_owners();
 
 public:
     // returns empty vector if keyspace_name not found.
@@ -362,6 +367,7 @@ future<token_metadata_impl> token_metadata_impl::clone_only_token_map(bool clone
         return do_for_each(_token_to_endpoint_map, [&ret] (const auto& p) {
             ret._token_to_endpoint_map.emplace(p);
         }).then([this, &ret] {
+            ret._nr_normal_token_owners = _nr_normal_token_owners;
             ret._endpoint_to_host_id_map = _endpoint_to_host_id_map;
         }).then([this, &ret] {
             ret._topology = _topology;
@@ -376,6 +382,7 @@ future<token_metadata_impl> token_metadata_impl::clone_only_token_map(bool clone
 
 future<> token_metadata_impl::clear_gently() noexcept {
     co_await utils::clear_gently(_token_to_endpoint_map);
+    update_normal_token_owners();
     co_await utils::clear_gently(_endpoint_to_host_id_map);
     co_await utils::clear_gently(_bootstrap_tokens);
     co_await utils::clear_gently(_leaving_endpoints);
@@ -475,6 +482,7 @@ future<> token_metadata_impl::update_normal_tokens(const std::unordered_map<inet
             }
         }
     }
+    update_normal_token_owners();
 
     // New tokens were added to _token_to_endpoint_map
     // so re-sort all tokens.
@@ -647,6 +655,7 @@ bool token_metadata_impl::is_any_node_being_replaced() const {
 void token_metadata_impl::remove_endpoint(inet_address endpoint) {
     remove_by_value(_bootstrap_tokens, endpoint);
     remove_by_value(_token_to_endpoint_map, endpoint);
+    update_normal_token_owners();
     _topology.remove_endpoint(endpoint);
     _leaving_endpoints.erase(endpoint);
     del_replacing_endpoint(endpoint);
@@ -904,11 +913,15 @@ future<> token_metadata_impl::update_pending_ranges(
 }
 
 size_t token_metadata_impl::count_normal_token_owners() const {
-    std::set<inet_address> eps;
+    return _nr_normal_token_owners;
+}
+
+void token_metadata_impl::update_normal_token_owners() {
+    std::unordered_set<inet_address> eps;
     for (auto [t, ep]: _token_to_endpoint_map) {
         eps.insert(ep);
     }
-    return eps.size();
+    _nr_normal_token_owners = eps.size();
 }
 
 void token_metadata_impl::add_leaving_endpoint(inet_address endpoint) {