anchor.diff:      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}),
  -    {0, rrd:create(get_gather_interval() * 1000000, 60, {timing, us})}.
  +    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))
  -        -> rrd:timing_type(T) when is_subtype(T, number()).
  +-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) ->
  
positive.code: 
  -behaviour(gen_component).
  
  -include("scalaris.hrl").
  
  % monitor process functions
  -export([start_link/1, init/1, on/2, check_config/0]).
  
  -type state() :: {Round::non_neg_integer(), Stats::rrd:rrd()}.
  -type message() ::
      {bench} |
      {propagate} |
      {get_node_details_response, node_details:node_details()} |
      {gather_stats, SourcePid::comm:mypid(), Round::pos_integer()} |
      {{get_rrd_response, Process::atom(), Key::string(), DB::rrd:rrd() | undefined}, {SourcePid::comm:mypid(), Round::pos_integer()}} |
      {gather_stats_response, Round::pos_integer(), Data::rrd:timing_type(number())} |
      {report_value, Round::pos_integer(), Stats::rrd:rrd()}.
  
  -spec run_bench() -> ok.
  run_bench() ->
  
negative.code: 
  %% @doc Message handler when the rm_loop module is fully initialized.
  -spec on(message(), state()) -> state().
  on({bench} = _Msg, State) ->
      ?TRACE1(_Msg, State),
      msg_delay:send_local(get_bench_interval(), self(), {bench}),
      run_bench(),
      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} = _Msg, State) ->
      ?TRACE1(_Msg, State),
      case is_leader(node_details:get(NodeDetails, my_range)) of
          false -> State;
          _ ->
              % start a new timeslot and gather stats...
              NewRound = State#state.round + 1,
              Msg = {send_to_group_member, monitor_perf, {gather_stats, comm:this(), NewRound}},
              bulkowner:issue_bulk_owner(intervals:all(), Msg),
              State1 = check_timeslots(State),
              broadcast_values(State, State1),
              State1#state{round = NewRound}
      end;
  
  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_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
          [] -> ok;
          _  -> comm:send(SourcePid, {gather_stats_response, Round, DataL})
      end,
      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.
  
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  