blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-25 11:06:43 +08:00
Code Issues Packages Projects Releases Wiki Activity

test fix dead lock

Browse Source
This commit is contained in:
wangchuxiao 2023-06-01 16:38:33 +08:00
parent dcd46e2363
commit d9d64d810b
3 changed files with 5 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
pkg/discoveryregistry/zookeeper/discover.go
View File

@ -71,7 +71,6 @@ func (s *ZkClient) GetConnsRemote(serviceName string) (conns []resolver.Address,
func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) { func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) {
s.lock.Lock() s.lock.Lock()
defer s.lock.Unlock()
opts = append(s.options, opts...) opts = append(s.options, opts...)
conns := s.localConns[serviceName] conns := s.localConns[serviceName]
if len(conns) == 0 { if len(conns) == 0 {
@ -82,6 +81,7 @@ func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grp
} }
s.localConns[serviceName] = conns s.localConns[serviceName] = conns
} }
s.lock.Unlock()
var ret []*grpc.ClientConn var ret []*grpc.ClientConn
for _, conn := range conns { for _, conn := range conns {
c, err := grpc.DialContext(ctx, conn.Addr, append(s.options, opts...)...) c, err := grpc.DialContext(ctx, conn.Addr, append(s.options, opts...)...)
@ -94,9 +94,8 @@ func (s *ZkClient) GetConns(ctx context.Context, serviceName string, opts ...grp
} }
func (s *ZkClient) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { func (s *ZkClient) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
// newOpts := append(s.options, grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, s.balancerName))) newOpts := append(s.options, grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, s.balancerName)))
// return grpc.DialContext(ctx, fmt.Sprintf("%s:///%s", s.scheme, serviceName), append(newOpts, opts...)...) return grpc.DialContext(ctx, fmt.Sprintf("%s:///%s", s.scheme, serviceName), append(newOpts, opts...)...)
return grpc.DialContext(ctx, fmt.Sprintf("%s:///%s", s.scheme, serviceName), append(s.options, opts...)...)
} }
func (s *ZkClient) GetFirstConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { func (s *ZkClient) GetFirstConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {

4
pkg/discoveryregistry/zookeeper/resolver.go
View File

@ -3,7 +3,6 @@ package openKeeper
import ( import (
"context" "context"
"strings" "strings"
"sync"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log" "github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"google.golang.org/grpc/resolver" "google.golang.org/grpc/resolver"
@ -13,7 +12,6 @@ type Resolver struct {
target resolver.Target target resolver.Target
cc resolver.ClientConn cc resolver.ClientConn
addrs []resolver.Address addrs []resolver.Address
lock sync.RWMutex
getConnsRemote func(serviceName string) (conns []resolver.Address, err error) getConnsRemote func(serviceName string) (conns []resolver.Address, err error)
} }
@ -25,9 +23,7 @@ func (r *Resolver) ResolveNow(o resolver.ResolveNowOptions) {
log.ZError(context.Background(), "resolve now error", err, "target", r.target) log.ZError(context.Background(), "resolve now error", err, "target", r.target)
return return
} }
r.lock.Lock()
r.addrs = newConns r.addrs = newConns
r.lock.Unlock()
r.cc.UpdateState(resolver.State{Addresses: r.addrs}) r.cc.UpdateState(resolver.State{Addresses: r.addrs})
} }

4
pkg/discoveryregistry/zookeeper/zk.go
View File

@ -103,8 +103,8 @@ func NewClient(zkServers []string, zkRoot string, options ...ZkOption) (*ZkClien
resolver.Register(client) resolver.Register(client)
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(2) wg.Add(2)
go client.refresh(&wg) // go client.refresh(&wg)
go client.watch(&wg) // go client.watch(&wg)
wg.Wait() wg.Wait()
return client, nil return client, nil
} }