%% @doc: Registers stored in one large binary
%%
%% This backend uses one plain Erlang binary to store registers. The
%% cost of rebuilding the binary is amortized by keeping a buffer of
%% inserts to perform in the future.

-module(hyper_binary).
-include_lib("eunit/include/eunit.hrl").
-behaviour(hyper_register).
%%-compile(native).

-export([new/1,
         set/3,
         compact/1,
         max_merge/1,
         max_merge/2,
         reduce_precision/2,
         bytes/1,
         register_sum/1,
         zero_count/1,
         encode_registers/1,
         decode_registers/2]).


-define(VALUE_SIZE, 6).
-define(MERGE_THRESHOLD, 0.05).

-record(buffer, {buf, buf_size, p, convert_threshold}).
-record(dense, {b, buf, buf_size, merge_threshold}).


new(P) ->
    new_buffer(P).

new_buffer(P) ->
    M = m(P),
    ConvertThreshold = M div (5 * 8), % 5 words for each entry
    #buffer{buf = [], buf_size = 0, p = P, convert_threshold = ConvertThreshold}.

new_dense(P) ->
    M = m(P),
    T = max(trunc(M * ?MERGE_THRESHOLD), 16),
    #dense{b = empty_binary(M), buf = [], buf_size = 0, merge_threshold = T}.


set(Index, Value, #buffer{buf = [{Index, OldValue} | Rest]} = Buffer) ->
    Buffer#buffer{buf = [{Index, max(Value, OldValue)} | Rest]};

set(Index, Value, #buffer{buf = Buf, buf_size = BufSize} = Buffer) ->
    NewBuffer = Buffer#buffer{buf = [{Index, Value} | Buf],
                              buf_size = BufSize + 1},
    case NewBuffer#buffer.buf_size < NewBuffer#buffer.convert_threshold of
        true ->
            NewBuffer;
        false ->
            buffer2dense(NewBuffer)
    end;

set(Index, Value, #dense{buf = Buf, buf_size = BufSize} = Dense) ->
    LeftOffset = Index * ?VALUE_SIZE,
    <<_:LeftOffset/bitstring, R:?VALUE_SIZE/integer, _/bitstring>> = Dense#dense.b,

    if
        R < Value ->
            New = Dense#dense{buf = [{Index, Value} | Buf],
                              buf_size = BufSize + 1},
            case New#dense.buf_size < Dense#dense.merge_threshold of
                true ->
                    New;
                false ->
                    compact(New)
            end;
        true ->
            Dense
    end.


compact(#buffer{} = Buffer) ->
    Buffer;

compact(#dense{b = B, buf = Buf} = Dense) ->
    NewB = merge_buf(B, max_registers(Buf)),
    Dense#dense{b = NewB, buf = [], buf_size = 0}.


max_merge([First | Rest]) ->
    lists:foldl(fun max_merge/2, First, Rest).

max_merge(#dense{b = SmallB, buf = SmallBuf, buf_size = SmallBufSize},
          #dense{b = BigB, buf = BigBuf, buf_size = BigBufSize} = Big) ->
    case SmallBufSize + BigBufSize < Big#dense.merge_threshold of
        true ->
            Merged = do_merge(SmallB, BigB, <<>>),
            Big#dense{b = Merged,
                      buf = SmallBuf ++ BigBuf,
                      buf_size = SmallBufSize + BigBufSize};
        false ->
            BigWithBuf = merge_buf(BigB, max_registers(SmallBuf ++ BigBuf)),
            Merged = do_merge(SmallB, BigWithBuf, <<>>),
            Big#dense{b = Merged,
                      buf = [],
                      buf_size = 0}
    end;

max_merge(#buffer{buf = Buf, buf_size = BufferSize},
          #dense{buf = DenseBuf, buf_size = DenseSize} = Dense) ->
    case BufferSize + DenseSize < Dense#dense.merge_threshold of
        true ->
            Dense#dense{buf = Buf ++ DenseBuf,
                        buf_size = BufferSize + DenseSize};
        false ->
            Merged = max_registers(DenseBuf ++ Buf),
            Dense#dense{buf = Merged,
                        buf_size = length(Merged)}
    end;

max_merge(#dense{} = Dense, #buffer{} = Buffer) ->
    max_merge(Buffer, Dense);

max_merge(#buffer{buf = LeftBuf, buf_size = LeftBufSize},
          #buffer{buf = RightBuf, buf_size = RightBufSize} = Right) ->
    case LeftBufSize + RightBufSize < Right#buffer.convert_threshold of
        true ->
            Right#buffer{buf = LeftBuf ++ RightBuf,
                         buf_size = LeftBufSize + RightBufSize};
        false ->
            Merged = max_registers(LeftBuf ++ RightBuf),
            NewRight = Right#buffer{buf = Merged,
                                    buf_size = length(Merged)},
            case NewRight#buffer.buf_size < NewRight#buffer.convert_threshold of
                true ->
                    NewRight;
                false ->
                    buffer2dense(NewRight)
            end
    end.


reduce_precision(NewP, #dense{} = Dense) ->
    Buf = register_fold(m(NewP), Dense),
    Empty = new_dense(NewP),
    compact(Empty#dense{buf = Buf});
reduce_precision(NewP, #buffer{} = Buffer) ->
    Buf = max_registers(register_fold(m(NewP), Buffer)),
    Empty = new_buffer(NewP),
    Empty#buffer{buf = Buf, buf_size = length(Buf)}.


%% fold an HLL to a lower number of registers `NewM` by projecting the values
%% onto their new index, see
%% http://research.neustar.biz/2012/09/12/set-operations-on-hlls-of-different-sizes/
%% NOTE: this function does not perform the max_registers step
register_fold(NewM, B) ->
    fold(fun (I, V, Acc) -> [{I rem NewM, V} | Acc] end, [], B).

register_sum(B) ->
    fold(fun (_, 0, Acc) -> Acc + 1.0;
             (_, 1, Acc) -> Acc + 0.5;
             (_, 2, Acc) -> Acc + 0.25;
             (_, 3, Acc) -> Acc + 0.125;
             (_, 4, Acc) -> Acc + 0.0625;
             (_, 5, Acc) -> Acc + 0.03125;
             (_, 6, Acc) -> Acc + 0.015625;
             (_, 7, Acc) -> Acc + 0.0078125;
             (_, 8, Acc) -> Acc + 0.00390625;
             (_, 9, Acc) -> Acc + 0.001953125;
             (_, V, Acc) -> Acc + math:pow(2, -V) end,
         0, B).

zero_count(B) ->
    fold(fun (_, 0, Acc) -> Acc + 1;
             (_, _, Acc) -> Acc
         end, 0, B).

encode_registers(#buffer{} = Buffer) ->
    encode_registers(buffer2dense(Buffer));

encode_registers(#dense{b = B}) ->
    << <<I:8/integer>> || <<I:?VALUE_SIZE/integer>> <= B >>.

decode_registers(AllBytes, P) ->
    M = m(P),
    Bytes = case AllBytes of
                <<B:M/binary>>    -> B;
                <<B:M/binary, 0>> -> B
            end,
    Dense = new_dense(P),
    Dense#dense{b = << <<I:?VALUE_SIZE/integer>> || <<I:8>> <= Bytes >>}.


bytes(#dense{b = B}) ->
    erlang:byte_size(B);
bytes(#buffer{} = Buffer) ->
    erts_debug:flat_size(Buffer) * 8.


%%
%% INTERNALS
%%

m(P) ->
    trunc(math:pow(2, P)).

empty_binary(M) ->
    list_to_bitstring([<<0:?VALUE_SIZE/integer>> || _ <-  lists:seq(0, M-1)]).

max_registers(Buf) ->
    lists:foldl(fun ({I, V}, Acc) ->
                          case orddict:find(I, Acc) of
                              {ok, R} when R >= V ->
                                  Acc;
                              _ ->
                                  orddict:store(I, V, Acc)
                          end
                  end, orddict:new(), lists:reverse(lists:sort(Buf))).


buffer2dense(#buffer{buf = Buf, p = P}) ->
    Dense = new_dense(P),
    Merged = merge_buf(Dense#dense.b, max_registers(Buf)),
    Dense#dense{b = Merged}.


do_merge(<<>>, <<>>, Acc) ->
    Acc;
do_merge(<<Left:?VALUE_SIZE/integer, SmallRest/bitstring>>,
         <<Right:?VALUE_SIZE/integer, BigRest/bitstring>>,
         Acc) ->
    do_merge(SmallRest, BigRest, <<Acc/bits, (max(Left, Right)):?VALUE_SIZE>>).


fold(F, Acc, #buffer{} = Buffer) ->
    fold(F, Acc, buffer2dense(Buffer));

fold(F, Acc, #dense{b = B, buf = Buf}) ->
    do_fold(F, Acc, merge_buf(B, max_registers(Buf)), 0).

do_fold(_, Acc, <<>>, _) ->
    Acc;
do_fold(F, Acc, <<Value:?VALUE_SIZE/integer, Rest/bitstring>>, Index) ->
    do_fold(F, F(Index, Value, Acc), Rest, Index+1).


merge_buf(B, L) ->
    merge_buf(B, L, -1, <<>>).

merge_buf(B, [], _PrevIndex, Acc) ->
    <<Acc/bitstring, B/bitstring>>;

merge_buf(B, [{Index, Value} | Rest], PrevIndex, Acc) ->
    I = (Index - PrevIndex - 1) * ?VALUE_SIZE,
    case B of
        <<Left:I/bitstring, OldValue:?VALUE_SIZE/integer, Right/bitstring>> ->
            case OldValue < Value of
                true ->
                    NewAcc = <<Acc/bitstring,
                               Left/bitstring,
                               Value:?VALUE_SIZE/integer>>,
                    merge_buf(Right, Rest, Index, NewAcc);
                false ->
                    NewAcc = <<Acc/bitstring,
                               Left/bitstring,
                               OldValue:?VALUE_SIZE/integer>>,
                    merge_buf(Right, Rest, Index, NewAcc)
            end;

        <<Left:I/bitstring>> ->
            <<Acc/bitstring, Left/bitstring, Value:?VALUE_SIZE/integer>>
    end.


%%
%% TESTS
%%


merge_test() ->
    P = 4, M = m(P),
    Tmp1 = [{1, 1},
            {3, 3},
            {9, 3},
            {15, 15}],
    Tmp2 = [{3, 5},
            {9, 2},
            {10, 5}],

    {buffer, [], 0, T, _} = new(P),

    {dense, Compact, [], 0, _} =
        compact({dense, empty_binary(M), Tmp1, length(Tmp1), T}),

    {dense, Compact2, [], 0, _} =
        compact({dense, Compact, Tmp2, length(Tmp2), T}),

    Expected = [0, 1, 0, 5,
                0, 0, 0, 0,
                0, 3, 5, 0,
                0, 0, 0, 15],
    Ints = [I || <<I:?VALUE_SIZE/integer>> <= Compact2],

    ?assertEqual(Expected, Ints).


serialize_test() ->
    H = compact(lists:foldl(fun (I, Acc) -> set(I, I, Acc) end,
                            new(4), lists:seq(0, 15))),

    ?assertEqual(<<0,1,2,3,
                   4,5,6,7,
                   8,9,10,11,
                   12,13,14,15>>, encode_registers(H)),
    ?assertEqual(H, decode_registers(encode_registers(H), 4)).

max_registers_test() ->
    ?assertEqual([{3, 3}], max_registers([{3, 1}, {3, 2}, {3, 3}])).