gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(flashnode): same cacheBlock may be created multiple times.

Browse Source 
close:#22922141

Signed-off-by: shuqiang-zheng <zhengshuqiang@oppo.com>
This commit is contained in:
shuqiang-zheng 2024-12-23 19:03:23 +08:00 committed by zhumingze1108
parent 110075c431
commit 76007c2d8d
3 changed files with 32 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
flashnode/cachengine/block.go
View File

@ -242,6 +242,7 @@ func (cb *CacheBlock) initFilePath(ttl int64, isLoad bool) (err error) {
cb.updateAllocSize(allocSize)
cb.maybeUpdateUsedSize(usedSize)
cb.notifyReady()
log.LogDebugf("action[initFilePath] volume(%v) inode(%v) offset(%v) version(%v) load ready", cb.volume, cb.inode, cb.fixedOffset, cb.version)
}
key := GenCacheBlockKey(cb.volume, cb.inode, cb.fixedOffset, cb.version)
@ -309,6 +310,7 @@ func (cb *CacheBlock) Init(sources []*proto.DataSource) {
return
}
cb.notifyReady()
log.LogDebugf("action[init] volume(%v) inode(%v) offset(%v) version(%v) init ready", cb.volume, cb.inode, cb.fixedOffset, cb.version)
}
func (cb *CacheBlock) prepareSource(ctx context.Context, sourceCh <-chan *proto.DataSource) (err error) {

32
flashnode/cachengine/engine.go
View File

@ -76,11 +76,12 @@ type CacheEngine struct {
dataPath string
config CacheConfig
cachePrepareTaskCh chan cachePrepareTask
cacheLoadTaskCh chan cacheLoadTask
lruCache LruCache
lruFhCache LruCache
readSourceFunc ReadExtentData
creatingCacheBlockMap sync.Map
cachePrepareTaskCh chan cachePrepareTask
cacheLoadTaskCh chan cacheLoadTask
lruCache LruCache
lruFhCache LruCache
readSourceFunc ReadExtentData
closeOnce sync.Once
closeCh chan struct{}
@ -368,6 +369,27 @@ func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64,
block = value.(*CacheBlock)
return
}
if !isLoad {
value, loaded := c.creatingCacheBlockMap.LoadOrStore(key, make(chan struct{}))
ch := value.(chan struct{})
if loaded {
select {
case <-ch:
}
v, getErr := c.lruCache.Get(key)
if getErr == nil {
block = v.(*CacheBlock)
return
}
return nil, fmt.Errorf("unable to get created cacheblock")
} else {
defer func() {
close(ch)
c.creatingCacheBlockMap.Delete(key)
}()
}
}
block = NewCacheBlock(c.dataPath, volume, inode, fixedOffset, version, allocSize, c.readSourceFunc)
if ttl <= 0 {
ttl = proto.DefaultCacheTTLSec

5
flashnode/flashnode_op.go
View File

@ -101,6 +101,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
volume = req.CacheRequest.Volume
cr := req.CacheRequest
log.LogDebugf("action[opCacheRead] req(%v)s start", req)
block, err := f.cacheEngine.GetCacheBlockForRead(volume, cr.Inode, cr.FixedFileOffset, cr.Version, req.Size_)
if err != nil {
log.LogWarnf("opCacheRead: GetCacheBlockForRead failed, req(%v) err(%v)", req, err)
@ -132,7 +133,7 @@ func (f *FlashNode) doStreamReadRequest(ctx context.Context, conn net.Conn, req
f.updateReadBytesMetric(req.Size_)
}
}()
log.LogDebugf("action[doStreamReadRequest] req(%v) start", req)
for needReplySize > 0 {
err = nil
reply := proto.NewPacket()
@ -158,7 +159,7 @@ func (f *FlashNode) doStreamReadRequest(ctx context.Context, conn net.Conn, req
reply.ExtentOffset = offset
p.Size = currReadSize
p.ExtentOffset = offset
log.LogDebugf("action[doStreamReadRequest] req(%v) block read start", req)
reply.CRC, err = block.Read(ctx, reply.Data[:], offset, int64(currReadSize))
if err != nil {
bufRelease()