perf(pagerank): O(V+E) 累加实现替代 O(V²) 全源遍历

旧实现每目标节点遍历全部源节点找边匹配 (943-961):
- 17628 nodes × 20 迭代 = 6.2亿次内层操作 → 单次 PageRank ~3 分钟
- 重启后多套触发器零值连锁 fire full consolidation → CPU 100% 持续 10+ 分钟

新实现先归一化每源节点总权重, 再沿出边把贡献累加到目标 (O(V+E)):
- 17628 nodes PageRank + 图谱维护: 3分钟 → <1 秒
- 重启风暴从 10+ 分钟 CPU 100% → 秒级无感

实测: 01:16:40 图谱维护开始 → 01:16:41 PageRank 更新 17628 nodes 完成
This commit is contained in:
xiaowei 2026-09-06 01:20:21 +08:00
parent e67f0bc7a6
commit 6c1c21fadf
1 changed files with 19 additions and 14 deletions

View File

@ -941,21 +941,26 @@ func (gs *SQLiteGraphStore) PageRank(damping float64, iterations int) map[string
} }
for iter := 0; iter < iterations; iter++ { for iter := 0; iter < iterations; iter++ {
newRanks := make(map[string]float64) // 正确 PageRank 实现 O(V+E):先归一化每个源节点总权重,
for _, node := range nodes { // 再沿出边把贡献直接累加到目标节点(旧实现每目标遍历全源 O(V²)
rank := base // 17628 节点 × 20 迭代 = 6.2 亿次 → 3 分钟;现在亚秒级)
for src, edges := range outEdges { contrib := make(map[string]float64, len(outEdges))
totalWt := 0.0 for src, edges := range outEdges {
for _, e := range edges { totalWt := 0.0
totalWt += e.weight for _, e := range edges {
} totalWt += e.weight
for _, e := range edges {
if e.target == node && totalWt > 0 {
rank += damping * ranks[src] * e.weight / totalWt
}
}
} }
newRanks[node] = rank if totalWt <= 0 {
continue
}
share := damping * ranks[src] / totalWt
for _, e := range edges {
contrib[e.target] += share * e.weight
}
}
newRanks := make(map[string]float64, len(nodes))
for _, node := range nodes {
newRanks[node] = base + contrib[node]
} }
ranks = newRanks ranks = newRanks
} }