memtable_list: clear_and_add: let caller clear the old memtables

As a follow up on b8263e5,
make clear_and_add synchronous yet again, and just return
the swapped list of memtables so that the caller (table::clear)
can clear them gently.

Refs scylladb#10424 (comment)

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

Closes scylladb#10540

Author: bhalevy

replica/database.cc

@@ -1593,17 +1593,12 @@ lw_shared_ptr<memtable> memtable_list::new_memtable() {
 }
 
 // Synchronously swaps the active memtable with a new, empty one,
-// then, clears the existing memtable(s) asynchronously.
+// returning the old memtables list.
 // Exception safe.
-future<> memtable_list::clear_and_add() {
+std::vector<replica::shared_memtable> memtable_list::clear_and_add() {
     std::vector<replica::shared_memtable> new_memtables;
     new_memtables.emplace_back(new_memtable());
-    auto old_memtables = std::exchange(_memtables, std::move(new_memtables));
-    // Now that the existing _memtables vector is swapped with new_memtables
-    // the function can yield for clearing the old_memtables.
-    for (auto& smt : old_memtables) {
-        co_await smt->clear_gently();
-    }
+    return std::exchange(_memtables, std::move(new_memtables));
 }
 
 } // namespace replica

replica/database.hh

@@ -221,9 +221,9 @@ public:
     }
 
     // Synchronously swaps the active memtable with a new, empty one,
-    // then, clears the existing memtable(s) asynchronously.
+    // returning the old memtables list.
     // Exception safe.
-    future<> clear_and_add();
+    std::vector<replica::shared_memtable> clear_and_add();
 
     size_t size() const {
         return _memtables.size();

replica/table.cc

@@ -1513,7 +1513,10 @@ future<> table::clear() {
             _commitlog->discard_completed_segments(_schema->id(), t->get_and_discard_rp_set());
         }
     }
-    co_await _memtables->clear_and_add();
+    auto old_memtables = _memtables->clear_and_add();
+    for (auto& smt : old_memtables) {
+        co_await smt->clear_gently();
+    }
     co_await _cache.invalidate(row_cache::external_updater([] { /* There is no underlying mutation source */ }));
 }