token_metadata: support host_id-based version

In this commit we enhance token_metadata with a pointer to the
new host_id-based generic_token_metadata specialisation (token_metadata2).
The idea is that in the following commits we'll go over all token_metadata
modifications and make the corresponding modifications to its new
host_id-based alternative.

The pointer to token_metadata2 is stored in the
generic_token_metadata::_new_value field. The pointer can be
mutable, immutable, or absent altogether (std::monostate).
It's mutable if this generic_token_metadata owns it, meaning
it was created using the generic_token_metadata(config cfg)
constructor. It's immutable if the
generic_token_metadata(lw_shared_ptr<const token_metadata2> new_value);
constructor was used. This means this old token_metadata is a wrapper for
new token_metadata and we can only use the get_new() method on it. The field
_new_value is empty for the new host_id-based token_metadata version.

The generic_token_metadata(std::unique_ptr<token_metadata_impl<NodeId>> impl, token_metadata2 new_value);
constructor is used for clone methods. We clone both versions,
and we need to pass a cloned token_metadata2 into constructor.

There are two overloads of get_new, for mutable and immutable
generic_token_metadata. Both of them throws an exception if
they can't get the appropriate pointer. There is also a
get_new_strong method, which returns an immutable owning
pointer. This is convenient since a lot of API's want an
owning pointer. We can't make the get_new/get_new_strong API
simpler and use get_new_strong everywhere since it mutate the
original generic_token_metadata by incrementing the reference
counter and this causes raises when it's passed between
shards in replicate_to_all_cores.

locator/token_metadata.cc

@@ -917,12 +917,36 @@ future<> topology_change_info<NodeId>::clear_gently() {
 
 template <typename NodeId>
 generic_token_metadata<NodeId>::generic_token_metadata(std::unique_ptr<token_metadata_impl<NodeId>> impl)
-    : _impl(std::move(impl)) {
+    : _impl(std::move(impl))
+{
+}
+
+template <typename NodeId>
+template <typename T>
+requires std::is_same_v<T, gms::inet_address>
+generic_token_metadata<NodeId>::generic_token_metadata(std::unique_ptr<token_metadata_impl<NodeId>> impl,
+    token_metadata2 new_value)
+    : _impl(std::move(impl))
+    , _new_value(make_token_metadata2_ptr(std::move(new_value)))
+{
+}
+
+template <typename NodeId>
+template <typename T>
+requires std::is_same_v<T, gms::inet_address>
+generic_token_metadata<NodeId>::generic_token_metadata(token_metadata2_ptr new_value)
+    : _impl(nullptr)
+    , _new_value(std::move(new_value))
+{
 }
 
 template <typename NodeId>
 generic_token_metadata<NodeId>::generic_token_metadata(config cfg)
-        : _impl(std::make_unique<token_metadata_impl<NodeId>>(std::move(cfg))) {
+        : _impl(std::make_unique<token_metadata_impl<NodeId>>(cfg))
+{
+    if constexpr (std::is_same_v<NodeId, gms::inet_address>) {
+        _new_value = make_token_metadata2_ptr(std::move(cfg));
+    }
 }
 
 template <typename NodeId>
@@ -1163,30 +1187,47 @@ void generic_token_metadata<NodeId>::del_replacing_endpoint(NodeId existing_node
 
 template <typename NodeId>
 future<generic_token_metadata<NodeId>> generic_token_metadata<NodeId>::clone_async() const noexcept {
-    return _impl->clone_async().then([] (std::unique_ptr<token_metadata_impl<NodeId>> impl) {
+    if constexpr (std::is_same_v<NodeId, gms::inet_address>) {
-        return make_ready_future<generic_token_metadata>(std::move(impl));
+        co_return !holds_alternative<std::monostate>(_new_value)
-    });
+                  ? generic_token_metadata(co_await _impl->clone_async(), co_await get_new()->clone_async())
+                  : generic_token_metadata(co_await _impl->clone_async());
+    } else {
+        co_return generic_token_metadata(co_await _impl->clone_async());
+    }
 }
 
 template <typename NodeId>
 future<generic_token_metadata<NodeId>>
 generic_token_metadata<NodeId>::clone_only_token_map() const noexcept {
-    return _impl->clone_only_token_map().then([] (std::unique_ptr<token_metadata_impl<NodeId>> impl) {
+    if constexpr (std::is_same_v<NodeId, gms::inet_address>) {
-        return generic_token_metadata(std::move(impl));
+        co_return !holds_alternative<std::monostate>(_new_value)
-    });
+                  ? generic_token_metadata(co_await _impl->clone_only_token_map(), co_await get_new()->clone_only_token_map())
+                  : generic_token_metadata(co_await _impl->clone_only_token_map());
+    } else {
+        co_return generic_token_metadata(co_await _impl->clone_only_token_map());
+    }
 }
 
 template <typename NodeId>
 future<generic_token_metadata<NodeId>>
 generic_token_metadata<NodeId>::clone_after_all_left() const noexcept {
-    return _impl->clone_after_all_left().then([] (std::unique_ptr<token_metadata_impl<NodeId>> impl) {
+    if constexpr (std::is_same_v<NodeId, gms::inet_address>) {
-        return generic_token_metadata(std::move(impl));
+        co_return !holds_alternative<std::monostate>(_new_value)
-    });
+                  ? generic_token_metadata(co_await _impl->clone_after_all_left(), co_await get_new()->clone_after_all_left())
+                  : generic_token_metadata(co_await _impl->clone_after_all_left());
+    } else {
+        co_return generic_token_metadata(co_await _impl->clone_after_all_left());
+    }
 }
 
 template <typename NodeId>
 future<> generic_token_metadata<NodeId>::clear_gently() noexcept {
-    return _impl->clear_gently();
+    co_await _impl->clear_gently();
+    if constexpr (std::is_same_v<NodeId, gms::inet_address>) {
+        if (holds_alternative<lw_shared_ptr<token_metadata2>>(_new_value)) {
+            co_await get_new()->clear_gently();
+        }
+    }
 }
 
 template <typename NodeId>
@@ -1409,5 +1450,10 @@ template class generic_token_metadata<locator::host_id>;
 template class generic_token_metadata<gms::inet_address>;
 template void host_id_or_endpoint::resolve(const token_metadata& tm);
 template void host_id_or_endpoint::resolve(const token_metadata2& tm);
+template token_metadata2* generic_token_metadata<gms::inet_address>::get_new<>();
+template const token_metadata2* generic_token_metadata<gms::inet_address>::get_new<>() const;
+template lw_shared_ptr<const token_metadata2> generic_token_metadata<gms::inet_address>::get_new_strong<>() const;
+template generic_token_metadata<gms::inet_address>::generic_token_metadata(std::unique_ptr<token_metadata_impl<gms::inet_address>>, token_metadata2);
+template generic_token_metadata<gms::inet_address>::generic_token_metadata(token_metadata2_ptr);
 
 } // namespace locator

locator/token_metadata.hh

@@ -93,6 +93,7 @@ public:
 template <typename NodeId = gms::inet_address>
 class generic_token_metadata final: public generic_token_metadata_base {
     std::unique_ptr<token_metadata_impl<NodeId>> _impl;
+    std::variant<std::monostate, lw_shared_ptr<token_metadata2>, lw_shared_ptr<const token_metadata2>> _new_value;
 private:
     friend class token_metadata_ring_splitter;
     class tokens_iterator {
@@ -119,6 +120,12 @@ private:
 public:
     generic_token_metadata(config cfg);
     explicit generic_token_metadata(std::unique_ptr<token_metadata_impl<NodeId>> impl);
+    template <typename T = NodeId>
+    requires std::is_same_v<T, gms::inet_address>
+    generic_token_metadata(std::unique_ptr<token_metadata_impl<NodeId>> impl, token_metadata2 new_value);
+    template <typename T = NodeId>
+    requires std::is_same_v<T, gms::inet_address>
+    generic_token_metadata(lw_shared_ptr<const token_metadata2> new_value);
     generic_token_metadata(generic_token_metadata&&) noexcept; // Can't use "= default;" - hits some static_assert in unique_ptr
     generic_token_metadata& operator=(generic_token_metadata&&) noexcept;
     ~generic_token_metadata();
@@ -140,6 +147,39 @@ public:
     const std::unordered_set<NodeId>& get_leaving_endpoints() const;
     const std::unordered_map<token, NodeId>& get_bootstrap_tokens() const;
 
+    template <typename T = NodeId>
+    requires std::is_same_v<T, gms::inet_address>
+    token_metadata2* get_new() {
+        if (holds_alternative<lw_shared_ptr<token_metadata2>>(_new_value)) {
+            return get<lw_shared_ptr<token_metadata2>>(_new_value).get();
+        }
+        throw_with_backtrace<std::runtime_error>("no mutable new value");
+    }
+
+    template <typename T = NodeId>
+    requires std::is_same_v<T, gms::inet_address>
+    const token_metadata2* get_new() const {
+        if (holds_alternative<lw_shared_ptr<token_metadata2>>(_new_value)) {
+            return get<lw_shared_ptr<token_metadata2>>(_new_value).get();
+        }
+        if (holds_alternative<lw_shared_ptr<const token_metadata2>>(_new_value)) {
+            return get<lw_shared_ptr<const token_metadata2>>(_new_value).get();
+        }
+        throw_with_backtrace<std::runtime_error>("no new value");
+    }
+
+    template <typename T = NodeId>
+    requires std::is_same_v<T, gms::inet_address>
+    lw_shared_ptr<const token_metadata2> get_new_strong() const {
+        if (holds_alternative<lw_shared_ptr<token_metadata2>>(_new_value)) {
+            return get<lw_shared_ptr<token_metadata2>>(_new_value);
+        }
+        if (holds_alternative<lw_shared_ptr<const token_metadata2>>(_new_value)) {
+            return get<lw_shared_ptr<const token_metadata2>>(_new_value);
+        }
+        throw_with_backtrace<std::runtime_error>("no new value");
+    }
+
     /**
      * Update or add endpoint given its inet_address and endpoint_dc_rack.
      */