修复BUG: 1、当存在较多任务时,系统会发生卡顿,加载对话或者加载界面时候存在明显卡顿,优化性能找出问题
This commit is contained in:
1 parent
cf0989c67d
commit
594436fddd
12 files changed
+576
-140
No files matched your search
@@ -991,6 +991,74 @@ func (d *DB) LastActivityAll() (map[int64]int64, error) {
|
||||
// GoalCounts is the goal summary for one exploration.
|
||||
type GoalCounts struct{ Total, Met int }
|
||||
|
||||
// TaskListMetrics contains the aggregates rendered in task lists.
|
||||
type TaskListMetrics struct {
|
||||
Tokens TokenUsage
|
||||
LastActivity int64
|
||||
Goals GoalCounts
|
||||
}
|
||||
|
||||
// TaskListMetricsAll returns list aggregates for every live task in one query.
|
||||
// The lateral lookups use the per-exploration indexes instead of grouping the
|
||||
// complete activity history on every poll.
|
||||
func (d *DB) TaskListMetricsAll() (map[int64]TaskListMetrics, error) {
|
||||
rows, err := d.Query(`
|
||||
SELECT task.exploration_id,
|
||||
COALESCE(token_metrics.input_tokens,0),
|
||||
COALESCE(token_metrics.output_tokens,0),
|
||||
COALESCE(token_metrics.cache_read_tokens,0),
|
||||
COALESCE(token_metrics.cache_write_tokens,0),
|
||||
COALESCE(latest_activity.created_at,0),
|
||||
COALESCE(goal_metrics.total,0),
|
||||
COALESCE(goal_metrics.met,0)
|
||||
FROM tasks task
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT SUM(input_tokens) AS input_tokens,
|
||||
SUM(output_tokens) AS output_tokens,
|
||||
SUM(cache_read_tokens) AS cache_read_tokens,
|
||||
SUM(cache_write_tokens) AS cache_write_tokens
|
||||
FROM activity
|
||||
WHERE exploration_id=task.exploration_id AND kind='result'
|
||||
) token_metrics ON true
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT EXTRACT(EPOCH FROM created_at)::bigint AS created_at
|
||||
FROM activity
|
||||
WHERE exploration_id=task.exploration_id
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 1
|
||||
) latest_activity ON true
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT COUNT(*) AS total,
|
||||
COUNT(*) FILTER (WHERE state='met') AS met
|
||||
FROM exploration_nodes
|
||||
WHERE exploration_id=task.exploration_id AND kind='goal'
|
||||
) goal_metrics ON true
|
||||
WHERE task.deleted_at IS NULL`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[int64]TaskListMetrics{}
|
||||
for rows.Next() {
|
||||
var explorationID int64
|
||||
var metrics TaskListMetrics
|
||||
if err := rows.Scan(
|
||||
&explorationID,
|
||||
&metrics.Tokens.InputTokens,
|
||||
&metrics.Tokens.OutputTokens,
|
||||
&metrics.Tokens.CacheReadTokens,
|
||||
&metrics.Tokens.CacheWriteTokens,
|
||||
&metrics.LastActivity,
|
||||
&metrics.Goals.Total,
|
||||
&metrics.Goals.Met,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[explorationID] = metrics
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// GoalCountsAll returns goal totals (total / met) per exploration id, one query
|
||||
// for all tasks — used by listTasks so the task list shows progress for every task.
|
||||
func (d *DB) GoalCountsAll() (map[int64]GoalCounts, error) {
|
||||
|
||||
@@ -271,6 +271,13 @@ CREATE INDEX IF NOT EXISTS idx_act_since ON activity(exploration_id, id);
|
||||
-- Main/Plan history pages filter by worker (both carry NULL node_id, so idx_act_node
|
||||
-- can't distinguish them); this covers reverse pagination of those sessions.
|
||||
CREATE INDEX IF NOT EXISTS idx_act_worker ON activity(exploration_id, worker, id);
|
||||
-- Task-list polls aggregate result usage and find the latest event repeatedly.
|
||||
-- Cover the token columns for index-only aggregation and the timestamp order for
|
||||
-- per-exploration latest-activity lookups.
|
||||
CREATE INDEX IF NOT EXISTS idx_act_result_usage ON activity(exploration_id)
|
||||
INCLUDE (input_tokens, output_tokens, cache_read_tokens, cache_write_tokens)
|
||||
WHERE kind='result';
|
||||
CREATE INDEX IF NOT EXISTS idx_act_latest ON activity(exploration_id, created_at DESC);
|
||||
|
||||
-- =====================================================================
|
||||
-- C. LLM profiles
|
||||
|
||||
+163
-3
@@ -52,9 +52,6 @@ func (d *DB) hydrateTaskContext(t *Task) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
t.LLMProfileID = legacyProfileID
|
||||
t.ActiveLLMProfileID = activeProfileID
|
||||
t.LLMChainRevision = revision
|
||||
sources, err := d.TaskSourceIDs(t.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -65,8 +62,22 @@ func (d *DB) hydrateTaskContext(t *Task) error {
|
||||
return err
|
||||
}
|
||||
t.CompanyIDs = companies
|
||||
applyTaskLLMContext(t, legacyProfileID, activeProfileID, revision, chain)
|
||||
return nil
|
||||
}
|
||||
|
||||
func applyTaskLLMContext(
|
||||
t *Task,
|
||||
legacyProfileID, activeProfileID *int64,
|
||||
revision int64,
|
||||
chain []TaskLLMProfile,
|
||||
) {
|
||||
t.LLMProfileID = legacyProfileID
|
||||
t.ActiveLLMProfileID = activeProfileID
|
||||
t.LLMChainRevision = revision
|
||||
t.LLMProfileIDs = make([]int64, 0, len(chain))
|
||||
t.LLMFailoverState = "default"
|
||||
t.LLMFailoverReason = ""
|
||||
var latest *TaskLLMProfile
|
||||
activeReady := false
|
||||
for i := range chain {
|
||||
@@ -90,6 +101,155 @@ func (d *DB) hydrateTaskContext(t *Task) error {
|
||||
if latest != nil {
|
||||
t.LLMFailoverReason = latest.LastError
|
||||
}
|
||||
}
|
||||
|
||||
type taskBatchContext struct {
|
||||
legacyProfileID *int64
|
||||
activeProfileID *int64
|
||||
revision int64
|
||||
chain []TaskLLMProfile
|
||||
sourceIDs []int64
|
||||
companyIDs []int64
|
||||
}
|
||||
|
||||
// hydrateTasksContext loads every task's LLM chain, source tasks, and company
|
||||
// scopes with three bulk queries. ListTasks used to issue these queries once per
|
||||
// task, making startup and task-list hydration grow as 3N+1 database round trips.
|
||||
func (d *DB) hydrateTasksContext(tasks []*Task) error {
|
||||
if len(tasks) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
ids := make([]int64, 0, len(tasks))
|
||||
contexts := make(map[int64]*taskBatchContext, len(tasks))
|
||||
byID := make(map[int64]*Task, len(tasks))
|
||||
for _, task := range tasks {
|
||||
if task == nil {
|
||||
continue
|
||||
}
|
||||
ids = append(ids, task.ID)
|
||||
contexts[task.ID] = &taskBatchContext{}
|
||||
byID[task.ID] = task
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
rows, err := d.Query(`
|
||||
SELECT t.id, t.llm_profile_id, t.active_llm_profile_id, t.llm_chain_revision,
|
||||
p.profile_id, p.position, p.status, COALESCE(p.last_error,''), p.exhausted_at
|
||||
FROM tasks t
|
||||
LEFT JOIN task_llm_profiles p ON p.task_id=t.id
|
||||
WHERE t.id=ANY($1::bigint[])
|
||||
ORDER BY t.id, p.position NULLS LAST`, ids)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for rows.Next() {
|
||||
var taskID, revision int64
|
||||
var legacy, active, profileID, position sql.NullInt64
|
||||
var status, lastError sql.NullString
|
||||
var exhaustedAt sql.NullTime
|
||||
if err := rows.Scan(
|
||||
&taskID, &legacy, &active, &revision,
|
||||
&profileID, &position, &status, &lastError, &exhaustedAt,
|
||||
); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
context := contexts[taskID]
|
||||
if context == nil {
|
||||
continue
|
||||
}
|
||||
context.revision = revision
|
||||
if legacy.Valid {
|
||||
id := legacy.Int64
|
||||
context.legacyProfileID = &id
|
||||
}
|
||||
if active.Valid {
|
||||
id := active.Int64
|
||||
context.activeProfileID = &id
|
||||
}
|
||||
if profileID.Valid {
|
||||
entry := TaskLLMProfile{
|
||||
ProfileID: profileID.Int64,
|
||||
Position: int(position.Int64),
|
||||
Status: status.String,
|
||||
LastError: lastError.String,
|
||||
}
|
||||
if exhaustedAt.Valid {
|
||||
ts := exhaustedAt.Time
|
||||
entry.ExhaustedAt = &ts
|
||||
}
|
||||
context.chain = append(context.chain, entry)
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
rows.Close()
|
||||
|
||||
rows, err = d.Query(`
|
||||
SELECT task_id, source_task_id
|
||||
FROM task_relations
|
||||
WHERE task_id=ANY($1::bigint[])
|
||||
ORDER BY task_id, created_at, source_task_id`, ids)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for rows.Next() {
|
||||
var taskID, sourceID int64
|
||||
if err := rows.Scan(&taskID, &sourceID); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
if context := contexts[taskID]; context != nil {
|
||||
context.sourceIDs = append(context.sourceIDs, sourceID)
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
rows.Close()
|
||||
|
||||
rows, err = d.Query(`
|
||||
SELECT task_id, company_id
|
||||
FROM task_scope
|
||||
WHERE task_id=ANY($1::bigint[]) AND kind='company' AND company_id IS NOT NULL
|
||||
ORDER BY task_id, id`, ids)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for rows.Next() {
|
||||
var taskID, companyID int64
|
||||
if err := rows.Scan(&taskID, &companyID); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
if context := contexts[taskID]; context != nil {
|
||||
context.companyIDs = append(context.companyIDs, companyID)
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
rows.Close()
|
||||
return err
|
||||
}
|
||||
rows.Close()
|
||||
|
||||
for id, context := range contexts {
|
||||
task := byID[id]
|
||||
task.SourceTaskIDs = context.sourceIDs
|
||||
task.CompanyIDs = context.companyIDs
|
||||
applyTaskLLMContext(
|
||||
task,
|
||||
context.legacyProfileID,
|
||||
context.activeProfileID,
|
||||
context.revision,
|
||||
context.chain,
|
||||
)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
package db
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestListTasksBulkHydratesTaskContext(t *testing.T) {
|
||||
d, err := Open(testDSN(t))
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) - skipping", err)
|
||||
}
|
||||
defer d.Close()
|
||||
|
||||
stamp := time.Now().UnixNano()
|
||||
profileID, err := d.SaveProfile(&LLMProfile{
|
||||
Name: fmt.Sprintf("bulk-list-profile-%d", stamp), Format: "openai", Model: "test", APIKey: "test-key",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = d.DeleteProfile(profileID) })
|
||||
source, err := d.CreateTask(fmt.Sprintf("bulk-list-source-%d", stamp), "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = d.DeleteTask(source.ID) })
|
||||
companyID, _, err := d.Companies().UpsertCompany(fmt.Sprintf("Bulk List Company %d", stamp), "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = d.Companies().DeleteCompany(companyID) })
|
||||
child, err := d.CreateTaskWithOptions("bulk-list-child", "goal", TaskCreateOptions{
|
||||
LLMProfileIDs: []int64{profileID},
|
||||
SourceTaskIDs: []int64{source.ID},
|
||||
CompanyIDs: []int64{companyID},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = d.DeleteTask(child.ID) })
|
||||
|
||||
tasks, err := d.ListTasks()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got *Task
|
||||
for _, task := range tasks {
|
||||
if task.ID == child.ID {
|
||||
got = task
|
||||
break
|
||||
}
|
||||
}
|
||||
if got == nil {
|
||||
t.Fatalf("task %d missing from list", child.ID)
|
||||
}
|
||||
if len(got.LLMProfileIDs) != 1 || got.LLMProfileIDs[0] != profileID ||
|
||||
got.ActiveLLMProfileID == nil || *got.ActiveLLMProfileID != profileID || got.LLMFailoverState != "ready" {
|
||||
t.Fatalf("LLM context was not bulk hydrated: %+v", got)
|
||||
}
|
||||
if len(got.SourceTaskIDs) != 1 || got.SourceTaskIDs[0] != source.ID {
|
||||
t.Fatalf("source task context was not bulk hydrated: %v", got.SourceTaskIDs)
|
||||
}
|
||||
if len(got.CompanyIDs) != 1 || got.CompanyIDs[0] != companyID {
|
||||
t.Fatalf("company context was not bulk hydrated: %v", got.CompanyIDs)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskListMetricsAll(t *testing.T) {
|
||||
d, err := Open(testDSN(t))
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) - skipping", err)
|
||||
}
|
||||
defer d.Close()
|
||||
|
||||
task, err := d.CreateTask("task-list-metrics", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = d.DeleteTask(task.ID) })
|
||||
|
||||
resultAt := time.Now().Add(-time.Second).Truncate(time.Second)
|
||||
latestAt := resultAt.Add(time.Second)
|
||||
if _, err := d.Exec(`
|
||||
INSERT INTO activity(
|
||||
exploration_id, kind, input_tokens, output_tokens,
|
||||
cache_read_tokens, cache_write_tokens, created_at
|
||||
) VALUES ($1,'result',11,7,3,2,$2), ($1,'tool_result',999,999,999,999,$3)`,
|
||||
task.ExplorationID, resultAt, latestAt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := d.Exec(`
|
||||
INSERT INTO exploration_nodes(exploration_id, kind, payload, state)
|
||||
VALUES ($1,'goal','{}','met'), ($1,'goal','{}','open')`, task.ExplorationID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
all, err := d.TaskListMetricsAll()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
metrics, ok := all[task.ExplorationID]
|
||||
if !ok {
|
||||
t.Fatalf("metrics for exploration %d missing", task.ExplorationID)
|
||||
}
|
||||
if metrics.Tokens.InputTokens != 11 || metrics.Tokens.OutputTokens != 7 ||
|
||||
metrics.Tokens.CacheReadTokens != 3 || metrics.Tokens.CacheWriteTokens != 2 {
|
||||
t.Fatalf("unexpected token metrics: %+v", metrics.Tokens)
|
||||
}
|
||||
if metrics.LastActivity != latestAt.Unix() {
|
||||
t.Fatalf("last activity=%d, want %d", metrics.LastActivity, latestAt.Unix())
|
||||
}
|
||||
if metrics.Goals.Total != 2 || metrics.Goals.Met != 1 {
|
||||
t.Fatalf("unexpected goal metrics: %+v", metrics.Goals)
|
||||
}
|
||||
}
|
||||
+2
-4
@@ -377,10 +377,8 @@ ORDER BY (pinned_at IS NOT NULL) DESC, pinned_at DESC NULLS LAST, id DESC`)
|
||||
return nil, err
|
||||
}
|
||||
rows.Close()
|
||||
for _, task := range out {
|
||||
if err := d.hydrateTaskContext(task); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := d.hydrateTasksContext(out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user