<commit_before>% @copyright 2011 Zuse Institute Berlin

%   Licensed under the Apache License, Version 2.0 (the "License");
%   you may not use this file except in compliance with the License.
%   You may obtain a copy of the License at
%
%       http://www.apache.org/licenses/LICENSE-2.0
%
%   Unless required by applicable law or agreed to in writing, software
%   distributed under the License is distributed on an "AS IS" BASIS,
%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%   See the License for the specific language governing permissions and
%   limitations under the License.

%%% @author <NAME> <<EMAIL>>
%%% @doc    RRD clone.
%%% @end
%% @version $Id$
-module(rrd).
-author('<EMAIL>').
-vsn('$Id$').

-include("scalaris.hrl").
-include("record_helpers.hrl").

-ifdef(with_export_type_support).
-export_type([rrd/0]).
-endif.

% external API with transparent time handling
-export([create/3, add_now/2, dump/1, dump_with/2]).

% external API without transparent time handling
-export([create/4, add/3]).

% internal API for the monitor process
-export([get_slot_start/2, reduce_timeslots/2, add_nonexisting_timeslots/2,
         get_type/1, get_slot_length/1, get_current_time/1]).

% for unit-testing only
-export([timestamp2us/1, us2timestamp/1]).

% gauge: record newest value of a time slot,
% counter: sum up all values of a time slot,
% event: record every event (incl. timestamp) in a time slot,
% timing: record time spans, store {sum(x), sum(x^2), count(x), min(x), max(x)}
-type timeseries_type() :: gauge | counter | event | timing.
-type fill_policy_type() :: set_undefined | keep_last_value.
-type time() :: util:time().
-type internal_time() :: non_neg_integer().
-type timespan() :: pos_integer().
-type update_fun(T, NewV) :: fun((Time::internal_time(), Old::T | undefined, NewV) -> T).

-record(rrd, {slot_length   = ?required(rrd, slot_length)   :: timespan(),
              count         = ?required(rrd, count)         :: pos_integer(),
              type          = ?required(rrd, type)          :: timeseries_type(),
              % index of current slot
              current_index = ?required(rrd, current_index) :: non_neg_integer(),
              % current slot starts at current_time and lasts for step_size
              % time units
              current_time  = ?required(rrd, current_time)  :: internal_time(),
              data          = ?required(rrd, data)          :: array(),
              fill_policy   = ?required(rrd, fill_policy)   :: fill_policy_type()
            }).

-opaque rrd() :: #rrd{}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
% External API with transparent time handling
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% @doc SlotLength in microseconds
-spec create(SlotLength::timespan(), Count::pos_integer(), Type::timeseries_type()) ->
    rrd().
create(SlotLength, Count, Type) ->
    create(SlotLength, Count, Type, os:timestamp()).

% @doc Note: gauge, counter and timing types accept only number() as value, event
%      accepts any value.
-spec add_now(Value::term(), rrd()) -> rrd().
add_now(Value, DB) ->
    add(os:timestamp(), Value, DB).

-spec dump(rrd()) -> [{From::time(), To::time(), term()}].
dump(DB) ->
    dump_with(DB, fun(_DB, From, To, X) -> {From, To, X} end).

-spec dump_with(rrd(), fun((rrd(), From::time(), To::time(), Value::term()) -> T)) -> [T].
dump_with(DB, F) ->
    SlotLength = DB#rrd.slot_length,
    CurrentIndex = DB#rrd.current_index,
    Count = DB#rrd.count,
    dump_internal(DB, (CurrentIndex + 1) rem Count, CurrentIndex,
                  DB#rrd.current_time - (Count - 1) * SlotLength, [], F).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
% Internal API (allows to specify timestamps explicitly)
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%@doc StepSize in milliseconds
-spec create(SlotLength::timespan(), Count::pos_integer(), Type::timeseries_type(),
             StartTime::time() | internal_time()) -> rrd().
create(SlotLength, Count, Type, {_, _, _} = StartTime) ->
    create(SlotLength, Count, Type, timestamp2us(StartTime));
create(SlotLength, Count, Type, StartTime) ->
    #rrd{slot_length = SlotLength, count = Count, type = Type, current_index = 0,
         current_time = StartTime, data = array:new(Count),
         fill_policy = set_undefined}.

-spec gauge_update_fun(Time::internal_time(), Old::T | undefined, New::T) -> T when T :: number().
gauge_update_fun(_Time, _Old, New) -> New.

-spec counter_update_fun(Time::internal_time(), Old::T | undefined, New::T) -> T when T :: number().
counter_update_fun(_Time, undefined, New) -> New;
counter_update_fun(_Time, Old, New) -> Old + New.

-spec event_update_fun(Time::internal_time(), Old::[{internal_time(), T}] | undefined, NewV)
        -> [{internal_time(), T | NewV}].
event_update_fun(Time, undefined, New) -> [{Time, New}];
event_update_fun(Time, Old, New) -> lists:append(Old, [{Time, New}]).

-spec timing_update_fun(Time::internal_time(), Old::{T, T, pos_integer(), T, T} | undefined, New::T)
        -> {T, T, pos_integer(), T, T} when T :: number().
timing_update_fun(_Time, undefined, New) -> {New, New*New, 1, New, New};
timing_update_fun(_Time, {Sum, Sum2, Count, Min, Max}, New) ->
    {Sum + New, Sum2 + New*New, Count + 1, erlang:min(Min, New), erlang:max(Max, New)}.

-spec keep_old_update_fun(Time::internal_time(), Old::T | undefined, NewV) -> T | NewV.
keep_old_update_fun(_Time, undefined, New) -> New;
keep_old_update_fun(_Time, Old, _New) -> Old.

% @doc Note: gauge, counter and timing types accept only number() as value, event
%      accepts any value.
-spec add(Time::time() | internal_time(), Value::term(), rrd()) -> rrd().
add({_, _, _} = ExternalTime, Value, DB) ->
    add(timestamp2us(ExternalTime), Value, DB);
add(Time, Value, DB) ->
    case DB#rrd.type of
        gauge ->
            add_with(Time, Value, DB, fun gauge_update_fun/3);
        counter ->
            add_with(Time, Value, DB, fun counter_update_fun/3);<commit_msg>- fixed compilation on Erlang <= R13B04<commit_after>% @copyright 2011 Zuse Institute Berlin

%   Licensed under the Apache License, Version 2.0 (the "License");
%   you may not use this file except in compliance with the License.
%   You may obtain a copy of the License at
%
%       http://www.apache.org/licenses/LICENSE-2.0
%
%   Unless required by applicable law or agreed to in writing, software
%   distributed under the License is distributed on an "AS IS" BASIS,
%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%   See the License for the specific language governing permissions and
%   limitations under the License.

%%% @author <NAME> <<EMAIL>>
%%% @doc    RRD clone.
%%% @end
%% @version $Id$
-module(rrd).
-author('<EMAIL>').
-vsn('$Id$').

-include("scalaris.hrl").
-include("record_helpers.hrl").

-ifdef(with_export_type_support).
-export_type([rrd/0]).
-endif.

% external API with transparent time handling
-export([create/3, add_now/2, dump/1, dump_with/2]).

% external API without transparent time handling
-export([create/4, add/3]).

% internal API for the monitor process
-export([get_slot_start/2, reduce_timeslots/2, add_nonexisting_timeslots/2,
         get_type/1, get_slot_length/1, get_current_time/1]).

% for unit-testing only
-export([timestamp2us/1, us2timestamp/1]).

% gauge: record newest value of a time slot,
% counter: sum up all values of a time slot,
% event: record every event (incl. timestamp) in a time slot,
% timing: record time spans, store {sum(x), sum(x^2), count(x), min(x), max(x)}
-type timeseries_type() :: gauge | counter | event | timing.
-type fill_policy_type() :: set_undefined | keep_last_value.
-type time() :: util:time().
-type internal_time() :: non_neg_integer().
-type timespan() :: pos_integer().
-type update_fun(T, NewV) :: fun((Time::internal_time(), Old::T | undefined, NewV) -> T).

-record(rrd, {slot_length   = ?required(rrd, slot_length)   :: timespan(),
              count         = ?required(rrd, count)         :: pos_integer(),
              type          = ?required(rrd, type)          :: timeseries_type(),
              % index of current slot
              current_index = ?required(rrd, current_index) :: non_neg_integer(),
              % current slot starts at current_time and lasts for step_size
              % time units
              current_time  = ?required(rrd, current_time)  :: internal_time(),
              data          = ?required(rrd, data)          :: array(),
              fill_policy   = ?required(rrd, fill_policy)   :: fill_policy_type()
            }).

-opaque rrd() :: #rrd{}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
% External API with transparent time handling
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% @doc SlotLength in microseconds
-spec create(SlotLength::timespan(), Count::pos_integer(), Type::timeseries_type()) ->
    rrd().
create(SlotLength, Count, Type) ->
    create(SlotLength, Count, Type, os:timestamp()).

% @doc Note: gauge, counter and timing types accept only number() as value, event
%      accepts any value.
-spec add_now(Value::term(), rrd()) -> rrd().
add_now(Value, DB) ->
    add(os:timestamp(), Value, DB).

-spec dump(rrd()) -> [{From::time(), To::time(), term()}].
dump(DB) ->
    dump_with(DB, fun(_DB, From, To, X) -> {From, To, X} end).

-spec dump_with(rrd(), fun((rrd(), From::time(), To::time(), Value::term()) -> T)) -> [T].
dump_with(DB, F) ->
    SlotLength = DB#rrd.slot_length,
    CurrentIndex = DB#rrd.current_index,
    Count = DB#rrd.count,
    dump_internal(DB, (CurrentIndex + 1) rem Count, CurrentIndex,
                  DB#rrd.current_time - (Count - 1) * SlotLength, [], F).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%
% Internal API (allows to specify timestamps explicitly)
%
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%@doc StepSize in milliseconds
-spec create(SlotLength::timespan(), Count::pos_integer(), Type::timeseries_type(),
             StartTime::time() | internal_time()) -> rrd().
create(SlotLength, Count, Type, {_, _, _} = StartTime) ->
    create(SlotLength, Count, Type, timestamp2us(StartTime));
create(SlotLength, Count, Type, StartTime) ->
    #rrd{slot_length = SlotLength, count = Count, type = Type, current_index = 0,
         current_time = StartTime, data = array:new(Count),
         fill_policy = set_undefined}.

-spec gauge_update_fun(Time::internal_time(), Old::T | undefined, New::T) -> T when is_subtype(T, number()).
gauge_update_fun(_Time, _Old, New) -> New.

-spec counter_update_fun(Time::internal_time(), Old::T | undefined, New::T) -> T when is_subtype(T, number()).
counter_update_fun(_Time, undefined, New) -> New;
counter_update_fun(_Time, Old, New) -> Old + New.

-spec event_update_fun(Time::internal_time(), Old::[{internal_time(), T}] | undefined, NewV)
        -> [{internal_time(), T | NewV}].
event_update_fun(Time, undefined, New) -> [{Time, New}];
event_update_fun(Time, Old, New) -> lists:append(Old, [{Time, New}]).

-spec timing_update_fun(Time::internal_time(), Old::{T, T, pos_integer(), T, T} | undefined, New::T)
        -> {T, T, pos_integer(), T, T} when is_subtype(T, number()).
timing_update_fun(_Time, undefined, New) -> {New, New*New, 1, New, New};
timing_update_fun(_Time, {Sum, Sum2, Count, Min, Max}, New) ->
    {Sum + New, Sum2 + New*New, Count + 1, erlang:min(Min, New), erlang:max(Max, New)}.

-spec keep_old_update_fun(Time::internal_time(), Old::T | undefined, NewV) -> T | NewV.
keep_old_update_fun(_Time, undefined, New) -> New;
keep_old_update_fun(_Time, Old, _New) -> Old.

% @doc Note: gauge, counter and timing types accept only number() as value, event
%      accepts any value.
-spec add(Time::time() | internal_time(), Value::term(), rrd()) -> rrd().
add({_, _, _} = ExternalTime, Value, DB) ->
    add(timestamp2us(ExternalTime), Value, DB);
add(Time, Value, DB) ->
    case DB#rrd.type of
        gauge ->
            add_with(Time, Value, DB, fun gauge_update_fun/3);
        counter ->
            add_with(Time, Value, DB, fun counter_update_fun/3);
{'fraction_non_alphanumeric': 0.20372771474878443, 'fraction_numerical': 0.006320907617504052, 'mean_word_length': 3.9502607300441235, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 77, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2581928', 'n_tokens_mistral': 3896, 'n_tokens_neox': 3832, 'n_words': 1074}