async-flush-wal: extract ManualFlushWritableFileWriter subclass

Move manual-flush double-buffering (SwapBuffer/FlushSwappedBuffer) and
the unbounded-append-buffer policy out of WritableFileWriter into a
dedicated ManualFlushWritableFileWriter subclass.  This keeps the base
class free of manual_wal_flush concerns and replaces runtime boolean
checks with virtual dispatch (AppendBufferSizeLimit,
ShouldImplicitFlushOnAppend).  CreateWAL now instantiates the subclass
when manual_wal_flush is enabled, and FlushWAL static_casts to call the
double-buffer methods.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: wolfkdy

db/db_impl/db_impl.cc

@@ -56,6 +56,7 @@
 #include "env/unique_id_gen.h"
 #include "file/file_util.h"
 #include "file/filename.h"
+#include "file/manual_flush_writable_file_writer.h"
 #include "file/random_access_file_reader.h"
 #include "file/sst_file_manager_impl.h"
 #include "logging/auto_roll_logger.h"
@@ -1442,10 +1443,14 @@ Status DBImpl::FlushWAL(bool sync) {
       cur_log_number = log.number;
       // Swap the buffer under the lock so AddRecord() can continue
       // writing to a fresh buffer without blocking.
-      cur_log_writer->file()->SwapBuffer();
+      // Safe: CreateWAL() instantiates ManualFlushWritableFileWriter when
+      // manual_wal_flush is true (which is a precondition for this branch).
+      static_cast<ManualFlushWritableFileWriter*>(cur_log_writer->file())->SwapBuffer();
     }
     // Write the swapped buffer to disk without holding the mutex.
-    io_s = cur_log_writer->file()->FlushSwappedBuffer();
+    // Safe: CreateWAL() instantiates ManualFlushWritableFileWriter when
+    // manual_wal_flush is true (which is a precondition for this branch).
+    io_s = static_cast<ManualFlushWritableFileWriter*>(cur_log_writer->file())->FlushSwappedBuffer();
     {
       InstrumentedMutexLock wl(&log_write_mutex_);
       // Find the log we marked, since logs_.back() may have changed

db/db_impl/db_impl_open.cc

@@ -16,6 +16,7 @@
 #include "file/filename.h"
 #include "file/read_write_util.h"
 #include "file/sst_file_manager_impl.h"
+#include "file/manual_flush_writable_file_writer.h"
 #include "file/writable_file_writer.h"
 #include "logging/logging.h"
 #include "monitoring/persistent_stats_history.h"
@@ -1739,12 +1740,20 @@ IOStatus DBImpl::CreateWAL(uint64_t log_file_num, uint64_t recycle_log_number,
 
     const auto& listeners = immutable_db_options_.listeners;
     FileTypeSet tmp_set = immutable_db_options_.checksum_handoff_file_types;
-    std::unique_ptr<WritableFileWriter> file_writer(new WritableFileWriter(
-        std::move(lfile), log_fname, opt_file_options,
-        immutable_db_options_.clock, io_tracer_, nullptr /* stats */, listeners,
-        nullptr, tmp_set.Contains(FileType::kWalFile),
-        tmp_set.Contains(FileType::kWalFile),
-        immutable_db_options_.manual_wal_flush));
+    std::unique_ptr<WritableFileWriter> file_writer;
+    if (immutable_db_options_.manual_wal_flush) {
+      file_writer.reset(new ManualFlushWritableFileWriter(
+          std::move(lfile), log_fname, opt_file_options,
+          immutable_db_options_.clock, io_tracer_, nullptr /* stats */,
+          listeners, nullptr, tmp_set.Contains(FileType::kWalFile),
+          tmp_set.Contains(FileType::kWalFile)));
+    } else {
+      file_writer.reset(new WritableFileWriter(
+          std::move(lfile), log_fname, opt_file_options,
+          immutable_db_options_.clock, io_tracer_, nullptr /* stats */,
+          listeners, nullptr, tmp_set.Contains(FileType::kWalFile),
+          tmp_set.Contains(FileType::kWalFile)));
+    }
     *new_log = new log::Writer(std::move(file_writer), log_file_num,
                                immutable_db_options_.recycle_log_file_num > 0,
                                immutable_db_options_.manual_wal_flush,

file/manual_flush_writable_file_writer.cc

@@ -0,0 +1,62 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+// This source code is licensed under both the GPLv2 (found in the
+// COPYING file in the root directory) and Apache 2.0 License
+// (found in the LICENSE.Apache file in the root directory).
+
+#include "file/manual_flush_writable_file_writer.h"
+
+#include "rocksdb/io_status.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+void ManualFlushWritableFileWriter::SwapBuffer() {
+  // Double-buffering is only for buffered I/O; direct I/O has its own path.
+  assert(!use_direct_io());
+  assert(flush_buf_.CurrentSize() == 0);
+  // Ping-pong: move buf_ → flush_buf_ for writing; reuse flush_buf_'s old
+  // allocation (now empty) as the new buf_ for incoming Append()s.
+  std::swap(buf_, flush_buf_);
+  // Ensure buf_ has an allocation (first swap, or after flush_buf_ was
+  // freshly constructed).
+  if (buf_.Capacity() == 0) {
+    buf_.Alignment(writable_file()->GetRequiredBufferAlignment());
+    buf_.AllocateNewBuffer(std::min((size_t)65536, max_buffer_size_));
+  }
+}
+
+IOStatus ManualFlushWritableFileWriter::FlushSwappedBuffer(
+    Env::IOPriority op_rate_limiter_priority) {
+  if (seen_error()) {
+    return AssertFalseAndGetStatusForPrevError();
+  }
+  // Double-buffering is only for buffered I/O; direct I/O has its own path.
+  assert(!use_direct_io());
+  IOStatus s;
+
+  if (flush_buf_.CurrentSize() > 0) {
+    // Use WriteToFile() rather than WriteBuffered() to avoid touching buf_,
+    // which may be receiving new Append()s concurrently.
+    s = WriteToFile(flush_buf_.BufferStart(), flush_buf_.CurrentSize(),
+                    op_rate_limiter_priority);
+    if (!s.ok()) {
+      set_seen_error();
+      flush_buf_.Clear();
+      return s;
+    }
+  }
+  flush_buf_.Clear();
+
+  {
+    IOOptions io_options;
+    io_options.rate_limiter_priority = DecideRateLimiterPriority(
+        writable_file()->GetIOPriority(), op_rate_limiter_priority);
+    s = writable_file()->Flush(io_options, nullptr);
+  }
+
+  if (!s.ok()) {
+    set_seen_error();
+  }
+  return s;
+}
+
+}  // namespace ROCKSDB_NAMESPACE

file/manual_flush_writable_file_writer.h

@@ -0,0 +1,68 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+// This source code is licensed under both the GPLv2 (found in the
+// COPYING file in the root directory) and Apache 2.0 License
+// (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+#include "file/writable_file_writer.h"
+#include "util/aligned_buffer.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+// ManualFlushWritableFileWriter extends WritableFileWriter for use with
+// manual_wal_flush mode.  It suppresses all implicit I/O during Append()
+// and provides double-buffering so FlushWAL() can swap the write buffer
+// under the mutex and then write to disk without holding it.
+//
+// Double-buffering protocol:
+//   1. Call SwapBuffer() while holding the write mutex:
+//      moves buf_ → flush_buf_, resets buf_ for new Append()s.
+//   2. Call FlushSwappedBuffer() WITHOUT holding the write mutex:
+//      writes flush_buf_ to the OS and issues Flush(); does not touch buf_.
+class ManualFlushWritableFileWriter : public WritableFileWriter {
+ public:
+  // Inherits the same constructor signature as WritableFileWriter but always
+  // operates in manual-flush mode (manual_flush parameter is not exposed).
+  ManualFlushWritableFileWriter(
+      std::unique_ptr<FSWritableFile>&& file,
+      const std::string& file_name,
+      const FileOptions& options,
+      SystemClock* clock = nullptr,
+      const std::shared_ptr<IOTracer>& io_tracer = nullptr,
+      Statistics* stats = nullptr,
+      const std::vector<std::shared_ptr<EventListener>>& listeners = {},
+      FileChecksumGenFactory* file_checksum_gen_factory = nullptr,
+      bool perform_data_verification = false,
+      bool buffered_data_with_checksum = false)
+      : WritableFileWriter(std::move(file), file_name, options, clock,
+                           io_tracer, stats, listeners,
+                           file_checksum_gen_factory,
+                           perform_data_verification,
+                           buffered_data_with_checksum) {}
+
+  // Swap buf_ into flush_buf_ so that FlushSwappedBuffer() can write it to
+  // disk without holding the mutex while Append()s continue into the fresh
+  // buf_.  Must be called under the write mutex.
+  void SwapBuffer();
+
+  // Write flush_buf_ to the OS and issue Flush().  Safe to call without any
+  // mutex because it does not touch buf_.  Must follow SwapBuffer().
+  IOStatus FlushSwappedBuffer(
+      Env::IOPriority op_rate_limiter_priority = Env::IO_TOTAL);
+
+ protected:
+  // Allow buf_ to grow without bound so that Append() never triggers
+  // implicit I/O.  FlushWAL() is responsible for draining the buffer.
+  size_t AppendBufferSizeLimit() const override { return size_t(-1); }
+
+  // Never flush implicitly during Append(); the caller is responsible for
+  // explicitly calling SwapBuffer() + FlushSwappedBuffer().
+  bool ShouldImplicitFlushOnAppend() const override { return false; }
+
+ private:
+  // Second buffer for ping-pong double-buffering.
+  // Non-empty only between SwapBuffer() and FlushSwappedBuffer() completing.
+  AlignedBuffer flush_buf_;
+};
+
+}  // namespace ROCKSDB_NAMESPACE

file/writable_file_writer.cc

@@ -71,9 +71,7 @@ IOStatus WritableFileWriter::Append(const Slice& data, uint32_t crc32c_checksum,
 
   // See whether we need to enlarge the buffer to avoid the flush
   if (buf_.Capacity() - buf_.CurrentSize() < left) {
-    // When manual_flush_ is true, allow buffer to grow beyond
-    // max_buffer_size_ to avoid any implicit I/O during Append().
-    size_t limit = manual_flush_ ? size_t(-1) : max_buffer_size_;
+    size_t limit = AppendBufferSizeLimit();
     for (size_t cap = buf_.Capacity();
          cap < limit;  // There is still room to increase
          cap *= 2) {
@@ -87,7 +85,7 @@ IOStatus WritableFileWriter::Append(const Slice& data, uint32_t crc32c_checksum,
   }
 
   // Flush only when buffered I/O
-  if (!use_direct_io() && !manual_flush_ &&
+  if (!use_direct_io() && ShouldImplicitFlushOnAppend() &&
       (buf_.Capacity() - buf_.CurrentSize()) < left) {
     if (buf_.CurrentSize() > 0) {
       s = Flush(op_rate_limiter_priority);
@@ -334,70 +332,12 @@ IOStatus WritableFileWriter::Close() {
   return s;
 }
 
-void WritableFileWriter::SwapBuffer() {
-  assert(manual_flush_);
-  // Double-buffering is only for buffered I/O; direct I/O has its own path.
-  assert(!use_direct_io());
-  assert(flush_buf_.CurrentSize() == 0);
-  // Ping-pong: reuse flush_buf_'s allocation as the new buf_,
-  // and move the full buf_ to flush_buf_ for writing to disk.
-  std::swap(buf_, flush_buf_);
-  // If buf_ has no allocation (first swap or after move), allocate one.
-  if (buf_.Capacity() == 0) {
-    buf_.Alignment(writable_file_->GetRequiredBufferAlignment());
-    buf_.AllocateNewBuffer(std::min((size_t)65536, max_buffer_size_));
-  }
-}
-
-IOStatus WritableFileWriter::FlushSwappedBuffer(
-    Env::IOPriority op_rate_limiter_priority) {
-  if (seen_error()) {
-    return AssertFalseAndGetStatusForPrevError();
-  }
-  assert(manual_flush_);
-  // Double-buffering is only for buffered I/O; direct I/O has its own path.
-  assert(!use_direct_io());
-  IOStatus s;
-
-  if (flush_buf_.CurrentSize() > 0) {
-    // Use WriteToFile() (not WriteBuffered()) to avoid touching buf_.
-    s = WriteToFile(flush_buf_.BufferStart(), flush_buf_.CurrentSize(),
-                    op_rate_limiter_priority);
-    if (!s.ok()) {
-      set_seen_error();
-      flush_buf_.Clear();
-      return s;
-    }
-  }
-  flush_buf_.Clear();
-
-  {
-    IOOptions io_options;
-    io_options.rate_limiter_priority =
-        WritableFileWriter::DecideRateLimiterPriority(
-            writable_file_->GetIOPriority(), op_rate_limiter_priority);
-    s = writable_file_->Flush(io_options, nullptr);
-  }
-
-  if (!s.ok()) {
-    set_seen_error();
-  }
-  return s;
-}
-
 // write out the cached data to the OS cache or storage if direct I/O
 // enabled
 IOStatus WritableFileWriter::Flush(Env::IOPriority op_rate_limiter_priority) {
   if (seen_error()) {
     return AssertFalseAndGetStatusForPrevError();
   }
-  // In manual_flush_ mode flush_buf_ is only non-empty while
-  // FlushSwappedBuffer() is running without the write mutex.  Callers that
-  // subsequently call Flush() (e.g. Sync -> Flush) must wait for
-  // IsFlushing() to clear before proceeding, so flush_buf_ must be empty by
-  // the time we get here.
-  assert(!manual_flush_ || flush_buf_.CurrentSize() == 0);
-
   IOStatus s;
   TEST_KILL_RANDOM_WITH_WEIGHT("WritableFileWriter::Flush:0", REDUCE_ODDS2);
 
@@ -708,31 +648,13 @@ IOStatus WritableFileWriter::WriteBuffered(
   if (seen_error()) {
     return AssertFalseAndGetStatusForPrevError();
   }
-  // In manual_flush_ mode flush_buf_ is only non-empty while
-  // FlushSwappedBuffer() is running concurrently.  WriteBuffered() must not
-  // be called in that window — it would race on the underlying file with
-  // FlushSwappedBuffer()'s WriteToFile().  Callers must wait for IsFlushing()
-  // to clear (via wal_io_cv_) before issuing any Flush/Sync that reaches here.
-  assert(!manual_flush_ || flush_buf_.CurrentSize() == 0);
 
   IOStatus s = WriteToFile(data, size, op_rate_limiter_priority);
+  buf_.Size(0);
+  buffered_data_crc32c_checksum_ = 0;
   if (!s.ok()) {
-    // If writable_file_->Append() failed, then the data may or may not
-    // exist in the underlying memory buffer, OS page cache, remote file
-    // system's buffer, etc. If WritableFileWriter keeps the data in
-    // buf_, then a future Close() or write retry may send the data to
-    // the underlying file again. If the data does exist in the
-    // underlying buffer and gets written to the file eventually despite
-    // returning error, the file may end up with two duplicate pieces of
-    // data. Therefore, clear the buf_ at the WritableFileWriter layer
-    // and let caller determine error handling.
-    buf_.Size(0);
-    buffered_data_crc32c_checksum_ = 0;
     set_seen_error();
-    return s;
   }
-  buf_.Size(0);
-  buffered_data_crc32c_checksum_ = 0;
   return s;
 }