diff --git a/pkg/fs/file.go b/pkg/fs/file.go index b6ef1bb..98061dd 100644 --- a/pkg/fs/file.go +++ b/pkg/fs/file.go @@ -42,6 +42,7 @@ func (p *preload) Read(data []byte) (n int, err error) { if p.f.hasChunk(n) { p.skipped++ _, err = p.f.Seek(int64(n), io.SeekCurrent) + n = 0 return } n, err = p.f.readToCache(data) @@ -49,8 +50,9 @@ func (p *preload) Read(data []byte) (n int, err error) { return } -func (f *File) Preload(ctx context.Context, fin func(), errf func(), stop func()) { +func (f *File) Preload(ctx context.Context, r *Rate, fin, errf, stop func()) { log := f.log + defer r.Stop() defer f.Close() if f.offline { log.V(2).Error(errors.New("no preload in offline mode"), "error preloading file") @@ -65,7 +67,8 @@ func (f *File) Preload(ctx context.Context, fin func(), errf func(), stop func() log.V(2).Info("preload started") p := &preload{f: f, ctx: ctx} - _, err := io.Copy(Discard, p) + r.SetReader(p) + _, err := io.Copy(Discard, r) if err == context.Canceled { log.V(2).Info("preload canceled", "skipped", p.skipped, "written", p.written) stop() diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go index 303d3b5..b670167 100644 --- a/pkg/fs/fs.go +++ b/pkg/fs/fs.go @@ -126,12 +126,7 @@ const ( TiB ) -func (fs *FS) FileSize(name string) (size string) { - fi, err := fs.Stat(name) - if err != nil { - return - } - s := fi.Size() +func FileSize(s int64) string { switch { case s >= TiB: return strconv.Itoa(int(s/TiB)) + " TiB" @@ -145,6 +140,14 @@ func (fs *FS) FileSize(name string) (size string) { return strconv.Itoa(int(s)) + " B" } +func (fs *FS) FileSize(name string) string { + fi, err := fs.Stat(name) + if err != nil { + return "" + } + return FileSize(fi.Size()) +} + func (fs *FS) RemoveDst(name string) error { return fs.dst.Remove(name) } diff --git a/pkg/fs/preload.go b/pkg/fs/preload.go index 945e44c..7030c41 100644 --- a/pkg/fs/preload.go +++ b/pkg/fs/preload.go @@ -4,7 +4,10 @@ package fs import ( "context" + "io" + "strconv" "sync" + "sync/atomic" "time" "github.com/go-logr/logr" @@ -54,14 +57,89 @@ func (q *queue) GetPreload(name string) *Preload { return nil } +type Rate struct { + r io.Reader + done chan struct{} + rate atomic.Pointer[string] + num int64 + last int64 + interval int +} + +var zeroRate = fmtRate(0) + +func newRate() *Rate { + r := &Rate{ + done: make(chan struct{}), + interval: 10, + } + r.rate.Store(&zeroRate) + return r +} + +func (r *Rate) SetReader(reader io.Reader) { + r.r = reader + go r.sampler() +} + +func (r *Rate) Read(b []byte) (n int, err error) { + n, err = r.r.Read(b) + atomic.AddInt64(&r.num, int64(n)) + return +} + +func fmtRate(r int) string { + switch { + case r >= GiB: + return strconv.FormatFloat(float64(r)/GiB, 'f', 2, 64) + " GiB/s" + case r >= MiB: + return strconv.FormatFloat(float64(r)/MiB, 'f', 2, 64) + " MiB/s" + case r >= KiB: + return strconv.FormatFloat(float64(r)/KiB, 'f', 2, 64) + " KiB/s" + } + return strconv.Itoa(r) + " B/s" +} + +func (r *Rate) sampler() { + t := time.NewTicker(time.Duration(r.interval) * time.Second) + for { + select { + case <-r.done: + t.Stop() + return + case <-t.C: + num := atomic.LoadInt64(&r.num) + rate := int(num-r.last) / r.interval + r.last = num + s := fmtRate(rate) + r.rate.Store(&s) + } + } +} + +func (r *Rate) String() string { + s := r.rate.Load() + if s == nil { + return "" + } + return *s +} + +func (r *Rate) Stop() { + close(r.done) + r.rate.Store(&zeroRate) +} + type Preload struct { ph *PreloadHandler + rate *Rate delay time.Duration Name string cancel func() Prio int Status int Errc int + Size int64 Running bool } @@ -132,6 +210,13 @@ func (ph *PreloadHandler) schedule() { } } +func (p *Preload) Rate() string { + if p.rate != nil { + return p.rate.String() + } + return "" +} + func (p *Preload) start() { if p.Running { return @@ -153,7 +238,10 @@ func (p *Preload) start() { ctx, cancel := context.WithCancel(context.Background()) p.Running = true p.cancel = cancel + p.rate = newRate() + p.Size = f.size() go f.Preload(ctx, + p.rate, func() { p.ph.fin <- name }, diff --git a/pkg/srv/interceptor.go b/pkg/srv/interceptor.go index 00f3d15..7ce7383 100644 --- a/pkg/srv/interceptor.go +++ b/pkg/srv/interceptor.go @@ -60,6 +60,7 @@ type file struct { URI template.HTML Anchor string Size string + Rate string Status int Prio int Errc int @@ -95,7 +96,7 @@ func (r *responseInterceptor) Status() int { return r.status } -func (r *responseInterceptor) GetPaths(path string, fs *fs.FS, relative bool) (dc dirContents, err error) { +func (r *responseInterceptor) GetPaths(path string, filesystem *fs.FS, relative bool) (dc dirContents, err error) { buf := r.buf.Bytes() buf = bytes.ReplaceAll(buf, []byte{'&'}, []byte("&")) err = xml.Unmarshal(buf, &dc) @@ -123,8 +124,8 @@ func (r *responseInterceptor) GetPaths(path string, fs *fs.FS, relative bool) (d Name: name, URI: uri, Anchor: anchor(p), - Size: fs.FileSize(full), - Status: fs.CacheStatus(full), + Size: filesystem.FileSize(full), + Status: filesystem.CacheStatus(full), }) } } @@ -133,12 +134,14 @@ func (r *responseInterceptor) GetPaths(path string, fs *fs.FS, relative bool) (d return } -func getPreloads(path string, fs *fs.FS) (dc dirContents) { +func getPreloads(path string, filesystem *fs.FS) (dc dirContents) { dc.Base = path - for _, p := range fs.Preloads() { + for _, p := range filesystem.Preloads() { dc.Files = append(dc.Files, file{ Name: template.HTML(p.Name), URI: template.HTML(p.Name), + Size: fs.FileSize(p.Size), + Rate: p.Rate(), Status: p.Status, Prio: p.Prio, Errc: p.Errc, diff --git a/pkg/srv/templates/preloads.html b/pkg/srv/templates/preloads.html index 2fd6b1a..3259872 100644 --- a/pkg/srv/templates/preloads.html +++ b/pkg/srv/templates/preloads.html @@ -24,6 +24,7 @@ Quota: {{.QuotaCur}} / {{.QuotaMax}} GiB [queued] {{end -}} {{end -}} + {{$s.Rate}} {{$s.Size}} {{$s.Prio}}[v][s][+][-]