fix(server): Support FLUSH(ALL) SYNC (#5821)

Author: dranikpg

src/server/db_slice.cc

@@ -893,7 +893,7 @@ void DbSlice::FlushSlots(const cluster::SlotRanges& slot_ranges) {
   }).Detach();
 }
 
-void DbSlice::FlushDbIndexes(const std::vector<DbIndex>& indexes) {
+util::fb2::Fiber DbSlice::FlushDbIndexes(const std::vector<DbIndex>& indexes) {
   bool clear_tiered = owner_->tiered_storage() != nullptr;
 
   if (clear_tiered)
@@ -902,7 +902,7 @@ void DbSlice::FlushDbIndexes(const std::vector<DbIndex>& indexes) {
   DbTableArray flush_db_arr(db_arr_.size());
 
   for (DbIndex index : indexes) {
-    if (!index) {
+    if (index == 0) {  // TODO: Async dealloc?
       owner_->search_indices()->DropAllIndices();
     }
 
@@ -925,20 +925,17 @@ void DbSlice::FlushDbIndexes(const std::vector<DbIndex>& indexes) {
                                           ServerState::kGlibcmalloc);
   };
 
-  fb2::Fiber("flush_dbs", std::move(cb)).Detach();
+  return {"flush_dbs", std::move(cb)};
 }
 
-void DbSlice::FlushDb(DbIndex db_ind) {
+util::fb2::Fiber DbSlice::FlushDb(DbIndex db_ind) {
   DVLOG(1) << "Flushing db " << db_ind;
 
   // clear client tracking map.
   client_tracking_map_.clear();
 
-  if (db_ind != kDbAll) {
-    // Flush a single database if a specific index is provided
-    FlushDbIndexes({db_ind});
-    return;
-  }
+  if (db_ind != kDbAll)  // Flush a single database if a specific index is provided
+    return FlushDbIndexes({db_ind});
 
   std::vector<DbIndex> indexes;
   indexes.reserve(db_arr_.size());
@@ -948,7 +945,7 @@ void DbSlice::FlushDb(DbIndex db_ind) {
     }
   }
 
-  FlushDbIndexes(indexes);
+  return FlushDbIndexes(indexes);
 }
 
 void DbSlice::AddExpire(DbIndex db_ind, const Iterator& main_it, uint64_t at) {

src/server/db_slice.h

@@ -331,7 +331,7 @@ class DbSlice {
   constexpr static DbIndex kDbAll = 0xFFFF;
 
   // Flushes db_ind or all databases if kDbAll is passed
-  void FlushDb(DbIndex db_ind);
+  util::fb2::Fiber FlushDb(DbIndex db_ind);
 
   // Flushes the data of given slot ranges.
   void FlushSlots(const cluster::SlotRanges& slot_ranges);
@@ -551,7 +551,7 @@ class DbSlice {
                                              bool force_update);
 
   void FlushSlotsFb(const cluster::SlotSet& slot_ids);
-  void FlushDbIndexes(const std::vector<DbIndex>& indexes);
+  util::fb2::Fiber FlushDbIndexes(const std::vector<DbIndex>& indexes);
 
   // Invalidate all watched keys in database. Used on FLUSH.
   void InvalidateDbWatches(DbIndex db_indx);

src/server/server_family.cc

@@ -1284,10 +1284,7 @@ void ServerFamily::FlushAll(Namespace* ns) {
   boost::intrusive_ptr<Transaction> flush_trans(new Transaction{cid});
   flush_trans->InitByArgs(ns, 0, {});
   VLOG(1) << "Performing flush";
-  error_code ec = Drakarys(flush_trans.get(), DbSlice::kDbAll);
-  if (ec) {
-    LOG(ERROR) << "Error flushing db " << ec.message();
-  }
+  Drakarys(flush_trans.get(), DbSlice::kDbAll, false);
 }
 
 // Load starts as many fibers as there are files to load each one separately.
@@ -2206,17 +2203,20 @@ bool ServerFamily::TEST_IsSaving() const {
   return is_saving.load(std::memory_order_relaxed);
 }
 
-error_code ServerFamily::Drakarys(Transaction* transaction, DbIndex db_ind) {
+void ServerFamily::Drakarys(Transaction* transaction, DbIndex db_ind, bool wait) {
   VLOG(1) << "Drakarys";
 
+  vector<fb2::Fiber> fibers(shard_set->size());
   transaction->Execute(
-      [db_ind](Transaction* t, EngineShard* shard) {
-        t->GetDbSlice(shard->shard_id()).FlushDb(db_ind);
+      [db_ind, &fibers](Transaction* t, EngineShard* shard) {
+        fibers[shard->shard_id()] = t->GetDbSlice(shard->shard_id()).FlushDb(db_ind);
         return OpStatus::OK;
       },
       true);
 
-  return error_code{};
+  auto action = wait ? &fb2::Fiber::JoinIfNeeded : &fb2::Fiber::Detach;
+  for (auto& f : fibers)
+    (f.*action)();
 }
 
 SaveInfoData ServerFamily::GetLastSaveInfo() const {
@@ -2286,20 +2286,14 @@ void ServerFamily::SendInvalidationMessages() const {
 }
 
 void ServerFamily::FlushDb(CmdArgList args, const CommandContext& cmd_cntx) {
-  DCHECK(cmd_cntx.tx);
-  Drakarys(cmd_cntx.tx, cmd_cntx.tx->GetDbIndex());
-  SendInvalidationMessages();
-  cmd_cntx.rb->SendOk();
-}
+  if (args.size() > 1)
+    return cmd_cntx.rb->SendError(kSyntaxErr);
 
-void ServerFamily::FlushAll(CmdArgList args, const CommandContext& cmd_cntx) {
-  if (args.size() > 1) {
-    cmd_cntx.rb->SendError(kSyntaxErr);
-    return;
-  }
+  bool sync = CmdArgParser{args}.Check("SYNC");
+  string_view cmd_name = cmd_cntx.tx->GetCId()->name();
+  DbIndex index = cmd_name == "FLUSHALL" ? DbSlice::kDbAll : cmd_cntx.tx->GetDbIndex();
 
-  DCHECK(cmd_cntx.tx);
-  Drakarys(cmd_cntx.tx, DbSlice::kDbAll);
+  Drakarys(cmd_cntx.tx, index, sync);
   SendInvalidationMessages();
   cmd_cntx.rb->SendOk();
 }
@@ -4037,10 +4031,10 @@ void ServerFamily::Register(CommandRegistry* registry) {
       << CI{"CONFIG", CO::ADMIN | CO::LOADING | CO::DANGEROUS, -2, 0, 0, acl::kConfig}.HFUNC(Config)
       << CI{"DBSIZE", CO::READONLY | CO::FAST | CO::LOADING, 1, 0, 0, acl::kDbSize}.HFUNC(DbSize)
       << CI{"DEBUG", CO::ADMIN | CO::LOADING, -2, 0, 0, acl::kDebug}.HFUNC(Debug)
-      << CI{"FLUSHDB", CO::WRITE | CO::GLOBAL_TRANS | CO::DANGEROUS, 1, 0, 0, acl::kFlushDB}.HFUNC(
+      << CI{"FLUSHDB", CO::WRITE | CO::GLOBAL_TRANS | CO::DANGEROUS, -1, 0, 0, acl::kFlushDB}.HFUNC(
              FlushDb)
       << CI{"FLUSHALL", CO::WRITE | CO::GLOBAL_TRANS | CO::DANGEROUS, -1, 0, 0, acl::kFlushAll}
-             .HFUNC(FlushAll)
+             .HFUNC(FlushDb)
       << CI{"INFO", CO::LOADING, -1, 0, 0, acl::kInfo}.HFUNC(Info)
       << CI{"HELLO", CO::LOADING, -1, 0, 0, acl::kHello}.HFUNC(Hello)
       << CI{"LASTSAVE", CO::LOADING | CO::FAST, 1, 0, 0, acl::kLastSave}.HFUNC(LastSave)

src/server/server_family.h

@@ -250,7 +250,8 @@ class ServerFamily {
 
   // Burns down and destroy all the data from the database.
   // if kDbAll is passed, burns all the databases to the ground.
-  std::error_code Drakarys(Transaction* transaction, DbIndex db_ind);
+  // `wait` makes it wait for all fibers to finish and decommit
+  void Drakarys(Transaction* transaction, DbIndex db_ind, bool wait);
 
   SaveInfoData GetLastSaveInfo() const;
 
@@ -344,7 +345,6 @@ class ServerFamily {
   void Dfly(CmdArgList args, const CommandContext& cmd_cntx);
   void Memory(CmdArgList args, const CommandContext& cmd_cntx);
   void FlushDb(CmdArgList args, const CommandContext& cmd_cntx);
-  void FlushAll(CmdArgList args, const CommandContext& cmd_cntx);
   void Info(CmdArgList args, const CommandContext& cmd_cntx) ABSL_LOCKS_EXCLUDED(replicaof_mu_);
   void Hello(CmdArgList args, const CommandContext& cmd_cntx);
   void LastSave(CmdArgList args, const CommandContext& cmd_cntx);

tests/dragonfly/memory_test.py

@@ -24,7 +24,7 @@
 # We limit to 5gb just in case to sanity check the gh runner. Otherwise, if we ask for too much
 # memory it might force the gh runner to run out of memory (since OOM killer might not even
 # get a chance to run).
-async def test_rss_used_mem_gap(df_factory, type, keys, val_size, elements):
+async def test_rss_used_mem_gap(df_factory: DflyInstanceFactory, type, keys, val_size, elements):
     dbfilename = f"dump_{tmp_file_name()}"
     instance = df_factory.create(
         proactor_threads=2,
@@ -74,9 +74,17 @@ async def check_memory():
 
     assert await client.execute_command("SAVE", "DF") == True
     assert await client.execute_command("DFLY", "LOAD", f"{dbfilename}-summary.dfs") == "OK"
-    #
+
     await check_memory()
-    await client.execute_command("FLUSHALL")
+
+    # FLUSHALL sync waits for flush to finish and decommit memory, so send INFO immediately after
+    p = client.pipeline(transaction=False)
+    p.execute_command("FLUSHALL", "SYNC")  # flushall(asynchronous=False) will just issue FLUSHALL$
+    p.info("memory")
+
+    info = (await p.execute())[-1]
+    assert info["used_memory"] < 2 * 1_000_000  # Table memory
+    assert info["used_memory_rss"] < min_rss / 10  # RSS must have been freed
 
 
 #