From 95f83dcfb45de4ed6b24ec89e2b13b7e57ff9e40 Mon Sep 17 00:00:00 2001 From: imbajin Date: Fri, 6 Nov 2026 01:34:19 +0800 Subject: [PATCH] fix(vermeer): optimize triangle-based algorithms Replace hash-heavy triangle and clustering implementations with degree-oriented sorted intersections, retain 64-bit counts, and harden communication and defensive edge handling. --- vermeer/algorithms/clustering_coeffcient.go | 110 +++---- vermeer/algorithms/triangle_count.go | 119 +++---- vermeer/algorithms/triangle_support.go | 183 +++++++++++ vermeer/algorithms/triangle_support_test.go | 298 ++++++++++++++++++ vermeer/apps/common/value_types.go | 20 ++ .../compute_clustering_coefficient.go | 2 +- .../test/functional/compute_trianglecount.go | 2 +- 7 files changed, 593 insertions(+), 141 deletions(-) create mode 100644 vermeer/algorithms/triangle_support.go create mode 100644 vermeer/algorithms/triangle_support_test.go create mode 100644 vermeer/apps/common/value_types.go diff --git a/vermeer/algorithms/clustering_coeffcient.go b/vermeer/algorithms/clustering_coeffcient.go index 25341ad95..a5e7e89c2 100644 --- a/vermeer/algorithms/clustering_coeffcient.go +++ b/vermeer/algorithms/clustering_coeffcient.go @@ -20,6 +20,7 @@ package algorithms import ( "vermeer/apps/common" "vermeer/apps/compute" + "vermeer/apps/options" "vermeer/apps/serialize" ) @@ -49,86 +50,67 @@ func (cc *ClusteringCoefficientMaker) DataNeeded() []string { type ClusteringCoefficientWorker struct { compute.WorkerComputerBase - ccValues []serialize.SFloat32 - totalNeighborSet []serialize.MapUint32Struct - selfNeighborSet []map[serialize.SUint32]struct{} + ccValues []serialize.SFloat32 + degrees []serialize.SUint32 + orientedNeighbors []serialize.SliceUint32 + localNeighbors []serialize.SliceUint32 } -func (ccw *ClusteringCoefficientWorker) VertexValue(i uint32) serialize.MarshalAble { - if ccw.WContext.Step == 1 { - return &ccw.totalNeighborSet[i] +func (ccw *ClusteringCoefficientWorker) VertexValue(vertexID uint32) serialize.MarshalAble { + if ccw.WContext.Output { + return &ccw.ccValues[vertexID] + } + switch ccw.WContext.Step { + case 1: + return &ccw.degrees[vertexID] + case 2: + return &ccw.orientedNeighbors[vertexID] + default: + return &noTriangleScatterValue } - return &ccw.ccValues[i] } func (ccw *ClusteringCoefficientWorker) Init() error { - ccw.ccValues = make([]serialize.SFloat32, ccw.WContext.GraphData.Vertex.TotalVertexCount()) - ccw.totalNeighborSet = make([]serialize.MapUint32Struct, ccw.WContext.GraphData.Vertex.TotalVertexCount()) - ccw.selfNeighborSet = make([]map[serialize.SUint32]struct{}, ccw.WContext.GraphData.VertexCount) + if err := validateTriangleMaxStep(options.GetInt(ccw.WContext.Params, + "compute.max_step")); err != nil { + return err + } + totalVertices := ccw.WContext.GraphData.Vertex.TotalVertexCount() + localVertices := ccw.WContext.GraphData.VertexCount + ccw.ccValues = make([]serialize.SFloat32, totalVertices) + ccw.degrees = make([]serialize.SUint32, totalVertices) + ccw.orientedNeighbors = make([]serialize.SliceUint32, totalVertices) + ccw.localNeighbors = make([]serialize.SliceUint32, localVertices) return nil } -func (ccw *ClusteringCoefficientWorker) Compute(vertexID uint32, pID int) { - _ = pID - vertID := vertexID - ccw.WContext.GraphData.VertIDStart +func (ccw *ClusteringCoefficientWorker) Compute(vertexID uint32, _ int) { + localID := vertexID - ccw.WContext.GraphData.VertIDStart switch ccw.WContext.Step { case 1: - inEdges := ccw.WContext.GraphData.Edges.GetInEdges(vertID) - outEdges := ccw.WContext.GraphData.Edges.GetOutEdges(vertID) - ccw.selfNeighborSet[vertID] = make(map[serialize.SUint32]struct{}, - len(inEdges)+len(outEdges)) - ccw.totalNeighborSet[vertexID] = make(map[serialize.SUint32]struct{}, - (len(inEdges)+len(outEdges))/2) - for _, nID := range inEdges { - if uint32(nID) == vertexID { - continue - } - ccw.selfNeighborSet[vertID][nID] = struct{}{} - if uint32(nID) < vertexID { - ccw.totalNeighborSet[vertexID][nID] = struct{}{} - } - } - for _, nID := range outEdges { - if uint32(nID) == vertexID { - continue - } - ccw.selfNeighborSet[vertID][nID] = struct{}{} - if uint32(nID) < vertexID { - ccw.totalNeighborSet[vertexID][nID] = struct{}{} - } - } + neighbors := sortedUndirectedNeighbors(vertexID, + uint32(len(ccw.degrees)), + ccw.WContext.GraphData.Edges.GetInEdges(localID), + ccw.WContext.GraphData.Edges.GetOutEdges(localID)) + ccw.localNeighbors[localID] = neighbors + ccw.degrees[vertexID] = serialize.SUint32(len(neighbors)) case 2: - length := len(ccw.selfNeighborSet[vertID]) - if length > 1 { - var triangleCount serialize.SUint32 - for fID := range ccw.selfNeighborSet[vertID] { - triangleCount += ccw.findCross(ccw.selfNeighborSet[vertID], ccw.totalNeighborSet[fID]) - } - ccw.ccValues[vertexID] = 2 * serialize.SFloat32(triangleCount) / - serialize.SFloat32((length)*(length-1)) - } else { + ccw.orientedNeighbors[vertexID] = orientedNeighbors(vertexID, + ccw.localNeighbors[localID], ccw.degrees) + case 3: + degree := len(ccw.localNeighbors[localID]) + if degree < 2 { ccw.ccValues[vertexID] = 0 + return } + triangles := countVertexTriangles(ccw.localNeighbors[localID], + ccw.orientedNeighbors) + denominator := float64(degree) * float64(degree-1) + ccw.ccValues[vertexID] = serialize.SFloat32( + 2 * float64(triangles) / denominator) } } -func (ccw *ClusteringCoefficientWorker) findCross(a map[serialize.SUint32]struct{}, b map[serialize.SUint32]struct{}) serialize.SUint32 { - var count serialize.SUint32 = 0 - if len(a) > len(b) { - for k := range b { - if _, ok := a[k]; ok { - count++ - } - } - } else { - for k := range a { - if _, ok := b[k]; ok { - count++ - } - } - } - return count -} func (ccw *ClusteringCoefficientWorker) OutputValueType() string { return common.HgValueTypeFloat } @@ -138,5 +120,5 @@ type ClusteringCoefficientMaster struct { } func (ccm *ClusteringCoefficientMaster) Compute() bool { - return ccm.MContext.Step != 2 + return ccm.MContext.Step != triangleComputeSteps } diff --git a/vermeer/algorithms/triangle_count.go b/vermeer/algorithms/triangle_count.go index 3dcc91f6c..ab64ad9ad 100644 --- a/vermeer/algorithms/triangle_count.go +++ b/vermeer/algorithms/triangle_count.go @@ -20,6 +20,7 @@ package algorithms import ( "vermeer/apps/common" "vermeer/apps/compute" + "vermeer/apps/options" "vermeer/apps/serialize" ) @@ -49,96 +50,64 @@ func (tc *TriangleCountMaker) DataNeeded() []string { type TriangleCountWorker struct { compute.WorkerComputerBase - triangleCount []serialize.SUint32 - totalNeighborSet []serialize.MapUint32Struct - selfNeighborSet []map[serialize.SUint32]struct{} + triangleCount []serialize.SInt64 + degrees []serialize.SUint32 + orientedNeighbors []serialize.SliceUint32 + localNeighbors []serialize.SliceUint32 } -func (tcw *TriangleCountWorker) VertexValue(i uint32) serialize.MarshalAble { - if tcw.WContext.Step == 1 { - return &tcw.totalNeighborSet[i] +func (tcw *TriangleCountWorker) VertexValue(vertexID uint32) serialize.MarshalAble { + if tcw.WContext.Output { + return &tcw.triangleCount[vertexID] + } + switch tcw.WContext.Step { + case 1: + return &tcw.degrees[vertexID] + case 2: + return &tcw.orientedNeighbors[vertexID] + default: + return &noTriangleScatterValue } - return &tcw.triangleCount[i] } func (tcw *TriangleCountWorker) Init() error { - tcw.triangleCount = make([]serialize.SUint32, tcw.WContext.GraphData.Vertex.TotalVertexCount()) - tcw.totalNeighborSet = make([]serialize.MapUint32Struct, tcw.WContext.GraphData.Vertex.TotalVertexCount()) - tcw.selfNeighborSet = make([]map[serialize.SUint32]struct{}, tcw.WContext.GraphData.VertexCount) + if err := validateTriangleMaxStep(options.GetInt(tcw.WContext.Params, + "compute.max_step")); err != nil { + return err + } + totalVertices := tcw.WContext.GraphData.Vertex.TotalVertexCount() + localVertices := tcw.WContext.GraphData.VertexCount + tcw.triangleCount = make([]serialize.SInt64, totalVertices) + tcw.degrees = make([]serialize.SUint32, totalVertices) + tcw.orientedNeighbors = make([]serialize.SliceUint32, totalVertices) + tcw.localNeighbors = make([]serialize.SliceUint32, localVertices) return nil } -func (tcw *TriangleCountWorker) Compute(vertexID uint32, pID int) { - _ = pID - vertID := vertexID - tcw.WContext.GraphData.VertIDStart +func (tcw *TriangleCountWorker) Compute(vertexID uint32, _ int) { + localID := vertexID - tcw.WContext.GraphData.VertIDStart switch tcw.WContext.Step { case 1: - //if tcw.WContext.GraphData.BothEdges != nil { - // tcw.selfNeighborSet[vertID] = make(map[serialize.SUint32]struct{}, - // len(tcw.WContext.GraphData.BothEdges[vertID])) - // tcw.totalNeighborSet[vertexID] = make(map[serialize.SUint32]struct{}, - // len(tcw.WContext.GraphData.BothEdges[vertID])/2) - // for _, nID := range tcw.WContext.GraphData.BothEdges[vertID] { - // if uint32(nID) == vertexID { - // continue - // } - // tcw.selfNeighborSet[vertID][nID] = struct{}{} - // if uint32(nID) < vertexID { - // tcw.totalNeighborSet[vertexID][nID] = struct{}{} - // } - // } - //} else { - inEdges := tcw.WContext.GraphData.Edges.GetInEdges(vertID) - outEdges := tcw.WContext.GraphData.Edges.GetOutEdges(vertID) - tcw.selfNeighborSet[vertID] = make(map[serialize.SUint32]struct{}, - len(inEdges)+len(outEdges)) - tcw.totalNeighborSet[vertexID] = make(map[serialize.SUint32]struct{}, - (len(inEdges)+len(outEdges))/2) - for _, nID := range inEdges { - if uint32(nID) == vertexID { - continue - } - tcw.selfNeighborSet[vertID][nID] = struct{}{} - if uint32(nID) < vertexID { - tcw.totalNeighborSet[vertexID][nID] = struct{}{} - } - } - for _, nID := range outEdges { - if uint32(nID) == vertexID { - continue - } - tcw.selfNeighborSet[vertID][nID] = struct{}{} - if uint32(nID) < vertexID { - tcw.totalNeighborSet[vertexID][nID] = struct{}{} - } - } - //} + neighbors := sortedUndirectedNeighbors(vertexID, + uint32(len(tcw.degrees)), + tcw.WContext.GraphData.Edges.GetInEdges(localID), + tcw.WContext.GraphData.Edges.GetOutEdges(localID)) + tcw.localNeighbors[localID] = neighbors + tcw.degrees[vertexID] = serialize.SUint32(len(neighbors)) case 2: - for fID := range tcw.selfNeighborSet[vertID] { - tcw.triangleCount[vertexID] += tcw.findCross(tcw.selfNeighborSet[vertID], tcw.totalNeighborSet[fID]) - } + tcw.orientedNeighbors[vertexID] = orientedNeighbors(vertexID, + tcw.localNeighbors[localID], tcw.degrees) + case 3: + // The largest simple graph addressable by uint32 vertex IDs has fewer + // than MaxInt64 triangles incident to one vertex, so this cast is safe. + tcw.triangleCount[vertexID] = serialize.SInt64( + countVertexTriangles(tcw.localNeighbors[localID], + tcw.orientedNeighbors)) } } -func (tcw *TriangleCountWorker) findCross(a map[serialize.SUint32]struct{}, b map[serialize.SUint32]struct{}) serialize.SUint32 { - var count serialize.SUint32 = 0 - if len(a) > len(b) { - for k := range b { - if _, ok := a[k]; ok { - count++ - } - } - } else { - for k := range a { - if _, ok := b[k]; ok { - count++ - } - } - } - return count -} func (tcw *TriangleCountWorker) OutputValueType() string { - return common.HgValueTypeInt + return common.HgValueTypeLong } type TriangleCountMaster struct { @@ -146,5 +115,5 @@ type TriangleCountMaster struct { } func (tcm *TriangleCountMaster) Compute() bool { - return tcm.MContext.Step != 2 + return tcm.MContext.Step != triangleComputeSteps } diff --git a/vermeer/algorithms/triangle_support.go b/vermeer/algorithms/triangle_support.go new file mode 100644 index 000000000..0614a215e --- /dev/null +++ b/vermeer/algorithms/triangle_support.go @@ -0,0 +1,183 @@ +/* +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/serialize" +) + +const triangleComputeSteps = 3 + +func validateTriangleMaxStep(maxStep int) error { + if maxStep < triangleComputeSteps { + return fmt.Errorf("triangle-based algorithms require compute.max_step >= %d, got %d", + triangleComputeSteps, maxStep) + } + return nil +} + +func sortedUndirectedNeighbors(vertexID, totalVertices uint32, inEdges, + outEdges serialize.SliceUint32) serialize.SliceUint32 { + neighbors := make(serialize.SliceUint32, 0, len(inEdges)+len(outEdges)) + for _, neighbor := range inEdges { + neighborID := uint32(neighbor) + if neighborID != vertexID && neighborID < totalVertices { + neighbors = append(neighbors, neighbor) + } + } + for _, neighbor := range outEdges { + neighborID := uint32(neighbor) + if neighborID != vertexID && neighborID < totalVertices { + neighbors = append(neighbors, neighbor) + } + } + if len(neighbors) < 2 { + return neighbors + } + + sort.Slice(neighbors, func(i, j int) bool { + return neighbors[i] < neighbors[j] + }) + write := 1 + for read := 1; read < len(neighbors); read++ { + if neighbors[read] == neighbors[write-1] { + continue + } + neighbors[write] = neighbors[read] + write++ + } + return neighbors[:write] +} + +func rankBefore(leftID, rightID uint32, degrees []serialize.SUint32) bool { + leftDegree := degrees[leftID] + rightDegree := degrees[rightID] + if leftDegree != rightDegree { + return leftDegree < rightDegree + } + return leftID < rightID +} + +func orientedNeighbors(vertexID uint32, neighbors serialize.SliceUint32, + degrees []serialize.SUint32) serialize.SliceUint32 { + count := 0 + for _, rawNeighbor := range neighbors { + neighborID := uint32(rawNeighbor) + if neighborID < uint32(len(degrees)) && + rankBefore(vertexID, neighborID, degrees) { + count++ + } + } + if count == 0 { + return nil + } + + oriented := make(serialize.SliceUint32, 0, count) + for _, rawNeighbor := range neighbors { + neighborID := uint32(rawNeighbor) + if neighborID < uint32(len(degrees)) && + rankBefore(vertexID, neighborID, degrees) { + oriented = append(oriented, rawNeighbor) + } + } + return oriented +} + +func sortedIntersectionCount(left, right serialize.SliceUint32) uint64 { + if len(left) == 0 || len(right) == 0 { + return 0 + } + if len(left) > len(right) { + left, right = right, left + } + + // A binary-search intersection avoids scanning a high-degree adjacency list + // once for every low-degree oriented neighbor. For similar-sized slices, the + // linear merge has better constants and cache locality. + if len(right) > 8*len(left) { + var count uint64 + for _, value := range left { + index := sort.Search(len(right), func(index int) bool { + return right[index] >= value + }) + if index < len(right) && right[index] == value { + count++ + } + } + return count + } + + var count uint64 + leftIndex := 0 + rightIndex := 0 + for leftIndex < len(left) && rightIndex < len(right) { + switch { + case left[leftIndex] < right[rightIndex]: + leftIndex++ + case left[leftIndex] > right[rightIndex]: + rightIndex++ + default: + count++ + leftIndex++ + rightIndex++ + } + } + return count +} + +func countVertexTriangles(neighbors serialize.SliceUint32, + orientedByVertex []serialize.SliceUint32) uint64 { + var count uint64 + for _, neighbor := range neighbors { + neighborID := uint32(neighbor) + if neighborID >= uint32(len(orientedByVertex)) { + continue + } + // For each pair of neighbors in a triangle, exactly one endpoint is + // earlier in the (degree, vertex-ID) order. Its oriented adjacency + // therefore counts the pair once, independent of the root vertex rank. + count += sortedIntersectionCount(neighbors, + orientedByVertex[neighborID]) + } + return count +} + +type emptyTriangleScatterValue struct{} + +func (*emptyTriangleScatterValue) Marshal([]byte) (int, error) { + return 0, nil +} + +func (*emptyTriangleScatterValue) Unmarshal(buffer []byte) (int, error) { + if len(buffer) != 0 { + return 0, fmt.Errorf("unexpected triangle scatter payload: %d bytes", len(buffer)) + } + return 0, nil +} + +func (*emptyTriangleScatterValue) ToString() string { + return "" +} + +func (*emptyTriangleScatterValue) PredictSize() int { + return 0 +} + +var noTriangleScatterValue emptyTriangleScatterValue diff --git a/vermeer/algorithms/triangle_support_test.go b/vermeer/algorithms/triangle_support_test.go new file mode 100644 index 000000000..43af1e600 --- /dev/null +++ b/vermeer/algorithms/triangle_support_test.go @@ -0,0 +1,298 @@ +/* +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" + "reflect" + "strconv" + "testing" + "vermeer/apps/common" + "vermeer/apps/options" + "vermeer/apps/serialize" + "vermeer/apps/structure" +) + +func init() { + options.Init() +} + +func TestSortedUndirectedNeighborsDeduplicatesAndDropsSelfLoops(t *testing.T) { + got := sortedUndirectedNeighbors(5, 8, + serialize.SliceUint32{7, 3, 5, 7}, + serialize.SliceUint32{2, 3, 5}) + want := serialize.SliceUint32{2, 3, 7} + if !reflect.DeepEqual(got, want) { + t.Fatalf("expected %v, got %v", want, got) + } +} + +func TestSortedUndirectedNeighborsDropsInvalidRemoteIDs(t *testing.T) { + got := sortedUndirectedNeighbors(0, 3, + serialize.SliceUint32{1, 99}, + serialize.SliceUint32{2, 99}) + want := serialize.SliceUint32{1, 2} + if !reflect.DeepEqual(got, want) { + t.Fatalf("expected %v after invalid-ID filtering, got %v", want, got) + } +} + +func TestDegreeOrientation(t *testing.T) { + degrees := []serialize.SUint32{3, 1, 3, 5} + neighbors := serialize.SliceUint32{0, 1, 2, 3} + got := orientedNeighbors(2, neighbors, degrees) + want := serialize.SliceUint32{3} + if !reflect.DeepEqual(got, want) { + t.Fatalf("expected %v, got %v", want, got) + } + if cap(got) != len(got) { + t.Fatalf("oriented neighbors retain capacity %d for %d values", cap(got), len(got)) + } + if !rankBefore(0, 2, degrees) || rankBefore(2, 0, degrees) { + t.Fatal("equal-degree IDs did not provide a stable total order") + } +} + +func TestSortedIntersectionCount(t *testing.T) { + tests := []struct { + name string + left serialize.SliceUint32 + right serialize.SliceUint32 + want uint64 + }{ + { + name: "linear_merge", + left: serialize.SliceUint32{1, 3, 4, 8}, + right: serialize.SliceUint32{0, 3, 8, 9}, + want: 2, + }, + { + name: "binary_search", + left: serialize.SliceUint32{5, 99}, + right: serialize.SliceUint32{ + 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, + 10, 11, 12, 13, 14, 15, 16, 17, 99, + }, + want: 2, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := sortedIntersectionCount(test.left, test.right); got != test.want { + t.Fatalf("expected %d, got %d", test.want, got) + } + }) + } +} + +func TestTriangleCountAndClusteringOnK4(t *testing.T) { + graph := completeUndirectedGraph(4) + + triangle := newTriangleTestWorker(t, graph, 3) + runThreeStepTriangleWorker(triangle, 4) + for vertexID, value := range triangle.triangleCount { + if value != 3 { + t.Fatalf("vertex %d expected 3 triangles, got %d", vertexID, value) + } + } + + coefficient := newClusteringTestWorker(t, graph, 3) + runThreeStepClusteringWorker(coefficient, 4) + for vertexID, value := range coefficient.ccValues { + if math.Abs(float64(value)-1) > 1e-6 { + t.Fatalf("vertex %d expected coefficient 1, got %v", vertexID, value) + } + } +} + +func TestTriangleCountAndClusteringDoNotCountOpenWedge(t *testing.T) { + graph := &structure.GraphData{ + VertexCount: 3, + Vertex: &structure.VertexMem{ + TotalVertex: make([]structure.Vertex, 3), + }, + Edges: &structure.EdgeMem{ + InEdges: serialize.TwoDimSliceUint32{ + {1}, {0, 2}, {1}, + }, + OutEdges: serialize.TwoDimSliceUint32{ + {1}, {0, 2}, {1}, + }, + }, + } + + triangle := newTriangleTestWorker(t, graph, 3) + runThreeStepTriangleWorker(triangle, 3) + for vertexID, value := range triangle.triangleCount { + if value != 0 { + t.Fatalf("open-wedge vertex %d reported %d triangles", vertexID, value) + } + } + + coefficient := newClusteringTestWorker(t, graph, 3) + runThreeStepClusteringWorker(coefficient, 3) + for vertexID, value := range coefficient.ccValues { + if value != 0 { + t.Fatalf("open-wedge vertex %d reported coefficient %v", vertexID, value) + } + } +} + +func TestTriangleWorkersTreatAsymmetricInputAsUndirected(t *testing.T) { + graph := &structure.GraphData{ + VertexCount: 3, + Vertex: &structure.VertexMem{ + TotalVertex: make([]structure.Vertex, 3), + }, + Edges: &structure.EdgeMem{ + // Directed input 0 -> 1 -> 2 -> 0. Unioning in/out adjacency + // produces one undirected triangle, matching the existing contract. + InEdges: serialize.TwoDimSliceUint32{ + {2}, {0}, {1}, + }, + OutEdges: serialize.TwoDimSliceUint32{ + {1}, {2}, {0}, + }, + }, + } + + worker := newTriangleTestWorker(t, graph, 3) + runThreeStepTriangleWorker(worker, 3) + for vertexID, value := range worker.triangleCount { + if value != 1 { + t.Fatalf("vertex %d expected one triangle, got %d", vertexID, value) + } + } +} + +func TestTriangleCountUsesInt64Output(t *testing.T) { + worker := &TriangleCountWorker{ + triangleCount: []serialize.SInt64{serialize.SInt64(math.MaxInt32) + 1}, + } + context := worker.MakeContext(nil) + context.Output = true + value, ok := worker.VertexValue(0).(*serialize.SInt64) + if !ok { + t.Fatalf("triangle output type is %T, expected *serialize.SInt64", + worker.VertexValue(0)) + } + if int64(*value) != int64(math.MaxInt32)+1 { + t.Fatalf("64-bit triangle count was truncated: %d", *value) + } + if worker.OutputValueType() != common.HgValueTypeLong { + t.Fatalf("triangle HugeGraph output type is %q, expected LONG", + worker.OutputValueType()) + } +} + +func TestTriangleStepThreeScatterPayloadIsEmpty(t *testing.T) { + worker := &TriangleCountWorker{} + context := worker.MakeContext(nil) + context.Step = 3 + value := worker.VertexValue(0) + if size := value.PredictSize(); size != 0 { + t.Fatalf("step-three scatter value predicts %d bytes, expected 0", size) + } + if _, err := value.Unmarshal([]byte{1}); err == nil { + t.Fatal("non-empty step-three scatter payload was accepted") + } +} + +func TestTriangleAlgorithmsRejectShortStepBudget(t *testing.T) { + if err := validateTriangleMaxStep(2); err == nil { + t.Fatal("compute.max_step=2 was accepted") + } + if err := validateTriangleMaxStep(3); err != nil { + t.Fatalf("compute.max_step=3 was rejected: %v", err) + } +} + +func completeUndirectedGraph(vertexCount uint32) *structure.GraphData { + inEdges := make(serialize.TwoDimSliceUint32, vertexCount) + outEdges := make(serialize.TwoDimSliceUint32, vertexCount) + for vertexID := uint32(0); vertexID < vertexCount; vertexID++ { + for neighborID := uint32(0); neighborID < vertexCount; neighborID++ { + if vertexID == neighborID { + continue + } + inEdges[vertexID] = append(inEdges[vertexID], + serialize.SUint32(neighborID)) + outEdges[vertexID] = append(outEdges[vertexID], + serialize.SUint32(neighborID)) + } + } + return &structure.GraphData{ + VertexCount: vertexCount, + Vertex: &structure.VertexMem{ + TotalVertex: make([]structure.Vertex, vertexCount), + }, + Edges: &structure.EdgeMem{ + InEdges: inEdges, + OutEdges: outEdges, + }, + } +} + +func newTriangleTestWorker(t *testing.T, graph *structure.GraphData, + maxStep int) *TriangleCountWorker { + t.Helper() + worker := &TriangleCountWorker{} + context := worker.MakeContext(map[string]string{ + "compute.max_step": strconv.Itoa(maxStep), + }) + context.GraphData = graph + context.Parallel = 1 + if err := worker.Init(); err != nil { + t.Fatalf("triangle init: %v", err) + } + return worker +} + +func newClusteringTestWorker(t *testing.T, graph *structure.GraphData, + maxStep int) *ClusteringCoefficientWorker { + t.Helper() + worker := &ClusteringCoefficientWorker{} + context := worker.MakeContext(map[string]string{ + "compute.max_step": strconv.Itoa(maxStep), + }) + context.GraphData = graph + context.Parallel = 1 + if err := worker.Init(); err != nil { + t.Fatalf("clustering init: %v", err) + } + return worker +} + +func runThreeStepTriangleWorker(worker *TriangleCountWorker, vertexCount uint32) { + for step := int32(1); step <= triangleComputeSteps; step++ { + worker.WContext.Step = step + for vertexID := uint32(0); vertexID < vertexCount; vertexID++ { + worker.Compute(vertexID, 0) + } + } +} + +func runThreeStepClusteringWorker(worker *ClusteringCoefficientWorker, + vertexCount uint32) { + for step := int32(1); step <= triangleComputeSteps; step++ { + worker.WContext.Step = step + for vertexID := uint32(0); vertexID < vertexCount; vertexID++ { + worker.Compute(vertexID, 0) + } + } +} diff --git a/vermeer/apps/common/value_types.go b/vermeer/apps/common/value_types.go new file mode 100644 index 000000000..87aad1494 --- /dev/null +++ b/vermeer/apps/common/value_types.go @@ -0,0 +1,20 @@ +/* +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 common + +const HgValueTypeLong = "LONG" diff --git a/vermeer/test/functional/compute_clustering_coefficient.go b/vermeer/test/functional/compute_clustering_coefficient.go index 4f7d4bfac..aecc38eb8 100644 --- a/vermeer/test/functional/compute_clustering_coefficient.go +++ b/vermeer/test/functional/compute_clustering_coefficient.go @@ -27,7 +27,7 @@ func (ct *ClusteringCoefficientTest) TaskComputeBody() map[string]string { //获取compute任务的body return map[string]string{ "compute.algorithm": "clustering_coefficient", - "compute.max_step": "2", + "compute.max_step": "3", "compute.parallel": "100", "output.parallel": "10", "output.file_path": "./data/" + ct.algoName, diff --git a/vermeer/test/functional/compute_trianglecount.go b/vermeer/test/functional/compute_trianglecount.go index 34edb7379..e7bfec812 100644 --- a/vermeer/test/functional/compute_trianglecount.go +++ b/vermeer/test/functional/compute_trianglecount.go @@ -27,7 +27,7 @@ func (tct *TriangleCountTest) TaskComputeBody() map[string]string { //获取compute任务的body return map[string]string{ "compute.algorithm": "triangle_count", - "compute.max_step": "2", + "compute.max_step": "3", "compute.parallel": "100", "output.file_path": "./data/" + tct.algoName, "output.type": "local",