Erlang RSS Memory

.gitignore

*.beam
*.csv

dump_memory_data.erl

-module(dump_memory_data).
% code:add_patha("/home/lbakken/Development/ferd/recon/ebin"), code:add_patha("/home/lbakken/Development/gists/erlang-rss").
% m(recon), m(dump_memory_data).
% f(Pids), Pids = dump_memory_data:start().

-export([start/0, start/1, stop/1,
         collect_data_loop/1, collect_data_loop/4,
         get_data_loop/1,
         get_memory/4]).

-record(memory_data, {erlang = 0,
                      recon = 0,
                      ps = 0,
                      statm = 0,
                      vmm = 0,
                      count = 0}).

-record(gdl_state, {cpid = undefined,
                    type = undefined,
                    os_type = undefined,
                    os_pid = undefined,
                    pagesize = undefined}).

start() ->
    start("dump_memory_data.csv").

start(File) ->
    prepare_file(File),
    CPid = spawn_link(dump_memory_data, collect_data_loop, [File]),

    OsType = os:type(),
    OsPid = os:getpid(),

    S0 = #gdl_state{cpid=CPid, type=erlang, os_type=OsType, os_pid=OsPid},
    P0 = spawn_link(dump_memory_data, get_data_loop, [S0]),

    S1 = #gdl_state{cpid=CPid, type=recon, os_type=OsType, os_pid=OsPid},
    P1 = spawn_link(dump_memory_data, get_data_loop, [S1]),

    S2 = #gdl_state{cpid=CPid, type=ps, os_type=OsType, os_pid=OsPid},
    P2 = spawn_link(dump_memory_data, get_data_loop, [S2]),

    S3 = #gdl_state{cpid=CPid, type=statm, os_type=OsType, os_pid=OsPid},
    P3 = spawn_link(dump_memory_data, get_data_loop, [S3]),

    S4 = #gdl_state{cpid=CPid, type=vmm, os_type=OsType, os_pid=OsPid},
    P4 = spawn_link(dump_memory_data, get_data_loop, [S4]),

    Pids = [P0, P1, P2, P3, P4],
    CPid ! {start, Pids},
    [CPid, P0, P1, P2, P3, P4].

stop(Pids) ->
    [ P ! stop || P <- Pids ].

prepare_file(File) ->
    file:write_file(File, <<"idx,erlang,recon,ps-rss,statm-rss,vmm\n">>, [write]).

exit_normal() ->
    io:format("[INFO] process ~p exiting normally~n", [self()]),
    exit(normal).

log_unknown_message(Unknown) ->
    io:format(standard_error, "[ERROR] unknown msg: ~p~n", [Unknown]).

one_second_from_now() ->
    erlang:monotonic_time(millisecond) +
        erlang:convert_time_unit(1, second, millisecond).

start_timers(Pids) ->
    start_timers(one_second_from_now(), Pids).

start_timers(_, []) ->
    ok;
start_timers(Time, [Pid|Rest]) ->
    erlang:send_after(Time, Pid, get_data, [{abs, true}]),
    start_timers(Time, Rest).

collect_data_loop(File) ->
    %% io:format("[DEBUG] collect_data_loop/1 ~p~n", [File]),
    receive
        {start, Pids} ->
            ok = start_timers(Pids),
            collect_data_loop(File, 0, Pids, #memory_data{});
        stop ->
            exit_normal();
        U0 ->
            log_unknown_message(U0)
    end.

collect_data_loop(File, Idx, Pids, Data0=#memory_data{count=C}) ->
    %% io:format("[DEBUG] collect_data_loop/4 ~p ~p ~p ~p~n", [File, Idx, Pids, Data0]),
    Data1 = receive
        {erlang, Mem0} ->
            Data0#memory_data{erlang=Mem0, count=C+1};
        {recon, Mem1} ->
            Data0#memory_data{recon=Mem1, count=C+1};
        {ps, Mem2} ->
            Data0#memory_data{ps=Mem2, count=C+1};
        {statm, Mem3} ->
            Data0#memory_data{statm=Mem3, count=C+1};
        {vmm, Mem4} ->
            Data0#memory_data{vmm=Mem4, count=C+1};
        stop ->
            exit_normal();
        U0 ->
            log_unknown_message(U0)
    end,
    maybe_write_data(File, Idx, Pids, Data1).

maybe_write_data(File, Idx, Pids, Data=#memory_data{count=Count}) ->
    FieldCount = record_info(size, memory_data) - 2,
    maybe_write_data(File, Idx, Pids, Data, Count =:= FieldCount).

maybe_write_data(File, Idx, Pids, Data, true) ->
    write_data(File, Idx, Data),
    ok = start_timers(Pids),
    collect_data_loop(File, Idx + 1, Pids, #memory_data{});
maybe_write_data(File, Idx, Pids, Data, false) ->
    collect_data_loop(File, Idx, Pids, Data).

get_data_loop(State=#gdl_state{os_type={unix,linux}, pagesize=undefined}) ->
    CmdOutput = os:cmd("getconf PAGESIZE"),
    PageSize = case re:run(CmdOutput, "^[0-9]+", [{capture, first, list}]) of
                   {match, [Match]} -> list_to_integer(Match);
                   _ ->
                       io:format(standard_error, "[ERROR] unexpected getconf output: ~p~n", [CmdOutput]),
                       4096
               end,
    get_data_loop(State#gdl_state{pagesize=PageSize});
get_data_loop(State=#gdl_state{cpid=CPid, type=Type, os_type=OsType, os_pid=OsPid, pagesize=PageSize}) ->
    receive
        get_data ->
            % io:format("[DEBUG] get_data ~p ~p~n", [Type, os:timestamp()]),
            {ok, Data} = get_memory(Type, OsType, OsPid, PageSize),
            CPid ! {Type, Data},
            get_data_loop(State);
        stop ->
            exit_normal();
        Unknown ->
            log_unknown_message(Unknown),
            get_data_loop(State)
    end.

write_data(File, Idx, Data) ->
    %% io:format("[DEBUG] write_data/3: ~p ~p ~p~n", [File, Idx, Data]),
    % idx
    S0 = integer_to_list(Idx),
    % erlang
    S1 = integer_to_list(Data#memory_data.erlang),
    % recon
    S2 = integer_to_list(Data#memory_data.recon),
    % ps
    S3 = integer_to_list(Data#memory_data.ps),
    % statm
    S4 = integer_to_list(Data#memory_data.statm),
    % vmm
    S5 = integer_to_list(Data#memory_data.vmm),
    Str = string:join([S0, S1, S2, S3, S4, S5], ",") ++ "\n",
    file:write_file(File, list_to_binary(Str), [write, append]).

get_memory(erlang, _, _, _) ->
    {ok, erlang:memory(total)};
get_memory(recon, _, _, _) ->
    {ok, recon_alloc:memory(allocated)};
get_memory(vmm, _, _, _) ->
    {ok, vm_memory_monitor:get_process_memory()};
get_memory(ps, {win32, _}, OsPid, _) ->
    Cmd = "wmic process where processid=" ++ OsPid ++ " get WorkingSetSize /value 2>&1",
    CmdOutput = os:cmd(Cmd),
    %% Memory usage is displayed in bytes
    case re:run(CmdOutput, "WorkingSetSize=([0-9]+)", [{capture, all_but_first, binary}]) of
        {match, [Match]} ->
            {ok, binary_to_integer(Match)};
        _ ->
            {error, {unexpected_output_from_command, Cmd, CmdOutput}}
    end;
get_memory(ps, _, OsPid, _) ->
    Cmd = "ps -p " ++ OsPid ++ " -o rss=",
    CmdOutput = os:cmd(Cmd),
    case re:run(CmdOutput, "[0-9]+", [{capture, first, list}]) of
        {match, [Match]} ->
            {ok, list_to_integer(Match) * 1024};
        _ ->
            {error, {unexpected_output_from_command, Cmd, CmdOutput}}
    end;
get_memory(statm, {unix, linux}, OsPid, PageSize) ->
    ProcFile = io_lib:format("/proc/~s/statm", [OsPid]),
    {ok, Data} = read_proc_file(ProcFile),
    [_|[RssPagesStr|_]] = string:split(Data, " ", all),
    Mem = list_to_integer(RssPagesStr) * PageSize,
    {ok, Mem};
get_memory(statm, _, _, _) ->
    {ok, 0}.

read_proc_file(ProcFile) ->
    {ok, F} = file:open(ProcFile, [read, raw]),
    {ok, Data} = file:read_line(F),
    ok = file:close(F),
    {ok, Data}.

erlang-rss.config

% vim:ft=erlang:

[
  {rabbit, [
      {loopback_users, []},
      {vm_memory_calculation_strategy, allocated}
    ]}
].

get-rss.escript

#!/usr/bin/env escript
%% -*- erlang -*-
%%! -smp enable -sname get_rss
main([String]) ->
    try
        output_rss(get_rss(String))
    catch
        _:_ ->
            usage()
    end;
main([]) ->
        output_rss(get_rss(os:getpid()));
main(_) ->
    usage().

usage() ->
    io:format("get-rss [pid]\n"),
    halt(1).

output_rss({error, Reason}) ->
    io:format("ERROR: ~p~n", [Reason]);
output_rss(Rss) ->
    io:format("RSS: ~p KiB~n", [Rss]).

get_rss(OsPid) ->
    ProcFile = filename:join(["/proc", OsPid, "smaps"]),
    {ok, Device} = file:open(ProcFile, [read]),
    Rss = 0,
    handle_read_line(file:read_line(Device), Device, Rss).

handle_read_line(eof, Device, Rss) ->
    file:close(Device),
    Rss;
handle_read_line({error, _}=Err, Device, _Rss) ->
    file:close(Device),
    Err;
handle_read_line({ok, Data}, Device, Rss) ->
    handle_string_str(string:str(Data, "Rss:"), Data, Device, Rss).

handle_string_str(1, Data, Device, Rss0) ->
    Value = list_to_integer(string:sub_word(Data, 2)),
    Rss = Rss0 + Value,
    handle_read_line(file:read_line(Device), Device, Rss);
handle_string_str(_, _Data, Device, Rss) ->
    handle_read_line(file:read_line(Device), Device, Rss).

vm_memory_allocators.erl

-module(vm_memory_allocators).

-export([total_size/0]).

instance_size(Items, Init) ->
    lists:foldl(fun
            ({_, [{blocks_size,BS,_,_}, {carriers_size,CS,_,_}]}, {Used, Total}) ->
                {Used+BS, Total+CS};
            ({_,[{blocks_size,BS},{carriers_size,CS}]}, {Used, Total}) ->
                {Used+BS, Total+CS};
            (_, Acc) -> Acc
        end, Init, Items).

allocator_size(Instances, Init) ->
    lists:foldl(
        fun({instance, _No, Items}, Acc) ->
            instance_size(Items, Acc)
        end,
        Init,
        Instances).

total_size() ->
    AllocNames = erlang:system_info(alloc_util_allocators),
    AllAllocators = erlang:system_info({allocator_sizes, AllocNames}),
    lists:foldl(
        fun({_, Instances}, Acc) -> allocator_size(Instances, Acc) end,
        {0, 0},
        AllAllocators).