From f99cc6ca94f82208901cb5b9d39779a8891e8429 Mon Sep 17 00:00:00 2001 From: Mohd Quamar Tyagi <104281681+Tyagiquamar@users.noreply.github.com> Date: Sun, 8 Nov 2026 18:33:52 +0530 Subject: [PATCH] volume.fsck: batch and deduplicate filer entry deletions (#12480) * volume.fsck: batch and deduplicate filer entry deletions In volume.fsck with -findMissingChunksInFiler and -reallyDeleteFilerEntries, deleting orphaned filer entries was slow and CPU intensive due to tiny worker channel buffers (128), lack of batch processing, redundant HTTP DELETE requests for multi-chunk files, and lock contention on parent directory tracking. Optimize by: - Adding startPurgeBatchWorkers with 4096-buffered channels to prevent producer stalls and batch opportunistic deletes up to 64 items. - Deduplicating missing chunks per file path during volume checks so that multi-chunk files are queued once. - Processing batches with in-batch path deduplication and aggregating parent directories before updating purgedDirs in a single lock acquisition. - Buffering temporary fid file writes (64KB) during the traversal to avoid millions of unbuffered syscalls. * volume.fsck: report a failed write of the filer fid files The fid files are written through a buffer now, so the last part of each one reaches the disk only in the deferred Flush. Its error was dropped: a full temp disk left a short fid file and the collection still succeeded, and the volume check would then take the chunks missing from it for orphans. Co-Authored-By: Claude Opus 5.5 * volume.fsck: retry a batched path whose earlier delete failed A path was marked done before its delete ran, so a second copy of it in the same batch was skipped after a transient failure and the entry stayed. Only skip paths already deleted. Co-Authored-By: Claude Opus 5.5 * volume.fsck: keep reporting a failed DELETE response read The non-verbose path now discards the body without checking the copy, so a connection cut after the headers went unreported. Co-Authored-By: Claude Opus 5.5 * volume.fsck: keep one purge worker starter and one delete function startPurgeWorkers and the single-path httpDelete were left only as wrappers the tests called. Give their names to the batch versions and adapt the tests. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/shell/command_volume_fsck.go | 136 ++++++++++++----- weed/shell/command_volume_fsck_test.go | 201 ++++++++++++++++++++++++- 2 files changed, 293 insertions(+), 44 deletions(-) diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index efabf1ceb1..c826615997 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -323,12 +323,23 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io. return nil } -func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, queuePurge func(purgeItem), collectModifyFromAtNs int64, cutoffFromAtNs int64) error { +func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, purgeAbsent bool, queuePurge func(purgeItem), collectModifyFromAtNs int64, cutoffFromAtNs int64) (err error) { if *c.verbose { fmt.Fprintf(c.writer, "checking each file from filer path /...\n") } files := make(map[uint32]*os.File) + writers := make(map[uint32]*bufio.Writer) + defer func() { + for vid, f := range files { + if flushErr := writers[vid].Flush(); flushErr != nil && err == nil { + err = fmt.Errorf("failed to write file %s: %w", f.Name(), flushErr) + } + if closeErr := f.Close(); closeErr != nil && err == nil { + err = fmt.Errorf("failed to close file %s: %w", f.Name(), closeErr) + } + } + }() for _, volumeIdToServer := range dataNodeVolumeIdToVInfo { for vid := range volumeIdToServer { if _, ok := files[vid]; ok { @@ -339,13 +350,9 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m return fmt.Errorf("failed to create file %s: %v", getFilerFileIdFile(c.tempFolder, vid), openErr) } files[vid] = dst + writers[vid] = bufio.NewWriterSize(dst, 64*1024) } } - defer func() { - for _, f := range files { - f.Close() - } - }() return doTraverseAndSaving(c.env, c.writer, "/", false, false, func(ctx context.Context, entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) { @@ -415,21 +422,21 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m buffer := make([]byte, readbufferSize) for item := range outputChan { i := item.(*Item) - if f, ok := files[i.vid]; ok { + if w, ok := writers[i.vid]; ok { util.Uint64toBytes(buffer, i.fileKey) util.Uint32toBytes(buffer[8:], i.cookie) util.Uint32toBytes(buffer[12:], uint32(len(i.path))) - if _, err := f.Write(buffer); err != nil { + if _, err := w.Write(buffer); err != nil { return err } - if _, err := f.Write([]byte(i.path)); err != nil { + if _, err := w.Write([]byte(i.path)); err != nil { return err } util.Uint32toBytes(buffer[:4], uint32(len(i.hardLinkId))) - if _, err := f.Write(buffer[:4]); err != nil { + if _, err := w.Write(buffer[:4]); err != nil { return err } - if _, err := f.Write(i.hardLinkId); err != nil { + if _, err := w.Write(i.hardLinkId); err != nil { return err } } else if *c.findMissingChunksInFiler { @@ -690,8 +697,9 @@ func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleI } defer fp.Close() - br := bufio.NewReader(fp) + br := bufio.NewReaderSize(fp, 64*1024) buffer := make([]byte, readbufferSize) + pathBuf := make([]byte, 512) var readSize int var readErr error item := &Item{vid: volumeId} @@ -709,15 +717,17 @@ func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleI item.fileKey = util.BytesToUint64(buffer[:8]) item.cookie = util.BytesToUint32(buffer[8:12]) pathSize := util.BytesToUint32(buffer[12:16]) - pathBytes := make([]byte, int(pathSize)) - n, err := io.ReadFull(br, pathBytes) + if int(pathSize) > len(pathBuf) { + pathBuf = make([]byte, pathSize) + } + n, err := io.ReadFull(br, pathBuf[:pathSize]) if err != nil { return fmt.Errorf("read fid path for volume %d,%x%08x: %w", volumeId, item.fileKey, item.cookie, err) } if n != int(pathSize) { return fmt.Errorf("read fid path size mismatch for volume %d,%x%08x: got %d want %d", volumeId, item.fileKey, item.cookie, n, pathSize) } - item.path = util.FullPath(pathBytes) + item.path = util.FullPath(string(pathBuf[:pathSize])) item.hardLinkId = nil if _, err := io.ReadFull(br, buffer[:4]); err != nil { return fmt.Errorf("read fid hard link id size for volume %d,%x%08x: %w", volumeId, item.fileKey, item.cookie, err) @@ -745,8 +755,13 @@ func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, vo if err = db.LoadFromIdx(getVolumeFileIdFile(c.tempFolder, dataNodeId, volumeId)); err != nil { return } + seenPaths := make(map[util.FullPath]struct{}) if err = c.readFilerFileIdFile(volumeId, func(needleId types.NeedleId, itemPath util.FullPath, hardLinkId []byte) { if _, found := db.Get(needleId); !found { + if _, seen := seenPaths[itemPath]; seen { + return + } + seenPaths[itemPath] = struct{}{} fmt.Fprintf(c.writer, "%s\n", itemPath) if queuePurge != nil { queuePurge(purgeItem{path: itemPath, hardLinkId: hardLinkId}) @@ -763,25 +778,44 @@ type purgeItem struct { hardLinkId []byte } -// startPurgeWorkers fans deletes across at least one worker and returns a -// queue function plus a finish function that closes the channels and waits -// for in-flight deletes. Paths that share a hard link id always land on the -// same worker, so their shared hard-link counter updates stay serialized — -// concurrent deletes of the same id could both read count N and write N-1, -// leaving the shared record behind after every path is gone. -func startPurgeWorkers(workers int, deleteFn func(util.FullPath)) (queue func(purgeItem), finish func()) { +const ( + purgeBatchSize = 64 + purgeChannelBufferSize = 4096 +) + +// startPurgeWorkers fans deletes across at least one worker, handing each +// one batches of the paths already queued for it, and returns a queue +// function plus a finish function that closes the channels and waits for +// in-flight deletes. Paths that share a hard link id always land on the +// same worker, so their shared hard-link counter updates stay serialized. +func startPurgeWorkers(workers int, deleteFn func([]purgeItem)) (queue func(purgeItem), finish func()) { if workers < 1 { workers = 1 } chans := make([]chan purgeItem, workers) var wg sync.WaitGroup for w := range chans { - chans[w] = make(chan purgeItem, 128) + chans[w] = make(chan purgeItem, purgeChannelBufferSize) wg.Add(1) go func(ch <-chan purgeItem) { defer wg.Done() + batch := make([]purgeItem, 0, purgeBatchSize) for it := range ch { - deleteFn(it.path) + batch = append(batch, it) + drain: + for len(batch) < purgeBatchSize { + select { + case next, ok := <-ch: + if !ok { + break drain + } + batch = append(batch, next) + default: + break drain + } + } + deleteFn(batch) + batch = batch[:0] } }(chans[w]) } @@ -799,11 +833,38 @@ func startPurgeWorkers(workers int, deleteFn func(util.FullPath)) (queue func(pu } } -func (c *commandVolumeFsck) httpDelete(path util.FullPath) { +func (c *commandVolumeFsck) httpDelete(items []purgeItem) { + if len(items) == 0 { + return + } + deleted := make(map[util.FullPath]struct{}, len(items)) + localPurgedDirs := make(map[util.FullPath]struct{}) + + for _, it := range items { + if _, ok := deleted[it.path]; ok { + continue + } + if c.doHttpDelete(it.path) { + deleted[it.path] = struct{}{} + dir, _ := it.path.DirAndName() + localPurgedDirs[util.FullPath(dir)] = struct{}{} + } + } + + if len(localPurgedDirs) > 0 { + c.purgedDirsLock.Lock() + for dir := range localPurgedDirs { + c.purgedDirs[dir] = struct{}{} + } + c.purgedDirsLock.Unlock() + } +} + +func (c *commandVolumeFsck) doHttpDelete(path util.FullPath) bool { req, err := http.NewRequest(http.MethodDelete, "", nil) if err != nil { fmt.Fprintf(c.writer, "HTTP delete request error: %v\n", err) - return + return false } req.URL = &url.URL{ @@ -824,28 +885,29 @@ func (c *commandVolumeFsck) httpDelete(path util.FullPath) { resp, err := util_http.GetGlobalHttpClient().Do(req) if err != nil { fmt.Fprintf(c.writer, "DELETE fetch error: %v\n", err) - return + return false } defer resp.Body.Close() - _, err = io.ReadAll(resp.Body) - if err != nil { - fmt.Fprintf(c.writer, "DELETE response error: %v\n", err) - } - if *c.verbose { + body, err := io.ReadAll(resp.Body) + if err != nil { + fmt.Fprintf(c.writer, "DELETE response error: %v\n", err) + } fmt.Fprintln(c.writer, "delete response Status : ", resp.Status) fmt.Fprintln(c.writer, "delete response Headers : ", resp.Header) + if len(body) > 0 { + fmt.Fprintln(c.writer, "delete response Body : ", string(body)) + } + } else if _, err := io.Copy(io.Discard, resp.Body); err != nil { + fmt.Fprintf(c.writer, "DELETE response error: %v\n", err) } if resp.StatusCode >= http.StatusMultipleChoices { fmt.Fprintf(c.writer, "DELETE %s failed: %s\n", path, resp.Status) - } else { - dir, _ := path.DirAndName() - c.purgedDirsLock.Lock() - c.purgedDirs[util.FullPath(dir)] = struct{}{} - c.purgedDirsLock.Unlock() + return false } + return true } // purgeEmptyDirectories removes the directories emptied by the purged entries, walking up while each parent is empty too. diff --git a/weed/shell/command_volume_fsck_test.go b/weed/shell/command_volume_fsck_test.go index 5993688436..33281872fd 100644 --- a/weed/shell/command_volume_fsck_test.go +++ b/weed/shell/command_volume_fsck_test.go @@ -64,8 +64,7 @@ func TestVolumeFsckHttpDeleteRecordsParentDirectory(t *testing.T) { env: &CommandEnv{option: &ShellOptions{FilerAddress: pb.ServerAddress(strings.TrimPrefix(filer.URL, "http://"))}}, } - c.httpDelete("/orphan/dir/deep/file.txt") - c.httpDelete("/locked/dir/file.txt") + c.httpDelete([]purgeItem{{path: "/orphan/dir/deep/file.txt"}, {path: "/locked/dir/file.txt"}}) if _, found := c.purgedDirs["/orphan/dir/deep"]; !found { t.Errorf("expected /orphan/dir/deep to be recorded, got %v", c.purgedDirs) @@ -267,10 +266,18 @@ func TestVolumeFsckPurgeEmptyDirectoriesKeepsChangedDirectory(t *testing.T) { // The purge pool may run at most `workers` deletes at once, and at least one // even when the flag is set to zero. +func eachPurgePath(deleteFn func(util.FullPath)) func([]purgeItem) { + return func(batch []purgeItem) { + for _, it := range batch { + deleteFn(it.path) + } + } +} + func TestVolumeFsckPurgeWorkerLimit(t *testing.T) { gate := make(chan struct{}) var inFlight, maxSeen, processed atomic.Int32 - queue, finish := startPurgeWorkers(3, func(util.FullPath) { + queue, finish := startPurgeWorkers(3, eachPurgePath(func(util.FullPath) { n := inFlight.Add(1) for { m := maxSeen.Load() @@ -281,7 +288,7 @@ func TestVolumeFsckPurgeWorkerLimit(t *testing.T) { <-gate inFlight.Add(-1) processed.Add(1) - }) + })) for i := 0; i < 10; i++ { queue(purgeItem{path: util.FullPath(fmt.Sprintf("/p/%d", i))}) @@ -306,7 +313,7 @@ func TestVolumeFsckPurgeWorkerLimit(t *testing.T) { func TestVolumeFsckPurgeWorkersAtLeastOne(t *testing.T) { var processed atomic.Int32 - queue, finish := startPurgeWorkers(0, func(util.FullPath) { processed.Add(1) }) + queue, finish := startPurgeWorkers(0, eachPurgePath(func(util.FullPath) { processed.Add(1) })) for i := 0; i < 5; i++ { queue(purgeItem{path: util.FullPath(fmt.Sprintf("/p/%d", i))}) } @@ -322,7 +329,7 @@ func TestVolumeFsckPurgeWorkersAtLeastOne(t *testing.T) { func TestVolumeFsckPurgeHardLinksSerialized(t *testing.T) { var inFlight, maxSeen, processed atomic.Int32 gate := make(chan struct{}) - queue, finish := startPurgeWorkers(8, func(util.FullPath) { + queue, finish := startPurgeWorkers(8, eachPurgePath(func(util.FullPath) { n := inFlight.Add(1) for { m := maxSeen.Load() @@ -333,7 +340,7 @@ func TestVolumeFsckPurgeHardLinksSerialized(t *testing.T) { <-gate inFlight.Add(-1) processed.Add(1) - }) + })) for i := 0; i < 8; i++ { queue(purgeItem{path: util.FullPath(fmt.Sprintf("/hl/%d", i)), hardLinkId: []byte{9, 9, 9}}) } @@ -514,3 +521,183 @@ func TestReadFilerFileIdFileRejectsTruncatedHardLinkId(t *testing.T) { t.Fatal("expected an error for a truncated hard link id size") } } + +func TestVolumeFsckPurgeBatchWorkersProcessesBatches(t *testing.T) { + var totalProcessed atomic.Int32 + var maxBatchSeen atomic.Int32 + var batchCount atomic.Int32 + + batchGate := make(chan struct{}) + queue, finish := startPurgeWorkers(1, func(batch []purgeItem) { + <-batchGate + count := int32(len(batch)) + totalProcessed.Add(count) + batchCount.Add(1) + for { + cur := maxBatchSeen.Load() + if count <= cur || maxBatchSeen.CompareAndSwap(cur, count) { + break + } + } + }) + + // Queue 10 items while worker is gated + for i := 0; i < 10; i++ { + queue(purgeItem{path: util.FullPath(fmt.Sprintf("/item/%d", i))}) + } + close(batchGate) + finish() + + if got := totalProcessed.Load(); got != 10 { + t.Fatalf("processed %d items, want 10", got) + } + if max := maxBatchSeen.Load(); max < 2 { + t.Fatalf("expected batch size >= 2 when draining queued items, got %d", max) + } +} + +func TestVolumeFsckCheckOneVolumeDeduplicatesMissingChunksForSamePath(t *testing.T) { + tempDir := t.TempDir() + var verbose bool + var buf bytes.Buffer + c := &commandVolumeFsck{tempFolder: tempDir, verbose: &verbose, writer: &buf} + + volumeId := uint32(8) + db := needle_map.NewMemDb() + if err := db.SaveToIdx(getVolumeFileIdFile(tempDir, "dn1", volumeId)); err != nil { + t.Fatalf("write empty volume idx: %v", err) + } + db.Close() + + f, err := os.Create(getFilerFileIdFile(tempDir, volumeId)) + if err != nil { + t.Fatalf("create fid file: %v", err) + } + writeFid := func(fileKey uint64, cookie uint32, p string, hardLinkId []byte) { + buffer := make([]byte, readbufferSize) + util.Uint64toBytes(buffer, fileKey) + util.Uint32toBytes(buffer[8:], cookie) + util.Uint32toBytes(buffer[12:], uint32(len(p))) + if _, err := f.Write(buffer); err != nil { + t.Fatalf("write fid header: %v", err) + } + if _, err := f.Write([]byte(p)); err != nil { + t.Fatalf("write fid path: %v", err) + } + util.Uint32toBytes(buffer[:4], uint32(len(hardLinkId))) + if _, err := f.Write(buffer[:4]); err != nil { + t.Fatalf("write fid hard link id size: %v", err) + } + if _, err := f.Write(hardLinkId); err != nil { + t.Fatalf("write fid hard link id: %v", err) + } + } + + // Two chunks for the same file, both missing from volume idx + writeFid(101, 1, "/orphan/same_file.txt", nil) + writeFid(102, 1, "/orphan/same_file.txt", nil) + // Another file + writeFid(103, 1, "/orphan/another_file.txt", nil) + if err := f.Close(); err != nil { + t.Fatalf("close fid file: %v", err) + } + + var queued []purgeItem + if err := c.oneVolumeFileIdsCheckOneVolume("dn1", volumeId, func(it purgeItem) { + queued = append(queued, it) + }); err != nil { + t.Fatalf("check volume: %v", err) + } + + if len(queued) != 2 { + t.Fatalf("queued %d purges, want 2 (deduplicated): %v", len(queued), queued) + } + if queued[0].path != "/orphan/same_file.txt" || queued[1].path != "/orphan/another_file.txt" { + t.Fatalf("queued paths %v, want /orphan/same_file.txt and /orphan/another_file.txt", queued) + } +} + +func TestVolumeFsckHttpDeleteBatchDeduplicatesAndRecordsDirs(t *testing.T) { + var requestCount, flakyCount atomic.Int32 + filer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestCount.Add(1) + if strings.HasPrefix(r.URL.Path, "/fail/") { + w.WriteHeader(http.StatusNotFound) + return + } + if strings.HasPrefix(r.URL.Path, "/flaky/") && flakyCount.Add(1) == 1 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusNoContent) + })) + defer filer.Close() + + verbose := false + var output bytes.Buffer + c := &commandVolumeFsck{ + verbose: &verbose, + writer: &output, + purgedDirs: make(map[util.FullPath]struct{}), + env: &CommandEnv{option: &ShellOptions{FilerAddress: pb.ServerAddress(strings.TrimPrefix(filer.URL, "http://"))}}, + } + + items := []purgeItem{ + {path: "/dir1/file1.txt"}, + {path: "/dir1/file1.txt"}, // duplicate within batch + {path: "/dir2/file2.txt"}, + {path: "/fail/file3.txt"}, + {path: "/flaky/file4.txt"}, + {path: "/flaky/file4.txt"}, // retried after the first delete failed + } + + c.httpDelete(items) + + if got := requestCount.Load(); got != 5 { + t.Fatalf("received %d requests, want 5", got) + } + if _, found := c.purgedDirs["/flaky"]; !found { + t.Errorf("expected /flaky to be recorded after the retried delete") + } + if _, found := c.purgedDirs["/dir1"]; !found { + t.Errorf("expected /dir1 to be recorded in purgedDirs") + } + if _, found := c.purgedDirs["/dir2"]; !found { + t.Errorf("expected /dir2 to be recorded in purgedDirs") + } + if _, found := c.purgedDirs["/fail"]; found { + t.Errorf("expected /fail not to be recorded in purgedDirs on 404") + } +} + +func TestVolumeFsckCollectFilerFileIdsReportsFailedFlush(t *testing.T) { + if _, err := os.Stat("/dev/full"); err != nil { + t.Skip("needs /dev/full") + } + tempDir := t.TempDir() + if err := os.Symlink("/dev/full", getFilerFileIdFile(tempDir, 7)); err != nil { + t.Fatalf("link fid file to /dev/full: %v", err) + } + directory := &filer_pb.Entry{IsDirectory: true} + stub := &stubFilerServer{entries: map[string]*filer_pb.Entry{ + "/": directory, + "/file": {Chunks: []*filer_pb.FileChunk{{Size: 1, Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 11, Cookie: 1}}}}, + }} + + verbose, findMissing, verify, force := false, false, false, false + c := &commandVolumeFsck{ + tempFolder: tempDir, + verbose: &verbose, + findMissingChunksInFiler: &findMissing, + verifyNeedle: &verify, + forcePurging: &force, + writer: io.Discard, + bucketsPath: "/buckets", + env: startStubFiler(t, stub), + } + + vinfo := map[string]map[uint32]VInfo{"dn1": {7: {}}} + if err := c.collectFilerFileIdAndPaths(vinfo, false, func(purgeItem) {}, 0, 0); err == nil { + t.Fatal("expected the failed fid file write to be reported") + } +}