Tsukikage7 commented on code in PR #1017: URL: https://github.com/apache/dubbo-go-pixiu/pull/1017#discussion_r3740996486
########## 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: 已修复，感谢指出。 现在 `ClusterManager.DeleteEndpoint` 会在删除 endpoint 后通知已注册的 removal handler，并传递 `cluster + endpoint address`。gRPC connection manager 收到通知后会： - 删除对应的 `cluster + endpoint` connection entry； - 关闭健康的旧 `grpc.ClientConn`； - 通过现有 `onRemove` 回调清理该 connection 对应的 descriptor cache。 此外，HTTP filter factory 在 reload/close 时会注销 handler；旧 factory 通过 request chain lease drain 后才关闭，避免 callback 和连接资源遗留。 相关修改位于 `pkg/server/cluster_manager.go`、`pkg/filter/http/grpcproxy/connection_manager.go`、`pkg/filter/http/grpcproxy/grpc.go` 和 filter manager 生命周期代码。benchmark 三套 suite 均通过。 ########## pkg/listener/http/http_listener.go: ########## @@ -83,7 +83,18 @@ func (ls *HttpListenerService) Start() error { } func (ls *HttpListenerService) Close() error { - return ls.srv.Close() + serverErr := error(nil) + if ls.srv != nil { + serverErr = ls.srv.Close() + } + filterErr := error(nil) + if ls.FilterChain != nil { + filterErr = ls.FilterChain.Close() Review Comment: 已修复，感谢指出。 当前实现增加了两层生命周期保护： - `HttpListenerService` 使用 `RWMutex` 保护 FilterChain；请求在 `RLock` 下执行，`Refresh` 在写锁下完成 swap，随后关闭旧 chain，因此正在执行的请求完成前不会释放旧资源； - HTTP `FilterManager` 为每个请求创建的 filter chain 持有 factory lease。reload 后旧 factory 先标记为 retired，只有所有活跃 chain 通过 `Release()` 释放后才调用 factory `Close()`，避免热刷新期间关闭仍在使用的连接。 对应代码：`pkg/listener/http/http_listener.go`、`pkg/common/extension/filter/filter_manager.go`、`pkg/common/extension/filter/filter_chain.go`、`pkg/common/http/manager.go`。相关包的 race 测试已通过。 ########## pkg/filter/http/grpcproxy/descriptor.go: ########## @@ -117,12 +128,79 @@ func (dr *Descriptor) getServerDescriptorSourceCtx(refCtx context.Context, cfg * default: err = errors.Errorf("found a value of type %s, which is not *grpc.ClientConn, ", t) } - return &serverSource{client: grpcreflect.NewClient(refCtx, reflectpb.NewServerReflectionClient(cc))}, err + if err != nil { + return nil, err + } + + // The reflection client is created per lookup and bound to the request + // context so every remote reflection RPC honors the request timeout. + // It must not be cached connection-scoped: grpcreflect reuses the root + // context for every RPC, and a cached client would lose the deadline and + // keep the per-request timeout from applying. The method descriptor + // cache in getMethodDescriptor below is what avoids repeating the + // reflection RPC after the first lookup. + return &serverSource{client: grpcreflect.NewClientV1Alpha(refCtx, reflectpb.NewServerReflectionClient(cc))}, nil } // nolint func (dr *Descriptor) getServerDescriptorSource(refCtx context.Context, cc *grpc.ClientConn) DescriptorSource { - return &serverSource{client: grpcreflect.NewClient(refCtx, reflectpb.NewServerReflectionClient(cc))} + return &serverSource{client: grpcreflect.NewClientV1Alpha(refCtx, reflectpb.NewServerReflectionClient(cc))} +} + +func (dr *Descriptor) removeConnection(cc *grpc.ClientConn) { + if cc == nil { + return + } + dr.methodMu.Lock() + delete(dr.methodDescs, cc) + dr.methodMu.Unlock() +} + +func (dr *Descriptor) Close() { + dr.methodMu.Lock() + dr.methodDescs = nil + dr.methodMu.Unlock() +} + +func (dr *Descriptor) getMethodDescriptor(source DescriptorSource, cc *grpc.ClientConn, service, method string) (*desc.MethodDescriptor, error) { + key := service + "\x00" + method + dr.methodMu.RLock() + if methods := dr.methodDescs[cc]; methods != nil { + if descriptor, ok := methods[key]; ok { + dr.methodMu.RUnlock() + return descriptor, nil + } + } + dr.methodMu.RUnlock() + + dr.methodMu.Lock() + defer dr.methodMu.Unlock() + if methods := dr.methodDescs[cc]; methods != nil { + if descriptor, ok := methods[key]; ok { + return descriptor, nil + } + } + + dscp, err := source.FindSymbol(service) Review Comment: 已修复，感谢指出。 `source.FindSymbol(service)` 已移出全局 `methodMu` 写锁。现在流程是： 1. 先在读锁下检查 descriptor cache； 2. 按 `connection + generation + service + method` 使用 `singleflight` 合并同一个 cache miss； 3. 在锁外执行网络 reflection； 4. 返回前再短暂获取写锁进行 double-check 和写入。 另外，connection 删除或 descriptor/factory close 时会递增 generation。已经开始的旧 reflection 即使之后返回，也不能把 descriptor 写回新一代 cache。 相关包 `go test -race -count=20` 已通过，`golangci-lint --new-from-rev=develop` 也为 0 issues。 -- 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]
