From 4f3cc29860a812a484c07398dc3dabf742d5a202 Mon Sep 17 00:00:00 2001
From: ZhaoMing <zhaoming.274@bytedance.com>
Date: Tue, 4 Aug 2020 19:43:54 +0800
Subject: [PATCH] Fix WAL orderless bug of enabling prepare_log_writer_num Fix
 WalManager's dysfunction of cleaning empty logs

---
 db/db_impl_write.cc | 20 ++++++++++++++------
 db/wal_manager.cc   | 21 ++++++++-------------
 db/wal_manager.h    |  4 ++--
 3 files changed, 24 insertions(+), 21 deletions(-)

diff --git a/db/db_impl_write.cc b/db/db_impl_write.cc
index bfe1c9b0b..56b5412f3 100644
--- a/db/db_impl_write.cc
+++ b/db/db_impl_write.cc
@@ -1516,12 +1516,20 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context) {
   // Log this later after lock release. It may be outdated, e.g., if background
   // flush happens before logging, but that should be ok.
   int num_imm_unflushed = cfd->imm()->NumNotFlushed();
-  if (creating_new_log && !log_writer_pool_.empty()) {
-    new_log = log_writer_pool_.front().release();
-    log_writer_pool_.pop_front();
-    new_log_number = new_log->get_log_number();
-
-  } else if (creating_new_log) {
+  if (creating_new_log) {
+    while (!log_writer_pool_.empty()) {
+      auto front = std::move(log_writer_pool_.front());
+      log_writer_pool_.pop_front();
+      if (front->get_log_number() > new_log_number) {
+        new_log = front.release();
+        new_log_number = new_log->get_log_number();
+        break;
+      } else {
+        logs_to_free_queue_.emplace_back(std::move(front.release()));
+      }
+    }
+  }
+  if (creating_new_log && new_log == nullptr) {
     uint64_t recycle_log_number = 0;
     if (!log_recycle_files_.empty()) {
       recycle_log_number = log_recycle_files_.front();
diff --git a/db/wal_manager.cc b/db/wal_manager.cc
index a7a277642..1d26b96af 100644
--- a/db/wal_manager.cc
+++ b/db/wal_manager.cc
@@ -86,14 +86,14 @@ Status WalManager::DeleteFile(const std::string& fname, uint64_t number) {
   return s;
 }
 
-Status WalManager::GetSortedWalFiles(VectorLogPtr& files) {
+Status WalManager::GetSortedWalFiles(VectorLogPtr& files, bool allow_empty) {
   // First get sorted files in db dir, then get sorted files from archived
   // dir, to avoid a race condition where a log file is moved to archived
   // dir in between.
   Status s;
   // list wal files in main db dir.
   VectorLogPtr logs;
-  s = GetWalsOfType(db_options_.wal_dir, logs, kAliveLogFile);
+  s = GetWalsOfType(db_options_.wal_dir, logs, kAliveLogFile, allow_empty);
   if (!s.ok()) {
     return s;
   }
@@ -109,7 +109,7 @@ Status WalManager::GetSortedWalFiles(VectorLogPtr& files) {
   std::string archivedir = ArchivalDirectory(db_options_.wal_dir);
   Status exists = env_->FileExists(archivedir);
   if (exists.ok()) {
-    s = GetWalsOfType(archivedir, files, kArchivedLogFile);
+    s = GetWalsOfType(archivedir, files, kArchivedLogFile, allow_empty);
     if (!s.ok()) {
       return s;
     }
@@ -135,11 +135,6 @@ Status WalManager::GetSortedWalFiles(VectorLogPtr& files) {
   // archived dir first, we would have missed the log file.
   files.erase(std::unique(files.begin(), files.end(), EqualLogByPointer()),
               files.end());
-  assert(std::is_sorted(
-      files.begin(), files.end(),
-      [](const std::unique_ptr<LogFile>& a, const std::unique_ptr<LogFile>& b) {
-        return a->StartSequence() < b->StartSequence();
-      }));
   return s;
 }
 
@@ -151,7 +146,7 @@ Status WalManager::GetUpdatesSince(
   //  Do binary search and open files and find the seq number.
 
   std::unique_ptr<VectorLogPtr> wal_files(new VectorLogPtr);
-  Status s = GetSortedWalFiles(*wal_files);
+  Status s = GetSortedWalFiles(*wal_files, false /* allow_empty */);
   if (!s.ok()) {
     return s;
   }
@@ -201,7 +196,7 @@ void WalManager::PurgeObsoleteWALFiles() {
   purge_wal_files_last_run_ = now_seconds;
 
   VectorLogPtr files;
-  s = GetSortedWalFiles(files);
+  s = GetSortedWalFiles(files, true /* allow_empty */);
   if (!s.ok()) {
     ROCKS_LOG_ERROR(db_options_.info_log, "GetSortedWalFiles fail: %s",
                     s.ToString().c_str());
@@ -295,8 +290,8 @@ void WalManager::ArchiveWALFile(const std::string& fname, uint64_t number) {
 }
 
 Status WalManager::GetWalsOfType(const std::string& path,
-                                 VectorLogPtr& log_files,
-                                 WalFileType log_type) {
+                                 VectorLogPtr& log_files, WalFileType log_type,
+                                 bool allow_empty) {
   std::vector<std::string> all_files;
   const Status status = env_->GetChildren(path, &all_files);
   if (!status.ok()) {
@@ -312,7 +307,7 @@ Status WalManager::GetWalsOfType(const std::string& path,
       if (!s.ok()) {
         return s;
       }
-      if (sequence == 0) {
+      if (!allow_empty && sequence == 0) {
         // empty file
         continue;
       }
diff --git a/db/wal_manager.h b/db/wal_manager.h
index 10778a044..549db46c4 100644
--- a/db/wal_manager.h
+++ b/db/wal_manager.h
@@ -43,7 +43,7 @@ class WalManager {
 
   uint64_t GetNextLogNumber(const uint64_t number);
 
-  Status GetSortedWalFiles(VectorLogPtr& files);
+  Status GetSortedWalFiles(VectorLogPtr& files, bool allow_empty = false);
 
   Status GetUpdatesSince(
       SequenceNumber seq_number, std::unique_ptr<TransactionLogIterator>* iter,
@@ -72,7 +72,7 @@ class WalManager {
 
  private:
   Status GetWalsOfType(const std::string& path, VectorLogPtr& log_files,
-                       WalFileType type);
+                       WalFileType type, bool allow_empty);
   // Requires: all_logs should be sorted with earliest log file first
   // Retains all log files in all_logs which contain updates with seq no.
   // Greater Than or Equal to the requested SequenceNumber.
-- 
GitLab