Packages
dqe
0.1.34
0.4.15
0.4.14
0.4.13
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.25
0.3.24
0.3.23
0.3.22
0.3.21
0.3.20
0.3.19
0.3.18
0.3.17
0.3.16
0.3.15
0.3.14
0.3.13
0.3.12
0.3.11
0.3.10
0.3.9
0.3.8
0.3.7
0.3.6
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.2.2
0.2.1
0.2.0
0.1.36
0.1.35
0.1.34
0.1.33
0.1.22
0.1.21
0.1.20
0.1.10
0.1.9
DalmatinerDB query engine
Current section
Files
Jump to
Current section
Files
src/dqe_histogram.erl
-module(dqe_histogram).
-behaviour(dflow).
-include_lib("mmath/include/mmath.hrl").
-export([init/1, describe/1, start/2, emit/3, done/2]).
-record(state, {
htv :: pos_integer(),
sf :: pos_integer(),
time :: pos_integer(),
acc = <<>> :: binary(),
resolution :: pos_integer()
}).
init([HTV, SF, SubQ, Time]) ->
{ok, #state{htv = HTV, sf = SF, time = Time}, SubQ}.
start({_Start, _Count}, State) ->
{ok, State}.
describe(#state{htv = HTV, sf = SF, time = Time}) ->
["histogram(", integer_to_list(HTV), integer_to_list(SF),
integer_to_list(Time), "ms)"].
%% When we get the first data we can calculate both the applied
%% time and the upwards resolution.
emit(Child, {realized, {Data, Resolution}},
State = #state{resolution = undefined, time = Time}) ->
Time1 = dqe_time:apply_times(Time, Resolution),
emit(Child, {Data, Resolution},
State#state{resolution = Time1 * Resolution, time = Time1});
emit(_Child, {realized, {Data, _R}},
State = #state{htv = HTV, sf = SF, acc = Acc, time = Time}) ->
case execute(HTV, SF, <<Data/binary, Acc/binary>>, Time, []) of
{Acc1, []} ->
{ok, State#state{acc = Acc1}};
{Acc2, Histograms} ->
Histograms1 = lists:reverse(Histograms),
{emit, {histograms, {Histograms1, State#state.resolution}},
State#state{acc = Acc2}}
end.
done(_Child, State = #state{acc = <<>>}) ->
{done, State};
done(_Child, State = #state{htv = HTV, sf = SF, acc = Acc}) ->
Data = mk_hist(HTV, SF, Acc),
{done, {histograms, {[Data], State#state.resolution}}, State#state{acc = <<>>}}.
execute(HTV, SF, Acc, T1, AccEmit) when byte_size(Acc) >= T1 * ?DATA_SIZE ->
MinSize = T1 * ?DATA_SIZE,
<<Data:MinSize/binary, Acc1/binary>> = Acc,
H = mk_hist(HTV, SF, Data),
execute(HTV, SF, Acc1, T1, [H | AccEmit]);
execute(_HTV, _SF, Acc, _, AccEmit) ->
{Acc, AccEmit}.
mk_hist(HVT, SF, Data) ->
{ok, H} = hdr_histogram:open(HVT, SF),
[ hdr_histogram:record(H, V)|| V <- mmath_bin:to_list(mmath_bin:derealize(Data))],
H.