gozerosandCursor
bd427b57a1
HTTP 用 Gin，配置用 Viper，日志用 zap，数据用 sqlx；保留原有接口与会商行为。 顺带补上就绪探活、请求号、滚动日志、systemd 与 Dockerfile。 Co-authored-by: Cursor <cursoragent@cursor.com>
76 lines
1.6 KiB
Go
76 lines
1.6 KiB
Go
package worker
|
|
|
|
import (
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"git.gozeros.com/wsk/eightbf-analysis-go/internal/domain"
|
|
"git.gozeros.com/wsk/eightbf-analysis-go/internal/logging"
|
|
"git.gozeros.com/wsk/eightbf-analysis-go/internal/repo"
|
|
)
|
|
|
|
// LiveWorker 定时扫 queued，最多并发 MaxInflight 场交给 LiveChainRunner。
|
|
type LiveWorker struct {
|
|
Repo *repo.JobRepo
|
|
Runner *LiveChainRunner
|
|
PollEvery time.Duration
|
|
MaxInflight int32
|
|
inflight int32
|
|
}
|
|
|
|
func (w *LiveWorker) Start() {
|
|
if w.PollEvery <= 0 {
|
|
w.PollEvery = time.Second
|
|
}
|
|
go func() {
|
|
t := time.NewTicker(w.PollEvery)
|
|
defer t.Stop()
|
|
for range t.C {
|
|
w.poll()
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (w *LiveWorker) poll() {
|
|
max := w.MaxInflight
|
|
if max < 1 {
|
|
max = 2
|
|
}
|
|
if atomic.LoadInt32(&w.inflight) >= max {
|
|
return
|
|
}
|
|
queued, err := w.Repo.FindQueuedJobs()
|
|
if err != nil {
|
|
logging.L.Warnw("LiveWorker 扫队列失败", "err", err)
|
|
return
|
|
}
|
|
for _, job := range queued {
|
|
if atomic.LoadInt32(&w.inflight) >= max {
|
|
break
|
|
}
|
|
job.Status = domain.StatusRunning
|
|
job.Phase = "claimed"
|
|
if err := w.Repo.UpdateProgress(job); err != nil {
|
|
logging.L.Warnw("LiveWorker 占坑失败", logging.KeyJobID, job.ID, "err", err)
|
|
continue
|
|
}
|
|
atomic.AddInt32(&w.inflight, 1)
|
|
jobID := job.ID
|
|
go func() {
|
|
defer atomic.AddInt32(&w.inflight, -1)
|
|
fresh, err := w.Repo.FindByID(jobID)
|
|
if err != nil || fresh == nil {
|
|
return
|
|
}
|
|
logging.L.Infow("开始真模型会商",
|
|
logging.KeyJobID, jobID,
|
|
logging.KeyRequestID, fresh.RequestID,
|
|
logging.KeyTaskID, fresh.TaskID,
|
|
logging.KeyModel, fresh.Model,
|
|
logging.KeyStatus, fresh.Status,
|
|
)
|
|
w.Runner.Run(fresh)
|
|
}()
|
|
}
|
|
}
|