在 F# 中比较两个字节数组的最快方法

Question

我正在编写一个应用程序，它使用 Kinect 照片数据并在 F# 中比较两个帧。我正在使用 Rob Miles Learn the Kinect Api 第 74 页作为路标，但我没有使用指针并且性能受到影响。 Kinect 帧的字节为 1,228,800 字节。我是这样写比较的：

member this.differenceCount(currentImageBytes) =
    if  previousImageBytes |> Seq.length = 0 then
        previousImageBytes <- currentImageBytes
        0
    else
        let bytes = Seq.zip previousImageBytes currentImageBytes
        let differenceCount = bytes |> Seq.mapi(fun i e -> i, e)
                                    |> Seq.filter(fun (i,e) -> i % 4 <> 0 )
                                    |> Seq.map snd
                                    |> Seq.filter(fun (p,c) -> p <> c)
                                    |> Seq.length
        previousImageBytes <- currentImageBytes
        differenceCount

当我运行它时，屏幕会滞后，因为（我认为）处理数组花费的时间太长。此外，错误率接近50%。

1) 我处理问题的方法有误吗？ 2) 有没有办法优化我的代码以加快速度？

Answer 1

您通过元组进行的序列映射/过滤会导致大量装箱开销。以下示例避免了装箱并并行工作，这（在我的机器上）要快 50 倍。

let parallelRanges =
    let kinectSize = 1228800
    let subSize = kinectSize / Environment.ProcessorCount
    let iMax = kinectSize - 1
    let steps = [| -1 .. subSize .. iMax |]
    steps.[steps.Length - 1] <- iMax
    steps |> Seq.pairwise |> Seq.toArray |> Array.map (fun (x, y) -> x + 1, y)

let countDiffs (prevBytes:byte[]) (curBytes:_[]) =
    let count (fromI, toI) =
        let rec aux i acc =
            if i > toI then acc
            elif i % 4 <> 0 && prevBytes.[i] <> curBytes.[i] then 
                aux (i + 1) (acc + 1)
            else aux (i + 1) acc
        aux fromI 0

    parallelRanges
    |> Array.Parallel.map count
    |> Array.sum

Answer 2

如果没有一些测试数据来分析和比较，我会做类似的事情

let len = Array.length previousImageBytes
let mutable count = 0
for i in 0 .. 4 .. (len-1) do
    if previousImageBytes.[i] <> currentImageBytes.[i] then
        count <- count+1

它对数据进行了一次而不是 5 次，并避免了速度较慢的 seq 函数。

Answer 3

我已经在 F# 中为我一直从事的项目尝试了几种不同的字节数组比较实现。

一般来说，对于这个特定的问题，我发现“更惯用的 F#”==“更慢”。所以我最终得到了这个：

// this code is very non-F#-ish.  but it's much faster than the
// idiomatic version which preceded it.

let Compare (x:byte[]) (y:byte[]) =
    let xlen = x.Length
    let ylen = y.Length
    let len = if xlen<ylen then xlen else ylen
    let mutable i = 0
    let mutable result = 0
    while i<len do
        let c = (int (x.[i])) - int (y.[i])
        if c <> 0 then
            i <- len+1 // breaks out of the loop, and signals that result is valid
            result <- c
        else
            i <- i + 1
    if i>len then result else (xlen - ylen)

我违反了几项良好函数式编程的准则。我使用 if-then-else 而不是 match。我有变量。我伪造了一个 break 语句，对循环索引变量进行了粗俗的更改。

尽管如此，这最终比任何更惯用的方法都要快得多。即使是像带有尾递归的闭包这样简单的东西也比较慢。 （分析器表明闭包涉及堆上的内存分配。）

包含此代码的项目在 GitHub 上：

https://github.com/ericsink/LSM

fs/lsm.fs 的提交历史应该包含我其他失败尝试的记录。