From 545843cce76c5c13df9e335119a84b2215d6a530 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 18:13:42 +0800 Subject: [PATCH 1/5] fix(vermeer): make closeness traversal race-free --- vermeer/algorithms/closeness_centrality.go | 179 ++++++++++++------ .../algorithms/closeness_centrality_test.go | 89 +++++++++ 2 files changed, 211 insertions(+), 57 deletions(-) create mode 100644 vermeer/algorithms/closeness_centrality_test.go diff --git a/vermeer/algorithms/closeness_centrality.go b/vermeer/algorithms/closeness_centrality.go index b5742d4c8..527513844 100644 --- a/vermeer/algorithms/closeness_centrality.go +++ b/vermeer/algorithms/closeness_centrality.go @@ -19,7 +19,9 @@ package algorithms import ( "fmt" + "math" "math/rand" + "sync/atomic" "vermeer/apps/common" "vermeer/apps/compute" "vermeer/apps/options" @@ -32,8 +34,7 @@ func init() { Algorithms = append(Algorithms, new(ClosenessCentralityMaker)) } -//normalized closeness centrality - +// normalized closeness centrality type ClosenessCentralityMaker struct { compute.AlgorithmMakerBase } @@ -54,6 +55,52 @@ func (cc *ClosenessCentralityMaker) DataNeeded() []string { return []string{} } +func validateProbability(name string, value float64) error { + if math.IsNaN(value) || math.IsInf(value, 0) || value <= 0 || value > 1 { + return fmt.Errorf("%s must be finite and in (0, 1], got %v", name, value) + } + return nil +} + +func parseBooleanFlag(name string, value int) (bool, error) { + switch value { + case 0: + return false, nil + case 1: + return true, nil + default: + return false, fmt.Errorf("%s must be 0 or 1, got %d", name, value) + } +} + +func newClosenessDistanceMap(vertexID uint32, capacity int) map[uint32]int32 { + distances := make(map[uint32]int32, capacity+1) + distances[vertexID] = 0 + return distances +} + +func closenessScore(distances map[uint32]int32, totalVertices uint32, + wfImproved bool) serialize.SFloat32 { + if totalVertices <= 1 || len(distances) <= 1 { + return 0 + } + + var distanceSum int64 + for _, distance := range distances { + distanceSum += int64(distance) + } + if distanceSum <= 0 { + return 0 + } + + reachable := len(distances) - 1 + score := float64(reachable) / float64(distanceSum) + if wfImproved { + score *= float64(reachable) / float64(totalVertices-1) + } + return serialize.SFloat32(score) +} + type ClosenessCentralityWorker struct { compute.WorkerComputerBase wfImproved bool @@ -64,47 +111,65 @@ type ClosenessCentralityWorker struct { dfsOutsideEdge [][]serialize.SUint32 neighborEdges []serialize.SliceUint32 pMap []map[uint32]int32 + finished []bool + unfinished int64 } func (ccw *ClosenessCentralityWorker) VertexValue(i uint32) serialize.MarshalAble { + if ccw.WContext.Output { + return &ccw.closenessValues[i] + } if ccw.WContext.Step == 1 { return &ccw.neighborEdges[i] } - return &ccw.closenessValues[i] } +func (ccw *ClosenessCentralityWorker) ValidateOutput() error { + unfinished := atomic.LoadInt64(&ccw.unfinished) + if unfinished != 0 { + return fmt.Errorf("closeness_centrality did not converge for %d local vertices; increase compute.max_step", + unfinished) + } + return nil +} + func (ccw *ClosenessCentralityWorker) Init() error { - ccw.closenessValues = make([]serialize.SFloat32, ccw.WContext.GraphData.Vertex.TotalVertexCount()) - ccw.neighborEdges = make([]serialize.SliceUint32, ccw.WContext.GraphData.Vertex.TotalVertexCount()) + ccw.closenessValues = make([]serialize.SFloat32, + ccw.WContext.GraphData.Vertex.TotalVertexCount()) + ccw.neighborEdges = make([]serialize.SliceUint32, + ccw.WContext.GraphData.Vertex.TotalVertexCount()) ccw.dfsOutsideEdge = make([][]serialize.SUint32, ccw.WContext.GraphData.VertexCount) ccw.stopSum = make([]serialize.SInt32, ccw.WContext.Parallel) ccw.pMap = make([]map[uint32]int32, ccw.WContext.GraphData.VertexCount) for i := range ccw.pMap { - ccw.pMap[i] = make(map[uint32]int32, len(ccw.WContext.GraphData.Edges.GetInEdges(uint32(i)))) - ccw.pMap[i][uint32(i)] = 0 + localID := uint32(i) + vertexID := ccw.WContext.GraphData.VertIDStart + localID + ccw.pMap[i] = newClosenessDistanceMap(vertexID, + len(ccw.WContext.GraphData.Edges.GetInEdges(localID))) } ccw.tempEdges = make([][]serialize.SUint32, ccw.WContext.Parallel) for i := range ccw.tempEdges { ccw.tempEdges[i] = make([]serialize.SUint32, 0, 100) } - ccw.sampleRate = options.GetFloat(ccw.WContext.Params, "closeness_centrality.sample_rate") - if ccw.sampleRate > 1.0 || ccw.sampleRate <= 0 { - logrus.Errorf("The param closeness_centrality.sample_rate must be in (0.0, 1.0], actual got '%v", - ccw.sampleRate) - return fmt.Errorf("the param closeness_centrality.sample_rate must be in (0.0, 1.0], actual got '%v", - ccw.sampleRate) - } - //wf_improved:bool. default true - // If True, scale by the fraction of nodes reachable. This gives the - // Wasserman and Faust improved formula. For single component graphs - // it is the same as the original formula. - ccw.wfImproved = true - wfImproved := options.GetInt(ccw.WContext.Params, "closeness_centrality.wf_improved") - if wfImproved != 1 { - ccw.wfImproved = false - } - ccw.WContext.CreateValue("stop_sum", compute.ValueTypeInt32, compute.CValueActionAggregate) + ccw.finished = make([]bool, ccw.WContext.GraphData.VertexCount) + atomic.StoreInt64(&ccw.unfinished, int64(ccw.WContext.GraphData.VertexCount)) + + ccw.sampleRate = options.GetFloat(ccw.WContext.Params, + "closeness_centrality.sample_rate") + if err := validateProbability("closeness_centrality.sample_rate", ccw.sampleRate); err != nil { + return err + } + + wfImproved, err := parseBooleanFlag("closeness_centrality.wf_improved", + options.GetInt(ccw.WContext.Params, "closeness_centrality.wf_improved")) + if err != nil { + return err + } + ccw.wfImproved = wfImproved + + ccw.WContext.CreateValue("stop_sum", compute.ValueTypeInt32, + compute.CValueActionAggregate) ccw.WContext.SetValue("stop_sum", serialize.SInt32(0)) return nil } @@ -117,60 +182,60 @@ func (ccw *ClosenessCentralityWorker) BeforeStep() { func (ccw *ClosenessCentralityWorker) Compute(vertexID uint32, pID int) { vertID := vertexID - ccw.WContext.GraphData.VertIDStart + if ccw.finished[vertID] { + return + } + switch ccw.WContext.Step { case 1: inEdges := ccw.WContext.GraphData.Edges.GetInEdges(vertID) ccw.neighborEdges[vertexID] = make([]serialize.SUint32, 0, len(inEdges)) for _, edge := range inEdges { - if !ccw.sample() || uint32(edge) == vertexID { + if !ccw.sample() { + continue + } + if _, seen := ccw.pMap[vertID][uint32(edge)]; seen { continue } ccw.neighborEdges[vertexID] = append(ccw.neighborEdges[vertexID], edge) ccw.pMap[vertID][uint32(edge)] = 1 } ccw.dfsOutsideEdge[vertID] = ccw.neighborEdges[vertexID] - ccw.stopSum[pID] = 1 default: for _, value := range ccw.dfsOutsideEdge[vertID] { - for _, nId := range ccw.neighborEdges[value] { - if _, ok := ccw.pMap[vertID][uint32(nId)]; !ok { - if !ccw.sample() { - continue - } - ccw.pMap[vertID][uint32(nId)] = ccw.pMap[vertID][uint32(value)] + 1 - ccw.tempEdges[pID] = append(ccw.tempEdges[pID], nId) - ccw.stopSum[pID] = 1 + for _, neighborID := range ccw.neighborEdges[value] { + if _, seen := ccw.pMap[vertID][uint32(neighborID)]; seen { + continue } + ccw.pMap[vertID][uint32(neighborID)] = + ccw.pMap[vertID][uint32(value)] + 1 + ccw.tempEdges[pID] = append(ccw.tempEdges[pID], neighborID) } } - ccw.dfsOutsideEdge[vertID] = make([]serialize.SUint32, len(ccw.tempEdges[pID])) - copy(ccw.dfsOutsideEdge[vertID], ccw.tempEdges[pID]) + ccw.dfsOutsideEdge[vertID] = append([]serialize.SUint32(nil), + ccw.tempEdges[pID]...) + ccw.tempEdges[pID] = ccw.tempEdges[pID][:0] } - ccw.tempEdges[pID] = ccw.tempEdges[pID][0:0] - var stopSum serialize.SInt32 - for _, v := range ccw.stopSum { - stopSum += v - } - if stopSum == 0 || options.GetInt(ccw.WContext.Params, "compute.max_step") == int(ccw.WContext.Step) { - //计算closenessValues - var sum int32 - for _, v := range ccw.pMap[vertID] { - sum += v - } - if sum > 0 && ccw.WContext.GraphData.Vertex.TotalVertexCount() > 1 { - ccw.closenessValues[vertexID] = - serialize.SFloat32(len(ccw.pMap[vertID])-1) / serialize.SFloat32(sum) - if ccw.wfImproved { - ccw.closenessValues[vertexID] *= serialize.SFloat32(len(ccw.pMap[vertID])-1) / - serialize.SFloat32(ccw.WContext.GraphData.Vertex.TotalVertexCount()-1) - } - ccw.closenessValues[vertexID] *= serialize.SFloat32(ccw.sampleRate) - } + + if len(ccw.dfsOutsideEdge[vertID]) == 0 { + ccw.finishVertex(vertID, vertexID) + return + } + ccw.stopSum[pID] = 1 +} + +func (ccw *ClosenessCentralityWorker) finishVertex(vertID, vertexID uint32) { + if ccw.finished[vertID] { + return } + ccw.closenessValues[vertexID] = closenessScore(ccw.pMap[vertID], + ccw.WContext.GraphData.Vertex.TotalVertexCount(), ccw.wfImproved) + ccw.finished[vertID] = true + atomic.AddInt64(&ccw.unfinished, -1) } func (ccw *ClosenessCentralityWorker) sample() bool { - return rand.Float64() < ccw.sampleRate + return ccw.sampleRate == 1 || rand.Float64() < ccw.sampleRate } func (ccw *ClosenessCentralityWorker) AfterStep() { diff --git a/vermeer/algorithms/closeness_centrality_test.go b/vermeer/algorithms/closeness_centrality_test.go new file mode 100644 index 000000000..6917bc946 --- /dev/null +++ b/vermeer/algorithms/closeness_centrality_test.go @@ -0,0 +1,89 @@ +/* +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" + "sync/atomic" + "testing" +) + +func TestClosenessDistanceMapUsesGlobalVertexID(t *testing.T) { + distances := newClosenessDistanceMap(101, 3) + if _, ok := distances[101]; !ok { + t.Fatalf("global source vertex 101 is missing: %v", distances) + } + if _, ok := distances[0]; ok { + t.Fatalf("local vertex ID leaked into distance map: %v", distances) + } +} + +func TestClosenessScore(t *testing.T) { + tests := []struct { + name string + distances map[uint32]int32 + total uint32 + improved bool + expected float64 + }{ + {"connected", map[uint32]int32{0: 0, 1: 1, 2: 2}, 3, true, 2.0 / 3.0}, + {"disconnected_standard", map[uint32]int32{0: 0, 1: 1}, 4, false, 1}, + {"disconnected_wf", map[uint32]int32{0: 0, 1: 1}, 4, true, 1.0 / 3.0}, + {"isolated", map[uint32]int32{0: 0}, 4, true, 0}, + {"wide_distance_sum", map[uint32]int32{0: 0, 1: math.MaxInt32, 2: math.MaxInt32}, 3, false, + 2.0 / float64(int64(math.MaxInt32)*2)}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + actual := float64(closenessScore(test.distances, test.total, test.improved)) + if math.Abs(actual-test.expected) > 1e-6*math.Max(1, math.Abs(test.expected)) { + t.Fatalf("expected %v, got %v", test.expected, actual) + } + }) + } +} + +func TestClosenessParameterValidation(t *testing.T) { + for _, value := range []float64{0, -1, 1.1, math.NaN(), math.Inf(1)} { + if err := validateProbability("sample_rate", value); err == nil { + t.Fatalf("expected invalid probability %v to be rejected", value) + } + } + for _, value := range []float64{0.1, 1} { + if err := validateProbability("sample_rate", value); err != nil { + t.Fatalf("valid probability %v rejected: %v", value, err) + } + } + for _, value := range []int{-1, 2} { + if _, err := parseBooleanFlag("wf_improved", value); err == nil { + t.Fatalf("expected invalid boolean flag %d to be rejected", value) + } + } +} + +func TestClosenessRejectsPartialOutput(t *testing.T) { + worker := &ClosenessCentralityWorker{} + atomic.StoreInt64(&worker.unfinished, 1) + if err := worker.ValidateOutput(); err == nil { + t.Fatal("expected partial output to be rejected") + } + atomic.StoreInt64(&worker.unfinished, 0) + if err := worker.ValidateOutput(); err != nil { + t.Fatalf("completed output rejected: %v", err) + } +} From 8ce3b28aa689b5fd2f63017dc38477458c7cb701 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 18:30:40 +0800 Subject: [PATCH 2/5] fix(vermeer): validate output with final aggregate values --- vermeer/apps/worker/compute_bl.go | 11 ++++ .../worker/compute_output_validation_test.go | 52 +++++++++++++++++++ vermeer/apps/worker/grpc_handlers.go | 2 +- 3 files changed, 64 insertions(+), 1 deletion(-) diff --git a/vermeer/apps/worker/compute_bl.go b/vermeer/apps/worker/compute_bl.go index 9cb31b2fc..03484649d 100644 --- a/vermeer/apps/worker/compute_bl.go +++ b/vermeer/apps/worker/compute_bl.go @@ -453,6 +453,14 @@ func (cb *ComputeBl) StepEnd(taskID int32, workerName string) { } func (cb *ComputeBl) RunOutput(taskID int32) { + cb.runOutput(taskID, nil) +} + +func (cb *ComputeBl) RunOutputWithValues(taskID int32, computeValues map[string][]byte) { + cb.runOutput(taskID, computeValues) +} + +func (cb *ComputeBl) runOutput(taskID int32, computeValues map[string][]byte) { defer func() { if r := recover(); r != nil { cb.SetStatusError(taskID, fmt.Sprintf("RunOutput panic recover panic:%v, stack message: %s", r, @@ -465,6 +473,9 @@ func (cb *ComputeBl) RunOutput(taskID int32) { if !cb.CheckAction(computeTask) { return } + if computeValues != nil { + computeTask.ComputeWorker.Context().UnmarshalValues(computeValues) + } if validator, ok := computeTask.ComputeWorker.(interface{ ValidateOutput() error }); ok { if err := validator.ValidateOutput(); err != nil { cb.SetStatusError(taskID, err.Error()) diff --git a/vermeer/apps/worker/compute_output_validation_test.go b/vermeer/apps/worker/compute_output_validation_test.go index 91c08f4a8..87379a0d6 100644 --- a/vermeer/apps/worker/compute_output_validation_test.go +++ b/vermeer/apps/worker/compute_output_validation_test.go @@ -27,6 +27,7 @@ import ( "vermeer/apps/compute" "vermeer/apps/options" pb "vermeer/apps/protos" + "vermeer/apps/serialize" "vermeer/apps/structure" "github.com/prometheus/client_golang/prometheus" @@ -43,6 +44,16 @@ func (w *invalidOutputWorker) ValidateOutput() error { return errors.New("increase compute.max_step") } +type finalValueOutputWorker struct { + invalidOutputWorker + seen serialize.SInt32 +} + +func (w *finalValueOutputWorker) ValidateOutput() error { + w.seen = w.WContext.GetValue("is_continue").(serialize.SInt32) + return errors.New("stop after validation") +} + type outputStatusClient struct { pb.MasterClient req *pb.ComputeTaskStatusReq @@ -122,3 +133,44 @@ func TestRunOutputValidatesBeforeWriting(t *testing.T) { } }) } + +func TestRunOutputAppliesFinalAggregateValuesBeforeValidation(t *testing.T) { + previousOptions := options.AllOptionsMap + previousRunning := common.PrometheusMetrics.TaskRunningCnt + previousMaster := ServiceWorker.MasterClient + options.Init() + ComputeTaskMgr.Init() + common.PrometheusMetrics.TaskRunningCnt = prometheus.NewGaugeVec( + prometheus.GaugeOpts{Name: "test_output_final_values_running"}, []string{"type"}) + t.Cleanup(func() { + ComputeTaskMgr.Init() + options.AllOptionsMap = previousOptions + common.PrometheusMetrics.TaskRunningCnt = previousRunning + ServiceWorker.MasterClient = previousMaster + }) + + status := new(outputStatusClient) + ServiceWorker.MasterClient = status + task := &structure.TaskInfo{ID: 9350, Type: structure.TaskTypeCompute, + Action: structure.ActionDoNoting, Params: map[string]string{"compute.algorithm": "sssp"}} + worker := new(finalValueOutputWorker) + ctx := worker.MakeContext(nil) + ctx.CreateValue("is_continue", compute.ValueTypeInt32, compute.CValueActionAggregate) + ctx.SetValue("is_continue", serialize.SInt32(0)) + ct := ComputeTaskMgr.MakeTask(task) + ct.ComputeWorker = worker + + finalValues := &compute.CContext{CValues: make(map[string]*compute.CValue)} + finalValues.CreateValue("is_continue", compute.ValueTypeInt32, compute.CValueActionAggregate) + finalValues.SetValue("is_continue", serialize.SInt32(1)) + new(ComputeBl).RunOutputWithValues(task.ID, finalValues.MarshalValues()) + + if worker.seen != 1 { + t.Fatalf("validator saw stale aggregate value %d, want 1", worker.seen) + } + if task.State != structure.TaskStateError || status.req == nil || + status.req.ErrorMsg != "stop after validation" { + t.Fatalf("expected validation failure after final-value refresh, state=%s request=%+v", + task.State, status.req) + } +} diff --git a/vermeer/apps/worker/grpc_handlers.go b/vermeer/apps/worker/grpc_handlers.go index c4a8366ee..c564361f8 100644 --- a/vermeer/apps/worker/grpc_handlers.go +++ b/vermeer/apps/worker/grpc_handlers.go @@ -403,7 +403,7 @@ func (rh *SuperStepHandler) HandleSuperStep() { } cb := ComputeBl{} if resp.Output { - go cb.RunOutput(resp.TaskId) + go cb.RunOutputWithValues(resp.TaskId, resp.ComputeValues) } else { go cb.RunSuperStep(resp.TaskId, resp.ComputeValues) } From 8c9adde22e72aa0ce1a08bbae2307615e902167c Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 22:30:36 +0800 Subject: [PATCH 3/5] fix(vermeer): make closeness completion globally atomic --- vermeer/algorithms/closeness_centrality.go | 107 ++++++++++++--------- 1 file changed, 59 insertions(+), 48 deletions(-) diff --git a/vermeer/algorithms/closeness_centrality.go b/vermeer/algorithms/closeness_centrality.go index 527513844..40115a24c 100644 --- a/vermeer/algorithms/closeness_centrality.go +++ b/vermeer/algorithms/closeness_centrality.go @@ -30,6 +30,8 @@ import ( "github.com/sirupsen/logrus" ) +const closenessUnfinishedValue = "closeness_unfinished" + func init() { Algorithms = append(Algorithms, new(ClosenessCentralityMaker)) } @@ -105,7 +107,6 @@ type ClosenessCentralityWorker struct { compute.WorkerComputerBase wfImproved bool sampleRate float64 - stopSum []serialize.SInt32 closenessValues []serialize.SFloat32 tempEdges [][]serialize.SUint32 dfsOutsideEdge [][]serialize.SUint32 @@ -126,22 +127,42 @@ func (ccw *ClosenessCentralityWorker) VertexValue(i uint32) serialize.MarshalAbl } func (ccw *ClosenessCentralityWorker) ValidateOutput() error { - unfinished := atomic.LoadInt64(&ccw.unfinished) + cValue, ok := ccw.WContext.CValues[closenessUnfinishedValue] + if !ok || cValue.Value == nil { + return fmt.Errorf("closeness_centrality output is missing global convergence state") + } + unfinished, ok := cValue.Value.(serialize.SInt64) + if !ok { + return fmt.Errorf("closeness_centrality output has invalid convergence state %T", + cValue.Value) + } if unfinished != 0 { - return fmt.Errorf("closeness_centrality did not converge for %d local vertices; increase compute.max_step", + return fmt.Errorf("closeness_centrality did not converge for %d vertices; increase compute.max_step", unfinished) } return nil } func (ccw *ClosenessCentralityWorker) Init() error { - ccw.closenessValues = make([]serialize.SFloat32, - ccw.WContext.GraphData.Vertex.TotalVertexCount()) - ccw.neighborEdges = make([]serialize.SliceUint32, - ccw.WContext.GraphData.Vertex.TotalVertexCount()) - ccw.dfsOutsideEdge = make([][]serialize.SUint32, ccw.WContext.GraphData.VertexCount) - ccw.stopSum = make([]serialize.SInt32, ccw.WContext.Parallel) - ccw.pMap = make([]map[uint32]int32, ccw.WContext.GraphData.VertexCount) + sampleRate := options.GetFloat(ccw.WContext.Params, + "closeness_centrality.sample_rate") + if err := validateProbability("closeness_centrality.sample_rate", sampleRate); err != nil { + return err + } + wfImproved, err := parseBooleanFlag("closeness_centrality.wf_improved", + options.GetInt(ccw.WContext.Params, "closeness_centrality.wf_improved")) + if err != nil { + return err + } + ccw.sampleRate = sampleRate + ccw.wfImproved = wfImproved + + totalVertices := ccw.WContext.GraphData.Vertex.TotalVertexCount() + localVertices := ccw.WContext.GraphData.VertexCount + ccw.closenessValues = make([]serialize.SFloat32, totalVertices) + ccw.neighborEdges = make([]serialize.SliceUint32, totalVertices) + ccw.dfsOutsideEdge = make([][]serialize.SUint32, localVertices) + ccw.pMap = make([]map[uint32]int32, localVertices) for i := range ccw.pMap { localID := uint32(i) vertexID := ccw.WContext.GraphData.VertIDStart + localID @@ -152,34 +173,15 @@ func (ccw *ClosenessCentralityWorker) Init() error { for i := range ccw.tempEdges { ccw.tempEdges[i] = make([]serialize.SUint32, 0, 100) } - ccw.finished = make([]bool, ccw.WContext.GraphData.VertexCount) - atomic.StoreInt64(&ccw.unfinished, int64(ccw.WContext.GraphData.VertexCount)) - - ccw.sampleRate = options.GetFloat(ccw.WContext.Params, - "closeness_centrality.sample_rate") - if err := validateProbability("closeness_centrality.sample_rate", ccw.sampleRate); err != nil { - return err - } - - wfImproved, err := parseBooleanFlag("closeness_centrality.wf_improved", - options.GetInt(ccw.WContext.Params, "closeness_centrality.wf_improved")) - if err != nil { - return err - } - ccw.wfImproved = wfImproved + ccw.finished = make([]bool, localVertices) + atomic.StoreInt64(&ccw.unfinished, int64(localVertices)) - ccw.WContext.CreateValue("stop_sum", compute.ValueTypeInt32, + ccw.WContext.CreateValue(closenessUnfinishedValue, compute.ValueTypeInt64, compute.CValueActionAggregate) - ccw.WContext.SetValue("stop_sum", serialize.SInt32(0)) + ccw.WContext.SetValue(closenessUnfinishedValue, serialize.SInt64(localVertices)) return nil } -func (ccw *ClosenessCentralityWorker) BeforeStep() { - for i := range ccw.stopSum { - ccw.stopSum[i] = 0 - } -} - func (ccw *ClosenessCentralityWorker) Compute(vertexID uint32, pID int) { vertID := vertexID - ccw.WContext.GraphData.VertIDStart if ccw.finished[vertID] { @@ -216,12 +218,29 @@ func (ccw *ClosenessCentralityWorker) Compute(vertexID uint32, pID int) { ccw.tempEdges[pID]...) ccw.tempEdges[pID] = ccw.tempEdges[pID][:0] } +} - if len(ccw.dfsOutsideEdge[vertID]) == 0 { - ccw.finishVertex(vertID, vertexID) - return +func (ccw *ClosenessCentralityWorker) AfterStep() { + for localID := range ccw.finished { + if ccw.finished[localID] || ccw.frontierCanExpand(uint32(localID)) { + continue + } + vertexID := ccw.WContext.GraphData.VertIDStart + uint32(localID) + ccw.finishVertex(uint32(localID), vertexID) + } + ccw.WContext.SetValue(closenessUnfinishedValue, + serialize.SInt64(atomic.LoadInt64(&ccw.unfinished))) +} + +func (ccw *ClosenessCentralityWorker) frontierCanExpand(vertID uint32) bool { + for _, frontierID := range ccw.dfsOutsideEdge[vertID] { + for _, neighborID := range ccw.neighborEdges[frontierID] { + if _, seen := ccw.pMap[vertID][uint32(neighborID)]; !seen { + return true + } + } } - ccw.stopSum[pID] = 1 + return false } func (ccw *ClosenessCentralityWorker) finishVertex(vertID, vertexID uint32) { @@ -238,14 +257,6 @@ func (ccw *ClosenessCentralityWorker) sample() bool { return ccw.sampleRate == 1 || rand.Float64() < ccw.sampleRate } -func (ccw *ClosenessCentralityWorker) AfterStep() { - var stopSum serialize.SInt32 - for i := range ccw.stopSum { - stopSum += ccw.stopSum[i] - } - ccw.WContext.SetValue("stop_sum", stopSum) -} - func (ccw *ClosenessCentralityWorker) OutputValueType() string { return common.HgValueTypeFloat } @@ -255,7 +266,7 @@ type ClosenessCentralityMaster struct { } func (ccm *ClosenessCentralityMaster) Compute() bool { - stopSum := ccm.MContext.GetValue("stop_sum") - logrus.Infof("stopSum:%v", stopSum) - return stopSum.(serialize.SInt32) != 0 + unfinished := ccm.MContext.GetValue(closenessUnfinishedValue).(serialize.SInt64) + logrus.Infof("closeness unfinished vertices: %v", unfinished) + return unfinished != 0 } From fb03d9c698ce7747208c476e181469946d0d8af6 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 22:30:58 +0800 Subject: [PATCH 4/5] test(vermeer): exercise distributed closeness lifecycle --- .../algorithms/closeness_centrality_test.go | 164 ++++++++++++++++-- 1 file changed, 154 insertions(+), 10 deletions(-) diff --git a/vermeer/algorithms/closeness_centrality_test.go b/vermeer/algorithms/closeness_centrality_test.go index 6917bc946..77fa746db 100644 --- a/vermeer/algorithms/closeness_centrality_test.go +++ b/vermeer/algorithms/closeness_centrality_test.go @@ -9,18 +9,22 @@ 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. +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" + "strings" "sync/atomic" "testing" + "vermeer/apps/compute" + "vermeer/apps/serialize" + "vermeer/apps/structure" ) func TestClosenessDistanceMapUsesGlobalVertexID(t *testing.T) { @@ -76,14 +80,154 @@ func TestClosenessParameterValidation(t *testing.T) { } } -func TestClosenessRejectsPartialOutput(t *testing.T) { +func TestClosenessInitValidatesBeforeAllocatingGraphState(t *testing.T) { + tests := []struct { + name string + params map[string]string + want string + }{ + { + name: "sample_rate", + params: map[string]string{ + "closeness_centrality.sample_rate": "0", + "closeness_centrality.wf_improved": "1", + }, + want: "sample_rate", + }, + { + name: "wf_improved", + params: map[string]string{ + "closeness_centrality.sample_rate": "1", + "closeness_centrality.wf_improved": "2", + }, + want: "wf_improved", + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + worker := &ClosenessCentralityWorker{} + ctx := worker.MakeContext(test.params) + // Invalid parameters must be rejected before Init dereferences graph + // storage or allocates per-vertex traversal state. + ctx.GraphData = &structure.GraphData{} + ctx.Parallel = 1 + + err := worker.Init() + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("expected %s validation error, got %v", test.want, err) + } + }) + } +} + +func TestClosenessTraversalFinishesAtEccentricity(t *testing.T) { + const maxStep = 4 + params := map[string]string{ + "closeness_centrality.sample_rate": "1", + "closeness_centrality.wf_improved": "1", + "compute.max_step": "4", + } + worker := &ClosenessCentralityWorker{} - atomic.StoreInt64(&worker.unfinished, 1) - if err := worker.ValidateOutput(); err == nil { - t.Fatal("expected partial output to be rejected") + ctx := worker.MakeContext(params) + ctx.GraphData = newClosenessPartitionGraph() + ctx.Parallel = 1 + if err := worker.Init(); err != nil { + t.Fatalf("worker init failed: %v", err) } - atomic.StoreInt64(&worker.unfinished, 0) + + master := &ClosenessCentralityMaster{} + masterContext := master.MakeContext(params) + masterContext.CreateValue(closenessUnfinishedValue, compute.ValueTypeInt64, + compute.CValueActionAggregate) + + expectedUnfinished := []serialize.SInt64{3, 2, 1, 0} + for step := 1; step <= maxStep; step++ { + ctx.Step = int32(step) + worker.BeforeStep() + for vertexID := uint32(2); vertexID < 5; vertexID++ { + worker.Compute(vertexID, 0) + } + if step == 1 { + // Step-1 scatter has made the two remote adjacency lists visible + // before AfterStep decides whether the current frontier can grow. + worker.neighborEdges[0] = nil + worker.neighborEdges[1] = serialize.SliceUint32{0} + } + worker.AfterStep() + + gotUnfinished := ctx.GetValue(closenessUnfinishedValue).(serialize.SInt64) + if gotUnfinished != expectedUnfinished[step-1] { + t.Fatalf("step %d: expected %d unfinished sources, got %d", + step, expectedUnfinished[step-1], gotUnfinished) + } + masterContext.SetValue(closenessUnfinishedValue, gotUnfinished) + gotContinue := master.Compute() + wantContinue := step < maxStep + if gotContinue != wantContinue { + t.Fatalf("step %d: expected continue=%v, got %v", + step, wantContinue, gotContinue) + } + } + if err := worker.ValidateOutput(); err != nil { - t.Fatalf("completed output rejected: %v", err) + t.Fatalf("complete traversal rejected at eccentricity: %v", err) + } + assertFloat32Close(t, worker.closenessValues[2], 1.0/3.0) + assertFloat32Close(t, worker.closenessValues[3], 3.0/8.0) + assertFloat32Close(t, worker.closenessValues[4], 2.0/5.0) +} + +func TestClosenessOutputUsesGlobalUnfinishedCount(t *testing.T) { + workers := []*ClosenessCentralityWorker{ + newClosenessValidationWorker(0, 1), + newClosenessValidationWorker(1, 1), + } + for i, worker := range workers { + if err := worker.ValidateOutput(); err == nil { + t.Fatalf("worker %d accepted output while another partition was unfinished", i) + } + } + + for i, worker := range workers { + worker.WContext.SetValue(closenessUnfinishedValue, serialize.SInt64(0)) + if err := worker.ValidateOutput(); err != nil { + t.Fatalf("worker %d rejected globally complete output: %v", i, err) + } + } +} + +func newClosenessPartitionGraph() *structure.GraphData { + return &structure.GraphData{ + VertIDStart: 2, + VertexCount: 3, + Vertex: &structure.VertexMem{ + TotalVertex: make([]structure.Vertex, 5), + }, + Edges: &structure.EdgeMem{ + // Global path: 0 -> 1 -> 2 -> 3 -> 4. This partition owns 2..4. + InEdges: serialize.TwoDimSliceUint32{ + {1}, + {2}, + {3}, + }, + }, + } +} + +func newClosenessValidationWorker(localUnfinished, globalUnfinished int64) *ClosenessCentralityWorker { + worker := &ClosenessCentralityWorker{} + ctx := worker.MakeContext(nil) + ctx.CreateValue(closenessUnfinishedValue, compute.ValueTypeInt64, + compute.CValueActionAggregate) + ctx.SetValue(closenessUnfinishedValue, serialize.SInt64(globalUnfinished)) + atomic.StoreInt64(&worker.unfinished, localUnfinished) + return worker +} + +func assertFloat32Close(t *testing.T, actual serialize.SFloat32, expected float64) { + t.Helper() + if math.Abs(float64(actual)-expected) > 1e-6 { + t.Fatalf("expected %v, got %v", expected, actual) } } From 0246855bb30d7c73d1807da408d493d51466b51e Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 22:40:20 +0800 Subject: [PATCH 5/5] test(vermeer): initialize closeness options in lifecycle tests --- vermeer/algorithms/closeness_centrality_test.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/vermeer/algorithms/closeness_centrality_test.go b/vermeer/algorithms/closeness_centrality_test.go index 77fa746db..5a03162c5 100644 --- a/vermeer/algorithms/closeness_centrality_test.go +++ b/vermeer/algorithms/closeness_centrality_test.go @@ -23,10 +23,20 @@ import ( "sync/atomic" "testing" "vermeer/apps/compute" + "vermeer/apps/options" "vermeer/apps/serialize" "vermeer/apps/structure" ) +func initClosenessTestOptions(t *testing.T) { + t.Helper() + previous := options.AllOptionsMap + options.Init() + t.Cleanup(func() { + options.AllOptionsMap = previous + }) +} + func TestClosenessDistanceMapUsesGlobalVertexID(t *testing.T) { distances := newClosenessDistanceMap(101, 3) if _, ok := distances[101]; !ok { @@ -81,6 +91,7 @@ func TestClosenessParameterValidation(t *testing.T) { } func TestClosenessInitValidatesBeforeAllocatingGraphState(t *testing.T) { + initClosenessTestOptions(t) tests := []struct { name string params map[string]string @@ -121,6 +132,7 @@ func TestClosenessInitValidatesBeforeAllocatingGraphState(t *testing.T) { } func TestClosenessTraversalFinishesAtEccentricity(t *testing.T) { + initClosenessTestOptions(t) const maxStep = 4 params := map[string]string{ "closeness_centrality.sample_rate": "1",