Skip to content

fix(vermeer): make Louvain output readiness explicit #372

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
imbajin merged 2 commits into from
Nov 5, 2026
Merged

fix(vermeer): make Louvain output readiness explicit #372

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
abbbae8
fix(vermeer): publish Louvain output atomically
imbajin Nov 5, 2026
471b2e3
test(vermeer): clarify Louvain readiness hardening
imbajin Nov 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions vermeer/algorithms/louvain.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,7 @@ type LouvainWorker struct {
firstStep bool
firstStepKI []float64
firstStepCommID []serialize.SUint32
outputReady int32
}

func (lw *LouvainWorker) VertexValue(i uint32) serialize.MarshalAble {
Expand All @@ -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
}
Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand Down
35 changes: 20 additions & 15 deletions vermeer/algorithms/louvain_output_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,28 +19,33 @@ package algorithms

import (
"strings"
"sync/atomic"
"testing"
)

func TestLouvainRejectsOutputBeforeFirstPass(t *testing.T) {
for _, test := range []struct {
name string
worker interface{ ValidateOutput() error }
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 {
name string
validate func() error
publish func()
}{
{"louvain", &LouvainWorker{firstStep: true}},
{"louvain_weighted", &LouvainWeightedWorker{firstStep: true}},
} {
{"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) {
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)
}
}
10 changes: 7 additions & 3 deletions vermeer/algorithms/louvain_weighted.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ type LouvainWeightedWorker struct {
firstStep bool
firstStepKI []float64
firstStepCommID []serialize.SUint32
outputReady int32

// property weight
useProperty bool
Expand All @@ -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
}
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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
Expand All @@ -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 {
Expand Down
Loading