From abbbae8e535f4302d28c2b0cadd13ba2bda26bc3 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 18:00:35 +0800 Subject: [PATCH 1/6] fix(vermeer): publish Louvain output atomically --- vermeer/algorithms/louvain.go | 10 ++++--- vermeer/algorithms/louvain_output_test.go | 32 ++++++++++++----------- vermeer/algorithms/louvain_weighted.go | 10 ++++--- 3 files changed, 31 insertions(+), 21 deletions(-) diff --git a/vermeer/algorithms/louvain.go b/vermeer/algorithms/louvain.go index 4223e3374..92fa1ba6d 100644 --- a/vermeer/algorithms/louvain.go +++ b/vermeer/algorithms/louvain.go @@ -111,6 +111,7 @@ type LouvainWorker struct { firstStep bool firstStepKI []float64 firstStepCommID []serialize.SUint32 + outputReady int32 } func (lw *LouvainWorker) VertexValue(i uint32) serialize.MarshalAble { @@ -126,8 +127,8 @@ func (lw *LouvainWorker) VertexValue(i uint32) serialize.MarshalAble { // ValidateOutput rejects early termination before community IDs are available. func (lw *LouvainWorker) ValidateOutput() error { - if lw.firstStep { - return fmt.Errorf("louvain did not finish its first pass; increase compute.max_step") + if atomic.LoadInt32(&lw.outputReady) == 0 { + return fmt.Errorf("louvain did not finish publishing community IDs; increase compute.max_step") } return nil } @@ -160,6 +161,7 @@ func (lw *LouvainWorker) Init() error { lw.emptyComm = serialize.SUint32(lw.WContext.GraphData.Vertex.TotalVertexCount() + 1) lw.firstStep = true + atomic.StoreInt32(&lw.outputReady, 0) return nil } @@ -261,11 +263,12 @@ func (lw *LouvainWorker) BeforeStep() { } update := lw.WContext.GetValue("update").(serialize.SInt32) if update > 0 { + atomic.StoreInt32(&lw.outputReady, 0) lw.deleteEmptyComm() if lw.firstStep { - lw.firstStep = false //初始化node lw.initLouvainNode() + lw.firstStep = false //free memory lw.firstStepCommID = nil lw.firstStepKI = nil @@ -274,6 +277,7 @@ func (lw *LouvainWorker) BeforeStep() { //生成新图 lw.genNewGraph() } + atomic.StoreInt32(&lw.outputReady, 1) lw.WContext.SetValue("mod_value", serialize.SFloat32(lw.calModularity())) } else { lw.optimizeMem(currCommIDs) diff --git a/vermeer/algorithms/louvain_output_test.go b/vermeer/algorithms/louvain_output_test.go index d1c91f575..9073c4daa 100644 --- a/vermeer/algorithms/louvain_output_test.go +++ b/vermeer/algorithms/louvain_output_test.go @@ -22,25 +22,27 @@ import ( "testing" ) -func TestLouvainRejectsOutputBeforeFirstPass(t *testing.T) { - for _, test := range []struct { - name string - worker interface{ ValidateOutput() error } +func TestLouvainRejectsOutputUntilCommunityIDsArePublished(t *testing.T) { + louvain := &LouvainWorker{firstStep: false} + weighted := &LouvainWeightedWorker{firstStep: false} + tests := []struct { + name string + validate func() error + publish func() }{ - {"louvain", &LouvainWorker{firstStep: true}}, - {"louvain_weighted", &LouvainWeightedWorker{firstStep: true}}, - } { + {"louvain", louvain.ValidateOutput, func() { louvain.outputReady = 1 }}, + {"louvain_weighted", weighted.ValidateOutput, func() { weighted.outputReady = 1 }}, + } + for _, test := range tests { t.Run(test.name, func(t *testing.T) { - err := test.worker.ValidateOutput() + err := test.validate() if err == nil || !strings.Contains(err.Error(), "increase compute.max_step") { - t.Fatalf("expected actionable early-output error, got %v", err) + t.Fatalf("expected actionable unpublished-output error, got %v", err) + } + test.publish() + if err := test.validate(); err != nil { + t.Fatalf("published output rejected: %v", err) } }) } - if err := (&LouvainWorker{firstStep: false}).ValidateOutput(); err != nil { - t.Fatalf("completed pass rejected: %v", err) - } - if err := (&LouvainWeightedWorker{firstStep: false}).ValidateOutput(); err != nil { - t.Fatalf("completed weighted pass rejected: %v", err) - } } diff --git a/vermeer/algorithms/louvain_weighted.go b/vermeer/algorithms/louvain_weighted.go index d26ebbad1..1da23c050 100644 --- a/vermeer/algorithms/louvain_weighted.go +++ b/vermeer/algorithms/louvain_weighted.go @@ -92,6 +92,7 @@ type LouvainWeightedWorker struct { firstStep bool firstStepKI []float64 firstStepCommID []serialize.SUint32 + outputReady int32 // property weight useProperty bool @@ -114,8 +115,8 @@ func (lw *LouvainWeightedWorker) VertexValue(i uint32) serialize.MarshalAble { // ValidateOutput rejects early termination before community IDs are available. func (lw *LouvainWeightedWorker) ValidateOutput() error { - if lw.firstStep { - return fmt.Errorf("louvain_weighted did not finish its first pass; increase compute.max_step") + if atomic.LoadInt32(&lw.outputReady) == 0 { + return fmt.Errorf("louvain_weighted did not finish publishing community IDs; increase compute.max_step") } return nil } @@ -149,6 +150,7 @@ func (lw *LouvainWeightedWorker) Init() error { lw.emptyComm = serialize.SUint32(lw.WContext.GraphData.Vertex.TotalVertexCount() + 1) lw.firstStep = true + atomic.StoreInt32(&lw.outputReady, 0) // load edge weight lw.edgeProperty = options.GetString(lw.WContext.Params, "louvain.edge_weight_property") @@ -276,11 +278,12 @@ func (lw *LouvainWeightedWorker) BeforeStep() { } update := lw.WContext.GetValue("update").(serialize.SInt32) if update > 0 { + atomic.StoreInt32(&lw.outputReady, 0) lw.deleteEmptyComm() if lw.firstStep { - lw.firstStep = false //初始化node lw.initLouvainNode() + lw.firstStep = false //free memory lw.firstStepCommID = nil lw.firstStepKI = nil @@ -289,6 +292,7 @@ func (lw *LouvainWeightedWorker) BeforeStep() { //生成新图 lw.genNewGraph() } + atomic.StoreInt32(&lw.outputReady, 1) lw.WContext.SetValue("mod_value", serialize.SFloat32(lw.calModularity())) } else { if rand.Float32() < 0.5 { From 9074fa5d9e50a8d4b801e17778f7f5ea91978b03 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 18:20:57 +0800 Subject: [PATCH 2/6] fix(vermeer): handle Louvain degenerate inputs --- vermeer/algorithms/louvain.go | 55 ++++++++- vermeer/algorithms/louvain_validation_test.go | 104 ++++++++++++++++++ vermeer/algorithms/louvain_weighted.go | 65 ++++++++++- 3 files changed, 216 insertions(+), 8 deletions(-) create mode 100644 vermeer/algorithms/louvain_validation_test.go diff --git a/vermeer/algorithms/louvain.go b/vermeer/algorithms/louvain.go index 92fa1ba6d..cd0e3df7d 100644 --- a/vermeer/algorithms/louvain.go +++ b/vermeer/algorithms/louvain.go @@ -81,6 +81,35 @@ type community struct { sigmaTot float64 } +func validatePositiveFinite(name string, value float64) error { + if math.IsNaN(value) || math.IsInf(value, 0) || value <= 0 { + return fmt.Errorf("%s must be finite and greater than 0, got %v", name, value) + } + return nil +} + +func validateNonNegativeFinite(name string, value float64) error { + if math.IsNaN(value) || math.IsInf(value, 0) || value < 0 { + return fmt.Errorf("%s must be finite and non-negative, got %v", name, value) + } + return nil +} + +func validatePositiveInt(name string, value int) error { + if value <= 0 { + return fmt.Errorf("%s must be greater than 0, got %d", name, value) + } + return nil +} + +func identityCommunityIDs(total uint32) []serialize.SUint32 { + ids := make([]serialize.SUint32, total) + for i := range ids { + ids[i] = serialize.SUint32(i) + } + return ids +} + type LouvainWorker struct { compute.WorkerComputerBase //nodeID @@ -139,6 +168,9 @@ func (lw *LouvainWorker) Init() error { lw.neighborEdges = make([]serialize.SliceUint32, lw.WContext.GraphData.Vertex.TotalVertexCount()) lw.resolution = options.GetFloat(lw.WContext.Params, "louvain.resolution") + if err := validatePositiveFinite("louvain.resolution", lw.resolution); err != nil { + return err + } lw.parallel = lw.WContext.Parallel if lw.parallel <= 0 { logrus.Infof("parallel value must be larger than 0, get: %v, set to defalut value :1", lw.parallel) @@ -179,6 +211,18 @@ func (lw *LouvainWorker) BeforeStep() { lw.firstStepKI[vertexID] = float64(len(edges)) lw.firstStepCommID[vertexID] = serialize.SUint32(vertexID) } + if lw.edgeNums == 0 { + lw.nodeID = identityCommunityIDs(lw.WContext.GraphData.Vertex.TotalVertexCount()) + lw.nodes = make(map[serialize.SUint32]*louvainNode) + lw.communities = make(map[serialize.SUint32]*community) + lw.firstStep = false + lw.firstStepKI = nil + lw.firstStepCommID = nil + lw.neighborEdges = nil + lw.WContext.SetValue("mod_value", serialize.SFloat32(0)) + atomic.StoreInt32(&lw.outputReady, 1) + return + } lw.resolution /= lw.edgeNums lw.WContext.SetValue("mod_value", serialize.SFloat32(-1)) } else if lw.WContext.Step > 2 { @@ -535,10 +579,7 @@ func (lw *LouvainWorker) initLouvainNode() { //gen a new graph for new step //可以并行 lw.nodes = make(map[serialize.SUint32]*louvainNode, len(lw.communities)) - lw.nodeID = make([]serialize.SUint32, lw.WContext.GraphData.Vertex.TotalVertexCount()) - for i := range lw.nodeID { - lw.nodeID[i] = serialize.SUint32(i) - } + lw.nodeID = identityCommunityIDs(lw.WContext.GraphData.Vertex.TotalVertexCount()) locker := &sync.Mutex{} commIDs := make([]serialize.SUint32, 0, len(lw.communities)) for commID := range lw.communities { @@ -667,7 +708,13 @@ type LouvainMaster struct { func (lm *LouvainMaster) Init() error { lm.threshold = options.GetFloat(lm.MContext.Params, "louvain.threshold") + if err := validateNonNegativeFinite("louvain.threshold", lm.threshold); err != nil { + return err + } lm.maxStep = options.GetInt(lm.MContext.Params, "louvain.step") + if err := validatePositiveInt("louvain.step", lm.maxStep); err != nil { + return err + } lm.prevModValue = math.MinInt32 lm.louvainStep = 1 return nil diff --git a/vermeer/algorithms/louvain_validation_test.go b/vermeer/algorithms/louvain_validation_test.go new file mode 100644 index 000000000..5457ffa46 --- /dev/null +++ b/vermeer/algorithms/louvain_validation_test.go @@ -0,0 +1,104 @@ +/* +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/compute" + "vermeer/apps/serialize" + "vermeer/apps/structure" +) + +func TestIdentityCommunityIDs(t *testing.T) { + ids := identityCommunityIDs(4) + if len(ids) != 4 { + t.Fatalf("expected 4 IDs, got %d", len(ids)) + } + for i, id := range ids { + if uint32(id) != uint32(i) { + t.Fatalf("vertex %d assigned to community %d", i, id) + } + } +} + +func TestLouvainFloatParameterValidation(t *testing.T) { + invalidPositive := []float64{0, -1, math.NaN(), math.Inf(1), math.Inf(-1)} + for _, value := range invalidPositive { + if err := validatePositiveFinite("value", value); err == nil { + t.Fatalf("expected positive validation to reject %v", value) + } + } + if err := validatePositiveFinite("value", 1); err != nil { + t.Fatalf("valid positive value rejected: %v", err) + } + + invalidNonNegative := []float64{-1, math.NaN(), math.Inf(1), math.Inf(-1)} + for _, value := range invalidNonNegative { + if err := validateNonNegativeFinite("value", value); err == nil { + t.Fatalf("expected non-negative validation to reject %v", value) + } + } + for _, value := range []float64{0, 1} { + if err := validateNonNegativeFinite("value", value); err != nil { + t.Fatalf("valid non-negative value %v rejected: %v", value, err) + } + } +} + +func TestLouvainStepValidation(t *testing.T) { + for _, value := range []int{0, -1} { + if err := validatePositiveInt("louvain.step", value); err == nil { + t.Fatalf("expected step validation to reject %d", value) + } + } + if err := validatePositiveInt("louvain.step", 1); err != nil { + t.Fatalf("valid step rejected: %v", err) + } +} + +func TestWeightedLouvainSkipsEffectivelyIsolatedVertex(t *testing.T) { + edges := &structure.EdgeMem{ + InEdges: serialize.TwoDimSliceUint32{{0}}, + OutEdges: serialize.TwoDimSliceUint32{{}}, + } + worker := &LouvainWeightedWorker{ + WorkerComputerBase: compute.WorkerComputerBase{ + WContext: &compute.CWContext{ + Step: 3, + GraphData: &structure.GraphData{ + VertIDStart: 0, + VertexCount: 1, + Edges: edges, + }, + }, + }, + firstStep: true, + firstStepKI: make([]float64, 1), + firstStepCommID: make([]serialize.SUint32, 1), + neighborEdges: []serialize.MapUint32Float32{make(serialize.MapUint32Float32)}, + communities: make(map[serialize.SUint32]*community), + node2comm: []map[serialize.SUint32]serialize.SUint32{make(map[serialize.SUint32]serialize.SUint32)}, + } + + worker.Compute(0, 0) + if len(worker.node2comm[0]) != 0 { + t.Fatalf("effectively isolated vertex was scheduled for a community move: %v", + worker.node2comm[0]) + } +} diff --git a/vermeer/algorithms/louvain_weighted.go b/vermeer/algorithms/louvain_weighted.go index 1da23c050..ae5553947 100644 --- a/vermeer/algorithms/louvain_weighted.go +++ b/vermeer/algorithms/louvain_weighted.go @@ -128,6 +128,9 @@ func (lw *LouvainWeightedWorker) Init() error { lw.inEdges = make([]serialize.MapUint32Float32, lw.WContext.GraphData.Vertex.TotalVertexCount()) lw.resolution = options.GetFloat(lw.WContext.Params, "louvain.resolution") + if err := validatePositiveFinite("louvain.resolution", lw.resolution); err != nil { + return err + } lw.parallel = lw.WContext.Parallel if lw.parallel <= 0 { logrus.Infof("parallel value must be larger than 0, get: %v, set to defalut value :1", lw.parallel) @@ -168,6 +171,32 @@ func (lw *LouvainWeightedWorker) Init() error { return fmt.Errorf("illegal edge weighted property type:%v", lw.edgeProperty) } lw.useProperty = true + if err := lw.validateEdgeWeights(); err != nil { + return err + } + } + return nil +} + +func (lw *LouvainWeightedWorker) validateEdgeWeights() error { + for localID := uint32(0); localID < lw.WContext.GraphData.VertexCount; localID++ { + inEdges := lw.WContext.GraphData.Edges.GetInEdges(localID) + for edgeIndex := range inEdges { + var weight float64 + switch lw.propertyType { + case structure.ValueTypeInt32: + weight = float64(lw.WContext.GraphData.InEdgesProperty.GetInt32Value( + lw.edgeProperty, localID, uint32(edgeIndex))) + case structure.ValueTypeFloat32: + weight = float64(lw.WContext.GraphData.InEdgesProperty.GetFloat32Value( + lw.edgeProperty, localID, uint32(edgeIndex))) + } + if err := validateNonNegativeFinite("louvain edge weight", weight); err != nil { + vertexID := lw.WContext.GraphData.VertIDStart + localID + return fmt.Errorf("property %q at vertex %d, incoming edge %d: %w", + lw.edgeProperty, vertexID, edgeIndex, err) + } + } } return nil } @@ -192,6 +221,18 @@ func (lw *LouvainWeightedWorker) BeforeStep() { lw.firstStepKI[vertexID] = wgtSum lw.firstStepCommID[vertexID] = serialize.SUint32(vertexID) } + if lw.edgeNums == 0 { + lw.nodeID = identityCommunityIDs(lw.WContext.GraphData.Vertex.TotalVertexCount()) + lw.nodes = make(map[serialize.SUint32]*louvainNode) + lw.communities = make(map[serialize.SUint32]*community) + lw.firstStep = false + lw.firstStepKI = nil + lw.firstStepCommID = nil + lw.neighborEdges = nil + lw.WContext.SetValue("mod_value", serialize.SFloat32(0)) + atomic.StoreInt32(&lw.outputReady, 1) + return + } lw.resolution /= lw.edgeNums logrus.Infof("edge nums:%v", lw.edgeNums) logrus.Infof("resolution:%v", lw.resolution) @@ -335,6 +376,10 @@ func (lw *LouvainWeightedWorker) Compute(vertexID uint32, pID int) { } } + if weight == 0 { + continue + } + lw.inEdges[vertexID][edge] += weight } } else if lw.WContext.Step == 2 { @@ -351,10 +396,19 @@ func (lw *LouvainWeightedWorker) Compute(vertexID uint32, pID int) { } trimMap[edge] = struct{}{} wgt := lw.inEdges[edge][serialize.SUint32(vertexID)] + if wgt == 0 { + continue + } lw.neighborEdges[vertexID][edge] += wgt } } else { if lw.firstStep { + // Zero-weight edges are removed while building neighborEdges. A vertex + // with no remaining effective edge is an identity community and must + // not enter the modularity optimization path. + if len(lw.neighborEdges[vertexID]) == 0 { + return + } //以vertex为基本单元计算 currCommID := lw.firstStepCommID[vertexID] kI := lw.firstStepKI[vertexID] @@ -572,10 +626,7 @@ func (lw *LouvainWeightedWorker) initLouvainNode() { //gen a new graph for new step //可以并行 lw.nodes = make(map[serialize.SUint32]*louvainNode, len(lw.communities)) - lw.nodeID = make([]serialize.SUint32, lw.WContext.GraphData.Vertex.TotalVertexCount()) - for i := range lw.nodeID { - lw.nodeID[i] = serialize.SUint32(i) - } + lw.nodeID = identityCommunityIDs(lw.WContext.GraphData.Vertex.TotalVertexCount()) locker := &sync.Mutex{} commIDs := make([]serialize.SUint32, 0, len(lw.communities)) for commID := range lw.communities { @@ -704,7 +755,13 @@ type LouvainWeightedMaster struct { func (lm *LouvainWeightedMaster) Init() error { lm.threshold = options.GetFloat(lm.MContext.Params, "louvain.threshold") + if err := validateNonNegativeFinite("louvain.threshold", lm.threshold); err != nil { + return err + } lm.maxStep = options.GetInt(lm.MContext.Params, "louvain.step") + if err := validatePositiveInt("louvain.step", lm.maxStep); err != nil { + return err + } lm.prevModValue = math.MinInt32 lm.louvainStep = 1 return nil From 471b2e3e187406bbfeac03f249bcb1d62e5a90af Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 22:22:19 +0800 Subject: [PATCH 3/6] test(vermeer): clarify Louvain readiness hardening --- vermeer/algorithms/louvain_output_test.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/vermeer/algorithms/louvain_output_test.go b/vermeer/algorithms/louvain_output_test.go index 9073c4daa..d439cf9ff 100644 --- a/vermeer/algorithms/louvain_output_test.go +++ b/vermeer/algorithms/louvain_output_test.go @@ -19,10 +19,13 @@ package algorithms import ( "strings" + "sync/atomic" "testing" ) -func TestLouvainRejectsOutputUntilCommunityIDsArePublished(t *testing.T) { +func TestLouvainDefensivelyRejectsUnpublishedCommunityIDs(t *testing.T) { + // The normal master/worker barrier completes BeforeStep before output starts. + // This test covers the defensive guard against malformed or early output state. louvain := &LouvainWorker{firstStep: false} weighted := &LouvainWeightedWorker{firstStep: false} tests := []struct { @@ -30,8 +33,8 @@ func TestLouvainRejectsOutputUntilCommunityIDsArePublished(t *testing.T) { validate func() error publish func() }{ - {"louvain", louvain.ValidateOutput, func() { louvain.outputReady = 1 }}, - {"louvain_weighted", weighted.ValidateOutput, func() { weighted.outputReady = 1 }}, + {"louvain", louvain.ValidateOutput, func() { atomic.StoreInt32(&louvain.outputReady, 1) }}, + {"louvain_weighted", weighted.ValidateOutput, func() { atomic.StoreInt32(&weighted.outputReady, 1) }}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { From eb6142843b62f80bebea6f037b11ee35ae98abe6 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 22:24:24 +0800 Subject: [PATCH 4/6] test(vermeer): cover Louvain degenerate lifecycle paths --- vermeer/algorithms/louvain_validation_test.go | 137 +++++++++++++++++- 1 file changed, 129 insertions(+), 8 deletions(-) diff --git a/vermeer/algorithms/louvain_validation_test.go b/vermeer/algorithms/louvain_validation_test.go index 5457ffa46..066bcadc7 100644 --- a/vermeer/algorithms/louvain_validation_test.go +++ b/vermeer/algorithms/louvain_validation_test.go @@ -19,6 +19,7 @@ package algorithms import ( "math" + "strings" "testing" "vermeer/apps/compute" "vermeer/apps/serialize" @@ -27,14 +28,7 @@ import ( func TestIdentityCommunityIDs(t *testing.T) { ids := identityCommunityIDs(4) - if len(ids) != 4 { - t.Fatalf("expected 4 IDs, got %d", len(ids)) - } - for i, id := range ids { - if uint32(id) != uint32(i) { - t.Fatalf("vertex %d assigned to community %d", i, id) - } - } + assertIdentityCommunityIDs(t, ids, 4) } func TestLouvainFloatParameterValidation(t *testing.T) { @@ -72,6 +66,109 @@ func TestLouvainStepValidation(t *testing.T) { } } +func TestLouvainZeroEffectiveEdgesPublishIdentityOutput(t *testing.T) { + t.Run("unweighted", func(t *testing.T) { + worker := &LouvainWorker{} + ctx := worker.MakeContext(map[string]string{"louvain.resolution": "1"}) + ctx.GraphData = newLouvainTestGraph(3) + ctx.Parallel = 1 + ctx.Step = 2 + + if err := worker.Init(); err != nil { + t.Fatalf("init failed: %v", err) + } + worker.BeforeStep() + + if err := worker.ValidateOutput(); err != nil { + t.Fatalf("zero-edge result was not published: %v", err) + } + assertIdentityCommunityIDs(t, worker.nodeID, 3) + if worker.firstStep { + t.Fatal("zero-edge early return left firstStep set") + } + if worker.neighborEdges != nil { + t.Fatal("zero-edge early return retained adjacency state") + } + if got := ctx.GetValue("mod_value").(serialize.SFloat32); got != 0 { + t.Fatalf("expected zero modularity, got %v", got) + } + }) + + t.Run("weighted", func(t *testing.T) { + worker := &LouvainWeightedWorker{} + ctx := worker.MakeContext(map[string]string{"louvain.resolution": "1"}) + ctx.GraphData = newLouvainTestGraph(3) + ctx.Parallel = 1 + ctx.Step = 3 + + if err := worker.Init(); err != nil { + t.Fatalf("init failed: %v", err) + } + worker.BeforeStep() + + if err := worker.ValidateOutput(); err != nil { + t.Fatalf("zero-weight result was not published: %v", err) + } + assertIdentityCommunityIDs(t, worker.nodeID, 3) + if worker.firstStep { + t.Fatal("zero-weight early return left firstStep set") + } + if worker.neighborEdges != nil { + t.Fatal("zero-weight early return retained adjacency state") + } + if got := ctx.GetValue("mod_value").(serialize.SFloat32); got != 0 { + t.Fatalf("expected zero modularity, got %v", got) + } + }) +} + +func TestWeightedLouvainInitRejectsInvalidEdgeWeights(t *testing.T) { + tests := []struct { + name string + weight serialize.SFloat32 + }{ + {"negative", serialize.SFloat32(-1)}, + {"nan", serialize.SFloat32(float32(math.NaN()))}, + {"positive_inf", serialize.SFloat32(float32(math.Inf(1)))}, + {"negative_inf", serialize.SFloat32(float32(math.Inf(-1)))}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + graph := newLouvainTestGraph(1) + graph.VertIDStart = 7 + graph.Edges = &structure.EdgeMem{ + InEdges: serialize.TwoDimSliceUint32{{0}}, + } + properties := structure.EdgeProperties{ + "weight": &structure.VValues{ + VType: structure.ValueTypeFloat32, + Values: [][]serialize.SFloat32{{test.weight}}, + }, + } + graph.InEdgesProperty = &properties + + worker := &LouvainWeightedWorker{} + ctx := worker.MakeContext(map[string]string{ + "louvain.resolution": "1", + "louvain.edge_weight_property": "weight", + }) + ctx.GraphData = graph + ctx.Parallel = 1 + + err := worker.Init() + if err == nil { + t.Fatalf("expected invalid edge weight %v to fail init", test.weight) + } + for _, expected := range []string{"weight", "vertex 7", "incoming edge 0"} { + if !strings.Contains(err.Error(), expected) { + t.Fatalf("error %q does not contain %q", err, expected) + } + } + }) + } +} + func TestWeightedLouvainSkipsEffectivelyIsolatedVertex(t *testing.T) { edges := &structure.EdgeMem{ InEdges: serialize.TwoDimSliceUint32{{0}}, @@ -102,3 +199,27 @@ func TestWeightedLouvainSkipsEffectivelyIsolatedVertex(t *testing.T) { worker.node2comm[0]) } } + +func newLouvainTestGraph(vertexCount uint32) *structure.GraphData { + return &structure.GraphData{ + VertexCount: vertexCount, + Vertex: &structure.VertexMem{ + TotalVertex: make([]structure.Vertex, vertexCount), + }, + Edges: &structure.EdgeMem{ + InEdges: make(serialize.TwoDimSliceUint32, vertexCount), + }, + } +} + +func assertIdentityCommunityIDs(t *testing.T, ids []serialize.SUint32, total uint32) { + t.Helper() + if uint32(len(ids)) != total { + t.Fatalf("expected %d IDs, got %d", total, len(ids)) + } + for i, id := range ids { + if uint32(id) != uint32(i) { + t.Fatalf("vertex %d assigned to community %d", i, id) + } + } +} From de3ba7264905890df76e25f198153103e013b141 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 22:24:43 +0800 Subject: [PATCH 5/6] test(vermeer): sync Louvain readiness clarification --- vermeer/algorithms/louvain_output_test.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/vermeer/algorithms/louvain_output_test.go b/vermeer/algorithms/louvain_output_test.go index 9073c4daa..d439cf9ff 100644 --- a/vermeer/algorithms/louvain_output_test.go +++ b/vermeer/algorithms/louvain_output_test.go @@ -19,10 +19,13 @@ package algorithms import ( "strings" + "sync/atomic" "testing" ) -func TestLouvainRejectsOutputUntilCommunityIDsArePublished(t *testing.T) { +func TestLouvainDefensivelyRejectsUnpublishedCommunityIDs(t *testing.T) { + // The normal master/worker barrier completes BeforeStep before output starts. + // This test covers the defensive guard against malformed or early output state. louvain := &LouvainWorker{firstStep: false} weighted := &LouvainWeightedWorker{firstStep: false} tests := []struct { @@ -30,8 +33,8 @@ func TestLouvainRejectsOutputUntilCommunityIDsArePublished(t *testing.T) { validate func() error publish func() }{ - {"louvain", louvain.ValidateOutput, func() { louvain.outputReady = 1 }}, - {"louvain_weighted", weighted.ValidateOutput, func() { weighted.outputReady = 1 }}, + {"louvain", louvain.ValidateOutput, func() { atomic.StoreInt32(&louvain.outputReady, 1) }}, + {"louvain_weighted", weighted.ValidateOutput, func() { atomic.StoreInt32(&weighted.outputReady, 1) }}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { From 43a37c83cd837ed0ff56dbe8b4e4605fce27337f Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 23:29:26 +0800 Subject: [PATCH 6/6] test(vermeer): initialize Louvain options --- vermeer/algorithms/louvain_validation_test.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/vermeer/algorithms/louvain_validation_test.go b/vermeer/algorithms/louvain_validation_test.go index 066bcadc7..3498eecd9 100644 --- a/vermeer/algorithms/louvain_validation_test.go +++ b/vermeer/algorithms/louvain_validation_test.go @@ -22,10 +22,15 @@ import ( "strings" "testing" "vermeer/apps/compute" + "vermeer/apps/options" "vermeer/apps/serialize" "vermeer/apps/structure" ) +func init() { + options.Init() +} + func TestIdentityCommunityIDs(t *testing.T) { ids := identityCommunityIDs(4) assertIdentityCommunityIDs(t, ids, 4)