From b50d4a9cabcf4c226079a9638b5c1a506ce8cc58 Mon Sep 17 00:00:00 2001 From: Yip Leo Date: Sat, 14 Nov 2026 13:56:44 +0800 Subject: [PATCH] mount: don't spend the close()/fsync commit deadline on uploads (#12641) * mount: start the close()/fsync commit deadline after the data upload Flush, Fsync and the truncate path of SetAttr create the 30 s metadataFlushTimeout context on entry and pass it to doFlush, so the clock also runs through FlushData, the upload of the dirty pages. That upload has its own timeouts and retries. When an assign through a stalled filer or a slow volume server takes most of the 30 s, CreateEntry then fails at once with "context deadline exceeded" and close() returns EIO for data that was already stored. The deadline is meant for the commit alone (#10102: "a healthy CreateEntry completes in well under a second"). doFlush now takes the commit timeout and starts the deadline once the data is uploaded. close() still cannot hang on a stuck filer: the commit keeps its 30 s bound. Release and the flush before a rename pass 0, as before (no overall bound, only the per-call filer timeouts). Seen in production on a training cluster: a 30 s stall of one replica's disk, and a filer that stalled for about 30 s, each failed close() with EIO although the upload had succeeded. Tests: close(), fsync() and a truncate below dirty data, each after an assign slower than the deadline (all three fail before this change), and a filer that never answers CreateEntry still fails close() at the deadline. Co-Authored-By: Claude Opus 5.5 * mount: charge the flush commit deadline only for its filer calls Starting the 30 s deadline after the data upload is not enough. flushMetadataToFiler does more than call the filer before it commits: it waits for the handle's locks, reads the file back and uploads it again when its chunks are bloated (maybeMergeChunks), and uploads a manifest once enough chunks have built up (MaybeManifestize). None of that takes the context. When it outlasts the deadline, CreateEntry again fails at once with "context deadline exceeded", and close() or fsync() returns an error for data that was stored. The deadline becomes a budget, commitBudget, charged only while a filer call of the commit runs: CreateEntry, and the read of a changed hard link before it. The attempts of one flush share it, and a flush is not retried once it is spent, so a stuck filer still fails close() after 30 s. Release, the flush before a rename and the async flush pass no budget, as before. Seen in production: the merge of a 1.8 MiB sqlite file with 183 chunks took 32 s because one chunk's read first tried a volume server that was down, and the commit failed 32 microseconds after the merge finished. Tests: a flush whose manifest upload is slower than the deadline commits (it fails without this change), and the filer calls of one flush share the budget. Co-Authored-By: Claude Fable 5.2 * mount: give the stalled-lookup refusal test the new doFlush argument TestContentFlushRefusalWithStalledLookup came in from master after this branch changed doFlush's signature, so the merged package did not build. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: Chris Lu --- weed/mount/content_flush_condition_test.go | 12 +- weed/mount/content_flush_recovery_test.go | 4 +- weed/mount/weedfs_async_flush.go | 2 +- weed/mount/weedfs_attr.go | 5 +- weed/mount/weedfs_attr_cas_test.go | 6 +- weed/mount/weedfs_file_io.go | 5 +- weed/mount/weedfs_file_sync.go | 101 +++++-- weed/mount/weedfs_flush_deadline_test.go | 276 ++++++++++++++++++ .../weedfs_invalidate_open_handle_test.go | 4 +- weed/mount/weedfs_link_open_handle_test.go | 10 +- weed/mount/weedfs_rename.go | 5 +- 11 files changed, 379 insertions(+), 51 deletions(-) create mode 100644 weed/mount/weedfs_flush_deadline_test.go diff --git a/weed/mount/content_flush_condition_test.go b/weed/mount/content_flush_condition_test.go index 68a5817670..0d47e1836b 100644 --- a/weed/mount/content_flush_condition_test.go +++ b/weed/mount/content_flush_condition_test.go @@ -125,7 +125,7 @@ func TestContentFlushPreservesPeerChunks(t *testing.T) { fh.dirtyMetadata = true fh.RememberPath(util.FullPath("/dir/file")) before := proto.Clone(current) - code := wfs.doFlush(context.Background(), fh, 1000, 20, false) + code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0) fake.mu.Lock() defer fake.mu.Unlock() if wantError { @@ -185,7 +185,7 @@ func TestContentFlushRepeatsOnOneHandle(t *testing.T) { e.Attributes.FileSize = i }) fh.dirtyMetadata = true - if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0); code != fuse.OK { t.Fatalf("flush %d: %v", i-1, code) } } @@ -226,7 +226,7 @@ func TestContentFlushRefusalWithStalledLookup(t *testing.T) { fh.dirtyMetadata = true fh.RememberPath(util.FullPath("/dir/file")) start := time.Now() - code := wfs.doFlush(context.Background(), fh, 1000, 20, false) + code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0) elapsed := time.Since(start) fake.mu.Lock() defer fake.mu.Unlock() @@ -298,7 +298,7 @@ func TestContentFlushMergeThenRepeat(t *testing.T) { fh.RememberPath(util.FullPath("/dir/file")) for i := uint64(4); i <= 5; i++ { fh.dirtyMetadata = true - if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0); code != fuse.OK { t.Fatalf("flush before chunk %d: %v", i, code) } fh.UpdateEntry(func(e *filer_pb.Entry) { @@ -307,7 +307,7 @@ func TestContentFlushMergeThenRepeat(t *testing.T) { }) } fh.dirtyMetadata = true - if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0); code != fuse.OK { t.Fatal(code) } fake.mu.Lock() @@ -473,7 +473,7 @@ func TestContentFlushRepublishesUnchangedEntry(t *testing.T) { fh.baseEntry.Store(observed) fh.dirtyMetadata = true fh.RememberPath(path) - if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0); code != fuse.OK { t.Fatal(code) } fake.mu.Lock() diff --git a/weed/mount/content_flush_recovery_test.go b/weed/mount/content_flush_recovery_test.go index ca4f582401..ecb372cfe4 100644 --- a/weed/mount/content_flush_recovery_test.go +++ b/weed/mount/content_flush_recovery_test.go @@ -141,7 +141,7 @@ func TestContentAsyncConflictKeepsRecoveryRecord(t *testing.T) { func TestContentRecoveryUsesFailedRequestNotLaterHandle(t *testing.T) { w, fh, _ := committedContentHandle(t, true) w.option.CacheDirForWrite = t.TempDir() - err := w.flushMetadataToFiler(context.Background(), fh, "/dir", "file", 1000, 20) + err := w.flushMetadataToFiler(context.Background(), fh, "/dir", "file", 1000, 20, nil) if err == nil { t.Fatal("expected refusal") } @@ -229,7 +229,7 @@ func TestContentRecoverySkipsAnotherUsersDirectory(t *testing.T) { if err := os.Mkdir(filepath.Join(w.option.CacheDirForWrite, "failed-metadata"), 0500); err != nil { t.Fatal(err) } - err := w.flushMetadataToFiler(context.Background(), fh, "/dir", "file", 1000, 20) + err := w.flushMetadataToFiler(context.Background(), fh, "/dir", "file", 1000, 20, nil) if err == nil { t.Fatal("expected refusal") } diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go index 383b4abeba..d3e7cbfa54 100644 --- a/weed/mount/weedfs_async_flush.go +++ b/weed/mount/weedfs_async_flush.go @@ -142,7 +142,7 @@ func (wfs *WFS) flushMetadataWithRetry(fh *FileHandle, dir, name string, fileFul attempts := 0 err := retryMetadataFlushUntil(context.Background(), func() error { attempts++ - return wfs.flushMetadataToFiler(context.Background(), fh, dir, name, fh.asyncFlushUid, fh.asyncFlushGid) + return wfs.flushMetadataToFiler(context.Background(), fh, dir, name, fh.asyncFlushUid, fh.asyncFlushGid, nil) }, isFilerUnreachable, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { if totalAttempts == 0 { glog.Warningf("completeAsyncFlush %s: filer unreachable, retrying metadata flush (attempt %d) after %v: %v", diff --git a/weed/mount/weedfs_attr.go b/weed/mount/weedfs_attr.go index 765574b9d5..c60cff7e8f 100644 --- a/weed/mount/weedfs_attr.go +++ b/weed/mount/weedfs_attr.go @@ -99,10 +99,7 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse // alone, pages beyond the new size come back with the next flush and // grow the file again, so turn them into chunks first. Runs before // the entry locks below: the flush takes its own. - ctx, cancelFunc := context.WithTimeout(context.Background(), metadataFlushTimeout) - flushStatus := wfs.doFlush(ctx, fh, input.Uid, input.Gid, false) - cancelFunc() - if flushStatus != fuse.OK { + if flushStatus := wfs.doFlush(context.Background(), fh, input.Uid, input.Gid, false, metadataFlushTimeout); flushStatus != fuse.OK { return flushStatus } } diff --git a/weed/mount/weedfs_attr_cas_test.go b/weed/mount/weedfs_attr_cas_test.go index 2fdede2d2c..f654334777 100644 --- a/weed/mount/weedfs_attr_cas_test.go +++ b/weed/mount/weedfs_attr_cas_test.go @@ -275,7 +275,7 @@ func TestAttributeOnlyOpsPreserveAcknowledgedAppends(t *testing.T) { t.Fatalf("reply uid %d, want the local id 1000", out.Attr.Uid) } if open { - if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0); code != fuse.OK { t.Fatalf("flush after attribute syscall: %v", code) } } @@ -395,7 +395,7 @@ func TestAcknowledgedAttributeSurvivesWriteDuringRPC(t *testing.T) { case <-time.After(10 * time.Second): t.Fatal("attribute RPC did not finish") } - if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code != fuse.OK { + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0); code != fuse.OK { t.Fatalf("flush: %v", code) } fake.mu.Lock() @@ -506,7 +506,7 @@ func TestAttributePublicationBeforeResponse(t *testing.T) { } flushDone := make(chan fuse.Status, 1) flushSettled, publicationEscaped := false, false - flush := func() { flushDone <- wfs.doFlush(context.Background(), fh, 1000, 20, false) } + flush := func() { flushDone <- wfs.doFlush(context.Background(), fh, 1000, 20, false, 0) } // Probe ownership while the committed response is held. On the old caller // no publication boundary exists, so force its whole-entry flush to commit // BEFORE releasing the response. With coordination, queue that same flush diff --git a/weed/mount/weedfs_file_io.go b/weed/mount/weedfs_file_io.go index fb1774f565..019e80d0c0 100644 --- a/weed/mount/weedfs_file_io.go +++ b/weed/mount/weedfs_file_io.go @@ -160,8 +160,9 @@ func (wfs *WFS) Release(cancel <-chan struct{}, in *fuse.ReleaseIn) { if fh := wfs.GetHandle(FileHandleId(in.Fh)); fh != nil { allowAsync := in.ReleaseFlags&fuse.FUSE_RELEASE_FLOCK_UNLOCK == 0 // Release is the last chance to persist the handle, so it must finish - // even if the triggering syscall was interrupted: non-cancellable context. - if status := wfs.doFlush(context.Background(), fh, in.Uid, in.Gid, allowAsync); status != fuse.OK { + // even if the triggering syscall was interrupted: non-cancellable context + // and no commit deadline. + if status := wfs.doFlush(context.Background(), fh, in.Uid, in.Gid, allowAsync, 0); status != fuse.OK { glog.Warningf("release fh %d inode %d: fallback flush failed: %v", in.Fh, in.NodeId, status) } } diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 491b2f0aec..a69016a102 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -17,11 +17,45 @@ import ( "google.golang.org/protobuf/proto" ) -// metadataFlushTimeout bounds a close()/fsync metadata flush so an overwhelmed -// filer cannot wedge the calling process forever. It is deliberately generous: -// a healthy CreateEntry completes in well under a second, so this only fires on -// a genuinely stuck filer, never on a normal flush. -const metadataFlushTimeout = 30 * time.Second +// metadataFlushTimeout bounds the metadata commit of a close()/fsync flush +// (CreateEntry and its retries) so an overwhelmed filer cannot wedge the +// calling process forever. It is deliberately generous: a healthy CreateEntry +// completes in well under a second, so this only fires on a genuinely stuck +// filer, never on a normal flush. It is the flush's commitBudget, so it does +// not cover the uploads that come first. A variable so tests can shorten it. +var metadataFlushTimeout = 30 * time.Second + +// commitBudget is the time one flush may spend in the filer calls of its +// metadata commit, added up over its attempts. Nothing else is charged to it: +// the data upload, the wait for the handle's locks, the merge of bloated +// chunks and the manifest upload can each outlast the whole budget. A deadline +// that also ran through that work was spent by it; CreateEntry then failed at +// once with "context deadline exceeded", and close() returned an error for +// data that was already stored. A nil budget sets no bound. The attempts run +// one after another, so it needs no lock. +type commitBudget struct { + left time.Duration +} + +// begin returns ctx limited to what is left of the budget, for one filer call. +// The returned func ends the call and charges the budget the time it took. +func (b *commitBudget) begin(ctx context.Context) (context.Context, func()) { + if b == nil { + return ctx, func() {} + } + start := time.Now() + ctx, cancel := context.WithTimeout(ctx, b.left) + return ctx, func() { + cancel() + b.left -= time.Since(start) + } +} + +// spent reports whether the budget has run out, so that another attempt could +// only fail. +func (b *commitBudget) spent() bool { + return b != nil && b.left <= 0 +} /** * Flush method @@ -79,17 +113,15 @@ func (wfs *WFS) Flush(cancel <-chan struct{}, in *fuse.FlushIn) fuse.Status { hasPosixLocks := wfs.hasPosixOwner(in.NodeId, in.LockOwner) allowAsync := !hasPosixLocks - // Bound the flush with a deadline instead of tying it to the FUSE cancel - // channel. A FUSE interrupt is not a process kill: Go's async preemption - // (SIGURG) makes a close() under load emit an interrupt on nearly every - // flush (see go-fuse RawFileSystem docs), so cancelling the in-flight + // Bound the metadata commit with a deadline instead of tying it to the FUSE + // cancel channel. A FUSE interrupt is not a process kill: Go's async + // preemption (SIGURG) makes a close() under load emit an interrupt on nearly + // every flush (see go-fuse RawFileSystem docs), so cancelling the in-flight // metadata CreateEntry on that interrupt turned healthy concurrent close()s // into EIO. The deadline still keeps close() from hanging forever against an - // overwhelmed filer without failing benign flushes. - ctx, cancelFunc := context.WithTimeout(context.Background(), metadataFlushTimeout) - defer cancelFunc() - - status := wfs.doFlush(ctx, fh, in.Uid, in.Gid, allowAsync) + // overwhelmed filer without failing benign flushes. doFlush charges it only + // for the commit's filer calls, so a slow upload cannot use it up. + status := wfs.doFlush(context.Background(), fh, in.Uid, in.Gid, allowAsync, metadataFlushTimeout) if in.LockOwner != 0 && wfs.releasePosixOwner(in.NodeId, in.LockOwner) { wfs.forgetLockOwner(in.NodeId, in.LockOwner) } @@ -122,15 +154,18 @@ func (wfs *WFS) Fsync(cancel <-chan struct{}, in *fuse.FsyncIn) (code fuse.Statu return fuse.ENOENT } - ctx, cancelFunc := context.WithTimeout(context.Background(), metadataFlushTimeout) - defer cancelFunc() - // Fsync is an explicit sync request — always flush synchronously - return wfs.doFlush(ctx, fh, in.Uid, in.Gid, false) + return wfs.doFlush(context.Background(), fh, in.Uid, in.Gid, false, metadataFlushTimeout) } -func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, allowAsync bool) fuse.Status { +// doFlush uploads the handle's dirty data, then commits its metadata to the +// filer. metadataTimeout is the commitBudget of that commit: what its filer +// calls may take over all attempts, whatever the uploads before and between +// them took. 0 sets no overall bound, leaving only the per-call filer +// timeouts. ctx should carry no deadline of its own: it would run through the +// uploads as well. +func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, allowAsync bool, metadataTimeout time.Duration) fuse.Status { // flush works at fh level fileFullPath := fh.FullPath() @@ -181,9 +216,17 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al return fuse.Status(syscall.ENOSPC) } + // The attempts share one budget, charged only for their filer calls. + var commit *commitBudget + if metadataTimeout > 0 { + commit = &commitBudget{left: metadataTimeout} + } + if err := retryMetadataFlushIf(ctx, func() error { - return wfs.flushMetadataToFiler(ctx, fh, dir, name, uid, gid) - }, isRetryableFilerError, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { + return wfs.flushMetadataToFiler(ctx, fh, dir, name, uid, gid, commit) + }, func(err error) bool { + return !commit.spent() && isRetryableFilerError(err) + }, func(nextAttempt, totalAttempts int, backoff time.Duration, err error) { glog.Warningf("%v fh %d flush: retrying metadata flush (attempt %d/%d) after %v: %v", fileFullPath, fh.fh, nextAttempt, totalAttempts, backoff, err) }); err != nil { @@ -204,7 +247,10 @@ func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, al // When -dlm is enabled, the distributed lock is already held by the FileHandle // from open-for-write through close, so no additional distributed lock is // needed here. The local fhLockTable lock below serializes within this mount. -func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, name string, uid, gid uint32) error { +// +// commit bounds the filer calls made here, over all attempts of one flush; nil +// leaves them to the per-call filer timeouts. +func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, name string, uid, gid uint32, commit *commitBudget) error { fileFullPath, dir, name := wfs.flushTarget(fh, dir, name) var unlock func() var fhActiveLock *util.ActiveLock @@ -240,7 +286,10 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n return nil } if fh.linkRefresh != nil { - if err := wfs.adoptStoredLink(ctx, fh, util.NewFullPath(dir, name), fh.linkRefresh); err != nil { + linkCtx, end := commit.begin(ctx) + err := wfs.adoptStoredLink(linkCtx, fh, util.NewFullPath(dir, name), fh.linkRefresh) + end() + if err != nil { return err } } @@ -303,7 +352,11 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry) wfs.mapPbIdFromLocalToFiler(request.Entry) - resp, merged, err := wfs.createContentEntry(ctx, fh, request, baseSnapshot, false) + // The chunk merge and the manifest above read and upload chunks, which can + // take longer than the whole commit budget. It is charged only from here. + commitCtx, end := commit.begin(ctx) + resp, merged, err := wfs.createContentEntry(commitCtx, fh, request, baseSnapshot, false) + end() if err != nil { glog.Errorf("fh flush create %s: %v", fileFullPath, err) return &contentPublicationError{cause: fmt.Errorf("fh flush create %s: %w", fileFullPath, err), request: proto.Clone(request).(*filer_pb.CreateEntryRequest)} diff --git a/weed/mount/weedfs_flush_deadline_test.go b/weed/mount/weedfs_flush_deadline_test.go new file mode 100644 index 0000000000..8272118fc4 --- /dev/null +++ b/weed/mount/weedfs_flush_deadline_test.go @@ -0,0 +1,276 @@ +package mount + +import ( + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/go-fuse/v2/fuse" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/wdclient" +) + +// slowDataFiler answers AssignVolume only after assignDelay, as a filer does +// while its master is slow or no volume slot is free, and points every +// assignment at a volume stub. CreateEntry, the metadata commit, answers at +// once; with stuckCommit it holds the request until the caller gives up, like +// a filer wedged on its store. +type slowDataFiler struct { + fakeFilerServer + volume string + assignDelay time.Duration + stuckCommit bool + assigns atomic.Int32 + commits atomic.Int32 + committedChunks atomic.Int32 + committedManifests atomic.Int32 +} + +func (s *slowDataFiler) AssignVolume(ctx context.Context, req *filer_pb.AssignVolumeRequest) (*filer_pb.AssignVolumeResponse, error) { + n := s.assigns.Add(1) + select { + case <-time.After(s.assignDelay): + case <-ctx.Done(): + return nil, ctx.Err() + } + return &filer_pb.AssignVolumeResponse{ + FileId: fmt.Sprintf("3,%02x637037d6", n), + Count: 1, + Location: &filer_pb.Location{Url: s.volume, PublicUrl: s.volume}, + }, nil +} + +func (s *slowDataFiler) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + s.commits.Add(1) + chunks := req.GetEntry().GetChunks() + manifests, _ := filer.SeparateManifestChunks(chunks) + s.committedChunks.Store(int32(len(chunks))) + s.committedManifests.Store(int32(len(manifests))) + if s.stuckCommit { + <-ctx.Done() + return nil, ctx.Err() + } + return s.fakeFilerServer.CreateEntry(ctx, req) +} + +// newVolumeStub stores chunk uploads the way a volume server answers them. +func newVolumeStub(t *testing.T) string { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reader, err := r.MultipartReader() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + part, err := reader.NextPart() + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + data, err := io.ReadAll(part) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + w.Header().Set("Content-MD5", part.Header.Get("Content-MD5")) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + fmt.Fprintf(w, `{"size":%d}`, len(data)) + })) + t.Cleanup(server.Close) + return strings.TrimPrefix(server.URL, "http://") +} + +// newSlowDataFlush opens /dir/file with dirty data not yet uploaded, written +// after the chunks the file already holds (stored), on a mount whose filer +// takes assignDelay to assign a chunk, and shortens the commit deadline to +// deadline. +func newSlowDataFlush(t *testing.T, assignDelay, deadline time.Duration, stuckCommit bool, stored ...*filer_pb.FileChunk) (*WFS, *FileHandle, *slowDataFiler) { + t.Helper() + + saved := metadataFlushTimeout + metadataFlushTimeout = deadline + t.Cleanup(func() { metadataFlushTimeout = saved }) + + wfs := newInvalidateTestWFS(t) + wfs.option.ConcurrentWriters = 4 + wfs.option.uniqueCacheDirForWrite = t.TempDir() + wfs.concurrentWriters = util.NewLimitedConcurrentExecutor(4) + + fake := &slowDataFiler{volume: newVolumeStub(t), assignDelay: assignDelay, stuckCommit: stuckCommit} + startFakeFiler(t, wfs, fake) + // Upload straight to the assigned volume server, as mounts do by default. + wfs.option.VolumeServerAccess = "direct" + wfs.filerClient = wdclient.NewFilerClient(wfs.option.FilerAddresses, wfs.option.GrpcDialOption, "") + + data := []byte("the last shard of a checkpoint, written just before close()") + offset := int64(filer.TotalSize(stored)) + inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) + fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, &filer_pb.Entry{ + Name: "file", + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(offset) + uint64(len(data)), FileMode: 0100644}, + Chunks: stored, + }, 0, 0) + t.Cleanup(func() { wfs.ReleaseHandle(fh.fh) }) + if err := fh.dirtyPages.AddPage(offset, data, true, time.Now().UnixNano()); err != nil { + t.Fatalf("AddPage: %v", err) + } + fh.dirtyMetadata = true + return wfs, fh, fake +} + +// storedSmallChunks returns n chunks a file already holds, back to back from +// offset 0, each small enough to count toward a manifest of small chunks. +func storedSmallChunks(n int) []*filer_pb.FileChunk { + const size = 1024 + chunks := make([]*filer_pb.FileChunk, n) + for i := range chunks { + chunks[i] = &filer_pb.FileChunk{ + FileId: fmt.Sprintf("4,%02x637037d6", i+1), + Offset: int64(i) * size, + Size: size, + ModifiedTsNs: int64(i + 1), + } + } + return chunks +} + +// checkCommitted asserts the flush made the given number of uploads, then +// committed once, an entry of one chunk. +func checkCommitted(t *testing.T, fh *FileHandle, fake *slowDataFiler, uploads int32) { + t.Helper() + if got := fake.assigns.Load(); got != uploads { + t.Fatalf("AssignVolume calls = %d, want %d", got, uploads) + } + if got := fake.commits.Load(); got != 1 { + t.Fatalf("CreateEntry calls = %d, want 1", got) + } + if got := fake.committedChunks.Load(); got != 1 { + t.Fatalf("committed chunks = %d, want 1", got) + } + if fh.dirtyMetadata { + t.Fatal("handle still dirty after the commit") + } +} + +// close() after an upload that outlasts the commit deadline still commits: +// the deadline bounds CreateEntry, which the filer answers at once. Started at +// the top of Flush, the deadline was spent by the slow assign, and close() +// returned EIO for data that was already stored. +func TestFlushCommitsAfterSlowUpload(t *testing.T) { + wfs, fh, fake := newSlowDataFlush(t, time.Second, 200*time.Millisecond, false) + + status := wfs.Flush(nil, &fuse.FlushIn{InHeader: fuse.InHeader{NodeId: fh.inode}, Fh: uint64(fh.fh)}) + if status != fuse.OK { + t.Fatalf("Flush = %v, want OK: the data is stored and the filer answers the commit at once", status) + } + checkCommitted(t, fh, fake, 1) +} + +// fsync() gets the same commit-only deadline as close(). +func TestFsyncCommitsAfterSlowUpload(t *testing.T) { + wfs, fh, fake := newSlowDataFlush(t, time.Second, 200*time.Millisecond, false) + + status := wfs.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: fh.inode}, Fh: uint64(fh.fh)}) + if status != fuse.OK { + t.Fatalf("Fsync = %v, want OK", status) + } + checkCommitted(t, fh, fake, 1) +} + +// A truncate below dirty data flushes the data first (see SetAttr); that +// flush gets the commit-only deadline too. +func TestTruncateFlushCommitsAfterSlowUpload(t *testing.T) { + wfs, fh, fake := newSlowDataFlush(t, time.Second, 200*time.Millisecond, false) + + in := &fuse.SetAttrIn{} + in.NodeId = fh.inode + in.Valid = fuse.FATTR_SIZE + in.Size = 8 + var out fuse.AttrOut + if status := wfs.SetAttr(nil, in, &out); status != fuse.OK { + t.Fatalf("SetAttr size down = %v, want OK", status) + } + if got := fake.commits.Load(); got != 1 { + t.Fatalf("CreateEntry calls = %d, want 1", got) + } + if out.Size != 8 { + t.Fatalf("size after truncate = %d, want 8", out.Size) + } +} + +// The data upload is not the only one a flush makes. One that completes a +// batch of small chunks folds them into a manifest and uploads that too, +// while it prepares the entry it is about to commit. That upload must not +// spend the commit deadline either. +func TestFlushCommitsAfterSlowManifestUpload(t *testing.T) { + // One chunk short of a batch: the chunk this flush uploads completes it. + stored := storedSmallChunks(filer.SmallChunkManifestBatch - 1) + wfs, fh, fake := newSlowDataFlush(t, time.Second, 200*time.Millisecond, false, stored...) + + status := wfs.Flush(nil, &fuse.FlushIn{InHeader: fuse.InHeader{NodeId: fh.inode}, Fh: uint64(fh.fh)}) + if status != fuse.OK { + t.Fatalf("Flush = %v, want OK: the data and the manifest are stored and the filer answers the commit at once", status) + } + checkCommitted(t, fh, fake, 2) // the data chunk, then the manifest + if got := fake.committedManifests.Load(); got != 1 { + t.Fatalf("committed manifests = %d, want 1: this flush no longer uploads a manifest", got) + } +} + +// The deadline still bounds the commit itself: against a filer that never +// answers CreateEntry, close() fails once the deadline passes instead of +// hanging, and the handle stays dirty for Release to try again. +func TestFlushCommitDeadlineStillBoundsAStuckFiler(t *testing.T) { + const deadline = 200 * time.Millisecond + wfs, fh, fake := newSlowDataFlush(t, 0, deadline, true) + + start := time.Now() + status := wfs.Flush(nil, &fuse.FlushIn{InHeader: fuse.InHeader{NodeId: fh.inode}, Fh: uint64(fh.fh)}) + elapsed := time.Since(start) + if status == fuse.OK { + t.Fatal("Flush = OK against a filer that never committed") + } + if elapsed < deadline || elapsed > deadline+5*time.Second { + t.Fatalf("Flush returned after %v, want about the %v commit deadline", elapsed, deadline) + } + if got := fake.commits.Load(); got != 1 { + t.Fatalf("CreateEntry calls = %d, want 1: no retry once the deadline passed", got) + } + if !fh.dirtyMetadata { + t.Fatal("handle marked clean after a failed commit") + } +} + +// The attempts of one flush share the budget: what one filer call used is gone +// for the next. A nil budget bounds nothing. +func TestCommitBudgetIsSharedByItsCalls(t *testing.T) { + budget := &commitBudget{left: 50 * time.Millisecond} + ctx, end := budget.begin(context.Background()) + <-ctx.Done() // a filer that never answers + end() + if !budget.spent() { + t.Fatalf("%v left after a call used the whole budget", budget.left) + } + ctx, end = budget.begin(context.Background()) + if ctx.Err() == nil { + t.Fatal("a call started with time left after the budget ran out") + } + end() + + var none *commitBudget + ctx, end = none.begin(context.Background()) + if _, bounded := ctx.Deadline(); bounded || none.spent() { + t.Fatal("a nil budget bounded the call") + } + end() +} diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index 5186c731e5..677c1720e4 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -1326,7 +1326,7 @@ func TestFlushAckCancelsPendingCopyEventAdoption(t *testing.T) { // A local flush lands: the ack at 3000 becomes the authoritative base. fh.dirtyMetadata = true - if status := wfs.doFlush(context.Background(), fh, 0, 0, false); status != fuse.OK { + if status := wfs.doFlush(context.Background(), fh, 0, 0, false, 0); status != fuse.OK { t.Fatalf("doFlush status = %v, want OK", status) } @@ -2283,7 +2283,7 @@ func TestPeerReplacementKeepsDirtyHandle(t *testing.T) { t.Fatalf("base %v moved to the replacement", base) } - if code := wfs.doFlush(context.Background(), fh, 1000, 20, false); code == fuse.OK { + if code := wfs.doFlush(context.Background(), fh, 1000, 20, false, 0); code == fuse.OK { t.Fatal("flush into the replacement acknowledged") } fake.mu.Lock() diff --git a/weed/mount/weedfs_link_open_handle_test.go b/weed/mount/weedfs_link_open_handle_test.go index a9018299a4..b887bb154b 100644 --- a/weed/mount/weedfs_link_open_handle_test.go +++ b/weed/mount/weedfs_link_open_handle_test.go @@ -193,7 +193,7 @@ func appendAndFlush(t *testing.T, wfs *WFS, fh *FileHandle, key uint64) fuse.Sta e.Attributes.FileSize = uint64(len(e.Chunks)) * 100 }) fh.dirtyMetadata = true - return wfs.doFlush(context.Background(), fh, 1000, 20, false) + return wfs.doFlush(context.Background(), fh, 1000, 20, false, 0) } // queueClose closes fh the way a writeback mount does. Its flush starts once @@ -334,7 +334,7 @@ func TestFlushResolvedBeforeAliasUnlinkReachesSurvivor(t *testing.T) { e.Attributes.FileSize = 200 }) fh.dirtyMetadata = true - if err := wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20); err != nil { + if err := wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20, nil); err != nil { t.Fatalf("flush resolved before the unlink: %v", err) } wantStored(t, fake, "final", 2, 1) @@ -356,7 +356,7 @@ func TestFlushWaitingThroughAliasUnlinkReachesSurvivor(t *testing.T) { table := wfs.metadataPublicationTable() held := table.AcquireLock("test", "/tmp", util.SharedLock) done := make(chan error, 1) - go func() { done <- wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20) }() + go func() { done <- wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20, nil) }() waitForPublicationWaiter(t, table, "/tmp") unlinkInRoot(t, wfs, "tmp") table.ReleaseLock("/tmp", held) @@ -385,7 +385,7 @@ func TestFlushWhoseNameKeepsChangingRetries(t *testing.T) { current := util.FullPath("/tmp") held := table.AcquireLock("test", current, util.SharedLock) done := make(chan error, 1) - go func() { done <- wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20) }() + go func() { done <- wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20, nil) }() var removed []string for i := 0; i < 4; i++ { waitForPublicationWaiter(t, table, current) @@ -405,7 +405,7 @@ func TestFlushWhoseNameKeepsChangingRetries(t *testing.T) { if err == nil || !isRetryableFilerError(err) { t.Fatalf("flush whose name changed during every lock wait: %v, want a retryable error", err) } - if err := wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20); err != nil { + if err := wfs.flushMetadataToFiler(context.Background(), fh, "/", "tmp", 1000, 20, nil); err != nil { t.Fatalf("retried flush: %v", err) } wantStored(t, fake, "final", 2, 1) diff --git a/weed/mount/weedfs_rename.go b/weed/mount/weedfs_rename.go index feb5049cee..1477b2ed8b 100644 --- a/weed/mount/weedfs_rename.go +++ b/weed/mount/weedfs_rename.go @@ -388,8 +388,9 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string // BEFORE any async flush interference. if fh, ok := wfs.fhMap.FindFileHandle(sourceInode); ok && fh.dirtyMetadata { glog.V(4).Infof("dir Rename %s: flushing deferred metadata before rename", oldPath) - // Prerequisite for the rename, so it must complete: non-cancellable context. - if flushStatus := wfs.doFlush(context.Background(), fh, oldEntry.Attributes.Uid, oldEntry.Attributes.Gid, false); flushStatus != fuse.OK { + // Prerequisite for the rename, so it must complete: non-cancellable + // context and no commit deadline. + if flushStatus := wfs.doFlush(context.Background(), fh, oldEntry.Attributes.Uid, oldEntry.Attributes.Gid, false, 0); flushStatus != fuse.OK { glog.Warningf("dir Rename %s: flush before rename failed: %v", oldPath, flushStatus) return flushStatus }