anchor.diff:  on({send_error, FailedTarget, {bulkowner, reply, Id, Target, Msg, Parents}, _Reason}, State) ->
       bulkowner:send_reply_failed(Id, Target, Msg, Parents, self(), FailedTarget),
       State.
  +
  +-spec get_range_data({ets, db_ets:db()} | [{nonempty_string(), term()},...],
  +                     intervals:interval()) -> [{nonempty_string(), term()}].
  +get_range_data({ets, ETS}, Range) ->
  +    lists:foldl(fun(Interval, Acc1) ->
  +                    db_ets:foldl(ETS,
  +                                 fun(E, Acc) -> [db_ets:get(ETS,E) | Acc] end,
  +                                 Acc1, Interval)
  +                end,
  +                [], intervals:get_simple_intervals(Range));
  +get_range_data(Data, Range) ->
  +    lists:filter(
  +      fun(Entry) ->
  +              intervals:in(element(1, Entry),
  +                           Range)
  +      end, Data).
  
positive.code:             false ->
                  case Msg of
                      {bulk_distribute, Proc, N, Env, Data} ->
                          {RangeData, _Rest} = lists:partition(
                                  fun(Entry) ->
                                      intervals:in(?RT:hash_key(element(1, Entry)),
                                                  Range)
                                  end, Data),
                          comm:send(node:pidX(Head),
                                {bulkowner, Id, Range,
                                {bulk_distribute, Proc, N, Env, RangeData}, Parents});
  
negative.code:     comm:send_local(DHTNode, {bulkowner, reply, Id, Target, Msg, Parents}).
  
  -spec issue_bulk_distribute(ID::uid:global_uid(), atom(), pos_integer(),
                             comm:message(), [{string(), term()}] | {ets, ets:tab()}) -> ok.
  issue_bulk_distribute(Id, Proc, Pos, Msg, Data) ->
      issue_bulk_distribute(Id, Proc, Pos, Msg, Data, intervals:all()).
  
  -spec issue_bulk_distribute(ID::uid:global_uid(), atom(), pos_integer(),
                             comm:message(), [{string(), term()}] | {ets, ets:tab()},
                             intervals:interval()) -> ok.
  issue_bulk_distribute(Id, Proc, Pos, Msg, Data, Interval) ->
      ?TRACE("~p ~p: bulk_distribute initiated to ~p", [self(), ?MODULE,
                                                        Interval]),
      DHTNode = pid_groups:find_a(dht_node),
      comm:send_local(DHTNode, {bulkowner, start, Id, Interval, {bulk_distribute, Proc,
                                                          Pos, Msg, Data}}).
  
  -spec send_reply(Id::uid:global_uid(), Target::comm:mypid(), Msg::comm:message(),
  