file_sorter.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,498 行 · 第 1/4 页

    check_run(IFun, F, FNs, W, L, I, {last, Last}).

ucheck_objs(IFun, F, FNs, W, L, I, LT, [?OBJ(T,_BT) | Os]) when T > LT ->
    ucheck_objs(IFun, F, FNs, W, L, I+1, T, Os);
ucheck_objs(IFun, F, FNs, W, L, I, _LT, [?OBJ(_T,BT) | _]) ->
    culprit_found(IFun, F, FNs, W, L, I, BT);
ucheck_objs(IFun, F, FNs, W, L, I, LT, []) ->
    check_run(IFun, F, FNs, W, L, I, {last, LT}).

rucheck_objs(IFun, F, FNs, W, L, I, LT, [?OBJ(T,_BT) | Os]) when T < LT ->
    rucheck_objs(IFun, F, FNs, W, L, I+1, T, Os);
rucheck_objs(IFun, F, FNs, W, L, I, _LT, [?OBJ(_T,BT) | _]) ->
    culprit_found(IFun, F, FNs, W, L, I, BT);
rucheck_objs(IFun, F, FNs, W, L, I, LT, []) ->
    check_run(IFun, F, FNs, W, L, I, {last, LT}).

ccheck_objs(IFun, F, FNs, W, L, I, LT, [?OBJ(T,BT) | Os], CF) ->
    case CF(LT, T) of
        true -> % LT =< T
            ccheck_objs(IFun, F, FNs, W, L, I+1, T, Os, CF);
        false -> % LT > T
            culprit_found(IFun, F, FNs, W, L, I, BT)
    end;
ccheck_objs(IFun, F, FNs, W, L, I, LT, [], _CF) ->
    check_run(IFun, F, FNs, W, L, I, {last, LT}).

uccheck_objs(IFun, F, FNs, W, L, I, LT, [?OBJ(T,BT) | Os], CF) ->
    case CF(LT, T) of
        true -> % LT =< T
            case CF(T, LT) of
                true -> % T equal to LT
                    culprit_found(IFun, F, FNs, W, L, I, BT);
                false -> % LT < T
                    uccheck_objs(IFun, F, FNs, W, L, I+1, T, Os, CF)
            end;
        false -> % LT > T
            culprit_found(IFun, F, FNs, W, L, I, BT)
    end;
uccheck_objs(IFun, F, FNs, W, L, I, LT, [], _CF) ->
    check_run(IFun, F, FNs, W, L, I, {last, LT}).

culprit_found(IFun, F, FNs, W, L, I, [_Size | BT]) ->
    IFun(close),
    check_files(FNs, W, [{F,I,binary_to_term(BT)} | L]).

files(_I, L, _LSz, #w{seq = 1}=W, []) ->
    %% No temporary files created, everything in L.
    case W#w.out of
        Fun when is_function(Fun) ->
            SL = internal_sort(L, W),
            W1 = outfun(binterm_objects(SL, []), W),
            NW = close_input(W1),
            outfun(close, NW);
        Out ->
            write_run(L, W, Out),
            ok
    end;
files(_I, L, _LSz, W, []) ->
    W1 = write_run(L, W),
    last_merge(lists:append(W1#w.runs), W1);
files(I, L, LSz, W, Fun) when is_function(Fun) ->
    NW = W#w{in = Fun},
    fun_run(I, L, LSz, NW, []);
files(I, L, LSz, W, [FileName | FileNames]) ->
    InFun = read_fun(FileName, user, W),
    NW = W#w{in = InFun},
    file_run(InFun, FileNames, I, L, LSz, NW).

file_run(InFun, FileNames, I, L, LSz, W) when LSz < W#w.runsize ->
    case InFun({I, L, LSz, W}) of
        {{I1, L1, LSz1}, InFun1} ->
            NW = W#w{in = InFun1},
            file_run(InFun1, FileNames, I1, L1, LSz1, NW);
        eof ->
            NW = W#w{in = undefined},
            files(I, L, LSz, NW, FileNames)
    end;
file_run(InFun, FileNames, I, L, _LSz, W) ->
    NW = write_run(L, W),
    file_run(InFun, FileNames, I, [], 0, NW).

fun_run(I, L, LSz, W, []) ->
    case infun(W) of
        {end_of_input, NW} ->
            files(I, L, LSz, NW, []);
        {cont, NW, Objs} ->
            fun_run(I, L, LSz, NW, Objs)
    end;
fun_run(I, L, LSz, W, Objs) when LSz < W#w.runsize ->
    {NI, NObjs, NL, NLSz} = fun_objs(Objs, L, LSz, W#w.runsize, I, W),
    fun_run(NI, NL, NLSz, W, NObjs);
fun_run(I, L, _LSz, W, Objs) ->
    NW = write_run(L, W),
    fun_run(I, [], 0, NW, Objs).

write_run([], W) ->
    W;
write_run(L, W) ->
    {W1, Temp} = next_temp(W),
    NW = write_run(L, W1, Temp),
    [R | Rs] = NW#w.runs,
    merge_runs([[Temp | R] | Rs], [], NW).

write_run(L, W, FileName) ->
    SL = internal_sort(L, W),
    BTs = binterms(SL, []), 
    {Fd, W1} = open_file(FileName, W),
    write(Fd, FileName, BTs, W1),
    close_file(Fd, W1).

%% Returns a list in reversed order.
internal_sort([]=L, _W) ->
    L;
internal_sort(L, #w{order = CFun, unique = Unique}) when is_function(CFun) ->
    Fun = fun(?OBJ(T1, _), ?OBJ(T2, _)) -> CFun(T1, T2) end,
    RL = lists:reverse(L),
    lists:reverse(if
                      Unique -> 
                          lists:usort(Fun, RL);
                      true -> 
                          lists:sort(Fun, RL)
                  end);
internal_sort(L, #w{unique = true, keypos = 0}=W) ->
    rev(lists:usort(L), W);
internal_sort(L, #w{unique = false, keypos = 0}=W) ->
    rev(lists:sort(L), W);
internal_sort(L, #w{unique = true}=W) ->
    rev(lists:ukeysort(1, lists:reverse(L)), W);
internal_sort(L, #w{unique = false}=W) ->
    rev(lists:keysort(1, lists:reverse(L)), W).

rev(L, #w{order = ascending}) ->
    lists:reverse(L);
rev(L, #w{order = descending}) ->
    L.

last_merge(R, W) when length(R) =< W#w.no_files ->
    case W#w.out of
        Fun when is_function(Fun) ->
            {Fs, W1} = init_merge(lists:reverse(R), 1, [], W),
            ?DEBUG("merging ~p~n", [lists:reverse(R)]),
            W2 = merge_files(Fs, [], 0, nolast, W1),
            NW = close_input(W2),
            outfun(close, NW);
        Out ->
            merge_files(R, W, Out),
            ok
    end;
last_merge(R, W) ->
    L = lists:sublist(R, W#w.no_files),
    {M, NW} = merge_files(L, W),
    last_merge([M | lists:nthtail(W#w.no_files, R)], NW).

merge_runs([R | Rs], NRs0, W) when length(R) < W#w.no_files ->
    NRs = lists:reverse(NRs0) ++ [R | Rs],
    W#w{runs = NRs};    
merge_runs([R], NRs0, W) ->
    {M, NW} = merge_files(R, W),
    NRs = [[] | lists:reverse([[M] | NRs0])],
    NW#w{runs = NRs};    
merge_runs([R, R1 | Rs], NRs0, W) ->
    {M, NW} = merge_files(R, W),
    merge_runs([[M | R1] | Rs], [[] | NRs0], NW).

merge_files(R, W) ->
    {W1, Temp} = next_temp(W),
    ?DEBUG("merging ~p~nto ~p~n", [lists:reverse(R), Temp]),
    {Temp, merge_files(R, W1, Temp)}.
    
merge_files(R, W, FileName) ->
    {Fs, W1} = init_merge(lists:reverse(R), 1, [], W),
    {Fd, W2} = open_file(FileName, W1),
    W3 = W2#w{wfd = {Fd, FileName}},
    W4 = merge_files(Fs, [], 0, nolast, W3),
    NW = W4#w{wfd = undefined},
    close_file(Fd, NW).

%% A file number, I, is used for making the merge phase stable.
init_merge([FN | FNs], I, Fs, W) ->
    IFun = case FN of
               _ when is_function(FN) ->
                   %% When and only when merge/2,3 or keymerge/3,4 was called.
                   FN;
               _ ->
                   read_fun(FN, fsort, W)
           end,
    W1 = W#w{temp = [IFun | lists:delete(FN, W#w.temp)]},
    case read_more(IFun, I, 0, W1) of
        {Ts, _LSz, NIFun, NW} ->
            InEtc = {I, NIFun},
            init_merge(FNs, I+1, [[Ts | InEtc] | Fs], NW);
        {eof, NW} -> % can only happen when merging files
            init_merge(FNs, I+1, Fs, NW)
    end;
init_merge([], _I, Fs0, #w{order = ascending}=W) ->
    {lists:sort(Fs0), W};
init_merge([], _I, Fs0, #w{order = descending}=W) ->
    {lists:reverse(lists:sort(Fs0)), W};
init_merge([], _I, Fs0, #w{order = Order}=W) when is_function(Order) ->
    {lists:sort(cfun_files(W#w.order), lists:reverse(Fs0)), W}.

cfun_files(CFun) ->
    fun(F1, F2) ->
            [[?OBJ(T1,_) | _] | _] = F1,
            [[?OBJ(T2,_) | _] | _] = F2,
            CFun(T1, T2)
    end.

%% The argument Last is used when unique = true. It is the last kept
%% element.
%% LSz is not the sum of the sizes of objects in L. Instead it is
%% the number of bytes read. After init_merge it is set to 0, which
%% means that the first chunk written may be quite large (it may take
%% a while before buffers are exhausted).
merge_files([F1, F2 | Fs], L0, LSz, Last0, W) when LSz < ?MERGESIZE ->
    [Ts0 | InEtc] = F1,
    Kind = merge_kind(W),
    {Last, L, Ts} = case {Last0, Kind} of
                        {{last, Lst}, Kind} -> 
                            {Lst, L0, Ts0};
                        {nolast, {ukmerge, _Kp}} -> 
                            [?OBJ(?SK(T, _I), BT) | Ts1] = Ts0,
                            {T, [BT], Ts1};
                        {nolast, {rukmerge, _Kp}} -> 
                            [?OBJ(?SK(T, _I), BT) | Ts1] = Ts0,
                            {{T, BT}, [], Ts1};
                        {nolast, _} ->
                            [?OBJ(T, BT) | Ts1] = Ts0,
                            {T, [BT], Ts1}
                    end,
    [[?OBJ(T2, BT2) | Ts2T] = Ts2 | InEtc2] = F2,
    {NInEtc, NFs, NL, NLast} = 
       case Kind of 
           umerge ->
               umerge_files(L, F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Last);
           {ukmerge, Kp} ->
               ukmerge_files(L, F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Kp, Last);
           merge ->
               merge_files(L, F2, Fs, InEtc2, BT2, Ts2T, Ts, InEtc, T2);
           rumerge ->
               rumerge_files(L, F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Last);
           {rukmerge, Kp} ->
               {Lt, LtBT} = Last,
               rukmerge_files(L, F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Kp, 
                              Lt, LtBT);
           rmerge ->
               rmerge_files(L, F2, Fs, InEtc2, BT2, Ts2T, Ts, InEtc, T2);
           {ucmerge, CF} ->
               {I2, _} = InEtc2,
               {I, _} = InEtc,
               ucmerge_files(L, F2, Fs, InEtc2, Ts2, I2, Ts, I, InEtc, T2, CF,
                             Last);
           {cmerge, CF} ->
               {I2, _} = InEtc2,
               {I, _} = InEtc,
               cmerge_files(L, F2, Fs, InEtc2, BT2, Ts2T, I2, Ts, I, InEtc, T2,
                            CF)
       end,
    read_chunk(NInEtc, NFs, NL, LSz, NLast, W);
merge_files([F1], L, LSz, Last, W) when LSz < ?MERGESIZE ->
    [Ts | InEtc] = F1,
    NL = last_file(Ts, L, Last, merge_kind(W), W),
    read_chunk(InEtc, [], NL, LSz, nolast, W);
merge_files([], [], 0, nolast, W) ->
    %% When merging files, ensure that the output fun (if there is
    %% one) is called at least once before closing.
    merge_write(W, []);
merge_files([], L, _LSz, Last, W) ->
    Last = nolast,
    merge_write(W, L);
merge_files(Fs, L, _LSz, Last, W) ->
    NW = merge_write(W, L),
    merge_files(Fs, [], 0, Last, NW).

merge_kind(#w{order = ascending, unique = true, keypos = 0}) ->
    umerge;
merge_kind(#w{order = ascending, unique = true, keypos = Kp}) ->
    {ukmerge, Kp};
merge_kind(#w{order = ascending, unique = false}) ->
    merge;
merge_kind(#w{order = descending, unique = true, keypos = 0}) ->
    rumerge;
merge_kind(#w{order = descending, unique = true, keypos = Kp}) ->
    {rukmerge, Kp};
merge_kind(#w{order = descending, unique = false}) ->
    rmerge;
merge_kind(#w{order = CF, unique = true}) ->
    {ucmerge, CF};
merge_kind(#w{order = CF, unique = false}) ->
    {cmerge, CF}.

merge_write(W, L) ->
     case {W#w.wfd, W#w.out} of
         {undefined, Fun} when is_function(Fun) ->
             outfun(objects(L, []), W);
         {{Fd, FileName}, _} ->
             write(Fd, FileName, lists:reverse(L), W),
             W
     end.

umerge_files(L, F2, Fs, InEtc2, Ts2, [?OBJ(T, _BT) | Ts], InEtc, T2, Last) 
            when T == Last ->
    umerge_files(L, F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Last);
umerge_files(L, F2, Fs, InEtc2, Ts2, [?OBJ(T, BT) | Ts], InEtc, T2, _Last) 
            when T =< T2 ->
    umerge_files([BT | L], F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, T);
umerge_files(L, F2, Fs, _InEtc2, _Ts2, [], InEtc, _T2, Last) ->
    {InEtc, [F2 | Fs], L, {last, Last}};
umerge_files(L, _F2, Fs, InEtc2, Ts2, Ts, InEtc, _T2, Last) ->
    [F3 | NFs] = insert([Ts | InEtc], Fs),
    [[?OBJ(T3,_BT3) | _] = Ts3 | InEtc3] = F3,
    umerge_files(L, F3, NFs, InEtc3, Ts3, Ts2, InEtc2, T3, Last).

rumerge_files(L, F2, Fs, InEtc2, Ts2, [?OBJ(T, _BT) | Ts], InEtc, T2, Last) 
            when T == Last ->
    rumerge_files(L, F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Last);
rumerge_files(L, F2, Fs, InEtc2, Ts2, [?OBJ(T, BT) | Ts], InEtc, T2, _Last) 
            when T >= T2 ->
    rumerge_files([BT | L], F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, T);
rumerge_files(L, F2, Fs, _InEtc2, _Ts2, [], InEtc, _T2, Last) ->
    {InEtc, [F2 | Fs], L, {last, Last}};
rumerge_files(L, _F2, Fs, InEtc2, Ts2, Ts, InEtc, _T2, Last) ->
    [F3 | NFs] = rinsert([Ts | InEtc], Fs),
    [[?OBJ(T3,_BT3) | _] = Ts3 | InEtc3] = F3,
    rumerge_files(L, F3, NFs, InEtc3, Ts3, Ts2, InEtc2, T3, Last).

merge_files(L, F2, Fs, InEtc2, BT2, Ts2, [?OBJ(T, BT) | Ts], InEtc, T2) 
            when T =< T2 ->
    merge_files([BT | L], F2, Fs, InEtc2, BT2, Ts2, Ts, InEtc, T2);
merge_files(L, F2, Fs, _InEtc2, _BT2, _Ts2, [], InEtc, _T2) ->
    {InEtc, [F2 | Fs], L, {last, foo}};
merge_files(L, _F2, Fs, InEtc2, BT2, Ts2, Ts, InEtc, _T2) ->
    L1 = [BT2 | L],
    [F3 | NFs] = insert([Ts | InEtc], Fs),
    [[?OBJ(T3,BT3) | Ts3] | InEtc3] = F3,
    merge_files(L1, F3, NFs, InEtc3, BT3, Ts3, Ts2, InEtc2, T3).

rmerge_files(L, F2, Fs, InEtc2, BT2, Ts2, [?OBJ(T, BT) | Ts], InEtc, T2) 
            when T >= T2 ->
    rmerge_files([BT | L], F2, Fs, InEtc2, BT2, Ts2, Ts, InEtc, T2);
rmerge_files(L, F2, Fs, _InEtc2, _BT2, _Ts2, [], InEtc, _T2) ->
    {InEtc, [F2 | Fs], L, {last, foo}};
rmerge_files(L, _F2, Fs, InEtc2, BT2, Ts2, Ts, InEtc, _T2) ->
    L1 = [BT2 | L],
    [F3 | NFs] = rinsert([Ts | InEtc], Fs),
    [[?OBJ(T3,BT3) | Ts3] | InEtc3] = F3,
    rmerge_files(L1, F3, NFs, InEtc3, BT3, Ts3, Ts2, InEtc2, T3).

ukmerge_files(L, F2, Fs, InEtc2, Ts2, [?OBJ(?SK(T, _I),_BT) | Ts], InEtc, 
              T2, Kp, Last) when T == Last ->
    ukmerge_files(L, F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Kp, Last);
ukmerge_files(L, F2, Fs, InEtc2, Ts2, [?OBJ(?SK(T0,_I)=T,BT) | Ts], InEtc, 
              T2, Kp, _Last) when T =< T2 ->
    ukmerge_files([BT | L], F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Kp, T0);
ukmerge_files(L, F2, Fs, _InEtc2, _Ts2, [], InEtc, _T2, _Kp, Last) ->
    {InEtc, [F2 | Fs], L, {last, Last}};
ukmerge_files(L, _F2, Fs, InEtc2, Ts2, Ts, InEtc, _T2, Kp, Last) ->
    [F3 | NFs] = insert([Ts | InEtc], Fs),
    [[?OBJ(T3,_BT3) | _] = Ts3 | InEtc3] = F3,
    ukmerge_files(L, F3, NFs, InEtc3, Ts3, Ts2, InEtc2, T3, Kp, Last).

rukmerge_files(L, F2, Fs, InEtc2, Ts2, [?OBJ(?SK(T, _I), BT) | Ts], InEtc, 
               T2, Kp, Last, _LastBT) when T == Last ->
    rukmerge_files(L, F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Kp, T, BT);
rukmerge_files(L, F2, Fs, InEtc2, Ts2, [?OBJ(?SK(T0, _I)=T, BT) | Ts], InEtc, 
               T2, Kp, _Last, LastBT) when T >= T2 ->
    rukmerge_files([LastBT|L], F2, Fs, InEtc2, Ts2, Ts, InEtc, T2, Kp, T0,BT);
rukmerge_files(L, F2, Fs, _InEtc2, _Ts2, [], InEtc, _T2, _Kp, Last, LastBT) ->
    {InEtc, [F2 | Fs], L, {last, {Last, LastBT}}};
rukmerge_files(L, _F2, Fs, InEtc2, Ts2, Ts, InEtc, _T2, Kp, Last, LastBT) ->
    [F3 | NFs] = rinsert([Ts | InEtc], Fs),
    [[?OBJ(T3,_BT3) | _] = Ts3 | InEtc3] = F3,
    rukmerge_files(L, F3, NFs, InEtc3, Ts3, Ts2, InEtc2, T3, Kp, Last,LastBT).

ucmerge_files(L, F2, Fs, InEtc2, Ts2, I2, [?OBJ(T, BT) | Ts] = Ts0, I,