gen_server源码阅读

gen_server是Erlang/OTP的通用服务器框架,使用最为广泛。源代码很少,只有几百行。抽空详细梳理了一遍流程,以简单的gen_server应用worker为例。

先看我们自己的worker.erl文件,根据OTP设计规范,既是回调模块,又是用户接口。对外导出 start_link/0,1 , query/1 , set/1 , show/1 , *stop/1*,同时实现gen_server行为模式的6个必要的回调函数:

%% gen_server

-module(worker).
-behaviour(gen_server).

-export([start_link/1, start_link/0, query/1, set/1, show/1, stop/0]).
-export([init/1,handle_call/3,handle_cast/2,handle_info/2,terminate/2,code_change/3]).

-record(state, {key::term(), value::term()}).

-define(SERVER, ?MODULE).
-define(PARAM, {'0x01f', 'test'}).

%% export
start_link() ->
    start_link(?PARAM).
start_link(Param) ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [Param], [{timeout, 10}]).
stop() ->
    gen_server:stop(?SERVER).
query(Key) ->
    gen_server:call(?SERVER, {query, Key}).
set({Key, Value}) ->
    gen_server:call(?SERVER, {set, Key, Value}).
show(Key) ->
    gen_server:cast(?SERVER, {show, Key}).

%% callbacks
init([{Key, Value}]) ->
    io:format("Saving key: ~p with value ~p, self: ~p~n", [Key, Value, self()]),
    {ok, #state{key=Key, value=Value}, 0}.

handle_call({query,Key}, From, State=#state{key=Key, value=Value}) ->
    io:format("query from ~p, self: ~p~n", [From, self()]),
    {reply, {ok, Value}, State};
handle_call({set,Key,Value}, From, State) ->
    io:format("set from ~p, self: ~p~n", [From, self()]),
    {reply, {ok, {Key, Value}}, State#state{key=Key, value=Value}}.

handle_cast({show, Key}, State=#state{key=Key, value=Value}) ->
    io:format("show, key ~p value ~p, self: ~p~n", [Key, Value, self()]),
    {noreply, State}.

handle_info(timeout, State) ->
    io:format("Timeout~n"),
    {noreply, State}.

terminate(Reason, _State) ->
    io:format("terminate reason ~p, self: ~p~n", [Reason, self()]),
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

有必要先看一下gen_server的注释和回调函数定义,简单介绍了gen_server设计理念和调用流程。源码永远比任何解释要精确。

%% gen_server.erl

%%% ---------------------------------------------------
%%%
%%% The idea behind THIS server is that the user module
%%% provides (different) functions to handle different
%%% kind of inputs. 
%%% If the Parent process terminates the Module:terminate/2
%%% function is called.
%%%
%%% The user module should export:
%%%
%%%   init(Args)  
%%%     ==> {ok, State}
%%%         {ok, State, Timeout}
%%%         ignore
%%%         {stop, Reason}
%%%
%%%   handle_call(Msg, {From, Tag}, State)
%%%
%%%    ==> {reply, Reply, State}
%%%        {reply, Reply, State, Timeout}
%%%        {noreply, State}
%%%        {noreply, State, Timeout}
%%%        {stop, Reason, Reply, State}  
%%%              Reason = normal | shutdown | Term terminate(State) is called
%%%
%%%   handle_cast(Msg, State)
%%%
%%%    ==> {noreply, State}
%%%        {noreply, State, Timeout}
%%%        {stop, Reason, State} 
%%%              Reason = normal | shutdown | Term terminate(State) is called
%%%
%%%   handle_info(Info, State) Info is e.g. {'EXIT', P, R}, {nodedown, N}, ...
%%%
%%%    ==> {noreply, State}
%%%        {noreply, State, Timeout}
%%%        {stop, Reason, State} 
%%%              Reason = normal | shutdown | Term, terminate(State) is called
%%%
%%%   terminate(Reason, State) Let the user module clean up
%%%        always called when server terminates
%%%
%%%    ==> ok
%%%
%%%
%%% The work flow (of the server) can be described as follows:
%%%
%%%   User module                          Generic
%%%   -----------                          -------
%%%     start            ----->             start
%%%     init             <-----              .
%%%
%%%                                         loop
%%%     handle_call      <-----              .
%%%                      ----->             reply
%%%
%%%     handle_cast      <-----              .
%%%
%%%     handle_info      <-----              .
%%%
%%%     terminate        <-----              .
%%%
%%%                      ----->             reply
%%%
%%%
%%% ---------------------------------------------------

-callback init(Args :: term()) ->
    {ok, State :: term()} | {ok, State :: term(), timeout() | hibernate} |
    {stop, Reason :: term()} | ignore.
-callback handle_call(Request :: term(), From :: {pid(), Tag :: term()},
                      State :: term()) ->
    {reply, Reply :: term(), NewState :: term()} |
    {reply, Reply :: term(), NewState :: term(), timeout() | hibernate} |
    {noreply, NewState :: term()} |
    {noreply, NewState :: term(), timeout() | hibernate} |
    {stop, Reason :: term(), Reply :: term(), NewState :: term()} |
    {stop, Reason :: term(), NewState :: term()}.
-callback handle_cast(Request :: term(), State :: term()) ->
    {noreply, NewState :: term()} |
    {noreply, NewState :: term(), timeout() | hibernate} |
    {stop, Reason :: term(), NewState :: term()}.
-callback handle_info(Info :: timeout | term(), State :: term()) ->
    {noreply, NewState :: term()} |
    {noreply, NewState :: term(), timeout() | hibernate} |
    {stop, Reason :: term(), NewState :: term()}.
-callback terminate(Reason :: (normal | shutdown | {shutdown, term()} |
                               term()),
                    State :: term()) ->
    term().
-callback code_change(OldVsn :: (term() | {down, term()}), State :: term(),
                      Extra :: term()) ->
    {ok, NewState :: term()} | {error, Reason :: term()}.
-callback format_status(Opt, StatusData) -> Status when
      Opt :: 'normal' | 'terminate',
      StatusData :: [PDict | State],
      PDict :: [{Key :: term(), Value :: term()}],
      State :: term(),
      Status :: term().

-optional_callbacks([format_status/2]).

gen_server(或者其他模式)的核心思想就是用户提供回调函数,gen_server框架提供进程管理、消息分发等,尽可能的把功能无关的代码封装在背后。

现在开始解释我们的五个导出函数如何和gen_server框架以及Erlang交互。

启动

gen_server:start_link({local, ?SERVER}, ?MODULE, [Param], [{timeout, 10}]).

调用gen_server模块的start_link/4,具体到代码:

%% gen_server.erl
start_link(Name, Mod, Args, Options) ->
    gen:start(?MODULE, link, Name, Mod, Args, Options).

又进一步调用了gen模块的start/6函数,link 表示链接程序,gen_server:start/3,4 调用时会传入nolink。注意传给gen:start的第一个参数是 ?MODULE, 即 gen_server原子。

%% gen.erl

%%-----------------------------------------------------------------
%% Starts a generic process.
%% start(GenMod, LinkP, Mod, Args, Options)
%% start(GenMod, LinkP, Name, Mod, Args, Options)
%%    GenMod = atom(), callback module implementing the 'real' fsm
%%    LinkP = link | nolink
%%    Name = {local, atom()} | {global, term()} | {via, atom(), term()}
%%    Args = term(), init arguments (to Mod:init/1)
%%    Options = [{timeout, Timeout} | {debug, [Flag]} | {spawn_opt, OptionList}]
%%      Flag = trace | log | {logfile, File} | statistics | debug
%%          (debug == log && statistics)
%% Returns: {ok, Pid} | ignore |{error, Reason} |
%%          {error, {already_started, Pid}} |
%%    The 'already_started' is returned only if Name is given 
%%-----------------------------------------------------------------

-spec start(module(), linkage(), emgr_name(), module(), term(), options()) ->
	start_ret().

start(GenMod, LinkP, Name, Mod, Args, Options) ->
    case where(Name) of
	undefined ->
	    do_spawn(GenMod, LinkP, Name, Mod, Args, Options);
	Pid ->
	    {error, {already_started, Pid}}
    end.
	
where({global, Name}) -> global:whereis_name(Name);
where({via, Module, Name}) -> Module:whereis_name(Name);
where({local, Name})  -> whereis(Name).

start函数首先判断Name是否已经被注册,这里判断的是worker原子,提供注册名的gen_server进程只能存在一个。

随后调用do_spawn/6函数,参数不变。GenMode是gen_server、LinkP是link、Name是{local, worker}、Mod是我们自己的模块名worker、Args我们传入的参数[Param]、Options是[{timeout, 10}]。

%% gen.erl
do_spawn(GenMod, link, Name, Mod, Args, Options) ->
    Time = timeout(Options),
    proc_lib:start_link(?MODULE, init_it,
			[GenMod, self(), self(), Name, Mod, Args, Options],
			Time,
			spawn_opts(Options));

timeout(Options) ->
    case lists:keyfind(timeout, 1, Options) of
	{_,Time} ->
	    Time;
	false ->
	    infinity
    end.

gen:do_spawn调用proc_lib:start_link/5,传入的?MODULE即gen,Time为10,我们自己通过worker传入的参数被放到第三个列表中。

%% proc_lib.erl
-spec start_link(Module, Function, Args, Time, SpawnOpts) -> Ret when
      Module :: module(),
      Function :: atom(),
      Args :: [term()],
      Time :: timeout(),
      SpawnOpts :: [spawn_option()],
      Ret :: term() | {error, Reason :: term()}.

start_link(M,F,A,Timeout,SpawnOpts) when is_atom(M), is_atom(F), is_list(A) ->
    Pid = ?MODULE:spawn_opt(M, F, A, ensure_link(SpawnOpts)),
    sync_wait(Pid, Timeout).

sync_wait(Pid, Timeout) ->
    receive
	{ack, Pid, Return} ->
	    Return;
	{'EXIT', Pid, Reason} ->
	    {error, Reason}
    after Timeout ->
	    unlink(Pid),
	    exit(Pid, kill),
	    flush(Pid),
	    {error, timeout}
    end.
	
ensure_link(SpawnOpts) ->
    case lists:member(link, SpawnOpts) of
	true -> 
	    SpawnOpts;
	false ->
	    [link|SpawnOpts]
    end.

这里首先通过调用proc_lib:spawn_opt/4创建一个进程,然后调用sync_wait(Pid, Timeout)同步调用,ensure_link确保link选项存在。sync_wait的receive消息接收我们回头介绍,先看进程的创建。

%% proc_lib.erl
-spec spawn_opt(Module, Function, Args, SpawnOpts) -> pid() when
      Module :: module(),
      Function :: atom(),
      Args :: [term()],
      SpawnOpts :: [spawn_option()].

spawn_opt(M, F, A, Opts) when is_atom(M), is_atom(F), is_list(A) ->
    Parent = get_my_name(),
    Ancestors = get_ancestors(),
    check_for_monitor(Opts),
    erlang:spawn_opt(?MODULE, init_p, [Parent,Ancestors,M,F,A], Opts).
	
get_my_name() ->
    case proc_info(self(),registered_name) of
	{registered_name,Name} -> Name;
	_                      -> self()
    end.

-spec get_ancestors() -> [pid()].

get_ancestors() ->
    case get('$ancestors') of
	A when is_list(A) -> A;
	_                 -> []
    end.
	
proc_info(Pid,Item) when node(Pid) =:= node() ->
    process_info(Pid,Item);
proc_info(Pid,Item) ->
    case lists:member(node(Pid),nodes()) of
	true ->
	    check(rpc:call(node(Pid), erlang, process_info, [Pid, Item]));
	_ ->
	    hidden
    end.

check({badrpc,nodedown}) -> undefined;
check({badrpc,Error})    -> Error;
check(Res)               -> Res.

check_for_monitor(SpawnOpts) ->
    case lists:member(monitor, SpawnOpts) of
	true ->
	    erlang:error(badarg);
	false ->
	    false
    end.

get_my_name返回当前进程pid或注册名,get_ancestors返回祖先进程列表,check_for_monitor确保monitor选项不存在。然后调用系统函数erlang:spawn_opt生成新进程。新的进程执行proc_lib:init_p(Parent, Ancestors, M, F, A)。

%% proc_lib.erl

-spec init_p(pid(), [pid()], atom(), atom(), [term()]) -> term().

init_p(Parent, Ancestors, M, F, A) when is_atom(M), is_atom(F), is_list(A) ->
    put('$ancestors', [Parent|Ancestors]),
    put('$initial_call', trans_init(M, F, A)),
    init_p_do_apply(M, F, A).

init_p_do_apply(M, F, A) ->
    try
	apply(M, F, A) 
    catch
	Class:Reason ->
	    exit_p(Class, Reason)
    end.
	
%% -----------------------------------------------------
%% Translate the initial call to some useful information.
%% -----------------------------------------------------

trans_init(gen,init_it,[gen_server,_,_,supervisor,{_,Module,_},_]) ->
    {supervisor,Module,1};
trans_init(gen,init_it,[gen_server,_,_,_,supervisor,{_,Module,_},_]) ->
    {supervisor,Module,1};
trans_init(gen,init_it,[gen_server,_,_,supervisor_bridge,[Module|_],_]) ->
    {supervisor_bridge,Module,1};
trans_init(gen,init_it,[gen_server,_,_,_,supervisor_bridge,[Module|_],_]) ->
    {supervisor_bridge,Module,1};
trans_init(gen,init_it,[gen_event|_]) ->
    {gen_event,init_it,6};
trans_init(gen,init_it,[_GenMod,_,_,Module,_,_]) when is_atom(Module) ->
    {Module,init,1};
trans_init(gen,init_it,[_GenMod,_,_,_,Module|_]) when is_atom(Module) ->
    {Module,init,1};
trans_init(M, F, A) when is_atom(M), is_atom(F) ->
    {M,F,length(A)}.

init_p首先更新新进程的祖先进程’$ancestors’环境变量和’$initial_call’,trans_init只是做一个简单的转换,这里的M 是 gen,F 是 init_it,A 是上面详细介绍过的参数[GenMod, self(), self(), Name, Mod, Args, Options],因此返回{Module, init, 1}。

init_p_do_apply直接调用M:F(A),即gen:init_it(GenMod, self(), self(), Name, Mod, Args, Options),注意这里的self()并不是新进程的pid,而是父进程pid,也就是调用sync_wait进程的pid,sync_wait的第一个参数Pid才是执行gen:init_it的新进程的pid。目前为止,其实只涉及到两个进程——运行gen_server框架的进程、新生成的进程。

回到gen模块,在新进程上下文执行init_it/7。

%%-----------------------------------------------------------------
%% Initiate the new process.
%% Register the name using the Rfunc function
%% Calls the Mod:init/Args function.
%% Finally an acknowledge is sent to Parent and the main
%% loop is entered.
%%-----------------------------------------------------------------
init_it(GenMod, Starter, Parent, Mod, Args, Options) ->
    init_it2(GenMod, Starter, Parent, self(), Mod, Args, Options).

init_it(GenMod, Starter, Parent, Name, Mod, Args, Options) ->
    case register_name(Name) of
	true ->
	    init_it2(GenMod, Starter, Parent, Name, Mod, Args, Options);
	{false, Pid} ->
	    proc_lib:init_ack(Starter, {error, {already_started, Pid}})
    end.

init_it2(GenMod, Starter, Parent, Name, Mod, Args, Options) ->
    GenMod:init_it(Starter, Parent, Name, Mod, Args, Options).

这里做了简单解释——初始化新进程、注册为{local, worker}、给父进程发送确认、进入loop循环。

init_it2调用GenMod:init_it(Starter, Parent, Name, Mod, Args, Options),Starter是父进程pid,而Parent在链接时跟Starter一样,不链接时是原子self。GenMod是gen_server,回到gen_server模块。

%%% ---------------------------------------------------
%%% Initiate the new process.
%%% Register the name using the Rfunc function
%%% Calls the Mod:init/Args function.
%%% Finally an acknowledge is sent to Parent and the main
%%% loop is entered.
%%% ---------------------------------------------------
init_it(Starter, self, Name, Mod, Args, Options) ->
    init_it(Starter, self(), Name, Mod, Args, Options);
init_it(Starter, Parent, Name0, Mod, Args, Options) ->
    Name = gen:name(Name0),
    Debug = gen:debug_options(Name, Options),
    case catch Mod:init(Args) of
	{ok, State} ->
	    proc_lib:init_ack(Starter, {ok, self()}), 	    
	    loop(Parent, Name, State, Mod, infinity, Debug);
	{ok, State, Timeout} ->
	    proc_lib:init_ack(Starter, {ok, self()}), 	    
	    loop(Parent, Name, State, Mod, Timeout, Debug);
	{stop, Reason} ->
	    %% For consistency, we must make sure that the
	    %% registered name (if any) is unregistered before
	    %% the parent process is notified about the failure.
	    %% (Otherwise, the parent process could get
	    %% an 'already_started' error if it immediately
	    %% tried starting the process again.)
	    gen:unregister_name(Name0),
	    proc_lib:init_ack(Starter, {error, Reason}),
	    exit(Reason);
	ignore ->
	    gen:unregister_name(Name0),
	    proc_lib:init_ack(Starter, ignore),
	    exit(normal);
	{'EXIT', Reason} ->
	    gen:unregister_name(Name0),
	    proc_lib:init_ack(Starter, {error, Reason}),
	    exit(Reason);
	Else ->
	    Error = {bad_return_value, Else},
	    proc_lib:init_ack(Starter, {error, Error}),
	    exit(Error)
    end.

调用gen:name返回注册名,在这里是worker;gen:debug_options和调试相关,先不讨论。有意思的是这里用到了老的catch风格代码,Mod:init(Args)即worker:init([Param]),终于回调到了我们自己写的代码。根据返回值,调用proc_lib:init_ack给Starter进程发送消息。

%% proc_lib.erl

-spec init_ack(Parent, Ret) -> 'ok' when
      Parent :: pid(),
      Ret :: term().

init_ack(Parent, Return) ->
    Parent ! {ack, self(), Return},
    ok.
	
sync_wait(Pid, Timeout) ->
    receive
	{ack, Pid, Return} ->
	    Return;
	{'EXIT', Pid, Reason} ->
	    {error, Reason}
    after Timeout ->
	    unlink(Pid),
	    exit(Pid, kill),
	    flush(Pid),
	    {error, timeout}
    end.

早先提到的proc_lib:start_link调用sync_wait等待确认,这时会根据收到的消息执行不同的操作,直接返回或者终止进程。我们传入的超时值,在这里发挥作用。

至此gen_server进程启动。简单总结:worker:start_link -> gen_server:start_link -> gen:start -> gen:do_spawn -> proc_lib:start_link -> proc_lib:spawn_opt -> erlang:spawn_opt => new process -> proc_lib:init_p -> gen:init_it -> gen_server:init_it -> worker:init -> send ack to parent -> gen_server:loop。

loop循环

总体来说,gen_server的启动还是很简单的,只是一连串的调用流程,最后进入gen_server 的loop循环。

loop(Parent, Name, State, Mod, Time, Debug) ->
    Msg = receive
	      Input ->
		    Input
	  after Time ->
		  timeout
	  end,
    decode_msg(Msg, Parent, Name, State, Mod, Time, Debug, false).
	
decode_msg(Msg, Parent, Name, State, Mod, Time, Debug, Hib) ->
    case Msg of
	{system, From, Req} ->
	    sys:handle_system_msg(Req, From, Parent, ?MODULE, Debug,
				  [Name, State, Mod, Time], Hib);
	{'EXIT', Parent, Reason} ->
	    terminate(Reason, Name, Msg, Mod, State, Debug);
	_Msg when Debug =:= [] ->
	    handle_msg(Msg, Parent, Name, State, Mod);
	_Msg ->
	    Debug1 = sys:handle_debug(Debug, fun print_event/3,
				      Name, {in, Msg}),
	    handle_msg(Msg, Parent, Name, State, Mod, Debug1)
    end.

新进程会阻塞在loop的receive,等到消息到来。我们假设已经收到消息,看下面的处理过程。对于{system, From, Req}系统消息,调用sys:handle_system_msg处理,用于在暂停、恢复或者终止进程等;{‘EXIT’, Parent, Reason}退出信号导致进程退出,并调用terminate函数;其他消息都会进入handle_msg函数。

进程退出

先来看进程退出,即gen_server:terminate函数。

-spec terminate(_, _, _, _, _, _) -> no_return().
terminate(Reason, Name, Msg, Mod, State, Debug) ->
    terminate(Reason, Reason, Name, Msg, Mod, State, Debug).

-spec terminate(_, _, _, _, _, _, _) -> no_return().
terminate(ExitReason, ReportReason, Name, Msg, Mod, State, Debug) ->
    Reply = try_terminate(Mod, ExitReason, State),
    case Reply of
	{'EXIT', ExitReason1, ReportReason1} ->
	    FmtState = format_status(terminate, Mod, get(), State),
	    error_info(ReportReason1, Name, Msg, FmtState, Debug),
	    exit(ExitReason1);
	_ ->
	    case ExitReason of
		normal ->
		    exit(normal);
		shutdown ->
		    exit(shutdown);
		{shutdown,_}=Shutdown ->
		    exit(Shutdown);
		_ ->
		    FmtState = format_status(terminate, Mod, get(), State),
		    error_info(ReportReason, Name, Msg, FmtState, Debug),
		    exit(ExitReason)
	    end
    end.
	
try_terminate(Mod, Reason, State) ->
    try
	{ok, Mod:terminate(Reason, State)}
    catch
	throw:R ->
	    {ok, R};
	error:R ->
	    Stacktrace = erlang:get_stacktrace(),
	    {'EXIT', {R, Stacktrace}, {R, Stacktrace}};
	exit:R ->
	    Stacktrace = erlang:get_stacktrace(),
	    {'EXIT', R, {R, Stacktrace}}
    end.

terminate函数会回调我们的worker:terminate,并调用format_status格式化输出信息。这里也可以看到可选回调format_status的作用。

format_status(Opt, Mod, PDict, State) ->
    DefStatus = case Opt of
		    terminate -> State;
		    _ -> [{data, [{"State", State}]}]
		end,
    case erlang:function_exported(Mod, format_status, 2) of
	true ->
	    case catch Mod:format_status(Opt, [PDict, State]) of
		{'EXIT', _} -> DefStatus;
		Else -> Else
	    end;
	_ ->
	    DefStatus
    end.

为了完整的讨论进程终止,我们顺便介绍gen_server:stop函数:

stop(Name) ->
    gen:stop(Name).

stop(Name, Reason, Timeout) ->
    gen:stop(Name, Reason, Timeout).

简单调用gen:stop函数:

stop(Process) ->
    stop(Process, normal, infinity).

stop(Process, Reason, Timeout)
  when Timeout =:= infinity; is_integer(Timeout), Timeout >= 0 ->
    Fun = fun(Pid) -> proc_lib:stop(Pid, Reason, Timeout) end,
    do_for_proc(Process, Fun).
	
%% Local or remote by pid
do_for_proc(Pid, Fun) when is_pid(Pid) ->
    Fun(Pid);
%% Local by name
do_for_proc(Name, Fun) when is_atom(Name) ->
    case whereis(Name) of
	Pid when is_pid(Pid) ->
	    Fun(Pid);
	undefined ->
	    exit(noproc)
    end;

do_for_proc会以pid为参数调用第二个参数fun函数。fun即 proc_lib:stop(Pid, Reason, Timeout)。

%% proc_lib.erl
-spec stop(Process, Reason, Timeout) -> 'ok' when
      Process :: pid() | RegName | {RegName,node()},
      RegName :: atom(),
      Reason :: term(),
      Timeout :: timeout().
stop(Process, Reason, Timeout) ->
    {Pid, Mref} = erlang:spawn_monitor(do_stop(Process, Reason)),
    receive
	{'DOWN', Mref, _, _, Reason} ->
	    ok;
	{'DOWN', Mref, _, _, {noproc,{sys,terminate,_}}} ->
	    exit(noproc);
	{'DOWN', Mref, _, _, CrashReason} ->
	    exit(CrashReason)
    after Timeout ->
	    exit(Pid, kill),
	    receive
		{'DOWN', Mref, _, _, _} ->
		    exit(timeout)
	    end
    end.

-spec do_stop(Process, Reason) -> Fun when
      Process :: pid() | RegName | {RegName,node()},
      RegName :: atom(),
      Reason :: term(),
      Fun :: fun(() -> no_return()).
do_stop(Process, Reason) ->
    fun() ->
	    Mref = erlang:monitor(process, Process),
	    ok = sys:terminate(Process, Reason, infinity),
	    receive
		{'DOWN', Mref, _, _, ExitReason} ->
		    exit(ExitReason)
	    end
    end.

这里会多次用到进程监控。proc_lib:stop生成并监控新进程 proc_lib:do_stop。do_stop也会监控要终止的进程,即worker,然后调用sys:terminate,它并不会直接终止进程,而是发消息:

terminate(Name, Reason, Timeout) ->
    send_system_msg(Name, {terminate, Reason}, Timeout).
	
send_system_msg(Name, Request, Timeout) ->
    case catch gen:call(Name, system, Request, Timeout) of
	{ok,Res} -> Res;
	{'EXIT', Reason} -> exit({Reason, mfa(Name, Request, Timeout)})
    end.

利用gen:call函数给Name - {local, worker}进程发消息,由之前介绍过的loop消息循环接收并处理,会收到一个系统消息,然后回调gen_server:system_terminate,然后回到gen_server:terminate。这里先略过详细流程,只要知道最后还是会由gen_server:terminate来处理。为什么要搞这么多监控呢?把复杂的任务放到新进程中,然后用一个简单的进程(简单到几乎不会崩溃)监控它,是Erlang最根本的哲学。

gen_server:call

worker.erl程序中,导出函数query和set调用了gen_server:call:

%% gen_server.erl
call(Name, Request) ->
    case catch gen:call(Name, '$gen_call', Request) of
	{ok,Res} ->
	    Res;
	{'EXIT',Reason} ->
	    exit({Reason, {?MODULE, call, [Name, Request]}})
    end.

call(Name, Request, Timeout) ->
    case catch gen:call(Name, '$gen_call', Request, Timeout) of
	{ok,Res} ->
	    Res;
	{'EXIT',Reason} ->
	    exit({Reason, {?MODULE, call, [Name, Request, Timeout]}})
    end.

会以参数[{local, worker}, ‘$gen_call’, Request]调用到gen:call——之前有过介绍,这里详细看一下代码:

call(Process, Label, Request) -> 
    call(Process, Label, Request, ?default_timeout).

call(Process, Label, Request, Timeout)
  when Timeout =:= infinity; is_integer(Timeout), Timeout >= 0 ->
    Fun = fun(Pid) -> do_call(Pid, Label, Request, Timeout) end,
    do_for_proc(Process, Fun).

do_call(Process, Label, Request, Timeout) ->
    try erlang:monitor(process, Process) of
	Mref ->
	    catch erlang:send(Process, {Label, {self(), Mref}, Request},
		  [noconnect]),
	    receive
		{Mref, Reply} ->
		    erlang:demonitor(Mref, [flush]),
		    {ok, Reply};
		{'DOWN', Mref, _, _, noconnection} ->
		    Node = get_node(Process),
		    exit({nodedown, Node});
		{'DOWN', Mref, _, _, Reason} ->
		    exit(Reason)
	    after Timeout ->
		    erlang:demonitor(Mref, [flush]),
		    exit(timeout)
	    end
    catch
	error:_ ->
	    Node = get_node(Process),
	    monitor_node(Node, true),
	    receive
		{nodedown, Node} -> 
		    monitor_node(Node, false),
		    exit({nodedown, Node})
	    after 0 -> 
		    Tag = make_ref(),
		    Process ! {Label, {self(), Tag}, Request},
		    wait_resp(Node, Tag, Timeout)
	    end
    end.

do_call执行真正的调用,也就是消息发送。首先monitor进程,然后调用erlang:send发送消息。如果进程在其他节点,需要监控节点的进程。erlang:send发送的消息由loop循环接收,gen_server:handle_msg最后详细介绍。

gen_server:cast

worker:show会调用gen_server:cast,发送异步消息:

cast({global,Name}, Request) ->
    catch global:send(Name, cast_msg(Request)),
    ok;
cast({via, Mod, Name}, Request) ->
    catch Mod:send(Name, cast_msg(Request)),
    ok;
cast({Name,Node}=Dest, Request) when is_atom(Name), is_atom(Node) -> 
    do_cast(Dest, Request);
cast(Dest, Request) when is_atom(Dest) ->
    do_cast(Dest, Request);
cast(Dest, Request) when is_pid(Dest) ->
    do_cast(Dest, Request).

do_cast(Dest, Request) -> 
    do_send(Dest, cast_msg(Request)),
    ok.
    
cast_msg(Request) -> {'$gen_cast',Request}.

do_send(Dest, Msg) ->
    case catch erlang:send(Dest, Msg, [noconnect]) of
	noconnect ->
	    spawn(erlang, send, [Dest,Msg]);
	Other ->
	    Other
    end.

异步消息只需要发送,不关心进程的响应。所以很简单,erlang:send发送’$gen_cast’标记的消息。

gen_server:handle_msg

消息收发的关键函数时handle_msg,所以现在介绍。

handle_msg({'$gen_call', From, Msg}, Parent, Name, State, Mod) ->
    Result = try_handle_call(Mod, Msg, From, State),
    case Result of
	{ok, {reply, Reply, NState}} ->
	    reply(From, Reply),
	    loop(Parent, Name, NState, Mod, infinity, []);
	{ok, {reply, Reply, NState, Time1}} ->
	    reply(From, Reply),
	    loop(Parent, Name, NState, Mod, Time1, []);
	{ok, {noreply, NState}} ->
	    loop(Parent, Name, NState, Mod, infinity, []);
	{ok, {noreply, NState, Time1}} ->
	    loop(Parent, Name, NState, Mod, Time1, []);
	{ok, {stop, Reason, Reply, NState}} ->
	    {'EXIT', R} = 
		(catch terminate(Reason, Name, Msg, Mod, NState, [])),
	    reply(From, Reply),
	    exit(R);
	Other -> handle_common_reply(Other, Parent, Name, Msg, Mod, State)
    end;
handle_msg(Msg, Parent, Name, State, Mod) ->
    Reply = try_dispatch(Msg, Mod, State),
    handle_common_reply(Reply, Parent, Name, Msg, Mod, State).
	
try_handle_call(Mod, Msg, From, State) ->
    try
	{ok, Mod:handle_call(Msg, From, State)}
    catch
	throw:R ->
	    {ok, R};
	error:R ->
	    Stacktrace = erlang:get_stacktrace(),
	    {'EXIT', {R, Stacktrace}, {R, Stacktrace}};
	exit:R ->
	    Stacktrace = erlang:get_stacktrace(),
	    {'EXIT', R, {R, Stacktrace}}
    end.
	
reply({To, Tag}, Reply) ->
    catch To ! {Tag, Reply}.

首先处理’$gen_call’消息,try_handle_call是try保护的回调调用,调用worker:handle_call,我们自己编写的回调函数,看到其实可以在handle_call中直接throw返回结果。调用reply发回响应,From的形式是{To, Tag},To表示pid或者注册名,Tag是monitor返回的引用值,这个消息会被之前介绍的gen:do_call接收,并返回{ok, Reply}到我们的应用程序。

返回其他值走向handle_common_reply。

其他消息走到第二个分支,调用try_dispatch。

try_dispatch({'$gen_cast', Msg}, Mod, State) ->
    try_dispatch(Mod, handle_cast, Msg, State);
try_dispatch(Info, Mod, State) ->
    try_dispatch(Mod, handle_info, Info, State).

try_dispatch(Mod, Func, Msg, State) ->
    try
	{ok, Mod:Func(Msg, State)}
    catch
	throw:R ->
	    {ok, R};
	error:R ->
	    Stacktrace = erlang:get_stacktrace(),
	    {'EXIT', {R, Stacktrace}, {R, Stacktrace}};
	exit:R ->
	    Stacktrace = erlang:get_stacktrace(),
	    {'EXIT', R, {R, Stacktrace}}
    end.

handle_cast和handle_info的使用在这里体现,它们返回的结果都会交由handle_common_reply处理:

handle_common_reply(Reply, Parent, Name, Msg, Mod, State) ->
    case Reply of
	{ok, {noreply, NState}} ->
	    loop(Parent, Name, NState, Mod, infinity, []);
	{ok, {noreply, NState, Time1}} ->
	    loop(Parent, Name, NState, Mod, Time1, []);
	{ok, {stop, Reason, NState}} ->
	    terminate(Reason, Name, Msg, Mod, NState, []);
	{'EXIT', ExitReason, ReportReason} ->
	    terminate(ExitReason, ReportReason, Name, Msg, Mod, State, []);
	{ok, BadReply} ->
	    terminate({bad_return_value, BadReply}, Name, Msg, Mod, State, [])
    end.

这里没什么特殊的,除了注意超时值,如果某次返回没有超时,就会更新为infinity;这也很好理解,尾递归情况,之前返回的参数不会影响后续调用。还有,handle_cast和handle_info的返回值只能是{ok, {noreply, NState…}}。随后接着进入loop循环。

最后

实际运行worker看一下效果:

Erlang/OTP 19 [erts-8.0] [64-bit] [smp:8:8] [async-threads:10]

Eshell V8.0  (abort with ^G)
1> worker:start_link().
Saving key: '0x01f' with value test, self: <0.58.0>
Timeout
{ok,<0.58.0>}
2> worker:show('0x01f').
show, key '0x01f' value test, self: <0.58.0>
ok
3> worker:query('0x01f').
query from {<0.56.0>,#Ref<0.0.1.59>}, self: <0.58.0>
{ok,test}
4> self().
<0.56.0>
5> 

唯一没有介绍的回调函数code_change由gen_server:system_code_change回调:

system_code_change([Name, State, Mod, Time], _Module, OldVsn, Extra) ->
    case catch Mod:code_change(OldVsn, State, Extra) of
	{ok, NewState} -> {ok, [Name, NewState, Mod, Time]};
	Else -> Else
    end.

总结一下。虽然篇幅很长,但几乎都是列源代码,只有少部分简单解释。因为Erlang实在是太简单了——这里简单的意思是简洁。很容易就能理解它的所有能力,而这种能力又非常稳定和强大。