anchor.diff:        traverse(RcvSnd, Iter2)
     end.
   
  --spec send_vivaldi_errors(pid_groups:groupname(), non_neg_integer()) -> ok | string().
  -send_vivaldi_errors(Group, Idx) ->
  -    case pid_groups:pid_of(Group, vivaldi) of
  -        failed ->
  -            ok;
  -        Vivaldi ->
  -            comm:send_local(Vivaldi, {get_coordinate, comm:this()}),
  -            receive
  -                ?SCALARIS_RECV(
  -                    {vivaldi_get_coordinate_response, _, Error}, %% ->
  -                    gmetric(both, lists:flatten(io_lib:format("vivaldi_error_~p", [Idx])), "float", Error, "error")
  -                  )
  -            end
  +-spec agg_check_pending(state()) -> state().
  +agg_check_pending(State) ->
  +    Pending = get_agg_pending(State),
  +    if
  +        Pending == 0 ->
  +            Load = get_agg_load(State),
  +            _ = gmetric(both, "kv pairs", "int32", Load, "pairs"),
  +            State;
  +        true ->
  +            State
       end.
   
  -% @doc get number of key-value pairs stored in given node
  --spec get_load(Pid::comm:erl_local_pid()) -> integer().
  -get_load(Pid) ->
  -    comm:send_local(Pid, {get_node_details, comm:this(), [load]}),
  -    receive
  -        ?SCALARIS_RECV(
  -            {get_node_details_response, Details}, %% ->
  -            node_details:get(Details, load)
  -          )
  -    after 2000 ->
  -            0
  -    end.
  +-spec get_last_active(state()) -> non_neg_integer().
  +get_last_active(State) ->
  +    element(1, State).
   
   -spec set_last_active(state()) -> state().
   set_last_active(State) ->
       setelement(1, State, erlang:now()).
  +
  +-spec get_trigger(state()) -> trigger:state().
  +get_trigger(State) ->
  +    element(2, State).
  +
  +-spec set_trigger(trigger:state(), state()) -> state().
  +set_trigger(Trigger, State) ->
  +    setelement(2, State, Trigger).
  +
  +-spec get_agg_id(state()) -> non_neg_integer().
  +get_agg_id(State) ->
  +    element(1, element(3, State)).
  +
  +-spec get_agg_load(state()) -> non_neg_integer().
  +get_agg_load(State) ->
  +    {_AggId, _Pending, Load} = element(3, State),
  +    Load.
  +
  +-spec set_agg_load(non_neg_integer(), state()) -> state().
  +set_agg_load(Load, State) ->
  +    {AggId, Pending, OldLoad} = element(3, State),
  +    setelement(3, State, {AggId, Pending - 1, OldLoad + Load}).
  +
  +-spec get_agg_pending(state()) -> non_neg_integer().
  +get_agg_pending(State) ->
  +    {_AggId, Pending, _Load} = element(3, State),
  +    Pending.
  +
  +-spec set_agg_pending(non_neg_integer(), state()) -> state().
  +set_agg_pending(NumPending, State) ->
  +    {AggId, _Pending, Load} = element(3, State),
  +    setelement(3, State, {AggId, NumPending, Load}).
  +
  +-spec inc_agg_id(state()) -> state().
  +inc_agg_id(State) ->
  +    {AggId, _Pending, _Load} = element(3,State),
  +    setelement(3, State, {AggId + 1, 0, 0}).
  +
  +%reset_load_aggregation(State) ->
  +%    {get_last_active(), 0}.
  
positive.code: 
  % @doc aggregate the number of key-value pairs
  %      and the amount of memory for this VM
  -spec send_dht_node_metrics() -> ok.
  send_dht_node_metrics() ->
      DHTNodes = pid_groups:find_all(dht_node),
      % Load of DHT Nodes
      Pairs = lists:foldl(fun (Pid, Agg) ->
                                  Agg + get_load(Pid)
                          end, 0, DHTNodes),
      _ = gmetric(both, "kv pairs", "int32", Pairs, "pairs"),
      % Memory Usage of DHT Nodes
      MemoryUsage = lists:sum([element(2, erlang:process_info(P, memory))
                               || P <- DHTNodes]),
      _ = gmetric(both, "Memory used by dht_nodes", "int32", MemoryUsage, "Bytes"),
      ok.
  
  -spec send_message_metrics() -> ok.
  send_message_metrics() ->
  
negative.code:                          X -> X
                       end,
      Trigger = trigger:init(trigger_periodic, fun () -> UpdateInterval end, ganglia_trigger),
      {_LastActive = 0, trigger:next(Trigger), {0, 0, 0}}.
  
  -spec on(message(), state()) -> state().
  on({ganglia_trigger}, State) ->
      Trigger = trigger:next(get_trigger(State)),
      StateNew = inc_agg_id(set_trigger(Trigger, State)),
      gen_component:post_op(StateNew, _Msg = {ganglia_periodic});
  
  on({ganglia_periodic}, State) ->
      NewState = send_dht_node_metrics(State),
      send_general_metrics(),
      send_message_metrics(),
      send_rrd_metrics(),
      send_vivaldi_errors(),
      set_last_active(NewState);
  
  on({ganglia_dht_load_aggregation, AggId, Msg} ,State) ->
      ?TRACE("~p: ~p~n", [AggId, Msg]),
      {get_node_details_response, [{load, Load}]} = Msg,
      CurAggId = get_agg_id(State),
      if
          % check for correct aggregation id
          AggId == CurAggId  ->
              NewState = set_agg_load(Load, State),
              agg_check_pending(NewState);
          % otherwise drop this message
          true -> State
      end;
  
  on({ganglia_vivaldi_response, DHTName, Msg}, State) ->
      {vivaldi_get_coordinate_response, _, Confidence} = Msg,
      gmetric(both, lists:flatten(io_lib:format("vivaldi_confidence_~s", [DHTName])), "float", Confidence, "Confidence"),
      State.
  
  send_general_metrics() ->
      % general erlang status information
  