从头开始合并列表列表

Question

我需要将已排序的列表合并为一个列表（列表的数量可能会有所不同）。作为 Erlang 的新手——我不知道漂亮的函数 lists:merge/1。所以我实现了自己的merge/1 函数。它的复杂性是 O(m*n) （m - 列表数，n - 列表中的平均元素数），我使用尾递归。这是我的功能：

-module( merge ).
-export( [ merge/1 ] ).

merge( ListOfLists ) ->
        merge( ListOfLists, [] ).

merge( [], Merged ) ->
        lists:reverse( Merged );
merge( ListOfLists, Merged ) ->
        [ [ Hfirst | Tfirst ] | ListOfLists_Tail ] = ListOfLists,
        % let's find list, which has minimal value of head
        % result would be a tuple { ListWithMinimalHead, Remainder_ListOfLists }
        { [ Hmin | Tmin ], ListOfLists_WithoutMinimalHead } =
        lists:foldl(
                fun( [ Hi | Ti ] = IncomingList, { [ Hmin | Tmin ], Acc } ) ->
                         case Hi < Hmin of
                                true ->
                                        % if incoming list has less value of head then swap it
                                        { [ Hi | Ti ], [ [ Hmin | Tmin ] | Acc ] };
                                false ->
                                        { [ Hmin | Tmin ], [ IncomingList | Acc ] }
                        end
                end,
                { [ Hfirst | Tfirst ], [] },
                ListOfLists_Tail ),
        % add minimal-valued head to accumulator, and go to next iteration
        case Tmin == [] of
                true ->
                        merge( ListOfLists_WithoutMinimalHead, [ Hmin | Merged ] );
                false ->
                        merge( [ Tmin | ListOfLists_WithoutMinimalHead ], [ Hmin | Merged ] )
        end.

但是，在我知道 lists:merge/1 之后 - 我决定测试我的解决方案的性能。

以下是一些结果：

1> c(merge).
{ok,merge}
2>
2> 
3> timer:tc( lists, merge, [ [ lists:seq(1,N) || N <- lists:seq(1,5) ]  ] ).   
{5,[1,1,1,1,1,2,2,2,2,3,3,3,4,4,5]}
3> 
3> timer:tc( merge, merge, [ [ lists:seq(1,N) || N <- lists:seq(1,5) ]  ] ). 
{564,[1,1,1,1,1,2,2,2,2,3,3,3,4,4,5]}
4> 
4> 
4> timer:tc( lists, merge, [ [ lists:seq(1,N) || N <- lists:seq(1,100) ]  ] ). 
{2559,
 [1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1|...]}
5>  
5> timer:tc( merge, merge, [ [ lists:seq(1,N) || N <- lists:seq(1,100) ]  ] ). 
{25186,
 [1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1|...]}
6> 
6> 
6> timer:tc( lists, merge, [ [ lists:seq(1,N) || N <- lists:seq(1,1000) ]  ] ). 
{153283,
 [1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1|...]}
7>  
7> timer:tc( merge, merge, [ [ lists:seq(1,N) || N <- lists:seq(1,1000) ]  ] ). 
{21676268,
 [1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1|...]}
8> 

0.153 秒给我留下了深刻的印象。 vs 21.676 秒。我的函数运行速度极慢。

我认为使用匿名函数会降低性能，但摆脱 fun 并没有帮助。

你能指出我在哪里犯了主要错误吗？或者为什么模块列表中的函数要快得多？

谢谢

Answer 1

不同之处在于算法的复杂性。 如果我没记错的话，您的算法是 O(m^2*n) 其中 n 是内部列表的长度，而 m是输入列表中的数字内部列表。 这是因为您的函数有效地遍历内部列表的整个列表以生成结果列表的一个元素。因此，对于您的测试示例，运行时间与 C1*N^3 成正比（在这种情况下，C1 是某个常数

但是，通常预排序列表的合并操作具有 O(n) 的复杂度，其中 n 是所有列表的总长度。因此，对于您的测试用例，复杂度应该是 O(n*m)，即它应该与 C2*N^2 成正比。

确实如您所见，当您的测试中的 N 增加 10 倍时，您的实现需要 860 倍以上的时间才能产生结果，而 'lists:merge/1' 只需要 53多倍的时间来合并输入。比率会因实际输入大小和“形状”而异，但总体趋势仍然是 N^3 与 N^2。

标准的 'lists:merge/1' 并不是那么简单：https://github.com/erlang/otp/blob/maint/lib/stdlib/src/lists.erl#L1441 ('merge/1' 只是调用 'mergel/1') 但实际上即使是一个简单的、未优化的、不是尾递归的“只是将头部列表与合并的尾部合并”的性能比您的实现要好得多：

merge2([]) ->
    [];
merge2([Ls|Lss]) ->
    merge2(Ls,merge2(Lss), []).

merge2([], Ls, Acc) ->
    lists:reverse(Acc) ++ Ls;
merge2(Ls, [], Acc) ->
    lists:reverse(Acc) ++ Ls;
merge2([H1|Ls1], [H2|_] = Ls2, Acc) when H1 =< H2 ->
    merge2(Ls1, Ls2, [H1|Acc]);
merge2(Ls1, [H2|Ls2], Acc) ->
    merge2(Ls1, Ls2, [H2|Acc]).

再一次，就像实践中经常发生的情况一样：任何优化的第一步都是查看算法。

UPD： 好吧，我的例子实际上也是 O(m^2*n) - 就复杂性而言，并不比你的好。我们在这里可能需要的是“分而治之”的方法，它应该将复杂性提高到 O(m*n*ln(n))

UPD2： 对上一次更新的更正和澄清： “分而治之”是指以下算法：

假设我们的输入列表中有 m 个排序列表，每个列表由 n 个元素组成。那么：

1. 将输入列表拆分为两个子列表，每个子列表中有 m/2 个列表
2. 以递归方式将此算法应用于它们中的每一个。
3. 使用标准 2 列表合并合并两个结果排序列表。

这个算法的渐近复杂度实际上是O(n*m*ln(m))，因为： 1.拆分操作在每个拆分级别上都是O(m)，因此可以忽略。 2. 合并操作在每一层都是O(m*n)：在上层（第一个拆分）我们需要合并两个列表，每个n*m/2 O(n*m) 的元素；在下一个级别（第二次拆分），我们需要进行两次独立的合并，每次合并两个 n*m/4 元素列表，这也是 O(m*n) 和以此类推，直到 m=2 或 m=1 3. 层数明显是log2(m) 所以得到的复杂度是O(n*m*ln(m))

事实上，这个算法可以被认为只是merge sort 的一种变体，它稍早“停止”分裂（因此它有 ln(m) 而不是 ln(m*n)) 并且当 n=1 时它变成了成熟的合并排序（而你的第一个算法实际上变成了selection sort）