Streaming of preloaded images.

Author: gsvgit

src/ImageProcessing/ImageProcessing.fs

@@ -167,8 +167,6 @@ let applyFilter (filter: float32[][]) (img: byte[,]) =
 
     Array2D.mapi (fun x y _ -> byte (processPixel x y)) img
 
-
-
 let applyFilterCpuParallel (filter: float32[][]) (img: byte[,]) =
     let imgH = img.GetLength 0
     let imgW = img.GetLength 1

src/ImageProcessing/ImageProcessing.fsproj

@@ -19,7 +19,6 @@
     <None Include="App.config" />
     <Compile Include="ImageProcessing.fs" />
     <Compile Include="Streaming.fs" />
-    <Compile Include="Matrices.fs" />
     <Compile Include="Main.fs" />
   </ItemGroup>
   <ItemGroup>

src/ImageProcessing/Main.fs

@@ -4,7 +4,7 @@ open Argu
 open Argu.ArguAttributes
 open Brahma.FSharp
 
-type Platforms = CPU = 1 | CPUParallel = 2 | NVidia = 3 | IntelGPU = 4 | AnyGPU = 5
+type Platforms = CPU = 1 | CPUParallel = 2 | Nvidia = 3 | IntelGPU = 4 | AnyGPU = 5
 
 [<CliPrefix(CliPrefix.DoubleDash)>]
 [<NoAppSettings>]
@@ -13,7 +13,7 @@ type ImageProcessingArguments =
     | Output of string
     | Platform of Platforms
     | WorkGroupSize of uint
-    | MatrixSize of uint    
+    | Stream of list<Platforms>
     with
     interface IArgParserTemplate with
         member arg.Usage =
@@ -22,7 +22,8 @@ type ImageProcessingArguments =
             | Output _ -> "File to store result."
             | Platform _ -> "Where to run."
             | WorkGroupSize _ -> "Work group size."
-            | MatrixSize _ -> "Number of columns (or rows). We use square matrices."
+            | Stream _ -> "Folder processing on multiple devices."
+
 module Main =
     //let pathToExamples = "/home/gsv/Projects/TestProj2020/src/ImgProcessing/Examples"
     //let inputFolder = System.IO.Path.Combine(pathToExamples, "input")
@@ -40,32 +41,45 @@ module Main =
         let output = results.GetResult(Output, defaultValue = "out.jpg")
         let platform = results.GetResult(Platform, defaultValue = Platforms.CPU)
         let workGroupSize = results.GetResult(WorkGroupSize, defaultValue = 64u)
-        let matrixSize = results.GetResult(MatrixSize, defaultValue = 512u)
+        let devicesForStream = results.GetResult(Stream, defaultValue = [])
 
         let filters = [
+            ImageProcessing.gaussianBlurKernel
+            ImageProcessing.gaussianBlurKernel
             ImageProcessing.gaussianBlurKernel
             ImageProcessing.gaussianBlurKernel
             ImageProcessing.edgesKernel
         ]
 
-(*        
-        let applyFiltersOnGPU =
+        
+        let applyFiltersOnGPU platform =
             let device =
                 match platform with 
                 | Platforms.AnyGPU -> ClDevice.GetFirstAppropriateDevice()
                 | _ -> 
                     let platform =
                         match platform with 
-                        | Platforms.NVidia -> Platform.Nvidia
+                        | Platforms.Nvidia -> Platform.Nvidia
                         | Platforms.IntelGPU -> Platform.Intel
                     ClDevice.GetAvailableDevices(platform = platform)
                     |> Seq.head
             printfn $"Device: %A{device.Name}"
 
-            let context = ClContext(device)
+            let context = ClContext device
             ImageProcessing.applyFiltersGPU context 64
-            *)
 
+        match devicesForStream with 
+        | hd :: tl ->           
+          let appliers = List.map applyFiltersOnGPU devicesForStream |> List.map (fun f -> f filters)
+          let imagesToProcess = 
+            Streaming.listAllFiles input
+            |> List.map (fun file -> Streaming.Img(ImageProcessing.loadAsImage file))
+          let start = System.DateTime.Now
+          Streaming.processAllLoadedFiles imagesToProcess appliers
+          //Streaming.processAllFiles input output appliers
+          printfn  $"TotalTime = %f{(System.DateTime.Now - start).TotalMilliseconds}"
+
+        (*
         match platform with
         | Platforms.CPU -> 
             let mutable image = ImageProcessing.loadAs2DArray input
@@ -83,7 +97,7 @@ module Main =
                 image <- ImageProcessing.applyFilterCpuParallel filter image
             printfn $"CPU processing time: {(System.DateTime.Now - start).TotalMilliseconds} ms"
             ImageProcessing.save2DByteArrayAsImage image output
-        
+        *)
         (*| _ ->             
             let start = System.DateTime.Now
             let grayscaleImage = ImageProcessing.loadAsImage input

src/ImageProcessing/Matrices.fs

@@ -10,9 +10,8 @@ type msg =
     | Img of Image
     | EOS of AsyncReplyChannel<unit>
 
-let imgSaver outDir =
-    let outFile (imgName: string) = System.IO.Path.Combine(outDir, imgName)
-
+let imgSaver saveImage =
+    
     MailboxProcessor.Start(fun inbox ->
         let rec loop () = async {
             let! msg = inbox.Receive()
@@ -23,12 +22,13 @@ let imgSaver outDir =
                 ch.Reply()
             | Img img ->
                 printfn $"Save: %A{img.Name}"
-                saveImage img (outFile img.Name)
+                saveImage img
                 return! loop ()
         }
         loop ()
     )
 
+
 let imgProcessor filterApplicator (imgSaver: MailboxProcessor<_>) =
     let filter = filterApplicator
     MailboxProcessor.Start(fun inbox ->
@@ -50,26 +50,62 @@ let imgProcessor filterApplicator (imgSaver: MailboxProcessor<_>) =
     )
 
 let processAllFiles inDir outDir filterApplicators =
+
     let mutable cnt = 0
 
+    let outFile (imgName: string) = System.IO.Path.Combine(outDir, imgName)
+
+    let saveImageToDir (img:Image) = saveImage img (outFile img.Name)
+
     let imgProcessors =
         filterApplicators
         |> List.map (fun x ->
-            let imgSaver = imgSaver outDir
+            let imgSaver = imgSaver saveImageToDir
             imgProcessor x imgSaver
         )
         |> Array.ofList
 
     let filesToProcess = listAllFiles inDir
 
+    while cnt < List.length filesToProcess do
+        let p = (imgProcessors |> Array.minBy (fun p -> p.CurrentQueueLength))
+        if p.CurrentQueueLength < 3 
+        then 
+           p.Post (Img(loadAsImage (filesToProcess[cnt])))
+           cnt <- cnt + 1
+
+(*
     for file in filesToProcess do
-        //while (imgProcessors |> Array.minBy (fun p -> p.CurrentQueueLength)).CurrentQueueLength > 3 do ()
         (imgProcessors
          |> Array.minBy (fun p -> p.CurrentQueueLength))
             .Post(Img(loadAsImage file))
+*)
+
+    for imgProcessor in imgProcessors do
+        imgProcessor.PostAndReply EOS
 
-        cnt <- cnt + 1
+
+let processAllLoadedFiles inImages filterApplicators =    
+    let result = new ResizeArray<_>()
+
+    let mutable cnt = 0
+
+    let saveImageToArr (img:Image) = result.Add img
+
+    let imgProcessors =
+        filterApplicators
+        |> List.map (fun x ->
+            let imgSaver = imgSaver saveImageToArr
+            imgProcessor x imgSaver
+        )
+        |> Array.ofList
+    
+    while cnt < List.length inImages do
+        let p = (imgProcessors |> Array.minBy (fun p -> p.CurrentQueueLength))
+        if p.CurrentQueueLength < 3 
+        then 
+           p.Post inImages[cnt]
+           cnt <- cnt + 1
 
     for imgProcessor in imgProcessors do
         imgProcessor.PostAndReply EOS
-//imgSaver.PostAndReply EOS