F#:从seq中删除重复项很慢

Question

我正在尝试编写一个函数,根据给定的相等函数确定连续重复,seq<'a> 但是有一个扭曲:我需要从一系列重复项中删除最后一个副本,使其成为结果序列.例如,如果我有一个序列[("a", 1); ("b", 2); ("b", 3); ("b", 4); ("c", 5)],并且我fun ((x1, y1),(x2, y2)) -> x1=x2用来检查相等性,我想看到的结果是[("a", 1); ("b", 4); ("c", 5)].这个函数的关键在于我有数据点进入,有时数据点合法地具有相同的时间戳,但我只关心最新的一个,所以我想抛弃前面的那些具有相同的时间戳.我实现的功能如下:

let rec dedupeTakingLast equalityFn prev s = seq {
     match ( Seq.isEmpty s ) with
     | true -> match prev with 
                 | None -> yield! s
                 | Some value -> yield value
     | false ->
         match prev with 
         | None -> yield! dedupeTakingLast equalityFn (Some (Seq.head s)) (Seq.tail s) 
         | Some prevValue -> 
             if not (equalityFn(prevValue, (Seq.head s))) then 
                 yield prevValue
             yield! dedupeTakingLast equalityFn (Some (Seq.head s)) (Seq.tail s)
}

在实际工作方面,它有效:

> [("a", 1); ("b", 2); ("b", 3); ("b", 4); ("c", 5)] 
  |> dedupeTakingLast (fun ((x1, y1),(x2, y2)) -> x1=x2) None 
  |> List.ofSeq;;
val it : (string * int) list = [("a", 1); ("b", 4); ("c", 5)]

但是,就性能而言,这是一场灾难:

> #time
List.init 1000 (fun _ -> 1) 
|> dedupeTakingLast (fun (x,y) -> x = y) None 
|> List.ofSeq
#time;;    
--> Timing now on    
Real: 00:00:09.958, CPU: 00:00:10.046, GC gen0: 54, gen1: 1, gen2: 1
val it : int list = [1]    
--> Timing now off

很明显,我在这里做了一件非常愚蠢的事,但我看不出它是什么.性能受到何种影响？我意识到有更好的实现,但我特别感兴趣的是理解为什么这个实现如此缓慢.

编辑:仅供参考,设法在Seq.仅依赖函数的功能样式中实现了一个不错的实现.性能还可以,大约是使用迭代器的@gradbot执行命令式实现的时间的1.6倍.似乎问题的根源是在我最初的努力中使用Seq.head和Seq.tail递归调用.

let dedupeTakingLastSeq equalityFn s = 
    s 
    |> Seq.map Some
    |> fun x -> Seq.append x [None]
    |> Seq.pairwise
    |> Seq.map (fun (x,y) -> 
            match (x,y) with 
            | (Some a, Some b) -> (if (equalityFn a b) then None else Some a)  
            | (_,None) -> x
            | _ -> None )
    |> Seq.choose id

Answer 1

问题在于你如何使用序列.所有这些产量,头部和尾部都在旋转迭代器分支的迭代器网络,当你在调用时最终实现它时List.ofSeq,你将比你应该更多地迭代你的输入序列.

其中每一个Seq.heads都不是简单地取一个序列的第一个元素 - 它取一个序列尾部序列尾部序列的尾部的第一个元素,依此类推.

检查一下 - 它将计算元素构造函数被调用的次数:

let count = ref 0

Seq.init 1000 (fun i -> count := !count + 1; 1) 
|> dedupeTakingLast (fun (x,y) -> x = y) None 
|> List.ofSeq

顺便说一句,只需将所有内容切换Seqs为Lists即时即可.

Answer 2

性能问题来自对Seq.tail的嵌套调用.这是Seq.tail的源代码

[<CompiledName("Tail")>]
let tail (source: seq<'T>) =
    checkNonNull "source" source
    seq { use e = source.GetEnumerator() 
          if not (e.MoveNext()) then 
              invalidArg "source" (SR.GetString(SR.notEnoughElements))
          while e.MoveNext() do
              yield e.Current }

如果调用Seq.tail(Seq.tail(Seq.tail(...)))编译器,则无法优化由其创建的枚举数GetEnumerator().后续返回的元素必须遍历每个嵌套序列和枚举器.这导致每个返回的元素必须在所有先前创建的序列中冒泡,并且所有这些序列也必须递增它们的内部状态.最终结果是运行时间为O(n ^ 2)而不是线性O(n).

不幸的是,目前无法用F#中的功能样式表示这一点.您可以使用列表(x :: xs)但不能使用序列.在语言获得对序列的更好的本机支持之前,不要在递归函数中使用Seq.tail.

使用单个枚举器将解决性能问题.

let RemoveDuplicatesKeepLast equals (items:seq<_>) =
    seq {
        use e = items.GetEnumerator()

        if e.MoveNext() then
            let mutable previous = e.Current

            while e.MoveNext() do
                if not (previous |> equals e.Current) then 
                    yield previous
                previous <- e.Current

            yield previous
    }

let test = [("a", 1); ("b", 2); ("b", 3); ("b", 4); ("c", 5)]
let FirstEqual a b = fst a = fst b

RemoveDuplicatesKeepLast FirstEqual test
|> printf "%A"

// output
// seq [("a", 1); ("b", 4); ("c", 5)]

这个答案的第一个版本有上面代码的递归版本,没有变异.