%%%-------------------------------------------------------------------
%%% File    : get_records.erl
%%% Author  : Jay Nelson <jay@duomark.com>
%%% Description : 
%%%   Benchmark for stream_split_rll/2 performance.
%%%
%%% Created : 26 Jun 2005 by Jay Nelson <jay@duomark.com>
%%%-------------------------------------------------------------------
-module(split_records).

%% API
-export([
	 gen_recs/1, time/2, test/2,
	 slow_collect/1, normal_collect/1, fast_collect/1, fastest_collect/1
	 ]).


%%====================================================================
%% Tests
%%====================================================================
gen_recs(small) ->  gen_recs(255, 25);
gen_recs(large) ->  gen_recs(255, 30000).

gen_recs(Length, Count) ->
    random:seed(),
    RecSizes = [ random:uniform(Length) || _N <- lists:seq(1,Count) ],
    Recs = [ list_to_binary([Size, Rec]) || Size <- RecSizes,
				    (Rec = lists:duplicate(Size,0)) =/= []  ],
    concat_binary(Recs).


time(Type, Size) ->
    Bin = gen_recs(Size),
    garbage_collect(),
    timer:tc(split_records, test, [Type, Bin]).


test(Type, DB) ->
    L = case Type of
	    slow     -> slow_collect(DB);
	    normal   -> normal_collect(DB);
	    fast     -> fast_collect(DB);
	    fastest  -> fastest_collect(DB)
	end,
    length(L).


%%====================================================================
%% API
%%====================================================================
slow_collect(DB) ->
    LDB = binary_to_list(DB),
    slow_collect(LDB, []).

slow_collect([], RecsCollected) ->
    lists:reverse(RecsCollected);
slow_collect([SizeRec | Rest], RecsCollected) ->
    {ThisRec, Remaining} = get_slow_rec(SizeRec, Rest),
    slow_collect(Remaining, [list_to_binary(ThisRec) | RecsCollected]).


normal_collect(DB) ->
    normal_collect(DB, []).

normal_collect(<<>>, RecsCollected) ->
    lists:reverse(RecsCollected);
normal_collect(<<SizeRec,  Rest/binary>>, RecsCollected) ->
    << ThisRec:SizeRec/binary, OtherRecs/binary>> = Rest,
    normal_collect(OtherRecs, [ThisRec | RecsCollected]).


fast_collect(DB) ->
    fast_collect(DB, 0, []).

fast_collect(RecSet, AlreadySeen, RecsCollected)  when AlreadySeen >= size(RecSet) ->
    lists:reverse(RecsCollected);
fast_collect(RecSet, AlreadySeen, RecsCollected) ->
    << _Skip:AlreadySeen/binary, SizeRec, Rest/binary>> = RecSet,
    <<ThisRec:SizeRec/binary, _OtherRecs/binary>> = Rest,
    fast_collect(RecSet, AlreadySeen + SizeRec + 1, [ThisRec | RecsCollected]).


fastest_collect(DB) ->
    stream_split_rll(DB, 1).


%%====================================================================
%% Internal functions
%%====================================================================
get_slow_rec(SizeRec, RestRecs) ->
    ThisRec = lists:sublist(RestRecs, 1, SizeRec),
    Remaining = lists:nthtail(SizeRec, RestRecs),
    {ThisRec, Remaining}.