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

fix(client): set DefaultRemoteCacheSameRegionTimeout to 20 ms

Browse Source 
with: #1000303671

Signed-off-by: clinx <chenlin1@oppo.com>
This commit is contained in:
clinx 2025-08-26 10:37:19 +08:00 committed by Victor1319
parent bf3370cc7b
commit cbb489d2ea
7 changed files with 19 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
proto/distributed_cache.go
View File

@ -12,7 +12,7 @@ import (
const (
PageSize = 4 * 1024
CACHE_BLOCK_PACKET_SIZE = 32 * PageSize
CACHE_BLOCK_PACKET_SIZE = 128 * PageSize
CACHE_BLOCK_SIZE = 1 << 20
CACHE_OBJECT_BLOCK_SIZE = 4 * 1024 * 1024
ReadCacheTimeout = 1 // second

2
proto/packet.go
View File

@ -453,7 +453,7 @@ const (
DefaultRemoteCacheHandleReadTimeout = 100 // ms
DefaultRemoteCacheExtentReadTimeout = 3000
DefaultRemoteCacheSameZoneTimeout = 400 // microsecond
DefaultRemoteCacheSameRegionTimeout = 2 // ms
DefaultRemoteCacheSameRegionTimeout = 100 // ms
)
// multi version operation

2
remotecache/flashgroupmanager/admin_task_manager.go
View File

@ -41,10 +41,8 @@ func newAdminTaskManager(targetAddr, clusterID string) (sender *AdminTaskManager
clusterID: clusterID,
TaskMap: make(map[string]*proto.AdminTask),
exitCh: make(chan struct{}, 1),
connPool: util.NewConnectPoolWithTimeout(idleConnTimeout, connectTimeout, false),
}
go sender.process()
return
}

8
remotecache/flashgroupmanager/admin_task_manager_test.go
View File

@ -8,6 +8,7 @@ import (
"time"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/stretchr/testify/assert"
)
@ -148,6 +149,7 @@ func TestAdminTaskManager_syncSendAdminTask(t *testing.T) {
// Create task manager instance
manager := newAdminTaskManager(targetAddr, "test_cluster")
manager.connPool = util.NewConnectPoolWithTimeoutAndCap(0, 5, idleConnTimeout, connectTimeout, false)
defer func() {
close(manager.exitCh)
time.Sleep(100 * time.Millisecond) // Wait for goroutine to exit
@ -196,6 +198,7 @@ func TestAdminTaskManager_syncSendAdminTask_WriteError(t *testing.T) {
}()
manager := newAdminTaskManager(targetAddr, "test_cluster")
manager.connPool = util.NewConnectPoolWithTimeoutAndCap(0, 5, idleConnTimeout, connectTimeout, false)
defer func() {
close(manager.exitCh)
time.Sleep(100 * time.Millisecond)
@ -237,6 +240,7 @@ func TestAdminTaskManager_syncSendAdminTask_ReadError(t *testing.T) {
}()
manager := newAdminTaskManager(targetAddr, "test_cluster")
manager.connPool = util.NewConnectPoolWithTimeoutAndCap(0, 5, idleConnTimeout, connectTimeout, false)
defer func() {
close(manager.exitCh)
time.Sleep(100 * time.Millisecond)
@ -279,6 +283,7 @@ func TestAdminTaskManager_syncSendAdminTask_Timeout(t *testing.T) {
}()
manager := newAdminTaskManager(targetAddr, "test_cluster")
manager.connPool = util.NewConnectPoolWithTimeoutAndCap(0, 5, idleConnTimeout, connectTimeout, false)
defer func() {
close(manager.exitCh)
time.Sleep(100 * time.Millisecond)
@ -297,6 +302,7 @@ func TestAdminTaskManager_syncSendAdminTask_InvalidTask(t *testing.T) {
// Test invalid task scenario: pass nil task
// Verify method's ability to handle abnormal input
manager := newAdminTaskManager("127.0.0.1:8080", "test_cluster")
manager.connPool = util.NewConnectPoolWithTimeoutAndCap(0, 5, idleConnTimeout, connectTimeout, false)
defer func() {
close(manager.exitCh)
time.Sleep(100 * time.Millisecond)
@ -352,6 +358,7 @@ func TestAdminTaskManager_syncSendAdminTask_EmptyTask(t *testing.T) {
}()
manager := newAdminTaskManager(targetAddr, "test_cluster")
manager.connPool = util.NewConnectPoolWithTimeoutAndCap(0, 5, idleConnTimeout, connectTimeout, false)
defer func() {
close(manager.exitCh)
time.Sleep(100 * time.Millisecond)
@ -418,6 +425,7 @@ func TestAdminTaskManager_syncSendAdminTask_ComplexRequest(t *testing.T) {
}()
manager := newAdminTaskManager(targetAddr, "test_cluster")
manager.connPool = util.NewConnectPoolWithTimeoutAndCap(0, 5, idleConnTimeout, connectTimeout, false)
defer func() {
close(manager.exitCh)
time.Sleep(100 * time.Millisecond)

2
remotecache/flashgroupmanager/flash_node.go
View File

@ -9,6 +9,7 @@ import (
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/httpclient"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/log"
)
@ -57,6 +58,7 @@ func NewFlashNode(addr, zoneName, clusterID, version string, isEnable bool) *Fla
node.IsEnable = isEnable
node.ReportTime = time.Now()
node.TaskManager = newAdminTaskManager(addr, clusterID)
node.TaskManager.connPool = util.NewConnectPoolWithTimeout(idleConnTimeout, connectTimeout, false)
return node
}

7
sdk/remotecache/client.go
View File

@ -1099,7 +1099,7 @@ func (reader *RemoteCacheReader) getReadObjectReply(p *proto.Packet) (length int
}
err = fmt.Errorf(string(p.Data))
if !proto.IsFlashNodeLimitError(err) {
log.LogWarnf("getReadObjectReply: ResultCode NOK, err(%v) ResultCode(%v)", err, p.ResultCode)
log.LogWarnf("getReadObjectReply:reqID(%v) ResultCode NOK, err(%v) ResultCode(%v)", reader.reqID, err, p.ResultCode)
}
return
}
@ -1157,6 +1157,9 @@ func (reader *RemoteCacheReader) read(p []byte) (n int, err error) {
alreadyReadLen := atomic.LoadInt64(&reader.alreadyReadLen)
if alreadyReadLen >= reader.needReadLen {
if log.EnableDebug() {
log.LogDebugf("RemoteCacheReader:reqID(%v) alreadyReadLen(%v) needReadLen(%v) get EOF", reader.reqID, alreadyReadLen, reader.needReadLen)
}
return 0, io.EOF
}
reader.conn.SetReadDeadline(time.Now().Add(time.Second * proto.ReadDeadlineTime))
@ -1229,7 +1232,7 @@ func (reader *RemoteCacheReader) Read(p []byte) (n int, err error) {
reader.size, err = reader.read(reader.buffer)
if reader.size == 0 {
if log.EnableDebug() {
log.LogDebugf("RemoteCacheReader:reqID(%v), from connection size 0", reader.reqID)
log.LogDebugf("RemoteCacheReader:reqID(%v), totalRead(%v) from connection size 0 err(%v)", reader.reqID, totalRead, err)
}
if err != nil {
return totalRead, err

5
util/conn_pool.go
View File

@ -92,11 +92,10 @@ func (cp *ConnectPool) GetConnect(targetAddr string) (c *net.TCPConn, err error)
pool, ok := cp.pools[targetAddr]
cp.RUnlock()
if !ok {
newPool := NewPool(cp.mincap, cp.maxcap, cp.timeout, cp.connectTimeout, targetAddr, cp.useMilliSecond)
cp.Lock()
pool, ok = cp.pools[targetAddr]
if !ok {
// pool = NewPool(cp.mincap, cp.maxcap, cp.timeout, cp.connectTimeout, targetAddr)
newPool := NewPool(cp.mincap, cp.maxcap, cp.timeout, cp.connectTimeout, targetAddr, cp.useMilliSecond)
pool = newPool
cp.pools[targetAddr] = pool
}
@ -222,7 +221,7 @@ func NewPool(min, max int, timeout, connectTimeout int64, target string, useMill
p.objects = make(chan *Object, max)
p.timeout = timeout
p.connectTimeout = connectTimeout
p.initAllConnect()
go p.initAllConnect()
return p
}