From a069baf48d84d23ea5d3542e0dc7d2ae1e419477 Mon Sep 17 00:00:00 2001 From: ston1th Date: Wed, 16 Mar 2022 01:58:10 +0100 Subject: [PATCH] added preload status and cancel --- TODO.txt | 2 ++ pkg/fs/chunk.go | 7 +++++++ pkg/fs/dir.go | 32 +++++++++++++++++++++++++++++ pkg/fs/file.go | 36 +++++++++++++++++++++++++-------- pkg/fs/fs.go | 46 ++++++++++++++++++++++++++++++------------ pkg/fs/metadata.go | 19 +++++++++++++++++ pkg/fs/webdav.go | 8 ++++++-- pkg/srv/interceptor.go | 12 +++++++++-- pkg/srv/srv.go | 15 +++++++++----- pkg/srv/templates.go | 11 +++++----- 10 files changed, 153 insertions(+), 35 deletions(-) create mode 100644 pkg/fs/dir.go diff --git a/TODO.txt b/TODO.txt index 9483a5b..2f5634e 100644 --- a/TODO.txt +++ b/TODO.txt @@ -1,3 +1,5 @@ * add disk quota + * preload: dont load file if quota exceeded * add LRU cache * implement directory listing cache +* clear cache if file sizes differ and > 0 diff --git a/pkg/fs/chunk.go b/pkg/fs/chunk.go index f04838b..bd85cac 100644 --- a/pkg/fs/chunk.go +++ b/pkg/fs/chunk.go @@ -30,6 +30,13 @@ func (cs *Chunks) Add(off int64, n int) { cs.merge() } +func (cs Chunks) Size() (sum int64) { + for _, c := range cs { + sum += (c[1] - c[0]) + } + return +} + func (cs *Chunks) merge() { c := *cs if len(c) < 2 { diff --git a/pkg/fs/dir.go b/pkg/fs/dir.go new file mode 100644 index 0000000..1bf1995 --- /dev/null +++ b/pkg/fs/dir.go @@ -0,0 +1,32 @@ +// Copyright (C) 2022 Marius Schellenberger + +package fs + +import ( + "io" + "os" +) + +type Dir struct { + f *os.File +} + +func (f *Dir) Read(_ []byte) (int, error) { + return 0, io.EOF +} + +func (f *Dir) Seek(_ int64, _ int) (int64, error) { + return 0, io.EOF +} + +func (f *Dir) Readdir(count int) ([]os.FileInfo, error) { + return f.f.Readdir(count) +} + +func (f *Dir) Stat() (os.FileInfo, error) { + return f.f.Stat() +} + +func (f *Dir) Close() error { + return f.f.Close() +} diff --git a/pkg/fs/file.go b/pkg/fs/file.go index 1048208..17356b4 100644 --- a/pkg/fs/file.go +++ b/pkg/fs/file.go @@ -3,6 +3,7 @@ package fs import ( + "context" "errors" "io" "os" @@ -22,9 +23,15 @@ type preload struct { f *File skipped int64 written int64 + ctx context.Context } func (p *preload) Read(data []byte) (n int, err error) { + select { + case <-p.ctx.Done(): + return 0, p.ctx.Err() + default: + } if p.f.offset >= p.f.size() { return 0, io.EOF } @@ -39,7 +46,7 @@ func (p *preload) Read(data []byte) (n int, err error) { return } -func (f *File) Preload(unlock func()) { +func (f *File) Preload(ctx context.Context, unlock func()) { log := f.log if f.offline { log.V(2).Error(errors.New("no preload in offline mode"), "error preloading file") @@ -47,8 +54,13 @@ func (f *File) Preload(unlock func()) { return } log.V(2).Info("preload started") - p := &preload{f: f} + p := &preload{f: f, ctx: ctx} _, err := io.Copy(io.Discard, p) + if err == context.Canceled { + log.V(2).Info("preload canceled", "skipped", p.skipped, "written", p.written) + unlock() + return + } if err != nil && err != io.EOF { log.Error(err, "error preloading file") } @@ -103,10 +115,12 @@ func (f *File) readWithoutCache(p []byte) (n int, err error) { } func (f *File) Seek(offset int64, whence int) (n int64, err error) { - n, err = f.f.Seek(offset, whence) - if err != nil { - f.log.Error(err, "error seeking source file") - return + if !f.offline { + n, err = f.f.Seek(offset, whence) + if err != nil { + f.log.Error(err, "error seeking source file") + return + } } switch whence { case io.SeekStart: @@ -117,14 +131,20 @@ func (f *File) Seek(offset int64, whence int) (n int64, err error) { return } -func (f *File) Readdir(count int) ([]os.FileInfo, error) { - return f.f.Readdir(count) +func (f *File) Readdir(_ int) ([]os.FileInfo, error) { + return nil, nil } func (f *File) Stat() (os.FileInfo, error) { + if f.offline { + return f.md.Stat() + } return f.f.Stat() } func (f *File) Close() error { + if f.offline { + return nil + } return f.f.Close() } diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go index fc1a146..a21f482 100644 --- a/pkg/fs/fs.go +++ b/pkg/fs/fs.go @@ -8,6 +8,7 @@ import ( "errors" "io" stdfs "io/fs" + "math" "net/http" "os" "path" @@ -28,6 +29,7 @@ type FS struct { mdf *os.File jenc *json.Encoder mm map[string]*Metadata + prem map[string]func() cancel func() } @@ -55,6 +57,7 @@ func NewFS(src, dst, metadata string, log logr.Logger) (fs *FS, err error) { dst: dst, mdf: mdf, mm: make(map[string]*Metadata), + prem: make(map[string]func()), jenc: json.NewEncoder(mdf), } err = json.NewDecoder(mdf).Decode(&fs.mm) @@ -79,7 +82,7 @@ func (fs *FS) initMetadata() { } if len(v.Chunks) == 0 { log.V(2).Info("removing empty file", "file", k) - err := fs.removeDst(k) + err := fs.RemoveDst(k) if err != nil { log.Error(err, "error removing empty file", "file", k) continue @@ -110,7 +113,7 @@ func (fs *FS) StatWithOffline(name string) (fi stdfs.FileInfo, offline bool, err return } -func (fs *FS) removeDst(name string) error { +func (fs *FS) RemoveDst(name string) error { _, dp := fs.paths(name) return os.Remove(dp) } @@ -120,25 +123,33 @@ func (fs *FS) statDst(name string) (fi stdfs.FileInfo, err error) { return os.Stat(dp) } -func (fs *FS) Open(name string) (http.File, error) { - return fs.open(name) -} - -func (fs *FS) OpenFile(name string) (*File, error) { - return fs.open(name) +func (fs *FS) CancelPreload(name string) { + if cancel, ok := fs.prem[name]; ok { + cancel() + } } func (fs *FS) Preload(name string) { - f, err := fs.open(name) + file, err := fs.Open(name) if err != nil { return } + f, ok := file.(*File) + if !ok { + return + } + if f.md.Preload() { fs.log.WithValues("file", name).Error(errors.New("preload is running"), "error starting preload") return } - unlock := func() { f.md.UnlockPreload() } - go f.Preload(unlock) + ctx, cancel := context.WithCancel(context.Background()) + fs.prem[name] = cancel + unlock := func() { + delete(fs.prem, name) + f.md.UnlockPreload() + } + go f.Preload(ctx, unlock) } func skipLog(name string) bool { @@ -152,7 +163,7 @@ func (fs *FS) paths(name string) (sp string, dp string) { return } -func (fs *FS) open(name string) (f *File, err error) { +func (fs *FS) Open(name string) (f http.File, err error) { log := fs.log.WithValues("file", name) sp, dp := fs.paths(name) offline := false @@ -186,7 +197,7 @@ func (fs *FS) open(name string) (f *File, err error) { offline = true log.V(2).Info("dir offline mode", "path", dp) } - return &File{f: sf, offline: offline}, err + return &Dir{f: sf}, err } md := fs.metadata(name, sfi.Size()) return &File{ @@ -244,6 +255,15 @@ func (fs *FS) metadata(name string, size int64) (md *Metadata) { return } +func (fs *FS) CacheStatus(name string) int { + fs.mu.RLock() + defer fs.mu.RUnlock() + if md, ok := fs.mm[name]; ok { + return int(math.Round(float64(md.Chunks.Size()) / float64(md.Size) * 100)) + } + return -1 +} + func (fs *FS) flusher(ctx context.Context) { for { select { diff --git a/pkg/fs/metadata.go b/pkg/fs/metadata.go index d8fec28..b2d434d 100644 --- a/pkg/fs/metadata.go +++ b/pkg/fs/metadata.go @@ -17,6 +17,15 @@ type Metadata struct { preload bool `json:"-"` } +func (md *Metadata) Delete() error { + md.mu.Lock() + defer md.mu.Unlock() + md.Chunks = Chunks{} + md.f.Close() + md.f = nil + return md.fs.RemoveDst(md.name) +} + func (md *Metadata) HasChunk(off int64, n int) bool { md.mu.RLock() defer md.mu.RUnlock() @@ -75,3 +84,13 @@ func (md *Metadata) UnlockPreload() { defer md.mu.Unlock() md.preload = false } + +func (md *Metadata) Stat() (os.FileInfo, error) { + if md.f == nil { + err := md.openCacheFile() + if err != nil { + return nil, err + } + } + return md.f.Stat() +} diff --git a/pkg/fs/webdav.go b/pkg/fs/webdav.go index 1065436..908ddc8 100644 --- a/pkg/fs/webdav.go +++ b/pkg/fs/webdav.go @@ -4,13 +4,14 @@ package fs import ( "context" + "net/http" "os" "golang.org/x/net/webdav" ) type WebDavFile struct { - *File + http.File } func (*WebDavFile) Write(_ []byte) (int, error) { @@ -41,7 +42,10 @@ func (w *WebDavFS) OpenFile(_ context.Context, name string, flags int, _ os.File flags&os.O_TRUNC == os.O_TRUNC { return nil, os.ErrPermission } - f, err := w.fs.OpenFile(name) + f, err := w.fs.Open(name) + if err != nil { + return nil, err + } return &WebDavFile{f}, err } diff --git a/pkg/srv/interceptor.go b/pkg/srv/interceptor.go index bfa7581..104d366 100644 --- a/pkg/srv/interceptor.go +++ b/pkg/srv/interceptor.go @@ -8,6 +8,8 @@ import ( "net/http" "sort" "strings" + + "cachefs/pkg/fs" ) type statusInterceptor struct { @@ -44,6 +46,7 @@ type dirContents struct { type file struct { Name string Anchor string + Status int } type files []file @@ -77,16 +80,21 @@ func (r *responseInterceptor) Status() int { return r.status } -func (r *responseInterceptor) GetPaths() (dir dirContents, err error) { +func (r *responseInterceptor) GetPaths(path string, fs *fs.FS) (dir dirContents, err error) { err = xml.Unmarshal(r.buf.Bytes(), &dir) if err != nil { return } + path = strings.TrimSuffix(path, "/") for _, p := range dir.AllPaths { if p[len(p)-1] == '/' { dir.Dirs = append(dir.Dirs, p) } else { - dir.Files = append(dir.Files, file{Name: p, Anchor: anchor(p)}) + dir.Files = append(dir.Files, file{ + Name: p, + Anchor: anchor(p), + Status: fs.CacheStatus(path + "/" + p), + }) } } sort.Strings(dir.Dirs) diff --git a/pkg/srv/srv.go b/pkg/srv/srv.go index 6e8d216..8510e05 100644 --- a/pkg/srv/srv.go +++ b/pkg/srv/srv.go @@ -57,7 +57,8 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.Error(w, msg, code) return } - if r.FormValue("v") == "t" { + option := r.FormValue("o") + if option == "v" { err = video.Execute(w, r.URL.Path) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) @@ -66,8 +67,12 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { } if !d.IsDir() { - if r.FormValue("p") == "t" { - fs.fs.Preload(p) + if option == "p" || option == "sp" { + if option == "p" { + fs.fs.Preload(p) + } else { + fs.fs.CancelPreload(p) + } rdir := "/" if i := strings.LastIndex(p, "/"); i > 0 { rdir = p[:i] @@ -75,7 +80,7 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.Redirect(w, r, rdir+"#"+anchor(p), http.StatusFound) return } - if r.FormValue("n") == "t" { + if option == "n" { fs.fs.NoCache.ServeHTTP(w, r) return } @@ -94,7 +99,7 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { if offline { w.Header().Del("Last-Modified") } - paths, err := i.GetPaths() + paths, err := i.GetPaths(p, fs.fs) if err == io.EOF { w.WriteHeader(i.Status()) return diff --git a/pkg/srv/templates.go b/pkg/srv/templates.go index 7daf277..3eac8ee 100644 --- a/pkg/srv/templates.go +++ b/pkg/srv/templates.go @@ -9,9 +9,9 @@ var ( @@ -20,16 +20,17 @@ var ( [v]: show video [n]: skip file caching [p]: preload file +[s]: stop preloading - + {{range $s := .Dirs -}} {{end -}} {{range $s := .Files -}} - + {{end -}}
PathOptions
PathOptions
../
{{$s}}
{{$s.Name}}[v] [n] [p]
[v] [n] [p] [sp] {{if ge $s.Status 0}}{{$s.Status}}%{{end}}