High concurrency CouchDB Metrics Update Benchmark

statbench.erl

-module(statbench).

-export([
    go_counter/2,
    go_gauge/2,
    go_hist/2
]).

go_counter(N, X) ->
    go(N, X, fun upstat_counter/1).

go_gauge(N, X) ->
    go(N, X, fun upstat_gauge/1).

go_hist(N, X) ->
    go(N, X, fun upstat_hist/1).

go(N, X, Fun) when is_function(Fun, 1) ->
    T0 = erlang:monotonic_time(),
    Workers = [spawn_monitor(fun() ->
        rand:seed(default, os:timestamp()),
        Fun(X)
    end) || _ <- lists:seq(1, N)],
    {_Pids, Refs} = lists:unzip(Workers),
    WorkersSet = sets:from_list(Refs, [{version,2}]),
    ok = wait_workers(WorkersSet),
    Dt = erlang:monotonic_time() - T0,
    erlang:convert_time_unit(Dt, native, millisecond).

upstat_hist(0) ->
    ok;
upstat_hist(Times) ->
    couch_stats:update_histogram([fsync, time], rand:uniform(100000)),
    upstat_hist(Times - 1).

upstat_gauge(0) ->
    ok;
upstat_gauge(Times) ->
    couch_stats:update_gauge([couch_replicator, jobs, total], rand:uniform(1000)),
    upstat_hist(Times - 1).

upstat_counter(0) ->
    ok;
upstat_counter(Times) ->
    couch_stats:increment_counter([fsync, count], rand:uniform(100000)),
    upstat_counter(Times - 1).

wait_workers(Workers) when is_map(Workers) ->
    case sets:size(Workers) of
        0 ->
            ok;
        _ ->
            receive
                {'DOWN', Ref, process, _, normal} ->
                    Workers1 = sets:del_element(Ref, Workers),
                    wait_workers(Workers1);
                {'DOWN', Ref, process, _, Err} ->
                    Workers1 = sets:del_element(Ref, Workers),
                    io:format("~n worker ~p crashed: ~p~n", [Ref, Err]),
                    wait_workers(Workers1)
            end
    end.