diff --git a/cmd/cachefs/main.go b/cmd/cachefs/main.go index 3c5aa38..2b4386b 100644 --- a/cmd/cachefs/main.go +++ b/cmd/cachefs/main.go @@ -29,6 +29,7 @@ var ( listenWebdav string listenCache string quota int64 + maxq int log logr.Logger ) @@ -43,13 +44,14 @@ func main() { flag.StringVar(&listenHttp, "listen", "127.0.0.1:8080", "listen addr:port") flag.StringVar(&listenWebdav, "webdav", "", "listen addr:port for webdav") flag.StringVar(&listenCache, "cache", "", "listen addr:port for cache only") + flag.IntVar(&maxq, "maxq", -1, "max parallel preloads") flag.Int64Var("a, "quota", 1, "max disk usage quota in GiB") flag.Parse() log = klogr.New().WithName("main") log.Info("starting cachefs", "version", version) - filesystem, err := fs.NewFS(quota*gib, src, dst, metadata, klogr.New().WithName("fs")) + filesystem, err := fs.NewFS(quota*gib, maxq, src, dst, metadata, klogr.New().WithName("fs")) if err != nil { klog.Fatalf("init failed: %s", err) } diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go index abf0598..72fb709 100644 --- a/pkg/fs/fs.go +++ b/pkg/fs/fs.go @@ -19,26 +19,44 @@ import ( "github.com/go-logr/logr" "golang.org/x/exp/maps" + "golang.org/x/exp/slices" ) type FS struct { - mu sync.RWMutex - log logr.Logger - NoCache http.Handler - src string - dst string - statCancel func() - flushCancel func() - mdf *os.File - jenc *json.Encoder - dc *DirCache - sc *StatCache - q *Quota - md MetadataMap - pm map[string]func() + mu sync.RWMutex + log logr.Logger + NoCache http.Handler + src string + dst string + statCancel func() + flushCancel func() + preloadCancel func() + mdf *os.File + jenc *json.Encoder + dc *DirCache + sc *StatCache + q *Quota + md MetadataMap + + pm map[string]*Preload + prec chan *Preload + pref chan string + preq int + maxq int } -func NewFS(max int64, src, dst, metadata string, log logr.Logger) (fs *FS, err error) { +type Preload struct { + Name string + cancel func() + Status int + Running bool +} + +type Preloads []*Preload + +func (Preloads) Less(i, j *Preload) bool { return i.Name < j.Name } + +func NewFS(max int64, maxq int, src, dst, metadata string, log logr.Logger) (fs *FS, err error) { if !filepath.IsAbs(src) { return nil, errors.New("src path is not absolute") } @@ -57,19 +75,30 @@ func NewFS(max int64, src, dst, metadata string, log logr.Logger) (fs *FS, err e } statCtx, statCancel := context.WithCancel(context.Background()) flushCtx, flushCancel := context.WithCancel(context.Background()) + preloadCtx, preloadCancel := context.WithCancel(context.Background()) + + if maxq < -1 { + maxq = -1 + } + fs = &FS{ - log: log, - NoCache: http.FileServer(http.Dir(src)), - src: src, - dst: dst, - mdf: mdf, - statCancel: statCancel, - flushCancel: flushCancel, - dc: NewDirCache(), - sc: NewStatCache(statCtx), - md: make(MetadataMap), - pm: make(map[string]func()), - jenc: json.NewEncoder(mdf), + log: log, + NoCache: http.FileServer(http.Dir(src)), + src: src, + dst: dst, + mdf: mdf, + statCancel: statCancel, + flushCancel: flushCancel, + preloadCancel: preloadCancel, + dc: NewDirCache(), + sc: NewStatCache(statCtx), + md: make(MetadataMap), + jenc: json.NewEncoder(mdf), + + pm: make(map[string]*Preload), + prec: make(chan *Preload, 1), + pref: make(chan string, 1), + maxq: maxq, } fs.q, err = NewQuota(max, fs, fs.log.WithName("quota")) if err != nil { @@ -82,6 +111,8 @@ func NewFS(max int64, src, dst, metadata string, log logr.Logger) (fs *FS, err e fs.initMetadata() fs.q.Init() go fs.flusher(flushCtx) + go fs.preload(preloadCtx) + go fs.preloadFinish(preloadCtx) return fs, err } @@ -173,32 +204,103 @@ func (fs *FS) OpenDst(name string) (f http.File, err error) { } func (fs *FS) CancelPreload(name string) { - if cancel, ok := fs.pm[name]; ok { - cancel() + fs.mu.Lock() + defer fs.mu.Unlock() + if p, ok := fs.pm[name]; ok && p.Running { + p.cancel() } } -func (fs *FS) Preload(name string) { - file, err := fs.Open(name) - if err != nil { - return +func (fs *FS) Preloads() (ps Preloads) { + fs.mu.RLock() + defer fs.mu.RUnlock() + for _, p := range fs.pm { + ps = append(ps, &Preload{ + Name: p.Name, + Running: p.Running, + Status: fs.CacheStatus(p.Name), + }) } - f, ok := file.(*File) + slices.SortFunc(ps, ps.Less) + return +} + +func (fs *FS) Preload(name string) { + fs.mu.Lock() + defer fs.mu.Unlock() + + p, ok := fs.pm[name] if !ok { + p = &Preload{Name: name} + fs.pm[name] = p + if fs.maxq == -1 || fs.preq < fs.maxq { + fs.prec <- p + } return } - if f.md.Preload() { + if p.Running { fs.log.WithValues("file", name).Error(errors.New("preload is running"), "error starting preload") + } + return +} + +func (fs *FS) nextPreload() { + if fs.preq == fs.maxq { return } - ctx, cancel := context.WithCancel(context.Background()) - fs.pm[name] = cancel - unlock := func() { - delete(fs.pm, name) - f.md.UnlockPreload() + for _, p := range fs.pm { + if p.Running { + continue + } + fs.prec <- p + return + } +} + +func (fs *FS) preload(ctx context.Context) { + var p *Preload + for { + select { + case <-ctx.Done(): + return + case p = <-fs.prec: + } + name := p.Name + file, err := fs.Open(name) + if err != nil { + fs.log.Error(err, "error staring next preload", "file", name) + return + } + f, ok := file.(*File) + if !ok { + continue + } + ctx, cancel := context.WithCancel(context.Background()) + fs.mu.Lock() + fs.preq++ + p.Running = true + p.cancel = cancel + fs.mu.Unlock() + go f.Preload(ctx, func() { + fs.pref <- name + }) + } +} + +func (fs *FS) preloadFinish(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case name := <-fs.pref: + fs.mu.Lock() + delete(fs.pm, name) + fs.preq-- + fs.nextPreload() + fs.mu.Unlock() + } } - go f.Preload(ctx, unlock) } func skipLog(name string) bool { @@ -390,15 +492,32 @@ func (fs *FS) flushMetadata() { } func (fs *FS) Close() { + fs.mu.Lock() + fs.preloadCancel() fs.flushCancel() fs.statCancel() - for _, cancel := range fs.pm { - cancel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + for { + select { + case <-ctx.Done(): + return + case <-fs.prec: + case <-fs.pref: + } + } + }() + for _, p := range fs.pm { + if p.Running { + p.cancel() + } } for _, md := range fs.md { md.f.Sync() md.f.Close() } + fs.mu.Unlock() fs.flushMetadata() } diff --git a/pkg/fs/metadata.go b/pkg/fs/metadata.go index b450e68..a59e0a2 100644 --- a/pkg/fs/metadata.go +++ b/pkg/fs/metadata.go @@ -16,14 +16,13 @@ func now() int64 { type MetadataMap map[string]*Metadata type Metadata struct { - mu sync.RWMutex `json:"-"` - fs *FS `json:"-"` - f *os.File `json:"-"` - name string `json:"-"` - Size int64 `json:"s"` - Atime int64 `json:"a"` - Chunks Chunks `json:"c"` - preload bool `json:"-"` + mu sync.RWMutex `json:"-"` + fs *FS `json:"-"` + f *os.File `json:"-"` + name string `json:"-"` + Size int64 `json:"s"` + Atime int64 `json:"a"` + Chunks Chunks `json:"c"` } func (md *Metadata) Delete() error { @@ -82,22 +81,6 @@ func (md *Metadata) openCacheFile() error { return err } -func (md *Metadata) Preload() bool { - md.mu.Lock() - defer md.mu.Unlock() - if md.preload { - return true - } - md.preload = true - return false -} - -func (md *Metadata) UnlockPreload() { - md.mu.Lock() - defer md.mu.Unlock() - md.preload = false -} - func (md *Metadata) Stat() (os.FileInfo, error) { if md.f == nil { err := md.openCacheFile() diff --git a/pkg/srv/interceptor.go b/pkg/srv/interceptor.go index 31ab393..2c7ef6f 100644 --- a/pkg/srv/interceptor.go +++ b/pkg/srv/interceptor.go @@ -56,10 +56,11 @@ type dirs []dir func (dirs) Less(i, j dir) bool { return i.Name < j.Name } type file struct { - Name template.HTML - URI template.HTML - Anchor string - Status int + Name template.HTML + URI template.HTML + Anchor string + Status int + Running bool } type files []file @@ -128,6 +129,20 @@ func (r *responseInterceptor) GetPaths(path string, fs *fs.FS, relative bool) (d return } +func getPreloads(path string, fs *fs.FS) (dc dirContents) { + dc.Base = path + for _, p := range fs.Preloads() { + dc.Files = append(dc.Files, file{ + Name: template.HTML(p.Name), + URI: template.HTML(p.Name), + Status: p.Status, + Running: p.Running, + }) + } + slices.SortFunc(dc.Files, dc.Files.Less) + return +} + func anchor(s string) string { if i := strings.LastIndex(s, "/"); i > 0 && i+1 < len(s) { s = s[i+1:] diff --git a/pkg/srv/srv.go b/pkg/srv/srv.go index 66fed6a..e8cde9a 100644 --- a/pkg/srv/srv.go +++ b/pkg/srv/srv.go @@ -95,6 +95,15 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { fs.h.ServeHTTP(w, r) return } + if option == "preloads" { + h.Set(csp, indexCSP) + err = preloads.Execute(w, getPreloads(p, fs.fs)) + if err != nil { + fs.log.Error(err, "error rendering preloads") + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } i := &responseInterceptor{w: w} r.Header.Del("If-Modified-Since") diff --git a/pkg/srv/templates.go b/pkg/srv/templates.go index 6d344c7..5707a0c 100644 --- a/pkg/srv/templates.go +++ b/pkg/srv/templates.go @@ -11,7 +11,6 @@ var ( - cachefs | {{.Base}} + cachefs | {{.Base}} ` index = template.Must(template.New("index").Parse(indexHead + ` @@ -114,7 +120,7 @@ span { 
[v]: show video
 [n]: skip file caching
 [p]: preload file
-[s]: stop preloading
+[Preloads] @@ -132,7 +138,33 @@ span { + + +{{end -}} +
../
{{$s.Name}} - {{if ge $s.Status 0}}{{$s.Status}}%{{end}}[v][n][p][s] + {{if ge $s.Status 0}}{{$s.Status}}%{{end}}[v][n][p] +
+{{if ge $s.Status 0}} +
+{{end -}} +
+ + +`)) + preloads = template.Must(template.New("preloads").Parse(indexHead + ` +
+ 
[v]: show video
+[s]: stop preloading
+ + + + + +{{range $s := .Files -}} + + +
/[dir]
{{$s.Name}} + {{if ge $s.Status 0}}{{$s.Status}}%{{end}}[running{{else}}yellow">[queued{{end}}][v][s]