From f3c339b2f5c9c2e2f4f2efdee9fa91d9ef51c69e Mon Sep 17 00:00:00 2001 From: imbajin Date: Fri, 6 Nov 2026 01:22:01 +0800 Subject: [PATCH 1/3] fix(vermeer): make betweenness memory- and step-safe Rebase the completed Brandes batch implementation onto the current master after the final-output lifecycle and preceding Vermeer fixes were merged upstream. --- vermeer/algorithms/betweenness_centrality.go | 600 +++++++++++++----- .../algorithms/betweenness_centrality_test.go | 271 ++++++++ 2 files changed, 724 insertions(+), 147 deletions(-) create mode 100644 vermeer/algorithms/betweenness_centrality_test.go diff --git a/vermeer/algorithms/betweenness_centrality.go b/vermeer/algorithms/betweenness_centrality.go index f8b28535c..093586b1b 100644 --- a/vermeer/algorithms/betweenness_centrality.go +++ b/vermeer/algorithms/betweenness_centrality.go @@ -19,7 +19,7 @@ package algorithms import ( "fmt" - "math/rand" + "math" "vermeer/apps/common" "vermeer/apps/compute" "vermeer/apps/options" @@ -28,6 +28,14 @@ import ( "github.com/sirupsen/logrus" ) +const ( + betweennessStopValue = "stop_sum" + betweennessLocalCountsValue = "betweenness_local_counts" + betweennessBatchCountValue = "betweenness_batch_count" + betweennessPartialValue = "betweenness_partial_values" + betweennessFinalValue = "betweenness_final_values" +) + func init() { Algorithms = append(Algorithms, new(BetweennessCentralityMaker)) } @@ -52,176 +60,405 @@ func (bc *BetweennessCentralityMaker) DataNeeded() []string { return []string{} } -type BetweennessCentralityWorker struct { - compute.WorkerComputerBase - useEndPoint bool - stopFlag bool - scale serialize.SFloat32 - sampleRate float64 - stopSum []serialize.SInt32 - betweennessValues []serialize.SFloat32 - tempEdges [][]serialize.SUint32 - neighborEdges []serialize.SliceUint32 - betweennessValueMap []serialize.MapUint32Float32 - bfsNewValues []map[uint32][][]uint32 - bfsOldValues []map[uint32][][]uint32 -} - -func (bcw *BetweennessCentralityWorker) VertexValue(i uint32) serialize.MarshalAble { - if bcw.WContext.Step == 1 { - return &bcw.neighborEdges[i] +func validateBetweennessSampleRate(value float64) error { + if math.IsNaN(value) || math.IsInf(value, 0) || value <= 0 || value > 1 { + return fmt.Errorf("betweenness_centrality.sample_rate must be finite and in (0, 1], got %v", + value) } - if bcw.stopFlag && bcw.WContext.Step > 1 { - return &bcw.betweennessValues[i] + return nil +} + +func parseBetweennessEndpoint(value int) (bool, error) { + switch value { + case 0: + return false, nil + case 1: + return true, nil + default: + return false, fmt.Errorf("betweenness_centrality.use_endpoint must be 0 or 1, got %d", + value) } - return &bcw.betweennessValueMap[i] } -func (bcw *BetweennessCentralityWorker) Init() error { - bcw.betweennessValues = make([]serialize.SFloat32, bcw.WContext.GraphData.Vertex.TotalVertexCount()) - bcw.neighborEdges = make([]serialize.SliceUint32, bcw.WContext.GraphData.Vertex.TotalVertexCount()) - bcw.betweennessValueMap = make([]serialize.MapUint32Float32, bcw.WContext.GraphData.Vertex.TotalVertexCount()) - for i := range bcw.betweennessValueMap { - bcw.betweennessValueMap[i] = make(map[serialize.SUint32]serialize.SFloat32) - } - bcw.tempEdges = make([][]serialize.SUint32, bcw.WContext.Parallel) - for i := range bcw.tempEdges { - bcw.tempEdges[i] = make([]serialize.SUint32, 0, 100) - } - bcw.bfsNewValues = make([]map[uint32][][]uint32, bcw.WContext.GraphData.VertexCount) - bcw.bfsOldValues = make([]map[uint32][][]uint32, bcw.WContext.GraphData.VertexCount) - bcw.stopSum = make([]serialize.SInt32, bcw.WContext.Parallel) - bcw.sampleRate = options.GetFloat(bcw.WContext.Params, "betweenness_centrality.sample_rate") - if bcw.sampleRate > 1.0 || bcw.sampleRate <= 0 { - logrus.Errorf("The param betweenness_centrality.sample_rate must be in (0.0, 1.0], actual got '%v", - bcw.sampleRate) - return fmt.Errorf("the param betweenness_centrality.sample_rate must be in (0.0, 1.0], actual got '%v", - bcw.sampleRate) - } - bcw.useEndPoint = false - if options.GetInt(bcw.WContext.Params, "betweenness_centrality.use_endpoint") == 1 { - bcw.useEndPoint = true - } - bcw.stopFlag = false - - n := serialize.SFloat32(bcw.WContext.GraphData.Vertex.TotalVertexCount()) - if n < 2 { - bcw.scale = 1 - } else if bcw.useEndPoint { - n *= serialize.SFloat32(bcw.sampleRate) - bcw.scale = 1 / serialize.SFloat32(n*(n-1)) +func betweennessScale(totalVertices uint32, useEndpoint bool, sampleRate float64) float64 { + if totalVertices <= 1 { + return 0 + } + n := float64(totalVertices) + var scale float64 + if useEndpoint { + scale = 1 / (n * (n - 1)) } else { - n *= serialize.SFloat32(bcw.sampleRate) - bcw.scale = 1 / serialize.SFloat32((n-1)*(n-2)) + if totalVertices <= 2 { + return 0 + } + scale = 1 / ((n - 1) * (n - 2)) } + return scale / sampleRate +} - bcw.WContext.CreateValue("stop_sum", compute.ValueTypeInt32, compute.CValueActionAggregate) - bcw.WContext.SetValue("stop_sum", serialize.SInt32(0)) - bcw.WContext.CreateValue("stop_flag", compute.ValueTypeInt32, compute.CValueActionAggregate) - bcw.WContext.SetValue("stop_flag", serialize.SInt32(0)) - return nil +func planBetweennessBatches(maxLocalVertices uint32, parallel, maxStep int) (int, error) { + if maxLocalVertices == 0 { + return 0, nil + } + if maxStep < 2 { + return 0, fmt.Errorf("betweenness_centrality requires compute.max_step >= 2, got %d", + maxStep) + } + if parallel <= 0 { + parallel = 1 + } + ideal := (uint64(maxLocalVertices) + uint64(parallel) - 1) / uint64(parallel) + available := uint64(maxStep - 1) + if ideal > available { + ideal = available + } + return int(ideal), nil } -func (bcw *BetweennessCentralityWorker) BeforeStep() { - for i := range bcw.stopSum { - bcw.stopSum[i] = 0 - } - if bcw.WContext.Step > 1 && bcw.WContext.GetValue("stop_flag").(serialize.SInt32) == 1 || - options.GetInt(bcw.WContext.Params, "compute.max_step") == int(bcw.WContext.Step) { - bcw.stopFlag = true - } - for i := range bcw.bfsNewValues { - bcw.bfsOldValues[i] = make(map[uint32][][]uint32) - for k, v := range bcw.bfsNewValues[i] { - bcw.bfsOldValues[i][k] = v - } - bcw.bfsNewValues[i] = make(map[uint32][][]uint32) - } - - if bcw.stopFlag { - rangeStart := bcw.WContext.GraphData.VertIDStart - rangeEnd := bcw.WContext.GraphData.VertIDStart + bcw.WContext.GraphData.VertexCount - for _, valueMap := range bcw.betweennessValueMap { - for k, v := range valueMap { - key := uint32(k) - if key >= rangeStart && key < rangeEnd { - bcw.betweennessValues[key] += v - } - } +func collectBetweennessPartitionCounts( + workerValues map[string]map[string]*compute.CValue) (serialize.SliceInt64, error) { + counts := make(serialize.SliceInt64, 0, len(workerValues)) + for workerName, values := range workerValues { + cValue, ok := values[betweennessLocalCountsValue] + if !ok || cValue.Value == nil { + return nil, fmt.Errorf("worker %s did not report its betweenness partition size", + workerName) } + count, ok := cValue.Value.(serialize.SInt64) + if !ok { + return nil, fmt.Errorf("worker %s reported invalid betweenness partition type %T", + workerName, cValue.Value) + } + counts = append(counts, count) } + return counts, nil +} +func summarizeBetweennessPartitions(counts serialize.SliceInt64) (uint32, uint32, error) { + var maxLocal uint64 + var total uint64 + for _, rawCount := range counts { + count := int64(rawCount) + if count < 0 { + return 0, 0, fmt.Errorf("negative betweenness partition size %d", count) + } + value := uint64(count) + total += value + if value > maxLocal { + maxLocal = value + } + } + if total > uint64(^uint32(0)) || maxLocal > uint64(^uint32(0)) { + return 0, 0, fmt.Errorf("betweenness partition sizes exceed uint32 vertex range") + } + return uint32(maxLocal), uint32(total), nil } -func (bcw *BetweennessCentralityWorker) Compute(vertexID uint32, pID int) { - if bcw.stopFlag { - return +func sampleBetweennessSource(vertexID uint32, sampleRate float64) bool { + if sampleRate >= 1 { + return true } - vertID := vertexID - bcw.WContext.GraphData.VertIDStart - switch bcw.WContext.Step { - case 1: - inEdges := bcw.WContext.GraphData.Edges.GetInEdges(vertID) - bcw.neighborEdges[vertexID] = make([]serialize.SUint32, 0, len(inEdges)) - for _, edge := range inEdges { - if !bcw.sample() || uint32(edge) == vertexID { + value := uint64(vertexID) + 0x9e3779b97f4a7c15 + value = (value ^ (value >> 30)) * 0xbf58476d1ce4e5b9 + value = (value ^ (value >> 27)) * 0x94d049bb133111eb + value ^= value >> 31 + return float64(value>>11) < sampleRate*float64(uint64(1)<<53) +} + +func logAddExp(left, right float64) float64 { + if math.IsInf(left, -1) { + return right + } + if math.IsInf(right, -1) { + return left + } + if left < right { + left, right = right, left + } + return left + math.Log1p(math.Exp(right-left)) +} + +func brandesContributions(source uint32, neighbors []serialize.SliceUint32, + useEndpoint bool, scale float64) map[uint32]float64 { + result := make(map[uint32]float64) + if scale == 0 || int(source) >= len(neighbors) { + return result + } + + predecessors := make(map[uint32][]uint32) + logSigma := map[uint32]float64{source: 0} + distance := map[uint32]int{source: 0} + queue := []uint32{source} + + for head := 0; head < len(queue); head++ { + vertex := queue[head] + nextDistance := distance[vertex] + 1 + for _, rawNeighbor := range neighbors[vertex] { + neighbor := uint32(rawNeighbor) + if int(neighbor) >= len(neighbors) { continue } - bcw.neighborEdges[vertexID] = append(bcw.neighborEdges[vertexID], edge) - bcw.bfsNewValues[vertID][uint32(edge)] = [][]uint32{} - bcw.betweennessValueMap[vertexID][edge] = 0 - if bcw.useEndPoint { - bcw.betweennessValueMap[vertexID][edge] += 1 * bcw.scale + neighborDistance, seen := distance[neighbor] + if !seen { + neighborDistance = nextDistance + distance[neighbor] = neighborDistance + queue = append(queue, neighbor) } - } - bcw.stopSum[pID] = 1 - default: - for k, v := range bcw.bfsOldValues[vertID] { - for _, nId := range bcw.neighborEdges[k] { - if !bcw.sample() || nId == serialize.SUint32(vertexID) { - continue - } - if _, ok := bcw.betweennessValueMap[vertexID][nId]; !ok { - bcw.stopSum[pID] = 1 - bcw.tempEdges[pID] = append(bcw.tempEdges[pID], nId) - for _, vi := range v { - newV := append(vi, k) - bcw.bfsNewValues[vertID][uint32(nId)] = append(bcw.bfsNewValues[vertID][uint32(nId)], newV) - } - if len(v) == 0 { - bcw.bfsNewValues[vertID][uint32(nId)] = append(bcw.bfsNewValues[vertID][uint32(nId)], []uint32{k}) - } - } + if neighborDistance != nextDistance { + continue } - } - for _, edge := range bcw.tempEdges[pID] { - bcw.betweennessValueMap[vertexID][edge] = 0 - if bcw.useEndPoint { - bcw.betweennessValueMap[vertexID][edge] += 1 * bcw.scale + currentLogSigma, ok := logSigma[neighbor] + if !ok { + currentLogSigma = math.Inf(-1) } + logSigma[neighbor] = logAddExp(currentLogSigma, logSigma[vertex]) + predecessors[neighbor] = append(predecessors[neighbor], vertex) + } + } + + dependency := make(map[uint32]float64, len(queue)) + if useEndpoint && len(queue) > 1 { + result[source] = float64(len(queue)-1) * scale + } + for index := len(queue) - 1; index >= 0; index-- { + vertex := queue[index] + for _, predecessor := range predecessors[vertex] { + pathRatio := math.Exp(logSigma[predecessor] - logSigma[vertex]) + dependency[predecessor] += pathRatio * (1 + dependency[vertex]) + } + if vertex == source { + continue + } + contribution := dependency[vertex] + if useEndpoint { + contribution++ } + contribution *= scale + if contribution != 0 { + result[vertex] = contribution + } + } + return result +} + +func sumBetweennessPartials(partials []serialize.SliceFloat32, + totalVertices uint32) (serialize.SliceFloat32, error) { + if totalVertices == 0 { + return serialize.SliceFloat32{}, nil } - bcw.tempEdges[pID] = bcw.tempEdges[pID][0:0] - for _, values := range bcw.bfsNewValues[vertID] { - for _, v := range values { - for _, u := range v { - bcw.betweennessValueMap[vertexID][serialize.SUint32(u)] += - 1.0 / serialize.SFloat32(len(values)) * bcw.scale + if len(partials) == 0 { + return nil, fmt.Errorf("no betweenness worker partials were reported") + } + + sums := make([]float64, totalVertices) + for workerIndex, partial := range partials { + if uint32(len(partial)) != totalVertices { + return nil, fmt.Errorf("betweenness worker %d reported %d values, expected %d", + workerIndex, len(partial), totalVertices) + } + for vertexID, value := range partial { + sums[vertexID] += float64(value) + } + } + + result := make(serialize.SliceFloat32, totalVertices) + for vertexID, value := range sums { + if math.IsNaN(value) || math.IsInf(value, 0) { + return nil, fmt.Errorf("betweenness result for vertex %d is not finite", vertexID) + } + result[vertexID] = serialize.SFloat32(value) + } + return result, nil +} + +type emptyBetweennessVertexValue struct{} + +func (*emptyBetweennessVertexValue) Marshal([]byte) (int, error) { + return 0, nil +} + +func (*emptyBetweennessVertexValue) Unmarshal([]byte) (int, error) { + return 0, nil +} + +func (*emptyBetweennessVertexValue) ToString() string { + return "" +} + +func (*emptyBetweennessVertexValue) PredictSize() int { + return 0 +} + +var noBetweennessVertexValue emptyBetweennessVertexValue + +type BetweennessCentralityWorker struct { + compute.WorkerComputerBase + useEndPoint bool + scale float64 + sampleRate float64 + maxStep int + batchCount int + currentBatch int + totalVertices uint32 + betweennessValues serialize.SliceFloat32 + neighborEdges []serialize.SliceUint32 + localValues []float64 + batchValues []map[uint32]float64 +} + +func (bcw *BetweennessCentralityWorker) VertexValue(vertexID uint32) serialize.MarshalAble { + if bcw.WContext.Output { + return &bcw.betweennessValues[vertexID] + } + if bcw.WContext.Step == 1 { + return &bcw.neighborEdges[vertexID] + } + // The framework still visits every local vertex while scattering. Returning a + // zero-byte value keeps the per-batch wire payload constant after adjacency + // has been exchanged in step 1. + return &noBetweennessVertexValue +} + +func (bcw *BetweennessCentralityWorker) ValidateOutput() error { + cValue, ok := bcw.WContext.CValues[betweennessFinalValue] + if !ok || cValue.Value == nil { + return fmt.Errorf("betweenness_centrality output is missing the global result") + } + values, ok := cValue.Value.(serialize.SliceFloat32) + if !ok { + return fmt.Errorf("betweenness_centrality output has invalid global result type %T", + cValue.Value) + } + if uint32(len(values)) != bcw.totalVertices { + return fmt.Errorf("betweenness_centrality did not finish all source batches: got %d global values, expected %d; increase compute.max_step", + len(values), bcw.totalVertices) + } + bcw.betweennessValues = values + return nil +} + +func (bcw *BetweennessCentralityWorker) Init() error { + bcw.sampleRate = options.GetFloat(bcw.WContext.Params, + "betweenness_centrality.sample_rate") + if err := validateBetweennessSampleRate(bcw.sampleRate); err != nil { + return err + } + useEndpoint, err := parseBetweennessEndpoint(options.GetInt(bcw.WContext.Params, + "betweenness_centrality.use_endpoint")) + if err != nil { + return err + } + bcw.maxStep = options.GetInt(bcw.WContext.Params, "compute.max_step") + if bcw.maxStep < 2 { + return fmt.Errorf("betweenness_centrality requires compute.max_step >= 2, got %d", + bcw.maxStep) + } + + bcw.useEndPoint = useEndpoint + bcw.totalVertices = bcw.WContext.GraphData.Vertex.TotalVertexCount() + localVertices := bcw.WContext.GraphData.VertexCount + bcw.scale = betweennessScale(bcw.totalVertices, bcw.useEndPoint, bcw.sampleRate) + bcw.betweennessValues = make(serialize.SliceFloat32, bcw.totalVertices) + bcw.neighborEdges = make([]serialize.SliceUint32, bcw.totalVertices) + bcw.localValues = make([]float64, bcw.totalVertices) + bcw.batchValues = make([]map[uint32]float64, bcw.WContext.Parallel) + + bcw.WContext.CreateValue(betweennessStopValue, compute.ValueTypeInt32, + compute.CValueActionAggregate) + bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1)) + // This value is intentionally Normal rather than Aggregate. Aggregate values + // are broadcast back to workers; feeding the global slice into the next round + // would multiply it by the worker count on every superstep. + bcw.WContext.CreateValue(betweennessLocalCountsValue, compute.ValueTypeInt64, + compute.CValueActionNormal) + bcw.WContext.SetValue(betweennessLocalCountsValue, serialize.SInt64(localVertices)) + bcw.WContext.CreateValue(betweennessBatchCountValue, compute.ValueTypeInt32, + compute.CValueActionNormal) + bcw.WContext.SetValue(betweennessBatchCountValue, serialize.SInt32(0)) + bcw.WContext.CreateValue(betweennessPartialValue, compute.ValueTypeSliceFloat32, + compute.CValueActionNormal) + bcw.WContext.SetValue(betweennessPartialValue, serialize.SliceFloat32{}) + bcw.WContext.CreateValue(betweennessFinalValue, compute.ValueTypeSliceFloat32, + compute.CValueActionNormal) + bcw.WContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + return nil +} + +func (bcw *BetweennessCentralityWorker) BeforeStep() { + if bcw.WContext.Step < 2 { + return + } + bcw.batchCount = int(bcw.WContext.GetValue(betweennessBatchCountValue).(serialize.SInt32)) + bcw.currentBatch = int(bcw.WContext.Step) - 2 + for partitionID := range bcw.batchValues { + bcw.batchValues[partitionID] = make(map[uint32]float64) + } +} + +func (bcw *BetweennessCentralityWorker) Compute(vertexID uint32, partitionID int) { + localID := vertexID - bcw.WContext.GraphData.VertIDStart + if bcw.WContext.Step == 1 { + inEdges := bcw.WContext.GraphData.Edges.GetInEdges(localID) + neighbors := make(serialize.SliceUint32, 0, len(inEdges)) + seen := make(map[serialize.SUint32]struct{}, len(inEdges)) + for _, neighbor := range inEdges { + if uint32(neighbor) == vertexID { + continue } + if _, exists := seen[neighbor]; exists { + continue + } + seen[neighbor] = struct{}{} + neighbors = append(neighbors, neighbor) } + bcw.neighborEdges[vertexID] = neighbors + return } + if bcw.batchCount <= 0 || bcw.currentBatch >= bcw.batchCount || + int(localID)%bcw.batchCount != bcw.currentBatch { + return + } + if !sampleBetweennessSource(vertexID, bcw.sampleRate) { + return + } + + contributions := brandesContributions(vertexID, bcw.neighborEdges, + bcw.useEndPoint, bcw.scale) + batch := bcw.batchValues[partitionID] + for target, value := range contributions { + batch[target] += value + } } func (bcw *BetweennessCentralityWorker) AfterStep() { - var stopSum serialize.SInt32 - for i := range bcw.stopSum { - stopSum += bcw.stopSum[i] + if bcw.WContext.Step == 1 { + bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1)) + return + } + + bcw.aggregateCurrentBatch() + if bcw.batchCount <= 0 || bcw.currentBatch+1 < bcw.batchCount { + bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1)) + return + } + + partial := make(serialize.SliceFloat32, bcw.totalVertices) + for vertexID, value := range bcw.localValues { + partial[vertexID] = serialize.SFloat32(value) } - bcw.WContext.SetValue("stop_sum", stopSum) + bcw.WContext.SetValue(betweennessPartialValue, partial) + bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(0)) } -func (bcw *BetweennessCentralityWorker) sample() bool { - return rand.Float64() < bcw.sampleRate +func (bcw *BetweennessCentralityWorker) aggregateCurrentBatch() { + for partitionID, values := range bcw.batchValues { + for target, value := range values { + bcw.localValues[target] += value + } + bcw.batchValues[partitionID] = nil + } } func (bcw *BetweennessCentralityWorker) OutputValueType() string { @@ -230,17 +467,86 @@ func (bcw *BetweennessCentralityWorker) OutputValueType() string { type BetweennessCentralityMaster struct { compute.MasterComputerBase + planned bool + totalVertices uint32 + batchCount int +} + +func (bcm *BetweennessCentralityMaster) Init() error { + maxStep := options.GetInt(bcm.MContext.Params, "compute.max_step") + if maxStep < 2 { + return fmt.Errorf("betweenness_centrality requires compute.max_step >= 2, got %d", + maxStep) + } + bcm.MContext.CreateValue(betweennessBatchCountValue, compute.ValueTypeInt32, + compute.CValueActionNormal) + bcm.MContext.SetValue(betweennessBatchCountValue, serialize.SInt32(0)) + bcm.MContext.CreateValue(betweennessFinalValue, compute.ValueTypeSliceFloat32, + compute.CValueActionNormal) + bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + return nil } func (bcm *BetweennessCentralityMaster) Compute() bool { - logrus.Infof("stopFlag:%v", bcm.MContext.GetValue("stop_flag").(serialize.SInt32)) - if bcm.MContext.GetValue("stop_flag").(serialize.SInt32) == serialize.SInt32(len(bcm.MContext.WorkerCValues)) { - return false + if !bcm.planned { + counts, err := collectBetweennessPartitionCounts(bcm.MContext.WorkerCValues) + if err != nil { + logrus.Errorf("collect betweenness partition sizes: %v", err) + return false + } + maxLocal, totalVertices, err := summarizeBetweennessPartitions(counts) + if err != nil { + logrus.Errorf("invalid betweenness partition summary: %v", err) + return false + } + bcm.totalVertices = totalVertices + bcm.batchCount, err = planBetweennessBatches(maxLocal, + options.GetInt(bcm.MContext.Params, "compute.parallel"), + options.GetInt(bcm.MContext.Params, "compute.max_step")) + if err != nil { + logrus.Errorf("plan betweenness batches: %v", err) + return false + } + bcm.MContext.SetValue(betweennessBatchCountValue, + serialize.SInt32(bcm.batchCount)) + bcm.planned = true + logrus.Infof("betweenness source batches: %d, largest partition: %d", + bcm.batchCount, maxLocal) + if bcm.totalVertices == 0 { + bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + return false + } + return true + } + + stopSum := bcm.MContext.GetValue(betweennessStopValue).(serialize.SInt32) + if stopSum != 0 { + return true + } + + partials := make([]serialize.SliceFloat32, 0, len(bcm.MContext.WorkerCValues)) + for workerName, values := range bcm.MContext.WorkerCValues { + cValue, ok := values[betweennessPartialValue] + if !ok { + logrus.Errorf("worker %s did not report betweenness partial values", workerName) + bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + return false + } + partial, ok := cValue.Value.(serialize.SliceFloat32) + if !ok { + logrus.Errorf("worker %s reported invalid betweenness partial type %T", + workerName, cValue.Value) + bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + return false + } + partials = append(partials, partial) } - stopSum := bcm.MContext.GetValue("stop_sum") - logrus.Infof("stopSum:%v", stopSum) - if stopSum.(serialize.SInt32) == serialize.SInt32(0) { - bcm.MContext.SetValue("stop_flag", serialize.SInt32(1)) + result, err := sumBetweennessPartials(partials, bcm.totalVertices) + if err != nil { + logrus.Errorf("aggregate betweenness partial values: %v", err) + bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + return false } - return true + bcm.MContext.SetValue(betweennessFinalValue, result) + return false } diff --git a/vermeer/algorithms/betweenness_centrality_test.go b/vermeer/algorithms/betweenness_centrality_test.go new file mode 100644 index 000000000..7c593a45b --- /dev/null +++ b/vermeer/algorithms/betweenness_centrality_test.go @@ -0,0 +1,271 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with this +work for additional information regarding copyright ownership. The ASF +licenses this file to You under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package algorithms + +import ( + "math" + "testing" + "vermeer/apps/buffer" + "vermeer/apps/compute" + "vermeer/apps/serialize" +) + +func TestBrandesBetweennessPath(t *testing.T) { + neighbors := []serialize.SliceUint32{{1}, {0, 2}, {1}} + scale := betweennessScale(3, false, 1) + values := make([]float64, 3) + for source := uint32(0); source < 3; source++ { + for vertex, value := range brandesContributions(source, neighbors, false, scale) { + values[vertex] += value + } + } + if math.Abs(values[1]-1) > 1e-6 { + t.Fatalf("middle vertex expected normalized betweenness 1, got %v", values[1]) + } + if values[0] != 0 || values[2] != 0 { + t.Fatalf("path endpoints must have zero betweenness, got %v", values) + } +} + +func TestBrandesSplitsEqualShortestPaths(t *testing.T) { + neighbors := []serialize.SliceUint32{{1, 2}, {3}, {3}, {}} + values := brandesContributions(0, neighbors, false, 1) + if math.Abs(values[1]-0.5) > 1e-6 || + math.Abs(values[2]-0.5) > 1e-6 { + t.Fatalf("diamond paths were not split equally: %v", values) + } +} + +func TestBrandesPathCountsStayFiniteBeyondFloat64Range(t *testing.T) { + const layers = 1030 + neighbors := make([]serialize.SliceUint32, 1+layers*2) + neighbors[0] = serialize.SliceUint32{1, 2} + for layer := 0; layer < layers-1; layer++ { + left := uint32(1 + layer*2) + right := left + 1 + nextLeft := left + 2 + nextRight := left + 3 + neighbors[left] = serialize.SliceUint32{serialize.SUint32(nextLeft), + serialize.SUint32(nextRight)} + neighbors[right] = serialize.SliceUint32{serialize.SUint32(nextLeft), + serialize.SUint32(nextRight)} + } + + values := brandesContributions(0, neighbors, false, 1) + for vertex, value := range values { + if math.IsNaN(value) || math.IsInf(value, 0) { + t.Fatalf("vertex %d has non-finite contribution %v", vertex, value) + } + } + lastLeft := uint32(len(neighbors) - 2) + lastRight := lastLeft + 1 + if math.Abs(values[lastLeft]-values[lastRight]) > 1e-9 { + t.Fatalf("symmetric final layer differs: left=%v right=%v", + values[lastLeft], values[lastRight]) + } +} + +func TestBrandesDisconnectedSourcesDoNotAllocateGlobalStacks(t *testing.T) { + const vertices = 4096 + neighbors := make([]serialize.SliceUint32, vertices) + for source := uint32(0); source < vertices; source++ { + if values := brandesContributions(source, neighbors, false, 1); len(values) != 0 { + t.Fatalf("isolated source %d produced contributions: %v", source, values) + } + } +} + +func TestBetweennessScaleAndValidation(t *testing.T) { + if scale := betweennessScale(2, false, 1); scale != 0 { + t.Fatalf("two-vertex non-endpoint normalization must be zero, got %v", scale) + } + if scale := betweennessScale(4, false, 0.5); math.Abs(scale-1.0/3.0) > 1e-12 { + t.Fatalf("unexpected sampled scale: %v", scale) + } + for _, value := range []float64{0, -1, 1.1, math.NaN(), math.Inf(1)} { + if err := validateBetweennessSampleRate(value); err == nil { + t.Fatalf("invalid sample rate %v accepted", value) + } + } + for _, value := range []int{-1, 2} { + if _, err := parseBetweennessEndpoint(value); err == nil { + t.Fatalf("invalid endpoint flag %d accepted", value) + } + } +} + +func TestBetweennessPartitionCountsStayWorkerLocal(t *testing.T) { + workerValues := map[string]map[string]*compute.CValue{ + "worker-a": { + betweennessLocalCountsValue: { + ValueType: compute.ValueTypeInt64, + Action: compute.CValueActionNormal, + Value: serialize.SInt64(9), + }, + }, + "worker-b": { + betweennessLocalCountsValue: { + ValueType: compute.ValueTypeInt64, + Action: compute.CValueActionNormal, + Value: serialize.SInt64(1), + }, + }, + } + counts, err := collectBetweennessPartitionCounts(workerValues) + if err != nil { + t.Fatalf("collect partition counts: %v", err) + } + maxLocal, total, err := summarizeBetweennessPartitions(counts) + if err != nil { + t.Fatalf("summarize partition counts: %v", err) + } + if maxLocal != 9 || total != 10 || len(counts) != 2 { + t.Fatalf("unexpected worker-local counts %v: max=%d total=%d", + counts, maxLocal, total) + } + + workerValues["worker-b"][betweennessLocalCountsValue].Value = + serialize.SliceInt64{9, 1} + if _, err := collectBetweennessPartitionCounts(workerValues); err == nil { + t.Fatal("a broadcast aggregate slice was accepted as a local partition size") + } +} + +func TestBetweennessBatchPlanUsesLargestActualPartition(t *testing.T) { + maxLocal, total, err := summarizeBetweennessPartitions( + serialize.SliceInt64{9, 1, 1}) + if err != nil { + t.Fatalf("partition summary failed: %v", err) + } + if maxLocal != 9 || total != 11 { + t.Fatalf("unexpected partition summary: max=%d total=%d", maxLocal, total) + } + + batches, err := planBetweennessBatches(maxLocal, 2, 10) + if err != nil { + t.Fatalf("batch plan failed: %v", err) + } + if batches != 5 { + t.Fatalf("expected 5 batches for largest partition 9 at parallelism 2, got %d", + batches) + } + counts := make([]int, batches) + for localID := 0; localID < int(maxLocal); localID++ { + counts[localID%batches]++ + } + for batch, count := range counts { + if count > 2 { + t.Fatalf("batch %d contains %d sources, parallel target is 2", batch, count) + } + } +} + +func TestBetweennessBatchPlanFitsStepBudget(t *testing.T) { + batches, err := planBetweennessBatches(1_000_000, 1, 10) + if err != nil { + t.Fatalf("default-step batch plan failed: %v", err) + } + if batches != 9 { + t.Fatalf("default max_step=10 must cap source batches at 9, got %d", batches) + } + if 1+batches > 10 { + t.Fatalf("adjacency plus source batches exceed max_step: %d", 1+batches) + } + + batches, err = planBetweennessBatches(9, 2, 3) + if err != nil { + t.Fatalf("short-step batch plan failed: %v", err) + } + if batches != 2 { + t.Fatalf("max_step=3 must use two source batches, got %d", batches) + } + if _, err := planBetweennessBatches(1, 1, 1); err == nil { + t.Fatal("max_step=1 must be rejected") + } +} + +func TestBetweennessInactiveBatchPayloadIsConstant(t *testing.T) { + var encoded buffer.EncodeBuffer + encoded.Init(1024) + offset := serialize.SUint32(100) + if err := encoded.Marshal(&offset); err != nil { + t.Fatalf("marshal offset: %v", err) + } + for i := 0; i < 10000; i++ { + if err := encoded.Marshal(&noBetweennessVertexValue); err != nil { + t.Fatalf("marshal empty value %d: %v", i, err) + } + } + if got := len(encoded.PayLoad()); got != 4 { + t.Fatalf("10000 inactive vertices produced %d payload bytes, expected 4", got) + } +} + +func TestBetweennessSourceSamplingIsDeterministic(t *testing.T) { + for vertexID := uint32(0); vertexID < 100; vertexID++ { + first := sampleBetweennessSource(vertexID, 0.25) + second := sampleBetweennessSource(vertexID, 0.25) + if first != second { + t.Fatalf("source %d sampling is not deterministic", vertexID) + } + if !sampleBetweennessSource(vertexID, 1) { + t.Fatalf("exact mode skipped source %d", vertexID) + } + } +} + +func TestSumBetweennessPartials(t *testing.T) { + result, err := sumBetweennessPartials([]serialize.SliceFloat32{ + {1, 2, 3}, + {0.5, 1, 1.5}, + }, 3) + if err != nil { + t.Fatalf("sum partials: %v", err) + } + expected := []float64{1.5, 3, 4.5} + for vertexID, value := range result { + if math.Abs(float64(value)-expected[vertexID]) > 1e-6 { + t.Fatalf("vertex %d expected %v, got %v", + vertexID, expected[vertexID], value) + } + } + if _, err := sumBetweennessPartials([]serialize.SliceFloat32{{1}}, 2); err == nil { + t.Fatal("mismatched partial length was accepted") + } +} + +func TestBetweennessOutputRequiresFinalGlobalVector(t *testing.T) { + worker := &BetweennessCentralityWorker{totalVertices: 3} + context := worker.MakeContext(nil) + context.CreateValue(betweennessFinalValue, compute.ValueTypeSliceFloat32, + compute.CValueActionNormal) + context.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + if err := worker.ValidateOutput(); err == nil { + t.Fatal("expected missing global result to reject output") + } + + expected := serialize.SliceFloat32{0, 1, 0} + context.SetValue(betweennessFinalValue, expected) + if err := worker.ValidateOutput(); err != nil { + t.Fatalf("completed global result rejected: %v", err) + } + if len(worker.betweennessValues) != len(expected) || + worker.betweennessValues[1] != expected[1] { + t.Fatalf("global result was not installed: %v", worker.betweennessValues) + } +} From 06110ac5cb46e7d4d8a561eec4f6bc4db9a22655 Mon Sep 17 00:00:00 2001 From: imbajin Date: Fri, 6 Nov 2026 01:44:44 +0800 Subject: [PATCH 2/3] fix(vermeer): stream betweenness reduction in bounded chunks --- vermeer/algorithms/betweenness_centrality.go | 572 +++++++++++-------- 1 file changed, 323 insertions(+), 249 deletions(-) diff --git a/vermeer/algorithms/betweenness_centrality.go b/vermeer/algorithms/betweenness_centrality.go index 093586b1b..3aee012db 100644 --- a/vermeer/algorithms/betweenness_centrality.go +++ b/vermeer/algorithms/betweenness_centrality.go @@ -18,8 +18,10 @@ under the License. package algorithms import ( + "encoding/binary" "fmt" "math" + "sync" "vermeer/apps/common" "vermeer/apps/compute" "vermeer/apps/options" @@ -29,11 +31,14 @@ import ( ) const ( - betweennessStopValue = "stop_sum" - betweennessLocalCountsValue = "betweenness_local_counts" - betweennessBatchCountValue = "betweenness_batch_count" - betweennessPartialValue = "betweenness_partial_values" - betweennessFinalValue = "betweenness_final_values" + betweennessStopValue = "stop_sum" + betweennessCompleteValue = "betweenness_complete" + + // A contribution chunk stays well below the 2 MiB scatter buffer. Several + // chunks can therefore be sent in one bounded RPC frame, while no result ever + // becomes a whole-graph compute value or a single multi-gigabyte message. + maxBetweennessChunkRecords = 65_536 + betweennessRecordSize = 8 // uint32 target + float32 worker partial ) func init() { @@ -68,6 +73,14 @@ func validateBetweennessSampleRate(value float64) error { return nil } +func validateBetweennessMaxStep(maxStep int) error { + if maxStep < 2 { + return fmt.Errorf("betweenness_centrality requires compute.max_step >= 2, got %d", + maxStep) + } + return nil +} + func parseBetweennessEndpoint(value int) (bool, error) { switch value { case 0: @@ -97,64 +110,6 @@ func betweennessScale(totalVertices uint32, useEndpoint bool, sampleRate float64 return scale / sampleRate } -func planBetweennessBatches(maxLocalVertices uint32, parallel, maxStep int) (int, error) { - if maxLocalVertices == 0 { - return 0, nil - } - if maxStep < 2 { - return 0, fmt.Errorf("betweenness_centrality requires compute.max_step >= 2, got %d", - maxStep) - } - if parallel <= 0 { - parallel = 1 - } - ideal := (uint64(maxLocalVertices) + uint64(parallel) - 1) / uint64(parallel) - available := uint64(maxStep - 1) - if ideal > available { - ideal = available - } - return int(ideal), nil -} - -func collectBetweennessPartitionCounts( - workerValues map[string]map[string]*compute.CValue) (serialize.SliceInt64, error) { - counts := make(serialize.SliceInt64, 0, len(workerValues)) - for workerName, values := range workerValues { - cValue, ok := values[betweennessLocalCountsValue] - if !ok || cValue.Value == nil { - return nil, fmt.Errorf("worker %s did not report its betweenness partition size", - workerName) - } - count, ok := cValue.Value.(serialize.SInt64) - if !ok { - return nil, fmt.Errorf("worker %s reported invalid betweenness partition type %T", - workerName, cValue.Value) - } - counts = append(counts, count) - } - return counts, nil -} - -func summarizeBetweennessPartitions(counts serialize.SliceInt64) (uint32, uint32, error) { - var maxLocal uint64 - var total uint64 - for _, rawCount := range counts { - count := int64(rawCount) - if count < 0 { - return 0, 0, fmt.Errorf("negative betweenness partition size %d", count) - } - value := uint64(count) - total += value - if value > maxLocal { - maxLocal = value - } - } - if total > uint64(^uint32(0)) || maxLocal > uint64(^uint32(0)) { - return 0, 0, fmt.Errorf("betweenness partition sizes exceed uint32 vertex range") - } - return uint32(maxLocal), uint32(total), nil -} - func sampleBetweennessSource(vertexID uint32, sampleRate float64) bool { if sampleRate >= 1 { return true @@ -166,6 +121,24 @@ func sampleBetweennessSource(vertexID uint32, sampleRate float64) bool { return float64(value>>11) < sampleRate*float64(uint64(1)<<53) } +func selectBetweennessSources(start, count uint32, sampleRate float64) []uint32 { + if count == 0 { + return nil + } + capacity := int(math.Ceil(float64(count) * sampleRate)) + if capacity < 1 { + capacity = 1 + } + sources := make([]uint32, 0, capacity) + for localID := uint32(0); localID < count; localID++ { + vertexID := start + localID + if sampleBetweennessSource(vertexID, sampleRate) { + sources = append(sources, vertexID) + } + } + return sources +} + func logAddExp(left, right float64) float64 { if math.IsInf(left, -1) { return right @@ -242,99 +215,228 @@ func brandesContributions(source uint32, neighbors []serialize.SliceUint32, return result } -func sumBetweennessPartials(partials []serialize.SliceFloat32, - totalVertices uint32) (serialize.SliceFloat32, error) { - if totalVertices == 0 { - return serialize.SliceFloat32{}, nil +// betweennessContributionChunk is already encoded so its retained memory is +// exactly the wire representation: a uint32 count followed by count pairs of +// (uint32 target, float32 worker partial). It never contains a whole-graph +// vector and each chunk has a fixed upper bound. +type betweennessContributionChunk struct { + payload []byte +} + +func (chunk *betweennessContributionChunk) Marshal(buffer []byte) (int, error) { + if len(buffer) < len(chunk.payload) { + return 0, fmt.Errorf("betweenness contribution buffer has %d bytes, need %d", + len(buffer), len(chunk.payload)) } - if len(partials) == 0 { - return nil, fmt.Errorf("no betweenness worker partials were reported") + return copy(buffer, chunk.payload), nil +} + +func (chunk *betweennessContributionChunk) Unmarshal(buffer []byte) (int, error) { + size, err := betweennessChunkSize(buffer) + if err != nil { + return 0, err } + chunk.payload = append(chunk.payload[:0], buffer[:size]...) + return size, nil +} - sums := make([]float64, totalVertices) - for workerIndex, partial := range partials { - if uint32(len(partial)) != totalVertices { - return nil, fmt.Errorf("betweenness worker %d reported %d values, expected %d", - workerIndex, len(partial), totalVertices) - } - for vertexID, value := range partial { - sums[vertexID] += float64(value) +func (chunk *betweennessContributionChunk) ToString() string { + if len(chunk.payload) < 4 { + return "0" + } + return fmt.Sprintf("%d", binary.LittleEndian.Uint32(chunk.payload[:4])) +} + +func (chunk *betweennessContributionChunk) PredictSize() int { + return len(chunk.payload) +} + +func betweennessChunkSize(buffer []byte) (int, error) { + if len(buffer) < 4 { + return 0, fmt.Errorf("betweenness contribution chunk is missing its count") + } + count := uint64(binary.LittleEndian.Uint32(buffer[:4])) + size := uint64(4) + count*betweennessRecordSize + if size > uint64(len(buffer)) { + return 0, fmt.Errorf("truncated betweenness contribution chunk: need %d bytes, got %d", + size, len(buffer)) + } + return int(size), nil +} + +func appendBetweennessRecord(payload []byte, target uint32, value float32) []byte { + offset := len(payload) + payload = append(payload, make([]byte, betweennessRecordSize)...) + binary.LittleEndian.PutUint32(payload[offset:offset+4], target) + binary.LittleEndian.PutUint32(payload[offset+4:offset+8], math.Float32bits(value)) + return payload +} + +func splitBetweennessContributions(partial []float64, localStart, + localCount uint32) ([]float64, []betweennessContributionChunk, error) { + local := make([]float64, localCount) + localEnd := uint64(localStart) + uint64(localCount) + chunks := make([]betweennessContributionChunk, 0) + var payload []byte + records := 0 + + flush := func() { + if records == 0 { + return } + binary.LittleEndian.PutUint32(payload[:4], uint32(records)) + chunks = append(chunks, betweennessContributionChunk{payload: payload}) + payload = nil + records = 0 } - result := make(serialize.SliceFloat32, totalVertices) - for vertexID, value := range sums { + for target, value := range partial { + if value == 0 { + continue + } if math.IsNaN(value) || math.IsInf(value, 0) { - return nil, fmt.Errorf("betweenness result for vertex %d is not finite", vertexID) + return nil, nil, fmt.Errorf("betweenness partial for vertex %d is not finite", target) + } + rounded := float32(value) + if math.IsInf(float64(rounded), 0) { + return nil, nil, fmt.Errorf("betweenness partial for vertex %d exceeds float32", target) + } + if rounded == 0 { + continue + } + + targetID := uint32(target) + if uint64(targetID) >= uint64(localStart) && uint64(targetID) < localEnd { + local[targetID-localStart] += float64(rounded) + continue + } + if records == 0 { + payload = make([]byte, 4, 4+maxBetweennessChunkRecords*betweennessRecordSize) + } + payload = appendBetweennessRecord(payload, targetID, rounded) + records++ + if records == maxBetweennessChunkRecords { + flush() } - result[vertexID] = serialize.SFloat32(value) } - return result, nil + flush() + return local, chunks, nil } -type emptyBetweennessVertexValue struct{} +type betweennessContributionReceiver struct { + worker *BetweennessCentralityWorker +} -func (*emptyBetweennessVertexValue) Marshal([]byte) (int, error) { +func (*betweennessContributionReceiver) Marshal([]byte) (int, error) { + // Inactive local vertex slots intentionally contribute no bytes. Active + // slots return a betweennessContributionChunk from VertexValue instead. return 0, nil } -func (*emptyBetweennessVertexValue) Unmarshal([]byte) (int, error) { - return 0, nil +func (receiver *betweennessContributionReceiver) Unmarshal(buffer []byte) (int, error) { + if receiver.worker == nil { + return 0, fmt.Errorf("betweenness contribution receiver is not initialized") + } + size, err := betweennessChunkSize(buffer) + if err != nil { + return 0, err + } + count := int(binary.LittleEndian.Uint32(buffer[:4])) + worker := receiver.worker + localEnd := uint64(worker.localStart) + uint64(worker.localCount) + + worker.valuesMu.Lock() + defer worker.valuesMu.Unlock() + for index := 0; index < count; index++ { + offset := 4 + index*betweennessRecordSize + target := binary.LittleEndian.Uint32(buffer[offset : offset+4]) + value := math.Float32frombits(binary.LittleEndian.Uint32( + buffer[offset+4 : offset+8])) + if math.IsNaN(float64(value)) || math.IsInf(float64(value), 0) { + return 0, fmt.Errorf("received non-finite betweenness value for vertex %d", target) + } + if uint64(target) < uint64(worker.localStart) || uint64(target) >= localEnd { + continue + } + worker.localValues[target-worker.localStart] += float64(value) + } + return size, nil } -func (*emptyBetweennessVertexValue) ToString() string { +func (*betweennessContributionReceiver) ToString() string { return "" } -func (*emptyBetweennessVertexValue) PredictSize() int { +func (*betweennessContributionReceiver) PredictSize() int { return 0 } -var noBetweennessVertexValue emptyBetweennessVertexValue - type BetweennessCentralityWorker struct { compute.WorkerComputerBase useEndPoint bool scale float64 sampleRate float64 - maxStep int - batchCount int - currentBatch int totalVertices uint32 - betweennessValues serialize.SliceFloat32 + localStart uint32 + localCount uint32 + sampledSources []uint32 + allSources bool + selectedSources int neighborEdges []serialize.SliceUint32 localValues []float64 - batchValues []map[uint32]float64 + betweennessValues serialize.SliceFloat32 + outgoingChunks []betweennessContributionChunk + outgoingOffset int + computed bool + valuesMu sync.Mutex + receiver betweennessContributionReceiver } func (bcw *BetweennessCentralityWorker) VertexValue(vertexID uint32) serialize.MarshalAble { if bcw.WContext.Output { - return &bcw.betweennessValues[vertexID] + if vertexID < bcw.localStart || vertexID >= bcw.localStart+bcw.localCount { + return &bcw.receiver + } + return &bcw.betweennessValues[vertexID-bcw.localStart] } if bcw.WContext.Step == 1 { return &bcw.neighborEdges[vertexID] } - // The framework still visits every local vertex while scattering. Returning a - // zero-byte value keeps the per-batch wire payload constant after adjacency - // has been exchanged in step 1. - return &noBetweennessVertexValue + + // The standard scatter loop provides one slot per local vertex. Fill the + // current window with bounded contribution chunks and leave every remaining + // slot at zero bytes. Receivers do not rely on the slot's vertex ID because + // each record carries its actual target ID. + if vertexID >= bcw.localStart && vertexID < bcw.localStart+bcw.localCount { + localID := int(vertexID - bcw.localStart) + window := bcw.outgoingWindowCount() + if localID < window { + return &bcw.outgoingChunks[bcw.outgoingOffset+localID] + } + } + return &bcw.receiver } func (bcw *BetweennessCentralityWorker) ValidateOutput() error { - cValue, ok := bcw.WContext.CValues[betweennessFinalValue] + cValue, ok := bcw.WContext.CValues[betweennessCompleteValue] if !ok || cValue.Value == nil { - return fmt.Errorf("betweenness_centrality output is missing the global result") + return fmt.Errorf("betweenness_centrality output is missing completion state") } - values, ok := cValue.Value.(serialize.SliceFloat32) - if !ok { - return fmt.Errorf("betweenness_centrality output has invalid global result type %T", - cValue.Value) + complete, ok := cValue.Value.(serialize.SInt32) + if !ok || complete != 1 { + return fmt.Errorf("betweenness_centrality did not finish contribution scattering; increase compute.max_step") } - if uint32(len(values)) != bcw.totalVertices { - return fmt.Errorf("betweenness_centrality did not finish all source batches: got %d global values, expected %d; increase compute.max_step", - len(values), bcw.totalVertices) + if uint32(len(bcw.localValues)) != bcw.localCount { + return fmt.Errorf("betweenness_centrality local result has %d values, expected %d", + len(bcw.localValues), bcw.localCount) + } + bcw.betweennessValues = make(serialize.SliceFloat32, bcw.localCount) + for localID, value := range bcw.localValues { + if math.IsNaN(value) || math.IsInf(value, 0) { + return fmt.Errorf("betweenness result for local vertex %d is not finite", localID) + } + bcw.betweennessValues[localID] = serialize.SFloat32(value) } - bcw.betweennessValues = values return nil } @@ -349,116 +451,150 @@ func (bcw *BetweennessCentralityWorker) Init() error { if err != nil { return err } - bcw.maxStep = options.GetInt(bcw.WContext.Params, "compute.max_step") - if bcw.maxStep < 2 { - return fmt.Errorf("betweenness_centrality requires compute.max_step >= 2, got %d", - bcw.maxStep) + if err := validateBetweennessMaxStep(options.GetInt(bcw.WContext.Params, + "compute.max_step")); err != nil { + return err } bcw.useEndPoint = useEndpoint bcw.totalVertices = bcw.WContext.GraphData.Vertex.TotalVertexCount() - localVertices := bcw.WContext.GraphData.VertexCount + bcw.localStart = bcw.WContext.GraphData.VertIDStart + bcw.localCount = bcw.WContext.GraphData.VertexCount bcw.scale = betweennessScale(bcw.totalVertices, bcw.useEndPoint, bcw.sampleRate) - bcw.betweennessValues = make(serialize.SliceFloat32, bcw.totalVertices) + if bcw.sampleRate >= 1 { + bcw.allSources = true + bcw.selectedSources = int(bcw.localCount) + } else { + bcw.sampledSources = selectBetweennessSources(bcw.localStart, bcw.localCount, + bcw.sampleRate) + bcw.selectedSources = len(bcw.sampledSources) + } bcw.neighborEdges = make([]serialize.SliceUint32, bcw.totalVertices) - bcw.localValues = make([]float64, bcw.totalVertices) - bcw.batchValues = make([]map[uint32]float64, bcw.WContext.Parallel) + bcw.localValues = make([]float64, bcw.localCount) + bcw.betweennessValues = make(serialize.SliceFloat32, bcw.localCount) + bcw.receiver.worker = bcw bcw.WContext.CreateValue(betweennessStopValue, compute.ValueTypeInt32, compute.CValueActionAggregate) bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1)) - // This value is intentionally Normal rather than Aggregate. Aggregate values - // are broadcast back to workers; feeding the global slice into the next round - // would multiply it by the worker count on every superstep. - bcw.WContext.CreateValue(betweennessLocalCountsValue, compute.ValueTypeInt64, + bcw.WContext.CreateValue(betweennessCompleteValue, compute.ValueTypeInt32, compute.CValueActionNormal) - bcw.WContext.SetValue(betweennessLocalCountsValue, serialize.SInt64(localVertices)) - bcw.WContext.CreateValue(betweennessBatchCountValue, compute.ValueTypeInt32, - compute.CValueActionNormal) - bcw.WContext.SetValue(betweennessBatchCountValue, serialize.SInt32(0)) - bcw.WContext.CreateValue(betweennessPartialValue, compute.ValueTypeSliceFloat32, - compute.CValueActionNormal) - bcw.WContext.SetValue(betweennessPartialValue, serialize.SliceFloat32{}) - bcw.WContext.CreateValue(betweennessFinalValue, compute.ValueTypeSliceFloat32, - compute.CValueActionNormal) - bcw.WContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + bcw.WContext.SetValue(betweennessCompleteValue, serialize.SInt32(0)) return nil } func (bcw *BetweennessCentralityWorker) BeforeStep() { - if bcw.WContext.Step < 2 { + if bcw.WContext.Step != 2 || bcw.computed { return } - bcw.batchCount = int(bcw.WContext.GetValue(betweennessBatchCountValue).(serialize.SInt32)) - bcw.currentBatch = int(bcw.WContext.Step) - 2 - for partitionID := range bcw.batchValues { - bcw.batchValues[partitionID] = make(map[uint32]float64) + partial := bcw.computeSampledSourcePartials() + local, chunks, err := splitBetweennessContributions(partial, + bcw.localStart, bcw.localCount) + if err != nil { + panic(err) } + bcw.localValues = local + bcw.outgoingChunks = chunks + bcw.outgoingOffset = 0 + bcw.computed = true + logrus.Infof("betweenness selected sources: %d, contribution chunks: %d", + bcw.selectedSources, len(bcw.outgoingChunks)) } -func (bcw *BetweennessCentralityWorker) Compute(vertexID uint32, partitionID int) { - localID := vertexID - bcw.WContext.GraphData.VertIDStart - if bcw.WContext.Step == 1 { - inEdges := bcw.WContext.GraphData.Edges.GetInEdges(localID) - neighbors := make(serialize.SliceUint32, 0, len(inEdges)) - seen := make(map[serialize.SUint32]struct{}, len(inEdges)) - for _, neighbor := range inEdges { - if uint32(neighbor) == vertexID { - continue - } - if _, exists := seen[neighbor]; exists { - continue - } - seen[neighbor] = struct{}{} - neighbors = append(neighbors, neighbor) - } - bcw.neighborEdges[vertexID] = neighbors - return - } - - if bcw.batchCount <= 0 || bcw.currentBatch >= bcw.batchCount || - int(localID)%bcw.batchCount != bcw.currentBatch { +func (bcw *BetweennessCentralityWorker) Compute(vertexID uint32, _ int) { + if bcw.WContext.Step != 1 { return } - if !sampleBetweennessSource(vertexID, bcw.sampleRate) { - return - } - - contributions := brandesContributions(vertexID, bcw.neighborEdges, - bcw.useEndPoint, bcw.scale) - batch := bcw.batchValues[partitionID] - for target, value := range contributions { - batch[target] += value + localID := vertexID - bcw.localStart + inEdges := bcw.WContext.GraphData.Edges.GetInEdges(localID) + neighbors := make(serialize.SliceUint32, 0, len(inEdges)) + seen := make(map[serialize.SUint32]struct{}, len(inEdges)) + for _, neighbor := range inEdges { + if uint32(neighbor) == vertexID || uint32(neighbor) >= bcw.totalVertices { + continue + } + if _, exists := seen[neighbor]; exists { + continue + } + seen[neighbor] = struct{}{} + neighbors = append(neighbors, neighbor) } + bcw.neighborEdges[vertexID] = neighbors } func (bcw *BetweennessCentralityWorker) AfterStep() { if bcw.WContext.Step == 1 { - bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1)) + if bcw.selectedSources == 0 || bcw.scale == 0 { + bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(0)) + } else { + bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1)) + } return } - bcw.aggregateCurrentBatch() - if bcw.batchCount <= 0 || bcw.currentBatch+1 < bcw.batchCount { + bcw.outgoingOffset += bcw.outgoingWindowCount() + if bcw.outgoingOffset < len(bcw.outgoingChunks) { bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(1)) return } + bcw.outgoingChunks = nil + bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(0)) +} - partial := make(serialize.SliceFloat32, bcw.totalVertices) - for vertexID, value := range bcw.localValues { - partial[vertexID] = serialize.SFloat32(value) +func (bcw *BetweennessCentralityWorker) outgoingWindowCount() int { + remaining := len(bcw.outgoingChunks) - bcw.outgoingOffset + if remaining <= 0 || bcw.localCount == 0 { + return 0 } - bcw.WContext.SetValue(betweennessPartialValue, partial) - bcw.WContext.SetValue(betweennessStopValue, serialize.SInt32(0)) + if remaining < int(bcw.localCount) { + return remaining + } + return int(bcw.localCount) } -func (bcw *BetweennessCentralityWorker) aggregateCurrentBatch() { - for partitionID, values := range bcw.batchValues { - for target, value := range values { - bcw.localValues[target] += value +func (bcw *BetweennessCentralityWorker) computeSampledSourcePartials() []float64 { + partial := make([]float64, bcw.totalVertices) + if bcw.selectedSources == 0 || bcw.scale == 0 { + return partial + } + parallel := bcw.WContext.Parallel + if parallel <= 0 { + parallel = 1 + } + if parallel > bcw.selectedSources { + parallel = bcw.selectedSources + } + + results := make(chan map[uint32]float64, parallel) + var workers sync.WaitGroup + workers.Add(parallel) + for workerID := 0; workerID < parallel; workerID++ { + go func(offset int) { + defer workers.Done() + for sourceIndex := offset; sourceIndex < bcw.selectedSources; sourceIndex += parallel { + results <- brandesContributions(bcw.selectedSource(sourceIndex), + bcw.neighborEdges, bcw.useEndPoint, bcw.scale) + } + }(workerID) + } + go func() { + workers.Wait() + close(results) + }() + + for contributions := range results { + for target, value := range contributions { + partial[target] += value } - bcw.batchValues[partitionID] = nil } + return partial +} + +func (bcw *BetweennessCentralityWorker) selectedSource(index int) uint32 { + if bcw.allSources { + return bcw.localStart + uint32(index) + } + return bcw.sampledSources[index] } func (bcw *BetweennessCentralityWorker) OutputValueType() string { @@ -467,86 +603,24 @@ func (bcw *BetweennessCentralityWorker) OutputValueType() string { type BetweennessCentralityMaster struct { compute.MasterComputerBase - planned bool - totalVertices uint32 - batchCount int } func (bcm *BetweennessCentralityMaster) Init() error { - maxStep := options.GetInt(bcm.MContext.Params, "compute.max_step") - if maxStep < 2 { - return fmt.Errorf("betweenness_centrality requires compute.max_step >= 2, got %d", - maxStep) + if err := validateBetweennessMaxStep(options.GetInt(bcm.MContext.Params, + "compute.max_step")); err != nil { + return err } - bcm.MContext.CreateValue(betweennessBatchCountValue, compute.ValueTypeInt32, + bcm.MContext.CreateValue(betweennessCompleteValue, compute.ValueTypeInt32, compute.CValueActionNormal) - bcm.MContext.SetValue(betweennessBatchCountValue, serialize.SInt32(0)) - bcm.MContext.CreateValue(betweennessFinalValue, compute.ValueTypeSliceFloat32, - compute.CValueActionNormal) - bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + bcm.MContext.SetValue(betweennessCompleteValue, serialize.SInt32(0)) return nil } func (bcm *BetweennessCentralityMaster) Compute() bool { - if !bcm.planned { - counts, err := collectBetweennessPartitionCounts(bcm.MContext.WorkerCValues) - if err != nil { - logrus.Errorf("collect betweenness partition sizes: %v", err) - return false - } - maxLocal, totalVertices, err := summarizeBetweennessPartitions(counts) - if err != nil { - logrus.Errorf("invalid betweenness partition summary: %v", err) - return false - } - bcm.totalVertices = totalVertices - bcm.batchCount, err = planBetweennessBatches(maxLocal, - options.GetInt(bcm.MContext.Params, "compute.parallel"), - options.GetInt(bcm.MContext.Params, "compute.max_step")) - if err != nil { - logrus.Errorf("plan betweenness batches: %v", err) - return false - } - bcm.MContext.SetValue(betweennessBatchCountValue, - serialize.SInt32(bcm.batchCount)) - bcm.planned = true - logrus.Infof("betweenness source batches: %d, largest partition: %d", - bcm.batchCount, maxLocal) - if bcm.totalVertices == 0 { - bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) - return false - } - return true - } - stopSum := bcm.MContext.GetValue(betweennessStopValue).(serialize.SInt32) if stopSum != 0 { return true } - - partials := make([]serialize.SliceFloat32, 0, len(bcm.MContext.WorkerCValues)) - for workerName, values := range bcm.MContext.WorkerCValues { - cValue, ok := values[betweennessPartialValue] - if !ok { - logrus.Errorf("worker %s did not report betweenness partial values", workerName) - bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) - return false - } - partial, ok := cValue.Value.(serialize.SliceFloat32) - if !ok { - logrus.Errorf("worker %s reported invalid betweenness partial type %T", - workerName, cValue.Value) - bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) - return false - } - partials = append(partials, partial) - } - result, err := sumBetweennessPartials(partials, bcm.totalVertices) - if err != nil { - logrus.Errorf("aggregate betweenness partial values: %v", err) - bcm.MContext.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) - return false - } - bcm.MContext.SetValue(betweennessFinalValue, result) + bcm.MContext.SetValue(betweennessCompleteValue, serialize.SInt32(1)) return false } From 5d6fb40e0ce35532d4e4ae2da82c85477e13b67e Mon Sep 17 00:00:00 2001 From: imbajin Date: Fri, 6 Nov 2026 01:45:19 +0800 Subject: [PATCH 3/3] test(vermeer): cover bounded betweenness reduction --- .../algorithms/betweenness_centrality_test.go | 280 +++++++++++------- 1 file changed, 176 insertions(+), 104 deletions(-) diff --git a/vermeer/algorithms/betweenness_centrality_test.go b/vermeer/algorithms/betweenness_centrality_test.go index 7c593a45b..fcced2076 100644 --- a/vermeer/algorithms/betweenness_centrality_test.go +++ b/vermeer/algorithms/betweenness_centrality_test.go @@ -18,6 +18,7 @@ limitations under the License. package algorithms import ( + "encoding/binary" "math" "testing" "vermeer/apps/buffer" @@ -107,108 +108,136 @@ func TestBetweennessScaleAndValidation(t *testing.T) { t.Fatalf("invalid endpoint flag %d accepted", value) } } -} - -func TestBetweennessPartitionCountsStayWorkerLocal(t *testing.T) { - workerValues := map[string]map[string]*compute.CValue{ - "worker-a": { - betweennessLocalCountsValue: { - ValueType: compute.ValueTypeInt64, - Action: compute.CValueActionNormal, - Value: serialize.SInt64(9), - }, - }, - "worker-b": { - betweennessLocalCountsValue: { - ValueType: compute.ValueTypeInt64, - Action: compute.CValueActionNormal, - Value: serialize.SInt64(1), - }, - }, - } - counts, err := collectBetweennessPartitionCounts(workerValues) - if err != nil { - t.Fatalf("collect partition counts: %v", err) + if err := validateBetweennessMaxStep(1); err == nil { + t.Fatal("compute.max_step=1 was accepted") } - maxLocal, total, err := summarizeBetweennessPartitions(counts) - if err != nil { - t.Fatalf("summarize partition counts: %v", err) + if err := validateBetweennessMaxStep(2); err != nil { + t.Fatalf("compute.max_step=2 was rejected: %v", err) } - if maxLocal != 9 || total != 10 || len(counts) != 2 { - t.Fatalf("unexpected worker-local counts %v: max=%d total=%d", - counts, maxLocal, total) +} + +func TestBetweennessSourceSamplingIsDeterministicAndCompacted(t *testing.T) { + first := selectBetweennessSources(100, 10_000, 0.01) + second := selectBetweennessSources(100, 10_000, 0.01) + if len(first) == 0 || len(first) >= 10_000 { + t.Fatalf("unexpected selected source count %d", len(first)) + } + if len(first) != len(second) { + t.Fatalf("deterministic selections differ in length: %d vs %d", + len(first), len(second)) + } + for index, source := range first { + if source != second[index] { + t.Fatalf("selection differs at %d: %d vs %d", + index, source, second[index]) + } + if !sampleBetweennessSource(source, 0.01) { + t.Fatalf("unselected source %d was compacted into the work list", source) + } } +} - workerValues["worker-b"][betweennessLocalCountsValue].Value = - serialize.SliceInt64{9, 1} - if _, err := collectBetweennessPartitionCounts(workerValues); err == nil { - t.Fatal("a broadcast aggregate slice was accepted as a local partition size") +func TestExactModeDoesNotRetainASecondSourceIDArray(t *testing.T) { + worker := &BetweennessCentralityWorker{ + localStart: 50, + localCount: 3, + allSources: true, + selectedSources: 3, + } + if len(worker.sampledSources) != 0 { + t.Fatalf("exact mode retained sampled source IDs: %v", worker.sampledSources) + } + for index, expected := range []uint32{50, 51, 52} { + if got := worker.selectedSource(index); got != expected { + t.Fatalf("source %d expected %d, got %d", index, expected, got) + } } } -func TestBetweennessBatchPlanUsesLargestActualPartition(t *testing.T) { - maxLocal, total, err := summarizeBetweennessPartitions( - serialize.SliceInt64{9, 1, 1}) +func TestSplitBetweennessContributionsKeepsLocalAndChunksRemote(t *testing.T) { + partial := []float64{0, 1.25, 2.5, 3.75} + local, chunks, err := splitBetweennessContributions(partial, 1, 2) if err != nil { - t.Fatalf("partition summary failed: %v", err) + t.Fatalf("split contributions: %v", err) } - if maxLocal != 9 || total != 11 { - t.Fatalf("unexpected partition summary: max=%d total=%d", maxLocal, total) + if len(local) != 2 || local[0] != 1.25 || local[1] != 2.5 { + t.Fatalf("unexpected local contribution slice: %v", local) } - - batches, err := planBetweennessBatches(maxLocal, 2, 10) - if err != nil { - t.Fatalf("batch plan failed: %v", err) + if len(chunks) != 1 { + t.Fatalf("expected one remote chunk, got %d", len(chunks)) + } + if count := binary.LittleEndian.Uint32(chunks[0].payload[:4]); count != 1 { + t.Fatalf("expected one remote record, got %d", count) } - if batches != 5 { - t.Fatalf("expected 5 batches for largest partition 9 at parallelism 2, got %d", - batches) + + remote := &BetweennessCentralityWorker{ + localStart: 3, + localCount: 1, + localValues: make([]float64, 1), } - counts := make([]int, batches) - for localID := 0; localID < int(maxLocal); localID++ { - counts[localID%batches]++ + receiver := betweennessContributionReceiver{worker: remote} + if _, err := receiver.Unmarshal(chunks[0].payload); err != nil { + t.Fatalf("receive contribution chunk: %v", err) } - for batch, count := range counts { - if count > 2 { - t.Fatalf("batch %d contains %d sources, parallel target is 2", batch, count) - } + if math.Abs(remote.localValues[0]-3.75) > 1e-6 { + t.Fatalf("remote contribution expected 3.75, got %v", remote.localValues[0]) } } -func TestBetweennessBatchPlanFitsStepBudget(t *testing.T) { - batches, err := planBetweennessBatches(1_000_000, 1, 10) +func TestBetweennessContributionChunksStayBounded(t *testing.T) { + partial := make([]float64, maxBetweennessChunkRecords+5) + for index := range partial { + partial[index] = float64(index + 1) + } + _, chunks, err := splitBetweennessContributions(partial, 0, 0) if err != nil { - t.Fatalf("default-step batch plan failed: %v", err) + t.Fatalf("split bounded chunks: %v", err) } - if batches != 9 { - t.Fatalf("default max_step=10 must cap source batches at 9, got %d", batches) + if len(chunks) != 2 { + t.Fatalf("expected two chunks, got %d", len(chunks)) } - if 1+batches > 10 { - t.Fatalf("adjacency plus source batches exceed max_step: %d", 1+batches) + maximum := 4 + maxBetweennessChunkRecords*betweennessRecordSize + for index, chunk := range chunks { + if chunk.PredictSize() > maximum { + t.Fatalf("chunk %d has %d bytes, maximum is %d", + index, chunk.PredictSize(), maximum) + } + if chunk.PredictSize() >= 2*1024*1024 { + t.Fatalf("chunk %d is not scatter-buffer bounded: %d bytes", + index, chunk.PredictSize()) + } } +} - batches, err = planBetweennessBatches(9, 2, 3) - if err != nil { - t.Fatalf("short-step batch plan failed: %v", err) +func TestBetweennessContributionChunkRejectsTruncation(t *testing.T) { + payload := make([]byte, 4) + binary.LittleEndian.PutUint32(payload, 1) + var chunk betweennessContributionChunk + if _, err := chunk.Unmarshal(payload); err == nil { + t.Fatal("truncated contribution chunk was accepted") } - if batches != 2 { - t.Fatalf("max_step=3 must use two source batches, got %d", batches) + + worker := &BetweennessCentralityWorker{ + localCount: 1, + localValues: make([]float64, 1), } - if _, err := planBetweennessBatches(1, 1, 1); err == nil { - t.Fatal("max_step=1 must be rejected") + receiver := betweennessContributionReceiver{worker: worker} + if _, err := receiver.Unmarshal(payload); err == nil { + t.Fatal("receiver accepted a truncated contribution chunk") } } -func TestBetweennessInactiveBatchPayloadIsConstant(t *testing.T) { +func TestBetweennessInactiveScatterPayloadIsConstant(t *testing.T) { var encoded buffer.EncodeBuffer encoded.Init(1024) offset := serialize.SUint32(100) if err := encoded.Marshal(&offset); err != nil { t.Fatalf("marshal offset: %v", err) } - for i := 0; i < 10000; i++ { - if err := encoded.Marshal(&noBetweennessVertexValue); err != nil { - t.Fatalf("marshal empty value %d: %v", i, err) + empty := &betweennessContributionReceiver{} + for index := 0; index < 10_000; index++ { + if err := encoded.Marshal(empty); err != nil { + t.Fatalf("marshal empty value %d: %v", index, err) } } if got := len(encoded.PayLoad()); got != 4 { @@ -216,56 +245,99 @@ func TestBetweennessInactiveBatchPayloadIsConstant(t *testing.T) { } } -func TestBetweennessSourceSamplingIsDeterministic(t *testing.T) { - for vertexID := uint32(0); vertexID < 100; vertexID++ { - first := sampleBetweennessSource(vertexID, 0.25) - second := sampleBetweennessSource(vertexID, 0.25) - if first != second { - t.Fatalf("source %d sampling is not deterministic", vertexID) +func TestBetweennessOutgoingChunksUseLocalVertexWindows(t *testing.T) { + worker := &BetweennessCentralityWorker{ + localStart: 10, + localCount: 2, + outgoingChunks: make([]betweennessContributionChunk, 5), + } + context := worker.MakeContext(nil) + context.Step = 2 + context.CreateValue(betweennessStopValue, compute.ValueTypeInt32, + compute.CValueActionAggregate) + context.SetValue(betweennessStopValue, serialize.SInt32(1)) + worker.receiver.worker = worker + + cases := []struct { + window int + offset int + }{ + {window: 2, offset: 2}, + {window: 2, offset: 4}, + {window: 1, offset: 5}, + } + for _, test := range cases { + if got := worker.outgoingWindowCount(); got != test.window { + t.Fatalf("offset %d expected window %d, got %d", + worker.outgoingOffset, test.window, got) } - if !sampleBetweennessSource(vertexID, 1) { - t.Fatalf("exact mode skipped source %d", vertexID) + if _, ok := worker.VertexValue(10).(*betweennessContributionChunk); !ok { + t.Fatalf("offset %d did not expose a bounded contribution chunk", + worker.outgoingOffset) } + worker.AfterStep() + if worker.outgoingOffset != test.offset { + t.Fatalf("expected outgoing offset %d, got %d", + test.offset, worker.outgoingOffset) + } + context.Step++ + } + if stop := context.GetValue(betweennessStopValue).(serialize.SInt32); stop != 0 { + t.Fatalf("final chunk window did not stop: %d", stop) } } -func TestSumBetweennessPartials(t *testing.T) { - result, err := sumBetweennessPartials([]serialize.SliceFloat32{ - {1, 2, 3}, - {0.5, 1, 1.5}, - }, 3) - if err != nil { - t.Fatalf("sum partials: %v", err) +func TestBetweennessMasterCompletesOnlyAfterScatterDrains(t *testing.T) { + master := &BetweennessCentralityMaster{} + context := master.MakeContext(map[string]string{"compute.max_step": "10"}) + if err := master.Init(); err != nil { + t.Fatalf("master init: %v", err) } - expected := []float64{1.5, 3, 4.5} - for vertexID, value := range result { - if math.Abs(float64(value)-expected[vertexID]) > 1e-6 { - t.Fatalf("vertex %d expected %v, got %v", - vertexID, expected[vertexID], value) - } + context.CreateValue(betweennessStopValue, compute.ValueTypeInt32, + compute.CValueActionAggregate) + context.SetValue(betweennessStopValue, serialize.SInt32(1)) + if !master.Compute() { + t.Fatal("master stopped while a worker still had contribution chunks") + } + if complete := context.GetValue(betweennessCompleteValue).(serialize.SInt32); complete != 0 { + t.Fatalf("completion was published early: %d", complete) + } + + context.SetValue(betweennessStopValue, serialize.SInt32(0)) + if master.Compute() { + t.Fatal("master continued after every worker drained its chunks") } - if _, err := sumBetweennessPartials([]serialize.SliceFloat32{{1}}, 2); err == nil { - t.Fatal("mismatched partial length was accepted") + if complete := context.GetValue(betweennessCompleteValue).(serialize.SInt32); complete != 1 { + t.Fatalf("completion was not published: %d", complete) } } -func TestBetweennessOutputRequiresFinalGlobalVector(t *testing.T) { - worker := &BetweennessCentralityWorker{totalVertices: 3} +func TestBetweennessOutputRequiresScalarCompletionAndUsesLocalRange(t *testing.T) { + worker := &BetweennessCentralityWorker{ + localStart: 10, + localCount: 2, + localValues: []float64{1.5, 2.5}, + } context := worker.MakeContext(nil) - context.CreateValue(betweennessFinalValue, compute.ValueTypeSliceFloat32, + context.CreateValue(betweennessCompleteValue, compute.ValueTypeInt32, compute.CValueActionNormal) - context.SetValue(betweennessFinalValue, serialize.SliceFloat32{}) + context.SetValue(betweennessCompleteValue, serialize.SInt32(0)) if err := worker.ValidateOutput(); err == nil { - t.Fatal("expected missing global result to reject output") + t.Fatal("incomplete scalar state was accepted") } - expected := serialize.SliceFloat32{0, 1, 0} - context.SetValue(betweennessFinalValue, expected) + context.SetValue(betweennessCompleteValue, serialize.SInt32(1)) if err := worker.ValidateOutput(); err != nil { - t.Fatalf("completed global result rejected: %v", err) + t.Fatalf("completed local output rejected: %v", err) } - if len(worker.betweennessValues) != len(expected) || - worker.betweennessValues[1] != expected[1] { - t.Fatalf("global result was not installed: %v", worker.betweennessValues) + context.Output = true + for vertexID, expected := range map[uint32]float32{10: 1.5, 11: 2.5} { + value, ok := worker.VertexValue(vertexID).(*serialize.SFloat32) + if !ok { + t.Fatalf("vertex %d output type is %T", vertexID, worker.VertexValue(vertexID)) + } + if float32(*value) != expected { + t.Fatalf("vertex %d expected %v, got %v", vertexID, expected, *value) + } } }