feat(client): add TCP keep-alive time configuration for connection pools. #1000402826

- Implement TcpAliveTime option in mount options
- Set TCP keep-alive time for stream and ahead read connection pools
- Update connection pool to accept timeout and minimum capacity parameters

Signed-off-by: Victor1319 <zengxuewei@oppo.com>
(cherry picked from commit d96c5db9a3)

client/fuse.go

@@ -401,6 +401,12 @@ func main() {
 		stream.SetReqChansize(int(opt.ReqChanCnt))
 	}
 
+	if opt.TcpAliveTime > 0 {
+		stream.StreamConnPool.SetPoolArgs(int64(opt.TcpAliveTime), 0)
+		stream.AheadReadConnPool.SetPoolArgs(int64(opt.TcpAliveTime), 0)
+		stream.StreamWriteConnPool.SetPoolArgs(int64(opt.TcpAliveTime), 0)
+	}
+
 	level := parseLogLevel(opt.Loglvl)
 	_, err = log.InitLog(opt.Logpath, opt.Volname, level, nil, log.DefaultLogLeftSpaceLimitRatio)
 	if err != nil {
@@ -1005,6 +1011,7 @@ func parseMountOption(cfg *config.Config) (*proto.MountOptions, error) {
 	opt.EnableAudit = GlobalMountOptions[proto.EnableAudit].GetBool()
 	opt.RequestTimeout = GlobalMountOptions[proto.RequestTimeout].GetInt64()
 	opt.ClientOpTimeOut = GlobalMountOptions[proto.ClientOpTimeOut].GetInt64()
+	opt.TcpAliveTime = GlobalMountOptions[proto.TcpAliveTime].GetInt64()
 	opt.FileSystemName = GlobalMountOptions[proto.FileSystemName].GetString()
 	opt.DisableMountSubtype = GlobalMountOptions[proto.DisableMountSubtype].GetBool()
 	opt.StreamRetryTimeout = int(GlobalMountOptions[proto.StreamRetryTimeOut].GetInt64())

proto/mount_options.go

@@ -93,6 +93,7 @@ const (
 	ReadDirLimit
 	MaxWarmUpConcurrency
 	StopWarmMeta
+	TcpAliveTime
 	MaxMountOption
 )
 
@@ -180,7 +181,7 @@ func InitMountOptions(opts []MountOption) {
 	opts[EnableAudit] = MountOption{"enableAudit", "enable client audit logging", "", true}
 	opts[RequestTimeout] = MountOption{"requestTimeout", "The Request Expiration Time", "", int64(0)}
 	opts[ClientOpTimeOut] = MountOption{"clientOpTimeOut", "client op time out in seconds", "", int64(300)}
-
+	opts[TcpAliveTime] = MountOption{"tcpAliveTime", "tcp alive time in seconds", "", int64(0)}
 	opts[FileSystemName] = MountOption{"fileSystemName", "The explicit name of the filesystem", "", ""}
 	opts[SnapshotReadVerSeq] = MountOption{"snapshotReadSeq", "Snapshot read seq", "", int64(0)} // default false
 	opts[DisableMountSubtype] = MountOption{"disableMountSubtype", "Disable Mount Subtype", "", false}
@@ -355,6 +356,7 @@ type MountOptions struct {
 	EnableAudit             bool
 	RequestTimeout          int64
 	ClientOpTimeOut         int64
+	TcpAliveTime            int64
 	FileSystemName          string
 	// TrashInterval                       int64
 	TrashDeleteExpiredDirGoroutineLimit int64

sdk/data/stream/stream_aheadread.go

@@ -388,7 +388,8 @@ func (arw *AheadReadWindow) addNextTask(offset int, stTime time.Time, reqID stri
 }
 
 func (arw *AheadReadWindow) doMultiAheadRead(offset int, req *ExtentRequest, dp *wrapper.DataPartition,
-	startTime time.Time, reqID string, storageClass uint32, winCnt int) {
+	startTime time.Time, reqID string, storageClass uint32, winCnt int,
+) {
 	curReq := &ExtentRequest{
 		FileOffset: req.FileOffset,
 		Size:       req.Size,
@@ -441,7 +442,8 @@ func (arw *AheadReadWindow) doMultiAheadRead(offset int, req *ExtentRequest, dp
 }
 
 func (arw *AheadReadWindow) getAheadReadTask(dp *wrapper.DataPartition, req *ExtentRequest, id int, size int,
-	storageClass uint32) *AheadReadTask {
+	storageClass uint32,
+) *AheadReadTask {
 	cacheOffset := id * int(arw.streamer.aheadReadBlockSize)
 	// tiny need to add ExtentOffset
 	if req.ExtentKey.ExtentId <= 64 {
@@ -597,11 +599,11 @@ func (s *Streamer) aheadRead(req *ExtentRequest, storageClass uint32) (readSize
 func sendToNode(host string, p *Packet, getReply GetReplyFunc) (err error) {
 	var conn *net.TCPConn
 	start := time.Now()
-	if conn, err = StreamConnPool.GetConnect(host); err != nil {
+	if conn, err = AheadReadConnPool.GetConnect(host); err != nil {
 		return
 	}
 	defer func() {
-		StreamConnPool.PutConnect(conn, err != nil)
+		AheadReadConnPool.PutConnect(conn, err != nil)
 		log.LogDebugf("sendToNode connect local(%v) remote(%v) cost(%v) err(%v)", conn.LocalAddr().String(),
 			conn.RemoteAddr().String(), time.Since(start).String(), err)
 	}()

sdk/data/stream/stream_conn.go

@@ -58,6 +58,7 @@ type StreamConn struct {
 
 var (
 	StreamConnPool      = util.NewConnectPool()
+	AheadReadConnPool   = util.NewConnectPool()
 	StreamWriteConnPool = util.NewConnectPool()
 )
 

util/conn_pool.go

@@ -87,6 +87,11 @@ func DailTimeOut(target string, timeout time.Duration) (c *net.TCPConn, err erro
 	return
 }
 
+func (cp *ConnectPool) SetPoolArgs(timeout int64, minCap int) {
+	cp.timeout = int64(time.Duration(timeout) * time.Second)
+	cp.mincap = minCap
+}
+
 func (cp *ConnectPool) GetConnect(targetAddr string) (c *net.TCPConn, err error) {
 	cp.RLock()
 	pool, ok := cp.pools[targetAddr]