Vấn đề hiệu suất thao tác hình ảnh F #

Question

Tôi hiện đang cố gắng cải thiện hiệu suất của một chương trình F # để làm cho nó nhanh như C# tương đương của nó. Chương trình này áp dụng một mảng bộ lọc cho một bộ đệm của các điểm ảnh. Truy cập vào bộ nhớ luôn được thực hiện bằng cách sử dụng con trỏ.

Đây là mã C# mà được áp dụng cho mỗi điểm ảnh của một hình ảnh:

unsafe private static byte getPixelValue(byte* buffer, double* filter, int filterLength, double filterSum) 
{ 
    double sum = 0.0; 
    for (int i = 0; i < filterLength; ++i) 
    { 
     sum += (*buffer) * (*filter); 
     ++buffer; 
     ++filter; 
    } 

    sum = sum/filterSum; 

    if (sum > 255) return 255; 
    if (sum < 0) return 0; 
    return (byte) sum; 
} 

Chiếc F # mã trông như thế này và phải mất ba lần miễn là chương trình C#:

let getPixelValue (buffer:nativeptr<byte>) (filterData:nativeptr<float>) filterLength filterSum : byte = 

    let rec accumulatePixel (acc:float) (buffer:nativeptr<byte>) (filter:nativeptr<float>) i = 
     if i > 0 then 
      let newAcc = acc + (float (NativePtr.read buffer) * (NativePtr.read filter)) 
      accumulatePixel newAcc (NativePtr.add buffer 1) (NativePtr.add filter 1) (i-1) 
     else 
      acc 

    let acc = (accumulatePixel 0.0 buffer filterData filterLength)/filterSum     

    match acc with 
     | _ when acc > 255.0 -> 255uy 
     | _ when acc < 0.0 -> 0uy 
     | _ -> byte acc 

Sử dụng các biến có thể thay đổi và một vòng lặp for trong F # không dẫn đến tốc độ giống như sử dụng đệ quy. Tất cả các dự án được cấu hình để chạy trong Chế độ phát hành với Tối ưu hóa mã được bật.

Làm cách nào để cải thiện hiệu suất của phiên bản F #?

EDIT:

Các nút cổ chai có vẻ là trong (NativePtr.get buffer offset). Nếu tôi thay thế mã này bằng một giá trị cố định và cũng thay thế mã tương ứng trong phiên bản C# bằng một giá trị cố định, tôi nhận được cùng tốc độ cho cả hai chương trình. Trong thực tế, trong C# tốc độ không thay đổi ở tất cả, nhưng trong F # nó làm cho một sự khác biệt rất lớn.

Hành vi này có thể bị thay đổi hoặc bắt nguồn từ sâu trong kiến ​​trúc của F # không?

EDIT 2:

tôi refactored mã một lần nữa để sử dụng cho-vòng. Tốc độ thực thi vẫn giữ nguyên:

let mutable acc <- 0.0 
let mutable f <- filterData 
let mutable b <- tBuffer 
for i in 1 .. filter.FilterLength do 
    acc <- acc + (float (NativePtr.read b)) * (NativePtr.read f) 
    f <- NativePtr.add f 1 
    b <- NativePtr.add b 1 

Nếu tôi so sánh các mã IL của một phiên bản sử dụng (NativePtr.read b) và một phiên bản đó là như nhau ngoại trừ việc nó sử dụng một giá trị cố định 111uy thay vì đọc nó từ con trỏ, Chỉ những dòng sau trong thay đổi mã IL:

111uy có IL-Code ldc.i4.s 0x6f (0,3 giây)

(NativePtr.read b) có dòng IL-Code ldloc.s b và ldobj uint8 (1,4 giây)

Để so sánh: C# thực hiện việc lọc trong 0,4 giây.

Thực tế là việc đọc bộ lọc không ảnh hưởng đến hiệu suất trong khi đọc từ bộ đệm hình ảnh không bằng cách nào đó gây nhầm lẫn. Trước khi tôi lọc một dòng hình ảnh, tôi sao chép dòng vào một bộ đệm có độ dài của một dòng. Đó là lý do tại sao các hoạt động đọc không được trải rộng trên toàn bộ hình ảnh nhưng nằm trong bộ đệm này, có kích thước khoảng 800 byte.

Answer 1

Nếu chúng ta nhìn vào mã IL thực tế của các vòng trong mà đi qua cả hai bộ đệm song song được tạo ra bởi biên dịch C# (phần liên quan):

L_0017: ldarg.0 
L_0018: ldc.i4.1 
L_0019: conv.i 
L_001a: add 
L_001b: starg.s buffer 
L_001d: ldarg.1 
L_001e: ldc.i4.8 
L_001f: conv.i 
L_0020: add 

và F # biên dịch:

L_0017: ldc.i4.1 
L_0018: conv.i 
L_0019: sizeof uint8 
L_001f: mul 
L_0020: add 
L_0021: ldarg.2 
L_0022: ldc.i4.1 
L_0023: conv.i 
L_0024: sizeof float64 
L_002a: mul 
L_002b: add 

chúng tôi sẽ nhận thấy rằng trong khi mã C# chỉ sử dụng toán tử add trong khi F # cần cả số mul và add. Nhưng rõ ràng trên mỗi bước chúng ta chỉ cần tăng con trỏ (bằng 'sizeof byte' và 'sizeof float' tương ứng), không tính địa chỉ (addrBase + (sizeof byte)) F # mul là không cần thiết (nó luôn nhân với 1).

Nguyên nhân cho rằng là C# định nghĩa ++ điều hành cho con trỏ trong khi F # cung cấp chỉ add : nativeptr<'T> -> int -> nativeptr<'T> điều hành:

[<NoDynamicInvocation>] 
let inline add (x : nativeptr<'a>) (n:int) : nativeptr<'a> = to_nativeint x + nativeint n * (# "sizeof !0" type('a) : nativeint #) |> of_nativeint 

Vì vậy, nó không phải là "bắt rễ sâu" trong F #, nó chỉ là module NativePtr thiếu inc và dec chức năng.

Btw, tôi nghi ngờ mẫu ở trên có thể được viết theo cách súc tích hơn nếu các đối số được chuyển thành mảng thay vì các con trỏ thô.

UPDATE:

Vì vậy, hiện đoạn mã sau chỉ có 1% tốc độ lên (có vẻ như để tạo ra rất giống với C# IL):

let getPixelValue (buffer:nativeptr<byte>) (filterData:nativeptr<float>) filterLength filterSum : byte = 

    let rec accumulatePixel (acc:float) (buffer:nativeptr<byte>) (filter:nativeptr<float>) i = 
     if i > 0 then 
      let newAcc = acc + (float (NativePtr.read buffer) * (NativePtr.read filter)) 
      accumulatePixel newAcc (NativePtr.ofNativeInt <| (NativePtr.toNativeInt buffer) + (nativeint 1)) (NativePtr.ofNativeInt <| (NativePtr.toNativeInt filter) + (nativeint 8)) (i-1) 
     else 
      acc 

    let acc = (accumulatePixel 0.0 buffer filterData filterLength)/filterSum     

    match acc with 
     | _ when acc > 255.0 -> 255uy 
     | _ when acc < 0.0 -> 0uy 
     | _ -> byte acc 

Một suy nghĩ: nó cũng có thể phụ thuộc vào số lượng các cuộc gọi đến getPixelValue thử nghiệm của bạn không (F # chia tách chức năng này thành hai phương pháp trong khi C# thực hiện nó trong một).

Có thể bạn đăng mã thử nghiệm của mình ở đây không?

Về mảng - Tôi mong đợi mã ít nhất ngắn gọn hơn (và không phải là unsafe).

CẬP NHẬT # 2:

Hình như các nút cổ chai thực tế đây là byte->float chuyển đổi.

C#:

L_0003: ldarg.1 
L_0004: ldind.u1 
L_0005: conv.r8 

F #:

L_000c: ldarg.1 
L_000d: ldobj uint8 
L_0012: conv.r.un 
L_0013: conv.r8 

Đối với một số lý do F # sử dụng đường dẫn sau: byte->float32->float64 trong khi C# chỉ byte->float64 làm. Không chắc chắn lý do tại sao, nhưng với sau hack phiên bản F # của tôi chạy với tốc độ tương tự như C# trên mẫu kiểm tra gradbot (BTW, cảm ơn gradbot để kiểm tra!):

let inline preadConvert (p : nativeptr<byte>) = (# "conv.r8" (# "ldobj !0" type (byte) p : byte #) : float #) 
let inline pinc (x : nativeptr<'a>) : nativeptr<'a> = NativePtr.toNativeInt x + (# "sizeof !0" type('a) : nativeint #) |> NativePtr.ofNativeInt 

let rec accumulatePixel_ed (acc, buffer, filter, i) = 
     if i > 0 then 
      accumulatePixel_ed 
       (acc + (preadConvert buffer) * (NativePtr.read filter), 
       (pinc buffer), 
       (pinc filter), 
       (i-1)) 
     else 
      acc 

Kết quả:

adrian 6374985677.162810 1408.870900 ms 
    gradbot 6374985677.162810 1218.908200 ms 
     C# 6374985677.162810 227.832800 ms 
C# Offset 6374985677.162810 224.921000 ms 
    mutable 6374985677.162810 1254.337300 ms 
    ed'ka 6374985677.162810 227.543100 ms 

CẬP NHẬT CUỐI Hóa ra là chúng ta có thể đạt được tốc độ như nhau ngay cả khi không có bất kỳ hacks:

let rec accumulatePixel_ed_last (acc, buffer, filter, i) = 
     if i > 0 then 
      accumulatePixel_ed_last 
       (acc + (float << int16 <| NativePtr.read buffer) * (NativePtr.read filter), 
       (NativePtr.add buffer 1), 
       (NativePtr.add filter 1), 
       (i-1)) 
     else 
      acc 

Tất cả chúng ta cần phải làm là chuyển đổi byte thành, nói int16 và sau đó nhập vào float. Bằng cách này, bạn sẽ tránh được hướng dẫn 'tốn kém' conv.r.un.

PS mã chuyển đổi có liên quan từ "prim-types.fs":

let inline float (x: ^a) = 
    (^a : (static member ToDouble : ^a -> float) (x)) 
    when ^a : float  = (# "" x : float #) 
    when ^a : float32 = (# "conv.r8" x : float #) 
    // [skipped] 
    when ^a : int16  = (# "conv.r8" x : float #) 
    // [skipped] 
    when ^a : byte  = (# "conv.r.un conv.r8" x : float #) 
    when ^a : decimal = (System.Convert.ToDouble((# "" x : decimal #))) 

Answer 2

So sánh này như thế nào? Nó có ít cuộc gọi đến NativePtr.

let getPixelValue (buffer:nativeptr<byte>) (filterData:nativeptr<float>) filterLength filterSum : byte = 
    let accumulatePixel (acc:float) (buffer:nativeptr<byte>) (filter:nativeptr<float>) length = 
     let rec accumulate acc offset = 
      if offset < length then 
       let newAcc = acc + (float (NativePtr.get buffer offset) * (NativePtr.get filter offset)) 
       accumulate newAcc (offset + 1) 
      else 
       acc 
     accumulate acc 0 

    let acc = (accumulatePixel 0.0 buffer filterData filterLength)/filterSum     

    match acc with 
     | _ when acc > 255.0 -> 255uy 
     | _ when acc < 0.0 -> 0uy 
     | _ -> byte acc 

Mã nguồn F # của NativePtr.

[<NoDynamicInvocation>] 
[<CompiledName("AddPointerInlined")>] 
let inline add (x : nativeptr<'T>) (n:int) : nativeptr<'T> = toNativeInt x + nativeint n * (# "sizeof !0" type('T) : nativeint #) |> ofNativeInt 

[<NoDynamicInvocation>] 
[<CompiledName("GetPointerInlined")>] 
let inline get (p : nativeptr<'T>) n = (# "ldobj !0" type ('T) (add p n) : 'T #) 

Answer 3

kết quả của tôi trên một thử nghiệm lớn hơn.

adrian 6374730426.098020 1561.102500 ms 
    gradbot 6374730426.098020 1842.768000 ms 
     C# 6374730426.098020 150.793500 ms 
C# Offset 6374730426.098020 150.318900 ms 
    mutable 6374730426.098020 1446.616700 ms 

F # mã kiểm tra

open Microsoft.FSharp.NativeInterop 
open System.Runtime.InteropServices 
open System.Diagnostics 

open AccumulatePixel 
#nowarn "9" 

let test size fn = 
    let bufferByte = Marshal.AllocHGlobal(size * 4) 
    let bufferFloat = Marshal.AllocHGlobal(size * 8) 

    let bi = NativePtr.ofNativeInt bufferByte 
    let bf = NativePtr.ofNativeInt bufferFloat 

    let random = System.Random() 

    for i in 1 .. size do 
     NativePtr.set bi i (byte <| random.Next() % 256) 
     NativePtr.set bf i (random.NextDouble()) 

    let duration (f, name) = 
     let stopWatch = Stopwatch.StartNew() 
     let time = f(0.0, bi, bf, size) 
     stopWatch.Stop() 
     printfn "%10s %f %f ms" name time stopWatch.Elapsed.TotalMilliseconds 

    List.iter duration fn 

    Marshal.FreeHGlobal bufferFloat 
    Marshal.FreeHGlobal bufferByte 

let rec accumulatePixel_adrian (acc, buffer, filter, i) = 
    if i > 0 then 
     let newAcc = acc + (float (NativePtr.read buffer) * (NativePtr.read filter)) 
     accumulatePixel_adrian (newAcc, (NativePtr.add buffer 1), (NativePtr.add filter 1), (i - 1)) 
    else 
     acc 

let accumulatePixel_gradbot (acc, buffer, filter, length) = 
    let rec accumulate acc offset = 
     if offset < length then 
      let newAcc = acc + (float (NativePtr.get buffer offset) * (NativePtr.get filter offset)) 
      accumulate newAcc (offset + 1) 
     else 
      acc 
    accumulate acc 0 

let accumulatePixel_mutable (acc, buffer, filter, length) = 
    let mutable acc = 0.0 
    let mutable f = filter 
    let mutable b = buffer 
    for i in 1 .. length do 
     acc <- acc + (float (NativePtr.read b)) * (NativePtr.read f) 
     f <- NativePtr.add f 1 
     b <- NativePtr.add b 1 
    acc 

[ 
    accumulatePixel_adrian, "adrian"; 
    accumulatePixel_gradbot, "gradbot"; 
    AccumulatePixel.getPixelValue, "C#"; 
    AccumulatePixel.getPixelValueOffset, "C# Offset"; 
    accumulatePixel_mutable, "mutable"; 
] 
|> test 100000000 

System.Console.ReadLine() |> ignore 

C# mã kiểm tra

namespace AccumulatePixel 
{ 
    public class AccumulatePixel 
    { 
     unsafe public static double getPixelValue(double sum, byte* buffer, double* filter, int filterLength) 
     { 
      for (int i = 0; i < filterLength; ++i) 
      { 
       sum += (*buffer) * (*filter); 
       ++buffer; 
       ++filter; 
      } 

      return sum; 
     } 

     unsafe public static double getPixelValueOffset(double sum, byte* buffer, double* filter, int filterLength) 
     { 
      for (int i = 0; i < filterLength; ++i) 
      { 
       sum += buffer[i] * filter[i]; 
      } 

      return sum; 
     } 
    } 
}