-module(ttt).
-include_lib("eunit/include/eunit.hrl").

-export([do/2,do_spl/2,do_work/4]).

-record(state,
                {
                    data=[],
                    x_num = 0,
                    childs = dict:new(),
                    count_in_batch= 10,
                    responces = gb_trees:empty(),
                    parent_pid = 0,
                    spawn_func= fun(A,B,C,D,E) -> spawn_chld(A,B,C,D,E) end
                }).

do(L, X)->
    do(L,X,[],0,length(L),[]).
%   Lst = [ P || P <- L, P<X],
%   Llen = length(Lst),
%   FirstPartLen = Llen div 2,
%   lists:append(lists:sublist(Lst, FirstPartLen),lists:reverse(lists:sublist(Lst,FirstPartLen+1,Llen-FirstPartLen))).


do([], _, Ret, _, _, Ret2)  ->
    lists:append(Ret,Ret2);
do([H|T], X, Ret, Cntr, Len, Ret2) when X>H, Cntr=:=(Len div 2-1) ->
    do(T,X,lists:reverse([H|Ret]),Cntr+1, Len, Ret2);
do([H|T], X, Ret, Cntr, Len, Ret2) when X>H, Cntr<(Len div 2-1) ->
    do(T, X, [H|Ret], Cntr+1, Len, Ret2);
do([H|T], X, Ret, Cntr, Len, Ret2) when X>H, Cntr>(Len div 2-1) ->
    do(T, X, Ret, Cntr+1, Len,[H|Ret2]);
do([_|T],X,Ret,Cntr,Len,Ret2)->
    do(T,X,Ret,Cntr,Len,Ret2).

%Removes all elements from Array for Count elements
%which are greater than X

% (List, X, Count, RetArray)
do_work(Arr, X, Count, RetFunc)->
    RetFunc(do_p(Arr,X,Count,[])).

do_p(_,_,0,Ret) ->
    lists:reverse(Ret);
do_p([],_,_,Ret) ->
    do_p([],0,0,Ret); %just call for done work
do_p([H|T], X,  Count, Ret) when X>H ->
    do_p(T, X, Count-1, [H | Ret]);
do_p([_|T], X, Count, Ret) ->
    do_p(T, X, Count-1, Ret).

do_spl(L,X)->
    MyPid=  self(),
    Pid = spawn(fun()-> process_flag(trap_exit, true), loop(#state{parent_pid=MyPid}) end),
    Pid!{splitme,{L, X}},
    receive
        {resp, R} ->
            {ok,R}
        after 10000 ->
            {timeout,[]}
    end.

make_split(Lst, X, CountForWorker, FuncCallb)->
    make_split(Lst, X, {CountForWorker,0,0}, dict:new(), FuncCallb).

make_split([], _,_,Chlds,_)->
    Chlds;

make_split(T, X, {CountForWorker, 0, CurIndex}, Chlds, FuncCallb)->
    NewChlds = FuncCallb(T, X, CountForWorker, Chlds, CurIndex),
    make_split(T, X, {CountForWorker, CountForWorker, CurIndex}, NewChlds, FuncCallb);

make_split([_|T], X, {CountForWorker, CurCnt, CurIndex}, Chlds, FuncCallb)->
    make_split(T, X, {CountForWorker, CurCnt-1, CurIndex+1}, Chlds, FuncCallb).


spawn_chld(L, X, Cnt, Chlds, CurIndex)->
    MyPid = self(),
    Pid = spawn_link(?MODULE, do_work, [L, X, Cnt,fun(Ret) -> MyPid ! {done,{self(),Ret}} end]),
    dict:store(Pid, {L, CurIndex}, Chlds).

%split arrays from dictionary
joinResult(Dct,Ln)->
    Iter = gb_trees:iterator(Dct),
    joinResult(gb_trees:next(Iter),[],[],Ln div 2).

joinResult(none,Result,RResult,_)->
    lists:append(Result,lists:reverse(RResult));
joinResult({Key,Val,Iter}, Result, RResult, Middle) when Middle =< 0 ->
    NRlst = lists:append(RResult,Val),
    joinResult(gb_trees:next(Iter),Result,NRlst,Middle);
joinResult({Key,Val,Iter}, Result,RResult, Middle)->
    NMiddle = Middle-length(Val),
    if
        NMiddle >= 0 ->
                Nlst = lists:append(Result,Val),
                joinResult(gb_trees:next(Iter),Nlst,RResult,NMiddle);
        true ->
                {P1,P2} = lists:split(0-NMiddle,Val),
                Nlst = lists:append(Result,P1),
                NRlst = lists:append(RResult,P2),
                joinResult(gb_trees:next(Iter),Nlst,NRlst,-1)
    end.

loop(State)->
    receive
        {splitme,{L,X}} ->
            NewState  = have_new_work(L,X,State, State#state.spawn_func),
            loop(NewState);

        {done, {Pid,L}} ->
            NewState = child_done(Pid, State, L),
            DSize =dict:size(NewState#state.childs),
            if
            DSize > 0 ->
                loop(NewState);
            true->
                Resps = NewState#state.responces,
                NewState#state.parent_pid ! {resp, joinResult(Resps,length(State#state.data))}
            end;

        {'EXIT', Pid, Reason} ->
            case Reason of
                normal -> loop(State);
                _ ->
                    NewState = child_dyed(Pid, State, fun(A,B,C,D,E) -> spawn_chld(A,B,C,D,E) end),
                    loop(NewState)
            end
    end.

%Got list L, Number X
%State is current state
%callb - function for spawning child
%spawn workers and change state
have_new_work(L, X, State, Callb)->
    New_Chlds = make_split(L, X, State#state.count_in_batch, Callb),
    State#state{childs=New_Chlds,x_num=X,data=L}.

child_done(Pid, State, L)->
    {_, Indx} = dict:fetch(Pid, State#state.childs),
    New_Resps = gb_trees:enter(Indx, L, State#state.responces),     %save result
    New_Chlds = dict:erase(Pid, State#state.childs),        %remove unneeded pid
    State#state{childs=New_Chlds,responces=New_Resps}.

child_dyed(Pid, State, CallBack) ->
    {Lst, Indx} = dict:fetch(Pid,State#state.childs),       %get dyed pid work
    %spawn next worker
    NewChlds = CallBack(Lst, State#state.x_num, State#state.count_in_batch, State#state.childs, Indx),
    State#state{childs=dict:erase(Pid,NewChlds)}.



%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%                                   TESTS
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

do_test()->
    ?assertEqual([2,3,5,4],do([2,3,4,5],10)),
    ?assertEqual([],do([2],1)),
    ?assertEqual([2],do([2],5)),
    ?assertEqual([2,5,3],do([2,3,5],10)),
    ?assertEqual([2,1,3],do([5,2,1,3],5)),
    ?assertEqual([3,2,4,3],do([3,2,5,3,4],5)).

do_work_test()->
    %only ensure that callback is called
    ?assertEqual([2,3,1],do_work([2,4,3,1],4,4,fun(Ret)-> Ret end)).

do_p_test()->
    ?assertEqual([],do_p([2,3,4,5],1,10,[])),   %empty returned
    ?assertEqual([2,3],do_p([2,3,4,5],4,2,[])), %only 2 values
    ?assertEqual([2,3,1],do_p([2,4,3,1],4,4,[])), %3 values with gaps and strong check that 4>4
    ?assertEqual([1],do_p([1,3,4,5],2,1,[])), %only one, and ok
    ?assertEqual([],do_p([2,3,4,5],2,1,[])), %only one and strong check 2>2
    ?assertEqual([1],do_p([1,3,4,5],2,1,[])), % check 2>1
    LSEq=lists:seq(10,100000,1),
    ?assertEqual(LSEq,do_p(LSEq,1000000+1,100000,[])). % check 2>1

test_callb_func([H|T], X, Cnt, Chlds, CurIndex)->
    dict:store(CurIndex,{H, X, Cnt},Chlds).

make_split_test_check(X,_,_,_) when X<0->
    ok;
make_split_test_check(0,_,_,_)->
    ok;
make_split_test_check(Cnt,Dict,M,{S,F})->
    io:format("~p ~n",[Cnt]),
    ?assertEqual({Cnt, S , F}, dict:fetch(Cnt - 1 ,Dict)),
    make_split_test_check(Cnt-M,Dict,M,{S,F}).
make_split_test()->
    Ret  = make_split([1,2,3,4,5],10,1,fun(A,B,C,D,E) -> test_callb_func(A,B,C,D,E) end ),
    ?assertEqual(5,dict:size(Ret)),
    make_split_test_check(5,Ret,1,{10,1}),

    Ret2  = make_split([1,2,3,4,5],10,2,fun(A,B,C,D,E) -> test_callb_func(A,B,C,D,E) end ),
    ?assertEqual(3,dict:size(Ret2)),
    ?assertEqual({1, 10, 2}, dict:fetch(0,Ret2)),
    ?assertEqual({3, 10, 2}, dict:fetch(2,Ret2)),
    ?assertEqual({5, 10, 2}, dict:fetch(4,Ret2)),

    Ret3  = make_split([1,2],10,20,fun(A,B,C,D,E) -> test_callb_func(A,B,C,D,E) end ),
    ?assertEqual(1,dict:size(Ret3)),
    ?assertEqual({1, 10, 20}, dict:fetch(0,Ret3)).

make_split2_test()->
    LSEq=lists:seq(1,100,1),
    Ret4  = make_split(LSEq,1000000000,10,fun(A,B,C,D,E) -> test_callb_func(A,B,C,D,E) end ),
    io:format("~p ~n",[dict:fetch_keys(Ret4)]),
    make_split_test_check(101-10,Ret4,10,{1000000000,10}).

joinResult_test()->
    Dct4 = gb_trees:empty(),
    Dct3 = gb_trees:enter(1,[2,3],Dct4),
    Dct2 = gb_trees:enter(2,[4],Dct3),
    Dct = gb_trees:enter(3,[5],Dct2),
    ?assertEqual([2,3,5,4],joinResult(Dct,4)).

joinResult2_test()->
    Dct4 = gb_trees:empty(),
    Dct3 = gb_trees:enter(3,[5],Dct4),
    Dct2 = gb_trees:enter(1,[2,3],Dct3),
    Dct = gb_trees:enter(2,[4],Dct2),
    ?assertEqual([2,3,5,4],joinResult(Dct,4)).

joinResult3_test()->
    Dct4 = gb_trees:empty(),
    Dct3 = gb_trees:enter(3,[5],Dct4),
    Dct2 = gb_trees:enter(1,[2,3],Dct3),
    Dct1 = gb_trees:enter(4,[6],Dct2),
    Dct = gb_trees:enter(2,[4],Dct1),
    ?assertEqual([2,3,6,5,4],joinResult(Dct,5)).

    %have_new_work(L, X, State, Callb)->
have_new_work_test()->
    State = have_new_work([1,2,3,4,5],10, #state{count_in_batch=2},fun(A,B,C,D,E) -> test_callb_func(A,B,C,D,E) end),
    Ret2 = State#state.childs,
    ?assertEqual(10,State#state.x_num),
    ?assertEqual({1, 10, 2}, dict:fetch(0,Ret2)),
    ?assertEqual({3, 10, 2}, dict:fetch(2,Ret2)),
    ?assertEqual({5, 10, 2}, dict:fetch(4,Ret2)).

child_done_test()->
    Dict2 = dict:new(),
    Dict = dict:store(1,{[2,3],5},Dict2),
    #state{childs=Childs,responces=Responces}= child_done(1,#state{childs=Dict},[1,20,40]),
    ?assertEqual([1,20,40],gb_trees:get(5,Responces)),
    ?assertEqual(0,dict:size(Childs)).

child_dyed_test()->
    Dict2 = dict:new(),
    Dict = dict:store(1,{[2,3],5},Dict2),
    #state{childs=Childs} = child_dyed(1,#state{childs=Dict},fun(A,B,C,D,E) -> test_callb_func(A,B,C,D,E) end),
    ?assertEqual(1,dict:size(Childs)),
    ?assertEqual({2,0,10},dict:fetch(5,Childs)).

spawn_chld_TST(L, X, Cnt, Chlds, CurIndex)->
    MyPid = self(),
    Rnd = random:uniform(10),
    if
        Rnd>5 -> Pid = spawn_link(?MODULE, do_work, [L, X, Cnt,fun(Ret) -> MyPid ! {done,{self(),Ret}} end]);
        true -> Pid = spawn_link(fun() ->  exit(tada) end)
    end,
    dict:store(Pid, {L, CurIndex}, Chlds).

do_spl_fake(L,X)->
    MyPid=  self(),
    Pid = spawn(fun()->
                    process_flag(trap_exit, true),
                    loop(#state{ parent_pid=MyPid,
                                spawn_func=fun(A,B,C,D,E) -> spawn_chld_TST(A,B,C,D,E) end
                                }
                        )
                end),
    Pid!{splitme,{L, X}},
    receive
        {resp, R} ->
            {ok,R}
        after 1000 ->
            {timeout,[]}
    end.

do_spl_test()->
    ?assertEqual({ok,[1,2,3]},do_spl([1,2,3],5)),
    ?assertEqual({ok,[1,2,3]},do_spl_fake([1,2,3],5)),
    ThLst = lists:seq(10,50,1),
    {ok,Out} = do_spl_fake(ThLst,100000+1),
    io:format("~p ~n",[Out]),
    ?assertEqual(do(ThLst,10000000000),Out).