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

update(client): debug and pass remote read process

Browse Source 
Signed-off-by: leonrayang <chl696@sina.com>
This commit is contained in:
leonrayang 2024-01-25 10:19:48 +08:00 committed by zhumingze1108
parent bf26b4bc3d
commit 3f6243d8b4
5 changed files with 153 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
flashnode/extent_reader.go
View File

@ -110,10 +110,11 @@ func readFromDataPartition(addr string, reqPacket *proto.Packet, afterReadFunc c
}
func getReadReply(conn *net.TCPConn, reqPacket *proto.Packet, afterReadFunc cachengine.ReadExtentAfter) (readBytes int, err error) {
buf, bufErr := proto.Buffers.Get(util.ReadBlockSize)
// todo: add cache block size buffer
buf, bufErr := proto.Buffers.Get(int(reqPacket.Size))
defer func() {
if bufErr == nil {
proto.Buffers.Put(buf[:util.ReadBlockSize])
proto.Buffers.Put(buf)
}
if err != nil {
log.LogWarnf("getReadReply: req(%v) readBytes(%v) err(%v)", reqPacket, readBytes, err)

162
sdk/data/stream/extent_client.go
View File

@ -207,19 +207,11 @@ type ExtentClient struct {
InnerReq bool
AheadRead *AheadReadCache
extentConfig *ExtentConfig
RcOldStatus bool
RcClusterEnabled bool
RcVolumeEnabled bool
RcPath string
RcAutoPrepare bool
RcTTL int64
RcReadTimeoutSec int64
RemoteCache *RemoteCache
rcPrepareCh chan *PrepareRemoteCacheRequest
stopOnce sync.Once
stopCh chan struct{}
wg sync.WaitGroup
extentConfig *ExtentConfig
RemoteCache RemoteCache
stopOnce sync.Once
stopCh chan struct{}
wg sync.WaitGroup
}
func (client *ExtentClient) UidIsLimited(uid uint32) bool {
@ -375,38 +367,35 @@ retry:
client.writeLimiter = rate.NewLimiter(writeLimit, defaultWriteLimitBurst)
client.AheadRead = NewAheadReadCache(config.AheadReadEnable, config.AheadReadTotalMem, config.AheadReadBlockTimeOut, config.AheadReadWindowCnt)
if config.MaxStreamerLimit <= 0 {
client.disableMetaCache = true
return
}
if config.MaxStreamerLimit > 0 {
if config.MaxStreamerLimit <= defaultStreamerLimit {
client.maxStreamerLimit = defaultStreamerLimit
} else if config.MaxStreamerLimit > defMaxStreamerLimit {
client.maxStreamerLimit = defMaxStreamerLimit
} else {
client.maxStreamerLimit = int(config.MaxStreamerLimit)
}
if config.MaxStreamerLimit <= defaultStreamerLimit {
client.maxStreamerLimit = defaultStreamerLimit
} else if config.MaxStreamerLimit > defMaxStreamerLimit {
client.maxStreamerLimit = defMaxStreamerLimit
client.maxStreamerLimit += fastStreamerEvictNum
log.LogInfof("max streamer limit %d", client.maxStreamerLimit)
client.streamerList = list.New()
go client.backgroundEvictStream()
} else {
client.maxStreamerLimit = int(config.MaxStreamerLimit)
client.disableMetaCache = true
}
client.maxStreamerLimit += fastStreamerEvictNum
log.LogInfof("max streamer limit %d", client.maxStreamerLimit)
client.streamerList = list.New()
client.stopCh = make(chan struct{})
client.wg.Add(1)
client.metaWrapper = config.MetaWrapper
if client.metaWrapper != nil {
client.metaWrapper.RemoteCacheBloom = client.RemoteCacheBloom
client.RemoteCacheBloom().AddUint64(0)
}
client.extentConfig = config
if client.IsRemoteCacheEnabled() {
client.InitRemoteCache()
}
client.rcPrepareCh = make(chan *PrepareRemoteCacheRequest, 1024)
client.stopCh = make(chan struct{})
client.wg.Add(1)
go client.DoPrepare()
go client.backgroundEvictStream()
client.RemoteCache.Init(client)
return
}
@ -431,82 +420,19 @@ func (client *ExtentClient) SetClientID(id uint64) (err error) {
}
func (client *ExtentClient) IsRemoteCacheEnabled() bool {
return client.RcClusterEnabled && client.RcVolumeEnabled
return client.RemoteCache.ClusterEnabled && client.RemoteCache.VolumeEnabled
}
func (client *ExtentClient) enableRemoteCacheCluster(enabled bool) {
client.RcOldStatus = client.IsRemoteCacheEnabled()
if client.RcClusterEnabled != enabled {
log.LogInfof("enableRemoteCacheCluster: %v -> %v", client.RcClusterEnabled, enabled)
client.RcClusterEnabled = enabled
client.RemoteCache.Status = client.IsRemoteCacheEnabled()
if client.RemoteCache.ClusterEnabled != enabled {
log.LogInfof("enableRemoteCacheCluster: %v -> %v", client.RemoteCache.ClusterEnabled, enabled)
client.RemoteCache.ClusterEnabled = enabled
}
}
func (client *ExtentClient) InitRemoteCache() (err error) {
cacheConfig := &CacheConfig{
Cluster: client.dataWrapper.ClusterName,
Volume: client.extentConfig.Volume,
Masters: client.extentConfig.Masters,
MW: client.metaWrapper,
readTimeoutSec: client.RcReadTimeoutSec,
}
if client.RemoteCache, err = NewRemoteCache(cacheConfig, client.enableRemoteCacheCluster); err != nil {
return
}
if !client.RemoteCache.ResetPathToBloom(client.RcPath) {
client.RcPath = ""
}
return
}
func (client *ExtentClient) UpdateRemoteCacheConfig(view *proto.SimpleVolView) {
if client.RcVolumeEnabled != view.RemoteCacheEnable {
log.LogInfof("RcVolumeEnabled: %v -> %v", client.RcVolumeEnabled, view.RemoteCacheEnable)
client.RcVolumeEnabled = view.RemoteCacheEnable
}
if client.RcOldStatus != client.IsRemoteCacheEnabled() {
log.LogInfof("enable: %v -> %v", client.RcOldStatus, client.IsRemoteCacheEnabled())
}
// RemoteCache may be nil if the first initialization failed, it will not be set nil anymore even if remote cache is disabled
if client.IsRemoteCacheEnabled() {
if !client.RcOldStatus || client.RemoteCache == nil {
log.LogInfof("initRemoteCache: enable(%v -> %v) RemoteCache isNil(%v)",
client.RcOldStatus, client.IsRemoteCacheEnabled(), client.RemoteCache == nil)
if err := client.InitRemoteCache(); err != nil {
log.LogErrorf("updateRemoteCacheConfig: initRemoteCache failed, err: %v", err)
return
}
}
} else if client.RcOldStatus && client.RemoteCache != nil {
rc := client.RemoteCache
client.RemoteCache = nil
rc.Stop()
log.LogInfo("stop RemoteCache")
}
if client.RcPath != view.RemoteCachePath {
oldPath := client.RcPath
client.RcPath = view.RemoteCachePath
if client.IsRemoteCacheEnabled() && client.RemoteCache != nil {
if !client.RemoteCache.ResetPathToBloom(view.RemoteCachePath) {
client.RcPath = ""
}
}
log.LogInfof("RcPath: %v -> %v, but(%v)", oldPath, view.RemoteCachePath, client.RcPath)
}
if client.RcAutoPrepare != view.RemoteCacheAutoPrepare {
log.LogInfof("RcAutoPrepare: %v -> %v", client.RcAutoPrepare, view.RemoteCacheAutoPrepare)
client.RcAutoPrepare = view.RemoteCacheAutoPrepare
}
if client.RcTTL != view.RemoteCacheTTL {
log.LogInfof("RcTTL: %d -> %d", client.RcTTL, view.RemoteCacheTTL)
client.RcTTL = view.RemoteCacheTTL
}
if client.RcReadTimeoutSec != view.RemoteCacheReadTimeoutSec {
log.LogInfof("RcReadTimeoutSec: %d -> %d", client.RcReadTimeoutSec, view.RemoteCacheReadTimeoutSec)
client.RcReadTimeoutSec = view.RemoteCacheReadTimeoutSec
}
client.RemoteCache.UpdateRemoteCacheConfig(client, view)
}
func (client *ExtentClient) GetVolumeName() string {
@ -1036,37 +962,13 @@ func (client *ExtentClient) UploadFlowInfo(clientInfo wrapper.SimpleClientInfo)
}
func (c *ExtentClient) RemoteCacheBloom() *bloom.BloomFilter {
if c.RemoteCache != nil {
return c.RemoteCache.GetRemoteCacheBloom()
}
return nil
return c.RemoteCache.GetRemoteCacheBloom()
}
func (c *ExtentClient) GetInodeBloomStatus(ino uint64) bool {
return c.RemoteCacheBloom().TestUint64(ino)
}
func (c *ExtentClient) DoPrepare() {
defer c.wg.Done()
workerWg := sync.WaitGroup{}
for range [5]struct{}{} {
workerWg.Add(1)
go func() {
defer workerWg.Done()
for {
select {
case <-c.stopCh:
return
case req := <-c.rcPrepareCh:
c.servePrepareRequest(req)
}
}
}()
}
workerWg.Wait()
}
func (c *ExtentClient) servePrepareRequest(prepareReq *PrepareRemoteCacheRequest) {
defer func() {
if err := recover(); err != nil {

125
sdk/data/stream/extent_flash_cache.go
View File

@ -73,7 +73,6 @@ type CacheConfig struct {
MW *meta.MetaWrapper
SameZoneWeight int
readTimeoutSec int64
}
type RemoteCache struct {
@ -89,41 +88,122 @@ type RemoteCache struct {
metaWrapper *meta.MetaWrapper
cacheBloom *bloom.BloomFilter
sameZoneWeight int
readTimeoutSec int
Status bool
ClusterEnabled bool
VolumeEnabled bool
Path string
AutoPrepare bool
TTL int64
ReadTimeoutSec int64
PrepareCh chan *PrepareRemoteCacheRequest
clusterEnable func(bool)
}
func NewRemoteCache(config *CacheConfig, clusterEnable func(bool)) (*RemoteCache, error) {
rc := new(RemoteCache)
rc.stopC = make(chan struct{})
rc.cluster = config.Cluster
rc.volname = config.Volume
rc.metaWrapper = config.MW
rc.flashGroups = btree.New(32)
rc.mc = master.NewMasterClient(config.Masters, false)
if config.SameZoneWeight <= 0 {
rc.sameZoneWeight = sameZoneWeight
} else {
rc.sameZoneWeight = config.SameZoneWeight
func (rc *RemoteCache) UpdateRemoteCacheConfig(client *ExtentClient, view *proto.SimpleVolView) {
if rc.VolumeEnabled != view.RemoteCacheEnable {
log.LogInfof("RcVolumeEnabled: %v -> %v", rc.VolumeEnabled, view.RemoteCacheEnable)
rc.VolumeEnabled = view.RemoteCacheEnable
}
rc.readTimeoutSec = int(config.readTimeoutSec)
if rc.Status != client.IsRemoteCacheEnabled() {
log.LogInfof("enable: %v -> %v", rc.Status, client.IsRemoteCacheEnabled())
}
// RemoteCache may be nil if the first initialization failed, it will not be set nil anymore even if remote cache is disabled
if client.IsRemoteCacheEnabled() {
if !rc.Status {
log.LogInfof("initRemoteCache: enable(%v -> %v)",
rc.Status, client.IsRemoteCacheEnabled())
if err := rc.Init(client); err != nil {
log.LogErrorf("updateRemoteCacheConfig: initRemoteCache failed, err: %v", err)
return
}
}
} else if client.RemoteCache.Status {
client.RemoteCache.Stop()
log.LogInfo("stop RemoteCache")
}
if rc.Path != view.RemoteCachePath {
oldPath := client.RemoteCache.Path
rc.Path = view.RemoteCachePath
if client.IsRemoteCacheEnabled() {
if !rc.ResetPathToBloom(view.RemoteCachePath) {
rc.Path = ""
}
}
log.LogInfof("RcPath: %v -> %v, but(%v)", oldPath, view.RemoteCachePath, rc.Path)
}
if rc.AutoPrepare != view.RemoteCacheAutoPrepare {
log.LogInfof("RcAutoPrepare: %v -> %v", rc.AutoPrepare, view.RemoteCacheAutoPrepare)
rc.AutoPrepare = view.RemoteCacheAutoPrepare
}
if rc.TTL != view.RemoteCacheTTL {
log.LogInfof("RcTTL: %d -> %d", rc.TTL, view.RemoteCacheTTL)
rc.TTL = view.RemoteCacheTTL
}
if rc.ReadTimeoutSec != view.RemoteCacheReadTimeoutSec {
log.LogInfof("RcReadTimeoutSec: %d -> %d", rc.ReadTimeoutSec, view.RemoteCacheReadTimeoutSec)
rc.ReadTimeoutSec = view.RemoteCacheReadTimeoutSec
}
}
func (rc *RemoteCache) DoRemoteCachePrepare(c *ExtentClient) {
defer c.wg.Done()
workerWg := sync.WaitGroup{}
for range [5]struct{}{} {
workerWg.Add(1)
go func() {
defer workerWg.Done()
for {
select {
case <-c.stopCh:
return
case req := <-c.RemoteCache.PrepareCh:
c.servePrepareRequest(req)
}
}
}()
}
workerWg.Wait()
}
func (rc *RemoteCache) Init(client *ExtentClient) (err error) {
log.LogDebugf("RemoteCache: Init")
fmt.Println("RemoteCache: Init")
rc.stopC = make(chan struct{})
rc.cluster = client.dataWrapper.ClusterName
rc.volname = client.extentConfig.Volume
rc.metaWrapper = client.metaWrapper
rc.flashGroups = btree.New(32)
rc.mc = master.NewMasterClient(client.extentConfig.Masters, false)
rc.readTimeoutSec = int(client.RemoteCache.ReadTimeoutSec)
if rc.readTimeoutSec <= 0 {
rc.readTimeoutSec = _connReadTimeoutSec
}
rc.clusterEnable = clusterEnable
err := rc.updateFlashGroups()
rc.clusterEnable = client.enableRemoteCacheCluster
err = rc.updateFlashGroups()
if err != nil {
return nil, err
log.LogDebugf("RemoteCache: Init err %v", err)
return
}
rc.conns = util.NewConnectPoolWithTimeoutAndCap(0, 10, _connIdelTimeout, int64(rc.readTimeoutSec))
rc.cacheBloom = bloom.New(BloomBits, BloomHashNum)
rc.wg.Add(1)
go rc.refresh()
return rc, nil
if !rc.ResetPathToBloom(client.RemoteCache.Path) {
rc.Path = ""
}
rc.PrepareCh = make(chan *PrepareRemoteCacheRequest, 1024)
go rc.DoRemoteCachePrepare(client)
log.LogDebugf("Init: NewRemoteCache sucess")
return
}
func (rc *RemoteCache) Read(ctx context.Context, fg *FlashGroup, inode uint64, req *CacheReadRequest) (read int, err error) {
@ -144,6 +224,7 @@ func (rc *RemoteCache) Read(ctx context.Context, fg *FlashGroup, inode uint64, r
}
defer func() {
if err != nil && (os.IsTimeout(err) || strings.Contains(err.Error(), syscall.ECONNREFUSED.Error())) {
log.LogErrorf("FlashGroup read: err %v", err)
moved = fg.moveToUnknownRank(addr)
}
}()
@ -208,6 +289,7 @@ func (rc *RemoteCache) Prepare(ctx context.Context, fg *FlashGroup, inode uint64
}
defer func() {
if err != nil && (os.IsTimeout(err) || strings.Contains(err.Error(), syscall.ECONNREFUSED.Error())) {
log.LogErrorf("FlashGroup Prepare: err %v", err)
moved = fg.moveToUnknownRank(addr)
}
}()
@ -247,6 +329,7 @@ func (rc *RemoteCache) Stop() {
}
func (rc *RemoteCache) GetRemoteCacheBloom() *bloom.BloomFilter {
log.LogDebugf("GetRemoteCacheBloom. cacheBloom %v", rc.cacheBloom)
return rc.cacheBloom
}
@ -343,7 +426,7 @@ func (rc *RemoteCache) updateFlashGroups() (err error) {
log.LogWarnf("updateFlashGroups: err(%v)", err)
return
}
log.LogDebugf("updateFlashGroups. get flashGroupView [%v]", fgv)
rc.clusterEnable(fgv.Enable)
if !fgv.Enable {
rc.flashGroups = newFlashGroups
@ -377,7 +460,7 @@ func (rc *RemoteCache) updateFlashGroups() (err error) {
}
func (rc *RemoteCache) ClassifyHostsByAvgDelay(fgID uint64, hosts []string) (sortedHosts map[ZoneRankType][]string) {
sortedHosts = make(map[ZoneRankType][]string, 0)
sortedHosts = make(map[ZoneRankType][]string)
for _, host := range hosts {
avgTime := time.Duration(0)

36
sdk/data/stream/stream_remote_cache.go
View File

@ -37,16 +37,17 @@ func (pr *PrepareRemoteCacheRequest) String() string {
}
func (s *Streamer) enableRemoteCache() bool {
return s.client.IsRemoteCacheEnabled() && s.client.RemoteCache != nil && s.bloomStatus
log.LogDebugf("Streamer inode %v parent %v enableRemoteCache bloomStatus, s.bloomStatus %v", s.inode, s.parentInode, s.bloomStatus)
return s.client.IsRemoteCacheEnabled() // && s.bloomStatus
}
func (s *Streamer) enableRemoteCacheAutoPrepare() bool {
return s.enableRemoteCache() && s.client.RcAutoPrepare
return s.enableRemoteCache() && s.client.RemoteCache.AutoPrepare
}
func (s *Streamer) sendToPrepareRomoteCacheChan(req *PrepareRemoteCacheRequest) {
select {
case s.client.rcPrepareCh <- req:
case s.client.RemoteCache.PrepareCh <- req:
default:
log.LogWarnf("sendToPrepareRomoteCacheChan: chan is full, discard req(%v)", req)
}
@ -114,30 +115,19 @@ func (s *Streamer) getFlashGroup(fixedFileOffset uint64) *FlashGroup {
func (s *Streamer) getDataSource(start, size, fixedFileOffset uint64, isRead bool) ([]*proto.DataSource, error) {
sources := make([]*proto.DataSource, 0)
var revisedRequests []*ExtentRequest
var data []byte
log.LogDebugf("getDataSource. start %v size %v fixedFileOffset %v isRead %v", start, size, fixedFileOffset, isRead)
eReqs := s.extents.PrepareReadRequests(int(fixedFileOffset), proto.CACHE_BLOCK_SIZE, nil)
for _, req := range eReqs {
if req.ExtentKey == nil {
for _, eReq := range eReqs {
log.LogDebugf("getDataSource. eReq %v", eReq)
if eReq.ExtentKey == nil {
continue
}
if req.ExtentKey.PartitionId == 0 || req.ExtentKey.ExtentId == 0 {
s.writeLock.Lock()
if err := s.IssueFlushRequest(); err != nil {
s.writeLock.Unlock()
if eReq.ExtentKey.PartitionId == 0 {
if eReq.ExtentKey.FileOffset+uint64(eReq.ExtentKey.Size) > start && eReq.ExtentKey.FileOffset < start+size {
err := fmt.Errorf("temporary ek, isRead[%v] start(%v) size(%v) eReq(%v) fixedOff(%v)", isRead, start, size, eReq, fixedFileOffset)
log.LogErrorf("getDataSource failed: err(%v)", err)
return nil, err
}
revisedRequests = s.extents.PrepareReadRequests(int(fixedFileOffset), proto.CACHE_BLOCK_SIZE, data)
s.writeLock.Unlock()
break
}
}
if revisedRequests != nil {
eReqs = revisedRequests
}
for _, eReq := range eReqs {
if eReq.ExtentKey == nil {
continue
}
@ -178,7 +168,7 @@ func (s *Streamer) prepareCacheRequests(offset, size uint64, data []byte) ([]*Ca
Volume: s.client.dataWrapper.VolName,
Inode: s.inode,
FixedFileOffset: fixedOff,
TTL: s.client.RcTTL,
TTL: s.client.RemoteCache.TTL,
Sources: sources,
Version: proto.ComputeSourcesVersion(sources),
}

1
sdk/meta/api.go
View File

@ -368,6 +368,7 @@ create_dentry:
func (mw *MetaWrapper) Lookup_ll(parentID uint64, name string) (inode uint64, mode uint32, err error) {
defer func() {
log.LogDebugf("Lookup_ll parent id %v, name %v ,inode %v", parentID, name, inode)
if err == nil && mw.RemoteCacheBloom != nil {
cacheBloom := mw.RemoteCacheBloom()
if cacheBloom.TestUint64(parentID) {