From 304135be2e8735ad432a1bd6345543ad61bdeffc Mon Sep 17 00:00:00 2001 From: ston1th Date: Wed, 17 Aug 2022 23:58:46 +0200 Subject: [PATCH] more async improvements and crypto bugfix --- cmd/cachefs/main.go | 1 + pkg/fs/file.go | 27 ++++++------ pkg/fs/fs.go | 15 +++++-- pkg/fs/metadata.go | 72 ++++++++++++++++++++++++-------- pkg/fs/preload.go | 78 ++++++++++++++++++++++++----------- pkg/provider/crypto/file.go | 45 +++++++++++--------- pkg/provider/crypto/reader.go | 2 - 7 files changed, 161 insertions(+), 79 deletions(-) diff --git a/cmd/cachefs/main.go b/cmd/cachefs/main.go index 81fc171..156a6fe 100644 --- a/cmd/cachefs/main.go +++ b/cmd/cachefs/main.go @@ -147,6 +147,7 @@ func main() { cancel() } filesystem.Close() + <-filesystem.Done() log.Info("cachefs shutdown completed") os.Exit(0) } diff --git a/pkg/fs/file.go b/pkg/fs/file.go index 8b2c08c..acfc50e 100644 --- a/pkg/fs/file.go +++ b/pkg/fs/file.go @@ -48,17 +48,17 @@ func (p *preload) Read(data []byte) (n int, err error) { return } -func (f *File) Preload(ctx context.Context, unlock func()) { +func (f *File) Preload(ctx context.Context, fin func(), errf func(), stop func()) { log := f.log defer f.Close() if f.offline { log.V(2).Error(errors.New("no preload in offline mode"), "error preloading file") - unlock() + fin() return } if f.md.FullyCached() { log.V(2).Info("skipped preload for fully cached file") - unlock() + fin() return } @@ -67,18 +67,24 @@ func (f *File) Preload(ctx context.Context, unlock func()) { _, err := io.Copy(Discard, p) if err == context.Canceled { log.V(2).Info("preload canceled", "skipped", p.skipped, "written", p.written) - // do not call unlock() to keep preload in list + stop() return } + perr := false if err != nil && err != io.EOF { log.Error(err, "error preloading file") + perr = true } err = f.md.Close() if err != nil { log.Error(err, "error closing cache file") } + if perr { + errf() + return + } log.V(2).Info("preload finished", "skipped", p.skipped, "written", p.written) - unlock() + fin() } func (f *File) Read(p []byte) (n int, err error) { @@ -130,13 +136,10 @@ func (f *File) readToCache(p []byte) (n int, err error) { // async cache write buf := make([]byte, n) copy(buf, p[:n]) - go func(p []byte, offset int64, n int) { - if n, err = f.md.WriteAt(p, offset); err != nil { - log.Error(err, "error writing cache file") - return - } - f.md.AddChunk(offset, n) - }(buf, f.offset, n) + if _, err = f.md.WriteAt(buf, f.offset); err != nil { + log.Error(err, "error writing cache file") + return + } } if err != nil && err != io.EOF { log.Error(err, "error reading source file") diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go index 5aceca9..f8ef58c 100644 --- a/pkg/fs/fs.go +++ b/pkg/fs/fs.go @@ -19,8 +19,8 @@ import ( type FS struct { log logr.Logger - done chan struct{} cancel func() + pCancel func() NoCache http.Handler src provider.FS dst provider.FS @@ -61,11 +61,12 @@ func NewFS(quota int64, max int, src, srckey, dst, dstkey, metadata string, bloc } ctx, cancel := context.WithCancel(context.Background()) + pCtx, pCancel := context.WithCancel(context.Background()) fs = &FS{ log: log, - done: make(chan struct{}), cancel: cancel, + pCancel: pCancel, NoCache: http.FileServer(http.Dir(src)), src: srcfs, dst: dstfs, @@ -80,7 +81,7 @@ func NewFS(quota int64, max int, src, srckey, dst, dstkey, metadata string, bloc if err != nil { return nil, fmt.Errorf("quota: %w", err) } - fs.ph = NewPreloadHandler(ctx, fs, max, log.WithName("preload")) + fs.ph = NewPreloadHandler(pCtx, fs, max, log.WithName("preload")) fs.sc = NewStatCache(ctx) return } @@ -265,8 +266,14 @@ func (fs *FS) CacheStatus(name string) int { } func (fs *FS) Close() { + fs.pCancel() + <-fs.ph.Done() + fs.cancel() fs.src.Close() fs.dst.Close() - <-fs.done +} + +func (fs *FS) Done() <-chan struct{} { + return fs.mh.Done() } diff --git a/pkg/fs/metadata.go b/pkg/fs/metadata.go index a36e292..4390be0 100644 --- a/pkg/fs/metadata.go +++ b/pkg/fs/metadata.go @@ -29,13 +29,14 @@ func now() int64 { } type MetadataHandler struct { - mu sync.RWMutex - log logr.Logger - md map[string]*Metadata - fs *FS - dst string - f *os.File - enc *json.Encoder + mu sync.RWMutex + done chan struct{} + log logr.Logger + md map[string]*Metadata + fs *FS + dst string + f *os.File + enc *json.Encoder } func MetadataGenerator(file string, dst provider.FS, fstat, encrypted bool) error { @@ -109,10 +110,11 @@ func NewMetadataHandler(ctx context.Context, fs *FS, file, dst string, log logr. return nil, errors.New("metadata path is not absolute") } mh = &MetadataHandler{ - log: log, - md: make(map[string]*Metadata), - fs: fs, - dst: dst, + log: log, + done: make(chan struct{}), + md: make(map[string]*Metadata), + fs: fs, + dst: dst, } mh.f, err = os.OpenFile(file, os.O_RDWR|os.O_CREATE, 0o640) if err != nil { @@ -230,6 +232,7 @@ func (mh *MetadataHandler) flusher(ctx context.Context) { select { case <-ctx.Done(): mh.close() + close(mh.done) return case <-time.After(time.Minute * 5): } @@ -302,6 +305,7 @@ func (mh *MetadataHandler) close() { mh.mu.Lock() for _, md := range mh.md { if md.f != nil { + close(md.wc) md.f.Sync() md.f.Close() } @@ -309,7 +313,10 @@ func (mh *MetadataHandler) close() { mh.mu.Unlock() mh.flush() mh.f.Close() - close(mh.fs.done) +} + +func (mh *MetadataHandler) Done() <-chan struct{} { + return mh.done } type Error string @@ -341,6 +348,8 @@ type Metadata struct { mu sync.RWMutex `json:"-"` fs *FS `json:"-"` f provider.File `json:"-"` + wc chan writeAt `json:"-"` + err error `json:"-"` name string `json:"-"` Size int64 `json:"s"` Atime int64 `json:"a"` @@ -350,6 +359,7 @@ type Metadata struct { func (md *Metadata) Close() error { md.mu.Lock() defer md.mu.Unlock() + close(md.wc) err := md.f.Close() md.f = nil return err @@ -359,6 +369,7 @@ func (md *Metadata) Delete() error { md.mu.Lock() defer md.mu.Unlock() md.Chunks = chunk.Chunks{} + close(md.wc) md.f.Close() md.f = nil atomic.StoreInt64(&md.Atime, now()) @@ -383,7 +394,7 @@ func (md *Metadata) ChunkSize() int64 { return md.Chunks.Size() } -func (md *Metadata) AddChunk(off int64, n int) { +func (md *Metadata) addChunk(off int64, n int) { md.mu.Lock() defer md.mu.Unlock() md.Chunks.Add(off, n) @@ -400,16 +411,39 @@ func (md *Metadata) AddChunk(off int64, n int) { // return md.f.ReadAt(p, pos) //} +type writeAt struct { + pos int64 + data []byte +} + +func (md *Metadata) writer() { + md.wc = make(chan writeAt, 10) + go func() { + for wa := range md.wc { + atomic.StoreInt64(&md.Atime, now()) + n, err := md.f.WriteAt(wa.data, wa.pos) + if err == nil { + md.addChunk(wa.pos, n) + md.fs.q.Add(n) + } else { + md.err = err + } + } + }() +} + func (md *Metadata) WriteAt(data []byte, pos int64) (int, error) { + if md.err != nil { + return 0, md.err + } if md.f == nil { err := md.openCacheFile() if err != nil { return 0, err } } - atomic.StoreInt64(&md.Atime, now()) - md.fs.q.Add(len(data)) - return md.f.WriteAt(data, pos) + md.wc <- writeAt{pos, data} + return len(data), nil } func (md *Metadata) openCacheFile() error { @@ -419,8 +453,12 @@ func (md *Metadata) openCacheFile() error { return nil } f, err := md.fs.openCacheFile(md.name, md.Size) + if err != nil { + return err + } md.f = f - return err + md.writer() + return nil } func (md *Metadata) Stat() (os.FileInfo, error) { diff --git a/pkg/fs/preload.go b/pkg/fs/preload.go index 5a741f2..2904ea9 100644 --- a/pkg/fs/preload.go +++ b/pkg/fs/preload.go @@ -36,7 +36,7 @@ func (q *queue) Add(ph *PreloadHandler, name string, prio int) { func (q *queue) Remove(name string) { for i, p := range *q { if p.Name == name { - p.stop() + p.stop(true) *q = slices.Delete(*q, i, i+1) q.Sort() return @@ -56,12 +56,15 @@ type Preload struct { type Preloads []Preload type PreloadHandler struct { - mu sync.RWMutex - log logr.Logger - fs *FS - q queue - fin chan string - max int + mu sync.RWMutex + log logr.Logger + fs *FS + q queue + fin chan string + err chan string + stop chan struct{} + done chan struct{} + max int } func NewPreloadHandler(ctx context.Context, fs *FS, max int, log logr.Logger) (ph *PreloadHandler) { @@ -69,13 +72,16 @@ func NewPreloadHandler(ctx context.Context, fs *FS, max int, log logr.Logger) (p max = 1 } ph = &PreloadHandler{ - log: log, - fs: fs, - q: make(queue, 0), - fin: make(chan string, 1), - max: max, + log: log, + fs: fs, + q: make(queue, 0), + fin: make(chan string, 1), + err: make(chan string, 1), + stop: make(chan struct{}, 1), + done: make(chan struct{}), + max: max, } - go ph.preloadFinish(ctx) + go ph.preloadStatus(ctx) return } @@ -109,7 +115,7 @@ func (ph *PreloadHandler) schedule() { if i < ph.max { ph.q[i].start() } else { - ph.q[i].stop() + ph.q[i].stop(true) } } } @@ -131,19 +137,30 @@ func (p *Preload) start() { ctx, cancel := context.WithCancel(context.Background()) p.Running = true p.cancel = cancel - go f.Preload(ctx, func() { - p.ph.fin <- name - }) + go f.Preload(ctx, + func() { + p.ph.fin <- name + }, + func() { + p.ph.err <- name + }, + func() { + p.ph.stop <- struct{}{} + }, + ) } -func (p *Preload) stop() { +func (p *Preload) stop(collect bool) { if p.Running { - p.cancel() p.Running = false + p.cancel() + if collect { + <-p.ph.stop + } } } -func (ph *PreloadHandler) preloadFinish(ctx context.Context) { +func (ph *PreloadHandler) preloadStatus(ctx context.Context) { for { select { case <-ctx.Done(): @@ -154,19 +171,32 @@ func (ph *PreloadHandler) preloadFinish(ctx context.Context) { ph.q.Remove(name) ph.schedule() ph.mu.Unlock() + case name := <-ph.err: + ph.mu.Lock() + ph.q.Add(ph, name, -1) + ph.schedule() + ph.mu.Unlock() } } } func (ph *PreloadHandler) close() { + ph.mu.Lock() + defer ph.mu.Unlock() + n := len(ph.q) for _, p := range ph.q { - p.stop() + p.stop(false) } - for { + for i := 0; i < n; i++ { select { case <-ph.fin: - default: - return + case <-ph.err: + case <-ph.stop: } } + close(ph.done) +} + +func (ph *PreloadHandler) Done() <-chan struct{} { + return ph.done } diff --git a/pkg/provider/crypto/file.go b/pkg/provider/crypto/file.go index e2f067b..2fa848d 100644 --- a/pkg/provider/crypto/file.go +++ b/pkg/provider/crypto/file.go @@ -139,7 +139,6 @@ func (f *file) Seek(offset int64, whence int) (n int64, err error) { return } f.nonce = [NonceSize]byte{} - //inc(&f.nonce, cn) put(&f.nonce, cn) f.r.nonce = f.nonce f.w.nonce = f.nonce @@ -184,26 +183,32 @@ func (f *file) ReadAt(p []byte, pos int64) (n int, err error) { func (f *file) WriteAt(data []byte, pos int64) (n int, err error) { f.wmu.Lock() defer f.wmu.Unlock() - cn, off, woff := align(pos) - c, ok := f.w.cm[cn] - if !ok { - c = &chunkWriter{offset: off + KDFNonceSize, w: f.w, cn: cn} - //inc(&c.nonce, cn) - put(&c.nonce, cn) - f.w.cm[cn] = c - if f.w.last == nil { - f.w.last = c - } else if cn > f.w.last.cn { - f.w.last = c + for n != len(data) { + cn, off, woff := align(pos + int64(n)) + c, ok := f.w.cm[cn] + if !ok { + c = &chunkWriter{offset: off + KDFNonceSize, w: f.w, cn: cn} + put(&c.nonce, cn) + f.w.cm[cn] = c + if f.w.last == nil { + f.w.last = c + } else if cn > f.w.last.cn { + f.w.last = c + } + } + w, e := c.writeAt(data[n:], woff) + n += w + if c.full() { + // async flush + go func() { + c.flush(notLastChunk) + }() + delete(f.w.cm, cn) + } + if e != nil { + err = e + return } - } - n, err = c.writeAt(data, woff) - if c.full() { - // async flush - go func() { - c.flush(notLastChunk) - }() - delete(f.w.cm, cn) } return } diff --git a/pkg/provider/crypto/reader.go b/pkg/provider/crypto/reader.go index 93e65f8..9134b11 100644 --- a/pkg/provider/crypto/reader.go +++ b/pkg/provider/crypto/reader.go @@ -97,7 +97,6 @@ func (r *reader) readChunk() (last bool, err error) { return false, err } - //outBuf := make([]byte, 0, ChunkSize) out, err := r.a.Open(r.outBuf, r.nonce[:], in, nil) if err != nil && !last { // Check if this was a full-length final chunk. @@ -110,7 +109,6 @@ func (r *reader) readChunk() (last bool, err error) { } incNonce(&r.nonce) - //r.unread = r.buf[:copy(r.buf[:], out)] size := FullSize if len(out) < FullSize { size = len(out)