Go to an entirely supervisor-based way of counting process memory. This has the disadvantage that we need to combine channel and connection memory, but we are far more inclusive, accounting for limiters, writers, supervisors etc. In particular this helps a lot when we have thousands of queues since the queue_sup ends up taking a lot of memory in its own right. This seems to bloat rabbit_mgmt_external_stats less (although still some) in the 100k queue case too. Also measure memory use of mgmt_db and msg_store processes and add them appropriately.

1 files changed, 39 insertions, 29 deletions

diff --git a/src/rabbit.erl b/src/rabbit.erl
index f9cb0d8c..26e4f1f4 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -747,13 +747,15 @@ start_fhc() ->
 
 %% Like erlang:memory(), but with awareness of rabbit-y things
 memory() ->
-    QPids = lists:append([pids(Q) || Q <- rabbit_amqqueue:list()]),
-    Conns = sum_proc_memory(rabbit_networking:connections_local()),
-    Chs = sum_proc_memory(rabbit_channel:list_local()),
-    Qs = sum_proc_memory(QPids),
+    ConnChs = sup_memory(rabbit_tcp_client_sup),
+    Qs = sup_memory(rabbit_amqqueue_sup) +
+        sup_memory(rabbit_mirror_queue_slave_sup),
     Mnesia = mnesia_memory(),
-    MsgIndex = ets_memory(rabbit_msg_store_ets_index),
-    MgmtDB = ets_memory(rabbit_mgmt_db),
+    MsgIndexETS = ets_memory(rabbit_msg_store_ets_index),
+    MsgIndexProc = pid_memory(msg_store_transient) +
+        pid_memory(msg_store_persistent),
+    MgmtDbETS = ets_memory(rabbit_mgmt_db),
+    MgmtDbProc = sup_memory(rabbit_mgmt_sup),
     [{total,     Total},
      {processes, Processes},
      {ets,       ETS},
@@ -762,29 +764,37 @@ memory() ->
      {code,      Code},
      {system,    System}] =
         erlang:memory([total, processes, ets, atom, binary, code, system]),
-    [{total,            Total},
-     {connection_procs, Conns},
-     {channel_procs,    Chs},
-     {queue_procs,      Qs},
-     {other_proc,       Processes - Conns - Chs - Qs},
-     {mnesia,           Mnesia},
-     {mgmt_db,          MgmtDB},
-     {msg_index,        MsgIndex},
-     {other_ets,        ETS - Mnesia - MsgIndex - MgmtDB},
-     {binary,           Bin},
-     {code,             Code},
-     {atom,             Atom},
-     {other_system,     System - ETS - Atom - Bin - Code}].
-
-sum_proc_memory(Pids) ->
-    lists:sum([Mem || P <- Pids, {memory, Mem} <- [process_info(P, memory)]]).
-
-pids(#amqqueue{pid = Pid, slave_pids = undefined}) ->
-    local_pids([Pid]);
-pids(#amqqueue{pid = Pid, slave_pids = SPids}) ->
-    local_pids([Pid | SPids]).
-
-local_pids(Pids) -> [Pid || Pid <- Pids, node(Pid) =:= node()].
+    [{total,                    Total},
+     {connection_channel_procs, ConnChs},
+     {queue_procs,              Qs},
+     {other_proc,               Processes - ConnChs - Qs - MsgIndexProc -
+          MgmtDbProc},
+     {mnesia,                   Mnesia},
+     {mgmt_db,                  MgmtDbETS + MgmtDbProc},
+     {msg_index,                MsgIndexETS + MsgIndexProc},
+     {other_ets,                ETS - Mnesia - MsgIndexETS - MgmtDbETS},
+     {binary,                   Bin},
+     {code,                     Code},
+     {atom,                     Atom},
+     {other_system,             System - ETS - Atom - Bin - Code}].
+
+sup_memory(Sup) ->
+    lists:sum([child_memory(P, T) || {_, P, T, _} <- sup_children(Sup)]) +
+        pid_memory(Sup).
+
+sup_children(Sup) ->
+    rabbit_misc:with_exit_handler(
+      rabbit_misc:const([]), fun () -> supervisor:which_children(Sup) end).
+
+pid_memory(Pid)  when is_pid(Pid)   -> element(2, process_info(Pid, memory));
+pid_memory(Name) when is_atom(Name) -> case whereis(Name) of
+                                           P when is_pid(P) -> pid_memory(P);
+                                           _                -> 0
+                                       end.
+
+child_memory(Pid, worker)     when is_pid (Pid) -> pid_memory(Pid);
+child_memory(Pid, supervisor) when is_pid (Pid) -> sup_memory(Pid);
+child_memory(_, _)                              -> 0.
 
 mnesia_memory() ->
     lists:sum([bytes(mnesia:table_info(Tab, memory)) ||