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 }