anchor.diff:                                           PartitionCount + VNode - 1)]).
   
   %% @private
  -%% @doc Find a minimal set of covering VNodes
  +%% @doc Find a minimal set of covering VNodes.
  +%% All parameters and return values are expressed as IDs in the [0,
  +%% RingSize) range.
  +%% Takes:
  +%%   A list of available vnode IDs
  +%%
  +%% Returns a list of {vnode_id, [partition_id,...]} tuples.
   find_coverage_vnodes([], _, Coverage) ->
       {ok, lists:sort(Coverage)};
  -find_coverage_vnodes(KeySpace, [], Coverage) ->
  -    {insufficient_vnodes_available, KeySpace, lists:sort(Coverage)};
  -find_coverage_vnodes(KeySpace, Available, Coverage) ->
  -    Res = next_vnode(KeySpace, Available),
  -    case Res of
  -        {0, _, _} -> % out of vnodes
  -            find_coverage_vnodes(KeySpace, [], Coverage);
  -        {_NumCovered, VNode, _} ->
  +find_coverage_vnodes(KeySpaces, [], Coverage) ->
  +    {insufficient_vnodes_available, KeySpaces, lists:sort(Coverage)};
  +find_coverage_vnodes(KeySpaces, Available, Coverage) ->
  +    case find_best_vnode_for_keyspace(KeySpaces, Available) of
  +        {error, no_coverage} ->
  +            %% Bail
  +            find_coverage_vnodes(KeySpaces, [], Coverage);
  +        VNode ->
               {value, {_, VNode, Covers}, UpdAvailable} = lists:keytake(VNode, 2, Available),
  -            UpdCoverage = [{VNode, ordsets:intersection(KeySpace, Covers)} | Coverage],
  -            UpdKeySpace = ordsets:subtract(KeySpace, Covers),
  -            find_coverage_vnodes(UpdKeySpace, UpdAvailable, UpdCoverage)
  +            UpdCoverage = [{VNode, ordsets:intersection(KeySpaces, Covers)} | Coverage],
  +            UpdKeySpaces = ordsets:subtract(KeySpaces, Covers),
  +            find_coverage_vnodes(UpdKeySpaces, UpdAvailable, UpdCoverage)
       end.
   
   %% @private
  -%% @doc Find the next vnode that covers the most of the
  -%% remaining keyspace. Use VNode id as tie breaker.
  -next_vnode(KeySpace, Available) ->
  +%% @doc Find the vnode that covers the most of the remaining
  +%% keyspace. Use VNode ID + offset (determined by request ID) as the
  +%% tiebreaker
  +find_best_vnode_for_keyspace(KeySpace, Available) ->
       CoverCount = [{covers(KeySpace, CoversKeys), VNode, TieBreaker} ||
                        {TieBreaker, VNode, CoversKeys} <- Available],
  -    hd(lists:sort(fun compare_next_vnode/2, CoverCount)).
  +    interpret_best_vnode(hd(lists:sort(fun compare_vnode_keyspaces/2,
  +                                       CoverCount))).
  +
  +%% @private
  +interpret_best_vnode({0, _, _}) ->
  +    {error, no_coverage};
  +interpret_best_vnode({_, VNode, _}) ->
  +    VNode.
   
   %% @private
   %% There is a potential optimization here once
  
positive.code: create_plan(VNodeTarget, NVal, PVC, ReqId, Service) ->
      {ok, CHBin} = riak_core_ring_manager:get_chash_bin(),
      PartitionCount = chashbin:num_partitions(CHBin),
      %% Create a coverage plan with the requested primary
      %% preference list VNode coverage.
      %% Get a list of the VNodes owned by any unavailble nodes
      DownVNodes = [Index ||
                       {Index, _Node}
                           <- riak_core_apl:offline_owners(Service, CHBin)],
      %% Calculate an offset based on the request id to offer
      %% the possibility of different sets of VNodes being
      %% used even when all nodes are available.
      Offset = ReqId rem NVal,
  
      RingIndexInc = chash:ring_increment(PartitionCount),
      AllKeySpaces = lists:seq(0, PartitionCount - 1),
      UnavailableKeySpaces = [(DownVNode div RingIndexInc) || DownVNode <- DownVNodes],
      %% Create function to map coverage keyspaces to
      %% actual VNode indexes and determine which VNode
      %% indexes should be filtered.
  
negative.code: %% @doc Count how many of CoversKeys appear in KeySpace
  covers(KeySpace, CoversKeys) ->
      ordsets:size(ordsets:intersection(KeySpace, CoversKeys)).
  
  %% @private
  %% Determines the number of non-mask bits in the 2^160 keyspace.
  %% Note that PartitionCount does not have to be ring size; we could be
  %% creating a coverage plan for subpartitions
  data_bits(PartitionCount) ->
      160 - round(math:log(PartitionCount) / math:log(2)).
  
  %% ===================================================================
  %% EUnit tests
  %% ===================================================================
  -ifdef(TEST).
  
  -define(SET(X), ordsets:from_list(X)).
  
  bits_test() ->
      %% 160 - log2(8)
      ?assertEqual(157, data_bits(8)),
      %% 160 - log2(65536)
      ?assertEqual(144, data_bits(65536)).
  
  n_keyspaces_test() ->
      %% First vnode in a cluster with ring size 64 should (with nval 3)
      %% cover keyspaces 61-63
      ?assertEqual([61, 62, 63], n_keyspaces(0, 3, 64)),
      %% 4th vnode in a cluster with ring size 8 should (with nval 5)
      %% cover the first 3 and last 2 keyspaces
      ?assertEqual([0, 1, 2, 6, 7], n_keyspaces(3, 5, 8)),
      %% First vnode in a cluster with a single partition should (with
      %% any nval) cover the only keyspace
      ?assertEqual([0], n_keyspaces(0, 1, 1)).
  
  covers_test() ->
      %% Count the overlap between the sets
      ?assertEqual(2, covers(?SET([1, 2]),
                             ?SET([0, 1, 2, 3]))),
      ?assertEqual(1, covers(?SET([1, 2]),
                             ?SET([0, 1]))),
      ?assertEqual(0, covers(?SET([1, 2, 3]),
                             ?SET([4, 5, 6, 7]))).
  
  best_vnode_test() ->
      %% Given two vnodes 0 and 7, pick 0 because it has more of the
      %% desired keyspaces
      ?assertEqual(0, find_best_vnode_for_keyspace(
                        ?SET([0, 1, 2, 3, 4]),
                        [{2, 0, ?SET([6, 7, 0, 1, 2])},
                         {1, 7, ?SET([5, 6, 7, 0, 1])}])),
      %% Given two vnodes 0 and 7, pick 7 because they cover the same
      %% keyspaces and 7 has the lower tiebreaker
      ?assertEqual(7, find_best_vnode_for_keyspace(
                        ?SET([0, 1, 2, 3, 4]),
                        [{2, 0, ?SET([6, 7, 0, 1, 2])},
                         {1, 7, ?SET([6, 7, 0, 1, 2])}])),
      %% Given two vnodes 0 and 7, pick 0 because they cover the same
      %% keyspaces and 0 has the lower tiebreaker
      ?assertEqual(0, find_best_vnode_for_keyspace(
                        ?SET([0, 1, 2, 3, 4]),
                        [{2, 0, ?SET([6, 7, 0, 1, 2])},
                         {3, 7, ?SET([6, 7, 0, 1, 2])}])).
  
  
  -endif.
  