diff options
Diffstat (limited to 'lib/kernel/src/net_kernel.erl')
| -rw-r--r-- | lib/kernel/src/net_kernel.erl | 261 |
1 files changed, 206 insertions, 55 deletions
diff --git a/lib/kernel/src/net_kernel.erl b/lib/kernel/src/net_kernel.erl index 35a54f591e..0c679e7349 100644 --- a/lib/kernel/src/net_kernel.erl +++ b/lib/kernel/src/net_kernel.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 1996-2013. All Rights Reserved. +%% Copyright Ericsson AB 1996-2016. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -53,18 +53,27 @@ -define(tckr_dbg(X), ok). -endif. -%% User Interface Exports --export([start/1, start_link/1, stop/0, - kernel_apply/3, +%% Documented API functions. + +-export([allow/1, + connect_node/1, monitor_nodes/1, monitor_nodes/2, + setopts/2, + getopts/2, + start/1, + stop/0]). + +%% Exports for internal use. + +-export([start_link/2, + kernel_apply/3, longnames/0, - allow/1, protocol_childspecs/0, epmd_module/0]). -export([connect/1, disconnect/1, hidden_connect/1, passive_cnct/1]). --export([connect_node/1, hidden_connect_node/1]). %% explicit connect +-export([hidden_connect_node/1]). %% explicit connect -export([set_net_ticktime/1, set_net_ticktime/2, get_net_ticktime/0]). -export([node_info/1, node_info/2, nodes_info/0, @@ -73,7 +82,8 @@ -export([publish_on_node/1, update_publish_nodes/1]). -%% Internal Exports +%% Internal exports for spawning processes. + -export([do_spawn/3, spawn_func/6, ticker/2, @@ -103,7 +113,7 @@ }). -record(listen, { - listen, %% listen pid + listen, %% listen socket accept, %% accepting pid address, %% #net_address module %% proto module @@ -341,18 +351,18 @@ request(Req) -> start(Args) -> erl_distribution:start(Args). -%% This is the main startup routine for net_kernel -%% The defaults are longnames and a ticktime of 15 secs to the tcp_drv. - -start_link([Name]) -> - start_link([Name, longnames]); +%% This is the main startup routine for net_kernel (only for internal +%% use by the Kernel application. -start_link([Name, LongOrShortNames]) -> - start_link([Name, LongOrShortNames, 15000]); +start_link([Name], CleanHalt) -> + start_link([Name, longnames], CleanHalt); +start_link([Name, LongOrShortNames], CleanHalt) -> + start_link([Name, LongOrShortNames, 15000], CleanHalt); -start_link([Name, LongOrShortNames, Ticktime]) -> - case gen_server:start_link({local, net_kernel}, net_kernel, - {Name, LongOrShortNames, Ticktime}, []) of +start_link([Name, LongOrShortNames, Ticktime], CleanHalt) -> + Args = {Name, LongOrShortNames, Ticktime, CleanHalt}, + case gen_server:start_link({local, net_kernel}, ?MODULE, + Args, []) of {ok, Pid} -> {ok, Pid}; {error, {already_started, Pid}} -> @@ -361,12 +371,9 @@ start_link([Name, LongOrShortNames, Ticktime]) -> exit(nodistribution) end. -%% auth:get_cookie should only be able to return an atom -%% tuple cookies are unknowns - -init({Name, LongOrShortNames, TickT}) -> +init({Name, LongOrShortNames, TickT, CleanHalt}) -> process_flag(trap_exit,true), - case init_node(Name, LongOrShortNames) of + case init_node(Name, LongOrShortNames, CleanHalt) of {ok, Node, Listeners} -> process_flag(priority, max), Ticktime = to_integer(TickT), @@ -379,7 +386,7 @@ init({Name, LongOrShortNames, TickT}) -> connections = ets:new(sys_dist,[named_table, protected, - {keypos, 2}]), + {keypos, #connection.node}]), listen = Listeners, allowed = [], verbose = 0 @@ -549,6 +556,38 @@ handle_call({new_ticktime,_T,_TP}, #state{tick = #tick_change{time = T}} = State) -> async_reply({reply, {ongoing_change_to, T}, State}, From); +handle_call({setopts, new, Opts}, From, State) -> + Ret = setopts_new(Opts, State), + async_reply({reply, Ret, State}, From); + +handle_call({setopts, Node, Opts}, From, State) -> + Return = + case ets:lookup(sys_dist, Node) of + [Conn] when Conn#connection.state =:= up -> + case call_owner(Conn#connection.owner, {setopts, Opts}) of + {ok, Ret} -> Ret; + _ -> {error, noconnection} + end; + + _ -> + {error, noconnection} + end, + async_reply({reply, Return, State}, From); + +handle_call({getopts, Node, Opts}, From, State) -> + Return = + case ets:lookup(sys_dist, Node) of + [Conn] when Conn#connection.state =:= up -> + case call_owner(Conn#connection.owner, {getopts, Opts}) of + {ok, Ret} -> Ret; + _ -> {error, noconnection} + end; + + _ -> + {error, noconnection} + end, + async_reply({reply, Return, State}, From); + handle_call(_Msg, _From, State) -> {noreply, State}. @@ -1201,12 +1240,12 @@ get_proto_mod(_Family, _Protocol, []) -> %% -------- Initialisation functions ------------------------ -init_node(Name, LongOrShortNames) -> - {NameWithoutHost,_Host} = lists:splitwith(fun($@)->false;(_)->true end, - atom_to_list(Name)), +init_node(Name, LongOrShortNames, CleanHalt) -> + {NameWithoutHost0,_Host} = split_node(Name), case create_name(Name, LongOrShortNames, 1) of {ok,Node} -> - case start_protos(list_to_atom(NameWithoutHost),Node) of + NameWithoutHost = list_to_atom(NameWithoutHost0), + case start_protos(NameWithoutHost, Node, CleanHalt) of {ok, Ls} -> {ok, Node, Ls}; Error -> @@ -1240,8 +1279,7 @@ create_name(Name, LongOrShortNames, Try) -> end. create_hostpart(Name, LongOrShortNames) -> - {Head,Host} = lists:splitwith(fun($@)->false;(_)->true end, - atom_to_list(Name)), + {Head,Host} = split_node(Name), Host1 = case {Host,LongOrShortNames} of {[$@,_|_],longnames} -> {ok,Host}; @@ -1268,6 +1306,9 @@ create_hostpart(Name, LongOrShortNames) -> end, {Head,Host1}. +split_node(Name) -> + lists:splitwith(fun(C) -> C =/= $@ end, atom_to_list(Name)). + %% %% %% @@ -1307,21 +1348,26 @@ epmd_module() -> %% Start all protocols %% -start_protos(Name,Node) -> +start_protos(Name, Node, CleanHalt) -> case init:get_argument(proto_dist) of {ok, [Protos]} -> - start_protos(Name,Protos, Node); + start_protos(Name, Protos, Node, CleanHalt); _ -> - start_protos(Name,["inet_tcp"], Node) + start_protos(Name, ["inet_tcp"], Node, CleanHalt) end. -start_protos(Name,Ps, Node) -> - case start_protos(Name, Ps, Node, []) of - [] -> {error, badarg}; - Ls -> {ok, Ls} +start_protos(Name, Ps, Node, CleanHalt) -> + case start_protos(Name, Ps, Node, [], CleanHalt) of + [] -> + case CleanHalt of + true -> halt(1); + false -> {error, badarg} + end; + Ls -> + {ok, Ls} end. -start_protos(Name, [Proto | Ps], Node, Ls) -> +start_protos(Name, [Proto | Ps], Node, Ls, CleanHalt) -> Mod = list_to_atom(Proto ++ "_dist"), case catch Mod:listen(Name) of {ok, {Socket, Address, Creation}} -> @@ -1334,33 +1380,48 @@ start_protos(Name, [Proto | Ps], Node, Ls) -> address = Address, accept = AcceptPid, module = Mod }, - start_protos(Name,Ps, Node, [L|Ls]); + start_protos(Name,Ps, Node, [L|Ls], CleanHalt); _ -> Mod:close(Socket), - error_logger:info_msg("Invalid node name: ~p~n", [Node]), - start_protos(Name, Ps, Node, Ls) + S = "invalid node name: " ++ atom_to_list(Node), + proto_error(CleanHalt, Proto, S), + start_protos(Name, Ps, Node, Ls, CleanHalt) end; {'EXIT', {undef,_}} -> - error_logger:info_msg("Protocol: ~tp: not supported~n", [Proto]), - start_protos(Name,Ps, Node, Ls); + proto_error(CleanHalt, Proto, "not supported"), + start_protos(Name, Ps, Node, Ls, CleanHalt); {'EXIT', Reason} -> - error_logger:info_msg("Protocol: ~tp: register error: ~tp~n", - [Proto, Reason]), - start_protos(Name,Ps, Node, Ls); + register_error(CleanHalt, Proto, Reason), + start_protos(Name, Ps, Node, Ls, CleanHalt); {error, duplicate_name} -> - error_logger:info_msg("Protocol: ~tp: the name " ++ - atom_to_list(Node) ++ - " seems to be in use by another Erlang node", - [Proto]), - start_protos(Name,Ps, Node, Ls); + S = "the name " ++ atom_to_list(Node) ++ + " seems to be in use by another Erlang node", + proto_error(CleanHalt, Proto, S), + start_protos(Name, Ps, Node, Ls, CleanHalt); {error, Reason} -> - error_logger:info_msg("Protocol: ~tp: register/listen error: ~tp~n", - [Proto, Reason]), - start_protos(Name,Ps, Node, Ls) + register_error(CleanHalt, Proto, Reason), + start_protos(Name, Ps, Node, Ls, CleanHalt) end; -start_protos(_,[], _Node, Ls) -> +start_protos(_, [], _Node, Ls, _CleanHalt) -> Ls. +register_error(false, Proto, Reason) -> + S = io_lib:format("register/listen error: ~p", [Reason]), + proto_error(false, Proto, lists:flatten(S)); +register_error(true, Proto, Reason) -> + S = "Protocol '" ++ Proto ++ "': register/listen error: ", + erlang:display_string(S), + erlang:display(Reason). + +proto_error(CleanHalt, Proto, String) -> + S = "Protocol '" ++ Proto ++ "': " ++ String ++ "\n", + case CleanHalt of + false -> + error_logger:info_msg(S); + true -> + erlang:display_string(S) + end. + set_node(Node, Creation) when node() =:= nonode@nohost -> case catch erlang:setnode(Node, Creation) of true -> @@ -1581,3 +1642,93 @@ async_gen_server_reply(From, Msg) -> {'EXIT', _} -> ok end. + +call_owner(Owner, Msg) -> + Mref = monitor(process, Owner), + Owner ! {self(), Mref, Msg}, + receive + {Mref, Reply} -> + erlang:demonitor(Mref, [flush]), + {ok, Reply}; + {'DOWN', Mref, _, _, _} -> + error + end. + + +-spec setopts(Node, Options) -> ok | {error, Reason} | ignored when + Node :: node() | new, + Options :: [inet:socket_setopt()], + Reason :: inet:posix() | noconnection. + +setopts(Node, Opts) when is_atom(Node), is_list(Opts) -> + request({setopts, Node, Opts}). + +setopts_new(Opts, State) -> + %% First try setopts on listening socket(s) + %% Bail out on failure. + %% If successful, we are pretty sure Opts are ok + %% and we continue with config params and pending connections. + case setopts_on_listen(Opts, State#state.listen) of + ok -> + setopts_new_1(Opts); + Fail -> Fail + end. + +setopts_on_listen(_, []) -> ok; +setopts_on_listen(Opts, [#listen {listen = LSocket, module = Mod} | T]) -> + try Mod:setopts(LSocket, Opts) of + ok -> + setopts_on_listen(Opts, T); + Fail -> Fail + catch + error:undef -> {error, enotsup} + end. + +setopts_new_1(Opts) -> + ConnectOpts = case application:get_env(kernel, inet_dist_connect_options) of + {ok, CO} -> CO; + _ -> [] + end, + application:set_env(kernel, inet_dist_connect_options, + merge_opts(Opts,ConnectOpts)), + ListenOpts = case application:get_env(kernel, inet_dist_listen_options) of + {ok, LO} -> LO; + _ -> [] + end, + application:set_env(kernel, inet_dist_listen_options, + merge_opts(Opts, ListenOpts)), + case lists:keyfind(nodelay, 1, Opts) of + {nodelay, ND} when is_boolean(ND) -> + application:set_env(kernel, dist_nodelay, ND); + _ -> ignore + end, + + %% Update any pending connections + PendingConns = ets:select(sys_dist, [{'_', + [{'=/=',{element,#connection.state,'$_'},up}], + ['$_']}]), + lists:foreach(fun(#connection{state = pending, owner = Owner}) -> + call_owner(Owner, {setopts, Opts}); + (#connection{state = up_pending, pending_owner = Owner}) -> + call_owner(Owner, {setopts, Opts}); + (_) -> ignore + end, PendingConns), + ok. + +merge_opts([], B) -> + B; +merge_opts([H|T], B0) -> + {Key, _} = H, + B1 = lists:filter(fun({K,_}) -> K =/= Key end, B0), + merge_opts(T, [H | B1]). + +-spec getopts(Node, Options) -> + {'ok', OptionValues} | {'error', Reason} | ignored when + Node :: node(), + Options :: [inet:socket_getopt()], + OptionValues :: [inet:socket_setopt()], + Reason :: inet:posix() | noconnection. + +getopts(Node, Opts) when is_atom(Node), is_list(Opts) -> + request({getopts, Node, Opts}). + |