milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: fix datarace in root_coord proxy_client_manager (#28796) 

bug: #28797

Signed-off-by: wayblink <anyang.wang@zilliz.com>
pull/29075/head
wayblink 2023-12-12 10:18:39 +08:00 committed by GitHub
parent ef18e6a532
commit 994b239161
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 12 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/rootcoord/proxy_client_manager.go
View File

@ -70,12 +70,18 @@ func newProxyClientManager(creator proxyCreator) *proxyClientManager {
}
}
func (p *proxyClientManager) GetProxyClients(sessions []*sessionutil.Session) {
func (p *proxyClientManager) AddProxyClients(sessions []*sessionutil.Session) {
for _, session := range sessions {
p.AddProxyClient(session)
}
}
func (p *proxyClientManager) GetProxyClients() map[int64]types.ProxyClient {
p.lock.RLock()
defer p.lock.RUnlock()
return p.proxyClient
}
func (p *proxyClientManager) AddProxyClient(session *sessionutil.Session) {
p.lock.RLock()
_, ok := p.proxyClient[session.ServerID]

4
internal/rootcoord/proxy_client_manager_test.go
View File

@ -95,7 +95,7 @@ func (p *proxyMock) RefreshPolicyInfoCache(ctx context.Context, req *proxypb.Ref
return merr.Success(), nil
}
func TestProxyClientManager_GetProxyClients(t *testing.T) {
func TestProxyClientManager_AddProxyClients(t *testing.T) {
paramtable.Init()
core, err := NewCore(context.Background(), nil)
@ -125,7 +125,7 @@ func TestProxyClientManager_GetProxyClients(t *testing.T) {
}
sessions := []*sessionutil.Session{session}
pcm.GetProxyClients(sessions)
pcm.AddProxyClients(sessions)
}
func TestProxyClientManager_AddProxyClient(t *testing.T) {

8
internal/rootcoord/root_coord.go
View File

@ -484,7 +484,7 @@ func (c *Core) initInternal() error {
c.ctx,
c.etcdCli,
c.chanTimeTick.initSessions,
c.proxyClientManager.GetProxyClients,
c.proxyClientManager.AddProxyClients,
)
c.proxyManager.AddSessionFunc(c.chanTimeTick.addSession, c.proxyClientManager.AddProxyClient)
c.proxyManager.DelSessionFunc(c.chanTimeTick.delSession, c.proxyClientManager.DelProxyClient)
@ -2779,10 +2779,9 @@ func (c *Core) CheckHealth(ctx context.Context, in *milvuspb.CheckHealthRequest)
mu := &sync.Mutex{}
group, ctx := errgroup.WithContext(ctx)
errReasons := make([]string, 0, len(c.proxyClientManager.proxyClient))
errReasons := make([]string, 0, c.proxyClientManager.GetProxyCount())
c.proxyClientManager.lock.RLock()
for nodeID, proxyClient := range c.proxyClientManager.proxyClient {
for nodeID, proxyClient := range c.proxyClientManager.GetProxyClients() {
nodeID := nodeID
proxyClient := proxyClient
group.Go(func() error {
@ -2800,7 +2799,6 @@ func (c *Core) CheckHealth(ctx context.Context, in *milvuspb.CheckHealthRequest)
return nil
})
}
c.proxyClientManager.lock.RUnlock()
err := group.Wait()
if err != nil || len(errReasons) != 0 {