seaweedfs/weed/filer/reader_cache.go

217 lines
4.9 KiB
Go
Raw Normal View History

2022-02-26 10:16:47 +00:00
package filer
import (
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
"github.com/seaweedfs/seaweedfs/weed/util/mem"
"github.com/seaweedfs/seaweedfs/weed/wdclient"
2022-02-26 10:16:47 +00:00
)
type ReaderCache struct {
chunkCache chunk_cache.ChunkCache
lookupFileIdFn wdclient.LookupFileIdFunctionType
sync.Mutex
downloaders map[string]*SingleChunkCacher
limit int
}
type SingleChunkCacher struct {
sync.Mutex
parent *ReaderCache
chunkFileId string
data []byte
err error
cipherKey []byte
isGzipped bool
chunkSize int
shouldCache bool
wg sync.WaitGroup
cacheStartedCh chan struct{}
completedTimeNew int64
2022-02-26 10:16:47 +00:00
}
func newReaderCache(limit int, chunkCache chunk_cache.ChunkCache, lookupFileIdFn wdclient.LookupFileIdFunctionType) *ReaderCache {
return &ReaderCache{
limit: limit,
chunkCache: chunkCache,
lookupFileIdFn: lookupFileIdFn,
downloaders: make(map[string]*SingleChunkCacher),
}
}
func (rc *ReaderCache) MaybeCache(chunkViews []*ChunkView) {
2022-02-26 10:16:47 +00:00
if rc.lookupFileIdFn == nil {
return
}
rc.Lock()
defer rc.Unlock()
if len(rc.downloaders) >= rc.limit {
return
}
for _, chunkView := range chunkViews {
if _, found := rc.downloaders[chunkView.FileId]; found {
continue
2022-02-26 10:16:47 +00:00
}
if len(rc.downloaders) >= rc.limit {
// abort when slots are filled
2022-02-26 10:16:47 +00:00
return
}
// glog.V(4).Infof("prefetch %s offset %d", chunkView.FileId, chunkView.LogicOffset)
// cache this chunk if not yet
cacher := newSingleChunkCacher(rc, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int(chunkView.ChunkSize), false)
go cacher.startCaching()
<-cacher.cacheStartedCh
rc.downloaders[chunkView.FileId] = cacher
}
2022-02-26 10:16:47 +00:00
return
}
func (rc *ReaderCache) ReadChunkAt(buffer []byte, fileId string, cipherKey []byte, isGzipped bool, offset int64, chunkSize int, shouldCache bool) (int, error) {
rc.Lock()
2022-02-26 10:16:47 +00:00
if cacher, found := rc.downloaders[fileId]; found {
if n, err := cacher.readChunkAt(buffer, offset); n != 0 && err == nil {
rc.Unlock()
return n, err
}
2022-02-26 10:16:47 +00:00
}
if shouldCache || rc.lookupFileIdFn == nil {
n, err := rc.chunkCache.ReadChunkAt(buffer, fileId, uint64(offset))
if n > 0 {
rc.Unlock()
2022-02-26 10:16:47 +00:00
return n, err
}
}
// clean up old downloaders
2022-02-26 10:16:47 +00:00
if len(rc.downloaders) >= rc.limit {
2022-08-26 23:55:15 +00:00
oldestFid, oldestTime := "", time.Now().UnixNano()
2022-02-26 10:16:47 +00:00
for fid, downloader := range rc.downloaders {
completedTime := atomic.LoadInt64(&downloader.completedTimeNew)
if completedTime > 0 && completedTime < oldestTime {
oldestFid, oldestTime = fid, completedTime
2022-02-26 10:16:47 +00:00
}
}
if oldestFid != "" {
oldDownloader := rc.downloaders[oldestFid]
delete(rc.downloaders, oldestFid)
oldDownloader.destroy()
}
}
// glog.V(4).Infof("cache1 %s", fileId)
2022-02-26 10:16:47 +00:00
cacher := newSingleChunkCacher(rc, fileId, cipherKey, isGzipped, chunkSize, shouldCache)
go cacher.startCaching()
<-cacher.cacheStartedCh
2022-02-26 10:16:47 +00:00
rc.downloaders[fileId] = cacher
rc.Unlock()
2022-02-26 10:16:47 +00:00
return cacher.readChunkAt(buffer, offset)
}
func (rc *ReaderCache) UnCache(fileId string) {
rc.Lock()
defer rc.Unlock()
// glog.V(4).Infof("uncache %s", fileId)
if downloader, found := rc.downloaders[fileId]; found {
downloader.destroy()
delete(rc.downloaders, fileId)
}
}
2022-02-26 10:16:47 +00:00
func (rc *ReaderCache) destroy() {
rc.Lock()
defer rc.Unlock()
for _, downloader := range rc.downloaders {
downloader.destroy()
}
}
func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte, isGzipped bool, chunkSize int, shouldCache bool) *SingleChunkCacher {
return &SingleChunkCacher{
parent: parent,
chunkFileId: fileId,
cipherKey: cipherKey,
isGzipped: isGzipped,
chunkSize: chunkSize,
shouldCache: shouldCache,
cacheStartedCh: make(chan struct{}),
2022-02-26 10:16:47 +00:00
}
}
func (s *SingleChunkCacher) startCaching() {
s.wg.Add(1)
defer s.wg.Done()
2022-02-26 10:16:47 +00:00
s.Lock()
defer s.Unlock()
s.cacheStartedCh <- struct{}{} // means this has been started
2022-02-26 10:16:47 +00:00
urlStrings, err := s.parent.lookupFileIdFn(s.chunkFileId)
if err != nil {
s.err = fmt.Errorf("operation LookupFileId %s failed, err: %v", s.chunkFileId, err)
return
}
s.data = mem.Allocate(s.chunkSize)
_, s.err = retriedFetchChunkData(s.data, urlStrings, s.cipherKey, s.isGzipped, true, 0)
if s.err != nil {
mem.Free(s.data)
s.data = nil
return
}
if s.shouldCache {
s.parent.chunkCache.SetChunk(s.chunkFileId, s.data)
}
2022-08-26 23:55:15 +00:00
atomic.StoreInt64(&s.completedTimeNew, time.Now().UnixNano())
2022-02-26 10:16:47 +00:00
return
}
func (s *SingleChunkCacher) destroy() {
// wait for all reads to finish before destroying the data
s.wg.Wait()
s.Lock()
defer s.Unlock()
2022-02-26 10:16:47 +00:00
if s.data != nil {
mem.Free(s.data)
s.data = nil
close(s.cacheStartedCh)
2022-02-26 10:16:47 +00:00
}
}
func (s *SingleChunkCacher) readChunkAt(buf []byte, offset int64) (int, error) {
s.wg.Add(1)
defer s.wg.Done()
s.Lock()
defer s.Unlock()
2022-02-26 10:16:47 +00:00
if s.err != nil {
return 0, s.err
}
if len(s.data) == 0 {
return 0, nil
}
return copy(buf, s.data[offset:]), nil
2022-02-26 10:16:47 +00:00
}