database: Merge the stop_database() into database::stop()

After stop_database() became shard-local, it's possible to merge
it with database::stop() as they are both called one after another
on scylla stop. In cql-test-env there are few more steps in
between, but they don't rely on the database being partially
stopped.

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

database.cc

@@ -2050,18 +2050,6 @@ future<> database::close_tables(table_kind kind_to_close) {
     });
 }
 
-future<> stop_database(sharded<database>& sdb) {
-    return sdb.invoke_on_all([](database& db) -> future<> {
-        co_await db.get_compaction_manager().stop();
-        co_await db._stop_barrier.arrive_and_wait();
-        // Closing a table can cause us to find a large partition. Since we want to record that, we have to close
-        // system.large_partitions after the regular tables.
-        co_await db.close_tables(database::table_kind::user);
-        co_await db.close_tables(database::table_kind::system);
-        co_await db.stop_large_data_handler();
-    });
-}
-
 future<> database::stop_large_data_handler() {
     return _large_data_handler->stop();
 }
@@ -2080,7 +2068,13 @@ future<> database::start() {
 
 future<>
 database::stop() {
-    assert(!_large_data_handler->running());
+    co_await _compaction_manager->stop();
+    co_await _stop_barrier.arrive_and_wait();
+    // Closing a table can cause us to find a large partition. Since we want to record that, we have to close
+    // system.large_partitions after the regular tables.
+    co_await close_tables(database::table_kind::user);
+    co_await close_tables(database::table_kind::system);
+    co_await _large_data_handler->stop();
 
     // try to ensure that CL has done disk flushing
     if (_commitlog) {

database.hh

@@ -1345,7 +1345,6 @@ private:
     gms::feature::listener_registration _infinite_bound_range_deletions_reg;
 
     std::unique_ptr<wasm::engine> _wasm_engine;
-    friend future<> stop_database(sharded<database>&); // temporary
     utils::cross_shard_barrier _stop_barrier;
 
 public:
@@ -1622,7 +1621,6 @@ public:
 };
 
 future<> start_large_data_handler(sharded<database>& db);
-future<> stop_database(sharded<database>& db);
 
 // Creates a streaming reader that reads from all shards.
 //

main.cc

@@ -891,11 +891,7 @@ int main(int ac, char** av) {
                 // #293 - do not stop anything - not even db (for real)
                 //return db.stop();
                 // call stop on each db instance, but leave the shareded<database> pointers alive.
-                stop_database(db).then([&db] {
-                    return db.invoke_on_all([](auto& db) {
-                        return db.stop();
-                    });
-                }).get();
+                db.invoke_on_all(&database::stop).get();
             });
 
             // We need to init commitlog on shard0 before it is inited on other shards

test/boost/gossip_test.cc

@@ -119,9 +119,6 @@ SEASTAR_TEST_CASE(test_boot_shutdown){
 
         db.start(std::ref(*cfg), dbcfg, std::ref(mm_notif), std::ref(feature_service), std::ref(token_metadata), std::ref(abort_sources), std::ref(sst_dir_semaphore)).get();
         auto stop_db = defer([&] { db.stop().get(); });
-        auto stop_database_d = defer([&db] {
-            stop_database(db).get();
-        });
 
         cdc_generation_service.start(std::ref(*cfg), std::ref(gms::get_gossiper()), std::ref(sys_dist_ks), std::ref(abort_sources), std::ref(token_metadata), std::ref(feature_service)).get();
         auto stop_cdc_generation_service = defer([&cdc_generation_service] {

test/lib/cql_test_env.cc

@@ -676,10 +676,6 @@ public:
 
             auto stop_system_keyspace = defer([] { db::qctx = {}; });
 
-            auto stop_database_d = defer([&db] {
-                stop_database(db).get();
-            });
-
             db::system_keyspace::init_local_cache().get();
             auto stop_local_cache = defer([] { db::system_keyspace::deinit_local_cache().get(); });