package mount import ( "bytes" "context" "errors" "fmt" "io" "syscall" "time" "github.com/seaweedfs/go-fuse/v2/fuse" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" "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 /** * Flush method * * This is called on each close() of the opened file. * * Since file descriptors can be duplicated (dup, dup2, fork), for * one open call there may be many flush calls. * * Filesystems shouldn't assume that flush will always be called * after some writes, or that if will be called at all. * * fi->fh will contain the value set by the open method, or will * be undefined if the open method didn't set any value. * * NOTE: the name of the method is misleading, since (unlike * fsync) the filesystem is not forced to flush pending writes. * One reason to flush data is if the filesystem wants to return * write errors during close. However, such use is non-portable * because POSIX does not require [close] to wait for delayed I/O to * complete. * * If the filesystem supports file locking operations (setlk, * getlk) it should remove all locks belonging to 'fi->owner'. * * If this request is answered with an error code of ENOSYS, * this is treated as success and future calls to flush() will * succeed automatically without being send to the filesystem * process. * * Valid replies: * fuse_reply_err * * @param req request handle * @param ino the inode number * @param fi file information * * [close]: http://pubs.opengroup.org/onlinepubs/9699919799/functions/close.html */ func (wfs *WFS) Flush(cancel <-chan struct{}, in *fuse.FlushIn) fuse.Status { fh := wfs.GetHandle(FileHandleId(in.Fh)) if fh == nil { // If handle is not found, it might have been already released // This is not an error condition for FLUSH if in.LockOwner != 0 && wfs.releasePosixOwner(in.NodeId, in.LockOwner) { wfs.forgetLockOwner(in.NodeId, in.LockOwner) } return fuse.OK } // FlushIn.LockOwner is populated by some FUSE kernels even when the process // did not hold byte-range locks. Only force the synchronous close path when // this owner actually has POSIX locks to release; otherwise writebackCache // would silently degrade to a blocking flush for ordinary close(). 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 // 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) if in.LockOwner != 0 && wfs.releasePosixOwner(in.NodeId, in.LockOwner) { wfs.forgetLockOwner(in.NodeId, in.LockOwner) } return status } /** * Synchronize file contents * * If the datasync parameter is non-zero, then only the user data * should be flushed, not the meta data. * * If this request is answered with an error code of ENOSYS, * this is treated as success and future calls to fsync() will * succeed automatically without being send to the filesystem * process. * * Valid replies: * fuse_reply_err * * @param req request handle * @param ino the inode number * @param datasync flag indicating if only data should be flushed * @param fi file information */ func (wfs *WFS) Fsync(cancel <-chan struct{}, in *fuse.FsyncIn) (code fuse.Status) { fh := wfs.GetHandle(FileHandleId(in.Fh)) if fh == nil { 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) } func (wfs *WFS) doFlush(ctx context.Context, fh *FileHandle, uid, gid uint32, allowAsync bool) fuse.Status { // flush works at fh level fileFullPath := fh.FullPath() fh.RememberPath(fileFullPath) dir, name := fileFullPath.DirAndName() // send the data to the OS glog.V(4).Infof("doFlush %s fh %d", fileFullPath, fh.fh) // When writebackCache is enabled and this is a close()-triggered Flush (not fsync), // defer the expensive data upload + metadata flush to a background goroutine. // This allows the calling process (e.g., rsync) to proceed to the next file immediately. // POSIX does not require close() to wait for delayed I/O to complete. if allowAsync && wfs.option.WritebackCache && fh.dirtyMetadata { if wfs.IsOverQuotaWithUncommitted() { return fuse.Status(syscall.ENOSPC) } fh.asyncFlushPending = true fh.asyncFlushUid = uid fh.asyncFlushGid = gid glog.V(3).Infof("doFlush async deferred %s fh %d", fileFullPath, fh.fh) return fuse.OK } // Synchronous flush path (normal mode, fsync, or no dirty data) fh.asyncFlushPending = false // Check quota including uncommitted writes for real-time enforcement isOverQuota := wfs.IsOverQuotaWithUncommitted() if !isOverQuota { if err := fh.dirtyPages.FlushData(); err != nil { glog.Errorf("%v doFlush: %v", fileFullPath, err) return writeErrorToFuseStatus(err) } } if !fh.dirtyMetadata { return fuse.OK } // Skip metadata flush if the file was unlinked while open. // The filer entry is already gone; flushing would recreate it. if fh.isDeleted { glog.V(3).Infof("doFlush %s fh %d: file was unlinked, skipping metadata flush", fileFullPath, fh.fh) return fuse.OK } if isOverQuota { return fuse.Status(syscall.ENOSPC) } 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) { glog.Warningf("%v fh %d flush: retrying metadata flush (attempt %d/%d) after %v: %v", fileFullPath, fh.fh, nextAttempt, totalAttempts, backoff, err) }); err != nil { glog.Errorf("%v fh %d flush: %v", fileFullPath, fh.fh, err) return grpcErrorToFuseStatus(err) } if IsDebugFileReadWrite { fh.mirrorFile.Sync() } return fuse.OK } // flushMetadataToFiler sends the file's chunk references and attributes to the filer. // This is shared between the synchronous doFlush path and the async flush completion. // // 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 { fileFullPath := fh.FullPath() // Attribute acknowledgement and local retention must complete before a // whole-entry flush can publish. Take this before the handle's active lock. unlock := wfs.lockMetadataPublication(fileFullPath) defer unlock() glog.V(4).Infof("flushMetadataToFiler %s/%s inode %d fh %d", dir, name, fh.inode, fh.fh) fhActiveLock := fh.wfs.fhLockTable.AcquireLock("doFlush", fh.fh, util.ExclusiveLock) defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) // Re-check under the lock: Unlink sets the flag under it, so a flush that // was already past the earlier check cannot write the entry back after // the delete removed it. if fh.isDeleted { glog.V(3).Infof("flushMetadataToFiler %s fh %d: file was unlinked, skipping", fileFullPath, fh.fh) return nil } entry := fh.GetEntry() entry.Name = name // this flush may be just after a rename operation if entry.Attributes != nil { entry.Attributes.Mime = fh.contentType if entry.Attributes.Uid == 0 { entry.Attributes.Uid = uid } if entry.Attributes.Gid == 0 { entry.Attributes.Gid = gid } // Do not stamp mtime/ctime here. Write/SetAttr already maintain // them on the entry; overwriting at flush time clobbered user-set // mtime (utimes/touch -m -d) once the deferred flush ran. } glog.V(4).Infof("%s set chunks: %v", fileFullPath, len(entry.GetChunks())) manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entry.GetChunks()) chunks, _, err := filer.CompactFileChunks(context.Background(), wfs.LookupFn(), nonManifestChunks) if err != nil { return fmt.Errorf("compact chunks of %s: %w", fileFullPath, err) } if mergedChunks, mergeErr := wfs.maybeMergeChunks(fileFullPath, chunks, manifestChunks); mergeErr != nil { glog.V(0).Infof("maybeMergeChunks %s: %v", fileFullPath, mergeErr) } else if mergedChunks != nil { chunks = mergedChunks manifestChunks = nil } // no chunk deleter here: a failed fold reports the blobs it saved chunks, manifestErr := filer.MaybeManifestize(wfs.saveDataAsChunk(fileFullPath), nil, chunks) if manifestErr != nil { // not good, but should be ok glog.V(0).Infof("MaybeManifestize: %v", manifestErr) } entry.Chunks = append(chunks, manifestChunks...) // Clone the proto entry for the filer request so that mapPbIdFromLocalToFiler // does not mutate the file handle's live entry. Without the clone, a concurrent // Lookup can observe filer-side uid/gid on the file handle entry and return it // to the kernel, which caches it and then rejects opens by the local user. requestEntry := proto.Clone(entry.GetEntry()).(*filer_pb.Entry) request := &filer_pb.CreateEntryRequest{ Directory: string(dir), Entry: requestEntry, Signatures: []int32{wfs.signature}, SkipCheckParentDirectory: true, } // Snapshot with local ids before the request mapping mutates the clone: // on ack this becomes the handle's base, judged against future events. baseSnapshot := proto.Clone(requestEntry).(*filer_pb.Entry) wfs.mapPbIdFromLocalToFiler(request.Entry) resp, merged, err := wfs.createContentEntry(ctx, fh, request, baseSnapshot) if err != nil { glog.Errorf("fh flush create %s: %v", fileFullPath, err) return fmt.Errorf("fh flush create %s: %w", fileFullPath, err) } if merged != nil { fh.rebaseEntry(merged, requestEntry) baseSnapshot = merged } event := resp.GetMetadataEvent() if event == nil { event = metadataUpdateEvent(string(dir), request.Entry) if event != nil { event.TsNs = ackVersionTsNs(resp) } } // The filer acknowledged this state at the event's log position (or, for // a no-op create, at the response's log position); older queued // subscription events must not roll the handle back. fh.setAuthoritativeBase(wfs.ackedBase(resp, baseSnapshot)) fh.advanceEntryVersion(ackVersionTsNs(resp), resp.GetLogSignature()) if applyErr := wfs.applyLocalMetadataEvent(context.Background(), event); applyErr != nil { glog.Warningf("flush %s: best-effort metadata apply failed: %v", fileFullPath, applyErr) wfs.inodeToPath.InvalidateChildrenCache(util.FullPath(dir)) } if err == nil { fh.dirtyMetadata = false } return err } // createContentEntry publishes a handle's whole entry only over the filer // state the handle observed. Deferred creates expect no entry; existing // handles compare the full base, including inode identity. The unary call // keeps a refused condition apart from an I/O failure; the mutation stream // reduces it to EIO. // // When the stored entry moved on, the handle's own changes since its base are // replayed onto it and published against it, so a peer's acknowledged chunks // survive. merged is that entry in local ids; nil when the request went as is. func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request *filer_pb.CreateEntryRequest, local *filer_pb.Entry) (resp *filer_pb.CreateEntryResponse, merged *filer_pb.Entry, err error) { base := fh.baseEntry.Load() guard := &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS} if base != nil { expected := proto.Clone(base).(*filer_pb.Entry) wfs.mapPbIdFromLocalToFiler(expected) guard = &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: expected} } sent := request.Entry for attempt := 0; ; attempt++ { request.Condition = &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{guard}} var stored *filer_pb.Entry err = wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { var callErr error stored = nil resp, callErr = filer_pb.CreateEntryWithResponse(ctx, client, request) if base == nil || attempt >= maxContentFlushMerges || !errors.Is(callErr, filer_pb.ErrPreconditionFailed) { return callErr } if found, lookupErr := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: request.Directory, Name: request.Entry.Name}); lookupErr == nil { stored = found.Entry } return callErr }) if err == nil && request.Entry != sent { merged = proto.Clone(request.Entry).(*filer_pb.Entry) wfs.mapPbIdFromFilerToLocal(merged) } if err == nil || stored == nil { return resp, merged, err } next, ok := mergeContentEntry(stored, base, local, sent, fh.truncatedSince(base)) if !ok { return resp, nil, err } glog.V(1).Infof("content flush %s/%s: entry changed since the handle's base, merging", request.Directory, request.Entry.Name) request.Entry = next guard = &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: stored} } } // maxContentFlushMerges bounds how often one flush rebases onto a moving entry. const maxContentFlushMerges = 3 // mergeContentEntry replays what local changed since base onto stored: the // chunks it added or dropped, and the attributes, extended keys and inline // content it changed, taking those values from sent (local in stored's id // space). Everything else keeps the stored value. A local truncation keeps // the handle's size and clips stored chunks at the lowest size truncated to // (truncatedTo, or -1); otherwise the stored size stands unless the handle // changed its size or wrote past it. // A different file at the path, inline content changed on both sides, // inline content next to chunks (reads stop at the inline bytes), or a local // whole-file rewrite (maybeMergeChunks, unstamped chunks) over a peer's chunk // change cannot be merged: the rewrite carries the handle's stale view. func mergeContentEntry(stored, base, local, sent *filer_pb.Entry, truncatedTo int64) (*filer_pb.Entry, bool) { if stored == nil || base == nil || stored.IsDirectory || !bytes.Equal(stored.HardLinkId, base.HardLinkId) { return nil, false } if inode := base.GetAttributes().GetInode(); inode != 0 && stored.GetAttributes().GetInode() != inode { return nil, false } merged := proto.Clone(stored).(*filer_pb.Entry) merged.Name = sent.Name if !bytes.Equal(local.Content, base.Content) { if !bytes.Equal(stored.Content, base.Content) { return nil, false } merged.Content = sent.Content } inBase := make(map[string]bool, len(base.Chunks)) for _, c := range base.Chunks { inBase[contentChunkKey(c)] = true } inLocal := make(map[string]bool, len(local.Chunks)) for _, c := range local.Chunks { inLocal[contentChunkKey(c)] = true } localSize := filer.FileSize(local) if truncatedTo < 0 && localSize < filer.FileSize(base) { truncatedTo = int64(localSize) } peerChanged := len(stored.Chunks) != len(base.Chunks) merged.Chunks = merged.Chunks[:0] for _, c := range stored.Chunks { key := contentChunkKey(c) peerChanged = peerChanged || !inBase[key] if inBase[key] && !inLocal[key] || truncatedTo >= 0 && c.Offset >= truncatedTo { continue } c = proto.Clone(c).(*filer_pb.FileChunk) if truncatedTo >= 0 && c.Offset+int64(c.Size) > truncatedTo { c.Size = uint64(truncatedTo - c.Offset) } merged.Chunks = append(merged.Chunks, c) } size := int64(filer.FileSize(stored)) if localSize != filer.FileSize(base) { size = max(size, int64(localSize)) } for _, c := range local.Chunks { if !inBase[contentChunkKey(c)] { if peerChanged && c.ModifiedTsNs == 0 && !c.IsChunkManifest { return nil, false } merged.Chunks = append(merged.Chunks, proto.Clone(c).(*filer_pb.FileChunk)) size = max(size, c.Offset+int64(c.Size)) } } if len(merged.Content) > 0 && len(merged.Chunks) > 0 { return nil, false } if merged.Attributes == nil { merged.Attributes = &filer_pb.FuseAttributes{} } baseAttrs, localAttrs := base.GetAttributes(), local.GetAttributes() if baseAttrs == nil { baseAttrs = &filer_pb.FuseAttributes{} } if localAttrs == nil { localAttrs = &filer_pb.FuseAttributes{} } m, b, l := merged.Attributes.ProtoReflect(), baseAttrs.ProtoReflect(), localAttrs.ProtoReflect() s := proto.Clone(sent.GetAttributes()).(*filer_pb.FuseAttributes).ProtoReflect() fields := m.Descriptor().Fields() for i := 0; i < fields.Len(); i++ { fd := fields.Get(i) if fd.Name() == "file_size" || l.Get(fd).Equal(b.Get(fd)) { continue } if s.Has(fd) { m.Set(fd, s.Get(fd)) } else { m.Clear(fd) } } for k, v := range local.Extended { if old, found := base.Extended[k]; !found || !bytes.Equal(old, v) { if merged.Extended == nil { merged.Extended = make(map[string][]byte) } merged.Extended[k] = sent.Extended[k] } } for k := range base.Extended { if _, found := local.Extended[k]; !found { delete(merged.Extended, k) } } if truncatedTo >= 0 { size = int64(localSize) } merged.Attributes.FileSize = uint64(size) return merged, true } // contentChunkKey identifies a chunk reference: truncation shrinks a chunk in // place, so the same file id at another size is a different reference. Both // file id encodings give one key, and the chunk is left as is. func contentChunkKey(c *filer_pb.FileChunk) string { fid := c.Fid if fid == nil { if parsed, err := filer_pb.ToFileIdObject(c.FileId); err == nil { fid = parsed } else { return fmt.Sprintf("%s@%d+%d", c.FileId, c.Offset, c.Size) } } return fmt.Sprintf("%d,%x,%x@%d+%d", fid.VolumeId, fid.FileKey, fid.Cookie, c.Offset, c.Size) } // ackedBase is the base an acknowledged create leaves: the entry the filer // stored, in local ids. The filer stamps commit_ts_ns, crtime, inode and atime // and may rewrite chunks, so the requested entry would never match it again. // A filer that sends no event leaves the requested entry. func (wfs *WFS) ackedBase(resp *filer_pb.CreateEntryResponse, requested *filer_pb.Entry) *filer_pb.Entry { stored := resp.GetMetadataEvent().GetEventNotification().GetNewEntry() if stored == nil { return requested } base := proto.Clone(stored).(*filer_pb.Entry) wfs.mapPbIdFromFilerToLocal(base) return base } // shouldMergeChunks reports whether the non-manifest chunks are bloated // enough to justify re-reading and re-uploading the file. The condition // is: sum of compacted chunk sizes > 2 * logical file size. func shouldMergeChunks(compactedChunks []*filer_pb.FileChunk, manifestChunks []*filer_pb.FileChunk) (totalChunkSize, fileSize uint64, merge bool) { for _, chunk := range compactedChunks { totalChunkSize += chunk.Size } // Count manifest coverage toward stored total. Each manifest holds // sub-chunks on volume servers that cover approximately Size bytes. // Without this, overlapping manifests accumulate undetected because // the merge condition only saw the (small) non-manifest chunk total. for _, chunk := range manifestChunks { totalChunkSize += chunk.Size } allChunks := make([]*filer_pb.FileChunk, 0, len(compactedChunks)+len(manifestChunks)) allChunks = append(allChunks, compactedChunks...) allChunks = append(allChunks, manifestChunks...) fileSize = filer.TotalSize(allChunks) merge = fileSize > 0 && totalChunkSize > 2*fileSize return } // maybeMergeChunks re-reads and re-uploads file data as properly sized chunks // when the total stored chunk data significantly exceeds the logical file size, // which happens after many random writes create partially-overlapping small chunks. func (wfs *WFS) maybeMergeChunks(fileFullPath util.FullPath, compactedChunks []*filer_pb.FileChunk, manifestChunks []*filer_pb.FileChunk) ([]*filer_pb.FileChunk, error) { totalChunkSize, fileSize, merge := shouldMergeChunks(compactedChunks, manifestChunks) if !merge { return nil, nil } glog.V(0).Infof("%.1fx chunk bloat detected on %s (%d chunks, %d stored vs %d content), merging", float64(totalChunkSize)/float64(fileSize), fileFullPath, len(compactedChunks), totalChunkSize, fileSize) ctx := context.Background() allChunks := make([]*filer_pb.FileChunk, 0, len(compactedChunks)+len(manifestChunks)) allChunks = append(allChunks, compactedChunks...) allChunks = append(allChunks, manifestChunks...) reader := filer.NewChunkStreamReaderFromLookup(ctx, wfs.LookupFn(), allChunks) defer reader.Close() saveFunc := wfs.saveDataAsChunk(fileFullPath) chunkSize := wfs.option.ChunkSizeLimit if int64(fileSize) < chunkSize { chunkSize = int64(fileSize) } var newChunks []*filer_pb.FileChunk var offset int64 buf := make([]byte, chunkSize) for { n, readErr := io.ReadFull(reader, buf) if n > 0 { chunk, uploadErr := saveFunc(bytes.NewReader(buf[:n]), "", offset, 0, uint64(n)) if uploadErr != nil { return nil, uploadErr } newChunks = append(newChunks, chunk) offset += int64(n) } if readErr != nil { if readErr == io.EOF || readErr == io.ErrUnexpectedEOF { break } return nil, readErr } } glog.V(0).Infof("merged %s: %d chunks -> %d chunks", fileFullPath, len(compactedChunks)+len(manifestChunks), len(newChunks)) return newChunks, nil }