anchor.diff:  -author('schuett@zib.de').
   -vsn('$Id$').
   
  +-define(TRACE(X, Y), io:format(X, Y)).
  +
   -include("scalaris.hrl").
   
   % public API:
  --export([issue_bulk_owner/3, issue_send_reply/4, issue_bulk_distribute/5,
  +-export([issue_bulk_owner/3, issue_send_reply/4,
  +         issue_bulk_distribute/5, issue_bulk_distribute/6,
            send_reply/5, send_reply_failed/6]).
   
   % only use inside the dht_node process:
  
positive.code:         false ->
              case Msg of
                  {bulk_distribute, Proc, N, Env, Data} ->
                      {SuccData, _Rest} = lists:partition(
                              fun(Entry) ->
                                  intervals:in(?RT:hash_key(element(1, Entry)),
                                              SuccIntI)
                              end, Data),
                      comm:send(node:pidX(nodelist:succ(Neighbors)),
                                {bulkowner, deliver, Id, SuccIntI,
                                {bulk_distribute, Proc, N, Env, SuccData}, Parents});
  
negative.code: 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).
  