Alanxtl commented on code in PR #26: URL: https://github.com/apache/dubbo-go-extensions/pull/26#discussion_r4109880662
########## filter/adaptivesvc/options.go: ########## @@ -0,0 +1,77 @@ +/* + * 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 adaptivesvc + +import ( + "fmt" +) + +import ( + "dubbo.apache.org/dubbo-go/v3/common/constant" + "dubbo.apache.org/dubbo-go/v3/common/extension" +) + +const adaptiveServiceExtensionName = "adaptive-service" + +var ( + _ extension.Config = (*Config)(nil) + _ extension.Option = adaptiveServiceOption{} +) + +// Config defines the adaptive service extension configuration. +type Config struct{} + +func (c *Config) Prefix() string { + return adaptiveServiceExtensionName +} + +func (c *Config) New() extension.Config { + return &Config{} +} + +func (c *Config) Init(scope extension.Scope) error { + if scope != extension.ServerScope { + return fmt.Errorf("adaptive service only supports server scope") Review Comment: 这里加一个logger ########## filter/adaptivesvc/limiter/hill_climbing.go: ########## @@ -0,0 +1,316 @@ +/* + * 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 limiter + +import ( + "math" + "sync" + "time" +) + +import ( + "github.com/dubbogo/gost/log/logger" + + "go.uber.org/atomic" +) + +var ( + _ Limiter = (*HillClimbing)(nil) + _ Updater = (*HillClimbingUpdater)(nil) +) + +// HillClimbing is a limiter using HillClimbing algorithm +type HillClimbing struct { + seq *atomic.Uint64 + round *atomic.Uint64 + + inflight *atomic.Uint64 + limitation *atomic.Uint64 + + mutex *sync.Mutex + // nextUpdateTime = lastUpdatedTime + updateInterval + updateInterval *atomic.Duration + lastUpdatedTime *atomic.Time + + // metrics of the current round + transactionNum *atomic.Uint64 + rttAvg *atomic.Float64 + + // best metrics in the history + bestMaxCapacity *atomic.Float64 + bestRTTAvg *atomic.Float64 + bestLimitation *atomic.Uint64 + bestTPS *atomic.Uint64 +} + +func NewHillClimbing() Limiter { + l := &HillClimbing{ + seq: new(atomic.Uint64), + round: new(atomic.Uint64), + inflight: new(atomic.Uint64), + limitation: atomic.NewUint64(initialLimitation), + mutex: new(sync.Mutex), + updateInterval: atomic.NewDuration(radicalPeriod), + lastUpdatedTime: atomic.NewTime(time.Now()), + transactionNum: new(atomic.Uint64), + rttAvg: new(atomic.Float64), + bestMaxCapacity: new(atomic.Float64), + bestRTTAvg: atomic.NewFloat64(math.MaxFloat64), + bestLimitation: new(atomic.Uint64), + bestTPS: new(atomic.Uint64), + } + + return l +} + +func (l *HillClimbing) Inflight() uint64 { + return l.inflight.Load() +} + +func (l *HillClimbing) Remaining() uint64 { + limitation := l.limitation.Load() + inflight := l.Inflight() + if limitation < inflight { + return 0 + } + return limitation - inflight +} + Review Comment: Acquire hill_climbing.go#L94-L99 先调用 Remaining() 判断容量，再在 NewHillClimbingUpdater hill_climbing.go#L109-L115 中增加 inflight。 并发请求可能同时通过容量检查，导致实际 inflight 超过 limitation，违背“达到容量后拒绝请求”的行为。建议使用 CAS，或在检查和增加之间加锁，并补充高并发测试。 ########## filter/adaptivesvc/README_CN.md: ########## @@ -0,0 +1,72 @@ +# 自适应服务限流 + +[English](./README.md) | 简体中文 + +dubbo-go 提供者侧自适应服务限流扩展。 + +该扩展保留当前 dubbo-go 内置实现的 key 和行为，同时允许应用显式依赖 `github.com/apache/dubbo-go-extensions`。 Review Comment: ```suggestion ``` ########## filter/adaptivesvc/README_CN.md: ########## @@ -0,0 +1,72 @@ +# 自适应服务限流 + +[English](./README.md) | 简体中文 + +dubbo-go 提供者侧自适应服务限流扩展。 + +该扩展保留当前 dubbo-go 内置实现的 key 和行为，同时允许应用显式依赖 `github.com/apache/dubbo-go-extensions`。 + +## 安装 + +```bash +go get github.com/apache/dubbo-go-extensions/filter/adaptivesvc +``` + +导入扩展，并通过扩展提供的 option 启用 adaptive service： + +```go +import ( + adaptivesvc "github.com/apache/dubbo-go-extensions/filter/adaptivesvc" + + "dubbo.apache.org/dubbo-go/v3/server" +) + +func main() { + server.NewServer( + server.WithExtension( + adaptivesvc.WithAdaptiveService(), + ), + ) +} +``` + +导入扩展会注册扩展配置和 `padasvc` filter。`server.WithExtension` 使用 server +scope 初始化扩展，由 dubbo-go 将该 filter 合并到 provider filter chain，并保留 +已经配置的其他 filter。 + +provider filter 只在 invocation 携带 `adaptive-service.enabled=1` 时执行限流。启用后，它会使用 hill-climbing limiter，并通过响应 attachment 返回 provider 状态： + +- `adaptive-service.remaining` +- `adaptive-service.inflight` + +consumer 侧需要在发出的 invocation 上设置启用 attachment，并读取 provider 返回的容量 attachment。consumer 侧 adaptive service 支持不在本 provider extension 的范围内。 + +## 限流效果 + +limiter 会为每个服务方法维护独立的自适应并发上限。当方法的当前并发达到上限时， +provider filter 会在请求进入业务处理逻辑前将其拒绝。该上限会根据实际吞吐量和请求 +延迟动态调整，从而把执行中的请求数控制在该方法当前可承受的容量内。 + +在 [dubbo-go#3347](https://github.com/apache/dubbo-go/pull/3347) 的 provider +保护实验中，客户端以 200 并发请求一个处理耗时为 200 ms 的方法。运行到第 10 秒时， +一组示例结果为：成功 610 次、限流拒绝 180 次、其他失败为 0，进入业务处理逻辑的 +最大并发数为 53： + +```text +elapsed=10s started=840 success=610 rejected=180 failed=0 qps=61.0 reject_rate=21.4% avg=205ms p95=240ms server_active=50 server_max_active=53 +``` Review Comment: 不用给外部链接 直接展示一下表格形式的数据就行了 ########## filter/adaptivesvc/options.go: ########## @@ -0,0 +1,77 @@ +/* + * 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 adaptivesvc + +import ( + "fmt" +) + +import ( + "dubbo.apache.org/dubbo-go/v3/common/constant" + "dubbo.apache.org/dubbo-go/v3/common/extension" +) + +const adaptiveServiceExtensionName = "adaptive-service" + +var ( + _ extension.Config = (*Config)(nil) + _ extension.Option = adaptiveServiceOption{} +) + +// Config defines the adaptive service extension configuration. +type Config struct{} + +func (c *Config) Prefix() string { + return adaptiveServiceExtensionName +} + +func (c *Config) New() extension.Config { + return &Config{} +} + +func (c *Config) Init(scope extension.Scope) error { + if scope != extension.ServerScope { + return fmt.Errorf("adaptive service only supports server scope") + } + return nil +} + +func (c *Config) FilterNames(scope extension.Scope) []string { + if scope != extension.ServerScope { + return nil + } + return []string{constant.AdaptiveServiceProviderFilterKey} +} + +type adaptiveServiceOption struct{} + +func (adaptiveServiceOption) Prefix() string { + return adaptiveServiceExtensionName +} + +func (adaptiveServiceOption) Apply(config extension.Config) error { + if _, ok := config.(*Config); !ok { + return fmt.Errorf("adaptive service received unexpected config type %T", config) + } + return nil +} + +// WithAdaptiveService enables adaptive service for a dubbo-go server. +func WithAdaptiveService() extension.Option { + return adaptiveServiceOption{} +} Review Comment: Please expose the provider-side adaptive-service configuration through `extension.Config` and typed `WithXxx` options. The current `Config` is empty and `adaptiveServiceOption.Apply` is a no-op, so the migrated extension only enables the filter and loses the previous verbose configuration capability. -- 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]
