added preload status and cancel

This commit is contained in:
ston1th 2022-03-16 01:58:10 +01:00
commit a069baf48d
10 changed files with 153 additions and 35 deletions

View file

@ -1,3 +1,5 @@
* add disk quota * add disk quota
* preload: dont load file if quota exceeded
* add LRU cache * add LRU cache
* implement directory listing cache * implement directory listing cache
* clear cache if file sizes differ and > 0

View file

@ -30,6 +30,13 @@ func (cs *Chunks) Add(off int64, n int) {
cs.merge() cs.merge()
} }
func (cs Chunks) Size() (sum int64) {
for _, c := range cs {
sum += (c[1] - c[0])
}
return
}
func (cs *Chunks) merge() { func (cs *Chunks) merge() {
c := *cs c := *cs
if len(c) < 2 { if len(c) < 2 {

32
pkg/fs/dir.go Normal file
View file

@ -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()
}

View file

@ -3,6 +3,7 @@
package fs package fs
import ( import (
"context"
"errors" "errors"
"io" "io"
"os" "os"
@ -22,9 +23,15 @@ type preload struct {
f *File f *File
skipped int64 skipped int64
written int64 written int64
ctx context.Context
} }
func (p *preload) Read(data []byte) (n int, err error) { 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() { if p.f.offset >= p.f.size() {
return 0, io.EOF return 0, io.EOF
} }
@ -39,7 +46,7 @@ func (p *preload) Read(data []byte) (n int, err error) {
return return
} }
func (f *File) Preload(unlock func()) { func (f *File) Preload(ctx context.Context, unlock func()) {
log := f.log log := f.log
if f.offline { if f.offline {
log.V(2).Error(errors.New("no preload in offline mode"), "error preloading file") 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 return
} }
log.V(2).Info("preload started") log.V(2).Info("preload started")
p := &preload{f: f} p := &preload{f: f, ctx: ctx}
_, err := io.Copy(io.Discard, p) _, 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 { if err != nil && err != io.EOF {
log.Error(err, "error preloading file") 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) { func (f *File) Seek(offset int64, whence int) (n int64, err error) {
n, err = f.f.Seek(offset, whence) if !f.offline {
if err != nil { n, err = f.f.Seek(offset, whence)
f.log.Error(err, "error seeking source file") if err != nil {
return f.log.Error(err, "error seeking source file")
return
}
} }
switch whence { switch whence {
case io.SeekStart: case io.SeekStart:
@ -117,14 +131,20 @@ func (f *File) Seek(offset int64, whence int) (n int64, err error) {
return return
} }
func (f *File) Readdir(count int) ([]os.FileInfo, error) { func (f *File) Readdir(_ int) ([]os.FileInfo, error) {
return f.f.Readdir(count) return nil, nil
} }
func (f *File) Stat() (os.FileInfo, error) { func (f *File) Stat() (os.FileInfo, error) {
if f.offline {
return f.md.Stat()
}
return f.f.Stat() return f.f.Stat()
} }
func (f *File) Close() error { func (f *File) Close() error {
if f.offline {
return nil
}
return f.f.Close() return f.f.Close()
} }

View file

@ -8,6 +8,7 @@ import (
"errors" "errors"
"io" "io"
stdfs "io/fs" stdfs "io/fs"
"math"
"net/http" "net/http"
"os" "os"
"path" "path"
@ -28,6 +29,7 @@ type FS struct {
mdf *os.File mdf *os.File
jenc *json.Encoder jenc *json.Encoder
mm map[string]*Metadata mm map[string]*Metadata
prem map[string]func()
cancel func() cancel func()
} }
@ -55,6 +57,7 @@ func NewFS(src, dst, metadata string, log logr.Logger) (fs *FS, err error) {
dst: dst, dst: dst,
mdf: mdf, mdf: mdf,
mm: make(map[string]*Metadata), mm: make(map[string]*Metadata),
prem: make(map[string]func()),
jenc: json.NewEncoder(mdf), jenc: json.NewEncoder(mdf),
} }
err = json.NewDecoder(mdf).Decode(&fs.mm) err = json.NewDecoder(mdf).Decode(&fs.mm)
@ -79,7 +82,7 @@ func (fs *FS) initMetadata() {
} }
if len(v.Chunks) == 0 { if len(v.Chunks) == 0 {
log.V(2).Info("removing empty file", "file", k) log.V(2).Info("removing empty file", "file", k)
err := fs.removeDst(k) err := fs.RemoveDst(k)
if err != nil { if err != nil {
log.Error(err, "error removing empty file", "file", k) log.Error(err, "error removing empty file", "file", k)
continue continue
@ -110,7 +113,7 @@ func (fs *FS) StatWithOffline(name string) (fi stdfs.FileInfo, offline bool, err
return return
} }
func (fs *FS) removeDst(name string) error { func (fs *FS) RemoveDst(name string) error {
_, dp := fs.paths(name) _, dp := fs.paths(name)
return os.Remove(dp) return os.Remove(dp)
} }
@ -120,25 +123,33 @@ func (fs *FS) statDst(name string) (fi stdfs.FileInfo, err error) {
return os.Stat(dp) return os.Stat(dp)
} }
func (fs *FS) Open(name string) (http.File, error) { func (fs *FS) CancelPreload(name string) {
return fs.open(name) if cancel, ok := fs.prem[name]; ok {
} cancel()
}
func (fs *FS) OpenFile(name string) (*File, error) {
return fs.open(name)
} }
func (fs *FS) Preload(name string) { func (fs *FS) Preload(name string) {
f, err := fs.open(name) file, err := fs.Open(name)
if err != nil { if err != nil {
return return
} }
f, ok := file.(*File)
if !ok {
return
}
if f.md.Preload() { if f.md.Preload() {
fs.log.WithValues("file", name).Error(errors.New("preload is running"), "error starting preload") fs.log.WithValues("file", name).Error(errors.New("preload is running"), "error starting preload")
return return
} }
unlock := func() { f.md.UnlockPreload() } ctx, cancel := context.WithCancel(context.Background())
go f.Preload(unlock) fs.prem[name] = cancel
unlock := func() {
delete(fs.prem, name)
f.md.UnlockPreload()
}
go f.Preload(ctx, unlock)
} }
func skipLog(name string) bool { func skipLog(name string) bool {
@ -152,7 +163,7 @@ func (fs *FS) paths(name string) (sp string, dp string) {
return 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) log := fs.log.WithValues("file", name)
sp, dp := fs.paths(name) sp, dp := fs.paths(name)
offline := false offline := false
@ -186,7 +197,7 @@ func (fs *FS) open(name string) (f *File, err error) {
offline = true offline = true
log.V(2).Info("dir offline mode", "path", dp) 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()) md := fs.metadata(name, sfi.Size())
return &File{ return &File{
@ -244,6 +255,15 @@ func (fs *FS) metadata(name string, size int64) (md *Metadata) {
return 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) { func (fs *FS) flusher(ctx context.Context) {
for { for {
select { select {

View file

@ -17,6 +17,15 @@ type Metadata struct {
preload bool `json:"-"` 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 { func (md *Metadata) HasChunk(off int64, n int) bool {
md.mu.RLock() md.mu.RLock()
defer md.mu.RUnlock() defer md.mu.RUnlock()
@ -75,3 +84,13 @@ func (md *Metadata) UnlockPreload() {
defer md.mu.Unlock() defer md.mu.Unlock()
md.preload = false 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()
}

View file

@ -4,13 +4,14 @@ package fs
import ( import (
"context" "context"
"net/http"
"os" "os"
"golang.org/x/net/webdav" "golang.org/x/net/webdav"
) )
type WebDavFile struct { type WebDavFile struct {
*File http.File
} }
func (*WebDavFile) Write(_ []byte) (int, error) { 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 { flags&os.O_TRUNC == os.O_TRUNC {
return nil, os.ErrPermission 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 return &WebDavFile{f}, err
} }

View file

@ -8,6 +8,8 @@ import (
"net/http" "net/http"
"sort" "sort"
"strings" "strings"
"cachefs/pkg/fs"
) )
type statusInterceptor struct { type statusInterceptor struct {
@ -44,6 +46,7 @@ type dirContents struct {
type file struct { type file struct {
Name string Name string
Anchor string Anchor string
Status int
} }
type files []file type files []file
@ -77,16 +80,21 @@ func (r *responseInterceptor) Status() int {
return r.status 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) err = xml.Unmarshal(r.buf.Bytes(), &dir)
if err != nil { if err != nil {
return return
} }
path = strings.TrimSuffix(path, "/")
for _, p := range dir.AllPaths { for _, p := range dir.AllPaths {
if p[len(p)-1] == '/' { if p[len(p)-1] == '/' {
dir.Dirs = append(dir.Dirs, p) dir.Dirs = append(dir.Dirs, p)
} else { } 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) sort.Strings(dir.Dirs)

View file

@ -57,7 +57,8 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, msg, code) http.Error(w, msg, code)
return return
} }
if r.FormValue("v") == "t" { option := r.FormValue("o")
if option == "v" {
err = video.Execute(w, r.URL.Path) err = video.Execute(w, r.URL.Path)
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) 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 !d.IsDir() {
if r.FormValue("p") == "t" { if option == "p" || option == "sp" {
fs.fs.Preload(p) if option == "p" {
fs.fs.Preload(p)
} else {
fs.fs.CancelPreload(p)
}
rdir := "/" rdir := "/"
if i := strings.LastIndex(p, "/"); i > 0 { if i := strings.LastIndex(p, "/"); i > 0 {
rdir = p[:i] 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) http.Redirect(w, r, rdir+"#"+anchor(p), http.StatusFound)
return return
} }
if r.FormValue("n") == "t" { if option == "n" {
fs.fs.NoCache.ServeHTTP(w, r) fs.fs.NoCache.ServeHTTP(w, r)
return return
} }
@ -94,7 +99,7 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if offline { if offline {
w.Header().Del("Last-Modified") w.Header().Del("Last-Modified")
} }
paths, err := i.GetPaths() paths, err := i.GetPaths(p, fs.fs)
if err == io.EOF { if err == io.EOF {
w.WriteHeader(i.Status()) w.WriteHeader(i.Status())
return return

View file

@ -9,9 +9,9 @@ var (
<html> <html>
<head> <head>
<style> <style>
body { text-align: center; padding: 0 150px 0 150px; font: 20px Helvetica, sans-serif; color: #333; } body { padding: 0 150px 0 150px; font: 20px Helvetica, sans-serif; color: #333; }
article { display: block; text-align: left; width: 650px; margin: 0 auto; } article { display: block; text-align: left; margin: 0 auto; }
table { text-align: left; } table { text-align: left; width: 100%; }
</style> </style>
</head> </head>
<body> <body>
@ -20,16 +20,17 @@ var (
[v]: show video [v]: show video
[n]: skip file caching [n]: skip file caching
[p]: preload file [p]: preload file
[s]: stop preloading
</pre> </pre>
<table> <table>
<tr><th>Path</th><th>Options</th></tr> <tr><th style="width:100%;">Path</th><th>Options</th></tr>
<tr><td><a href="../">../</a></td><td></td></tr> <tr><td><a href="../">../</a></td><td></td></tr>
{{range $s := .Dirs -}} {{range $s := .Dirs -}}
<tr><td><a href="{{$s}}">{{$s}}</a></td><td></td></tr> <tr><td><a href="{{$s}}">{{$s}}</a></td><td></td></tr>
{{end -}} {{end -}}
{{range $s := .Files -}} {{range $s := .Files -}}
<tr><td><a id="{{$s.Anchor}}" href="{{$s.Name}}">{{$s.Name}}</a></td> <tr><td><a id="{{$s.Anchor}}" href="{{$s.Name}}">{{$s.Name}}</a></td>
<td><a href="{{$s.Name}}?v=t">[v]</a>&nbsp;<a href="{{$s.Name}}?n=t">[n]</a>&nbsp;<a href="{{$s.Name}}?p=t">[p]</a></td></tr> <td><a href="{{$s.Name}}?o=v">[v]</a>&nbsp;<a href="{{$s.Name}}?o=n">[n]</a>&nbsp;<a href="{{$s.Name}}?o=p">[p]</a>&nbsp;<a href="{{$s.Name}}?o=sp">[sp]</a>&nbsp;{{if ge $s.Status 0}}{{$s.Status}}%{{end}}</td></tr>
{{end -}} {{end -}}
</table> </table>
</article> </article>