perf: 渠道 MaxConcurrency 改配置后排空热生效,无需重启
信号量条目带上容量,TryAcquire 发现容量变更且旧槽已排空时就地换新。 release 闭包持有旧 chan,换新后在途请求释放进旧槽是无害的。
This commit is contained in:
1 parent
a3197e2146
commit
bb20ebbed0
2 files changed
+48
-7
No files matched your search
@@ -25,7 +25,7 @@ type Service struct {
|
||||
enc *crypto.Encryptor
|
||||
|
||||
mu sync.Mutex
|
||||
sems map[uint64]chan struct{}
|
||||
sems map[uint64]*semEntry
|
||||
runt map[uint64]*runtimeHealth
|
||||
|
||||
// 运行期冷却参数;NewService 给默认值,测试可覆盖。
|
||||
@@ -44,7 +44,7 @@ type runtimeHealth struct {
|
||||
func NewService(db *gorm.DB, enc *crypto.Encryptor) *Service {
|
||||
return &Service{
|
||||
db: db, enc: enc,
|
||||
sems: map[uint64]chan struct{}{},
|
||||
sems: map[uint64]*semEntry{},
|
||||
runt: map[uint64]*runtimeHealth{},
|
||||
failThreshold: 3,
|
||||
cooldownFor: 30 * time.Second,
|
||||
@@ -179,6 +179,14 @@ func pickWeight(c Candidate) int {
|
||||
return bw * cw
|
||||
}
|
||||
|
||||
// semEntry 渠道并发信号量。容量变更(管理后台改 MaxConcurrency)时在旧槽排空后
|
||||
// 就地换新,改配置不再需要重启。release 闭包持有旧 chan:换新后旧持有者释放进旧
|
||||
// chan 是无害的(新 chan 重新从 0 计数)。
|
||||
type semEntry struct {
|
||||
sem chan struct{}
|
||||
cap int
|
||||
}
|
||||
|
||||
// TryAcquire 尝试获取渠道并发槽;渠道满载返回 false(调用方可溢出到其他渠道)。
|
||||
// MaxConcurrency<=0 视为不限制。
|
||||
func (s *Service) TryAcquire(ch *store.Channel) (func(), bool) {
|
||||
@@ -186,11 +194,16 @@ func (s *Service) TryAcquire(ch *store.Channel) (func(), bool) {
|
||||
return func() {}, true
|
||||
}
|
||||
s.mu.Lock()
|
||||
sem, ok := s.sems[ch.ID]
|
||||
e, ok := s.sems[ch.ID]
|
||||
// 容量变了且旧槽已排空(len 即当前持有数)→ 换新;还有在途请求就先沿用旧槽
|
||||
if !ok {
|
||||
sem = make(chan struct{}, ch.MaxConcurrency)
|
||||
s.sems[ch.ID] = sem
|
||||
e = &semEntry{sem: make(chan struct{}, ch.MaxConcurrency), cap: ch.MaxConcurrency}
|
||||
s.sems[ch.ID] = e
|
||||
} else if e.cap != ch.MaxConcurrency && len(e.sem) == 0 {
|
||||
e = &semEntry{sem: make(chan struct{}, ch.MaxConcurrency), cap: ch.MaxConcurrency}
|
||||
s.sems[ch.ID] = e
|
||||
}
|
||||
sem := e.sem
|
||||
s.mu.Unlock()
|
||||
select {
|
||||
case sem <- struct{}{}:
|
||||
|
||||
@@ -308,8 +308,8 @@ func TestPickBindingWeightMultiplier(t *testing.T) {
|
||||
db.Create(&m)
|
||||
var chs []store.Channel
|
||||
db.Order("id").Find(&chs)
|
||||
db.Create(&store.ChannelModelBinding{ChannelID: chs[0].ID, ModelID: m.ID, UpstreamModel: "mm"}) // 默认 weight=1
|
||||
db.Create(&store.ChannelModelBinding{ChannelID: chs[1].ID, ModelID: m.ID, UpstreamModel: "mm", Weight: 100}) // 显式放大
|
||||
db.Create(&store.ChannelModelBinding{ChannelID: chs[0].ID, ModelID: m.ID, UpstreamModel: "mm"}) // 默认 weight=1
|
||||
db.Create(&store.ChannelModelBinding{ChannelID: chs[1].ID, ModelID: m.ID, UpstreamModel: "mm", Weight: 100}) // 显式放大
|
||||
|
||||
s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890"))
|
||||
cands := s.Candidates("mm")
|
||||
@@ -326,3 +326,31 @@ func TestPickBindingWeightMultiplier(t *testing.T) {
|
||||
t.Fatalf("权重 100:1 下 boosted 命中 %d/200,应接近 200", boosted)
|
||||
}
|
||||
}
|
||||
|
||||
// MaxConcurrency 改配置后在旧槽排空时热生效,无需重启。
|
||||
func TestTryAcquireCapacityChange(t *testing.T) {
|
||||
db := newTestDB(t)
|
||||
id := seedChannel(t, db, "c", 1, 0, 1)
|
||||
var ch store.Channel
|
||||
db.First(&ch, id)
|
||||
s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890"))
|
||||
|
||||
release, ok := s.TryAcquire(&ch)
|
||||
if !ok {
|
||||
t.Fatal("容量 1 首次获取应成功")
|
||||
}
|
||||
// 容量 1 → 2:旧槽被占用,本次应沿用旧容量(拿不到)
|
||||
ch.MaxConcurrency = 2
|
||||
if _, ok := s.TryAcquire(&ch); ok {
|
||||
t.Fatal("旧槽占用期间不应扩容")
|
||||
}
|
||||
release()
|
||||
// 排空后再取:新容量生效,连取 2 个都成功
|
||||
r1, ok1 := s.TryAcquire(&ch)
|
||||
r2, ok2 := s.TryAcquire(&ch)
|
||||
if !ok1 || !ok2 {
|
||||
t.Fatal("排空后容量 2 应允许两个并发")
|
||||
}
|
||||
r1()
|
||||
r2()
|
||||
}
|
||||
Reference in new issue
Block a user