fix:修复balancer任务协程hang死问题

balancer.go

@@ -171,7 +171,7 @@ func (p *polarisNamingBalancer) createSubConnection(key string, addr resolver.Ad
 	}
 	// is a new address (not existing in b.subConns).
 	sc, err := p.cc.NewSubConn(
-		[]resolver.Address{addr}, balancer.NewSubConnOptions{HealthCheckEnabled: true})
+		[]resolver.Address{addr}, balancer.NewSubConnOptions{HealthCheckEnabled: false})
 	if err != nil {
 		GetLogger().Error("[Polaris][Balancer] failed to create new SubConn: %v", err)
 		return
@@ -245,6 +245,8 @@ func (p *polarisNamingBalancer) ResolverError(err error) {
 		// report an error.
 		return
 	}
+	p.rwMutex.RLock()
+	defer p.rwMutex.RUnlock()
 	p.regeneratePicker(nil)
 	p.cc.UpdateState(balancer.State{
 		ConnectivityState: p.state,
@@ -308,8 +310,6 @@ func (p *polarisNamingBalancer) regeneratePicker(options *dialOptions) {
 		return
 	}
 	readySCs := make(map[string]balancer.SubConn)
-	p.rwMutex.RLock()
-	defer p.rwMutex.RUnlock()
 	// Filter out all ready SCs from full subConn map.
 	for addr, sc := range p.subConns {
 		if st, ok := p.scStates[sc]; ok && st == connectivity.Ready {

examples/quickstart/consumer/main.go

@@ -22,6 +22,9 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -41,11 +44,16 @@ func main() {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	polaris.GetLogger().SetLevel(polaris.LogDebug)
+
 	conn, err := polaris.DialContext(ctx, "polaris://QuickStartEchoServerGRPC",
 		polaris.WithGRPCDialOptions(grpc.WithTransportCredentials(insecure.NewCredentials())),
 		polaris.WithDisableRouter(),
 		polaris.WithDisableCircuitBreaker(),
 	)
+
+	conn.Close()
+
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -84,8 +92,24 @@ func main() {
 		_, _ = w.Write([]byte(resp.GetValue()))
 	}
 	http.HandleFunc("/echo", indexHandler)
-	if err := http.ListenAndServe(fmt.Sprintf(":%d", listenPort), nil); nil != err {
-		log.Fatal(err)
-	}
+	go func() {
+		if err := http.ListenAndServe(fmt.Sprintf(":%d", listenPort), nil); nil != err {
+			log.Fatal(err)
+		}
+	}()
+	runMainLoop()
+}
 
+func runMainLoop() {
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, []os.Signal{
+		syscall.SIGINT, syscall.SIGTERM,
+		syscall.SIGSEGV,
+	}...)
+
+	for s := range ch {
+		log.Printf("catch signal(%+v), stop servers", s)
+		polaris.ClosePolarisContext()
+		return
+	}
 }

examples/quickstart/consumer/polaris.yaml

@@ -1,4 +1,18 @@
 global:
   serverConnector:
     addresses:
-      - 127.0.0.1:8091
+      - 127.0.0.1:8091
+  statReporter:
+    #描述：是否将统计信息上报至monitor
+    #类型：bool
+    enable: true
+    #描述：启用的统计上报插件类型
+    #类型：list
+    #范围：已经注册的统计上报插件的名字
+    chain:
+      - prometheus
+    plugin:
+      prometheus:
+        type: push
+        address: 127.0.0.1:9091
+        interval: 10s

global.go

@@ -42,7 +42,7 @@ var (
 	// DefaultNamespace default namespace when namespace is not set
 	DefaultNamespace = "default"
 	// DefaultTTL default ttl value when ttl is not set
-	DefaultTTL = 20
+	DefaultTTL = 5
 	// DefaultGracefulStopMaxWaitDuration default stop max wait duration when not set
 	DefaultGracefulStopMaxWaitDuration = 30 * time.Second
 	// DefaultDelayStopWaitDuration default delay time before stop
@@ -57,16 +57,31 @@ const (
 )
 
 var (
+	ctxRef              = 0
 	polarisContext      api.SDKContext
 	polarisConfig       config.Configuration
 	mutexPolarisContext sync.Mutex
 	oncePolarisConfig   sync.Once
 )
 
+func ClosePolarisContext() {
+	mutexPolarisContext.Lock()
+	defer mutexPolarisContext.Unlock()
+	if nil == polarisContext {
+		return
+	}
+	ctxRef--
+	if ctxRef == 0 {
+		polarisContext.Destroy()
+		polarisContext = nil
+	}
+}
+
 // PolarisContext get or init the global polaris context
 func PolarisContext() (api.SDKContext, error) {
 	mutexPolarisContext.Lock()
 	defer mutexPolarisContext.Unlock()
+	ctxRef++
 	if nil != polarisContext {
 		return polarisContext, nil
 	}

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/natefinch/lumberjack v2.0.0+incompatible
-	github.com/polarismesh/polaris-go v1.6.0-beta.5
+	github.com/polarismesh/polaris-go v1.6.0-alpha.7
 	github.com/polarismesh/specification v1.5.1
 	github.com/prometheus/client_golang v1.19.1 // indirect
 	github.com/prometheus/common v0.54.0 // indirect

go.sum

@@ -1695,8 +1695,8 @@ github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZ
 github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/polarismesh/polaris-go v1.6.0-beta.5 h1:llucvfydWlFWTNeABHbbuVL2ijR7AITx8UG02tx0c/Y=
-github.com/polarismesh/polaris-go v1.6.0-beta.5/go.mod h1:CuXO9bhHGjSoOIMWr4NXf3bJAkRBp5YoM7ibBzENC+c=
+github.com/polarismesh/polaris-go v1.6.0-alpha.7 h1:cWyx5BC6+JurBNCV3EaeCjd6U8edY6KrG6HkkBhWMR0=
+github.com/polarismesh/polaris-go v1.6.0-alpha.7/go.mod h1:CuXO9bhHGjSoOIMWr4NXf3bJAkRBp5YoM7ibBzENC+c=
 github.com/polarismesh/specification v1.4.1/go.mod h1:rDvMMtl5qebPmqiBLNa5Ps0XtwkP31ZLirbH4kXA0YU=
 github.com/polarismesh/specification v1.5.1 h1:cJ2m0RBepdopGo/e3UpKdsab3NpDZnw5IsVTB1sFc5I=
 github.com/polarismesh/specification v1.5.1/go.mod h1:rDvMMtl5qebPmqiBLNa5Ps0XtwkP31ZLirbH4kXA0YU=

resolver.go

@@ -55,7 +55,14 @@ func RegisterResolverInterceptor(i ResolverInterceptor) {
 	resolverInterceptors = append(resolverInterceptors, i)
 }
 
+func NewResolver(ctx api.SDKContext) *resolverBuilder {
+	return &resolverBuilder{
+		sdkCtx: ctx,
+	}
+}
+
 type resolverBuilder struct {
+	sdkCtx api.SDKContext
 }
 
 // Scheme polaris scheme
@@ -104,12 +111,14 @@ func (rb *resolverBuilder) Build(
 		return nil, err
 	}
 
-	sdkCtx, err := PolarisContext()
-	if nil != err {
-		return nil, err
+	if rb.sdkCtx == nil {
+		sdkCtx, err := PolarisContext()
+		if nil != err {
+			return nil, err
+		}
+		rb.sdkCtx = sdkCtx
 	}
-
-	options.SDKContext = sdkCtx
+	options.SDKContext = rb.sdkCtx
 
 	ctx, cancel := context.WithCancel(context.Background())
 	d := &polarisNamingResolver{
@@ -120,7 +129,7 @@ func (rb *resolverBuilder) Build(
 		options:  options,
 		host:     host,
 		port:     port,
-		consumer: api.NewConsumerAPIByContext(sdkCtx),
+		consumer: api.NewConsumerAPIByContext(rb.sdkCtx),
 		eventCh:  make(chan struct{}, 1),
 	}
 	go d.watcher()
@@ -251,9 +260,10 @@ func (pr *polarisNamingResolver) watcher() {
 				pr.options.Namespace, pr.host)
 			return
 		case <-pr.eventCh:
+			pr.doRefresh()
 		case <-ticker.C:
+			pr.doRefresh()
 		}
-		pr.doRefresh()
 	}
 }