diff --git a/pkg/fs/metadata.go b/pkg/fs/metadata.go index 57c37fc..3b8b8ae 100644 --- a/pkg/fs/metadata.go +++ b/pkg/fs/metadata.go @@ -304,11 +304,12 @@ func (mh *MetadataHandler) flush() { 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() - } + md.Close() + //if md.f != nil { + // close(md.wc) + // md.f.Sync() + // md.f.Close() + //} } mh.mu.Unlock() mh.flush() @@ -345,30 +346,43 @@ func IsIOErr(err error) bool { } 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"` - Chunks chunk.Chunks `json:"c"` + mu sync.Mutex `json:"-"` + cmu sync.RWMutex `json:"-"` + fs *FS `json:"-"` + f provider.File `json:"-"` + wc chan writeAt `json:"-"` + done chan struct{} `json:"-"` + err atomic.Pointer[mdErr] `json:"-"` + name string `json:"-"` + Size int64 `json:"s"` + Atime int64 `json:"a"` + Chunks chunk.Chunks `json:"c"` } -func (md *Metadata) Close() error { +func (md *Metadata) Close() (err error) { md.mu.Lock() defer md.mu.Unlock() - close(md.wc) - err := md.f.Close() - md.f = nil - return err + if md.wc != nil { + close(md.wc) + if md.done != nil { + <-md.done + } + md.wc = nil + } + if md.f != nil { + md.f.Sync() + err = md.f.Close() + md.f = nil + } + return } func (md *Metadata) Delete() error { md.mu.Lock() defer md.mu.Unlock() + md.cmu.Lock() md.Chunks = chunk.Chunks{} + md.cmu.Unlock() close(md.wc) md.f.Close() md.f = nil @@ -377,26 +391,26 @@ func (md *Metadata) Delete() error { } func (md *Metadata) FullyCached() bool { - md.mu.RLock() - defer md.mu.RUnlock() + md.cmu.RLock() + defer md.cmu.RUnlock() return len(md.Chunks) == 1 && md.Size == md.Chunks[0][1] } func (md *Metadata) HasChunk(off int64, n int) bool { - md.mu.RLock() - defer md.mu.RUnlock() + md.cmu.RLock() + defer md.cmu.RUnlock() return md.Chunks.Exists(off, n, md.Size) } func (md *Metadata) ChunkSize() int64 { - md.mu.RLock() - defer md.mu.RUnlock() + md.cmu.RLock() + defer md.cmu.RUnlock() return md.Chunks.Size() } func (md *Metadata) addChunk(off int64, n int) { - md.mu.Lock() - defer md.mu.Unlock() + md.cmu.Lock() + defer md.cmu.Unlock() md.Chunks.Add(off, n) } @@ -411,14 +425,24 @@ func (md *Metadata) addChunk(off int64, n int) { // return md.f.ReadAt(p, pos) //} +type mdErr struct { + err error +} + type writeAt struct { pos int64 data []byte ret *[]byte } -func (md *Metadata) writer() { +func (md *Metadata) resetChans() { md.wc = make(chan writeAt, 10) + md.done = make(chan struct{}) +} + +func (md *Metadata) writer() { + md.resetChans() + md.err.Store(nil) go func() { for wa := range md.wc { atomic.StoreInt64(&md.Atime, now()) @@ -427,18 +451,20 @@ func (md *Metadata) writer() { md.addChunk(wa.pos, n) md.fs.q.Add(n) } else { - md.err = err + md.err.Store(&mdErr{err}) } if wa.ret != nil { streamingPool.Put(wa.ret) } } + close(md.done) }() } func (md *Metadata) WriteAt(data []byte, pos int64) (int, error) { - if md.err != nil { - return 0, md.err + err := md.err.Load() + if err != nil && err.err != nil { + return 0, err.err } if md.f == nil { err := md.openCacheFile()