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(flash): pull handleReadTimeout when flashnode start

Browse Source 
with: #1000347719

Signed-off-by: clinx <chenlin1@oppo.com>
This commit is contained in:
clinx 2025-09-18 17:31:14 +08:00
parent 53edbd9d19
commit dfa3e22413
6 changed files with 15 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
master/api_service.go
View File

@ -1178,6 +1178,7 @@ func (m *Server) getIPAddr(w http.ResponseWriter, r *http.Request) {
DataNodeAutoRepairLimitRate: autoRepairRate,
DpMaxRepairErrCnt: dpMaxRepairErrCnt,
DirChildrenNumLimit: dirChildrenNumLimit,
FlashReadTimeout: m.cluster.cfg.flashNodeHandleReadTimeout,
// Ip: strings.Split(r.RemoteAddr, ":")[0],
Ip: iputil.RealIP(r),
EbsAddr: m.bStoreAddr,

1
proto/admin_proto.go
View File

@ -633,6 +633,7 @@ type ClusterInfo struct {
DataNodeAutoRepairLimitRate uint64
DpMaxRepairErrCnt uint64
DirChildrenNumLimit uint32
FlashReadTimeout int
EbsAddr string
ServicePath string
ClusterUuid string

5
remotecache/flashgroupmanager/api_service.go
View File

@ -93,8 +93,9 @@ func (m *FlashGroupManager) getIPAddr(w http.ResponseWriter, r *http.Request) {
doStatAndMetric(proto.AdminGetIP, metric, nil, nil)
}()
cInfo := &proto.ClusterInfo{
Cluster: m.clusterName,
Ip: iputil.RealIP(r),
Cluster: m.clusterName,
FlashReadTimeout: int(m.cluster.cfg.RemoteCacheReadTimeout),
Ip: iputil.RealIP(r),
}
sendOkReply(w, r, newSuccessHTTPReply(cInfo))
}

6
remotecache/flashnode/flashnode.go
View File

@ -544,7 +544,7 @@ func (f *FlashNode) startCacheEngine() (err error) {
log.LogErrorf("startCacheEngine failed:%v", err)
return
}
f.SetTimeout(proto.DefaultRemoteCacheHandleReadTimeout, proto.DefaultRemoteCacheExtentReadTimeout)
f.cacheEngine.SetReadDataNodeTimeout(proto.DefaultRemoteCacheExtentReadTimeout)
f.cacheEngine.StartCachePrepareWorkers(f.limitWrite, f.prepareLoadRoutineNum)
return f.cacheEngine.Start()
}
@ -578,6 +578,10 @@ func (f *FlashNode) register() error {
break
}
f.clusterID = ci.Cluster
if ci.FlashReadTimeout != 0 {
log.LogInfof("FlashNode load handleReadTimeout from %d(ms) to %d(ms)", f.handleReadTimeout, ci.FlashReadTimeout)
f.handleReadTimeout = ci.FlashReadTimeout
}
f.localAddr = fmt.Sprintf("%s:%v", localIP, f.listen)
nodeID, err := f.mc.NodeAPI().AddFlashNode(f.localAddr, f.zoneName, "")

8
sdk/remotecache/client.go
View File

@ -53,7 +53,6 @@ const (
DefaultFirstRetryPercent = 80
DefaultSubsequentRetryPercent = 10
DefaultSecondRetryPercent = 20
DefaultFlowLimit = 5 * 1024 * 1024 * 1024 // 5GB default
)
type AddressPingStats struct {
@ -201,10 +200,11 @@ func NewRemoteCacheClient(config *ClientConfig) (rc *RemoteCacheClient, err erro
} else {
rc.connWorkers = config.ConnWorkers
}
if config.FlowLimit == 0 {
config.FlowLimit = DefaultFlowLimit
if config.FlowLimit > 0 {
rc.flowLimiter = rate.NewLimiter(rate.Limit(config.FlowLimit), int(config.FlowLimit/2))
} else {
rc.flowLimiter = rate.NewLimiter(rate.Inf, 0)
}
rc.flowLimiter = rate.NewLimiter(rate.Limit(config.FlowLimit), int(config.FlowLimit/2))
rc.connPutChan = make(chan *ConnPutTask, DefaultConnPutChanSize)
rc.readPool = util.NewGTaskPool(DefaultReadPoolSize)
rc.readPool.SetMaxDeltaRunning(512)

2
tool/remotecache-benchmark/main.go
View File

@ -219,7 +219,7 @@ func main() {
disableBatch := flag.Bool("disable-batch", false, "Disable batch operations")
activateTime := flag.Int64("activate-time", 200, "Activate time in microseconds")
flashConnWorkers := flag.Int("flash-conn-workers", 64, "Number of flash connection workers")
flowLimit := flag.Int64("flow-limit", 5*1024*1024*1024, "Flow limit in bytes per second (default: 5GB)")
flowLimit := flag.Int64("flow-limit", 0, "Flow limit in bytes per second (default: 0 limiter is disable)")
flag.Parse()
fmt.Printf("Parsed command-line arguments:\n")