messaging, topology: Keep shared_token_metadata* on messaging

Messaging will need to call topology methods to compare DC/RACK of peers
with local node. Topology now resides on token metadata, so messaging
needs to get the dependency reference.

However, messaging only needs the topology when it's up and running, so
instead of producing a life-time reference, add a pointer, that's set up
on .start_listen(), before any client pops up, and is cleared on
.shutdown() after all connections are dropped.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>

main.cc

@@ -1362,7 +1362,9 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
             });
 
             with_scheduling_group(maintenance_scheduling_group, [&] {
-                return messaging.invoke_on_all(&netw::messaging_service::start_listen);
+                return messaging.invoke_on_all([&token_metadata] (auto& netw) {
+                    return netw.start_listen(token_metadata.local());
+                });
             }).get();
 
             with_scheduling_group(maintenance_scheduling_group, [&] {

message/messaging_service.cc

@@ -236,7 +236,8 @@ rpc_resource_limits(size_t memory_limit) {
     return limits;
 }
 
-future<> messaging_service::start_listen() {
+future<> messaging_service::start_listen(locator::shared_token_metadata& stm) {
+    _token_metadata = &stm;
     if (_credentials_builder && !_credentials) {
         return _credentials_builder->build_reloadable_server_credentials([](const std::unordered_set<sstring>& files, std::exception_ptr ep) {
             if (ep) {
@@ -429,7 +430,8 @@ future<> messaging_service::stop_client() {
 
 future<> messaging_service::shutdown() {
     _shutting_down = true;
-    return when_all(stop_nontls_server(), stop_tls_server(), stop_client()).discard_result();
+    co_await when_all(stop_nontls_server(), stop_tls_server(), stop_client()).discard_result();
+    _token_metadata = nullptr;
 }
 
 future<> messaging_service::stop() {

message/messaging_service.hh

@@ -51,6 +51,10 @@ namespace db::view {
 class update_backlog;
 }
 
+namespace locator {
+class shared_token_metadata;
+}
+
 class frozen_mutation;
 class frozen_schema;
 class canonical_mutation;
@@ -291,6 +295,7 @@ private:
     };
 private:
     config _cfg;
+    locator::shared_token_metadata* _token_metadata = nullptr;
     // map: Node broadcast address -> Node internal IP, and the reversed mapping, for communication within the same data center
     std::unordered_map<gms::inet_address, gms::inet_address> _preferred_ip_cache, _preferred_to_endpoint;
     std::unique_ptr<rpc_protocol_wrapper> _rpc;
@@ -317,7 +322,7 @@ public:
     messaging_service(config cfg, scheduling_config scfg, std::shared_ptr<seastar::tls::credentials_builder>);
     ~messaging_service();
 
-    future<> start_listen();
+    future<> start_listen(locator::shared_token_metadata& stm);
     uint16_t port();
     gms::inet_address listen_address();
     future<> shutdown();