goofys.go

// Copyright 2015 - 2017 Ka-Hing Cheung
// Copyright 2021 Yandex LLC
// Copyright 2024 Tigris Data, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package core

import (
	"context"
	"fmt"
	"math/rand"
	"net/http"
	"net/url"
	"os"
	"runtime/debug"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/jacobsa/fuse/fuseops"
	"github.com/rs/zerolog"
	"github.com/tigrisdata/tigrisfs/core/cfg"
	"github.com/tigrisdata/tigrisfs/log"
)

// goofys is a Filey System written in Go. All the backend data is
// stored on S3 as is. It's a Filey System instead of a File System
// because it makes minimal effort at being POSIX
// compliant. Particularly things that are difficult to support on S3
// or would translate into more than one round-trip would either fail
// (rename non-empty dir) or faked (no per-file permission). goofys
// does not have a on disk data cache, and consistency model is
// close-to-open.

type Goofys struct {
	bucket string

	flags *cfg.FlagStorage

	umask uint32

	rootAttrs InodeAttributes

	bufferPool *BufferPool
	wantFree   int32

	shutdown   int32
	shutdownCh chan struct{}

	// A lock protecting the state of the file system struct itself (distinct
	// from per-inode locks). Should be always taken after any inode locks.
	mu sync.RWMutex

	flusherMu    sync.Mutex
	flusherCond  *sync.Cond
	flushPending int32

	// The next inode ID to hand out. We assume that this will never overflow,
	// since even if we were handing out inode IDs at 4 GHz, it would still take
	// over a century to do so.
	//
	// GUARDED_BY(mu)
	nextInodeID fuseops.InodeID

	// The collection of live inodes, keyed by inode ID. No ID less than
	// fuseops.RootInodeID is ever used.
	//
	// INVARIANT: For all keys k, fuseops.RootInodeID <= k < nextInodeID
	// INVARIANT: For all keys k, inodes[k].ID() == k
	// INVARIANT: inodes[fuseops.RootInodeID] is missing or of type inode.DirInode
	// INVARIANT: For all v, if IsDirName(v.Name()) then v is inode.DirInode
	//
	// GUARDED_BY(mu)
	inodes map[fuseops.InodeID]*Inode

	inodesByTime map[int64]map[fuseops.InodeID]bool

	// Inflight changes are tracked to skip them in parallel listings
	// Required because we don't have guarantees about listing & change ordering
	inflightListingId int
	inflightListings  map[int]map[string]bool
	inflightChanges   map[string]int

	nextHandleID fuseops.HandleID
	dirHandles   map[fuseops.HandleID]*DirHandle

	fileHandles map[fuseops.HandleID]*FileHandle

	activeFlushers  int64
	flushRetrySet   int32
	hasNewWrites    uint64
	flushPriorities []int64

	forgotCnt uint32

	cleanQueue BufferQueue
	inodeQueue InodeQueue

	zeroBuf []byte

	diskFdQueue *FDQueue

	stats OpStats

	NotifyCallback func(notifications []interface{})

	cloud atomic.Pointer[StorageBackend]
}

func (g *Goofys) setCloud(cloud StorageBackend) {
	g.cloud.Store(&cloud)
}

func (g *Goofys) getCloud() StorageBackend {
	if g == nil {
		return nil
	}
	return *g.cloud.Load()
}

type OpStats struct {
	reads          int64
	readHits       int64
	writes         int64
	flushes        int64
	metadataReads  int64
	metadataWrites int64
	noops          int64
	evicts         int64
	ts             time.Time
}

var (
	s3Log   = log.GetLogger("s3")
	mainLog = log.GetLogger("main")
	fuseLog = log.GetLogger("fuse")
)

func NewBackend(bucket string, flags *cfg.FlagStorage) (cloud StorageBackend, err error) {
	if flags.Backend == nil {
		flags.Backend = (&cfg.S3Config{}).Init()
	}

	if config, ok := flags.Backend.(*cfg.AZBlobConfig); ok {
		cloud, err = NewAZBlob(bucket, config)
	} else if config, ok := flags.Backend.(*cfg.ADLv1Config); ok {
		cloud, err = NewADLv1(bucket, flags, config)
	} else if config, ok := flags.Backend.(*cfg.ADLv2Config); ok {
		cloud, err = NewADLv2(bucket, flags, config)
	} else if config, ok := flags.Backend.(*cfg.S3Config); ok {
		if strings.HasSuffix(flags.Endpoint, "/storage.googleapis.com") {
			cloud, err = NewGCS3(bucket, flags, config)
		} else {
			cloud, err = NewS3(bucket, flags, config)
		}
	} else {
		err = fmt.Errorf("Unknown backend config: %T", flags.Backend)
	}

	return
}

type BucketSpec struct {
	Scheme string
	Bucket string
	Prefix string
}

func ParseBucketSpec(bucket string) (spec BucketSpec, err error) {
	if strings.Contains(bucket, "://") {
		var u *url.URL
		u, err = url.Parse(bucket)
		if err != nil {
			return
		}

		spec.Scheme = u.Scheme
		spec.Bucket = u.Host
		if u.User != nil {
			// wasb url can be wasb://container@storage-end-point
			// we want to return the entire thing as bucket
			spec.Bucket = u.User.String() + "@" + u.Host
		}
		spec.Prefix = u.Path
	} else {
		spec.Scheme = "s3"

		colon := strings.Index(bucket, ":")
		if colon != -1 {
			spec.Prefix = bucket[colon+1:]
			spec.Bucket = bucket[0:colon]
		} else {
			spec.Bucket = bucket
		}
	}

	spec.Prefix = strings.Trim(spec.Prefix, "/")
	if spec.Prefix != "" {
		spec.Prefix += "/"
	}
	return
}

func NewGoofys(ctx context.Context, bucketName string, flags *cfg.FlagStorage) (*Goofys, error) {
	if flags.DebugFuse || flags.DebugMain {
		mainLog.SetLevel(zerolog.DebugLevel)
	}
	if flags.DebugFuse {
		fuseLog.SetLevel(zerolog.DebugLevel)
	}
	if flags.DebugS3 {
		log.SetCloudLogLevel(zerolog.DebugLevel)
	}
	if flags.Backend == nil {
		if spec, err := ParseBucketSpec(bucketName); err == nil {
			switch spec.Scheme {
			case "adl":
				auth, err := cfg.AzureAuthorizerConfig{
					Log: log.GetLogger("adlv1"),
				}.Authorizer()
				if err != nil {
					err = fmt.Errorf("couldn't load azure credentials: %v",
						err)
					return nil, err
				}
				flags.Backend = &cfg.ADLv1Config{
					Endpoint:   spec.Bucket,
					Authorizer: auth,
				}
				// adlv1 doesn't really have bucket
				// names, but we will rebuild the
				// prefix
				bucketName = ""
				if spec.Prefix != "" {
					bucketName = ":" + spec.Prefix
				}
			case "wasb":
				config, err := cfg.AzureBlobConfig(flags.Endpoint, spec.Bucket, "blob")
				if err != nil {
					return nil, err
				}
				flags.Backend = &config
				if config.Container != "" {
					bucketName = config.Container
				} else {
					bucketName = spec.Bucket
				}
				if config.Prefix != "" {
					spec.Prefix = config.Prefix
				}
				if spec.Prefix != "" {
					bucketName += ":" + spec.Prefix
				}
			case "abfs":
				config, err := cfg.AzureBlobConfig(flags.Endpoint, spec.Bucket, "dfs")
				if err != nil {
					return nil, err
				}
				flags.Backend = &config
				/*
					if config.Container != "" {
						bucketName = config.Container
					} else {
						bucketName = spec.Bucket
					}
					if config.Prefix != "" {
						spec.Prefix = config.Prefix
					}
					if spec.Prefix != "" {
						bucketName += ":" + spec.Prefix
					}
				*/

				flags.Backend = &cfg.ADLv2Config{
					Endpoint:   config.Endpoint,
					Authorizer: &config,
				}
				bucketName = spec.Bucket
				if spec.Prefix != "" {
					bucketName += ":" + spec.Prefix
				}
			}
		}
	}
	return newGoofys(ctx, bucketName, flags, NewBackend)
}

func newGoofys(ctx context.Context, bucket string, flags *cfg.FlagStorage,
	newBackend func(string, *cfg.FlagStorage) (StorageBackend, error),
) (*Goofys, error) {
	// Set up the basic struct.
	fs := &Goofys{
		bucket:           bucket,
		flags:            flags,
		umask:            0o122,
		shutdownCh:       make(chan struct{}),
		zeroBuf:          make([]byte, 1048576),
		inflightChanges:  make(map[string]int),
		inflightListings: make(map[int]map[string]bool),
		stats: OpStats{
			ts: time.Now(),
		},
		flushPriorities: make([]int64, MAX_FLUSH_PRIORITY+1),
	}

	var prefix string
	colon := strings.Index(bucket, ":")
	if colon != -1 {
		prefix = bucket[colon+1:]
		prefix = strings.Trim(prefix, "/")
		if prefix != "" {
			prefix += "/"
		}

		fs.bucket = bucket[0:colon]
		bucket = fs.bucket
	}

	if flags.DebugS3 {
		s3Log.SetLevel(zerolog.DebugLevel)
	}

	cloud, err := newBackend(bucket, flags)
	if err != nil {
		return nil, fmt.Errorf("Unable to setup backend: %v", err)
	}

	randomObjectName := prefix + (RandStringBytesMaskImprSrc(32))
	err = cloud.Init(randomObjectName)
	if err != nil {
		return nil, fmt.Errorf("Unable to access '%v': %v", bucket, err)
	}
	_, _ = cloud.MultipartExpire(&MultipartExpireInput{})

	now := time.Now()
	fs.rootAttrs = InodeAttributes{
		Size:  4096,
		Ctime: now,
		Mtime: now,
	}

	if os.Getenv("GOGC") == "" {
		// Set garbage collection ratio to 20 instead of 100 by default.
		debug.SetGCPercent(20)
	}

	fs.bufferPool = NewBufferPool(int64(flags.MemoryLimit), uint64(flags.GCInterval)<<20)
	fs.bufferPool.FreeSomeCleanBuffers = func(size int64) (int64, bool) {
		return fs.FreeSomeCleanBuffers(size)
	}

	fs.nextInodeID = fuseops.RootInodeID + 1
	fs.inodes = make(map[fuseops.InodeID]*Inode)
	fs.inodesByTime = make(map[int64]map[fuseops.InodeID]bool)
	fs.setCloud(cloud)
	root := NewInode(fs, nil, "")
	root.refcnt = 1
	root.Id = fuseops.RootInodeID
	root.ToDir()
	root.dir.mountPrefix = prefix
	root.userMetadata = make(map[string][]byte)
	root.Attributes.Mtime = fs.rootAttrs.Mtime
	root.Attributes.Ctime = fs.rootAttrs.Ctime

	fs.inodes[fuseops.RootInodeID] = root

	fs.nextHandleID = 1
	fs.dirHandles = make(map[fuseops.HandleID]*DirHandle)

	fs.fileHandles = make(map[fuseops.HandleID]*FileHandle)

	fs.flusherCond = sync.NewCond(&fs.flusherMu)
	go fs.Flusher()
	if fs.flags.StatsInterval > 0 {
		go fs.StatPrinter()
	}

	if fs.flags.CachePath != "" {
		fs.diskFdQueue = NewFDQueue(int(fs.flags.MaxDiskCacheFD))
		if fs.flags.MaxDiskCacheFD > 0 {
			go fs.FDCloser()
		}
	}

	go fs.MetaEvictor()

	return fs, nil
}

func (fs *Goofys) Shutdown() {
	atomic.StoreInt32(&fs.shutdown, 1)
	close(fs.shutdownCh)
	fs.WakeupFlusher()
	if fs.diskFdQueue != nil {
		fs.diskFdQueue.cond.Broadcast()
	}
}

// from https://stackoverflow.com/questions/22892120/how-to-generate-a-random-string-of-a-fixed-length-in-golang
func RandStringBytesMaskImprSrc(n int) string {
	const letterBytes = "abcdefghijklmnopqrstuvwxyz0123456789"
	const (
		letterIdxBits = 6                    // 6 bits to represent a letter index
		letterIdxMask = 1<<letterIdxBits - 1 // All 1-bits, as many as letterIdxBits
		letterIdxMax  = 63 / letterIdxBits   // # of letter indices fitting in 63 bits
	)
	src := rand.NewSource(time.Now().UnixNano())
	b := make([]byte, n)
	// A src.Int63() generates 63 random bits, enough for letterIdxMax characters!
	for i, cache, remain := n-1, src.Int63(), letterIdxMax; i >= 0; {
		if remain == 0 {
			cache, remain = src.Int63(), letterIdxMax
		}
		if idx := int(cache & letterIdxMask); idx < len(letterBytes) {
			b[i] = letterBytes[idx]
			i--
		}
		cache >>= letterIdxBits
		remain--
	}

	return string(b)
}

func (fs *Goofys) SigUsr1() {
	fs.mu.RLock()

	mainLog.Infof("forgot %v inodes", fs.forgotCnt)
	mainLog.Infof("%v inodes", len(fs.inodes))
	fs.mu.RUnlock()
	debug.FreeOSMemory()
}

// Find the given inode. Panic if it doesn't exist.
//
// LOCKS_EXCLUDED(fs.mu)
func (fs *Goofys) getInodeOrDie(id fuseops.InodeID) (inode *Inode) {
	fs.mu.RLock()
	inode = fs.inodes[id]
	fs.mu.RUnlock()
	if inode == nil {
		panic(fmt.Sprintf("Unknown inode: %v", id))
	}

	return
}

func (fs *Goofys) AddDirHandle(dh *DirHandle) fuseops.HandleID {
	fs.mu.Lock()
	handleID := fs.nextHandleID
	fs.nextHandleID++
	fs.dirHandles[handleID] = dh
	fs.mu.Unlock()
	return handleID
}

func (fs *Goofys) AddFileHandle(fh *FileHandle) fuseops.HandleID {
	fs.mu.Lock()
	handleID := fs.nextHandleID
	fs.nextHandleID++
	fs.fileHandles[handleID] = fh
	fs.mu.Unlock()
	return handleID
}

func (fs *Goofys) StatPrinter() {
	for atomic.LoadInt32(&fs.shutdown) == 0 {
		select {
		case <-time.After(fs.flags.StatsInterval):
		case <-fs.shutdownCh:
			return
		}
		now := time.Now()
		d := now.Sub(fs.stats.ts).Seconds()
		reads := atomic.SwapInt64(&fs.stats.reads, 0)
		readHits := atomic.SwapInt64(&fs.stats.readHits, 0)
		writes := atomic.SwapInt64(&fs.stats.writes, 0)
		flushes := atomic.SwapInt64(&fs.stats.flushes, 0)
		metadataReads := atomic.SwapInt64(&fs.stats.metadataReads, 0)
		metadataWrites := atomic.SwapInt64(&fs.stats.metadataWrites, 0)
		noops := atomic.SwapInt64(&fs.stats.noops, 0)
		evicts := atomic.SwapInt64(&fs.stats.evicts, 0)
		fs.mu.RLock()
		inodeCount := len(fs.inodes)
		fs.mu.RUnlock()
		fs.stats.ts = now
		readsOr1 := float64(reads)
		if reads == 0 {
			readsOr1 = 1
		}
		mainLog.Infof(
			"I/O: %.2f read/s, %.2f %% hits, %.2f write/s; metadata: %.2f read/s, %.2f write/s, %.2f noop/s, %v alive, %.2f evict/s; %.2f flush/s",
			float64(reads)/d,
			float64(readHits)/readsOr1*100,
			float64(writes)/d,
			float64(metadataReads)/d,
			float64(metadataWrites)/d,
			float64(noops)/d,
			inodeCount,
			float64(evicts)/d,
			float64(flushes)/d,
		)
	}
}

// Close unneeded cache FDs
func (fs *Goofys) FDCloser() {
	for atomic.LoadInt32(&fs.shutdown) == 0 {
		fs.diskFdQueue.CloseExtra()
	}
}

// Try to reclaim some clean buffers
func (fs *Goofys) FreeSomeCleanBuffers(origSize int64) (int64, bool) {
	freed := int64(0)
	// Free at least 5 MB
	size := origSize
	if size < 5*1024*1024 {
		size = 5 * 1024 * 1024
	}
	var inode *Inode
	var cleanEnd, cleanQueueID uint64
	for freed < size {
		inode, cleanEnd, cleanQueueID = fs.cleanQueue.NextClean(cleanQueueID)
		if cleanQueueID == 0 {
			break
		}
		inode.mu.Lock()
		toFs := -1
		buf := inode.buffers.Get(cleanEnd)
		// Never evict buffers flushed in an incomplete (last) part
		if buf != nil && (buf.state == BUF_CLEAN || buf.state == BUF_FLUSHED_FULL) &&
			buf.ptr != nil && !inode.IsRangeLocked(buf.offset, buf.length, false) {
			fs.tryEvictToDisk(inode, buf, &toFs)
			allocated, _ := inode.buffers.EvictFromMemory(buf)
			if allocated != 0 {
				_ = fs.bufferPool.UseUnlocked(allocated, false)
				freed -= allocated
			}
		}
		inode.mu.Unlock()
		if freed >= size {
			break
		}
	}
	haveDirty := fs.inodeQueue.Size() > 0
	if freed < origSize && haveDirty {
		fs.bufferPool.mu.Unlock()
		atomic.AddInt32(&fs.wantFree, 1)
		fs.WakeupFlusherAndWait(true)
		atomic.AddInt32(&fs.wantFree, -1)
		fs.bufferPool.mu.Lock()
	}
	return freed, haveDirty
}

// FIXME: Implement disk cache size limit, add another btree.Map-based
// "LRU" queue to delete old files from the disk.
func (fs *Goofys) tryEvictToDisk(inode *Inode, buf *FileBuffer, toFs *int) {
	if fs.flags.CachePath != "" && !buf.onDisk {
		if *toFs == -1 {
			*toFs = 1
		}
		if *toFs > 0 {
			// Evict to disk
			err := inode.OpenCacheFD()
			if err != nil {
				*toFs = 0
			} else {
				_, err := inode.DiskCacheFD.WriteAt(buf.data, int64(buf.offset))
				if err != nil {
					*toFs = 0
					mainLog.Errorf("Couldn't write %v bytes at offset %v to %v: %v",
						len(buf.data), buf.offset, fs.flags.CachePath+"/"+inode.FullName(), err)
				} else {
					buf.onDisk = true
				}
			}
		}
	}
}

func (fs *Goofys) WakeupFlusherAndWait(wait bool) {
	fs.flusherMu.Lock()
	if fs.flushPending == 0 {
		fs.flushPending = 1
		fs.flusherCond.Broadcast()
	}
	if wait {
		// Wait for any result
		fs.flusherCond.Wait()
	}
	fs.flusherMu.Unlock()
}

func (fs *Goofys) WakeupFlusher() {
	fs.WakeupFlusherAndWait(false)
}

func (fs *Goofys) ScheduleRetryFlush() {
	if atomic.CompareAndSwapInt32(&fs.flushRetrySet, 0, 1) {
		time.AfterFunc(fs.flags.RetryInterval, func() {
			atomic.StoreInt32(&fs.flushRetrySet, 0)
			// Wakeup flusher after retry interval
			fs.WakeupFlusher()
		})
	}
}

// Flusher goroutine.
// Overall algorithm:
//  1. File opened => reads and writes just populate cache
//  2. File closed => flush it
//     Created or fully overwritten =>
//     => Less than 5 MB => upload in a single part
//     => More than 5 MB => upload using multipart
//     Updated => CURRENTLY:
//     => Less than 5 MB => upload in a single part
//     => More than 5 MB => update using multipart copy
//     Also we can't update less than 5 MB because it's the minimal part size
//  3. Fsync triggered => intermediate full flush (same algorithm)
//  4. Dirty memory limit reached => without on-disk cache we have to flush the whole object.
//     With on-disk cache we can unload some dirty buffers to disk.
func (fs *Goofys) Flusher() {
	var inodeID, nextQueueID uint64
	priority := 1
	for atomic.LoadInt32(&fs.shutdown) == 0 {
		fs.flusherMu.Lock()
		if fs.flushPending == 0 {
			fs.flusherCond.Wait()
		}
		fs.flushPending = 0
		fs.flusherMu.Unlock()
		attempts := 1
		if priority > 1 || priority == 1 && nextQueueID != 0 {
			attempts = 2
		}
		curPriorityOk := false
		for i := 1; i <= priority; i++ {
			curPriorityOk = curPriorityOk || atomic.LoadInt64(&fs.flushPriorities[priority]) > 0
		}
		for attempts > 0 && atomic.LoadInt64(&fs.activeFlushers) < atomic.LoadInt64(&fs.flags.MaxFlushers) {
			inodeID, nextQueueID = fs.inodeQueue.Next(nextQueueID)
			if inodeID == 0 {
				if curPriorityOk {
					break
				}
				priority++
				if priority > MAX_FLUSH_PRIORITY {
					attempts--
					priority = 1
				}
			} else {
				if atomic.CompareAndSwapUint64(&fs.hasNewWrites, 1, 0) {
					// restart from the beginning
					//inodeID, nextQueueID = 0, 0
					priority = 1
					attempts = 1
					curPriorityOk = atomic.LoadInt64(&fs.flushPriorities[1]) > 0
					continue
				}
				fs.mu.RLock()
				inode := fs.inodes[fuseops.InodeID(inodeID)]
				fs.mu.RUnlock()
				started := false
				if inode != nil {
					started = inode.TryFlush(priority)
				}
				curPriorityOk = curPriorityOk || started
			}
		}
	}
}

func (fs *Goofys) EvictEntry(id fuseops.InodeID) bool {
	fs.mu.RLock()
	childTmp := fs.inodes[id]
	fs.mu.RUnlock()
	if childTmp == nil ||
		childTmp.Id == fuseops.RootInodeID ||
		atomic.LoadInt32(&childTmp.fileHandles) > 0 ||
		atomic.LoadInt32(&childTmp.CacheState) > ST_DEAD ||
		childTmp.isDir() && atomic.LoadInt64(&childTmp.dir.ModifiedChildren) > 0 {
		return false
	}
	if !childTmp.mu.TryLock() {
		return false
	}
	// We CAN evict inodes which are still referenced by the kernel,
	// but only if they're expired!
	if childTmp.ExpireTime.After(time.Now()) {
		childTmp.mu.Unlock()
		return false
	}
	tmpParent := childTmp.Parent
	// Respect locking order: parent before child, inode before fs
	childTmp.mu.Unlock()
	if !tmpParent.mu.TryLock() {
		return false
	}
	if !childTmp.mu.TryLock() {
		tmpParent.mu.Unlock()
		return false
	}
	if childTmp.Parent != tmpParent ||
		atomic.LoadInt32(&tmpParent.fileHandles) > 0 {
		childTmp.mu.Unlock()
		tmpParent.mu.Unlock()
		return false
	}
	found := tmpParent.findChildUnlocked(childTmp.Name)
	if found == childTmp {
		tmpParent.removeChildUnlocked(childTmp)
		// Mark directory listing as unfinished
		tmpParent.dir.DirTime = time.Time{}
		tmpParent.dir.forgetDuringList = true
	}
	childTmp.resetCache()
	childTmp.SetCacheState(ST_DEAD)
	// Drop inode
	fs.mu.Lock()
	childTmp.resetExpireTime()
	delete(fs.inodes, childTmp.Id)
	fs.forgotCnt += 1
	fs.mu.Unlock()
	childTmp.mu.Unlock()
	tmpParent.mu.Unlock()
	return true
}

func (fs *Goofys) MetaEvictor() {
	retry := false
	var seen map[fuseops.InodeID]bool
	for atomic.LoadInt32(&fs.shutdown) == 0 {
		if !retry {
			select {
			case <-time.After(1 * time.Second):
			case <-fs.shutdownCh:
				return
			}
			seen = make(map[fuseops.InodeID]bool)
		}
		// Try to keep the number of cached inodes under control %)
		fs.mu.RLock()
		totalInodes := len(fs.inodes)
		toEvict := (totalInodes - fs.flags.EntryLimit) * 2
		if toEvict < 0 {
			fs.mu.RUnlock()
			retry = false
			continue
		}
		if toEvict < fs.flags.EntryLimit/100 {
			toEvict = fs.flags.EntryLimit / 100
		}
		if toEvict < 10 {
			toEvict = 10
		}
		expireUnix := time.Now().Add(-fs.flags.StatCacheTTL).Unix()
		var scan []fuseops.InodeID
		for tm, inodes := range fs.inodesByTime {
			if tm < expireUnix {
				for inode := range inodes {
					if !seen[inode] {
						scan = append(scan, inode)
					}
					if len(scan) >= toEvict {
						break
					}
				}
			}
			if len(scan) >= toEvict {
				break
			}
		}
		fs.mu.RUnlock()
		evicted := 0
		for _, id := range scan {
			if fs.EvictEntry(id) {
				evicted++
			} else {
				seen[id] = true
			}
		}
		retry = len(scan) >= toEvict && totalInodes > fs.flags.EntryLimit
		atomic.AddInt64(&fs.stats.evicts, int64(evicted))
		if len(scan) > 0 {
			mainLog.Debugf("metadata cache: alive %v, scanned %v, evicted %v", totalInodes, len(scan), evicted)
		}
	}
}

type Mount struct {
	// Mount Point relative to goofys's root mount.
	name    string
	cloud   StorageBackend
	prefix  string
	mounted bool
}

func (fs *Goofys) mount(mp *Inode, b *Mount) {
	if b.mounted {
		return
	}

	name := strings.Trim(b.name, "/")

	// create path for the mount. AttrTime is set to TIME_MAX so
	// they will never expire and be removed. But DirTime is not
	// so we will still consult the underlining cloud for listing
	// (which will then be merged with the cached result)

	for {
		idx := strings.Index(name, "/")
		if idx == -1 {
			break
		}
		dirName := name[0:idx]
		name = name[idx+1:]

		mp.mu.Lock()
		dirInode := mp.findChildUnlocked(dirName)
		if dirInode == nil {
			dirInode = NewInode(fs, mp, dirName)
			dirInode.ToDir()
			dirInode.SetAttrTime(TIME_MAX)
			dirInode.userMetadata = make(map[string][]byte)

			fs.insertInode(mp, dirInode)
		}
		mp.mu.Unlock()
		mp = dirInode
	}

	mp.mu.Lock()
	defer mp.mu.Unlock()

	prev := mp.findChildUnlocked(name)
	if prev == nil {
		mountInode := NewInode(fs, mp, name)
		mountInode.ToDir()
		mountInode.dir.mountPrefix = b.prefix
		mountInode.SetAttrTime(TIME_MAX)
		mountInode.userMetadata = make(map[string][]byte)

		fs.insertInode(mp, mountInode)

		prev = mountInode
	} else {
		if !prev.isDir() {
			panic(fmt.Sprintf("inode %v is not a directory", prev.FullName()))
		}

		// This inode might have some cached data from a parent mount.
		// Clear this cache by resetting the DirTime.
		// Note: resetDirTimeRec should be called without holding the lock.
		prev.resetDirTimeRec()
		prev.mu.Lock()
		defer prev.mu.Unlock()
		prev.dir.mountPrefix = b.prefix
		prev.SetAttrTime(TIME_MAX)

	}
	prev.addModified(1)
	fuseLog.Infof("mounted /%v", prev.FullName())
	b.mounted = true
}

func (fs *Goofys) MountAll(mounts []*Mount) {
	root := fs.getInodeOrDie(fuseops.RootInodeID)

	for _, m := range mounts {
		fs.mount(root, m)
	}
}

func (fs *Goofys) Mount(mount *Mount) {
	root := fs.getInodeOrDie(fuseops.RootInodeID)
	fs.mount(root, mount)
}

func (fs *Goofys) Unmount(mountPoint string) {
	mp := fs.getInodeOrDie(fuseops.RootInodeID)

	fuseLog.Infof("Attempting to unmount %v", mountPoint)
	path := strings.Split(strings.Trim(mountPoint, "/"), "/")
	for _, localName := range path {
		dirInode := mp.findChild(localName)
		if dirInode == nil || !dirInode.isDir() {
			fuseLog.Errorf("Failed to find directory:%v while unmounting %v. "+
				"Ignoring the unmount operation.", localName, mountPoint)
			return
		}
		mp = dirInode
	}
	mp.addModified(-1)
	mp.ResetForUnmount()
}

func (fs *Goofys) RefreshInodeCache(inode *Inode) error {
	inode.mu.Lock()
	parent := inode.Parent
	parentId := fuseops.InodeID(0)
	if parent != nil {
		parentId = parent.Id
	}
	name := inode.Name
	inodeId := inode.Id
	inode.mu.Unlock()
	inode.resetDirTimeRec()
	var mappedErr error
	var notifications []interface{}
	if parent == nil {
		// For regular directories it's enough to send one invalidation
		// message, the kernel will send forgets for their children and
		// everything will be refreshed just fine.
		// But root directory is a special case: we should invalidate all
		// inodes in it ourselves. Basically this means that we have to do
		// a listing and notify the kernel about every file in the root
		// directory.
		dh := inode.OpenDir()
		dh.mu.Lock()
		for {
			en, err := dh.ReadDir()
			if err != nil {
				mappedErr = mapAwsError(err)
				break
			}
			if en == nil {
				break
			}
			if dh.lastInternalOffset >= 2 {
				// Delete notifications are sent by ReadDir() itself
				notifications = append(notifications, &fuseops.NotifyInvalEntry{
					Parent: inode.Id,
					Name:   en.Name,
				})
			}
			dh.Next(en.Name)
		}
		_ = dh.CloseDir()
		dh.mu.Unlock()
		if fs.NotifyCallback != nil {
			fs.NotifyCallback(notifications)
		}
		return mappedErr
	}
	// Use recheckInodeByName to ensure we work with the current child instance
	// This handles cases where the inode passed to RefreshInodeCache might be
	// a stale reference from fs.inodes while parent.dir.Children has a newer instance
	_, err := parent.recheckInodeByName(name)
	mappedErr = mapAwsError(err)
	if mappedErr == syscall.ENOENT {
		notifications = append(notifications, &fuseops.NotifyDelete{
			Parent: parentId,
			Child:  inodeId,
			Name:   name,
		})
	} else {
		notifications = append(notifications, &fuseops.NotifyInvalEntry{
			Parent: parentId,
			Name:   name,
		})
	}
	if fs.NotifyCallback != nil {
		fs.NotifyCallback(notifications)
	}
	if mappedErr == syscall.ENOENT {
		// We don't mind if the file disappeared
		return nil
	}
	return mappedErr
}

// FIXME: Add similar write backoff (now it's handled by file/dir code)
func ReadBackoff(flags *cfg.FlagStorage, try func(attempt int) error) (err error) {
	interval := flags.ReadRetryInterval
	attempt := 1
	for {
		err = try(attempt)
		if err != nil {
			if shouldRetry(err) && (flags.ReadRetryAttempts < 1 || attempt < flags.ReadRetryAttempts) {
				attempt++
				time.Sleep(interval)
				interval = time.Duration(flags.ReadRetryMultiplier * float64(interval))
				if interval > flags.ReadRetryMax {
					interval = flags.ReadRetryMax