anchor.diff:           max_size/1,
            clear/1,
            clear_bkey/2,
  -         destroy/1]).
  +         destroy/1,
  +         table_sizes/1]).
  +-export([init/1,
  +         handle_call/3,
  +         handle_cast/2,
  +         handle_info/2,
  +         terminate/2]).
   
   -record(kv_lru, {max_size,
                    bucket_idx,
  
positive.code:                        ts}).
  
  new(0) ->
      #kv_lru{max_size=0};
  new(Size) ->
      IdxName = pid_to_list(self()) ++ "_cache_age_idx",
      BucketIdxName = pid_to_list(self()) ++ "_bucket_idx",
      CacheName = pid_to_list(self()) ++ "_cache",
      Idx = ets:new(list_to_atom(IdxName), [ordered_set, public]),
      BucketIdx = ets:new(list_to_atom(BucketIdxName), [bag, public]),
      Cache = ets:new(list_to_atom(CacheName), [public, {keypos, 2}]),
      #kv_lru{max_size=Size, age_idx=Idx, bucket_idx=BucketIdx, cache=Cache}.
  
  put(#kv_lru{max_size=0}, _BKey, _Key, _Value) ->
      ok;
  put(#kv_lru{max_size=MaxSize, age_idx=Idx, bucket_idx=BucketIdx,
              cache=Cache}, BKey, Key, Value) ->
      remove_existing(Idx, BucketIdx, Cache, BKey, Key),
      insert_value(Idx, BucketIdx, Cache, BKey, Key, Value),
      prune_oldest_if_needed(MaxSize, Idx, BucketIdx, Cache).
  
  fetch(#kv_lru{max_size=0}, _BKey, _Key) ->
      notfound;
  fetch(#kv_lru{cache=Cache}=LRU, BKey, Key) ->
      case fetch_value(Cache, BKey, Key) of
          notfound ->
              notfound;
          Value ->
              %% Do a put to update the timestamp in the cache
              riak_kv_lru:put(LRU, BKey, Key, Value),
              Value
      end.
  
  remove(#kv_lru{max_size=0}, _BKey, _Key) ->
      ok;
  remove(#kv_lru{age_idx=Idx, bucket_idx=BucketIdx, cache=Cache}, BKey, Key) ->
      remove_existing(Idx, BucketIdx, Cache, BKey, Key),
      ok.
  
  size(#kv_lru{max_size=0}) ->
      0;
  size(#kv_lru{age_idx=Idx}) ->
      ets:info(Idx, size).
  
  max_size(#kv_lru{max_size=MaxSize}) ->
      MaxSize.
  
  clear(#kv_lru{max_size=0}) ->
      ok;
  clear(#kv_lru{age_idx=Idx, cache=Cache}) ->
      ets:delete_all_objects(Idx),
      ets:delete_all_objects(Cache),
      ok.
  
  clear_bkey(#kv_lru{max_size=0}, _BKey) ->
      ok;
  clear_bkey(#kv_lru{bucket_idx=BucketIdx}=LRU, BKey) ->
      R = ets:match(BucketIdx, {BKey, '$1'}),
       case R of
          [] ->
              ok;
          Keys ->
              [remove(LRU, BKey, Key) || [Key] <- Keys],
              ok
      end.
  
  destroy(#kv_lru{max_size=0}) ->
      ok;
  destroy(#kv_lru{age_idx=Idx, bucket_idx=BucketIdx, cache=Cache}) ->
      ets:delete(Idx),
      ets:delete(BucketIdx),
      ets:delete(Cache),
      ok.
  
  %% Internal functions
  remove_existing(Idx, BucketIdx, Cache, BKey, Key) ->
  
negative.code:     [F(X) || X <- lists:seq(1,10)],
      consistency_check(C).
  
  %% Make sure that riak_kv_lru is correct under concurrent modification
  %% by spawning 10 processes that each do 1000 puts on the same LRU, then
  %% checking that the size limit of the cache has been respected
  %% (added to check https://issues.basho.com/show_bug.cgi?id=969)
  concurrency_test() ->
      Size = 10,
      C = riak_kv_lru:new(Size),
      Pids = [ spawn_link(concurrent_incrementer(C, K, self()))
               || K <- lists:seq(10000, 10010) ],
      wait_for_incrementers(Pids),
      consistency_check(C),
      ?assertEqual(Size, riak_kv_lru:size(C)).
  
  concurrent_incrementer(C, K, Test) ->
      fun() ->
              [ riak_kv_lru:put(C, N+K, N+K, N+K)
                || N <- lists:seq(1, 1000) ],
              Test ! {increment_done, self()}
      end.
  
  wait_for_incrementers([]) -> ok;
  wait_for_incrementers(Pids) ->
      receive {increment_done, Pid} ->
              wait_for_incrementers(lists:delete(Pid, Pids))
      after 5000 ->
              throw(incrementer_timeout)
      end.
  
  consistency_check(LRU) ->
      Ts = table_sizes(LRU),
      %% make sure all tables report same size
      UniqueSizes = lists:usort([ Size || {_Name, Size} <- Ts]),
      ?assertEqual(1, length(UniqueSizes)).
  
  -endif.
  