BFS Push/Pull, SSSP, SpMSpV

benchmarks/GraphBLAS-sharp.Benchmarks/BenchmarksBFS.fs

@@ -29,7 +29,7 @@ type BFSBenchmarks<'matrixT, 'elem when 'matrixT :> IDeviceMemObject and 'elem :
 
     let source = 0
 
-    member val ResultVector = Unchecked.defaultof<ClArray<'elem option>> with get,set
+    member val ResultVector = Unchecked.defaultof<ClArray<int option>> with get,set
 
     [<ParamsSource("AvaliableContexts")>]
     member val OclContextInfo = Unchecked.defaultof<Utils.BenchmarkContext * int> with get, set
@@ -129,6 +129,64 @@ type BFSBenchmarksWithoutDataTransfer() =
         this.BFS()
         this.Processor.PostAndReply(Msg.MsgNotifyMe)
 
+type SSSPBenchmarksWithoutDataTransfer() =
+
+    inherit BFSBenchmarks<ClMatrix.CSR<int>, int>(
+        (fun context wgSize -> SSSP.run context wgSize),
+        int,
+        (fun _ -> Utils.nextInt (System.Random())),
+        Matrix.ToBackendCSR)
+
+    static member InputMatricesProvider =
+        BFSBenchmarks<_,_>.InputMatricesProviderBuilder "BFSBenchmarks.txt"
+
+    [<GlobalSetup>]
+    override this.GlobalSetup() =
+        this.ReadMatrix ()
+        this.LoadMatrixToGPU ()
+
+    [<IterationCleanup>]
+    override this.IterationCleanup() =
+        this.ClearResult()
+
+    [<GlobalCleanup>]
+    override this.GlobalCleanup() =
+        this.ClearInputMatrix()
+
+    [<Benchmark>]
+    override this.Benchmark() =
+        this.BFS()
+        this.Processor.PostAndReply(Msg.MsgNotifyMe)
+
+type BFSPushPullBenchmarksWithoutDataTransfer() =
+
+    inherit BFSBenchmarks<ClMatrix.CSR<bool>, bool>(
+        (fun context wgSize -> BFS.singleSourcePushPull context ArithmeticOperations.boolSum ArithmeticOperations.boolMul wgSize),
+        (fun _ -> true),
+        (fun _ -> true),
+        Matrix.ToBackendCSR)
+
+    static member InputMatricesProvider =
+        BFSBenchmarks<_,_>.InputMatricesProviderBuilder "BFSBenchmarks.txt"
+
+    [<GlobalSetup>]
+    override this.GlobalSetup() =
+        this.ReadMatrix ()
+        this.LoadMatrixToGPU ()
+
+    [<IterationCleanup>]
+    override this.IterationCleanup() =
+        this.ClearResult()
+
+    [<GlobalCleanup>]
+    override this.GlobalCleanup() =
+        this.ClearInputMatrix()
+
+    [<Benchmark>]
+    override this.Benchmark() =
+        this.BFS()
+        this.Processor.PostAndReply(Msg.MsgNotifyMe)
+
 type BFSBenchmarksWithDataTransfer<'matrixT, 'elem when 'matrixT :> IDeviceMemObject and 'elem : struct>(
         buildFunToBenchmark,
         converter: string -> 'elem,

benchmarks/GraphBLAS-sharp.Benchmarks/Program.fs

@@ -4,7 +4,9 @@ open BenchmarkDotNet.Running
 [<EntryPoint>]
 let main argv =
     let benchmarks =
-        BenchmarkSwitcher [| typeof<BFSBenchmarksWithoutDataTransfer> |]
+        BenchmarkSwitcher [| typeof<BFSBenchmarksWithoutDataTransfer>
+                             typeof<BFSPushPullBenchmarksWithoutDataTransfer>
+                             typeof<SSSPBenchmarksWithoutDataTransfer> |]
 
     benchmarks.Run argv |> ignore
     0

src/GraphBLAS-sharp.Backend/Algorithms/BFS.fs

@@ -11,6 +11,7 @@ open GraphBLAS.FSharp.Backend.Vector.Dense
 open GraphBLAS.FSharp.Backend.Objects.ClContext
 open GraphBLAS.FSharp.Backend.Objects.ArraysExtensions
 open GraphBLAS.FSharp.Backend.Objects.ClCell
+open GraphBLAS.FSharp.Backend.Vector.Sparse
 
 module BFS =
     let singleSource
@@ -40,7 +41,7 @@ module BFS =
         fun (queue: MailboxProcessor<Msg>) (matrix: ClMatrix.CSR<'a>) (source: int) ->
             let vertexCount = matrix.RowCount
 
-            let levels = zeroCreate queue HostInterop vertexCount
+            let levels = zeroCreate queue DeviceOnly vertexCount
 
             let frontier =
                 ofList queue DeviceOnly Dense vertexCount [ source, 1 ]
@@ -71,3 +72,183 @@ module BFS =
 
                 levels
             | _ -> failwith "Not implemented"
+
+
+    let singleSourceSparse
+        (clContext: ClContext)
+        (add: Expr<bool option -> bool option -> bool option>)
+        (mul: Expr<bool option -> bool option -> bool option>)
+        workGroupSize
+        =
+
+        let spMSpV =
+            SpMSpV.run clContext add mul workGroupSize
+
+        let zeroCreate =
+            ClArray.zeroCreate clContext workGroupSize
+
+        let ofList = Vector.ofList clContext workGroupSize
+
+        let maskComplemented =
+            SparseVector.map2SparseDense clContext Mask.complementedOp workGroupSize
+
+        let fillSubVectorTo =
+            DenseVector.assignBySparseMaskInplace clContext (Convert.assignToOption Mask.assign) workGroupSize
+
+        let containsNonZero =
+            SparseVector.exists clContext workGroupSize (Predicates.notEquals false)
+
+        fun (queue: MailboxProcessor<Msg>) (matrix: ClMatrix.CSR<bool>) (source: int) ->
+            let vertexCount = matrix.RowCount
+
+            let levels = zeroCreate queue HostInterop vertexCount
+
+            let mutable frontier =
+                ofList queue DeviceOnly Sparse vertexCount [ source, true ]
+
+            let mutable level = 0
+            let mutable stop = false
+
+            while not stop do
+                match frontier with
+                | ClVector.Sparse front ->
+                    level <- level + 1
+
+                    //Assigning new level values
+                    fillSubVectorTo queue levels front (clContext.CreateClCell level) levels
+
+                    //Getting new frontier
+                    let newFrontier = spMSpV queue matrix front
+
+                    frontier.Dispose queue
+
+                    frontier <- ClVector.Sparse(maskComplemented queue DeviceOnly newFrontier levels)
+
+                    newFrontier.Dispose queue
+
+                    //Checking if front is empty
+                    match frontier with
+                    | ClVector.Sparse front ->
+                        stop <-
+                            not
+                            <| (containsNonZero queue front).ToHostAndFree queue
+                    | _ -> ()
+
+                | _ -> failwith "Not implemented"
+
+            frontier.Dispose queue
+
+            levels
+
+    let singleSourcePushPull
+        (clContext: ClContext)
+        (add: Expr<bool option -> bool option -> bool option>)
+        (mul: Expr<bool option -> bool option -> bool option>)
+        workGroupSize
+        =
+
+        let SPARSITY = 0.001f
+
+        let spMVTo =
+            SpMV.runTo clContext add mul workGroupSize
+
+        let spMSpV =
+            SpMSpV.run clContext add mul workGroupSize
+
+        let zeroCreate =
+            ClArray.zeroCreate clContext workGroupSize
+
+        let ofList = Vector.ofList clContext workGroupSize
+
+        let maskComplementedTo =
+            DenseVector.map2Inplace clContext Mask.complementedOp workGroupSize
+
+        let maskComplemented =
+            SparseVector.map2SparseDense clContext Mask.complementedOp workGroupSize
+
+        let fillSubVectorDenseTo =
+            DenseVector.assignByMaskInplace clContext (Convert.assignToOption Mask.assign) workGroupSize
+
+        let fillSubVectorSparseTo =
+            DenseVector.assignBySparseMaskInplace clContext (Convert.assignToOption Mask.assign) workGroupSize
+
+        let containsNonZeroSparse =
+            SparseVector.exists clContext workGroupSize (Predicates.notEquals false)
+
+        let toSparse =
+            DenseVector.toSparse clContext workGroupSize
+
+        let toDense =
+            SparseVector.toDense clContext workGroupSize
+
+        let countNNZ =
+            ClArray.count clContext workGroupSize Predicates.isSome
+
+        fun (queue: MailboxProcessor<Msg>) (matrix: ClMatrix.CSR<bool>) (source: int) ->
+            let vertexCount = matrix.RowCount
+
+            let levels = zeroCreate queue HostInterop vertexCount
+
+            let mutable frontier =
+                ofList queue DeviceOnly Sparse vertexCount [ source, true ]
+
+            let mutable level = 0
+            let mutable stop = false
+
+            while not stop do
+                level <- level + 1
+
+                match frontier with
+                | ClVector.Sparse front ->
+                    //Assigning new level values
+                    fillSubVectorSparseTo queue levels front (clContext.CreateClCell level) levels
+
+                    //Getting new frontier
+                    let newFrontier = spMSpV queue matrix front
+
+                    frontier.Dispose queue
+
+                    let newMaskedFrontier =
+                        maskComplemented queue DeviceOnly newFrontier levels
+
+                    newFrontier.Dispose queue
+
+                    //Checking if front is empty
+                    stop <-
+                        not
+                        <| (containsNonZeroSparse queue newMaskedFrontier)
+                            .ToHostAndFree queue
+
+                    if not stop then
+                        //Push/pull
+                        if ((float32 newMaskedFrontier.NNZ)
+                            / (float32 newMaskedFrontier.Size)
+                            <= SPARSITY) then
+                            frontier <- ClVector.Sparse newMaskedFrontier
+                        else
+                            frontier <- ClVector.Dense(toDense queue DeviceOnly newMaskedFrontier)
+                            newMaskedFrontier.Dispose queue
+
+                | ClVector.Dense front ->
+                    //Assigning new level values
+                    fillSubVectorDenseTo queue levels front (clContext.CreateClCell level) levels
+
+                    //Getting new frontier
+                    spMVTo queue matrix front front
+
+                    maskComplementedTo queue front levels front
+
+                    //Emptiness check
+                    let NNZ = countNNZ queue front
+
+                    stop <- NNZ = 0
+
+                    //Push/pull
+                    if not stop then
+                        if ((float32 NNZ) / (float32 front.Length) <= SPARSITY) then
+                            frontier <- ClVector.Sparse(toSparse queue DeviceOnly front)
+                            front.Dispose queue
+
+            frontier.Dispose queue
+
+            levels

src/GraphBLAS-sharp.Backend/Algorithms/SSSP.fs

@@ -0,0 +1,95 @@
+namespace GraphBLAS.FSharp.Backend.Algorithms
+
+open GraphBLAS.FSharp.Backend
+open Brahma.FSharp
+open FSharp.Quotations
+open GraphBLAS.FSharp.Backend.Objects
+open GraphBLAS.FSharp.Backend.Common
+open GraphBLAS.FSharp.Backend.Quotes
+open GraphBLAS.FSharp.Backend.Vector
+open GraphBLAS.FSharp.Backend.Vector.Dense
+open GraphBLAS.FSharp.Backend.Objects.ClContext
+open GraphBLAS.FSharp.Backend.Objects.ArraysExtensions
+open GraphBLAS.FSharp.Backend.Objects.ClCell
+
+module SSSP =
+    let run (clContext: ClContext) workGroupSize =
+
+        let less = ArithmeticOperations.less<int>
+        let min = ArithmeticOperations.min<int>
+        let plus = ArithmeticOperations.intSumExplicit
+
+        let spMVTo =
+            SpMV.runTo clContext min plus workGroupSize
+
+        let create = ClArray.create clContext workGroupSize
+
+        let createMask = ClArray.create clContext workGroupSize
+
+        let ofList = Vector.ofList clContext workGroupSize
+
+        let eWiseMulLess =
+            ClArray.map2Inplace clContext workGroupSize less
+
+        let eWiseAddMin =
+            ClArray.map2Inplace clContext workGroupSize min
+
+        let fillSubVectorTo =
+            DenseVector.assignByMaskInplace clContext (Convert.assignToOption Mask.assignComplemented) workGroupSize
+
+        let containsNonZero =
+            ClArray.exists clContext workGroupSize Predicates.isSome
+
+        fun (queue: MailboxProcessor<Msg>) (matrix: ClMatrix.CSR<int>) (source: int) ->
+            let vertexCount = matrix.RowCount
+
+            //None is System.Int32.MaxValue
+            let distance =
+                ofList queue DeviceOnly Dense vertexCount [ source, 0 ]
+
+            let mutable f1 =
+                ofList queue DeviceOnly Dense vertexCount [ source, 0 ]
+
+            let mutable f2 =
+                create queue DeviceOnly vertexCount None
+                |> ClVector.Dense
+
+            let m =
+                createMask queue DeviceOnly vertexCount None
+                |> ClVector.Dense
+
+            let mutable stop = false
+
+            while not stop do
+                match f1, f2, distance, m with
+                | ClVector.Dense front1, ClVector.Dense front2, ClVector.Dense dist, ClVector.Dense mask ->
+                    //Getting new frontier
+                    spMVTo queue matrix front1 front2
+
+                    //Checking which distances were updated
+                    eWiseMulLess queue front2 dist mask
+                    //Updating
+                    eWiseAddMin queue dist front2 dist
+
+                    //Filtering unproductive vertices
+                    fillSubVectorTo queue front2 mask (clContext.CreateClCell 0) front2
+
+                    //Swap fronts
+                    let temp = f1
+                    f1 <- f2
+                    f2 <- temp
+
+                    //Checking if no distances were updated
+                    stop <-
+                        not
+                        <| (containsNonZero queue mask).ToHostAndFree(queue)
+
+                | _ -> failwith "not implemented"
+
+            f1.Dispose queue
+            f2.Dispose queue
+            m.Dispose queue
+
+            match distance with
+            | ClVector.Dense dist -> dist
+            | _ -> failwith "not implemented"