diff --git a/vermeer/algorithms/cycle_boolean_rank_test.go b/vermeer/algorithms/cycle_boolean_rank_test.go new file mode 100644 index 000000000..cefc2c052 --- /dev/null +++ b/vermeer/algorithms/cycle_boolean_rank_test.go @@ -0,0 +1,37 @@ +/* +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 "testing" + +func TestBooleanCycleModeSkipsCanonicalRanks(t *testing.T) { + worker := newCycleTestWorker(t, Boolean) + if len(worker.cycleRanks) != 0 { + t.Fatalf("boolean mode allocated %d canonical ranks", len(worker.cycleRanks)) + } + + state := newCycleSearchState(3) + _, found := boundedCycleSearch(0, directedTriangleAdjacency(), nil, &state, + cycleSearchOptions{ + minLen: 3, maxLen: 3, maxResults: 1, + canonical: false, collect: false, + }) + if found != 1 { + t.Fatalf("boolean search without ranks found %d cycles, expected 1", found) + } +} diff --git a/vermeer/algorithms/cycle_detection.go b/vermeer/algorithms/cycle_detection.go index d7023a9b4..95babdede 100644 --- a/vermeer/algorithms/cycle_detection.go +++ b/vermeer/algorithms/cycle_detection.go @@ -27,6 +27,8 @@ import ( "github.com/sirupsen/logrus" ) +const cycleCountValue = "cycle_num" + func init() { Algorithms = append(Algorithms, new(CycleDetectionMaker)) } @@ -61,13 +63,12 @@ const ( type CycleDetectionWorker struct { compute.WorkerComputerBase - mode cyclesMode - maxLen int - minLen int - limit int + parameters cycleParameters hasCycle []serialize.SUint8 inEdges []serialize.SliceUint32 cycleList []serialize.TwoDimSliceString + cycleRanks []uint32 + searchStates []cycleSearchState useVertexFilter bool useEdgeFilter bool vertexFilter *compute.VertexFilter @@ -75,186 +76,179 @@ type CycleDetectionWorker struct { } func (cdw *CycleDetectionWorker) Init() error { - cdw.inEdges = make([]serialize.SliceUint32, cdw.WContext.GraphData.Vertex.TotalVertexCount()) - cdw.maxLen = options.GetInt(cdw.WContext.Params, "cycle.max_length") - cdw.minLen = options.GetInt(cdw.WContext.Params, "cycle.min_length") - cdw.mode = cyclesMode(options.GetString(cdw.WContext.Params, "cycle.mode")) - switch cdw.mode { - case All: - cdw.cycleList = make([]serialize.TwoDimSliceString, cdw.WContext.GraphData.VertexCount) - case Limit: - cdw.cycleList = make([]serialize.TwoDimSliceString, cdw.WContext.GraphData.VertexCount) - cdw.limit = options.GetInt(cdw.WContext.Params, "cycle.max_cycles") + totalVertices := cdw.WContext.GraphData.Vertex.TotalVertexCount() + parameters, err := parseCycleParameters(cdw.WContext.Params, totalVertices) + if err != nil { + return err + } + cdw.parameters = parameters + cdw.inEdges = make([]serialize.SliceUint32, totalVertices) + + if parameters.mode != Boolean { + cdw.cycleRanks = buildCycleRanksByID(totalVertices, + func(vertexID uint32) string { + return cdw.WContext.GraphData.Vertex.GetVertex(vertexID).ID + }) + } + parallel := cdw.WContext.Parallel + if parallel <= 0 { + parallel = 1 + } + cdw.searchStates = make([]cycleSearchState, parallel) + for partitionID := range cdw.searchStates { + cdw.searchStates[partitionID] = newCycleSearchState(parameters.maxLen) + } + + switch parameters.mode { + case All, Limit: + cdw.cycleList = make([]serialize.TwoDimSliceString, + cdw.WContext.GraphData.VertexCount) case Boolean: - cdw.hasCycle = make([]serialize.SUint8, cdw.WContext.GraphData.VertexCount) - default: - return fmt.Errorf("cycle detection mode must be 'all', 'limit', 'boolean'. not: %v", cdw.mode) + cdw.hasCycle = make([]serialize.SUint8, + cdw.WContext.GraphData.VertexCount) } - cdw.WContext.CreateValue("cycle_num", compute.ValueTypeInt32, compute.CValueActionAggregate) - cdw.WContext.SetValue("cycle_num", serialize.SInt32(0)) + cdw.WContext.CreateValue(cycleCountValue, compute.ValueTypeInt64, + compute.CValueActionAggregate) + cdw.WContext.SetValue(cycleCountValue, serialize.SInt64(0)) + vertexExprStr := options.GetString(cdw.WContext.Params, "filter.vertex_expr") - filterVertexProps := options.GetSliceString(cdw.WContext.Params, "filter.vertex_properties") + filterVertexProps := options.GetSliceString(cdw.WContext.Params, + "filter.vertex_properties") edgeExprStr := options.GetString(cdw.WContext.Params, "filter.edge_expr") - filterEdgeProps := options.GetSliceString(cdw.WContext.Params, "filter.edge_properties") + filterEdgeProps := options.GetSliceString(cdw.WContext.Params, + "filter.edge_properties") if vertexExprStr != "" && len(filterVertexProps) > 0 { cdw.useVertexFilter = true cdw.vertexFilter = &compute.VertexFilter{} - err := cdw.vertexFilter.Init(vertexExprStr, filterVertexProps, cdw.WContext.GraphData.VertexProperty) - if err != nil { - logrus.Errorf("vertex filter init error:%v", err) - return fmt.Errorf("vertex filter init error:%w", err) + if err := cdw.vertexFilter.Init(vertexExprStr, filterVertexProps, + cdw.WContext.GraphData.VertexProperty); err != nil { + return fmt.Errorf("vertex filter init error: %w", err) } } if edgeExprStr != "" && len(filterEdgeProps) > 0 { cdw.useEdgeFilter = true cdw.edgeFilter = &compute.EdgeFilter{} - err := cdw.edgeFilter.Init(edgeExprStr, filterEdgeProps, cdw.WContext.GraphData.InEdgesProperty) - if err != nil { - logrus.Errorf("edge filter init error:%v", err) - return fmt.Errorf("edge filter init error:%w", err) + if err := cdw.edgeFilter.Init(edgeExprStr, filterEdgeProps, + cdw.WContext.GraphData.InEdgesProperty); err != nil { + return fmt.Errorf("edge filter init error: %w", err) } } return nil } -func (cdw *CycleDetectionWorker) VertexValue(i uint32) serialize.MarshalAble { +func (cdw *CycleDetectionWorker) VertexValue(vertexID uint32) serialize.MarshalAble { if cdw.WContext.Output { - switch cdw.mode { + localID := vertexID - cdw.WContext.GraphData.VertIDStart + switch cdw.parameters.mode { case All, Limit: - return &cdw.cycleList[i-cdw.WContext.GraphData.VertIDStart] + return &cdw.cycleList[localID] case Boolean: - return &cdw.hasCycle[i-cdw.WContext.GraphData.VertIDStart] + return &cdw.hasCycle[localID] } } if cdw.WContext.Step == 1 { - return &cdw.inEdges[i] + return &cdw.inEdges[vertexID] } - nilValue := serialize.SInt32(0) - return &nilValue + return &noCycleScatterValue } -func (cdw *CycleDetectionWorker) BeforeStep() { -} - -func (cdw *CycleDetectionWorker) Compute(vertexID uint32, pID int) { +func (cdw *CycleDetectionWorker) Compute(vertexID uint32, partitionID int) { if cdw.useVertexFilter && !cdw.vertexFilter.Filter(vertexID) { return } - _ = pID - vIDx := vertexID - cdw.WContext.GraphData.VertIDStart + localID := vertexID - cdw.WContext.GraphData.VertIDStart if cdw.WContext.Step == 1 { - cdw.inEdges[vertexID] = make([]serialize.SUint32, 0, len(cdw.WContext.GraphData.Edges.GetInEdges(vIDx))) - tempMap := make(map[serialize.SUint32]struct{}) - for idx, edge := range cdw.WContext.GraphData.Edges.GetInEdges(vIDx) { - if cdw.useVertexFilter && !cdw.vertexFilter.Filter(uint32(edge)) || - cdw.useEdgeFilter && !cdw.edgeFilter.Filter(vIDx, uint32(idx)) { - continue - } - if _, ok := tempMap[edge]; ok || vertexID == uint32(edge) { - continue - } - tempMap[edge] = struct{}{} - cdw.inEdges[vertexID] = append(cdw.inEdges[vertexID], edge) - } - } else { - switch cdw.mode { - case All, Limit: - for _, edge := range cdw.inEdges[vertexID] { - if cdw.WContext.GraphData.Vertex.GetVertex(vertexID).ID < cdw.WContext.GraphData.Vertex.GetVertex(uint32(edge)).ID { - cdw.dfs(serialize.SUint32(vertexID), edge, []serialize.SUint32{edge}) - } - } - case Boolean: - for _, edge := range cdw.inEdges[vertexID] { - cdw.dfs(serialize.SUint32(vertexID), edge, []serialize.SUint32{edge}) - } - } + cdw.buildFilteredInEdges(vertexID, localID) + return } -} - -func (cdw *CycleDetectionWorker) dfs(rootID serialize.SUint32, vertexID serialize.SUint32, stack []serialize.SUint32) { - //logrus.Infof("root:%v, vID:%v, stack:%v", rootID, vertexID, stack) - if len(stack) > cdw.maxLen { + if cdw.WContext.Step != 2 { return } - temp := make(map[serialize.SUint32]struct{}, len(stack)) - for _, node := range stack { - if _, ok := temp[node]; ok { - return + + state := &cdw.searchStates[partitionID] + options := cycleSearchOptions{ + minLen: cdw.parameters.minLen, + maxLen: cdw.parameters.maxLen, + canonical: cdw.parameters.mode != Boolean, + collect: cdw.parameters.mode != Boolean, + } + switch cdw.parameters.mode { + case All: + cycles, _ := boundedCycleSearch(vertexID, cdw.inEdges, + cdw.cycleRanks, state, options) + cdw.cycleList[localID] = cdw.externalCycleIDs(cycles) + case Limit: + options.maxResults = cdw.parameters.limit + cycles, _ := boundedCycleSearch(vertexID, cdw.inEdges, + cdw.cycleRanks, state, options) + cdw.cycleList[localID] = cdw.externalCycleIDs(cycles) + case Boolean: + options.maxResults = 1 + _, found := boundedCycleSearch(vertexID, cdw.inEdges, + cdw.cycleRanks, state, options) + if found > 0 { + cdw.hasCycle[localID] = 1 } - temp[node] = struct{}{} } +} - for _, edge := range cdw.inEdges[vertexID] { - if _, ok := temp[edge]; ok { +func (cdw *CycleDetectionWorker) buildFilteredInEdges(vertexID, localID uint32) { + rawEdges := cdw.WContext.GraphData.Edges.GetInEdges(localID) + neighbors := make(serialize.SliceUint32, 0, len(rawEdges)) + for edgeIndex, edge := range rawEdges { + neighborID := uint32(edge) + if neighborID == vertexID { continue } - switch cdw.mode { - case All: - if cdw.WContext.GraphData.Vertex.GetVertex(uint32(rootID)).ID < cdw.WContext.GraphData.Vertex.GetVertex(uint32(edge)).ID { - cdw.dfs(rootID, edge, append(stack, edge)) - } else if rootID == edge && len(stack) >= cdw.minLen { - vIDx := uint32(rootID) - cdw.WContext.GraphData.VertIDStart - newList := make([]serialize.SString, 0, len(stack)) - for _, node := range stack { - newList = append(newList, serialize.SString(cdw.WContext.GraphData.Vertex.GetVertex(uint32(node)).ID)) - } - cdw.cycleList[vIDx] = append(cdw.cycleList[vIDx], newList) - } - case Limit: - vIDx := uint32(rootID) - cdw.WContext.GraphData.VertIDStart - if len(cdw.cycleList[vIDx]) >= cdw.limit { - return - } - if cdw.WContext.GraphData.Vertex.GetVertex(uint32(rootID)).ID < cdw.WContext.GraphData.Vertex.GetVertex(uint32(edge)).ID { - cdw.dfs(rootID, edge, append(stack, edge)) - } else if rootID == edge && len(stack) >= cdw.minLen { - vIDx := uint32(rootID) - cdw.WContext.GraphData.VertIDStart - newList := make([]serialize.SString, 0, len(stack)) - for _, node := range stack { - newList = append(newList, serialize.SString(cdw.WContext.GraphData.Vertex.GetVertex(uint32(node)).ID)) - } - cdw.cycleList[vIDx] = append(cdw.cycleList[vIDx], newList) - if len(cdw.cycleList[vIDx]) >= cdw.limit { - return - } - } - case Boolean: - if cdw.hasCycle[uint32(rootID)-cdw.WContext.GraphData.VertIDStart] == 1 { - return - } - if rootID != edge { - cdw.dfs(rootID, edge, append(stack, edge)) - } else if len(stack) >= cdw.minLen { - cdw.hasCycle[uint32(rootID)-cdw.WContext.GraphData.VertIDStart] = 1 - return - } + if cdw.useVertexFilter && !cdw.vertexFilter.Filter(neighborID) { + continue + } + if cdw.useEdgeFilter && !cdw.edgeFilter.Filter(localID, + uint32(edgeIndex)) { + continue } + neighbors = append(neighbors, edge) } + cdw.inEdges[vertexID] = sortedUniqueCycleNeighbors(neighbors) +} + +func (cdw *CycleDetectionWorker) externalCycleIDs(cycles [][]uint32) serialize.TwoDimSliceString { + result := make(serialize.TwoDimSliceString, 0, len(cycles)) + for _, cycle := range cycles { + ids := make(serialize.SliceString, len(cycle)) + for index, vertexID := range cycle { + ids[index] = serialize.SString( + cdw.WContext.GraphData.Vertex.GetVertex(vertexID).ID) + } + result = append(result, ids) + } + return result } func (cdw *CycleDetectionWorker) AfterStep() { - if cdw.WContext.Step > 1 { - cycleCount := 0 - switch cdw.mode { - case All, Limit: - for _, list := range cdw.cycleList { - cycleCount += len(list) - } - case Boolean: - for _, has := range cdw.hasCycle { - if has == 1 { - cycleCount++ - } + if cdw.WContext.Step != 2 { + return + } + var cycleCount int64 + switch cdw.parameters.mode { + case All, Limit: + for _, cycles := range cdw.cycleList { + cycleCount += int64(len(cycles)) + } + case Boolean: + for _, hasCycle := range cdw.hasCycle { + if hasCycle == 1 { + cycleCount++ } } - - //logrus.Infof("cycle count:%v", cycleCount) - cdw.WContext.SetValue("cycle_num", serialize.SInt32(cycleCount)) } + cdw.WContext.SetValue(cycleCountValue, serialize.SInt64(cycleCount)) } func (cdw *CycleDetectionWorker) OutputValueType() string { + if cdw.parameters.mode == Boolean { + return common.HgValueTypeInt + } return common.HgValueTypeString } @@ -262,15 +256,11 @@ type CycleDetectionMaster struct { compute.MasterComputerBase } -func (cdm *CycleDetectionMaster) Init() error { - return nil -} - func (cdm *CycleDetectionMaster) Compute() bool { - if cdm.MContext.Step == 2 { - cycleNum := cdm.MContext.GetValue("cycle_num").(serialize.SInt32) - logrus.Infof("cycles num:%v", cycleNum) - return false + if cdm.MContext.Step != 2 { + return true } - return true + cycleCount := cdm.MContext.GetValue(cycleCountValue).(serialize.SInt64) + logrus.Infof("cycles num:%v", cycleCount) + return false } diff --git a/vermeer/algorithms/cycle_detection_test.go b/vermeer/algorithms/cycle_detection_test.go new file mode 100644 index 000000000..111a0bc6c --- /dev/null +++ b/vermeer/algorithms/cycle_detection_test.go @@ -0,0 +1,278 @@ +/* +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 ( + "reflect" + "testing" + "vermeer/apps/common" + "vermeer/apps/options" + "vermeer/apps/serialize" + "vermeer/apps/structure" +) + +func init() { + options.Init() +} + +func TestParseCycleParameters(t *testing.T) { + parameters, err := parseCycleParameters(map[string]string{ + "compute.max_step": "2", + "cycle.mode": "all", + "cycle.min_length": "0", + "cycle.max_length": "10", + "cycle.max_cycles": "10", + }, 4) + if err != nil { + t.Fatalf("valid parameters rejected: %v", err) + } + if parameters.minLen != 2 || parameters.maxLen != 4 { + t.Fatalf("unexpected normalized bounds: %+v", parameters) + } + + tests := []struct { + name string + params map[string]string + }{ + { + name: "short_compute_budget", + params: map[string]string{ + "compute.max_step": "1", "cycle.mode": "all", + "cycle.min_length": "2", "cycle.max_length": "3", + }, + }, + { + name: "max_below_two", + params: map[string]string{ + "compute.max_step": "2", "cycle.mode": "all", + "cycle.min_length": "0", "cycle.max_length": "1", + }, + }, + { + name: "min_above_max", + params: map[string]string{ + "compute.max_step": "2", "cycle.mode": "all", + "cycle.min_length": "4", "cycle.max_length": "3", + }, + }, + { + name: "zero_limit", + params: map[string]string{ + "compute.max_step": "2", "cycle.mode": "limit", + "cycle.min_length": "2", "cycle.max_length": "3", + "cycle.max_cycles": "0", + }, + }, + { + name: "unknown_mode", + params: map[string]string{ + "compute.max_step": "2", "cycle.mode": "unknown", + "cycle.min_length": "2", "cycle.max_length": "3", + }, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if _, err := parseCycleParameters(test.params, 10); err == nil { + t.Fatal("invalid parameters were accepted") + } + }) + } +} + +func TestBuildCycleRanksUsesExternalVertexIDs(t *testing.T) { + ranks := buildCycleRanks([]string{"2", "10", "1"}) + want := []uint32{2, 1, 0} + if !reflect.DeepEqual(ranks, want) { + t.Fatalf("expected lexical external-ID ranks %v, got %v", want, ranks) + } +} + +func TestSortedUniqueCycleNeighbors(t *testing.T) { + got := sortedUniqueCycleNeighbors(serialize.SliceUint32{3, 1, 3, 2, 1}) + want := serialize.SliceUint32{1, 2, 3} + if !reflect.DeepEqual(got, want) { + t.Fatalf("expected %v, got %v", want, got) + } +} + +func TestBoundedCycleSearchUsesActualCycleLength(t *testing.T) { + adjacency := directedTriangleAdjacency() + ranks := []uint32{0, 1, 2} + state := newCycleSearchState(3) + + cycles, found := boundedCycleSearch(0, adjacency, ranks, &state, + cycleSearchOptions{ + minLen: 3, maxLen: 3, canonical: true, collect: true, + }) + if found != 1 || len(cycles) != 1 || + !reflect.DeepEqual(cycles[0], []uint32{2, 1}) { + t.Fatalf("expected one length-3 cycle, found=%d cycles=%v", found, cycles) + } + + _, found = boundedCycleSearch(0, adjacency, ranks, &state, + cycleSearchOptions{ + minLen: 2, maxLen: 2, canonical: true, collect: true, + }) + if found != 0 { + t.Fatalf("length-3 cycle passed max_length=2: %d", found) + } + + _, found = boundedCycleSearch(1, adjacency, ranks, &state, + cycleSearchOptions{ + minLen: 3, maxLen: 3, canonical: true, collect: true, + }) + if found != 0 { + t.Fatalf("non-minimum canonical root emitted the same cycle: %d", found) + } +} + +func TestBooleanCycleSearchFindsEveryCycleVertex(t *testing.T) { + adjacency := directedTriangleAdjacency() + ranks := []uint32{0, 1, 2} + state := newCycleSearchState(3) + for rootID := uint32(0); rootID < 3; rootID++ { + cycles, found := boundedCycleSearch(rootID, adjacency, ranks, &state, + cycleSearchOptions{ + minLen: 3, maxLen: 3, maxResults: 1, + canonical: false, collect: false, + }) + if found != 1 || len(cycles) != 0 { + t.Fatalf("root %d: expected boolean short-circuit, found=%d cycles=%v", + rootID, found, cycles) + } + } +} + +func TestLimitModeStopsAtRequestedResultCount(t *testing.T) { + adjacency := []serialize.SliceUint32{ + {1, 2, 3}, + {0, 2, 3}, + {0, 1, 3}, + {0, 1, 2}, + } + ranks := []uint32{0, 1, 2, 3} + state := newCycleSearchState(4) + cycles, found := boundedCycleSearch(0, adjacency, ranks, &state, + cycleSearchOptions{ + minLen: 2, maxLen: 4, maxResults: 1, + canonical: true, collect: true, + }) + if found != 1 || len(cycles) != 1 { + t.Fatalf("expected exactly one collected cycle, found=%d cycles=%v", + found, cycles) + } + + // Early return leaves internal slices populated, so run again to verify reset + // safely reuses the same per-partition state. + cycles, found = boundedCycleSearch(0, adjacency, ranks, &state, + cycleSearchOptions{ + minLen: 2, maxLen: 4, maxResults: 1, + canonical: true, collect: true, + }) + if found != 1 || len(cycles) != 1 { + t.Fatalf("reused state changed result, found=%d cycles=%v", found, cycles) + } +} + +func TestCycleWorkerLifecycleAndCanonicalOutput(t *testing.T) { + worker := newCycleTestWorker(t, All) + for step := int32(1); step <= 2; step++ { + worker.WContext.Step = step + for vertexID := uint32(0); vertexID < 3; vertexID++ { + worker.Compute(vertexID, 0) + } + worker.AfterStep() + } + + // External IDs are ["2", "10", "1"], so global vertex 2 owns the cycle. + if len(worker.cycleList[0]) != 0 || len(worker.cycleList[1]) != 0 || + len(worker.cycleList[2]) != 1 { + t.Fatalf("unexpected canonical cycle ownership: %v", worker.cycleList) + } + if count := worker.WContext.GetValue(cycleCountValue).(serialize.SInt64); count != 1 { + t.Fatalf("expected one aggregated local cycle, got %d", count) + } +} + +func TestCycleBooleanLifecycleAndOutputType(t *testing.T) { + worker := newCycleTestWorker(t, Boolean) + for step := int32(1); step <= 2; step++ { + worker.WContext.Step = step + for vertexID := uint32(0); vertexID < 3; vertexID++ { + worker.Compute(vertexID, 0) + } + worker.AfterStep() + } + for vertexID, value := range worker.hasCycle { + if value != 1 { + t.Fatalf("vertex %d was not marked as cyclic: %v", vertexID, worker.hasCycle) + } + } + if worker.OutputValueType() != common.HgValueTypeInt { + t.Fatalf("boolean mode output type is %q", worker.OutputValueType()) + } + + worker.parameters.mode = All + if worker.OutputValueType() != common.HgValueTypeString { + t.Fatalf("list mode output type is %q", worker.OutputValueType()) + } +} + +func TestCycleStepTwoScatterPayloadIsEmpty(t *testing.T) { + worker := &CycleDetectionWorker{} + context := worker.MakeContext(nil) + context.Step = 2 + if size := worker.VertexValue(0).PredictSize(); size != 0 { + t.Fatalf("step-two scatter value predicts %d bytes, expected 0", size) + } +} + +func directedTriangleAdjacency() []serialize.SliceUint32 { + // Reversed traversal adjacency: 0 <- 2 <- 1 <- 0. + return []serialize.SliceUint32{{2}, {0}, {1}} +} + +func newCycleTestWorker(t *testing.T, mode cyclesMode) *CycleDetectionWorker { + t.Helper() + graph := &structure.GraphData{ + VertexCount: 3, + Vertex: &structure.VertexMem{ + TotalVertex: []structure.Vertex{ + {ID: "2"}, {ID: "10"}, {ID: "1"}, + }, + }, + Edges: &structure.EdgeMem{ + InEdges: serialize.TwoDimSliceUint32{{2}, {0}, {1}}, + }, + } + worker := &CycleDetectionWorker{} + context := worker.MakeContext(map[string]string{ + "compute.max_step": "2", + "cycle.mode": string(mode), + "cycle.min_length": "3", + "cycle.max_length": "3", + "cycle.max_cycles": "1", + }) + context.GraphData = graph + context.Parallel = 1 + if err := worker.Init(); err != nil { + t.Fatalf("cycle worker init: %v", err) + } + return worker +} diff --git a/vermeer/algorithms/cycle_search.go b/vermeer/algorithms/cycle_search.go new file mode 100644 index 000000000..6bc3d8428 --- /dev/null +++ b/vermeer/algorithms/cycle_search.go @@ -0,0 +1,246 @@ +/* +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 ( + "fmt" + "sort" + "vermeer/apps/options" + "vermeer/apps/serialize" +) + +type cycleParameters struct { + mode cyclesMode + minLen int + maxLen int + limit int +} + +func parseCycleParameters(params map[string]string, totalVertices uint32) (cycleParameters, error) { + result := cycleParameters{ + mode: cyclesMode(options.GetString(params, "cycle.mode")), + minLen: options.GetInt(params, "cycle.min_length"), + maxLen: options.GetInt(params, "cycle.max_length"), + limit: options.GetInt(params, "cycle.max_cycles"), + } + if options.GetInt(params, "compute.max_step") < 2 { + return result, fmt.Errorf("cycle_detection requires compute.max_step >= 2") + } + if result.minLen < 0 { + return result, fmt.Errorf("cycle.min_length must be >= 0, got %d", result.minLen) + } + if result.maxLen < 2 { + return result, fmt.Errorf("cycle.max_length must be >= 2, got %d", result.maxLen) + } + // Self-loops are excluded, so the shortest supported simple directed cycle + // contains two distinct vertices. Keep the historical default min_length=0 + // as an alias for the natural lower bound. + if result.minLen < 2 { + result.minLen = 2 + } + if result.minLen > result.maxLen { + return result, fmt.Errorf("cycle.min_length (%d) must not exceed cycle.max_length (%d)", + result.minLen, result.maxLen) + } + if totalVertices >= 2 && result.maxLen > int(totalVertices) { + result.maxLen = int(totalVertices) + } + switch result.mode { + case All, Boolean: + case Limit: + if result.limit <= 0 { + return result, fmt.Errorf("cycle.max_cycles must be > 0 in limit mode, got %d", + result.limit) + } + default: + return result, fmt.Errorf("cycle detection mode must be 'all', 'limit', or 'boolean', got %q", + result.mode) + } + return result, nil +} + +func buildCycleRanks(vertexIDs []string) []uint32 { + return buildCycleRanksByID(uint32(len(vertexIDs)), func(vertexID uint32) string { + return vertexIDs[vertexID] + }) +} + +// buildCycleRanksByID reads external IDs from their existing graph storage. +// It intentionally allocates only one temporary uint32 permutation and the +// retained uint32 rank vector, rather than copying one string header per vertex. +func buildCycleRanksByID(totalVertices uint32, + vertexIDAt func(uint32) string) []uint32 { + order := make([]uint32, totalVertices) + for vertexID := range order { + order[vertexID] = uint32(vertexID) + } + sort.Slice(order, func(left, right int) bool { + leftID := order[left] + rightID := order[right] + leftExternalID := vertexIDAt(leftID) + rightExternalID := vertexIDAt(rightID) + if leftExternalID != rightExternalID { + return leftExternalID < rightExternalID + } + return leftID < rightID + }) + ranks := make([]uint32, totalVertices) + for rank, vertexID := range order { + ranks[vertexID] = uint32(rank) + } + return ranks +} + +func sortedUniqueCycleNeighbors(neighbors serialize.SliceUint32) serialize.SliceUint32 { + if len(neighbors) < 2 { + return neighbors + } + sort.Slice(neighbors, func(left, right int) bool { + return neighbors[left] < neighbors[right] + }) + write := 1 + for read := 1; read < len(neighbors); read++ { + if neighbors[read] == neighbors[write-1] { + continue + } + neighbors[write] = neighbors[read] + write++ + } + return neighbors[:write] +} + +type cycleFrame struct { + vertexID uint32 + nextEdge int +} + +type cycleSearchState struct { + visited map[uint32]struct{} + path []uint32 + frames []cycleFrame +} + +func newCycleSearchState(maxLen int) cycleSearchState { + capacity := maxLen + if capacity > 64 { + capacity = 64 + } + return cycleSearchState{ + visited: make(map[uint32]struct{}, capacity), + path: make([]uint32, 0, capacity), + frames: make([]cycleFrame, 0, capacity), + } +} + +func (state *cycleSearchState) reset(rootID uint32) { + clear(state.visited) + state.path = append(state.path[:0], rootID) + state.frames = append(state.frames[:0], cycleFrame{vertexID: rootID}) + state.visited[rootID] = struct{}{} +} + +type cycleSearchOptions struct { + minLen int + maxLen int + maxResults int + canonical bool + collect bool +} + +func boundedCycleSearch(rootID uint32, adjacency []serialize.SliceUint32, + ranks []uint32, state *cycleSearchState, + options cycleSearchOptions) ([][]uint32, int) { + if int(rootID) >= len(adjacency) || options.maxLen < 2 { + return nil, 0 + } + state.reset(rootID) + var rootRank uint32 + if options.canonical { + if int(rootID) >= len(ranks) { + return nil, 0 + } + rootRank = ranks[rootID] + } + var cycles [][]uint32 + found := 0 + + for len(state.frames) > 0 { + frameIndex := len(state.frames) - 1 + frame := &state.frames[frameIndex] + neighbors := adjacency[frame.vertexID] + if frame.nextEdge >= len(neighbors) { + vertexID := frame.vertexID + state.frames = state.frames[:frameIndex] + state.path = state.path[:len(state.path)-1] + delete(state.visited, vertexID) + continue + } + + neighborID := uint32(neighbors[frame.nextEdge]) + frame.nextEdge++ + if neighborID == rootID { + cycleLen := len(state.path) + if cycleLen >= options.minLen && cycleLen <= options.maxLen { + found++ + if options.collect { + cycle := make([]uint32, len(state.path)-1) + copy(cycle, state.path[1:]) + cycles = append(cycles, cycle) + } + if options.maxResults > 0 && found >= options.maxResults { + return cycles, found + } + } + continue + } + if neighborID >= uint32(len(adjacency)) || len(state.path) >= options.maxLen { + continue + } + if options.canonical && ranks[neighborID] <= rootRank { + continue + } + if _, seen := state.visited[neighborID]; seen { + continue + } + + state.visited[neighborID] = struct{}{} + state.path = append(state.path, neighborID) + state.frames = append(state.frames, cycleFrame{vertexID: neighborID}) + } + return cycles, found +} + +type emptyCycleScatterValue struct{} + +func (*emptyCycleScatterValue) Marshal([]byte) (int, error) { + return 0, nil +} + +func (*emptyCycleScatterValue) Unmarshal([]byte) (int, error) { + return 0, nil +} + +func (*emptyCycleScatterValue) ToString() string { + return "" +} + +func (*emptyCycleScatterValue) PredictSize() int { + return 0 +} + +var noCycleScatterValue emptyCycleScatterValue diff --git a/vermeer/algorithms/cycle_search_regression_test.go b/vermeer/algorithms/cycle_search_regression_test.go new file mode 100644 index 000000000..35b43dd95 --- /dev/null +++ b/vermeer/algorithms/cycle_search_regression_test.go @@ -0,0 +1,66 @@ +/* +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 ( + "testing" + "vermeer/apps/serialize" +) + +func TestParseCycleParametersRejectsNegativeMinimum(t *testing.T) { + _, err := parseCycleParameters(map[string]string{ + "compute.max_step": "2", + "cycle.mode": "all", + "cycle.min_length": "-1", + "cycle.max_length": "3", + }, 3) + if err == nil { + t.Fatal("negative cycle.min_length was accepted") + } +} + +func TestBoundedCycleSearchPreventsRepeatedVertices(t *testing.T) { + // The only walk from 0 is 0 -> 1 -> 2 -> 1. It repeats vertex 1 and never + // returns to the root, so it must not be reported as a simple cycle. + adjacency := []serialize.SliceUint32{{1}, {2}, {1}} + state := newCycleSearchState(6) + cycles, found := boundedCycleSearch(0, adjacency, nil, &state, + cycleSearchOptions{ + minLen: 2, maxLen: 6, canonical: false, collect: true, + }) + if found != 0 || len(cycles) != 0 { + t.Fatalf("repeated-vertex walk was reported as a cycle: found=%d cycles=%v", + found, cycles) + } +} + +func TestBoundedCycleSearchSkipsInvalidRemoteIDs(t *testing.T) { + // 0 <-> 1 is a valid two-cycle. The out-of-range ID must be ignored before + // canonical-rank lookup and must not affect the valid result. + adjacency := []serialize.SliceUint32{{1, 99}, {0}} + ranks := []uint32{0, 1} + state := newCycleSearchState(2) + cycles, found := boundedCycleSearch(0, adjacency, ranks, &state, + cycleSearchOptions{ + minLen: 2, maxLen: 2, canonical: true, collect: true, + }) + if found != 1 || len(cycles) != 1 || len(cycles[0]) != 1 || cycles[0][0] != 1 { + t.Fatalf("invalid remote ID changed valid cycle result: found=%d cycles=%v", + found, cycles) + } +}