diff --git a/fs/cache.go b/fs/cache.go index 7920a60d..95de8f65 100644 --- a/fs/cache.go +++ b/fs/cache.go @@ -37,6 +37,11 @@ type Filesystem struct { // tracks currently open directories opendirsM sync.RWMutex opendirs map[uint64][]*Inode + + // content downloads currently in progress, keyed by item ID. Closed + // when the download finishes. + downloadsM sync.Mutex + downloads map[string]chan struct{} } // boltdb buckets @@ -107,6 +112,7 @@ func NewFilesystem(auth *graph.Auth, cacheDir string) *Filesystem { db: db, auth: auth, opendirs: make(map[uint64][]*Inode), + downloads: make(map[string]chan struct{}), } rootItem, err := graph.GetItem("root", auth) diff --git a/fs/fs.go b/fs/fs.go index 1c8fbc81..cb68f288 100644 --- a/fs/fs.go +++ b/fs/fs.go @@ -12,6 +12,7 @@ import ( "github.com/hanwen/go-fuse/v2/fuse" "github.com/jstaf/onedriver/fs/graph" + "github.com/rs/zerolog" "github.com/rs/zerolog/log" ) @@ -482,13 +483,6 @@ func (f *Filesystem) Open(cancel <-chan struct{}, in *fuse.OpenIn, out *fuse.Ope ctx.Debug().Msg("") - // we have something on disk- - // verify content against what we're supposed to have - inode.Lock() - defer inode.Unlock() - // stay locked until end to prevent multiple Opens() from competing for - // downloads of the same file. - // try grabbing from disk fd, err := f.content.Open(id) if err != nil { @@ -501,20 +495,80 @@ func (f *Filesystem) Open(cancel <-chan struct{}, in *fuse.OpenIn, out *fuse.Ope return fuse.OK } - if inode.VerifyChecksum(graph.QuickXORHashStream(fd)) { - // disk content is only used if the checksums match - ctx.Info().Msg("Found content in cache.") + // The inode lock is only held briefly to read or update metadata, never + // while hashing or downloading, so GetAttr/Lookup/ReadDirPlus on this + // inode are not blocked by a slow download. Concurrent Opens of the same + // file are serialized by a per-ID download lock instead. + for attempt := 0; ; attempt++ { + inode.RLock() + remoteHash := "" + if inode.DriveItem.File != nil { + remoteHash = inode.DriveItem.File.Hashes.QuickXorHash + } + inode.RUnlock() - // we check size ourselves in case the API file sizes are WRONG (it happens) - st, err := fd.Stat() - if err != nil { - ctx.Error().Err(err).Msg("Could not fetch file stats.") - return fuse.EIO + if remoteHash != "" && strings.EqualFold(remoteHash, graph.QuickXORHashStream(fd)) { + // disk content is only used if the checksums match + ctx.Info().Msg("Found content in cache.") + + // we check size ourselves in case the API file sizes are WRONG (it happens) + st, err := fd.Stat() + if err != nil { + ctx.Error().Err(err).Msg("Could not fetch file stats.") + return fuse.EIO + } + inode.Lock() + inode.DriveItem.Size = uint64(st.Size()) + inode.Unlock() + return fuse.OK } - inode.DriveItem.Size = uint64(st.Size()) - return fuse.OK + + unlock, waited := f.lockDownload(id) + if waited { + // someone else just finished downloading this file, re-check the cache + continue + } + status := f.downloadContent(ctx, inode, fd, remoteHash) + unlock() + if status == fuse.EAGAIN && attempt < 3 { + // the inode was replaced while downloading (for instance by a + // delta racing a Lookup), so retry with the current one + if inode = f.GetID(id); inode == nil { + return fuse.ENOENT + } + continue + } + return status } +} + +// lockDownload takes the download lock for an ID. If another download of the +// same ID is already running, it blocks until that one is done and returns +// waited=true. Otherwise the caller owns the lock and must call unlock. +func (f *Filesystem) lockDownload(id string) (unlock func(), waited bool) { + f.downloadsM.Lock() + if ch, ok := f.downloads[id]; ok { + f.downloadsM.Unlock() + <-ch + return nil, true + } + ch := make(chan struct{}) + f.downloads[id] = ch + f.downloadsM.Unlock() + return func() { + f.downloadsM.Lock() + delete(f.downloads, id) + f.downloadsM.Unlock() + close(ch) + }, false +} +// downloadContent fetches an item's content from the API without holding the +// inode lock and installs it into the content cache file fd, unless the inode +// changed while downloading. It returns EAGAIN if the inode was replaced and +// ENOENT if it was deleted. +func (f *Filesystem) downloadContent(ctx zerolog.Logger, inode *Inode, fd *os.File, remoteHash string) fuse.Status { + id := inode.ID() ctx.Info().Msg( "Not using cached item due to file hash mismatch, fetching content from API.", ) @@ -527,13 +581,33 @@ func (f *Filesystem) Open(cancel <-chan struct{}, in *fuse.OpenIn, out *fuse.Ope return fuse.EIO } defer f.content.Delete(tempID) + temp.Seek(0, 0) + temp.Truncate(0) // replace content only on a match size, err := graph.GetItemContentStream(id, f.auth, temp) - if err != nil || !inode.VerifyChecksum(graph.QuickXORHashStream(temp)) { + if err != nil || remoteHash == "" || + !strings.EqualFold(remoteHash, graph.QuickXORHashStream(temp)) { ctx.Error().Err(err).Msg("Failed to fetch remote content.") return fuse.EREMOTEIO } + + inode.Lock() + defer inode.Unlock() + if current := f.GetID(id); current != inode { + // Deleted, re-ID'd or replaced (e.g. by a delta) while downloading. + // Don't report success: the cache file was never filled, so the + // caller would read empty content. + ctx.Warn().Msg("Inode changed during download, discarding content.") + if current == nil { + return fuse.ENOENT + } + return fuse.EAGAIN + } + if inode.hasChanges { + ctx.Warn().Msg("Local changes made during download, keeping local content.") + return fuse.OK + } temp.Seek(0, 0) // being explicit, even though already done in hashstream func fd.Seek(0, 0) fd.Truncate(0)