anchor.diff:  
   %% @doc Message handler when the rm_loop module is fully initialized.
   -spec on(message(), state()) -> state().
  -on({bench}, State) ->
  +on({bench} = _Msg, State) ->
  +    ?TRACE1(_Msg, State),
       msg_delay:send_local(get_bench_interval(), self(), {bench}),
       run_bench(),
       State;
   
  -on({propagate}, State) ->
  +on({propagate} = _Msg, State) ->
  +    ?TRACE1(_Msg, State),
       msg_delay:send_local(get_gather_interval(), self(), {propagate}),
       DHT_Node = pid_groups:get_my(dht_node),
       comm:send_local(DHT_Node, {get_node_details, comm:this(), [my_range]}),
       State;
   
  -on({get_node_details_response, NodeDetails}, State) ->
  -    MyRange = node_details:get(NodeDetails, my_range),
  -    case is_leader(MyRange) of
  +on({get_node_details_response, NodeDetails} = _Msg, State) ->
  +    ?TRACE1(_Msg, State),
  +    case is_leader(node_details:get(NodeDetails, my_range)) of
           false -> State;
           _ ->
               % start a new timeslot and gather stats...
  -            {Round, Stats} = State,
  -            NewRound = Round + 1,
  +            NewRound = State#state.round + 1,
               Msg = {send_to_group_member, monitor_perf, {gather_stats, comm:this(), NewRound}},
               bulkowner:issue_bulk_owner(intervals:all(), Msg),
  -            NewStats = rrd:check_timeslot_now(Stats),
  -            broadcast_value(Stats, NewStats, Round),
  -            {NewRound, NewStats}
  +            State1 = check_timeslots(State),
  +            broadcast_values(State, State1),
  +            State1#state{round = NewRound}
       end;
   
  -on({gather_stats, SourcePid, Round}, State) ->
  +on({gather_stats, SourcePid, Round} = _Msg, State) ->
  +    ?TRACE1(_Msg, State),
       This = comm:this_with_cookie({SourcePid, Round}),
       comm:send_local(pid_groups:get_my(monitor),
  -                    {get_rrd, ?MODULE, "read_read", This}),
  -    State;
  -
  -on({{get_rrd_response, _Process, _Key, undefined}, _Cookie}, State) ->
  -    State;
  -
  -on({{get_rrd_response, _Process, _Key, DB}, {SourcePid, Round}}, State) ->
  -    DB2 = rrd:reduce_timeslots(1, DB),
  -    DataL = rrd:dump_with(DB2, fun(_DB, _From, _To, X) -> X end),
  +                    {get_rrds, [{?MODULE, "read_read"}, {dht_node, "lookup_hops"}], This}),
  +    State;
  +
  +on({{get_rrds_response, DBs}, {SourcePid, Round}} = _Msg, State) ->
  +    ?TRACE1(_Msg, State),
  +    DataL = lists:flatten(
  +              [begin
  +                   DB2 = rrd:reduce_timeslots(1, DB),
  +                   case rrd:dump_with(DB2, fun(_DB, _From, _To, X) -> X end) of
  +                       []     -> [];
  +                       [Data] -> {Process, Key, Data}
  +                   end
  +               end || {Process, Key, DB} <- DBs, DB =/= undefined]),
       case DataL of
  -        [Data] -> comm:send(SourcePid, {gather_stats_response, Round, Data});
  -        []     -> ok
  +        [] -> ok;
  +        _  -> comm:send(SourcePid, {gather_stats_response, Round, DataL})
       end,
       State;
   
  -on({gather_stats_response, Round, Data}, {Round, Stats}) ->
  -    Time = rrd:get_current_time(Stats),
  -    {Round, rrd:add_with(Time, Data, Stats, fun timing_update_fun/3)};
  -on({gather_stats_response, _Round, _Data}, State) ->
  -    State;
  -
  -on({report_value, Round, Stats}, {OldRound, OldStats}) when OldRound < Round ->
  -    Time = rrd:get_current_time(Stats),
  -    DataL = rrd:dump_with(Stats, fun(_DB, _From, _To, X) -> X end),
  -    NewStats = case DataL of
  -        [Data] -> rrd:add_with(Time, Data, OldStats, fun timing_update_fun/3);
  -        []     -> OldStats
  -    end,
  -    {Round, NewStats};
  -
  -on({report_value, _Round, _Stats}, State) ->
  +on({gather_stats_response, Round, DataL} = _Msg, State) when Round =:= State#state.round->
  +    ?TRACE1(_Msg, State),
  +    lists:foldl(
  +      fun(Data, A) ->
  +              case Data of
  +                  {?MODULE, "read_read", PerfRR} ->
  +                      DB = A#state.perf_rr,
  +                      T = rrd:get_current_time(DB),
  +                      A#state{perf_rr = rrd:add_with(T, PerfRR, DB, fun timing_update_fun/3)};
  +                  {dht_node, "lookup_hops", PerfLH} ->
  +                      DB = A#state.perf_lh,
  +                      T = rrd:get_current_time(DB),
  +                      A#state{perf_lh = rrd:add_with(T, PerfLH, DB, fun timing_update_fun/3)}
  +              end
  +      end, State, DataL);
  +on({gather_stats_response, _Round, _Data} = _Msg, State) ->
  +    ?TRACE1(_Msg, State),
  +    State;
  +
  +on({report_value, OtherState} = _Msg, State) when OtherState#state.round > State#state.round ->
  +    ?TRACE1(_Msg, State),
  +    integrate_values(OtherState, State);
  +
  +on({report_value, _OtherState} = _Msg, State) ->
  +    ?TRACE1(_Msg, State),
       State.
   
   %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
positive.code:     {Sum + NewSum, Sum2 + NewSum2, Count + NewCount,
       erlang:min(Min, NewMin), erlang:max(Max, NewMax), NewHist2}.
  
  broadcast_value(OldStats, Stats, OldRound) ->
      % broadcast the latest value only if a new time slot was started
      SlotOld = rrd:get_slot_start(0, OldStats),
      SlotNew = rrd:get_slot_start(0, Stats),
      case SlotNew of
          SlotOld -> ok; %nothing to do
          _  -> % new slot -> broadcast latest value only:
              DB2 = rrd:reduce_timeslots(1, OldStats),
              Msg = {send_to_group_member, monitor_perf, {report_value, OldRound, DB2}},
              bulkowner:issue_bulk_owner(intervals:all(), Msg)
      end.
  
  %% @doc Checks whether the node is the current leader.
  
negative.code:     FirstDelay = randoms:rand_uniform(1, get_bench_interval() + 1),
      msg_delay:send_local(FirstDelay, self(), {bench}),
      msg_delay:send_local(get_gather_interval(), self(), {propagate}),
      Now = os:timestamp(),
      #state{perf_rr = rrd:create(get_gather_interval() * 1000000, 60, {timing, us}, Now),
             perf_lh = rrd:create(get_gather_interval() * 1000000, 60, {timing, count}, Now)}.
  
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  % Miscellaneous
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
  -spec timing_update_fun(Time::rrd:internal_time(), Old::rrd:timing_type(T) | undefined, New::rrd:timing_type(T) | undefined)
          -> rrd:timing_type(T) | undefined when is_subtype(T, number()).
  timing_update_fun(_Time, undefined, New) ->
      New;
  timing_update_fun(_Time, Old, undefined) ->
      Old;
  timing_update_fun(_Time, {Sum, Sum2, Count, Min, Max, Hist},
                    {NewSum, NewSum2, NewCount, NewMin, NewMax, NewHist}) ->
      NewHist2 = lists:foldl(fun({V, C}, Acc) ->
  