generic_server: coroutinize server::shutdown()

By turning server::shutdown() into a coroutine, we need not dynamically
allocate "nr_conn".

Verified as follows:

(1) In terminal #1:

    build/Dev/scylla --overprovisioned --developer-mode=yes \
        --memory=2G --smp=1 --default-log-level error \
        --logger-log-level cql_server=debug:cql_server_controller=debug

> INFO  [...] cql_server_controller - Starting listening for CQL clients
>                                     on 127.0.0.1:9042 (unencrypted,
>                                     non-shard-aware)
> INFO  [...] cql_server_controller - Starting listening for CQL clients
>                                     on 127.0.0.1:19042 (unencrypted,
>                                     shard-aware)

(2) In terminals #2 and #3:

    tools/cqlsh/bin/cqlsh.py

(3) Press ^C in terminal #1:

> DEBUG [...] cql_server - abort accept nr_total=2
> DEBUG [...] cql_server - abort accept 1 out of 2 done
> DEBUG [...] cql_server - abort accept 2 out of 2 done
> DEBUG [...] cql_server - shutdown connection nr_total=4
> DEBUG [...] cql_server - shutdown connection 1 out of 4 done
> DEBUG [...] cql_server - shutdown connection 2 out of 4 done
> DEBUG [...] cql_server - shutdown connection 3 out of 4 done
> DEBUG [...] cql_server - shutdown connection 4 out of 4 done
> INFO  [...] cql_server_controller - CQL server stopped

This patch is best viewed with "git show --word-diff=color".

Suggested-by: Benny Halevy <[email protected]>
Signed-off-by: Laszlo Ersek <[email protected]>

Author: Laszlo Ersek

generic_server.cc

@@ -11,7 +11,7 @@
 
 #include <fmt/ranges.h>
 #include <seastar/core/when_all.hh>
-#include <seastar/core/loop.hh>
+#include <seastar/coroutine/parallel_for_each.hh>
 #include <seastar/core/reactor.hh>
 
 namespace generic_server {
@@ -119,7 +119,7 @@ future<> server::stop() {
 
 future<> server::shutdown() {
     if (_gate.is_closed()) {
-        return make_ready_future<>();
+        co_return;
     }
     _all_connections_stopped = _gate.close();
     size_t nr = 0;
@@ -129,16 +129,14 @@ future<> server::shutdown() {
         l.abort_accept();
         _logger.debug("abort accept {} out of {} done", ++nr, nr_total);
     }
-    auto nr_conn = make_lw_shared<size_t>(0);
+    size_t nr_conn = 0;
     auto nr_conn_total = _connections_list.size();
     _logger.debug("shutdown connection nr_total={}", nr_conn_total);
-    return parallel_for_each(_connections_list.begin(), _connections_list.end(), [this, nr_conn, nr_conn_total] (auto&& c) {
-        return c.shutdown().then([this, nr_conn, nr_conn_total] {
-            _logger.debug("shutdown connection {} out of {} done", ++(*nr_conn), nr_conn_total);
-        });
-    }).then([this] {
-        return std::move(_listeners_stopped);
+    co_await coroutine::parallel_for_each(_connections_list, [&] (auto&& c) -> future<> {
+        co_await c.shutdown();
+        _logger.debug("shutdown connection {} out of {} done", ++nr_conn, nr_conn_total);
     });
+    co_await std::move(_listeners_stopped);
 }
 
 future<>