anchor.diff:  
   -export([new/3]).
   -export([new/4]).
  +
  +-record(state, {
  +  ip :: string(),
  +  bucket :: binary(),
  +  resolver :: atom(),
  +  pid = no_pid :: pid() | no_pid,
  +  key_vals = no_keys :: dict:dict() | no_keys,
  +  map_fun_str :: string()
  +}).
   
   new(DbIp, Bucket, Resolver) ->
     new(DbIp, Bucket, Resolver, default_map_fun()).
  
positive.code:   {Dt, {KeyVals, Size}} = timer:tc(fun() -> DoPrep() end),
    lager:info("dsdl_riak prep (num_elements=~p, size=~p, prep_time_s=~p).",
               [dict:size(KeyVals), Size, Dt / (1000 * 1000)]),
    {Size, {Ip, Bucket, Resolver, Pid, KeyVals, MapFunStr}}.
  
  %% Get the list of {Key, HashedVal} tuples.
  get({_Ip, _Bucket, _Resolver, _Pid, KeyVals, _MapFunStr}) ->
    dict:to_list(KeyVals).
  
  %% Given a list L of {Key, HashedVal} tuples, return the list of {Key,
  %% Val}, i.e. unhashed values.
  get_vals({_Ip, Bucket, Resolver, Pid, _KeyVals, _MapFunStr}, L) ->
    [{Key, riak_ops:get(Pid, Bucket, Key, Resolver)} || {Key, _HashedVal} <- L].
  
  %% Store a {Key, Value} pair.
  put({Ip, Bucket, Resolver, Pid, KeyVals, MapFunStr}, {Key, Val}) ->
    NewVal = riak_ops:put(Pid, Bucket, Key, Val, Resolver),
    Hash = crypto:hash(sha, term_to_binary(NewVal)),
    {Ip, Bucket, Resolver, Pid, dict:store(Key, Hash, KeyVals), MapFunStr}.
  
  %% Cleanup after syncing is done.
  unprep({Ip, Bucket, Resolver, Pid, _Keys, MapFunStr}) ->
    riak_ops:disconnect(Pid),
    {Ip, Bucket, Resolver, no_pid, no_keys, MapFunStr}.
  
  map(Pid, Bucket, MapFunStr) ->
    Res = riakc_pb_socket:mapred(Pid, Bucket, [{map, {strfun, MapFunStr}, "myarg",
  
negative.code: %% and Sha is the SHA-128 value of the object, or the empty list
  %% if the value should be ignored.
  new(DbIp, Bucket, Resolver, MapFunStr) ->
    State = #state{
      ip = DbIp,
      bucket = Bucket,
      resolver = Resolver,
      map_fun_str = MapFunStr
    },
    {State, fun prep/1, fun get/1, fun get_vals/2, fun put/2, fun unprep/1}.
  
  %%%_* Internal =========================================================
  
  %% Prepare to start syncing.
  prep(#state{ip=Ip, bucket=Bucket, map_fun_str=MapFunStr}=State) ->
    Pid = riak_ops:connect(Ip),
    DoPrep = fun() ->
                 KeyValsSize = map(Pid, Bucket, MapFunStr),
  