memtable_list: safely futurize clear_and_add

Following a4be927
that reverted 2325c56
due to scylladb#10421, this patch reintroduces an async version
of memtable_list::clear_and_add that calls clear_gently
safely after replacing the _memtables vector with a new one
so that writes and flushes can continue in he foreground
while the old memtables are cleared.

Fixes scylladb#10281

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

Author: bhalevy

replica/database.cc

@@ -1574,6 +1574,20 @@ lw_shared_ptr<memtable> memtable_list::new_memtable() {
     return make_lw_shared<memtable>(_current_schema(), *_dirty_memory_manager, _table_stats, this, _compaction_scheduling_group);
 }
 
+// Synchronously swaps the active memtable with a new, empty one,
+// then, clears the existing memtable(s) asynchronously.
+// Exception safe.
+future<> 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();
+    }
+}
+
 } // namespace replica
 
 future<flush_permit> flush_permit::reacquire_sstable_write_permit() && {

replica/database.hh

@@ -219,17 +219,10 @@ public:
         }
     }
 
-    // Clears the active memtable and adds a new, empty one.
+    // Synchronously swaps the active memtable with a new, empty one,
+    // then, clears the existing memtable(s) asynchronously.
     // Exception safe.
-    void clear_and_add() {
-        auto mt = new_memtable();
-        _memtables.clear();
-        // emplace_back might throw only if _memtables was empty
-        // on entry. Otherwise, we rely on clear() not to release
-        // the vector capacity (See https://en.cppreference.com/w/cpp/container/vector/clear)
-        // and lw_shared_ptr being nothrow move constructible.
-        _memtables.emplace_back(std::move(mt));
-    }
+    future<> clear_and_add();
 
     size_t size() const {
         return _memtables.size();

replica/table.cc

@@ -1502,8 +1502,8 @@ future<> table::clear() {
             _commitlog->discard_completed_segments(_schema->id(), t->get_and_discard_rp_set());
         }
     }
-    _memtables->clear_and_add();
-    return _cache.invalidate(row_cache::external_updater([] { /* There is no underlying mutation source */ }));
+    co_await _memtables->clear_and_add();
+    co_await _cache.invalidate(row_cache::external_updater([] { /* There is no underlying mutation source */ }));
 }
 
 // NOTE: does not need to be futurized, but might eventually, depending on