From bb20ebbed0ababb68d21008135c893195c3b53d7 Mon Sep 17 00:00:00 2001
From: Sakurasan <26715255+Sakurasan@users.noreply.github.com>
Date: Fri, 30 Oct 2026 01:42:22 +0800
Subject: [PATCH] =?UTF-8?q?perf:=20=E6=B8=A0=E9=81=93=20MaxConcurrency=20?=
 =?UTF-8?q?=E6=94=B9=E9=85=8D=E7=BD=AE=E5=90=8E=E6=8E=92=E7=A9=BA=E7=83=AD?=
 =?UTF-8?q?=E7=94=9F=E6=95=88=EF=BC=8C=E6=97=A0=E9=9C=80=E9=87=8D=E5=90=AF?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

信号量条目带上容量，TryAcquire 发现容量变更且旧槽已排空时就地换新。
release 闭包持有旧 chan，换新后在途请求释放进旧槽是无害的。
---
 server/internal/channel/channel.go      | 23 ++++++++++++++----
 server/internal/channel/channel_test.go | 32 +++++++++++++++++++++++--
 2 files changed, 48 insertions(+), 7 deletions(-)

diff --git a/server/internal/channel/channel.go b/server/internal/channel/channel.go
index 0a90cee..ea95101 100644
--- a/server/internal/channel/channel.go
+++ b/server/internal/channel/channel.go
@@ -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{}{}:
diff --git a/server/internal/channel/channel_test.go b/server/internal/channel/channel_test.go
index d8bc382..6911d4c 100644
--- a/server/internal/channel/channel_test.go
+++ b/server/internal/channel/channel_test.go
@@ -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()
+}