added preloads page
This commit is contained in:
parent
b5ae5504a7
commit
333d5cb149
6 changed files with 235 additions and 75 deletions
205
pkg/fs/fs.go
205
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()
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue