Anonymous avatar Anonymous committed f936ade

keep "exclusion list" of already-handed-off-and-empty partitions, cleanup

Comments (0)

Files changed (6)

apps/riak/src/riak_connect.erl

     end.
 
 ensure_vnodes_started(Ring) ->
-    VNodes2Start = case length(riak_ring:all_members(Ring)) of
-       1 -> riak_ring:my_indices(Ring);
-       _ -> [riak_ring:random_other_index(Ring)| riak_ring:my_indices(Ring)]
-    end,
-    [begin
-        gen_server:cast({riak_vnode_master, node()}, {start_vnode, I}) 
-    end|| I <- VNodes2Start].
+    AllMembers = riak_ring:all_members(Ring),
+    VNodes2Start = 
+        case {length(AllMembers), hd(AllMembers) =:= node()} of
+            {1, true} -> riak_ring:my_indices(Ring);
+            _ -> 
+                {ok, Excl} = gen_server:call(riak_vnode_master, get_exclusions),
+                case riak_ring:random_other_index(Ring, Excl) of
+                    no_indices ->
+                        case length(Excl) =:= riak_ring:num_partitions(Ring) of
+                            true ->
+                                exit;
+                            false ->
+                                riak_ring:my_indices(Ring)
+                        end;
+                    RO ->
+                        [RO | riak_ring:my_indices(Ring)]
+                end
+        end,
+    case VNodes2Start of
+        exit ->
+            riak:stop("node removal completed, exiting.");
+        _ ->
+            [begin
+                 gen_server:cast({riak_vnode_master, node()}, {start_vnode, I}) 
+             end|| I <- VNodes2Start]
+    end.
 
 remove_from_cluster(ExitingNode) ->
     % Set the remote node to stop claiming.

apps/riak/src/riak_handoff_receiver.erl

 
 process_message(0, MsgData, State) ->
     <<Partition:160/integer>> = MsgData,
-    {ok, VNode} = gen_server2:call(riak_vnode_master, {get_vnode, Partition}),  
+    {ok, VNode} = gen_server2:call(riak_vnode_master, {get_vnode, Partition}, 60000),  
     State#state{partition=Partition, vnode=VNode};
 process_message(1, MsgData, State=#state{vnode=VNode}) ->
     % header of 1 is a riakobject_pb

apps/riak/src/riak_handoff_sender.erl

 -include("riakserver_pb.hrl").
 
 start_link(TargetNode, Partition, BKeyList) ->
-    io:format("about to acquire lock~n"),
     case global:set_lock({handoff_token, {node(), Partition}}, [node()], 0) of
         true ->
             Self = self(),
             {ok, spawn_link(fun() -> start_fold(TargetNode, Partition, BKeyList, Self) end)};
         false ->
-            io:format("failed to acquire handoff lock token~n"),
             {error, locked}
     end.
             

apps/riak/src/riak_ring.erl

 	 owner_node/1,all_members/1,num_partitions/1,all_owners/1,
          transfer_node/3, rename_node/3, reconcile/2, my_indices/1,
 	 index_owner/2,diff_nodes/2,random_node/1, random_other_index/1,
+         random_other_index/2,
          get_meta/2, update_meta/3, equal_rings/2]).	 
 
 % @type riak_ring(). The opaque data type used for partition ownership.
         _ -> lists:nth(crypto:rand_uniform(1, length(L)+1), L)
     end.
 
+random_other_index(State, Exclude) when is_list(Exclude) ->
+    L = [I || {I, Owner} <- ?MODULE:all_owners(State),
+              Owner =/= node(),
+              not lists:member(I, Exclude)],
+    case L of
+        [] -> no_indices;
+        _ -> lists:nth(crypto:rand_uniform(1, length(L)+1), L)
+    end.
+
 % @doc Return the node that owns the given index.
 % @spec index_owner(State :: chstate(), Idx :: integer()) -> Node :: term()
 index_owner(State, Idx) ->

apps/riak/src/riak_vnode.erl

     end.
 
 %% @private
-delete_and_exit(StateData=#state{mod=Mod, modstate=ModState}) ->
+delete_and_exit(StateData=#state{idx=Idx, mod=Mod, modstate=ModState}) ->
     ok = Mod:drop(ModState),
+    gen_server:cast(riak_vnode_master, {add_exclusion, Idx}),
     {stop, normal, StateData}.
 
 %%%%%%%%%% in active state, we process normal client requests
 
 %% @private
 do_fold(Fun, Acc0, _State=#state{idx=Idx,mod=Mod, modstate=ModState}) ->
-    io:format("~p do_fold~n",[Idx]),
     Mod:fold(ModState, Fun, Acc0).
 
 %% @private

apps/riak/src/riak_vnode_master.erl

 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
 	 terminate/2, code_change/3]).
 -record(idxrec, {idx, pid, monref}).
--record(state, {idxtab}).
+-record(state, {idxtab, excl=ordsets:new()}).
 
 start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
 %% @private
 init([]) -> {ok, #state{idxtab=ets:new(riak_vnode_idx,[{keypos,2}])}}.
 
+
 %% @private
-handle_cast({start_vnode, Partition}, State) ->
+handle_cast({start_vnode, Partition}, State=#state{excl=Excl}) ->
     _Pid = get_vnode(Partition, State),
-    {noreply, State};
+    {noreply, State#state{excl=ordsets:del_element(Partition, Excl)}};
 handle_cast({vnode_map, {Partition,_Node},
              {ClientPid,QTerm,BKey,KeyData}}, State) ->
     Pid = get_vnode(Partition, State),
             {FSM_pid, Bucket, ReqID}}, State) ->
     Pid = get_vnode(Partition, State),
     gen_fsm:send_event(Pid, {list_bucket, FSM_pid, Bucket, ReqID}),
-    {noreply, State}.
-                  
-
+    {noreply, State};
+handle_cast({add_exclusion, Partition}, State=#state{excl=Excl}) ->
+    {noreply, State#state{excl=ordsets:add_element(Partition, Excl)}}.
 
 %% @private
 handle_call(all_possible_vnodes, _From, State) ->
       fun() -> gen_fsm:send_all_state_event(Pid, {fold, {Fun,Acc0,From}}) end),
     {noreply, State};
 handle_call({get_vnode, Partition}, _From, State) ->
-    {reply, {ok, get_vnode(Partition, State)}, State}.
+    {reply, {ok, get_vnode(Partition, State)}, State};
+handle_call(get_exclusions, _From, State=#state{excl=Excl}) ->
+    {reply, {ok, ordsets:to_list(Excl)}, State}.
 %% @private
 handle_info({'DOWN', MonRef, process, _P, _I}, State) ->
     delmon(MonRef, State),
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.