Tsukikage7 commented on code in PR #1017: URL: https://github.com/apache/dubbo-go-pixiu/pull/1017#discussion_r3801315177
########## pkg/filter/http/grpcproxy/connection_manager.go: ########## @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package grpcproxy + +import ( + "context" + "fmt" + "sync" + "time" +) + +import ( + "golang.org/x/sync/singleflight" + + "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" + "google.golang.org/grpc/credentials/insecure" +) + +const defaultGRPCDialTimeout = 5 * time.Second + +type grpcConnectionDialer func(context.Context, string) (*grpc.ClientConn, error) + +// grpcConnectionManager owns long-lived backend connections for the HTTP gRPC +// proxy. A grpc.ClientConn is safe for concurrent use and multiplexes calls +// over HTTP/2, so a sync.Pool is both unnecessary and incorrect here. +type grpcConnectionManager struct { + connections sync.Map + creates singleflight.Group + dial grpcConnectionDialer + dialTimeout time.Duration + onRemove func(*grpc.ClientConn) + + mu sync.Mutex + closed bool +} + +func newGRPCConnectionManager() *grpcConnectionManager { + return &grpcConnectionManager{ + dial: dialGRPCConnection, + dialTimeout: defaultGRPCDialTimeout, + } +} + +func dialGRPCConnection(ctx context.Context, endpoint string) (*grpc.ClientConn, error) { + return grpc.DialContext( //nolint:staticcheck // SA1019: the context is required to enforce the dial timeout. + ctx, + endpoint, + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) +} + +func (m *grpcConnectionManager) Get(ctx context.Context, key, endpoint string) (*grpc.ClientConn, error) { + if key == "" || endpoint == "" { + return nil, fmt.Errorf("grpc connection key and endpoint must not be empty") + } + if ctx == nil { + ctx = context.Background() + } + + if conn, ok := m.loadHealthy(key); ok { + return conn, nil + } + + result := m.creates.DoChan(key, func() (any, error) { + if conn, ok := m.loadHealthy(key); ok { + return conn, nil + } + + m.mu.Lock() + if m.closed { + m.mu.Unlock() + return nil, fmt.Errorf("grpc connection manager is closed") + } + dial := m.dial + dialTimeout := m.dialTimeout + m.mu.Unlock() + + dialCtx, cancel := context.WithTimeout(context.Background(), dialTimeout) + defer cancel() + conn, err := dial(dialCtx, endpoint) + if err != nil { + return nil, err + } + + m.mu.Lock() + closed := m.closed + if !closed { + m.connections.Store(key, conn) Review Comment: Fixed in 95cc6c58. `endpointEventVers` and `endpointGenerations` now use a bounded 1024-entry LRU tombstone set. Removed endpoints are evicted only after their connection and in-flight dial references are gone, so an in-flight dial still cannot publish after removal. Request-only state created by a failed or canceled `Get` is also reclaimed. Added regression coverage for 4096 unique endpoint churn, request-only state cleanup, and the existing removed-during-dial path. Verified with `go test ./pkg/... -gcflags=-l -race -count=1`, `go vet ./pkg/...`, and `go build ./...`. ########## pkg/listener/http/http_listener.go: ########## @@ -96,13 +112,30 @@ func (ls *HttpListenerService) ShutDown(wg any) error { cancel() wg.(*sync.WaitGroup).Done() }() - return ls.srv.Shutdown(ctx) + serverErr := ls.srv.Shutdown(ctx) + filterErr := error(nil) + ls.filterMu.Lock() + filterChain := ls.FilterChain + ls.FilterChain = nil + ls.filterMu.Unlock() Review Comment: Fixed in 95cc6c58. After `http.Server.Shutdown(ctx)` returns at the deadline, the listener now uses `TryLock` and defers filter-chain detach/close until active requests release `filterMu.RLock`, instead of synchronously waiting on `filterMu.Lock` in the shutdown path. Added a regression test covering an active request read lock and deferred cleanup. Verified with `go test ./pkg/... -gcflags=-l -race -count=1`, `go vet ./pkg/...`, and `go build ./...`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
