Skip to content

fix: reconnect attempts after successful pool fill #140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AlexStocks merged 4 commits into from
Aug 13, 2026
Merged

fix: reconnect attempts after successful pool fill #140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
cd64a48
fix: reset reconnect attempts after successful pool fill
222twotwotwo Aug 7, 2026
22461bf
test: cover reconnect attempts reset after successful connection
Aug 9, 2026
80d64f0
Merge branch 'apache:master' into fix-disable-reconnect-after-enough-…
222twotwotwo Aug 12, 2026
66aedd3
fix: bound reconnect attempts after session init failure
Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
266 changes: 134 additions & 132 deletions client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import (
"crypto/x509"
"encoding/pem"
"fmt"
"math"
"net"
"os"
"strings"
Expand Down Expand Up @@ -153,29 +152,32 @@ func (c *client) dialTCP() Session {
conn net.Conn
)

for {
if c.IsClosed() {
return nil
}
if c.sslEnabled {
if sslConfig, buildTlsConfErr := c.tlsConfigBuilder.BuildTlsConfig(); buildTlsConfErr == nil && sslConfig != nil {
d := &net.Dialer{Timeout: connectTimeout}
conn, err = tls.DialWithDialer(d, "tcp", c.addr, sslConfig)
}
if c.IsClosed() {
return nil
}
if c.sslEnabled {
sslConfig, buildTlsConfErr := c.tlsConfigBuilder.BuildTlsConfig()
if buildTlsConfErr != nil {
err = buildTlsConfErr
} else if sslConfig == nil {
err = fmt.Errorf("tlsConfigBuilder returned nil config without error")
} else {
conn, err = net.DialTimeout("tcp", c.addr, connectTimeout)
}
if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
_ = conn.Close()
err = errSelfConnect
d := &net.Dialer{Timeout: connectTimeout}
conn, err = tls.DialWithDialer(d, "tcp", c.addr, sslConfig)
}
if err == nil {
return newTCPSession(conn, c)
}

log.Infof("net.DialTimeout(addr:%s, timeout:%v) = error:%+v", c.addr, connectTimeout, perrors.WithStack(err))
<-gxtime.After(connectInterval)
} else {
conn, err = net.DialTimeout("tcp", c.addr, connectTimeout)
}
if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
_ = conn.Close()
err = errSelfConnect
}
if err == nil {
return newTCPSession(conn, c)
}

log.Infof("net.DialTimeout(addr:%s, timeout:%v) = error:%+v", c.addr, connectTimeout, perrors.WithStack(err))
return nil
}

func (c *client) dialUDP() Session {
Expand All @@ -194,46 +196,41 @@ func (c *client) dialUDP() Session {
buf = *bufp
localAddr = &net.UDPAddr{IP: net.IPv4zero, Port: 0}
peerAddr, _ = net.ResolveUDPAddr("udp", c.addr)
for {
if c.IsClosed() {
return nil
}
conn, err = net.DialUDP("udp", localAddr, peerAddr)
if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
_ = conn.Close()
err = errSelfConnect
}
if err != nil {
log.Warnf("net.DialTimeout(addr:%s, timeout:%v) = error:%+v", c.addr, perrors.WithStack(err))
<-gxtime.After(connectInterval)
continue
}
if c.IsClosed() {
return nil
}
conn, err = net.DialUDP("udp", localAddr, peerAddr)
if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
_ = conn.Close()
err = errSelfConnect
}
if err != nil {
log.Warnf("net.DialUDP(addr:%s) = error:%+v", c.addr, perrors.WithStack(err))
return nil
}

// check connection alive by write/read action
if err := conn.SetWriteDeadline(time.Now().Add(1e9)); err != nil {
log.Warnf("failed to set write deadline: %+v", err)
}
if length, err = conn.Write(connectPingPackage[:]); err != nil {
_ = conn.Close()
log.Warnf("conn.Write(%s) = {length:%d, err:%+v}", string(connectPingPackage), length, perrors.WithStack(err))
<-gxtime.After(connectInterval)
continue
}
if err := conn.SetReadDeadline(time.Now().Add(1e9)); err != nil {
log.Warnf("failed to set read deadline: %+v", err)
}
length, err = conn.Read(buf)
if netErr, ok := perrors.Cause(err).(net.Error); ok && netErr.Timeout() {
err = nil
}
if err != nil {
log.Infof("conn{%#v}.Read() = {length:%d, err:%+v}", conn, length, perrors.WithStack(err))
_ = conn.Close()
<-gxtime.After(connectInterval)
continue
}
return newUDPSession(conn, c)
// check connection alive by write/read action
if err := conn.SetWriteDeadline(time.Now().Add(1e9)); err != nil {
log.Warnf("failed to set write deadline: %+v", err)
}
if length, err = conn.Write(connectPingPackage[:]); err != nil {
_ = conn.Close()
log.Warnf("conn.Write(%s) = {length:%d, err:%+v}", string(connectPingPackage), length, perrors.WithStack(err))
return nil
}
if err := conn.SetReadDeadline(time.Now().Add(1e9)); err != nil {
log.Warnf("failed to set read deadline: %+v", err)
}
length, err = conn.Read(buf)
if netErr, ok := perrors.Cause(err).(net.Error); ok && netErr.Timeout() {
err = nil
}
if err != nil {
log.Infof("conn{%#v}.Read() = {length:%d, err:%+v}", conn, length, perrors.WithStack(err))
_ = conn.Close()
return nil
}
return newUDPSession(conn, c)
}

func (c *client) dialWS() Session {
Expand All @@ -245,28 +242,25 @@ func (c *client) dialWS() Session {
)

dialer.EnableCompression = true
for {
if c.IsClosed() {
return nil
}
conn, _, err = dialer.Dial(c.addr, nil)
log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err))
if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
_ = conn.Close()
err = errSelfConnect
}
if err == nil {
ss = newWSSession(conn, c)
if ss.(*session).maxMsgLen > 0 {
conn.SetReadLimit(int64(ss.(*session).maxMsgLen))
}

return ss
if c.IsClosed() {
return nil
}
conn, _, err = dialer.Dial(c.addr, nil)
if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
_ = conn.Close()
err = errSelfConnect
}
if err == nil {
ss = newWSSession(conn, c)
if ss.(*session).maxMsgLen > 0 {
conn.SetReadLimit(int64(ss.(*session).maxMsgLen))
}

log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err))
<-gxtime.After(connectInterval)
return ss
}

log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err))
return nil
}

func (c *client) dialWSS() Session {
Expand Down Expand Up @@ -323,28 +317,26 @@ func (c *client) dialWSS() Session {

// dialer.EnableCompression = true
dialer.TLSClientConfig = config
for {
if c.IsClosed() {
return nil
}
conn, _, err = dialer.Dial(c.addr, nil)
if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
_ = conn.Close()
err = errSelfConnect
}
if err == nil {
ss = newWSSession(conn, c)
if ss.(*session).maxMsgLen > 0 {
conn.SetReadLimit(int64(ss.(*session).maxMsgLen))
}
ss.SetName(defaultWSSSessionName)

return ss
if c.IsClosed() {
return nil
}
conn, _, err = dialer.Dial(c.addr, nil)
if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
_ = conn.Close()
err = errSelfConnect
}
if err == nil {
ss = newWSSession(conn, c)
if ss.(*session).maxMsgLen > 0 {
conn.SetReadLimit(int64(ss.(*session).maxMsgLen))
}
ss.SetName(defaultWSSSessionName)

log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err))
<-gxtime.After(connectInterval)
return ss
}

log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err))
return nil
}

func (c *client) dial() Session {
Expand Down Expand Up @@ -381,36 +373,36 @@ func (c *client) sessionNum() int {
return num
}

func (c *client) connect() {
func (c *client) connect() bool {
var (
err error
ss Session
)

for {
ss = c.dial()
if ss == nil {
// client has been closed
break
}
err = c.newSession(ss)
if err == nil {
ss.(*session).run()
c.Lock()
if c.ssMap == nil {
c.Unlock()
break
}
c.ssMap[ss] = struct{}{}
ss = c.dial()
if ss == nil {
// client has been closed
return false
}
err = c.newSession(ss)
if err == nil {
ss.(*session).run()
c.Lock()
if c.ssMap == nil {
c.Unlock()
ss.SetAttribute(sessionClientKey, c)
ss.SetAttribute(ignoreReconnectKey, false)
break
ss.Close()
return false
}
// don't distinguish between tcp connection and websocket connection. Because
// gorilla/websocket/conn.go:(Conn)Close also invoke net.Conn.Close()
_ = ss.Conn().Close()
c.ssMap[ss] = struct{}{}
c.Unlock()
ss.SetAttribute(sessionClientKey, c)
ss.SetAttribute(ignoreReconnectKey, false)
return true
}
// don't distinguish between tcp connection and websocket connection. Because
// gorilla/websocket/conn.go:(Conn)Close also invoke net.Conn.Close()
_ = ss.Conn().Close()
return false

@AlexStocks AlexStocks Aug 12, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] 这个 return false 只有 dial 失败并跳出 connect() 内部循环后才会执行。只要 dial 成功而 c.newSession(ss) 持续返回错误，当前实现就会关闭连接后立即在 connect() 内再次 dial，永远不会回到 reConnect() 增加 reconnectAttempts 或执行 backoff。当前 Head 的定向测试配置最大尝试次数为 2，但 RunEventLoop 超过 500ms 仍未返回。请让 connect() 一次只完成一次 dial/初始化尝试；newSession 失败时关闭 session 并返回 false，由外层统一计数和退避，同时补“dial 成功、newSession 连续失败”的回归测试。

@222twotwotwo 222twotwotwo Aug 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

收到，已将connect()从内部循环改为单次尝试，并补充测试

}

// there are two methods to keep connection pool. the first approach is like
Expand All @@ -429,10 +421,7 @@ func (c *client) RunEventLoop(newSession NewSessionCallback) {

// a for-loop connect to make sure the connection pool is valid
func (c *client) reConnect() {
var (
sessionNum, reconnectAttempts int
maxReconnectInterval int64
)
var reconnectAttempts int
reconnectInterval := c.reconnectInterval
if reconnectInterval == 0 {
reconnectInterval = defaultReconnectInterval
Expand All @@ -442,21 +431,34 @@ func (c *client) reConnect() {
maxReconnectAttempts = defaultMaxReconnectAttempts
}
connPoolSize := c.number
for {
for reconnectAttempts < maxReconnectAttempts {
if c.IsClosed() {
log.Warnf("client{peer:%s} goroutine exit now.", c.addr)
break
return
}

sessionNum = c.sessionNum()
if connPoolSize <= sessionNum || maxReconnectAttempts < reconnectAttempts {
//exit reconnect when the number of connection pools is sufficient or the current reconnection attempts exceeds the max reconnection attempts.
break
if connPoolSize <= c.sessionNum() {
return
}
if c.connect() {
reconnectAttempts = 0
continue
}
c.connect()
reconnectAttempts++
maxReconnectInterval = int64(math.Min(float64(reconnectAttempts), float64(maxBackOffTimes))) * int64(reconnectInterval)
<-gxtime.After(time.Duration(maxReconnectInterval))
if c.IsClosed() || connPoolSize <= c.sessionNum() || reconnectAttempts >= maxReconnectAttempts {
return
}

backOffTimes := reconnectAttempts
if maxBackOffTimes < backOffTimes {
backOffTimes = maxBackOffTimes
}
backoff := time.Duration(int64(backOffTimes) * int64(reconnectInterval))
select {
case <-c.done:
return
case <-gxtime.After(backoff):
}
}
}

Expand Down
Loading
Loading