<commit_before>%%%-------------------------------------------------------------------
%%% @author <NAME> <<EMAIL>>
%%% @copyright (C) 2014, <NAME>
%%% @doc
%%%
%%% @end
%%% Created : 24 Sep 2014 by <NAME> <<EMAIL>>
%%%-------------------------------------------------------------------
-module(metric_io).

-behaviour(gen_server).

%% API
-export([start_link/1,
         empty/1, fold/3, delete/1, close/1,
         buckets/1, metrics/2, delete/2,
         read/7, write/5, write/6]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).

-define(SERVER, ?MODULE).
-define(WEEK, 604800). %% Seconds in a week.
-define(MAX_Q_LEN, 20).

-record(state, {
          partition,
          node,
          mstore=gb_trees:empty(),
          dir,
          fold_size
         }).

%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Starts the server
%%
%% @spec start_link() -> {ok, Pid} | ignore | {error, Error}
%% @end
%%--------------------------------------------------------------------
start_link(Partition) ->
    gen_server:start_link(?MODULE, [Partition], []).

write(Pid, Bucket, Metric, Time, Value) ->
    write(Pid, Bucket, Metric, Time, Value, ?MAX_Q_LEN).

write(Pid, Bucket, Metric, Time, Value, MaxLen) ->
    case erlang:process_info(Pid, message_queue_len) of
        {message_queue_len, N} when N > MaxLen ->
            swrite(Pid, Bucket, Metric, Time, Value);
        _ ->
            gen_server:cast(Pid, {write, Bucket, Metric, Time, Value})
    end.

swrite(Pid, Bucket, Metric, Time, Value) ->
    gen_server:call(Pid, {write, Bucket, Metric, Time, Value}).

read(Pid, Bucket, Metric, Time, Count, ReqID, Sender) ->
    gen_server:cast(Pid, {read, Bucket, Metric, Time, Count, ReqID, Sender}).

buckets(Pid) ->
    gen_server:call(Pid, buckets).

metrics(Pid, Bucket) ->
    gen_server:call(Pid, {metrics, Bucket}).

fold(Pid, Fun, Acc0) ->
    gen_server:call(Pid, {fold, Fun, Acc0}).

empty(Pid) ->
    gen_server:call(Pid, empty).

delete(Pid) ->
    gen_server:call(Pid, delete).

close(Pid) ->
    gen_server:call(Pid, close).

delete(Pid, Bucket) ->
    gen_server:call(Pid, {delete, Bucket}).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the server
%%
%% @spec init(Args) -> {ok, State} |
%%                     {ok, State, Timeout} |
%%                     ignore |
%%                     {stop, Reason}
%% @end
%%--------------------------------------------------------------------
init([Partition]) ->
    process_flag(trap_exit, true),
    DataDir = case application:get_env(riak_core, platform_data_dir) of
                  {ok, DD} ->
                      DD;
                  _ ->
                      "data"
              end,
    FoldSize = case application:get_env(metric_vnode, handoff_chunk) of
                   {ok, FS} ->
                       FS;
                   _ ->
                       10*1024
               end,
    PartitionDir = [DataDir, $/,  integer_to_list(Partition)],

    {ok, #state{ partition = Partition,
                 node = node(),
                 dir = PartitionDir,
                 fold_size = FoldSize
               }}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%%
%% @spec handle_call(Request, From, State) ->
%%                                   {reply, Reply, State} |
%%                                   {reply, Reply, State, Timeout} |
%%                                   {noreply, State} |
%%                                   {noreply, State, Timeout} |
%%                                   {stop, Reason, Reply, State} |
%%                                   {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------
%% fold_fun(Fun, Bucket) ->
%%     fun(Metric, Time, V,
%%         {[{_Metric2, _}|_] = AccL, AccIn})
%%           when Metric =/= _Metric2 ->
%%             AccOut = Fun({Bucket, {Metric, Time}}, V, AccIn)
%%     end.

-record(facc,
        {
          metric,
          size = 0,
          hacc,
          lacc = [],
          bucket,
          acc_fun
        }).
-define(FOLD_SIZE, 50).

fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric2,
                   lacc = []}) when
      Metric =/= Metric2 ->
    Acc#facc{
      metric = Metric,
      size = mmath_bin:length(V),
      lacc = [{Time, V}]};
fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric2,
                   bucket = Bucket,
                   lacc = AccL,
                   acc_fun = Fun,
                   hacc = AccIn}) when
      Metric =/= Metric2 ->
    AccOut = Fun({Bucket, Metric2}, lists:reverse(AccL), AccIn),
    Acc#facc{
      metric = Metric,
      size = mmath_bin:length(V),
      hacc = AccOut,
      lacc = [{Time, V}]};
fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric,
                   bucket = Bucket,
                   size = Size,
                   lacc = AccL,
                   acc_fun = Fun,
                   hacc = AccIn}) when
      Size > ?FOLD_SIZE ->
    AccOut = Fun({Bucket, Metric}, lists:reverse(AccL), AccIn),
    Acc#facc{
      size = mmath_bin:length(V),
      hacc = AccOut,
      lacc = [{Time, V}]};
fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric,
                   size = Size,
                   lacc = AccL}) ->
    Acc#facc{
      size = Size + mmath_bin:length(V),
      lacc = [{Time, V} | AccL]}.

handle_call({fold, Fun, Acc0}, _From,
            State = #state{fold_size = FoldSize, partition = Partition}) ->

    DataDir = application:get_env(riak_core, platform_data_dir, "data"),
    PartitionDir = [DataDir, $/,  integer_to_list(Partition)],
    case file:list_dir(PartitionDir) of
        {ok, Buckets} ->
            AsyncWork =<commit_msg>Combining multiple handoffs in one.<commit_after>%%%-------------------------------------------------------------------
%%% @author <NAME> <<EMAIL>>
%%% @copyright (C) 2014, <NAME>
%%% @doc
%%%
%%% @end
%%% Created : 24 Sep 2014 by <NAME> <<EMAIL>>
%%%-------------------------------------------------------------------
-module(metric_io).

-behaviour(gen_server).

%% API
-export([start_link/1,
         empty/1, fold/3, delete/1, close/1,
         buckets/1, metrics/2, delete/2,
         read/7, write/5, write/6]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).

-define(SERVER, ?MODULE).
-define(WEEK, 604800). %% Seconds in a week.
-define(MAX_Q_LEN, 20).

-record(state, {
          partition,
          node,
          mstore=gb_trees:empty(),
          dir,
          fold_size
         }).

%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Starts the server
%%
%% @spec start_link() -> {ok, Pid} | ignore | {error, Error}
%% @end
%%--------------------------------------------------------------------
start_link(Partition) ->
    gen_server:start_link(?MODULE, [Partition], []).

write(Pid, Bucket, Metric, Time, Value) ->
    write(Pid, Bucket, Metric, Time, Value, ?MAX_Q_LEN).

write(Pid, Bucket, Metric, Time, Value, MaxLen) ->
    case erlang:process_info(Pid, message_queue_len) of
        {message_queue_len, N} when N > MaxLen ->
            swrite(Pid, Bucket, Metric, Time, Value);
        _ ->
            gen_server:cast(Pid, {write, Bucket, Metric, Time, Value})
    end.

swrite(Pid, Bucket, Metric, Time, Value) ->
    gen_server:call(Pid, {write, Bucket, Metric, Time, Value}).

read(Pid, Bucket, Metric, Time, Count, ReqID, Sender) ->
    gen_server:cast(Pid, {read, Bucket, Metric, Time, Count, ReqID, Sender}).

buckets(Pid) ->
    gen_server:call(Pid, buckets).

metrics(Pid, Bucket) ->
    gen_server:call(Pid, {metrics, Bucket}).

fold(Pid, Fun, Acc0) ->
    gen_server:call(Pid, {fold, Fun, Acc0}).

empty(Pid) ->
    gen_server:call(Pid, empty).

delete(Pid) ->
    gen_server:call(Pid, delete).

close(Pid) ->
    gen_server:call(Pid, close).

delete(Pid, Bucket) ->
    gen_server:call(Pid, {delete, Bucket}).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the server
%%
%% @spec init(Args) -> {ok, State} |
%%                     {ok, State, Timeout} |
%%                     ignore |
%%                     {stop, Reason}
%% @end
%%--------------------------------------------------------------------
init([Partition]) ->
    process_flag(trap_exit, true),
    DataDir = case application:get_env(riak_core, platform_data_dir) of
                  {ok, DD} ->
                      DD;
                  _ ->
                      "data"
              end,
    FoldSize = case application:get_env(metric_vnode, handoff_chunk) of
                   {ok, FS} ->
                       FS;
                   _ ->
                       10*1024
               end,
    PartitionDir = [DataDir, $/,  integer_to_list(Partition)],

    {ok, #state{ partition = Partition,
                 node = node(),
                 dir = PartitionDir,
                 fold_size = FoldSize
               }}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%%
%% @spec handle_call(Request, From, State) ->
%%                                   {reply, Reply, State} |
%%                                   {reply, Reply, State, Timeout} |
%%                                   {noreply, State} |
%%                                   {noreply, State, Timeout} |
%%                                   {stop, Reason, Reply, State} |
%%                                   {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------
%% fold_fun(Fun, Bucket) ->
%%     fun(Metric, Time, V,
%%         {[{_Metric2, _}|_] = AccL, AccIn})
%%           when Metric =/= _Metric2 ->
%%             AccOut = Fun({Bucket, {Metric, Time}}, V, AccIn)
%%     end.

-record(facc,
        {
          metric,
          size = 0,
          hacc,
          lacc = [],
          bucket,
          acc_fun,
          last
        }).

-define(FOLD_SIZE, 50).
-define(MAX_DELTA, 5).

fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric2,
                   lacc = []}) when
      Metric =/= Metric2 ->
    Acc#facc{
      metric = Metric,
      size = mmath_bin:length(V),
      lacc = [{Time, V}]};
fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric2,
                   bucket = Bucket,
                   lacc = AccL,
                   acc_fun = Fun,
                   hacc = AccIn}) when
      Metric =/= Metric2 ->
    AccOut = Fun({Bucket, Metric2}, lists:reverse(AccL), AccIn),
    Acc#facc{
      metric = Metric,
      size = mmath_bin:length(V),
      hacc = AccOut,
      lacc = [{Time, V}]};
fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric,
                   bucket = Bucket,
                   size = _Size,
                   lacc = AccL,
                   acc_fun = Fun,
                   hacc = AccIn}) when
      _Size > ?FOLD_SIZE ->
    AccOut = Fun({Bucket, Metric}, lists:reverse(AccL), AccIn),
    Size = mmath_bin:length(V),
    Acc#facc{
      size = Size,
      hacc = AccOut,
      last = Time + Size,
      lacc = [{Time, V}]};

fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric,
                   size = Size,
                   last = Last,
                   lacc = [{T0, AccE} | AccL]}) when
      Time - Last =< ?MAX_DELTA ->
    Delta = Time - Last,
    ThisSize = mmath_bin:length(V),
    AccV = <<AccE/binary, (mmath_bin:empty(Delta))/binary, V/binary>>,
    Acc#facc{
      size = Size + ThisSize + Delta,
      last = Last + Delta + ThisSize,
      lacc = [{T0, AccV} | AccL]};

fold_fun(Metric, Time, V,
         Acc =
             #facc{metric = Metric,
                   size = Size,
                   lacc = AccL}) ->
    ThisSize = mmath_bin:length(V),
    Acc#facc{
      size = Size + ThisSize,
      last = Time + ThisSize,
      lacc = [{Time, V} | AccL]}.

handle_call({fold, Fun, Acc0}, _From,
            State = #state{fold_size = FoldSize, partition = Partition}) ->

    DataDir = application:get_env(riak_core, platform_data_dir, "data"),
    PartitionDir = [DataDir, $/,  integer_to_list(Partition)],
    case file:list_dir(PartitionDir) of
        {ok, Buckets} ->
            AsyncWork =
{'fraction_non_alphanumeric': 0.25656901238296587, 'fraction_numerical': 0.00838115372999094, 'mean_word_length': 1.4877911344853494, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 84, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4494150', 'n_tokens_mistral': 4102, 'n_tokens_neox': 3894, 'n_words': 851}