Current section
Files
Jump to
Current section
Files
src/riak_core_vnode.erl
%%
%% Copyright (c) 2007-2010 Basho Technologies, Inc. All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License. You may obtain
%% a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied. See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------
-module(riak_core_vnode).
-behaviour(gen_fsm_compat).
-include("riak_core_vnode.hrl").
-export([start_link/3,
start_link/4,
wait_for_init/1,
send_command/2,
send_command_after/2]).
-export([init/1,
started/2,
started/3,
active/2,
active/3,
handle_event/3,
handle_sync_event/4,
handle_info/3,
terminate/3,
code_change/4]).
-export([reply/2, monitor/1]).
-export([get_mod_index/1,
get_modstate/1,
set_forwarding/2,
trigger_handoff/2,
trigger_handoff/3,
trigger_delete/1,
core_status/1,
handoff_error/3]).
-export([cast_finish_handoff/1,
send_an_event/2,
send_req/2,
send_all_proxy_req/2,
cancel_handoff/1,
handoff_complete/1,
resize_transfer_complete/2,
handoff_data/3,
unregistered/1]).
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-export([test_link/2, current_state/1]).
-endif.
-ifdef(PULSE).
-compile(export_all).
-compile({parse_transform, pulse_instrument}).
-compile({pulse_replace_module,
[{gen_fsm_compat, pulse_gen_fsm},
{gen_server, pulse_gen_server}]}).
-endif.
-define(NORMAL_REASON(R),
R == normal orelse
R == shutdown orelse
is_tuple(R) andalso element(1, R) == shutdown).
-export_type([vnode_opt/0, pool_opt/0]).
-type vnode_opt() :: pool_opt().
-type pool_opt() :: {pool, WorkerModule :: module(),
PoolSize :: pos_integer(), WorkerArgs :: [term()]}.
-callback init([partition()]) -> {ok,
ModState :: term()} |
{ok, ModState :: term(), [vnode_opt()]} |
{error, Reason :: term()}.
-callback handle_command(Request :: term(),
Sender :: sender(), ModState :: term()) -> continue |
{reply,
Reply ::
term(),
NewModState ::
term()} |
{noreply,
NewModState ::
term()} |
{async,
Work ::
function(),
From ::
sender(),
NewModState ::
term()} |
{stop,
Reason ::
term(),
NewModState ::
term()}.
-callback handle_coverage(Request :: term(),
keyspaces(), Sender :: sender(),
ModState :: term()) -> continue |
{reply, Reply :: term(),
NewModState :: term()} |
{noreply,
NewModState :: term()} |
{async, Work :: function(),
From :: sender(),
NewModState :: term()} |
{stop, Reason :: term(),
NewModState :: term()}.
-callback handle_exit(pid(), Reason :: term(),
ModState :: term()) -> {noreply,
NewModState :: term()} |
{stop, Reason :: term(),
NewModState :: term()}.
-callback handoff_starting(handoff_dest(),
ModState :: term()) -> {boolean(),
NewModState :: term()}.
-callback handoff_cancelled(ModState :: term()) -> {ok,
NewModState :: term()}.
-callback handoff_finished(handoff_dest(),
ModState :: term()) -> {ok, NewModState :: term()}.
-callback handle_handoff_command(Request :: term(),
Sender :: sender(),
ModState :: term()) -> {reply, Reply :: term(),
NewModState ::
term()} |
{noreply,
NewModState ::
term()} |
{async,
Work :: function(),
From :: sender(),
NewModState ::
term()} |
{forward,
NewModState ::
term()} |
{drop,
NewModState ::
term()} |
{stop, Reason :: term(),
NewModState :: term()}.
-callback handle_handoff_data(binary(),
ModState :: term()) -> {reply,
ok |
{error, Reason :: term()},
NewModState :: term()}.
-callback encode_handoff_item(Key :: term(),
Value :: term()) -> corrupted | binary().
-callback is_empty(ModState :: term()) -> {boolean(),
NewModState :: term()} |
{false, Size :: pos_integer(),
NewModState :: term()}.
-callback terminate(Reason :: term(),
ModState :: term()) -> ok.
-callback delete(ModState :: term()) -> {ok,
NewModState :: term()}.
%% This commands are not executed inside the VNode, instead they are
%% part of the vnode_proxy contract.
%%
%% The vnode_proxy will drop requests in an overload situation, when
%% his happens one of the two handle_overload_* commands in the
%% vnode module is called. This call happens **from the vnode proxy**
%%
%% These calls are wrapped in a catch() meaning that when they don't
%% exist they will quietly fail. However the catch is hugely expensive
%% leading to the sitaution that when there already is a overload
%% the vnode proxy gets even worst overloaded.
%%
%% This is pretty bad since the proxy is supposed to protect against
%% exactly this overload.
%%
%% So yea sorry, you're going to be forced to implement them, if nothing
%% else just nop them out.
%%
%% BUT DO NOT call expensive functions from them there is a special hell
%% for people doing that! (it's called overflowing message queue hell and is
%% really nasty!)
-callback handle_overload_command(Request :: term(),
Sender :: sender(), Idx :: partition()) -> ok.
-callback handle_overload_info(Request :: term(),
Idx :: partition()) -> ok.
%% handle_exit/3 is an optional behaviour callback that can be implemented.
%% It will be called in the case that a process that is linked to the vnode
%% process dies and allows the module using the behaviour to take appropriate
%% action. It is called by handle_info when it receives an {'EXIT', Pid, Reason}
%% message and the function signature is: handle_exit(Pid, Reason, State).
%%
%% It should return a tuple indicating the next state for the fsm. For a list of
%% valid return types see the documentation for the gen_fsm_compat handle_info callback.
%%
%% Here is what the spec for handle_exit/3 would look like:
%% -spec handle_exit(pid(), atom(), term()) ->
%% {noreply, term()} |
%% {stop, term(), term()}
%% handle_info/2 is an optional behaviour callback too.
%% It will be called in the case when a vnode receives any other message
%% than an EXIT message.
%% The function signature is: handle_info(Info, State).
%% It should return a tuple of the form {ok, NextState}
%%
%% Here is what the spec for handle_info/2 would look like:
%% -spec handle_info(term(), term()) -> {ok, term()}
-define(DEFAULT_TIMEOUT, 60000).
-define(LOCK_RETRY_TIMEOUT, 10000).
%% ========
%% API
%% ========
start_link(Mod, Index, Forward) ->
start_link(Mod, Index, 0, Forward).
start_link(Mod, Index, InitialInactivityTimeout,
Forward) ->
gen_fsm_compat:start_link(?MODULE,
[Mod, Index, InitialInactivityTimeout, Forward],
[]).
%% #1 - State started
wait_for_init(Vnode) ->
gen_fsm_compat:sync_send_event(Vnode,
wait_for_init,
infinity).
%% #2 -
%% Send a command message for the vnode module by Pid -
%% typically to do some deferred processing after returning yourself
send_command(Pid, Request) ->
gen_fsm_compat:send_event(Pid,
#riak_vnode_req_v1{request = Request}).
%% #3 -
handoff_error(Vnode, Err, Reason) ->
gen_fsm_compat:send_event(Vnode,
{handoff_error, Err, Reason}).
%% #4 -
get_mod_index(VNode) ->
gen_fsm_compat:sync_send_all_state_event(VNode,
get_mod_index).
%% #5
set_forwarding(VNode, ForwardTo) ->
gen_fsm_compat:send_all_state_event(VNode,
{set_forwarding, ForwardTo}).
%% #6
trigger_handoff(VNode, TargetIdx, TargetNode) ->
gen_fsm_compat:send_all_state_event(VNode,
{trigger_handoff,
TargetIdx,
TargetNode}).
%% #7
trigger_handoff(VNode, TargetNode) ->
gen_fsm_compat:send_all_state_event(VNode,
{trigger_handoff, TargetNode}).
%% #8
trigger_delete(VNode) ->
gen_fsm_compat:send_all_state_event(VNode,
trigger_delete).
%% #9
core_status(VNode) ->
gen_fsm_compat:sync_send_all_state_event(VNode,
core_status).
%% #10
%% Sends a command to the FSM that called it after Time
%% has passed.
-spec send_command_after(integer(),
term()) -> reference().
send_command_after(Time, Request) ->
gen_fsm_compat:send_event_after(Time,
#riak_vnode_req_v1{request = Request}).
%%%%%%% %new APIs
%% #11 - riak_core_vnode_manager - handle_vnode_event
cast_finish_handoff(VNode) ->
gen_fsm_compat:send_all_state_event(VNode,
finish_handoff).
%% #12 - riak_core_vnode_manager - handle_vnode_event
cancel_handoff(VNode) ->
gen_fsm_compat:send_all_state_event(VNode,
cancel_handoff).
%% #13 - riak_core_vnode_master - send_an_event
send_an_event(VNode, Event) ->
gen_fsm_compat:send_event(VNode, Event).
%% #14 - riak_core_vnode_master - handle_cast/handle_call
%riak_core_vnode_master - command2
%riak_core_vnode_proxy - handle_call
send_req(VNode, Req) ->
gen_fsm_compat:send_event(VNode, Req).
%% #15 - riak_core_vnode_master - handle_call
send_all_proxy_req(VNode, Req) ->
gen_fsm_compat:send_all_state_event(VNode, Req).
%% #16 - riak:core_handoff_sender - start_fold_
-spec handoff_complete(VNode :: pid()) -> ok.
handoff_complete(VNode) ->
gen_fsm_compat:send_event(VNode, handoff_complete).
%% #17 - riak:core_handoff_sender - start_fold_
-spec resize_transfer_complete(VNode :: pid(),
NotSentAcc :: term()) -> ok.
resize_transfer_complete(VNode, NotSentAcc) ->
gen_fsm_compat:send_event(VNode,
{resize_transfer_complete, NotSentAcc}).
%% #18 - riak_core_handoff_receiver - process_message
handoff_data(VNode, MsgData, VNodeTimeout) ->
gen_fsm_compat:sync_send_all_state_event(VNode,
{handoff_data, MsgData},
VNodeTimeout).
%% #19 - riak_core_vnode_proxy - handle_cast
unregistered(VNode) ->
gen_fsm_compat:send_event(VNode, unregistered).
%% @doc Send a reply to a vnode request. If
%% the Ref is undefined just send the reply
%% for compatibility with pre-0.12 requestors.
%% If Ref is defined, send it along with the
%% reply.
%% NOTE: We *always* send the reply using unreliable delivery.
%%
-spec reply(sender(), term()) -> any().
reply({fsm, undefined, From}, Reply) ->
riak_core_send_msg:send_event_unreliable(From, Reply);
reply({fsm, Ref, From}, Reply) ->
riak_core_send_msg:send_event_unreliable(From,
{Ref, Reply});
reply({server, undefined, From}, Reply) ->
riak_core_send_msg:reply_unreliable(From, Reply);
reply({server, Ref, From}, Reply) ->
riak_core_send_msg:reply_unreliable(From, {Ref, Reply});
reply({raw, Ref, From}, Reply) ->
riak_core_send_msg:bang_unreliable(From, {Ref, Reply});
reply(ignore, _Reply) -> ok.
%% @doc Set up a monitor for the pid named by a {@type sender()} vnode
%% argument. If `Sender' was the atom `ignore', this function sets up
%% a monitor on `self()' in order to return a valid (if useless)
%% monitor reference.
-spec monitor(Sender :: sender()) -> Monitor ::
reference().
monitor({fsm, _, From}) ->
erlang:monitor(process, From);
monitor({server, _, {Pid, _Ref}}) ->
erlang:monitor(process, Pid);
monitor({raw, _, From}) ->
erlang:monitor(process, From);
monitor(ignore) -> erlang:monitor(process, self()).
%% ========================
%% ========
%% State, Mode, Init, Terminate
%% ========
%% ========================
-record(state,
{index :: partition(),
mod :: module(),
modstate :: term(),
forward :: node() | [{integer(), node()}],
handoff_target = none :: none | {integer(), node()},
handoff_pid :: pid() | undefined,
handoff_type ::
riak_core_handoff_manager:ho_type() | undefined,
pool_pid :: pid() | undefined,
pool_config :: tuple() | undefined,
manager_event_timer :: reference() | undefined,
inactivity_timeout :: non_neg_integer()}).
init([Module,
Index,
InitialInactivityTimeout,
Forward]) ->
process_flag(trap_exit, true),
State = #state{index = Index, mod = Module,
forward = Forward,
inactivity_timeout = InitialInactivityTimeout},
{ok, started, State, 0}.
terminate(Reason, _StateName,
#state{mod = Module, modstate = ModState,
pool_pid = Pool}) ->
%% Shutdown if the pool is still alive and a normal `Reason' is
%% given - there could be a race on delivery of the unregistered
%% event and successfully shutting down the pool.
try case is_pid(Pool) andalso
is_process_alive(Pool) andalso (?NORMAL_REASON(Reason))
of
true ->
riak_core_vnode_worker_pool:shutdown_pool(Pool, 60000);
_ -> ok
end
catch
Type:Reason:Stacktrace ->
logger:error("Error while shutting down vnode worker "
"pool ~p:~p trace : ~p",
[Type, Reason, Stacktrace])
after
case ModState of
%% Handoff completed, Module:delete has been called, now terminate.
{deleted, ModState1} ->
Module:terminate(Reason, ModState1);
_ -> Module:terminate(Reason, ModState)
end
end.
code_change(_OldVsn, StateName, State, _Extra) ->
{ok, StateName, State}.
%% ========================
%% ========
%% States
%% ========
%% ========================
%% started
%% ========
started(timeout,
State = #state{inactivity_timeout =
InitialInactivityTimeout}) ->
case do_init(State) of
{ok, State2} ->
{next_state, active, State2, InitialInactivityTimeout};
{error, Reason} -> {stop, Reason}
end.
started(wait_for_init, _From,
State = #state{inactivity_timeout =
InitialInactivityTimeout}) ->
case do_init(State) of
{ok, State2} ->
{reply, ok, active, State2, InitialInactivityTimeout};
{error, Reason} -> {stop, Reason}
end.
%%active
%%%%%%%%%%%%
active(timeout,
State = #state{mod = Module, index = Idx}) ->
riak_core_vnode_manager:vnode_event(Module,
Idx,
self(),
inactive),
continue(State);
active(#riak_coverage_req_v1{keyspaces = KeySpaces,
request = Request, sender = Sender},
State) ->
%% Coverage request handled in handoff and non-handoff. Will be forwarded if set.
vnode_coverage(Sender, Request, KeySpaces, State);
active(#riak_vnode_req_v1{sender = Sender,
request = {resize_forward, Request}},
State) ->
vnode_command(Sender, Request, State);
active(#riak_vnode_req_v1{sender = Sender,
request = Request},
State = #state{handoff_target = HT})
when HT =:= none ->
forward_or_vnode_command(Sender, Request, State);
active(#riak_vnode_req_v1{sender = Sender,
request = Request},
State = #state{handoff_type = resize,
handoff_target = {HOIdx, HONode}, index = Index,
forward = Forward, mod = Module}) ->
RequestHash = Module:request_hash(Request),
case RequestHash of
%% will never have enough information to forward request so only handle locally
undefined -> vnode_command(Sender, Request, State);
_ ->
{ok, R} = riak_core_ring_manager:get_my_ring(),
FutureIndex = riak_core_ring:future_index(RequestHash,
Index,
R),
case FutureIndex of
%% request for portion of keyspace currently being transferred
HOIdx ->
vnode_handoff_command(Sender,
Request,
{HOIdx, HONode},
State);
%% some portions of keyspace already transferred
_Other when is_list(Forward) ->
vnode_resize_command(Sender,
Request,
FutureIndex,
State);
%% some portions of keyspace not already transferred
_Other -> vnode_command(Sender, Request, State)
end
end;
active(#riak_vnode_req_v1{sender = Sender,
request = Request},
State) ->
vnode_handoff_command(Sender,
Request,
State#state.handoff_target,
State);
active(handoff_complete, State) ->
State2 = start_manager_event_timer(handoff_complete,
State),
continue(State2);
active({resize_transfer_complete, SeenIdxs},
State = #state{mod = Module, modstate = ModState,
handoff_target = Target}) ->
case Target of
none -> continue(State);
_ ->
%% TODO: refactor similarties w/ finish_handoff handle_event
{ok, NewModState} = Module:handoff_finished(Target,
ModState),
finish_handoff(SeenIdxs,
State#state{modstate = NewModState})
end;
active({handoff_error, _Err, _Reason}, State) ->
State2 = start_manager_event_timer(handoff_error,
State),
continue(State2);
active({send_manager_event, Event}, State) ->
State2 = start_manager_event_timer(Event, State),
continue(State2);
active({trigger_handoff, TargetNode}, State) ->
active({trigger_handoff, State#state.index, TargetNode},
State);
active({trigger_handoff, TargetIdx, TargetNode},
State) ->
maybe_handoff(TargetIdx, TargetNode, State);
active(trigger_delete,
State = #state{mod = Module, modstate = ModState,
index = Idx}) ->
case mark_delete_complete(Idx, Module) of
{ok, _NewRing} ->
{ok, NewModState} = Module:delete(ModState),
logger:debug("~p ~p vnode deleted", [Idx, Module]);
_ -> NewModState = ModState
end,
maybe_shutdown_pool(State),
riak_core_vnode_manager:unregister_vnode(Idx, Module),
continue(State#state{modstate =
{deleted, NewModState}});
active(unregistered,
State = #state{mod = Module, index = Index}) ->
%% Add exclusion so the ring handler will not try to spin this vnode
%% up until it receives traffic.
riak_core_handoff_manager:add_exclusion(Module, Index),
logger:debug("~p ~p vnode excluded and unregistered.",
[Index, Module]),
{stop,
normal,
State#state{handoff_target = none,
handoff_type = undefined, pool_pid = undefined}}.
active(_Event, _From, State) ->
Reply = ok,
{reply,
Reply,
active,
State,
State#state.inactivity_timeout}.
%% handle_event
%%%%%%%%%%%%%%%%
handle_event({set_forwarding, undefined}, _StateName,
State = #state{modstate = {deleted, _ModState}}) ->
%% The vnode must forward requests when in the deleted state, therefore
%% ignore requests to stop forwarding.
continue(State);
handle_event({set_forwarding, ForwardTo}, _StateName,
State) ->
logger:debug("vnode fwd :: ~p/~p :: ~p -> ~p~n",
[State#state.mod,
State#state.index,
State#state.forward,
ForwardTo]),
State2 = mod_set_forwarding(ForwardTo, State),
continue(State2#state{forward = ForwardTo});
handle_event(finish_handoff, _StateName,
State = #state{modstate = {deleted, _ModState}}) ->
stop_manager_event_timer(State),
continue(State#state{handoff_target = none});
handle_event(finish_handoff, _StateName,
State = #state{mod = Module, modstate = ModState,
handoff_target = Target}) ->
stop_manager_event_timer(State),
case Target of
none -> continue(State);
_ ->
{ok, NewModState} = Module:handoff_finished(Target,
ModState),
finish_handoff(State#state{modstate = NewModState})
end;
handle_event(cancel_handoff, _StateName,
State = #state{mod = Module, modstate = ModState}) ->
%% it would be nice to pass {Err, Reason} to the vnode but the
%% API doesn't currently allow for that.
stop_manager_event_timer(State),
case State#state.handoff_target of
none -> continue(State);
_ ->
{ok, NewModState} = Module:handoff_cancelled(ModState),
continue(State#state{handoff_target = none,
handoff_type = undefined,
modstate = NewModState})
end;
handle_event({trigger_handoff, TargetNode}, StateName,
State) ->
handle_event({trigger_handoff,
State#state.index,
TargetNode},
StateName,
State);
handle_event({trigger_handoff, _TargetIdx, _TargetNode},
_StateName,
State = #state{modstate = {deleted, _ModState}}) ->
continue(State);
handle_event(R = {trigger_handoff,
_TargetIdx,
_TargetNode},
_StateName, State) ->
active(R, State);
handle_event(trigger_delete, _StateName,
State = #state{modstate = {deleted, _}}) ->
continue(State);
handle_event(trigger_delete, _StateName, State) ->
active(trigger_delete, State);
handle_event(R = #riak_vnode_req_v1{}, _StateName,
State) ->
active(R, State);
handle_event(R = #riak_coverage_req_v1{}, _StateName,
State) ->
active(R, State).
%%handle_sync_event
%%%%%%%%%%%%%%%%%%%%
handle_sync_event(current_state, _From, StateName,
State) ->
{reply, {StateName, State}, StateName, State};
handle_sync_event(get_mod_index, _From, StateName,
State = #state{index = Idx, mod = Mod}) ->
{reply,
{Mod, Idx},
StateName,
State,
State#state.inactivity_timeout};
handle_sync_event({handoff_data, _BinObj}, _From,
StateName,
State = #state{modstate = {deleted, _ModState}}) ->
{reply,
{error, vnode_exiting},
StateName,
State,
State#state.inactivity_timeout};
handle_sync_event({handoff_data, BinObj}, _From,
StateName,
State = #state{mod = Module, modstate = ModState}) ->
case Module:handle_handoff_data(BinObj, ModState) of
{reply, ok, NewModState} ->
{reply,
ok,
StateName,
State#state{modstate = NewModState},
State#state.inactivity_timeout};
{reply, {error, Err}, NewModState} ->
logger:error("~p failed to store handoff obj: ~p",
[Module, Err]),
{reply,
{error, Err},
StateName,
State#state{modstate = NewModState},
State#state.inactivity_timeout}
end;
handle_sync_event(core_status, _From, StateName,
State = #state{index = Index, mod = Module,
modstate = ModState, handoff_target = HT,
forward = FN}) ->
Mode = case {FN, HT} of
{undefined, none} -> active;
{undefined, HT} -> handoff;
{FN, none} -> forward;
_ -> undefined
end,
Status = [{index, Index}, {mod, Module}] ++
case FN of
undefined -> [];
_ -> [{forward, FN}]
end
++
case HT of
none -> [];
_ -> [{handoff_target, HT}]
end
++
case ModState of
{deleted, _} -> [deleted];
_ -> []
end,
{reply,
{Mode, Status},
StateName,
State,
State#state.inactivity_timeout}.
%%handle_info
%%%%%%%%%%%%%%
handle_info({'$vnode_proxy_ping', From, Ref, Msgs},
StateName, State) ->
riak_core_vnode_proxy:cast(From,
{vnode_proxy_pong, Ref, Msgs}),
{next_state,
StateName,
State,
State#state.inactivity_timeout};
handle_info({'EXIT', Pid, Reason}, _StateName,
State = #state{mod = Module, index = Index,
pool_pid = Pid, pool_config = PoolConfig}) ->
case Reason of
Reason when Reason == normal; Reason == shutdown ->
continue(State#state{pool_pid = undefined});
_ ->
logger:error("~p ~p worker pool crashed ~p\n",
[Index, Module, Reason]),
{pool, WorkerModule, PoolSize, WorkerArgs} = PoolConfig,
logger:debug("starting worker pool ~p with size of "
"~p for vnode ~p.",
[WorkerModule, PoolSize, Index]),
{ok, NewPoolPid} =
riak_core_vnode_worker_pool:start_link(WorkerModule,
PoolSize,
Index,
WorkerArgs,
worker_props),
continue(State#state{pool_pid = NewPoolPid})
end;
handle_info({'DOWN', _Ref, process, _Pid, normal},
_StateName, State = #state{modstate = {deleted, _}}) ->
%% these messages are produced by riak_kv_vnode's aae tree
%% monitors; they are harmless, so don't yell about them. also
%% only dustbin them in the deleted modstate, because pipe vnodes
%% need them in other states
continue(State);
handle_info(Info, _StateName,
State = #state{mod = Module, modstate = {deleted, _},
index = Index}) ->
logger:info("~p ~p ignored handle_info ~p - vnode "
"unregistering\n",
[Index, Module, Info]),
continue(State);
handle_info({'EXIT', Pid, Reason}, StateName,
State = #state{mod = Module, modstate = ModState}) ->
%% A linked processes has died so use the
%% handle_exit callback to allow the vnode
%% process to take appropriate action.
%% If the function is not implemented default
%% to crashing the process.
try case Module:handle_exit(Pid, Reason, ModState) of
{noreply, NewModState} ->
{next_state,
StateName,
State#state{modstate = NewModState},
State#state.inactivity_timeout};
{stop, Reason1, NewModState} ->
{stop, Reason1, State#state{modstate = NewModState}}
end
catch
_ErrorType:undef -> {stop, linked_process_crash, State}
end;
handle_info(Info, StateName,
State = #state{mod = Module, modstate = ModState}) ->
case erlang:function_exported(Module, handle_info, 2) of
true ->
{ok, NewModState} = Module:handle_info(Info, ModState),
{next_state,
StateName,
State#state{modstate = NewModState},
State#state.inactivity_timeout};
false ->
{next_state,
StateName,
State,
State#state.inactivity_timeout}
end.
%% ========================
%% ========
%% Internal Helper Functions
%% ========
%% ========================
do_init(State = #state{index = Index, mod = Module,
forward = Forward}) ->
{ModState, Props} = case Module:init([Index]) of
{ok, MS} -> {MS, []};
{ok, MS, P} -> {MS, P};
{error, R} -> {error, R}
end,
case {ModState, Props} of
{error, Reason} -> {error, Reason};
_ ->
PoolConfig = case lists:keyfind(pool, 1, Props) of
{pool, WorkerModule, PoolSize, WorkerArgs} =
PoolCfg ->
logger:debug("starting worker pool ~p with size of "
"~p~n",
[WorkerModule, PoolSize]),
{ok, PoolPid} =
riak_core_vnode_worker_pool:start_link(WorkerModule,
PoolSize,
Index,
WorkerArgs,
worker_props),
PoolCfg;
_ -> PoolPid = undefined
end,
riak_core_handoff_manager:remove_exclusion(Module,
Index),
Timeout = application:get_env(riak_core,
vnode_inactivity_timeout,
?DEFAULT_TIMEOUT),
Timeout2 = Timeout + rand:uniform(Timeout),
State2 = State#state{modstate = ModState,
inactivity_timeout = Timeout2,
pool_pid = PoolPid, pool_config = PoolConfig},
logger:debug("vnode :: ~p/~p :: ~p~n",
[Module, Index, Forward]),
State3 = mod_set_forwarding(Forward, State2),
{ok, State3}
end.
continue(State) ->
{next_state,
active,
State,
State#state.inactivity_timeout}.
continue(State, NewModState) ->
continue(State#state{modstate = NewModState}).
%% Active vnodes operate in three states: normal, handoff, and forwarding.
%%
%% In the normal state, vnode commands are passed to handle_command. When
%% a handoff is triggered, handoff_target is set and the vnode
%% is said to be in the handoff state.
%%
%% In the handoff state, vnode commands are passed to handle_handoff_command.
%% However, a vnode may be blocked during handoff (and therefore not servicing
%% commands) if the handoff procedure is blocking (eg. in riak_kv when not
%% using async fold).
%%
%% After handoff, a vnode may move into forwarding state. The forwarding state
%% is a product of the new gossip/membership code and will not occur if the
%% node is running in legacy mode. The forwarding state represents the case
%% where the vnode has already handed its data off to the new owner, but the
%% new owner is not yet listed as the current owner in the ring. This may occur
%% because additional vnodes are still waiting to handoff their data to the
%% new owner, or simply because the ring has yet to converge on the new owner.
%% In the forwarding state, all vnode commands and coverage commands are
%% forwarded to the new owner for processing.
%%
%% The above becomes a bit more complicated when the vnode takes part in resizing
%% the ring, since several transfers with a single vnode as the source are necessary
%% to complete the operation. A vnode will remain in the handoff state, for, potentially,
%% more than one transfer and may be in the handoff state despite there being no active
%% transfers with this vnode as the source. During this time requests that can be forwarded
%% to a partition for which the transfer has already completed, are forwarded. All other
%% requests are passed to handle_handoff_command.
forward_or_vnode_command(Sender, Request,
State = #state{forward = Forward, mod = Module,
index = Index}) ->
Resizing = is_list(Forward),
RequestHash = case Resizing of
true -> Module:request_hash(Request);
false -> undefined
end,
case {Forward, RequestHash} of
%% typical vnode operation, no forwarding set, handle request locally
{undefined, _} -> vnode_command(Sender, Request, State);
%% implicit forwarding after ownership transfer/hinted handoff
{F, _} when not is_list(F) ->
vnode_forward(implicit,
{Index, Forward},
Sender,
Request,
State),
continue(State);
%% during resize we can't forward a request w/o request hash, always handle locally
{_, undefined} -> vnode_command(Sender, Request, State);
%% possible forwarding during ring resizing
{_, _} ->
{ok, R} = riak_core_ring_manager:get_my_ring(),
FutureIndex = riak_core_ring:future_index(RequestHash,
Index,
R),
vnode_resize_command(Sender,
Request,
FutureIndex,
State)
end.
vnode_command(_Sender, _Request,
State = #state{modstate = {deleted, _}}) ->
continue(State);
vnode_command(Sender, Request,
State = #state{mod = Module, modstate = ModState,
pool_pid = Pool}) ->
case catch Module:handle_command(Request,
Sender,
ModState)
of
{'EXIT', ExitReason} ->
reply(Sender, {vnode_error, ExitReason}),
logger:error("~p command failed ~p",
[Module, ExitReason]),
{stop, ExitReason, State#state{modstate = ModState}};
continue -> continue(State, ModState);
{reply, Reply, NewModState} ->
reply(Sender, Reply),
continue(State, NewModState);
{noreply, NewModState} -> continue(State, NewModState);
{async, Work, From, NewModState} ->
%% dispatch some work to the vnode worker pool
%% the result is sent back to 'From'
riak_core_vnode_worker_pool:handle_work(Pool,
Work,
From),
continue(State, NewModState);
{stop, Reason, NewModState} ->
{stop, Reason, State#state{modstate = NewModState}}
end.
vnode_coverage(Sender, Request, KeySpaces,
State = #state{index = Index, mod = Module,
modstate = ModState, pool_pid = Pool,
forward = Forward}) ->
%% Check if we should forward
case Forward of
undefined ->
Action = Module:handle_coverage(Request,
KeySpaces,
Sender,
ModState);
%% handle coverage requests locally during ring resize
Forwards when is_list(Forwards) ->
Action = Module:handle_coverage(Request,
KeySpaces,
Sender,
ModState);
NextOwner ->
logger:debug("Forwarding coverage ~p -> ~p: ~p~n",
[node(), NextOwner, Index]),
riak_core_vnode_master:coverage(Request,
{Index, NextOwner},
KeySpaces,
Sender,
riak_core_vnode_master:reg_name(Module)),
Action = continue
end,
case Action of
continue -> continue(State, ModState);
{reply, Reply, NewModState} ->
reply(Sender, Reply),
continue(State, NewModState);
{noreply, NewModState} -> continue(State, NewModState);
{async, Work, From, NewModState} ->
%% dispatch some work to the vnode worker pool
%% the result is sent back to 'From'
riak_core_vnode_worker_pool:handle_work(Pool,
Work,
From),
continue(State, NewModState);
{stop, Reason, NewModState} ->
{stop, Reason, State#state{modstate = NewModState}}
end.
vnode_handoff_command(Sender, Request, ForwardTo,
State = #state{mod = Module, modstate = ModState,
handoff_target = HOTarget,
handoff_type = HOType, pool_pid = Pool}) ->
case Module:handle_handoff_command(Request,
Sender,
ModState)
of
{reply, Reply, NewModState} ->
reply(Sender, Reply),
continue(State, NewModState);
{noreply, NewModState} -> continue(State, NewModState);
{async, Work, From, NewModState} ->
%% dispatch some work to the vnode worker pool
%% the result is sent back to 'From'
riak_core_vnode_worker_pool:handle_work(Pool,
Work,
From),
continue(State, NewModState);
{forward, NewModState} ->
forward_request(HOType,
Request,
HOTarget,
ForwardTo,
Sender,
State),
continue(State, NewModState);
{forward, NewReq, NewModState} ->
forward_request(HOType,
NewReq,
HOTarget,
ForwardTo,
Sender,
State),
continue(State, NewModState);
{drop, NewModState} -> continue(State, NewModState);
{stop, Reason, NewModState} ->
{stop, Reason, State#state{modstate = NewModState}}
end.
%% @private wrap the request for resize forwards, and use the resize
%% target.
forward_request(resize, Request, _HOTarget,
ResizeTarget, Sender, State) ->
%% resize op and transfer ongoing
vnode_forward(resize,
ResizeTarget,
Sender,
{resize_forward, Request},
State);
forward_request(undefined, Request, _HOTarget,
ResizeTarget, Sender, State) ->
%% resize op ongoing, no resize transfer ongoing, arrive here
%% via forward_or_vnode_command
vnode_forward(resize,
ResizeTarget,
Sender,
{resize_forward, Request},
State);
forward_request(_, Request, HOTarget, _ResizeTarget,
Sender, State) ->
%% normal explicit forwarding during owhership transfer
vnode_forward(explicit,
HOTarget,
Sender,
Request,
State).
vnode_forward(Type, ForwardTo, Sender, Request,
State) ->
logger:debug("Forwarding (~p) {~p,~p} -> ~p~n",
[Type, State#state.index, node(), ForwardTo]),
riak_core_vnode_master:command_unreliable(ForwardTo,
Request,
Sender,
riak_core_vnode_master:reg_name(State#state.mod)).
%% @doc during ring resizing if we have completed a transfer to the index that will
%% handle request in future ring we forward to it. Otherwise we delegate
%% to the local vnode like other requests during handoff
vnode_resize_command(Sender, Request, FutureIndex,
State = #state{forward = Forward})
when is_list(Forward) ->
case lists:keyfind(FutureIndex, 1, Forward) of
false -> vnode_command(Sender, Request, State);
{FutureIndex, FutureOwner} ->
vnode_handoff_command(Sender,
Request,
{FutureIndex, FutureOwner},
State)
end.
%% This code lives in riak_core_vnode rather than riak_core_vnode_manager
%% because the ring_trans call is a synchronous call to the ring manager,
%% and it is better to block an individual vnode rather than the vnode
%% manager. Blocking the manager can impact all vnodes. This code is safe
%% to execute on multiple parallel vnodes because of the synchronization
%% afforded by having all ring changes go through the single ring manager.
mark_handoff_complete(SrcIdx, Target, SeenIdxs, Mod,
resize) ->
Prev = node(),
Source = {SrcIdx, Prev},
TransFun = fun (Ring, _) ->
Owner = riak_core_ring:index_owner(Ring, SrcIdx),
Status = riak_core_ring:resize_transfer_status(Ring,
Source,
Target,
Mod),
case {Owner, Status} of
{Prev, awaiting} ->
F = fun (SeenIdx, RingAcc) ->
riak_core_ring:schedule_resize_transfer(RingAcc,
Source,
SeenIdx)
end,
Ring2 = lists:foldl(F,
Ring,
ordsets:to_list(SeenIdxs)),
Ring3 =
riak_core_ring:resize_transfer_complete(Ring2,
Source,
Target,
Mod),
%% local ring optimization (see below)
{set_only, Ring3};
_ -> ignore
end
end,
Result = riak_core_ring_manager:ring_trans(TransFun,
[]),
case Result of
{ok, _NewRing} -> resize;
_ -> continue
end;
mark_handoff_complete(Idx, {Idx, New}, [], Mod, _) ->
Prev = node(),
Result = riak_core_ring_manager:ring_trans(fun (Ring,
_) ->
Owner =
riak_core_ring:index_owner(Ring,
Idx),
{_, NextOwner, Status} =
riak_core_ring:next_owner(Ring,
Idx,
Mod),
NewStatus =
riak_core_ring:member_status(Ring,
New),
case {Owner,
NextOwner,
NewStatus,
Status}
of
{Prev,
New,
_,
awaiting} ->
Ring2 =
riak_core_ring:handoff_complete(Ring,
Idx,
Mod),
%% Optimization. Only alter the local ring without
%% triggering a gossip, thus implicitly coalescing
%% multiple vnode handoff completion events. In the
%% future we should decouple vnode handoff state from
%% the ring structure in order to make gossip independent
%% of ring size.
{set_only,
Ring2};
_ -> ignore
end
end,
[]),
case Result of
{ok, NewRing} -> NewRing = NewRing;
_ ->
{ok, NewRing} = riak_core_ring_manager:get_my_ring()
end,
Owner = riak_core_ring:index_owner(NewRing, Idx),
{_, NextOwner, Status} =
riak_core_ring:next_owner(NewRing, Idx, Mod),
NewStatus = riak_core_ring:member_status(NewRing, New),
case {Owner, NextOwner, NewStatus, Status} of
{_, _, invalid, _} ->
%% Handing off to invalid node, don't give-up data.
continue;
{Prev, New, _, _} -> forward;
{Prev, _, _, _} ->
%% Handoff wasn't to node that is scheduled in next, so no change.
continue;
{_, _, _, _} -> shutdown
end.
finish_handoff(State) -> finish_handoff([], State).
finish_handoff(SeenIdxs,
State = #state{mod = Module, modstate = ModState,
index = Idx, handoff_target = Target,
handoff_type = HOType}) ->
case mark_handoff_complete(Idx,
Target,
SeenIdxs,
Module,
HOType)
of
continue ->
continue(State#state{handoff_target = none,
handoff_type = undefined});
resize ->
CurrentForwarding = resize_forwarding(State),
NewForwarding = [Target | CurrentForwarding],
State2 = mod_set_forwarding(NewForwarding, State),
continue(State2#state{handoff_target = none,
handoff_type = undefined,
forward = NewForwarding});
Res when Res == forward; Res == shutdown ->
{_, HN} = Target,
%% Have to issue the delete now. Once unregistered the
%% vnode master will spin up a new vnode on demand.
%% Shutdown the async pool beforehand, don't want callbacks
%% running on non-existant data.
maybe_shutdown_pool(State),
{ok, NewModState} = Module:delete(ModState),
logger:debug("~p ~p vnode finished handoff and deleted.",
[Idx, Module]),
riak_core_vnode_manager:unregister_vnode(Idx, Module),
logger:debug("vnode hn/fwd :: ~p/~p :: ~p -> ~p~n",
[State#state.mod,
State#state.index,
State#state.forward,
HN]),
State2 = mod_set_forwarding(HN, State),
continue(State2#state{modstate =
{deleted,
NewModState}, % like to fail if used
handoff_target = none,
handoff_type = undefined, forward = HN})
end.
maybe_shutdown_pool(#state{pool_pid = Pool}) ->
case is_pid(Pool) of
true ->
%% state.pool_pid will be cleaned up by handle_info message.
riak_core_vnode_worker_pool:shutdown_pool(Pool, 60000);
_ -> ok
end.
resize_forwarding(#state{forward = F})
when is_list(F) ->
F;
resize_forwarding(_) -> [].
mark_delete_complete(Idx, Mod) ->
Result = riak_core_ring_manager:ring_trans(fun (Ring,
_) ->
Type =
riak_core_ring:vnode_type(Ring,
Idx),
{_, Next, Status} =
riak_core_ring:next_owner(Ring,
Idx),
case {Type, Next, Status}
of
{resized_primary,
'$delete',
awaiting} ->
Ring3 =
riak_core_ring:deletion_complete(Ring,
Idx,
Mod),
%% Use local ring optimization like mark_handoff_complete
{set_only,
Ring3};
{{fallback, _},
'$delete',
awaiting} ->
Ring3 =
riak_core_ring:deletion_complete(Ring,
Idx,
Mod),
%% Use local ring optimization like mark_handoff_complete
{set_only,
Ring3};
_ -> ignore
end
end,
[]),
Result.
maybe_handoff(_TargetIdx, _TargetNode,
State = #state{modstate = {deleted, _}}) ->
%% Modstate has been deleted, waiting for unregistered. No handoff.
continue(State);
maybe_handoff(TargetIdx, TargetNode,
State = #state{index = Idx, mod = Module,
modstate = ModState,
handoff_target = CurrentTarget,
handoff_pid = HPid}) ->
Target = {TargetIdx, TargetNode},
ExistingHO = is_pid(HPid) andalso
is_process_alive(HPid),
ValidHN = case CurrentTarget of
none -> true;
Target -> not ExistingHO;
_ ->
logger:info("~s/~b: handoff request to ~p before "
"finishing handoff to ~p",
[Module, Idx, Target, CurrentTarget]),
not ExistingHO
end,
case ValidHN of
true ->
{ok, R} = riak_core_ring_manager:get_my_ring(),
Resizing = riak_core_ring:is_resizing(R),
Primary = riak_core_ring:is_primary(R, {Idx, node()}),
HOType = case {Resizing, Primary} of
{true, _} -> resize;
{_, true} -> ownership;
{_, false} -> hinted
end,
case Module:handoff_starting({HOType, Target}, ModState)
of
{true, NewModState} ->
start_handoff(HOType,
TargetIdx,
TargetNode,
State#state{modstate = NewModState});
{false, NewModState} -> continue(State, NewModState)
end;
false -> continue(State)
end.
start_handoff(HOType, TargetIdx, TargetNode,
State = #state{mod = Module, modstate = ModState}) ->
case Module:is_empty(ModState) of
{true, NewModState} ->
finish_handoff(State#state{modstate = NewModState,
handoff_type = HOType,
handoff_target =
{TargetIdx, TargetNode}});
{false, Size, NewModState} ->
State2 = State#state{modstate = NewModState},
NewState = start_outbound(HOType,
TargetIdx,
TargetNode,
[{size, Size}],
State2),
continue(NewState);
{false, NewModState} ->
State2 = State#state{modstate = NewModState},
NewState = start_outbound(HOType,
TargetIdx,
TargetNode,
[],
State2),
continue(NewState)
end.
start_outbound(HOType, TargetIdx, TargetNode, Opts,
State = #state{index = Idx, mod = Module}) ->
case riak_core_handoff_manager:add_outbound(HOType,
Module,
Idx,
TargetIdx,
TargetNode,
self(),
Opts)
of
{ok, Pid} ->
State#state{handoff_pid = Pid, handoff_type = HOType,
handoff_target = {TargetIdx, TargetNode}};
{error, _Reason} ->
{ok, NewModState} =
Module:handoff_cancelled(State#state.modstate),
State#state{modstate = NewModState}
end.
%% Individual vnode processes and the vnode manager are tightly coupled. When
%% vnode events occur, the vnode must ensure that the events are forwarded to
%% the vnode manager, which will make a state change decision and send an
%% appropriate message back to the vnode. To minimize blocking, asynchronous
%% messaging is used. It is possible for the vnode manager to crash and miss
%% messages sent by the vnode. Therefore, the vnode periodically resends event
%% messages until an appropriate message is received back from the vnode
%% manager. The event timer functions below implement this logic.
start_manager_event_timer(Event,
State = #state{mod = Module, index = Idx}) ->
riak_core_vnode_manager:vnode_event(Module,
Idx,
self(),
Event),
stop_manager_event_timer(State),
T2 = gen_fsm_compat:send_event_after(30000,
{send_manager_event, Event}),
State#state{manager_event_timer = T2}.
stop_manager_event_timer(#state{manager_event_timer =
undefined}) ->
ok;
stop_manager_event_timer(#state{manager_event_timer =
T}) ->
_ = gen_fsm_compat:cancel_timer(T),
ok.
mod_set_forwarding(_Forward,
State = #state{modstate = {deleted, _}}) ->
State;
mod_set_forwarding(Forward,
State = #state{mod = Module, modstate = ModState}) ->
case lists:member({set_vnode_forwarding, 2},
Module:module_info(exports))
of
true ->
NewModState = Module:set_vnode_forwarding(Forward,
ModState),
State#state{modstate = NewModState};
false -> State
end.
%% ===================================================================
%% Test API
%% ===================================================================
-type state() :: #state{}.
%% @doc Reveal the underlying module state for testing
-spec get_modstate(pid()) -> {atom(), state()}.
get_modstate(Pid) ->
{_StateName, State} =
gen_fsm_compat:sync_send_all_state_event(Pid,
current_state),
{State#state.mod, State#state.modstate}.
-ifdef(TEST).
%% Start the garbage collection server
test_link(Mod, Index) ->
gen_fsm_compat:start_link(?MODULE,
[Mod, Index, 0, node()],
[]).
%% Get the current state of the fsm for testing inspection
-spec current_state(pid()) -> {atom(), state()} |
{error, term()}.
current_state(Pid) ->
gen_fsm_compat:sync_send_all_state_event(Pid,
current_state).
wait_for_process_death(Pid) ->
wait_for_process_death(Pid, is_process_alive(Pid)).
wait_for_process_death(Pid, true) ->
wait_for_process_death(Pid, is_process_alive(Pid));
wait_for_process_death(_Pid, false) -> ok.
wait_for_state_update(OriginalStateData, Pid) ->
{_, CurrentStateData} = (?MODULE):current_state(Pid),
wait_for_state_update(OriginalStateData,
CurrentStateData,
Pid).
wait_for_state_update(OriginalStateData,
OriginalStateData, Pid) ->
{_, CurrentStateData} = (?MODULE):current_state(Pid),
wait_for_state_update(OriginalStateData,
CurrentStateData,
Pid);
wait_for_state_update(_OriginalState, _StateData,
_Pid) ->
ok.
%% ===================================================================
%% Test
%% ===================================================================
pool_death_test() ->
%% expect error log
error_logger:tty(false),
meck:unload(),
meck:new(test_vnode, [non_strict, no_link]),
meck:expect(test_vnode,
init,
fun (_) -> {ok, [], [{pool, test_pool_mod, 1, []}]}
end),
meck:expect(test_vnode,
terminate,
fun (_, _) -> normal end),
meck:new(test_pool_mod, [non_strict, no_link]),
meck:expect(test_pool_mod,
init_worker,
fun (_, _, _) -> {ok, []} end),
{ok, Pid} = riak_core_vnode:test_link(test_vnode, 0),
{_, StateData1} = riak_core_vnode:current_state(Pid),
PoolPid1 = StateData1#state.pool_pid,
exit(PoolPid1, kill),
wait_for_process_death(PoolPid1),
?assertNot((is_process_alive(PoolPid1))),
wait_for_state_update(StateData1, Pid),
{_, StateData2} = riak_core_vnode:current_state(Pid),
PoolPid2 = StateData2#state.pool_pid,
?assertNot((PoolPid2 =:= undefined)),
exit(Pid, normal),
wait_for_process_death(Pid),
meck:validate(test_pool_mod),
meck:validate(test_vnode).
-endif.