file_sorter.erl

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

              InEtc, T2, CF, Last) when I < I2 ->
    case CF(T, T2) of
        true -> % T =< T2
            case CF(T, Last) of
                true ->
                    ucmerge_files(L, F2, Fs, InEtc2, Ts2, I2, Ts, I, InEtc, T2,
                                  CF, Last);
                false ->
                    ucmerge_files([BT | L], F2, Fs, InEtc2, Ts2, I2, Ts, I,
                                  InEtc, T2, CF, T)
            end;
       false -> % T > T2
          [F3 | NFs] = cinsert([Ts0 | InEtc], Fs, CF),
          [[?OBJ(T3,_BT3) | _] = Ts3 | {I3,_} = InEtc3] = F3,
          ucmerge_files(L, F3, NFs, InEtc3, Ts3, I3, Ts2, I2, InEtc2, T3, CF, Last)
    end;
ucmerge_files(L, F2, Fs, InEtc2, Ts2, I2, [?OBJ(T, BT) | Ts] = Ts0, I,
              InEtc, T2, CF, Last) -> % when I2 < I
    case CF(T2, T) of
        true -> % T2 =< T
            [F3 | NFs] = cinsert([Ts0 | InEtc], Fs, CF),
            [[?OBJ(T3,_BT3) | _] = Ts3 | {I3,_} = InEtc3] = F3,
            ucmerge_files(L, F3, NFs, InEtc3, Ts3, I3, Ts2, I2, InEtc2, T3, 
                          CF, Last);
       false -> % T < T2
            case CF(T, Last) of
                true ->
                    ucmerge_files(L, F2, Fs, InEtc2, Ts2, I2, Ts, I, InEtc, T2,
                                  CF, Last);
                false ->
                    ucmerge_files([BT | L], F2, Fs, InEtc2, Ts2, I2, Ts, I,
                                  InEtc, T2, CF, T)
            end
    end;
ucmerge_files(L, F2, Fs, _InEtc2, _Ts2, _I2, [], _I, InEtc, _T2, _CF, Last) ->
    {InEtc, [F2 | Fs], L, {last, Last}}.

cmerge_files(L, F2, Fs, InEtc2, BT2, Ts2, I2, [?OBJ(T, BT) | Ts] = Ts0, I,
             InEtc, T2, CF) when I < I2 ->
    case CF(T, T2) of
       true -> % T =< T2
          cmerge_files([BT|L], F2, Fs, InEtc2, BT2, Ts2, I2, Ts, I, InEtc, T2, CF);
       false -> % T > T2
          L1 = [BT2 | L],
          [F3 | NFs] = cinsert([Ts0 | InEtc], Fs, CF),
          [[?OBJ(T3,BT3) | Ts3] | {I3,_} = InEtc3] = F3,
          cmerge_files(L1, F3, NFs, InEtc3, BT3, Ts3, I3, Ts2, I2, InEtc2, T3, CF)
    end;
cmerge_files(L, F2, Fs, InEtc2, BT2, Ts2, I2, [?OBJ(T, BT) | Ts] = Ts0, I,
             InEtc, T2, CF) -> % when I2 < I
    case CF(T2, T) of
       true -> % T2 =< T
          L1 = [BT2 | L],
          [F3 | NFs] = cinsert([Ts0 | InEtc], Fs, CF),
          [[?OBJ(T3,BT3) | Ts3] | {I3,_} = InEtc3] = F3,
          cmerge_files(L1, F3, NFs, InEtc3, BT3, Ts3, I3, Ts2, I2, InEtc2, T3, CF);
       false -> % T < T2
          cmerge_files([BT|L], F2, Fs, InEtc2, BT2, Ts2, I2, Ts, I, InEtc, T2, CF)
    end;
cmerge_files(L, F2, Fs, _InEtc2, _BT2, _Ts2, _I2, [], _I, InEtc, _T2, _CF) ->
    {InEtc, [F2 | Fs], L, {last, foo}}.

last_file(Ts, L, {last, T}, {ukmerge,_}, _W) ->
    kulast_file(Ts, T, L);
last_file(Ts, L, {last, {T,BT}}, {rukmerge,_}, _W) ->
    ruklast_file(Ts, T, BT, L);
last_file(Ts, L, {last, T}, {ucmerge,CF}, _W) ->
    uclast_file(Ts, T, CF, L);
last_file(Ts, L, {last, T}, _Kind, #w{unique = true}) ->
    ulast_file(Ts, T, L);
last_file(Ts, L, _Last, _Kind, _W) ->
    last_file(Ts, L).

ulast_file([?OBJ(T, _BT) | Ts], Last, L) when Last == T ->
    last_file(Ts, L);
ulast_file(Ts, _Last, L) ->
    last_file(Ts, L).
    
kulast_file([?OBJ(?SK(T, _I), _BT) | Ts], Last, L) when Last == T ->
    last_file(Ts, L);
kulast_file(Ts, _Last, L) ->
    last_file(Ts, L).

ruklast_file([?OBJ(?SK(T, _I), BT) | Ts], Last, _LastBT, L) when Last == T ->
    last_file(Ts, [BT | L]);
ruklast_file(Ts, _Last, LastBT, L) ->
    last_file(Ts, [LastBT | L]).

uclast_file([?OBJ(T, BT) | Ts], Last, CF, L) ->
    case CF(T, Last) of
        true ->
            last_file(Ts, L);
        false ->
            last_file(Ts, [BT | L])
    end.

last_file([?OBJ(_Ta, BTa), ?OBJ(_Tb, BTb) | Ts], L) ->
    last_file(Ts, [BTb, BTa | L]);
last_file([?OBJ(_T, BT) | Ts], L) ->
    last_file(Ts, [BT | L]);
last_file([], L) ->
    L.
    
%% OK for 16 files.
insert(A, [X1, X2, X3, X4 | Xs]) when A > X4 ->
    [X1, X2, X3, X4 | insert(A, Xs)];
insert(A, [X1, X2, X3 | T]) when A > X3 ->
    [X1, X2, X3, A | T];
insert(A, [X1, X2 | Xs]) when A > X2 ->
    [X1, X2, A | Xs];
insert(A, [X1 | T]) when A > X1 ->
    [X1, A | T];
insert(A, Xs) ->
    [A | Xs].

rinsert(A, [X1, X2, X3, X4 | Xs]) when A < X4 ->
    [X1, X2, X3, X4 | rinsert(A, Xs)];
rinsert(A, [X1, X2, X3 | T]) when A < X3 ->
    [X1, X2, X3, A | T];
rinsert(A, [X1, X2 | Xs]) when A < X2 ->
    [X1, X2, A | Xs];
rinsert(A, [X1 | T]) when A < X1 ->
    [X1, A | T];
rinsert(A, Xs) ->
    [A | Xs].

-define(CINSERT(F, A, T1, T2),
        case cfun(CF, F, A) of
            true -> [F, A | T2];
            false -> [A | T1]
        end).

cinsert(A, [F1 | [F2 | [F3 | [F4 | Fs]=T4]=T3]=T2]=T1, CF) ->
    case cfun(CF, F4, A) of
        true -> [F1, F2, F3, F4 | cinsert(A, Fs, CF)];
        false -> 
            case cfun(CF, F2, A) of
                true -> [F1, F2 | ?CINSERT(F3, A, T3, T4)];
                false -> ?CINSERT(F1, A, T1, T2)
            end
    end;
cinsert(A, [F1 | [F2 | Fs]=T2]=T1, CF) ->
    case cfun(CF, F2, A) of
        true -> [F1, F2 | cinsert(A, Fs, CF)];
        false -> ?CINSERT(F1, A, T1, T2)
    end;
cinsert(A, [F | Fs]=T, CF) ->
    ?CINSERT(F, A, T, Fs);
cinsert(A, _, _CF) ->
    [A].

%% Inlined.
cfun(CF, F1, F2) ->
    [[?OBJ(T1,_) | _] | {I1,_}] = F1,
    [[?OBJ(T2,_) | _] | {I2,_}] = F2,
    if 
        I1 < I2 ->
            CF(T1, T2);
        true -> % I2 < I1
            not CF(T2, T1)
    end.

binterm_objects([?OBJ(_T, [_Sz | BT]) | Ts], L) ->
    binterm_objects(Ts, [BT | L]);
binterm_objects([], L) ->
    L.

objects([[_Sz | BT] | Ts], L) ->
    objects(Ts, [BT | L]);
objects([], L) ->
    L.

binterms([?OBJ(_T1, BT1), ?OBJ(_T2, BT2) | Ts], L) ->
    binterms(Ts, [BT2, BT1 | L]);
binterms([?OBJ(_T, BT) | Ts], L) ->
    binterms(Ts, [BT | L]);
binterms([], L) ->
    L.

read_chunk(InEtc, Fs, L, LSz, Last, W) ->
    {I, IFun} = InEtc,
    case read_more(IFun, I, LSz, W) of
        {Ts, NLSz, NIFun, #w{order = ascending}=NW} ->
            NInEtc = {I, NIFun},
            NFs = insert([Ts | NInEtc], Fs),
            merge_files(NFs, L, NLSz, Last, NW);
        {Ts, NLSz, NIFun, #w{order = descending}=NW} ->
            NInEtc = {I, NIFun},
            NFs = rinsert([Ts | NInEtc], Fs),
            merge_files(NFs, L, NLSz, Last, NW);
        {Ts, NLSz, NIFun, NW} ->
            NInEtc = {I, NIFun},
            NFs = cinsert([Ts | NInEtc], Fs, NW#w.order),
            merge_files(NFs, L, NLSz, Last, NW);
        {eof, NW} ->
            merge_files(Fs, L, LSz, Last, NW)
    end.

%% -> {[{term() | binary()}], NewLSz, NewIFun, NewW} | eof | throw(Error)
read_more(IFun, I, LSz, W) ->
    case IFun({{merge, I}, [], LSz, W}) of
        {{_, [], NLSz}, NIFun} ->
            read_more(NIFun, I, NLSz, W);
        {{_, L, NLSz}, NInFun} ->
            NW = case lists:member(IFun, W#w.temp) of
                     true ->
                         %% temporary file
                         W#w{temp = [NInFun | lists:delete(IFun, W#w.temp)]};
                     false ->
                         %% input file
                         W
                 end,
            {lists:reverse(L), NLSz, NInFun, NW};
        eof ->
            %% already closed.
            NW = W#w{temp = lists:delete(IFun, W#w.temp)},
            {eof, NW}
    end.

read_fun(FileName, Owner, W) ->
    case file:open(FileName, [raw, binary, read, compressed]) of
        {ok, Fd} ->
            read_fun2(Fd, <<>>, 0, FileName, Owner);
        Error ->
            file_error(FileName, Error, W)
    end.

read_fun2(Fd, Bin, Size, FileName, Owner) ->
    fun(close) ->
            close_read_fun(Fd, FileName, Owner);
       ({I, L, LSz, W}) ->
            case read_objs(Fd, FileName, I, L, Bin, Size, LSz, W) of
                {{I1, L1, Bin1, Size1}, LSz1} ->
                    NIFun = read_fun2(Fd, Bin1, Size1, FileName, Owner),
                    {{I1, L1, LSz1}, NIFun};
                eof ->
                    close_read_fun(Fd, FileName, Owner),
                    eof
            end
    end.
       
close_read_fun(Fd, _FileName, user) ->
    file:close(Fd);
close_read_fun(Fd, FileName, fsort) ->
    file:close(Fd),
    file:delete(FileName).

read_objs(Fd, FileName, I, L, Bin0, Size0, LSz, W) ->
    Max = max(Size0, ?CHUNKSIZE),
    BSz0 = size(Bin0),
    Min = Size0 - BSz0 + W#w.hdlen, % Min > 0
    NoBytes = max(Min, Max),
    case read(Fd, FileName, NoBytes, W) of
        {ok, Bin} ->
            BSz = size(Bin),
            NLSz = LSz + BSz,
            case catch file_loop(L, I, Bin0, Bin, Size0, BSz0, BSz, Min, W)
                of
                {'EXIT', _R} ->
                    error({error, {bad_object, FileName}}, W);
                Reply ->
                    {Reply, NLSz}
            end;
        eof when size(Bin0) =:= 0 ->
            eof;
        eof ->
            error({error, {premature_eof, FileName}}, W)
    end.            

file_loop(L, I, _B1, B2, Sz, 0, _B2Sz, _Min, W) ->
    file_loop(L, I, B2, Sz, W);
file_loop(L, I, B1, B2, Sz, _B1Sz, B2Sz, Min, W) when B2Sz > Min ->
    {B3, B4} = split_binary(B2, Min),
    {I1, L1, <<>>, Sz1} = file_loop(L, I, list_to_binary([B1, B3]), Sz, W),
    file_loop(L1, I1, B4, Sz1, W);
file_loop(L, I, B1, B2, Sz, _B1Sz, _B2Sz, _Min, W) ->
    file_loop(L, I, list_to_binary([B1, B2]), Sz, W).

file_loop(L, I, B, Sz, W) ->
    #w{keypos = Kp, format = Format, hdlen = HdLen} = W,
    file_loop1(L, I, B, Sz, Kp, Format, HdLen).

file_loop1(L, I, HB, 0, Kp, F, HdLen) ->
    <<Size:HdLen/unit:8, B/binary>> = HB,
    file_loop2(L, I, B, Size, <<Size:HdLen/unit:8>>, Kp, F, HdLen);
file_loop1(L, I, B, Sz, Kp, F, HdLen) ->
    file_loop2(L, I, B, Sz, <<Sz:HdLen/unit:8>>, Kp, F, HdLen).

file_loop2(L, _I, B, Sz, SzB, 0, binary, HdLen) ->
    {NL, NB, NSz, NSzB} = file_binloop(L, Sz, SzB, B, HdLen),
    if 
        size(NB) =:= NSz ->
            <<Bin:NSz/binary>> = NB,
            {0, [?OBJ(Bin, [NSzB | Bin]) | NL], <<>>, 0};
        true ->
            {0, NL, NB, NSz}
    end;
file_loop2(L, _I, B, Sz, SzB, 0, Fun, HdLen) ->
    file_binterm_loop(L, Sz, SzB, B, Fun, HdLen);
file_loop2(L, {merge, I}, B, Sz, SzB, Kp, Fun, HdLen) -> % when Kp =/= 0
    merge_loop(Kp, I, L, Sz, SzB, B, Fun, HdLen);
file_loop2(L, I, B, Sz, SzB, Kp, Fun, HdLen) when is_integer(I) ->
    key_loop(Kp, I, L, Sz, SzB, B, Fun, HdLen).

file_binloop(L, Size, SizeB, B, HL) ->
    case B of
        <<Bin:Size/binary, NSizeB:HL/binary, R/binary>> ->
            <<NSize:HL/unit:8>> = NSizeB,
            file_binloop([?OBJ(Bin, [SizeB | Bin]) | L], NSize, NSizeB, R, HL);
        _ ->
            {L, B, Size, SizeB}
    end.

file_binterm_loop(L, Size, SizeB, B, Fun, HL) ->
    case B of
        <<BinTerm:Size/binary, NSizeB:HL/binary, R/binary>> ->
            <<NSize:HL/unit:8>> = NSizeB,
            BT = [SizeB | BinTerm],
            Term = Fun(BinTerm),
            file_binterm_loop([?OBJ(Term, BT) | L], NSize, NSizeB, R, Fun, HL);
        <<BinTerm:Size/binary>> ->
            Term = Fun(BinTerm),
            NL = [?OBJ(Term, [SizeB | BinTerm]) | L], 
            {0, NL, <<>>, 0};
        _ ->
            {0, L, B, Size}
    end.

key_loop(KeyPos, I, L, Size, SizeB, B, Fun, HL) ->
    case B of
        <<BinTerm:Size/binary, NSizeB:HL/binary, R/binary>> ->
            <<NSize:HL/unit:8>> = NSizeB,
            BT = [SizeB | BinTerm],
            UniqueKey = make_key(KeyPos, Fun(BinTerm)),
            E = ?OBJ(UniqueKey, BT),
            key_loop(KeyPos, I+1, [E | L], NSize, NSizeB, R, Fun, HL);
        <<BinTerm:Size/binary>> ->
            UniqueKey = make_key(KeyPos, Fun(BinTerm)),
            NL = [?OBJ(UniqueKey, [SizeB | BinTerm]) | L], 
            {I+1, NL, <<>>, 0};
        _ ->
            {I, L, B, Size}
    end.

merge_loop(KeyPos, I, L, Size, SizeB, B, Fun, HL) ->
    case B of
        <<BinTerm:Size/binary, NSizeB:HL/binary, R/binary>> ->
            <<NSize:HL/unit:8>> = NSizeB,
            BT = [SizeB | BinTerm],
            UniqueKey = make_stable_key(KeyPos, I, Fun(BinTerm)),
            E = ?OBJ(UniqueKey, BT),
            merge_loop(KeyPos, I, [E | L], NSize, NSizeB, R, Fun, HL);
        <<BinTerm:Size/binary>> ->
            UniqueKey = make_stable_key(KeyPos, I, Fun(BinTerm)),
            NL = [?OBJ(UniqueKey, [SizeB | BinTerm]) | L], 
            {{merge, I}, NL, <<>>, 0};
        _ ->
            {{merge, I}, L, B, Size}
    end.

fun_objs(Objs, L, LSz, NoBytes, I, W) ->
    #w{keypos = Keypos, format = Format, hdlen = HL} = W,
    case catch fun_loop(Objs, L, LSz, NoBytes, I, Keypos, Format, HL) of
        {'EXIT', _R} ->
            error({error, bad_object}, W);
        Reply ->
            Reply
    end.

fun_loop(Objs, L, LSz, RunSize, _I, 0, binary, HdLen) ->
    fun_binloop(Objs, L, LSz, RunSize, HdLen);
fun_loop(Objs, L, LSz, RunSize, _I, 0, Fun, HdLen) ->
    fun_loop(Objs, L, LSz, RunSize, Fun, HdLen);
fun_loop(Objs, L, LSz, RunSize, {merge, I}, Keypos, Fun, HdLen) ->
    fun_mergeloop(Objs, L, LSz, RunSize, I, Keypos, Fun, HdLen);