feat: BatchIO interface for swappable flusher I/O backend

New package batchio/ with BatchIO interface (PreadBatch, PwriteBatch,
Fsync, LinkedWriteFsync) and standard sequential implementation.

Flusher refactored to use BatchIO: WAL header reads, WAL entry reads,
and extent writes are now batched through the interface. With the
default NewStandard() backend, behavior is identical to before.

UseIOUring config field added for future io_uring opt-in (Linux 5.6+).
9 interface tests, all existing blockvol tests pass unchanged.

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

weed/storage/blockvol/batchio/batchio.go

@@ -0,0 +1,46 @@
+// Package batchio provides a swappable I/O backend for batch pread/pwrite/fsync.
+//
+// The standard implementation uses sequential os.File calls (identical to current
+// code). An optional io_uring implementation (Linux 5.6+, build-tagged) can batch
+// these into fewer syscalls for higher throughput.
+//
+// Usage:
+//
+//	bio := batchio.NewStandard()  // or batchio.NewIOUring(256) on Linux
+//	defer bio.Close()
+//	bio.PreadBatch(fd, ops)
+//	bio.PwriteBatch(fd, ops)
+//	bio.Fsync(fd)
+package batchio
+
+import "os"
+
+// Op represents a single I/O operation: read or write buf at offset.
+type Op struct {
+	Buf    []byte
+	Offset int64
+}
+
+// BatchIO batches pread/pwrite/fsync operations.
+// All methods are safe for concurrent use from a single goroutine.
+// Callers must not share a BatchIO across goroutines without external locking.
+type BatchIO interface {
+	// PreadBatch reads multiple regions from fd. Each Op.Buf is filled with
+	// data from Op.Offset. Returns the first error encountered.
+	PreadBatch(fd *os.File, ops []Op) error
+
+	// PwriteBatch writes multiple regions to fd. Each Op.Buf is written at
+	// Op.Offset. Returns the first error encountered.
+	PwriteBatch(fd *os.File, ops []Op) error
+
+	// Fsync issues fdatasync on the file.
+	Fsync(fd *os.File) error
+
+	// LinkedWriteFsync writes buf at offset then fsyncs, as an atomic pair.
+	// On io_uring this is a linked SQE chain (one syscall).
+	// On standard, this is sequential write + fdatasync.
+	LinkedWriteFsync(fd *os.File, buf []byte, offset int64) error
+
+	// Close releases resources (io_uring ring, etc). No-op for standard.
+	Close() error
+}

weed/storage/blockvol/batchio/standard.go

@@ -0,0 +1,46 @@
+package batchio
+
+import "os"
+
+// standardBatchIO implements BatchIO with sequential os.File calls.
+// This is functionally identical to calling ReadAt/WriteAt/Sync directly.
+type standardBatchIO struct{}
+
+// NewStandard returns a BatchIO that uses sequential pread/pwrite/fdatasync.
+// This is the default (and only) implementation on non-Linux platforms.
+func NewStandard() BatchIO {
+	return &standardBatchIO{}
+}
+
+func (s *standardBatchIO) PreadBatch(fd *os.File, ops []Op) error {
+	for i := range ops {
+		if _, err := fd.ReadAt(ops[i].Buf, ops[i].Offset); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *standardBatchIO) PwriteBatch(fd *os.File, ops []Op) error {
+	for i := range ops {
+		if _, err := fd.WriteAt(ops[i].Buf, ops[i].Offset); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *standardBatchIO) Fsync(fd *os.File) error {
+	return fd.Sync()
+}
+
+func (s *standardBatchIO) LinkedWriteFsync(fd *os.File, buf []byte, offset int64) error {
+	if _, err := fd.WriteAt(buf, offset); err != nil {
+		return err
+	}
+	return fd.Sync()
+}
+
+func (s *standardBatchIO) Close() error {
+	return nil
+}

weed/storage/blockvol/batchio/standard_test.go

@@ -0,0 +1,218 @@
+package batchio
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func tempFile(t *testing.T) *os.File {
+	t.Helper()
+	f, err := os.CreateTemp(t.TempDir(), "batchio-test-*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { f.Close() })
+	return f
+}
+
+func TestStandard_PwriteBatch(t *testing.T) {
+	f := tempFile(t)
+	bio := NewStandard()
+	defer bio.Close()
+
+	ops := []Op{
+		{Buf: []byte("AAAA"), Offset: 0},
+		{Buf: []byte("BBBB"), Offset: 4096},
+		{Buf: []byte("CCCC"), Offset: 8192},
+	}
+	if err := bio.PwriteBatch(f, ops); err != nil {
+		t.Fatalf("PwriteBatch: %v", err)
+	}
+
+	// Verify each region.
+	for _, op := range ops {
+		got := make([]byte, len(op.Buf))
+		if _, err := f.ReadAt(got, op.Offset); err != nil {
+			t.Fatalf("ReadAt offset %d: %v", op.Offset, err)
+		}
+		if !bytes.Equal(got, op.Buf) {
+			t.Errorf("offset %d: got %q, want %q", op.Offset, got, op.Buf)
+		}
+	}
+}
+
+func TestStandard_PreadBatch(t *testing.T) {
+	f := tempFile(t)
+	bio := NewStandard()
+	defer bio.Close()
+
+	// Write known data.
+	data := []byte("Hello, World! This is batch I/O testing data.")
+	if _, err := f.WriteAt(data, 0); err != nil {
+		t.Fatal(err)
+	}
+
+	ops := []Op{
+		{Buf: make([]byte, 5), Offset: 0},
+		{Buf: make([]byte, 6), Offset: 7},
+	}
+	if err := bio.PreadBatch(f, ops); err != nil {
+		t.Fatalf("PreadBatch: %v", err)
+	}
+
+	if string(ops[0].Buf) != "Hello" {
+		t.Errorf("op[0]: got %q, want %q", ops[0].Buf, "Hello")
+	}
+	if string(ops[1].Buf) != "World!" {
+		t.Errorf("op[1]: got %q, want %q", ops[1].Buf, "World!")
+	}
+}
+
+func TestStandard_Fsync(t *testing.T) {
+	f := tempFile(t)
+	bio := NewStandard()
+	defer bio.Close()
+
+	if _, err := f.WriteAt([]byte("data"), 0); err != nil {
+		t.Fatal(err)
+	}
+	if err := bio.Fsync(f); err != nil {
+		t.Fatalf("Fsync: %v", err)
+	}
+}
+
+func TestStandard_LinkedWriteFsync(t *testing.T) {
+	f := tempFile(t)
+	bio := NewStandard()
+	defer bio.Close()
+
+	data := []byte("durable write")
+	if err := bio.LinkedWriteFsync(f, data, 100); err != nil {
+		t.Fatalf("LinkedWriteFsync: %v", err)
+	}
+
+	// Verify data persisted.
+	got := make([]byte, len(data))
+	if _, err := f.ReadAt(got, 100); err != nil {
+		t.Fatalf("ReadAt: %v", err)
+	}
+	if !bytes.Equal(got, data) {
+		t.Errorf("got %q, want %q", got, data)
+	}
+}
+
+func TestStandard_EmptyBatch(t *testing.T) {
+	f := tempFile(t)
+	bio := NewStandard()
+	defer bio.Close()
+
+	// Empty ops should succeed.
+	if err := bio.PreadBatch(f, nil); err != nil {
+		t.Errorf("PreadBatch(nil): %v", err)
+	}
+	if err := bio.PwriteBatch(f, nil); err != nil {
+		t.Errorf("PwriteBatch(nil): %v", err)
+	}
+	if err := bio.PreadBatch(f, []Op{}); err != nil {
+		t.Errorf("PreadBatch([]): %v", err)
+	}
+	if err := bio.PwriteBatch(f, []Op{}); err != nil {
+		t.Errorf("PwriteBatch([]): %v", err)
+	}
+}
+
+func TestStandard_Close(t *testing.T) {
+	bio := NewStandard()
+	if err := bio.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	// Double close should be safe.
+	if err := bio.Close(); err != nil {
+		t.Fatalf("Close (2nd): %v", err)
+	}
+}
+
+func TestStandard_LargeBatch(t *testing.T) {
+	f := tempFile(t)
+	bio := NewStandard()
+	defer bio.Close()
+
+	// Write 100 blocks of 4KB each.
+	const blockSize = 4096
+	const numBlocks = 100
+
+	ops := make([]Op, numBlocks)
+	for i := range ops {
+		buf := make([]byte, blockSize)
+		for j := range buf {
+			buf[j] = byte(i)
+		}
+		ops[i] = Op{Buf: buf, Offset: int64(i) * blockSize}
+	}
+
+	if err := bio.PwriteBatch(f, ops); err != nil {
+		t.Fatalf("PwriteBatch: %v", err)
+	}
+	if err := bio.Fsync(f); err != nil {
+		t.Fatalf("Fsync: %v", err)
+	}
+
+	// Read back and verify.
+	readOps := make([]Op, numBlocks)
+	for i := range readOps {
+		readOps[i] = Op{Buf: make([]byte, blockSize), Offset: int64(i) * blockSize}
+	}
+	if err := bio.PreadBatch(f, readOps); err != nil {
+		t.Fatalf("PreadBatch: %v", err)
+	}
+
+	for i, op := range readOps {
+		expected := byte(i)
+		for j, b := range op.Buf {
+			if b != expected {
+				t.Fatalf("block %d byte %d: got %d, want %d", i, j, b, expected)
+			}
+		}
+	}
+}
+
+func TestStandard_PreadBatch_ErrorOnInvalidFd(t *testing.T) {
+	bio := NewStandard()
+	defer bio.Close()
+
+	// Create and close a file to get an invalid fd.
+	f, err := os.CreateTemp(t.TempDir(), "batchio-bad-*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.Close()
+
+	ops := []Op{{Buf: make([]byte, 10), Offset: 0}}
+	if err := bio.PreadBatch(f, ops); err == nil {
+		t.Error("expected error reading from closed file")
+	}
+}
+
+func TestStandard_PwriteBatch_ErrorOnReadOnly(t *testing.T) {
+	bio := NewStandard()
+	defer bio.Close()
+
+	// Create a file, write data, close, reopen read-only.
+	dir := t.TempDir()
+	path := filepath.Join(dir, "readonly")
+	if err := os.WriteFile(path, []byte("data"), 0644); err != nil {
+		t.Fatal(err)
+	}
+	f, err := os.Open(path) // read-only
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+
+	ops := []Op{{Buf: []byte("new"), Offset: 0}}
+	if err := bio.PwriteBatch(f, ops); err == nil {
+		t.Error("expected error writing to read-only file")
+	}
+}

weed/storage/blockvol/config.go

@@ -19,6 +19,7 @@ type BlockVolConfig struct {
 	WALSoftWatermark       float64       // WAL fraction above which writes begin throttling (default 0.7)
 	WALHardWatermark       float64       // WAL fraction above which writes block until drain (default 0.9)
 	WALMaxConcurrentWrites int           // max concurrent writers in WAL append path (default 16)
+	UseIOUring             bool          // opt-in: use io_uring for batch flusher I/O (Linux 5.6+ only)
 }
 
 // DefaultConfig returns a BlockVolConfig with production defaults.

weed/storage/blockvol/flusher.go

@@ -7,6 +7,8 @@ import (
 	"os"
 	"sync"
 	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/blockvol/batchio"
 )
 
 // Flusher copies WAL entries to the extent region and frees WAL space.
@@ -35,6 +37,7 @@ type Flusher struct {
 	snapMu    sync.RWMutex
 	snapshots []*activeSnapshot
 
+	bio     batchio.BatchIO // batch I/O backend (default: standard sequential)
 	logger  *log.Logger
 	lastErr bool // true if last FlushOnce returned error
 	metrics *EngineMetrics
@@ -55,6 +58,7 @@ type FlusherConfig struct {
 	Interval    time.Duration    // default 100ms
 	Logger      *log.Logger      // optional; defaults to log.Default()
 	Metrics     *EngineMetrics   // optional; if nil, no metrics recorded
+	BatchIO     batchio.BatchIO  // optional; defaults to batchio.NewStandard()
 }
 
 // NewFlusher creates a flusher. Call Run() in a goroutine.
@@ -65,6 +69,9 @@ func NewFlusher(cfg FlusherConfig) *Flusher {
 	if cfg.Logger == nil {
 		cfg.Logger = log.Default()
 	}
+	if cfg.BatchIO == nil {
+		cfg.BatchIO = batchio.NewStandard()
+	}
 	return &Flusher{
 		fd:             cfg.FD,
 		super:          cfg.Super,
@@ -74,6 +81,7 @@ func NewFlusher(cfg FlusherConfig) *Flusher {
 		walSize:        cfg.Super.WALSize,
 		blockSize:      cfg.Super.BlockSize,
 		extentStart:    cfg.Super.WALOffset + cfg.Super.WALSize,
+		bio:            cfg.BatchIO,
 		logger:         cfg.Logger,
 		metrics:        cfg.Metrics,
 		checkpointLSN:  cfg.Super.WALCheckpointLSN,
@@ -248,69 +256,53 @@ func (f *Flusher) flushOnceLocked() error {
 		}
 	}
 
-	// --- Phase 2: Extent writes (existing, unchanged) ---
+	// --- Phase 2: Extent writes via BatchIO ---
 	var maxLSN uint64
 	var maxWALEnd uint64
 
-	for _, e := range entries {
-		// Read the WAL entry and copy data to extent region.
-		headerBuf := make([]byte, walEntryHeaderSize)
-		absWALOff := int64(f.walOffset + e.WalOffset)
-		if _, err := f.fd.ReadAt(headerBuf, absWALOff); err != nil {
-			return fmt.Errorf("flusher: read WAL header at %d: %w", absWALOff, err)
+	// Step 2a: Batch-read WAL headers.
+	headerOps := make([]batchio.Op, len(entries))
+	for i, e := range entries {
+		headerOps[i] = batchio.Op{
+			Buf:    make([]byte, walEntryHeaderSize),
+			Offset: int64(f.walOffset + e.WalOffset),
 		}
+	}
+	if err := f.bio.PreadBatch(f.fd, headerOps); err != nil {
+		return fmt.Errorf("flusher: batch read WAL headers: %w", err)
+	}
 
-		// WAL reuse guard: validate LSN before trusting the entry.
-		entryLSN := binary.LittleEndian.Uint64(headerBuf[0:8])
+	// Step 2b: Identify entries needing full WAL read, batch-read them.
+	type pendingEntry struct {
+		idx       int    // index into entries
+		entryType uint8
+		entryLen  int
+	}
+	var pending []pendingEntry
+
+	for i, e := range entries {
+		hdr := headerOps[i].Buf
+		entryLSN := binary.LittleEndian.Uint64(hdr[0:8])
 		if entryLSN != e.Lsn {
-			continue // stale --WAL slot reused, skip this entry
+			continue // stale — WAL slot reused
 		}
 
-		// Parse entry type and length.
-		entryType := headerBuf[16] // Type at LSN(8)+Epoch(8)=16
-		dataLen := parseLength(headerBuf)
-
-		if entryType == EntryTypeWrite && dataLen > 0 {
-			// Read full entry.
-			entryLen := walEntryHeaderSize + int(dataLen)
-			fullBuf := make([]byte, entryLen)
-			if _, err := f.fd.ReadAt(fullBuf, absWALOff); err != nil {
-				return fmt.Errorf("flusher: read WAL entry at %d: %w", absWALOff, err)
-			}
-
-			entry, err := DecodeWALEntry(fullBuf)
-			if err != nil {
-				continue // corrupt or partially overwritten --skip
-			}
-
-			if e.Lba < entry.LBA {
-				continue // LBA mismatch --stale entry
-			}
-			blockIdx := e.Lba - entry.LBA
-			dataStart := blockIdx * uint64(f.blockSize)
-			if dataStart+uint64(f.blockSize) <= uint64(len(entry.Data)) {
-				extentOff := int64(f.extentStart + e.Lba*uint64(f.blockSize))
-				blockData := entry.Data[dataStart : dataStart+uint64(f.blockSize)]
-				if _, err := f.fd.WriteAt(blockData, extentOff); err != nil {
-					return fmt.Errorf("flusher: write extent at LBA %d: %w", e.Lba, err)
-				}
-			}
-
-			walEnd := e.WalOffset + uint64(entryLen)
-			if walEnd > maxWALEnd {
-				maxWALEnd = walEnd
+		entryType := hdr[16]
+		if entryType == EntryTypeWrite {
+			dataLen := parseLength(hdr)
+			if dataLen > 0 {
+				pending = append(pending, pendingEntry{
+					idx:       i,
+					entryType: entryType,
+					entryLen:  walEntryHeaderSize + int(dataLen),
+				})
 			}
 		} else if entryType == EntryTypeTrim {
-			zeroBlock := make([]byte, f.blockSize)
-			extentOff := int64(f.extentStart + e.Lba*uint64(f.blockSize))
-			if _, err := f.fd.WriteAt(zeroBlock, extentOff); err != nil {
-				return fmt.Errorf("flusher: zero extent at LBA %d: %w", e.Lba, err)
-			}
-
-			walEnd := e.WalOffset + uint64(walEntryHeaderSize)
-			if walEnd > maxWALEnd {
-				maxWALEnd = walEnd
-			}
+			pending = append(pending, pendingEntry{
+				idx:       i,
+				entryType: entryType,
+				entryLen:  walEntryHeaderSize,
+			})
 		}
 
 		if e.Lsn > maxLSN {
@@ -318,8 +310,76 @@ func (f *Flusher) flushOnceLocked() error {
 		}
 	}
 
-	// Fsync extent writes.
-	if err := f.fd.Sync(); err != nil {
+	// Batch-read full WAL entries for write ops.
+	var walReadOps []batchio.Op
+	for _, p := range pending {
+		if p.entryType == EntryTypeWrite {
+			walReadOps = append(walReadOps, batchio.Op{
+				Buf:    make([]byte, p.entryLen),
+				Offset: int64(f.walOffset + entries[p.idx].WalOffset),
+			})
+		}
+	}
+	if len(walReadOps) > 0 {
+		if err := f.bio.PreadBatch(f.fd, walReadOps); err != nil {
+			return fmt.Errorf("flusher: batch read WAL entries: %w", err)
+		}
+	}
+
+	// Step 2c: Decode entries and build extent write ops.
+	var extentWriteOps []batchio.Op
+	walReadI := 0
+
+	for _, p := range pending {
+		e := entries[p.idx]
+		if p.entryType == EntryTypeWrite {
+			fullBuf := walReadOps[walReadI].Buf
+			walReadI++
+
+			entry, err := DecodeWALEntry(fullBuf)
+			if err != nil {
+				continue
+			}
+			if e.Lba < entry.LBA {
+				continue
+			}
+			blockIdx := e.Lba - entry.LBA
+			dataStart := blockIdx * uint64(f.blockSize)
+			if dataStart+uint64(f.blockSize) <= uint64(len(entry.Data)) {
+				extentOff := int64(f.extentStart + e.Lba*uint64(f.blockSize))
+				blockData := entry.Data[dataStart : dataStart+uint64(f.blockSize)]
+				extentWriteOps = append(extentWriteOps, batchio.Op{
+					Buf:    blockData,
+					Offset: extentOff,
+				})
+			}
+
+			walEnd := e.WalOffset + uint64(p.entryLen)
+			if walEnd > maxWALEnd {
+				maxWALEnd = walEnd
+			}
+		} else if p.entryType == EntryTypeTrim {
+			zeroBlock := make([]byte, f.blockSize)
+			extentOff := int64(f.extentStart + e.Lba*uint64(f.blockSize))
+			extentWriteOps = append(extentWriteOps, batchio.Op{
+				Buf:    zeroBlock,
+				Offset: extentOff,
+			})
+
+			walEnd := e.WalOffset + uint64(walEntryHeaderSize)
+			if walEnd > maxWALEnd {
+				maxWALEnd = walEnd
+			}
+		}
+	}
+
+	// Step 2d: Batch-write extents + fsync.
+	if len(extentWriteOps) > 0 {
+		if err := f.bio.PwriteBatch(f.fd, extentWriteOps); err != nil {
+			return fmt.Errorf("flusher: batch write extents: %w", err)
+		}
+	}
+	if err := f.bio.Fsync(f.fd); err != nil {
 		return fmt.Errorf("flusher: fsync extent: %w", err)
 	}