<commit_before>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2007-2012 Basho Technologies, Inc.  All Rights Reserved.
%%
%% -------------------------------------------------------------------

%%
%% Please note that the in order
%% to run basho_bench_driver_moss
%% `auth_bypass` must be set to true
%% on the riak_moss app.config

-module(basho_bench_driver_moss).

-define(BLOCK, (1024*1024)).
-define(VERYLONG_TIMEOUT, (300*1000)).
-define(REPORTFUN_APPKEY, report_fun).
-define(START_KEY, {?MODULE, start}).
-define(BLOCKSIZE_KEY, {?MODULE, block_size}).

-export([new/1,
         run/4]).
-export([bigfile_valgen/2]).

-record(state, { client_id,
                 hosts,
                 bucket,
                 report_fun,
                 working_op,
                 req_id
               }).

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

-spec new(integer()) -> {ok, term()}.
new(ID) ->
    %% For large numbers of load generators, it's a good idea to
    %% stagger things out a little bit to avoid thundering herd
    %% problems at the server side.
    timer:sleep(basho_bench_config:get(moss_new_delay, 10) * (ID - 1)),

    application:start(ibrowse),
    application:start(crypto),

    %% The IP, port and path we'll be testing

    %% TODO:
    %% We'll start with a single
    %% IP for now
    Ip  = basho_bench_config:get(moss_raw_ip, "127.0.0.1"),
    Port = basho_bench_config:get(moss_raw_port, 8080),
    Disconnect = basho_bench_config:get(moss_disconnect_frequency, infinity),
    erlang:put(disconnect_freq, Disconnect),

    %% Get our measurement units: op/sec, Byte/sec, KByte/sec, MByte/sec
    {RF_name, ReportFun} =
        %% We need to be really careful with these custom units things.
        %% Use floats for everything.
        case (catch basho_bench_config:get(moss_measurement_units)) of
            N = byte_sec  -> {N,      fun(X) -> X / 1 end};
            N = kbyte_sec -> {N,      fun(X) -> X / 1024 end};
            N = mbyte_sec -> {N,      fun(X) -> X / (1024 * 1024) end};
            _             -> {op_sec, fun(_) -> 1.0 end}
        end,
    if ID == 1 ->
            lager:log(info, self(), "Reporting factor = ~p\n", [RF_name]);
       true ->
            ok
    end,
    %% For ops other than get, basho_bench doesn't give us an explicit
    %% way to pass ReportFun to the necessary place in a purely
    %% functional way.
    application:set_env(?MODULE, ?REPORTFUN_APPKEY, ReportFun),

    OpsList = basho_bench_config:get(operations, []),
    case RF_name /= op_sec andalso
         lists:keymember(delete, 1, OpsList) andalso
         length(OpsList) > 1 of
        true ->
            lager:log(
              warn, self(),
              "Mixing delete and non-delete operations together with "
              "~p measurements unit can yield nonsense results!\n\n",
              [RF_name]);
        false ->
            ok
    end,

    Hosts = [{Ip, Port}],

    {ok, #state{client_id=ID, hosts=Hosts,
                bucket = basho_bench_config:get(moss_bucket, "test"),
                report_fun = ReportFun}}.

%% This module does some crazy stuff, but it's there for a reason.
%% The reason is that basho_bench is expecting the run() function to
%% do something that takes a few/several seconds at most.  It is not
%% expecting run() to finish after minutes/hours/days of runtime,
%% which is quite possible when testing 5GB files or larger.
%%
%% So, we adopt a strategy where run() will return after ?BLOCK of
%% data received/sent.  That will allow R to create graphs of the
%% latency of ?BLOCK chunk retrievals.  We will do our own throughput
%% reporting to allow R to graph total throughput in terms of
%% KByte/sec or MByte/sec -- we do it via a kludge, but at least it
%% works.
%%
%% This scheme will also allow b_b's built-in rate-limiting scheme to
%% work, though on a granularity of ?BLOCK increments per second,
%% which is probably still too sucky and may need finer control.

-spec run(atom(), fun(), fun(), term()) -> {{ok, number()}, term()} | {error, term(), term()}.

run(get, KeyGen, _ValueGen, #state{working_op = undefined,
                                   bucket = Bucket} = State) ->
    {NextHost, S2} = next_host(State),
    {Host, Port} = NextHost,
    Key = KeyGen(),
    Url = url(Host, Port, Bucket, Key),
    case do_get_first_unit({Host, Port}, Url, [], S2) of
        {ok, S3} ->
            {{silent, ok}, S3};
        Else ->
            Else
    end;
run(_, _KeyGen, _ValueGen, #state{working_op = get} = State) ->
    case do_get_loop(State) of
        {ok, S2} ->
            {{silent, ok}, S2};
        Else ->
            Else
    end;
run(Op, KeyGen, ValueGen, State) ->
    run2(Op, KeyGen, ValueGen, State).

run2(insert, KeyGen, ValueGen, #state{bucket = Bucket} = State) ->
    {NextHost, S2} = next_host(State),
    ConnInfo = NextHost,
    case insert(KeyGen, ValueGen, ConnInfo, Bucket) of
        ok ->
            {{silent, ok}, S2};
        {error, Reason} ->
            {error, Reason, S2}
    end;
run2(delete, KeyGen, _ValueGen, #state{bucket = Bucket} = State) ->
    {NextHost, S2} = next_host(State),
    {Host, Port} = NextHost,
    Key = KeyGen(),
    Url = url(Host, Port, Bucket, Key),
    case do_delete({Host, Port}, Url, []) of
        ok ->
            {ok, S2};
        {error, Reason} ->
            {error, Reason, S2}
    end;
run2(Op, _KeyGen, _ValueGen, State) ->
    {error, {unknown_op, Op}, State}.

%% The ibrowse client can take a {fun-arity-1, Arg::term()} tuple for
%% a value and use the fun to generate body bytes on-the-fly.
%%
%% Proplist:
%%
%% file_size -- Size of the S3 file that we'll be uploading to the
%%              S3/S3-like service.  Right now, this is a constant value.
%%              If a variable file size is needed, we'll have to add code
%%              to do it.  No default: this property *must* be defined.
%% max_rate_per_chunk -- Maximum rate per second that an
%%                       'ibrowse_chunk_size' sized piece of data will
%%                       be transmitted.
%%                       NOTE: This rate is global/shared across all
%%                             worker processes.
%%                       Default = no limit.
%% ibrowse_chunk_size -- Amount of data that the ibrowse data chunk
%%                       creation function will create.  Default = 1MByte.

bigfile_valgen(Id, Props) ->
    if Id == 1 ->
            lager:log(info, "~s value gen props: ~p\n", [?MODULE, Props]);
       true ->
            ok
    end,
    FileSize = proplists:get_value(file_size, Props, file_size_undefined),
    ChunkSize = proplists:get_value(ibrowse_chunk_size, Props, 1024*1024),
    MaxRate = proplists:get_value(max_rate_per_chunk, Props, 9999999) /
        basho_bench_config:get(concurrent),
    MaxRateSleepUS = trunc((1 / MaxRate) * 1000000),

    Chunk = list_to_binary(lists:duplicate(ChunkSize, 42)),
    {fun(get_content_length) ->
             FileSize;
        (Start) when Start < FileSize ->
             case erlang:get(?START_KEY) of
                 undefined ->
                     ok; %% This is our first call, so no timing to report.
                 StartT ->
                     DiffT = timer:now_diff(now(), StartT),<commit_msg>Fix minor lager formatting error in basho_bench_driver_moss.erl<commit_after>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2007-2012 Basho Technologies, Inc.  All Rights Reserved.
%%
%% -------------------------------------------------------------------

%%
%% Please note that the in order
%% to run basho_bench_driver_moss
%% `auth_bypass` must be set to true
%% on the riak_moss app.config

-module(basho_bench_driver_moss).

-define(BLOCK, (1024*1024)).
-define(VERYLONG_TIMEOUT, (300*1000)).
-define(REPORTFUN_APPKEY, report_fun).
-define(START_KEY, {?MODULE, start}).
-define(BLOCKSIZE_KEY, {?MODULE, block_size}).

-export([new/1,
         run/4]).
-export([bigfile_valgen/2]).

-record(state, { client_id,
                 hosts,
                 bucket,
                 report_fun,
                 working_op,
                 req_id
               }).

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

-spec new(integer()) -> {ok, term()}.
new(ID) ->
    %% For large numbers of load generators, it's a good idea to
    %% stagger things out a little bit to avoid thundering herd
    %% problems at the server side.
    timer:sleep(basho_bench_config:get(moss_new_delay, 10) * (ID - 1)),

    application:start(ibrowse),
    application:start(crypto),

    %% The IP, port and path we'll be testing

    %% TODO:
    %% We'll start with a single
    %% IP for now
    Ip  = basho_bench_config:get(moss_raw_ip, "127.0.0.1"),
    Port = basho_bench_config:get(moss_raw_port, 8080),
    Disconnect = basho_bench_config:get(moss_disconnect_frequency, infinity),
    erlang:put(disconnect_freq, Disconnect),

    %% Get our measurement units: op/sec, Byte/sec, KByte/sec, MByte/sec
    {RF_name, ReportFun} =
        %% We need to be really careful with these custom units things.
        %% Use floats for everything.
        case (catch basho_bench_config:get(moss_measurement_units)) of
            N = byte_sec  -> {N,      fun(X) -> X / 1 end};
            N = kbyte_sec -> {N,      fun(X) -> X / 1024 end};
            N = mbyte_sec -> {N,      fun(X) -> X / (1024 * 1024) end};
            _             -> {op_sec, fun(_) -> 1.0 end}
        end,
    if ID == 1 ->
            lager:log(info, self(), "Reporting factor = ~p\n", [RF_name]);
       true ->
            ok
    end,
    %% For ops other than get, basho_bench doesn't give us an explicit
    %% way to pass ReportFun to the necessary place in a purely
    %% functional way.
    application:set_env(?MODULE, ?REPORTFUN_APPKEY, ReportFun),

    OpsList = basho_bench_config:get(operations, []),
    case RF_name /= op_sec andalso
         lists:keymember(delete, 1, OpsList) andalso
         length(OpsList) > 1 of
        true ->
            lager:log(
              warn, self(),
              "Mixing delete and non-delete operations together with "
              "~p measurements unit can yield nonsense results!\n\n",
              [RF_name]);
        false ->
            ok
    end,

    Hosts = [{Ip, Port}],

    {ok, #state{client_id=ID, hosts=Hosts,
                bucket = basho_bench_config:get(moss_bucket, "test"),
                report_fun = ReportFun}}.

%% This module does some crazy stuff, but it's there for a reason.
%% The reason is that basho_bench is expecting the run() function to
%% do something that takes a few/several seconds at most.  It is not
%% expecting run() to finish after minutes/hours/days of runtime,
%% which is quite possible when testing 5GB files or larger.
%%
%% So, we adopt a strategy where run() will return after ?BLOCK of
%% data received/sent.  That will allow R to create graphs of the
%% latency of ?BLOCK chunk retrievals.  We will do our own throughput
%% reporting to allow R to graph total throughput in terms of
%% KByte/sec or MByte/sec -- we do it via a kludge, but at least it
%% works.
%%
%% This scheme will also allow b_b's built-in rate-limiting scheme to
%% work, though on a granularity of ?BLOCK increments per second,
%% which is probably still too sucky and may need finer control.

-spec run(atom(), fun(), fun(), term()) -> {{ok, number()}, term()} | {error, term(), term()}.

run(get, KeyGen, _ValueGen, #state{working_op = undefined,
                                   bucket = Bucket} = State) ->
    {NextHost, S2} = next_host(State),
    {Host, Port} = NextHost,
    Key = KeyGen(),
    Url = url(Host, Port, Bucket, Key),
    case do_get_first_unit({Host, Port}, Url, [], S2) of
        {ok, S3} ->
            {{silent, ok}, S3};
        Else ->
            Else
    end;
run(_, _KeyGen, _ValueGen, #state{working_op = get} = State) ->
    case do_get_loop(State) of
        {ok, S2} ->
            {{silent, ok}, S2};
        Else ->
            Else
    end;
run(Op, KeyGen, ValueGen, State) ->
    run2(Op, KeyGen, ValueGen, State).

run2(insert, KeyGen, ValueGen, #state{bucket = Bucket} = State) ->
    {NextHost, S2} = next_host(State),
    ConnInfo = NextHost,
    case insert(KeyGen, ValueGen, ConnInfo, Bucket) of
        ok ->
            {{silent, ok}, S2};
        {error, Reason} ->
            {error, Reason, S2}
    end;
run2(delete, KeyGen, _ValueGen, #state{bucket = Bucket} = State) ->
    {NextHost, S2} = next_host(State),
    {Host, Port} = NextHost,
    Key = KeyGen(),
    Url = url(Host, Port, Bucket, Key),
    case do_delete({Host, Port}, Url, []) of
        ok ->
            {ok, S2};
        {error, Reason} ->
            {error, Reason, S2}
    end;
run2(Op, _KeyGen, _ValueGen, State) ->
    {error, {unknown_op, Op}, State}.

%% The ibrowse client can take a {fun-arity-1, Arg::term()} tuple for
%% a value and use the fun to generate body bytes on-the-fly.
%%
%% Proplist:
%%
%% file_size -- Size of the S3 file that we'll be uploading to the
%%              S3/S3-like service.  Right now, this is a constant value.
%%              If a variable file size is needed, we'll have to add code
%%              to do it.  No default: this property *must* be defined.
%% max_rate_per_chunk -- Maximum rate per second that an
%%                       'ibrowse_chunk_size' sized piece of data will
%%                       be transmitted.
%%                       NOTE: This rate is global/shared across all
%%                             worker processes.
%%                       Default = no limit.
%% ibrowse_chunk_size -- Amount of data that the ibrowse data chunk
%%                       creation function will create.  Default = 1MByte.

bigfile_valgen(Id, Props) ->
    if Id == 1 ->
            lager:log(info, "~p value gen props: ~p\n", [?MODULE, Props]);
       true ->
            ok
    end,
    FileSize = proplists:get_value(file_size, Props, file_size_undefined),
    ChunkSize = proplists:get_value(ibrowse_chunk_size, Props, 1024*1024),
    MaxRate = proplists:get_value(max_rate_per_chunk, Props, 9999999) /
        basho_bench_config:get(concurrent),
    MaxRateSleepUS = trunc((1 / MaxRate) * 1000000),

    Chunk = list_to_binary(lists:duplicate(ChunkSize, 42)),
    {fun(get_content_length) ->
             FileSize;
        (Start) when Start < FileSize ->
             case erlang:get(?START_KEY) of
                 undefined ->
                     ok; %% This is our first call, so no timing to report.
                 StartT ->
                     DiffT = timer:now_diff(now(), StartT),
{'fraction_non_alphanumeric': 0.15553910351325886, 'fraction_numerical': 0.014133799973078476, 'mean_word_length': 2.220416124837451, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 77, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1535799', 'n_tokens_mistral': 4785, 'n_tokens_neox': 4451, 'n_words': 1423}