perf(chat,update): пул воркеров per-user вместо сериальной очереди + HEAD-проба обновлений
- chat.Router: ограниченный пул chatWorkers=4 воркеров + FIFO-очереди per-user (userState/workerLoop/runUser). Порядок сообщений одного UserID сохраняется; разные пользователи обрабатываются параллельно (до 4 одновременных LLM-вызовов), long-poll Telegram не блокируется чужим аналитиком. Backpressure по jobs — только на перегруженного пользователя. - app.FreeChat: sessions под sync.Mutex (защита от data race при параллельных воркерах роутера). - update: ResolveLatest проверяет наличие бинаря HEAD-пробой без скачивания тела (fallback GET Range 0-0 при 405/501), сортировка версий по id убыв.; один общий http.Client (keep-alive) вместо нового на каждый запрос. - тесты: порядок/параллелизм per-user в router, HEAD-без-тела и фоллбэк на версию без бинаря в update. - память Serena: инварианты Router/update, примечания по форматированию на Windows.
This commit is contained in:
@@ -21,36 +21,90 @@ type Router struct {
|
||||
// Hook, вызываемый на каждое входящее событие (обычно → process_turn).
|
||||
onUserMsg func(Incoming)
|
||||
|
||||
// Асинхронная обработка входящих: handleIncoming кладёт событие в канал,
|
||||
// воркер-горутина последовательно вызывает onUserMsg. Благодаря этому
|
||||
// long-poll цикл канала (Telegram) не блокируется на время долгого
|
||||
// вызова аналитика и продолжает принимать новые сообщения.
|
||||
incoming chan Incoming
|
||||
// Асинхронная обработка входящих ограниченным пулом воркеров с
|
||||
// упорядоченными очередями per-user (см. chatWorkers, userState).
|
||||
// Благодаря этому long-poll цикл канала (Telegram) не блокируется на время
|
||||
// долгого вызова аналитика, а сообщения разных пользователей не сериализуются
|
||||
// друг за другом: каждый активный пользователь занимает своего воркера.
|
||||
jobs chan *userState
|
||||
users map[UserID]*userState
|
||||
userMu sync.Mutex
|
||||
|
||||
// processed — число обработанных воркером событий (для синхронизации
|
||||
// тестов с асинхронной очередью: WaitProcessed ждёт обработку события).
|
||||
processed atomic.Int64
|
||||
}
|
||||
|
||||
// userState — FIFO-очередь входящих одного пользователя. В каждый момент
|
||||
// для пользователя активен ровно один воркер (scheduled), поэтому порядок
|
||||
// обработки его сообщений сохраняется, а параллелизм достигается между
|
||||
// разными пользователями.
|
||||
type userState struct {
|
||||
mu sync.Mutex
|
||||
pending []Incoming
|
||||
scheduled bool
|
||||
}
|
||||
|
||||
// chatWorkers — число воркеров обработки входящих. Ограничивает количество
|
||||
// одновременных тяжёлых LLM-вызовов (аналитик/свободный чат), чтобы поток
|
||||
// каналов не упирался в один долгий вызов.
|
||||
const chatWorkers = 4
|
||||
|
||||
// NewRouter создаёт роутер. onUserMsg — колбэк обработки входящего.
|
||||
func NewRouter(onUserMsg func(Incoming)) *Router {
|
||||
if onUserMsg == nil {
|
||||
onUserMsg = func(Incoming) {}
|
||||
}
|
||||
r := &Router{
|
||||
sessions: map[UserID]any{},
|
||||
routes: map[UserID]Route{},
|
||||
pending: map[UserID]PendingQ{},
|
||||
sessions: map[UserID]any{},
|
||||
routes: map[UserID]Route{},
|
||||
pending: map[UserID]PendingQ{},
|
||||
onUserMsg: onUserMsg,
|
||||
incoming: make(chan Incoming, 256),
|
||||
jobs: make(chan *userState, chatWorkers),
|
||||
users: make(map[UserID]*userState),
|
||||
}
|
||||
for i := 0; i < chatWorkers; i++ {
|
||||
go r.workerLoop()
|
||||
}
|
||||
go r.processLoop()
|
||||
return r
|
||||
}
|
||||
|
||||
// processLoop — воркер асинхронной обработки входящих (FIFO).
|
||||
func (r *Router) processLoop() {
|
||||
for inc := range r.incoming {
|
||||
// userStateOf возвращает очередь пользователя (создаёт при первом сообщении).
|
||||
// Очереди живут вечно — по одной маленькой структуре на пользователя/вкладку.
|
||||
func (r *Router) userStateOf(uid UserID) *userState {
|
||||
r.userMu.Lock()
|
||||
defer r.userMu.Unlock()
|
||||
st, ok := r.users[uid]
|
||||
if !ok {
|
||||
st = &userState{}
|
||||
r.users[uid] = st
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
// workerLoop — воркер пула: берёт пользователя из общей очереди и дренит его.
|
||||
func (r *Router) workerLoop() {
|
||||
for st := range r.jobs {
|
||||
r.runUser(st)
|
||||
}
|
||||
}
|
||||
|
||||
// runUser обрабатывает все накопленные сообщения пользователя по порядку.
|
||||
// По исчерпании очереди снимает scheduled — следующий handleIncoming вновь
|
||||
// поставит пользователя в jobs. Возвращается в workerLoop, чтобы тот взял
|
||||
// следующего пользователя из общей очереди.
|
||||
func (r *Router) runUser(st *userState) {
|
||||
for {
|
||||
st.mu.Lock()
|
||||
if len(st.pending) == 0 {
|
||||
st.scheduled = false
|
||||
st.mu.Unlock()
|
||||
return
|
||||
}
|
||||
inc := st.pending[0]
|
||||
st.pending = st.pending[1:]
|
||||
st.mu.Unlock()
|
||||
|
||||
r.onUserMsg(inc)
|
||||
r.processed.Add(1)
|
||||
}
|
||||
@@ -99,9 +153,19 @@ func (r *Router) handleIncoming(inc Incoming) {
|
||||
}
|
||||
r.mu.Unlock()
|
||||
|
||||
// Асинхронная обработка: кладём событие в очередь воркера и сразу
|
||||
// возвращаемся, не блокируя вызывающий long-poll цикл канала.
|
||||
r.incoming <- inc
|
||||
// Асинхронная обработка: кладём событие в FIFO-очередь пользователя и
|
||||
// сразу возвращаемся, не блокируя вызывающий long-poll цикл канала.
|
||||
// Если пользователь ещё не обрабатывается — ставим его в общую очередь
|
||||
// пула воркеров. Backpressure по jobs блокирует только перегруженного
|
||||
// пользователя (его собственную горутину канала), не весь роутер.
|
||||
st := r.userStateOf(inc.UserID)
|
||||
st.mu.Lock()
|
||||
st.pending = append(st.pending, inc)
|
||||
if !st.scheduled {
|
||||
st.scheduled = true
|
||||
r.jobs <- st
|
||||
}
|
||||
st.mu.Unlock()
|
||||
}
|
||||
|
||||
// Send уведомляет пользователя через текущий маршрут. M1 (нет маршрута) — no-op,
|
||||
|
||||
@@ -9,8 +9,8 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
uidA UserID = "u-a"
|
||||
uidB UserID = "u-b"
|
||||
uidA UserID = "u-a"
|
||||
uidB UserID = "u-b"
|
||||
tg Address = "tg://123"
|
||||
tui Address = "tui://local"
|
||||
)
|
||||
@@ -213,3 +213,105 @@ func TestRouter_Ask_PendingNotConsumedFromOtherAddr(t *testing.T) {
|
||||
t.Fatal("pending должен остаться (ответ из другого адреса)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRouter_PerUserOrdering проверяет, что сообщения одного пользователя
|
||||
// обрабатываются строго в порядке поступления (пул воркеров не перемешивает).
|
||||
func TestRouter_PerUserOrdering(t *testing.T) {
|
||||
cb := newFakeOnMsg()
|
||||
r := NewRouter(cb.h)
|
||||
tgCh := newFakeChannel(tg)
|
||||
_ = r.Attach(tgCh)
|
||||
|
||||
for _, txt := range []string{"1", "2", "3"} {
|
||||
tgCh.emit(uidA, tg, txt)
|
||||
}
|
||||
if !cb.wait(3) {
|
||||
t.Fatal("сообщения не обработаны за таймаут")
|
||||
}
|
||||
for i, want := range []string{"1", "2", "3"} {
|
||||
if got := cb.get(i).Msg.Text; got != want {
|
||||
t.Errorf("порядок обработки нарушен: idx %d = %q, want %q", i, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestRouter_ParallelismAcrossUsers проверяет, что пока обработчик одного
|
||||
// пользователя заблокирован (долгий LLM-вызов), сообщение другого пользователя
|
||||
// обрабатывается в другом воркере, а второе сообщение того же пользователя —
|
||||
// ждёт своей очереди (per-user порядок).
|
||||
func TestRouter_ParallelismAcrossUsers(t *testing.T) {
|
||||
r := NewRouter(nil)
|
||||
|
||||
tgCh := newFakeChannel(tg)
|
||||
tuiCh := newFakeChannel(tui)
|
||||
_ = r.Attach(tgCh)
|
||||
_ = r.Attach(tuiCh)
|
||||
|
||||
blocked := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
var muLocal sync.Mutex
|
||||
seen := make([]string, 0, 3)
|
||||
signal := make(chan struct{}, 8)
|
||||
|
||||
h := func(inc Incoming) {
|
||||
if inc.Msg.Text == "block" {
|
||||
close(blocked)
|
||||
<-release // держим воркера, пока не отпустим
|
||||
}
|
||||
muLocal.Lock()
|
||||
seen = append(seen, inc.Msg.Text)
|
||||
muLocal.Unlock()
|
||||
signal <- struct{}{}
|
||||
}
|
||||
r.onUserMsg = h
|
||||
|
||||
snapshot := func() []string {
|
||||
muLocal.Lock()
|
||||
defer muLocal.Unlock()
|
||||
return append([]string(nil), seen...)
|
||||
}
|
||||
waitFor := func(n int) bool {
|
||||
deadline := time.After(2 * time.Second)
|
||||
for len(snapshot()) < n {
|
||||
select {
|
||||
case <-signal:
|
||||
case <-deadline:
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// первое сообщение A блокирует своего воркера
|
||||
tgCh.emit(uidA, tg, "block")
|
||||
<-blocked
|
||||
|
||||
// B обрабатывается параллельно, пока A висит
|
||||
tuiCh.emit(uidB, tui, "B1")
|
||||
select {
|
||||
case <-signal:
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Fatal("B не обработан, пока блокирован A — чат-путь снова сериализован")
|
||||
}
|
||||
if got := snapshot(); len(got) != 1 || got[0] != "B1" {
|
||||
t.Fatalf("ожидали обработку B1, got %v", got)
|
||||
}
|
||||
|
||||
// второе сообщение A НЕ обрабатывается, пока занят воркер A (порядок per-user)
|
||||
tgCh.emit(uidA, tg, "a2")
|
||||
select {
|
||||
case <-signal:
|
||||
t.Fatal("сообщение A обработано ДО освобождения A — нарушен per-user порядок")
|
||||
case <-time.After(80 * time.Millisecond):
|
||||
}
|
||||
|
||||
// отпускаем A → дообрабатывается a2
|
||||
close(release)
|
||||
if !waitFor(3) {
|
||||
t.Fatal("итоговые сообщения не обработаны")
|
||||
}
|
||||
got := snapshot()
|
||||
if got[2] != "a2" {
|
||||
t.Errorf("порядок персональной очереди нарушен: pos2 = %q, want a2", got[2])
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user