From abbbae8e535f4302d28c2b0cadd13ba2bda26bc3 Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 18:00:35 +0800 Subject: [PATCH 1/2] 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 471b2e3e187406bbfeac03f249bcb1d62e5a90af Mon Sep 17 00:00:00 2001 From: imbajin Date: Thu, 5 Nov 2026 22:22:19 +0800 Subject: [PATCH 2/2] 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) {