Skip to content

Fix releasefile handle leak #1098

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
foodprocessor merged 7 commits into from
Oct 27, 2026
Merged

Fix releasefile handle leak #1098

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
16d7b4b
Release handle resources in all components and simplify logging (push…
foodprocessor Oct 23, 2026
d19acb7
Retry failed uploads on close
foodprocessor Oct 24, 2026
08bc161
Fixup
foodprocessor Oct 24, 2026
37a08b6
Let ENOSPC kill uploads.
foodprocessor Oct 24, 2026
7507936
Improve readability and error handling in pending op code.
foodprocessor Oct 27, 2026
c05d24e
Allow ENOSPC to be returned to the application!
foodprocessor Oct 27, 2026
14061fb
Merge remote-tracking branch 'origin/main' into fix/releasefile-handl…
foodprocessor Oct 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions component/block_cache/block_cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -621,21 +621,21 @@ func (bc *BlockCache) releaseFileInternal(options internal.ReleaseFileOptions) e

defer bc.fileCloseOpt.Done()

var err error
if options.Handle.Dirty() {
log.Info(
"BlockCache::CloseFile : name=%s, handle=%d dirty. Flushing the file.",
options.Handle.Path,
options.Handle.ID,
)
err := bc.FlushFile(
err = bc.FlushFile(
internal.FlushFileOptions{Handle: options.Handle, CloseInProgress: true},
) //nolint
)
if err != nil {
log.Err(
"BlockCache::ReleaseFileInternal : failed to flush file %s",
options.Handle.Path,
)
return err
}
}

Expand Down Expand Up @@ -669,7 +669,7 @@ func (bc *BlockCache) releaseFileInternal(options internal.ReleaseFileOptions) e
}
options.Handle.Buffers.Cooked = nil

return nil
return err
}

func (bc *BlockCache) getBlockSize(fileSize uint64, block *Block) uint64 {
Expand Down
98 changes: 55 additions & 43 deletions component/file_cache/async.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"fmt"
"os"
"path/filepath"
"syscall"
"time"

"github.com/Seagate/cloudfuse/common"
Expand Down Expand Up @@ -233,12 +234,6 @@ func (fc *FileCache) addPendingOp(name string, value pendingFlags) {
}
}

// syncsPendingOps : whether servicePendingOps is running. Queueing an op without it
// would pin the cached file forever, since eviction skips pending ops.
func (fc *FileCache) syncsPendingOps() bool {
return len(fc.schedule) > 0 || fc.offlineAccess
}

// persistent background thread function
func (fc *FileCache) servicePendingOps() {
// grows while upload cycles keep failing, resets as soon as one succeeds
Expand Down Expand Up @@ -296,35 +291,33 @@ func (fc *FileCache) servicePendingOps() {
// runPendingOpCycle : sync pending ops with cloud storage until the upload window
// closes or an operation fails. Returns the number of ops attempted, and the error
// that ended the cycle, if any.
func (fc *FileCache) runPendingOpCycle() (int, error) {
numFilesProcessed := 0
var cycleErr error

func (fc *FileCache) runPendingOpCycle() (numFilesProcessed int, err error) {
fc.pendingOps.Range(func(key, value any) bool {
select {
case <-fc.componentStopping:
return false
case <-fc.uploadWindow():
name := key.(string)
numFilesProcessed++
if !fc.updateObject(name, value.(pendingFlags)) {
// record the failure so the caller backs off instead of busy-retrying
cycleErr = fmt.Errorf("sync failed for %s", name)
return false
}
return true
err = fc.updateObject(name, value.(pendingFlags))
return err == nil
default:
return false // upload window ended
}
})

return numFilesProcessed, cycleErr
return numFilesProcessed, err
}

// synchronize pending operation with cloud storage
// returns false if we should stop iterating (offline, etc.)
func (fc *FileCache) updateObject(name string, flags pendingFlags) bool {
log.Trace("FileCache::updateObject : %s", name)
// returns the cloud error *if and only if* the op stays pending
func (fc *FileCache) updateObject(name string, flags pendingFlags) error {
log.Trace(
"FileCache::updateObject : %s, isDir=%t, isDeletion=%t",
name,
flags.isDir,
flags.isDeletion,
)

// lock the file
flock := fc.fileLocks.Get(name)
Expand All @@ -334,7 +327,7 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) bool {
// don't double upload
_, stillPending := fc.pendingOps.Load(name)
if !stillPending {
return true
return nil
}

// look up file (or folder!)
Expand All @@ -343,11 +336,16 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) bool {
localMissing := os.IsNotExist(localErr)
// in case of inconsistency, local state takes precedence (except to prevent incorrect deletions)
if !flags.isDeletion && localErr != nil {
log.Err("FileCache::updateObject : %s stat failed. Here's why: %v", name, localErr)
// there is no file to upload, so consider the inconsistency resolved
fc.pendingOps.Delete(name)
// this file failed, but we can continue with the rest of the pending ops
return true
if localMissing {
log.Err("FileCache::updateObject : %s not found. Pending operation dropped.", localPath)
fc.pendingOps.Delete(name)
// do not interrupt other pending operations
return nil
} else {
// other errors (EIO, EACCESS, out of resources, etc.)
log.Err("FileCache::updateObject : %s %v", name, localErr)
return localErr
}
}
if flags.isDeletion && !localMissing {
log.Err("FileCache::updateObject : %s exists. Ignoring deletion flag!", name)
Expand All @@ -357,48 +355,62 @@ func (fc *FileCache) updateObject(name string, flags pendingFlags) bool {
}

// update cloud
op := "deletion"
objType := "directory"
var op, objType string
var cloudErr error
if localMissing {
op = "deletion"
if flags.isDeletion && fc.notInCloud(name) {
// nothing to do
log.Info("FileCache::updateObject : %s skipping cloud deletion (not in cloud)", name)
fc.pendingOps.Delete(name)
return true
}
if flags.isDir {
// delete folder
} else if flags.isDir {
objType = "directory"
options := internal.DeleteDirOptions{Name: name}
cloudErr = fc.NextComponent().DeleteDir(options)
} else {
// delete file
objType = "file"
options := internal.DeleteFileOptions{Name: name}
cloudErr = fc.NextComponent().DeleteFile(options)
}
// ignore ENOENT
if isNotExist(cloudErr) {
log.Debug("FileCache::updateObject : %s does not exist in cloud", name)
cloudErr = nil
}
} else {
op = "creation/update"
if info.IsDir() {
// upload folder
objType = "directory"
options := internal.CreateDirOptions{Name: name, Mode: info.Mode()}
cloudErr = fc.NextComponent().CreateDir(options)
} else {
// upload file
objType = "file"
cloudErr = fc.uploadFile(name)
}
// ignore EEXIST
if errors.Is(cloudErr, os.ErrExist) {
log.Debug("FileCache::updateObject : %s already exists in cloud", name)
cloudErr = nil
}
}

// drop any ops that will never succeed
if errors.Is(cloudErr, syscall.ENAMETOOLONG) {
// the key is too long for the backend, so retrying can never succeed
log.Err("FileCache::updateObject : %s key too long, abandoning upload", name)
fc.pendingOps.Delete(name)
// do not interrupt other pending operations
return nil
}
// handle errors

// any remaining errors should interrupt all pending operations and delay retries
if cloudErr != nil {
log.Err("FileCache::updateObject : %s %s %s failed [%v]", name, objType, op, cloudErr)
return false
err := fmt.Errorf("%s %s %s failed [%w]", name, objType, op, cloudErr)
log.Err("FileCache::updateObject : %v", err)
return err
}

// update state
log.Info("FileCache::updateObject : %s sync successful", name)
fc.pendingOps.Delete(name)

return true
return nil
}

// returns true if we *know* that this entity does not exist in cloud storage
Expand Down
59 changes: 18 additions & 41 deletions component/file_cache/file_cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ type FileCache struct {
policyTrace bool
missedChmodList sync.Map // uses object name (common.JoinUnixFilepath)
pendingOps sync.Map // uses object name (common.JoinUnixFilepath)
pendingOpAdded chan struct{} // signals when an offline operation is queued
pendingOpAdded chan struct{} // signals when a pending operation is queued
mountPath string // uses os.Separator (filepath.Join)
allowOther bool
offloadIO bool
Expand Down Expand Up @@ -194,10 +194,8 @@ func (fc *FileCache) Start(ctx context.Context) error {
} else {
close(fc.startScheduledUploads)
}
if fc.syncsPendingOps() {
fc.pendingOpAdded = make(chan struct{}, 1)
fc.pendingOpsWorker.Go(fc.servicePendingOps)
}
fc.pendingOpAdded = make(chan struct{}, 1)
fc.pendingOpsWorker.Go(fc.servicePendingOps)

return nil
}
Expand Down Expand Up @@ -1833,40 +1831,26 @@ func (fc *FileCache) releaseFileInternal(
// if file has not been interactively read or written to by end user, then there is no cached file to close.
_, noCachedHandle := options.Handle.GetValue("openFileOptions")

var err error
if !noCachedHandle {
// update the cache policy
_ = fc.FileUsed(options.Handle.Path)
// sync
if options.Handle.Dirty() {
// flush the local file
err := fc.flushFileLocal(options.Handle)
if err != nil {
return err
}
// upload
// flock is already locked
flushOptions := internal.FlushFileOptions{Handle: options.Handle, CloseInProgress: true}
err = fc.flushFileCloud(flushOptions)
if err != nil {
return err
}
}
// close
f := options.Handle.GetFileObject()
if f == nil {
log.Err("FileCache::releaseFileInternal : %s missing fd in handle", options.Handle.Path)
return syscall.EBADF
}
err := f.Close()
if err != nil {
log.Err(
"FileCache::releaseFileInternal : %s (%d) close failed [%v]",
// upload (flock is already locked)
flushOptions := internal.FlushFileOptions{Handle: options.Handle, CloseInProgress: true}
Comment thread
jfantinhardesty marked this conversation as resolved.
Resolved
err = fc.flushFileCloud(flushOptions)
// the host owns capacity, so ENOSPC isn't retried behind its back
if err != nil && !errors.Is(err, syscall.ENOSPC) {
// the handle can't carry the retry past close, so the queue must (this also blocks eviction)
log.Warn(
"FileCache::releaseFileInternal : %s upload queued for retry",
options.Handle.Path,
int(f.Fd()),
err,
)
return err
fc.addPendingOp(options.Handle.Path, pendingFlags{})
}
// clears the flag if the upload failed, since the handle is going away
fc.clearHandleDirty(options.Handle)
// close the local file
err = errors.Join(err, options.Handle.GetFileObject().Close())
}

flock.Dec()
Expand All @@ -1876,7 +1860,7 @@ func (fc *FileCache) releaseFileInternal(
flock.LazyOpen = false
}

return nil
return err
}

// ReadInBuffer: Read the local file into a buffer
Expand Down Expand Up @@ -2105,18 +2089,11 @@ func (fc *FileCache) flushFileCloud(options internal.FlushFileOptions) error {
case err == nil:
fc.clearHandleDirty(options.Handle)
fc.pendingOps.Delete(options.Handle.Path)
case errors.Is(err, syscall.ENOSPC):
log.Err("FileCache::flushFileCloud : %s upload failed [%v]", options.Handle.Path, err)
case isOffline(err) && fc.offlineAccess:
log.Warn("FileCache::flushFileCloud : %s upload delayed (offline)", options.Handle.Path)
// add file to upload queue
fc.addPendingOp(options.Handle.Path, pendingFlags{})
err = nil
case fc.syncsPendingOps():
log.Err("FileCache::flushFileCloud : %s upload failed [%v]", options.Handle.Path, err)
// queue for retry, which also protects the cached data from eviction.
// the error is still returned - the caller must not assume the data is safe
fc.addPendingOp(options.Handle.Path, pendingFlags{})
default:
log.Err("FileCache::flushFileCloud : %s upload failed [%v]", options.Handle.Path, err)
}
Expand Down
Loading
Loading