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): check the type of error returned by flashnode

Browse Source 
close:#23047275

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-02-18 14:14:21 +08:00 committed by zhumingze1108
parent b4a9c85ae6
commit 596b79c286
4 changed files with 22 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
flashnode/cachengine/engine.go
View File

@ -64,8 +64,9 @@ type Disk struct {
}
type cachePrepareTask struct {
request *proto.CacheRequest
reqID int64
request *proto.CacheRequest
reqID int64
clientIP string
}
type cacheLoadTask struct {
@ -639,6 +640,10 @@ func (c *CacheEngine) startCachePrepareWorkers() {
return
case task := <-c.cachePrepareTaskCh:
r := task.request
if _, err := c.CreateBlock(r, task.clientIP, true); err != nil {
log.LogWarnf("action[startCachePrepareWorkers] ReqID(%d) create block failed, err:%v", task.reqID, err)
continue
}
block, err := c.PeekCacheBlock(GenCacheBlockKey(r.Volume, r.Inode, r.FixedFileOffset, r.Version))
if err != nil {
log.LogWarnf("action[startCachePrepareWorkers] ReqID(%d) cache block not found, err:%v", task.reqID, err)
@ -652,13 +657,10 @@ func (c *CacheEngine) startCachePrepareWorkers() {
}
func (c *CacheEngine) PrepareCache(reqID int64, req *proto.CacheRequest, clientIP string) (err error) {
if _, err = c.CreateBlock(req, clientIP, true); err != nil {
return
}
select {
case c.cachePrepareTaskCh <- cachePrepareTask{reqID: reqID, request: req}:
case c.cachePrepareTaskCh <- cachePrepareTask{reqID: reqID, request: req, clientIP: clientIP}:
default:
log.LogWarn("action[PrepareCache] cachePrepareTaskCh has been full")
log.LogDebugf("action[PrepareCache] cachePrepareTaskCh has been full")
}
return
}

2
flashnode/flashnode_op_test.go
View File

@ -91,7 +91,7 @@ func testTCPCachePrepare(t *testing.T) {
p.MarshalDataPb(prepare)
require.NoError(t, p.WriteToConn(conn))
require.NoError(t, r.ReadFromConn(conn, 3))
require.Equal(t, proto.OpErr, r.ResultCode) // No DataSource
require.Equal(t, proto.OpOk, r.ResultCode) // No DataSource
prepare.CacheRequest.Sources = []*proto.DataSource{{
FileOffset: 0,

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

@ -20,11 +20,9 @@ import (
"fmt"
"hash/crc32"
"net"
"os"
"runtime/debug"
"strings"
"sync"
"syscall"
"time"
"github.com/cubefs/cubefs/proto"
@ -264,7 +262,7 @@ func (rc *RemoteCache) Read(ctx context.Context, fg *FlashGroup, inode uint64, r
}
if conn, err = rc.conns.GetConnect(addr); err != nil {
log.LogWarnf("FlashGroup read: get connection failed, addr(%v) reqPacket(%v) err(%v) remoteCacheFollowerRead(%v)", addr, req, err, rc.remoteCacheFollowerRead)
moved = fg.moveToUnknownRank(addr)
moved = fg.moveToUnknownRank(addr, err)
if rc.remoteCacheFollowerRead {
log.LogInfof("Retrying due to GetConnect of addr(%v) failure err(%v)", addr, err)
continue
@ -275,7 +273,7 @@ func (rc *RemoteCache) Read(ctx context.Context, fg *FlashGroup, inode uint64, r
if err = reqPacket.WriteToConn(conn); err != nil {
log.LogWarnf("FlashGroup Read: failed to write to addr(%v) err(%v) remoteCacheFollowerRead(%v)", addr, err, rc.remoteCacheFollowerRead)
rc.conns.PutConnect(conn, err != nil)
moved = fg.moveToUnknownRank(addr)
moved = fg.moveToUnknownRank(addr, err)
if rc.remoteCacheFollowerRead {
log.LogInfof("Retrying due to write to addr(%v) failure err(%v)", addr, err)
continue
@ -285,7 +283,7 @@ func (rc *RemoteCache) Read(ctx context.Context, fg *FlashGroup, inode uint64, r
if read, err = rc.getReadReply(conn, reqPacket, req); err != nil {
log.LogWarnf("FlashGroup Read: getReadReply from addr(%v) err(%v) remoteCacheFollowerRead(%v)", addr, err, rc.remoteCacheFollowerRead)
rc.conns.PutConnect(conn, err != nil)
moved = fg.moveToUnknownRank(addr)
moved = fg.moveToUnknownRank(addr, err)
if rc.remoteCacheFollowerRead {
log.LogInfof("Retrying due to getReadReply from addr(%v) failure err(%v)", addr, err)
continue
@ -341,9 +339,8 @@ func (rc *RemoteCache) Prepare(ctx context.Context, fg *FlashGroup, inode uint64
return
}
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)
if err != nil {
moved = fg.moveToUnknownRank(addr, err)
}
}()
if conn, err = rc.conns.GetConnect(addr); err != nil {

8
sdk/data/stream/flash_group.go
View File

@ -16,8 +16,11 @@ package stream
import (
"fmt"
"os"
"strings"
"sync"
"sync/atomic"
"syscall"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util/btree"
@ -74,7 +77,10 @@ func (fg *FlashGroup) getFlashHost() (host string) {
return
}
func (fg *FlashGroup) moveToUnknownRank(addr string) bool {
func (fg *FlashGroup) moveToUnknownRank(addr string, err error) bool {
if err != nil && (os.IsTimeout(err) || strings.Contains(err.Error(), syscall.ECONNREFUSED.Error())) {
return false
}
fg.hostLock.Lock()
defer fg.hostLock.Unlock()