This is an automated email from the ASF dual-hosted git repository.
Alanxtl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new 130b2a686 fix(getty): fix the getty poll bug described at #3509 (#3592)
130b2a686 is described below
commit 130b2a686841f319ee95ff8052e693cfa9731eeb
Author: Xuetao Li <[email protected]>
AuthorDate: Tue Sep 1 17:14:39 2026 +0800
fix(getty): fix the getty poll bug described at #3509 (#3592)
* fix 3509
* import format
* fmt
* fix format
* fix comment
* fix comment
* fix comment
* fix race
---
remoting/getty/getty_client.go | 5 +-
remoting/getty/pool.go | 28 +++--
remoting/getty/pool_test.go | 272 +++++++++++++++++++++++++++++++++++++++++
3 files changed, 297 insertions(+), 8 deletions(-)
diff --git a/remoting/getty/getty_client.go b/remoting/getty/getty_client.go
index ee6432d36..f2b7bec43 100644
--- a/remoting/getty/getty_client.go
+++ b/remoting/getty/getty_client.go
@@ -229,7 +229,7 @@ func (c *Client) RequestContext(ctx context.Context,
request *remoting.Request,
if timeout <= 0 {
timeout = c.opts.RequestTimeout
}
- _, session, err := c.selectSession(c.addr)
+ rpcClient, session, err := c.selectSession(c.addr)
if err != nil {
return perrors.WithStack(err)
}
@@ -256,6 +256,9 @@ func (c *Client) RequestContext(ctx context.Context,
request *remoting.Request,
defer timer.Stop()
select {
case <-timer.C:
+
remoting.RemovePendingResponse(remoting.SequenceType(request.ID))
+ rpcClient.removeSession(session)
+ go session.Close()
return perrors.WithStack(errClientReadTimeout)
case <-response.Done:
err = response.Err
diff --git a/remoting/getty/pool.go b/remoting/getty/pool.go
index 7434e11df..623923172 100644
--- a/remoting/getty/pool.go
+++ b/remoting/getty/pool.go
@@ -19,6 +19,7 @@ package getty
import (
"crypto/tls"
+ "errors"
"fmt"
"math/rand"
"net"
@@ -184,11 +185,16 @@ func (c *gettyRPCClient) selectSession() getty.Session {
if c.sessions == nil {
return nil
}
- count := len(c.sessions)
- if count == 0 {
+ available := make([]getty.Session, 0, len(c.sessions))
+ for _, s := range c.sessions {
+ if s != nil && s.session != nil && !s.session.IsClosed() {
+ available = append(available, s.session)
+ }
+ }
+ if len(available) == 0 {
return nil
}
- return c.sessions[rand.Int31n(int32(count))].session
+ return available[rand.Int31n(int32(len(available)))] // NOSONAR
}
func (c *gettyRPCClient) addSession(session getty.Session) {
@@ -211,6 +217,7 @@ func (c *gettyRPCClient) removeSession(session
getty.Session) {
}
var removeFlag bool
+ var removed bool
func() {
c.lock.Lock()
defer c.lock.Unlock()
@@ -219,12 +226,16 @@ func (c *gettyRPCClient) removeSession(session
getty.Session) {
}
for i, s := range c.sessions {
- if s.session == session {
+ if s != nil && s.session == session {
c.sessions = append(c.sessions[:i],
c.sessions[i+1:]...)
+ removed = true
logger.Debugf("[Remoting][Getty] delete
session=%s index=%d", session.Stat(), i)
break
}
}
+ if !removed {
+ return
+ }
logger.Infof("[Remoting][Getty] after remove session=%s, left
session number=%d", session.Stat(), len(c.sessions))
if len(c.sessions) == 0 {
removeFlag = true
@@ -289,8 +300,9 @@ func (c *gettyRPCClient) isAvailable() bool {
}
func (c *gettyRPCClient) close() error {
- closeErr := perrors.Errorf("close gettyRPCClient{%#v} again", c)
+ firstClose := false
c.once.Do(func() {
+ firstClose = true
var (
gettyClient getty.Client
sessions []*rpcSession
@@ -320,7 +332,9 @@ func (c *gettyRPCClient) close() error {
}
}()
- closeErr = nil
})
- return closeErr
+ if !firstClose {
+ return errors.New("close gettyRPCClient again")
+ }
+ return nil
}
diff --git a/remoting/getty/pool_test.go b/remoting/getty/pool_test.go
index 0559e06d6..a1d8c74ec 100644
--- a/remoting/getty/pool_test.go
+++ b/remoting/getty/pool_test.go
@@ -18,15 +18,26 @@
package getty
import (
+ "net"
"sync"
+ "sync/atomic"
"testing"
+ "time"
)
import (
+ gettylib "github.com/apache/dubbo-getty"
+
+ perrors "github.com/pkg/errors"
+
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
+import (
+ "dubbo.apache.org/dubbo-go/v3/remoting"
+)
+
func TestGettyRPCClientUpdateActive(t *testing.T) {
client := &gettyRPCClient{}
client.updateActive(1234567890)
@@ -132,3 +143,264 @@ func TestGettyRPCClientLifecycle(t *testing.T) {
require.NoError(t, client.close())
assert.Equal(t, int64(0), client.active.Load())
}
+
+type stubSession struct {
+ closed atomic.Bool
+ writes atomic.Int32
+ onWrite func()
+ onClose func()
+ closeOnce sync.Once
+}
+
+func (s *stubSession) ID() uint32 { return 1 }
+func (s *stubSession) SetCompressType(gettylib.CompressType) {}
+func (s *stubSession) LocalAddr() string { return
"127.0.0.1:12345" }
+func (s *stubSession) RemoteAddr() string { return
"127.0.0.1:20880" }
+func (s *stubSession) IncReadPkgNum() {}
+func (s *stubSession) IncWritePkgNum() {}
+func (s *stubSession) UpdateActive() {}
+func (s *stubSession) GetActive() time.Time { return
time.Now() }
+func (s *stubSession) ReadTimeout() time.Duration { return
time.Second }
+func (s *stubSession) SetReadTimeout(time.Duration) {}
+func (s *stubSession) WriteTimeout() time.Duration { return
time.Second }
+func (s *stubSession) SetWriteTimeout(time.Duration) {}
+func (s *stubSession) Send(any) (int, error) { return 0, nil
}
+func (s *stubSession) CloseConn(int) {}
+func (s *stubSession) SetSession(gettylib.Session) {}
+func (s *stubSession) Reset() {}
+func (s *stubSession) Conn() net.Conn { return nil }
+func (s *stubSession) Stat() string { return
"stub-session" }
+func (s *stubSession) IsClosed() bool { return
s.closed.Load() }
+func (s *stubSession) EndPoint() gettylib.EndPoint { return nil }
+func (s *stubSession) SetMaxMsgLen(int) {}
+func (s *stubSession) SetName(string) {}
+func (s *stubSession) SetEventListener(gettylib.EventListener) {}
+func (s *stubSession) SetPkgHandler(gettylib.ReadWriter) {}
+func (s *stubSession) SetReader(gettylib.Reader) {}
+func (s *stubSession) SetWriter(gettylib.Writer) {}
+func (s *stubSession) SetCronPeriod(int) {}
+func (s *stubSession) SetWaitTime(time.Duration) {}
+func (s *stubSession) GetAttribute(any) any { return nil }
+func (s *stubSession) SetAttribute(any, any) {}
+func (s *stubSession) RemoveAttribute(any) {}
+func (s *stubSession) WritePkg(pkg any, timeout time.Duration) (int, int,
error) {
+ s.writes.Add(1)
+ if s.onWrite != nil {
+ s.onWrite()
+ }
+ return 1, 1, nil
+}
+func (s *stubSession) WriteBytes([]byte) (int, error) { return 0, nil }
+func (s *stubSession) WriteBytesArray(...[]byte) (int, error) { return 0, nil }
+func (s *stubSession) Close() {
+ s.closeOnce.Do(func() {
+ s.closed.Store(true)
+ if s.onClose != nil {
+ s.onClose()
+ }
+ })
+}
+
+// This test case verifies the scenario described at
https://github.com/apache/dubbo-go/issues/3509.
+func TestReadTimeoutRemovesHalfDeadSession(t *testing.T) {
+ sess := &stubSession{}
+ client := &Client{addr: "127.0.0.1:20880"}
+ rpcClient := &gettyRPCClient{rpcClient: client, sessions:
[]*rpcSession{{session: sess}}}
+ client.gettyClient = rpcClient
+
+ req := remoting.NewRequest("2.0.2")
+ req.TwoWay = true
+ rsp := remoting.NewPendingResponse(req.ID)
+ remoting.AddPendingResponse(rsp)
+
+ err := client.Request(req, 10*time.Millisecond, rsp)
+ require.Error(t, err)
+ require.ErrorIs(t, err, errClientReadTimeout)
+ assert.Eventually(t, sess.IsClosed, time.Second, time.Millisecond,
"timed out session should be closed")
+ assert.Equal(t, int32(1), sess.writes.Load())
+
+ assert.Nil(t, rpcClient.selectSession())
+ assert.Empty(t, rpcClient.sessions)
+ assert.Nil(t, client.gettyClient, "the connection handle should be
reset after the last session is removed")
+ assert.Nil(t,
remoting.GetPendingResponse(remoting.SequenceType(req.ID)))
+ assert.False(t, client.closed.Load(), "a timed out session must not
close the client")
+}
+
+func TestIssueClosedSessionIsNotSelected(t *testing.T) {
+ sess := &stubSession{}
+ sess.Close()
+ client := &gettyRPCClient{sessions: []*rpcSession{{session: sess}}}
+
+ selected := client.selectSession()
+ assert.Nil(t, selected)
+}
+
+func TestDelayedOnCloseDoesNotResetReplacement(t *testing.T) {
+ client := &Client{addr: "127.0.0.1:20880"}
+ oldSession := &stubSession{}
+ oldPool := &gettyRPCClient{rpcClient: client, sessions:
[]*rpcSession{{session: oldSession}}}
+ client.gettyClient = oldPool
+
+ closeStarted := make(chan struct{})
+ allowOnClose := make(chan struct{})
+ onCloseDone := make(chan struct{})
+ oldSession.onClose = func() {
+ close(closeStarted)
+ <-allowOnClose
+ oldPool.removeSession(oldSession)
+ close(onCloseDone)
+ }
+
+ request := remoting.NewRequest("2.0.2")
+ request.TwoWay = true
+ response := remoting.NewPendingResponse(request.ID)
+ remoting.AddPendingResponse(response)
+
+ err := client.Request(request, 10*time.Millisecond, response)
+ require.ErrorIs(t, err, errClientReadTimeout)
+ <-closeStarted
+
+ replacement := &gettyRPCClient{rpcClient: client, sessions:
[]*rpcSession{{session: &stubSession{}}}}
+ client.gettyClientMux.Lock()
+ client.gettyClient = replacement
+ client.gettyClientMux.Unlock()
+
+ client.resetRpcConn(oldPool)
+ client.gettyClientMux.RLock()
+ assert.Same(t, replacement, client.gettyClient)
+ client.gettyClientMux.RUnlock()
+
+ close(allowOnClose)
+ select {
+ case <-onCloseDone:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for delayed OnClose")
+ }
+
+ client.gettyClientMux.RLock()
+ assert.Same(t, replacement, client.gettyClient)
+ client.gettyClientMux.RUnlock()
+}
+
+func TestTimeoutResetWithConcurrentSelectSession(t *testing.T) {
+ client := &Client{addr: "127.0.0.1:20880"}
+ oldSession := &stubSession{}
+ oldPool := &gettyRPCClient{rpcClient: client, sessions:
[]*rpcSession{{session: oldSession}}}
+ client.gettyClient = oldPool
+
+ closeStarted := make(chan struct{})
+ allowOnClose := make(chan struct{})
+ onCloseDone := make(chan struct{})
+ oldSession.onClose = func() {
+ close(closeStarted)
+ <-allowOnClose
+ oldPool.removeSession(oldSession)
+ close(onCloseDone)
+ }
+
+ request := remoting.NewRequest("2.0.2")
+ request.TwoWay = true
+ response := remoting.NewPendingResponse(request.ID)
+ remoting.AddPendingResponse(response)
+ require.ErrorIs(t, client.Request(request, 10*time.Millisecond,
response), errClientReadTimeout)
+ <-closeStarted
+ client.gettyClientMux.RLock()
+ assert.Nil(t, client.gettyClient)
+ client.gettyClientMux.RUnlock()
+
+ replacementSession := &stubSession{}
+ replacement := &gettyRPCClient{rpcClient: client, sessions:
[]*rpcSession{{session: replacementSession}}}
+ client.gettyClientMux.Lock()
+ client.gettyClient = replacement
+ client.gettyClientMux.Unlock()
+
+ selectDone := make(chan struct{})
+ selectErr := make(chan error, 1)
+ go func() {
+ defer close(selectDone)
+ for range 100 {
+ selectedClient, selectedSession, err :=
client.selectSession(client.addr)
+ if err != nil {
+ selectErr <- err
+ return
+ }
+ if selectedClient != replacement || selectedSession !=
replacementSession {
+ selectErr <- perrors.New("selectSession
returned a stale connection")
+ return
+ }
+ }
+ }()
+
+ close(allowOnClose)
+ select {
+ case err := <-selectErr:
+ t.Fatal(err)
+ case <-selectDone:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for concurrent selectSession")
+ }
+ select {
+ case <-onCloseDone:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for delayed OnClose")
+ }
+}
+
+func TestRequestTimeoutConcurrentWithClose(t *testing.T) {
+ client := &Client{addr: "127.0.0.1:20880"}
+ sess := &stubSession{}
+ rpcClient := &gettyRPCClient{rpcClient: client, sessions:
[]*rpcSession{{session: sess}}}
+ client.gettyClient = rpcClient
+
+ writeStarted := make(chan struct{})
+ sess.onWrite = func() {
+ close(writeStarted)
+ }
+
+ request := remoting.NewRequest("2.0.2")
+ request.TwoWay = true
+ response := remoting.NewPendingResponse(request.ID)
+ remoting.AddPendingResponse(response)
+ requestDone := make(chan error, 1)
+ go func() {
+ requestDone <- client.Request(request, 20*time.Millisecond,
response)
+ }()
+
+ select {
+ case <-writeStarted:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for request write")
+ }
+
+ // Hold the pointer lock before starting Close. Both Close and the
timeout
+ // reset must wait for this lock before accessing gettyClient.
+ client.gettyClientMux.Lock()
+ closeDone := make(chan struct{})
+ go func() {
+ client.Close()
+ close(closeDone)
+ }()
+
+ select {
+ case <-closeDone:
+ t.Fatal("Close must wait for gettyClientMux")
+ case <-time.After(100 * time.Millisecond):
+ }
+ client.gettyClientMux.Unlock()
+
+ select {
+ case err := <-requestDone:
+ require.ErrorIs(t, err, errClientReadTimeout)
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for request timeout")
+ }
+ select {
+ case <-closeDone:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for client close")
+ }
+ client.gettyClientMux.RLock()
+ assert.Nil(t, client.gettyClient)
+ client.gettyClientMux.RUnlock()
+ assert.True(t, client.closed.Load())
+}
