diff options
Diffstat (limited to 'lib/kernel/src/inet_tcp_dist.erl')
| -rw-r--r-- | lib/kernel/src/inet_tcp_dist.erl | 157 |
1 files changed, 88 insertions, 69 deletions
diff --git a/lib/kernel/src/inet_tcp_dist.erl b/lib/kernel/src/inet_tcp_dist.erl index 1bdc1c9ed8..64b28bb49b 100644 --- a/lib/kernel/src/inet_tcp_dist.erl +++ b/lib/kernel/src/inet_tcp_dist.erl @@ -24,9 +24,13 @@ -export([listen/1, accept/1, accept_connection/5, setup/5, close/1, select/1, is_node_name/1]). +%% Generalized dist API +-export([gen_listen/2, gen_accept/2, gen_accept_connection/6, + gen_setup/6, gen_select/2]). + %% internal exports --export([accept_loop/2,do_accept/6,do_setup/6,getstat/1,tick/1]). +-export([accept_loop/3,do_accept/7,do_setup/7,getstat/1]). -import(error_logger,[error_msg/2]). @@ -34,15 +38,6 @@ --define(to_port(Socket, Data, Opts), - case inet_tcp:send(Socket, Data, Opts) of - {error, closed} -> - self() ! {tcp_closed, Socket}, - {error, closed}; - R -> - R - end). - -include("dist.hrl"). -include("dist_util.hrl"). @@ -53,8 +48,15 @@ %% ------------------------------------------------------------ select(Node) -> + gen_select(inet_tcp, Node). + +gen_select(Driver, Node) -> case split_node(atom_to_list(Node), $@, []) of - [_,_Host] -> true; + [_, Host] -> + case inet:getaddr(Host, Driver:family()) of + {ok,_} -> true; + _ -> false + end; _ -> false end. @@ -64,9 +66,12 @@ select(Node) -> %% ------------------------------------------------------------ listen(Name) -> - case do_listen([{active, false}, {packet,2}, {reuseaddr, true}]) of + gen_listen(inet_tcp, Name). + +gen_listen(Driver, Name) -> + case do_listen(Driver, [{active, false}, {packet,2}, {reuseaddr, true}]) of {ok, Socket} -> - TcpAddress = get_tcp_address(Socket), + TcpAddress = get_tcp_address(Driver, Socket), {_,Port} = TcpAddress#net_address.address, case erl_epmd:register_node(Name, Port) of {ok, Creation} -> @@ -78,7 +83,7 @@ listen(Name) -> Error end. -do_listen(Options) -> +do_listen(Driver, Options) -> {First,Last} = case application:get_env(kernel,inet_dist_listen_min) of {ok,N} when is_integer(N) -> case application:get_env(kernel, @@ -91,14 +96,14 @@ do_listen(Options) -> _ -> {0,0} end, - do_listen(First, Last, listen_options([{backlog,128}|Options])). + do_listen(Driver, First, Last, listen_options([{backlog,128}|Options])). -do_listen(First,Last,_) when First > Last -> +do_listen(_Driver, First,Last,_) when First > Last -> {error,eaddrinuse}; -do_listen(First,Last,Options) -> - case inet_tcp:listen(First, Options) of +do_listen(Driver, First,Last,Options) -> + case Driver:listen(First, Options) of {error, eaddrinuse} -> - do_listen(First+1,Last,Options); + do_listen(Driver, First+1,Last,Options); Other -> Other end. @@ -124,23 +129,26 @@ listen_options(Opts0) -> %% ------------------------------------------------------------ accept(Listen) -> - spawn_opt(?MODULE, accept_loop, [self(), Listen], [link, {priority, max}]). + gen_accept(inet_tcp, Listen). -accept_loop(Kernel, Listen) -> - case inet_tcp:accept(Listen) of +gen_accept(Driver, Listen) -> + spawn_opt(?MODULE, accept_loop, [Driver, self(), Listen], [link, {priority, max}]). + +accept_loop(Driver, Kernel, Listen) -> + case Driver:accept(Listen) of {ok, Socket} -> - Kernel ! {accept,self(),Socket,inet,tcp}, - _ = controller(Kernel, Socket), - accept_loop(Kernel, Listen); + Kernel ! {accept,self(),Socket,Driver:family(),tcp}, + _ = controller(Driver, Kernel, Socket), + accept_loop(Driver, Kernel, Listen); Error -> exit(Error) end. -controller(Kernel, Socket) -> +controller(Driver, Kernel, Socket) -> receive {Kernel, controller, Pid} -> flush_controller(Pid, Socket), - inet_tcp:controlling_process(Socket, Pid), + Driver:controlling_process(Socket, Pid), flush_controller(Pid, Socket), Pid ! {self(), controller}; {Kernel, unsupported_protocol} -> @@ -165,15 +173,18 @@ flush_controller(Pid, Socket) -> %% ------------------------------------------------------------ accept_connection(AcceptPid, Socket, MyNode, Allowed, SetupTime) -> + gen_accept_connection(inet_tcp, AcceptPid, Socket, MyNode, Allowed, SetupTime). + +gen_accept_connection(Driver, AcceptPid, Socket, MyNode, Allowed, SetupTime) -> spawn_opt(?MODULE, do_accept, - [self(), AcceptPid, Socket, MyNode, Allowed, SetupTime], + [Driver, self(), AcceptPid, Socket, MyNode, Allowed, SetupTime], [link, {priority, max}]). -do_accept(Kernel, AcceptPid, Socket, MyNode, Allowed, SetupTime) -> +do_accept(Driver, Kernel, AcceptPid, Socket, MyNode, Allowed, SetupTime) -> receive {AcceptPid, controller} -> Timer = dist_util:start_timer(SetupTime), - case check_ip(Socket) of + case check_ip(Driver, Socket) of true -> HSData = #hs_data{ kernel_pid = Kernel, @@ -182,9 +193,8 @@ do_accept(Kernel, AcceptPid, Socket, MyNode, Allowed, SetupTime) -> timer = Timer, this_flags = 0, allowed = Allowed, - f_send = fun(S,D) -> inet_tcp:send(S,D) end, - f_recv = fun(S,N,T) -> inet_tcp:recv(S,N,T) - end, + f_send = fun Driver:send/2, + f_recv = fun Driver:recv/3, f_setopts_pre_nodeup = fun(S) -> inet:setopts(S, @@ -203,8 +213,8 @@ do_accept(Kernel, AcceptPid, Socket, MyNode, Allowed, SetupTime) -> f_getll = fun(S) -> inet:getll(S) end, - f_address = fun get_remote_id/2, - mf_tick = fun ?MODULE:tick/1, + f_address = fun(S, Node) -> get_remote_id(Driver, S, Node) end, + mf_tick = fun(S) -> tick(Driver, S) end, mf_getstat = fun ?MODULE:getstat/1 }, dist_util:handshake_other_started(HSData); @@ -235,13 +245,13 @@ nodelay() -> %% ------------------------------------------------------------ %% Get remote information about a Socket. %% ------------------------------------------------------------ -get_remote_id(Socket, Node) -> +get_remote_id(Driver, Socket, Node) -> case inet:peername(Socket) of {ok,Address} -> case split_node(atom_to_list(Node), $@, []) of [_,Host] -> #net_address{address=Address,host=Host, - protocol=tcp,family=inet}; + protocol=tcp,family=Driver:family()}; _ -> %% No '@' or more than one '@' in node name. ?shutdown(no_node) @@ -256,14 +266,18 @@ get_remote_id(Socket, Node) -> %% ------------------------------------------------------------ setup(Node, Type, MyNode, LongOrShortNames,SetupTime) -> + gen_setup(inet_tcp, Node, Type, MyNode, LongOrShortNames, SetupTime). + +gen_setup(Driver, Node, Type, MyNode, LongOrShortNames, SetupTime) -> spawn_opt(?MODULE, do_setup, - [self(), Node, Type, MyNode, LongOrShortNames, SetupTime], + [Driver, self(), Node, Type, MyNode, LongOrShortNames, SetupTime], [link, {priority, max}]). -do_setup(Kernel, Node, Type, MyNode, LongOrShortNames,SetupTime) -> +do_setup(Driver, Kernel, Node, Type, MyNode, LongOrShortNames, SetupTime) -> ?trace("~p~n",[{inet_tcp_dist,self(),setup,Node}]), - [Name, Address] = splitnode(Node, LongOrShortNames), - case inet:getaddr(Address, inet) of + [Name, Address] = splitnode(Driver, Node, LongOrShortNames), + AddressFamily = Driver:family(), + case inet:getaddr(Address, AddressFamily) of {ok, Ip} -> Timer = dist_util:start_timer(SetupTime), case erl_epmd:port_please(Name, Ip) of @@ -272,7 +286,7 @@ do_setup(Kernel, Node, Type, MyNode, LongOrShortNames,SetupTime) -> [Node,Version]), dist_util:reset_timer(Timer), case - inet_tcp:connect( + Driver:connect( Ip, TcpPort, connect_options([{active, false}, {packet, 2}])) of @@ -285,8 +299,8 @@ do_setup(Kernel, Node, Type, MyNode, LongOrShortNames,SetupTime) -> timer = Timer, this_flags = 0, other_version = Version, - f_send = fun inet_tcp:send/2, - f_recv = fun inet_tcp:recv/3, + f_send = fun Driver:send/2, + f_recv = fun Driver:recv/3, f_setopts_pre_nodeup = fun(S) -> inet:setopts @@ -311,9 +325,9 @@ do_setup(Kernel, Node, Type, MyNode, LongOrShortNames,SetupTime) -> address = {Ip,TcpPort}, host = Address, protocol = tcp, - family = inet} + family = AddressFamily} end, - mf_tick = fun ?MODULE:tick/1, + mf_tick = fun(S) -> tick(Driver, S) end, mf_getstat = fun ?MODULE:getstat/1, request_type = Type }, @@ -353,18 +367,23 @@ close(Socket) -> %% If Node is illegal terminate the connection setup!! -splitnode(Node, LongOrShortNames) -> +splitnode(Driver, Node, LongOrShortNames) -> case split_node(atom_to_list(Node), $@, []) of [Name|Tail] when Tail =/= [] -> Host = lists:append(Tail), case split_node(Host, $., []) of [_] when LongOrShortNames =:= longnames -> - error_msg("** System running to use " - "fully qualified " - "hostnames **~n" - "** Hostname ~s is illegal **~n", - [Host]), - ?shutdown(Node); + case Driver:parse_address(Host) of + {ok, _} -> + [Name, Host]; + _ -> + error_msg("** System running to use " + "fully qualified " + "hostnames **~n" + "** Hostname ~s is illegal **~n", + [Host]), + ?shutdown(Node) + end; L when length(L) > 1, LongOrShortNames =:= shortnames -> error_msg("** System NOT running to use fully qualified " "hostnames **~n" @@ -390,26 +409,26 @@ split_node([], _, Ack) -> [lists:reverse(Ack)]. %% ------------------------------------------------------------ %% Fetch local information about a Socket. %% ------------------------------------------------------------ -get_tcp_address(Socket) -> +get_tcp_address(Driver, Socket) -> {ok, Address} = inet:sockname(Socket), {ok, Host} = inet:gethostname(), #net_address { address = Address, host = Host, protocol = tcp, - family = inet + family = Driver:family() }. %% ------------------------------------------------------------ %% Do only accept new connection attempts from nodes at our %% own LAN, if the check_ip environment parameter is true. %% ------------------------------------------------------------ -check_ip(Socket) -> +check_ip(Driver, Socket) -> case application:get_env(check_ip) of {ok, true} -> case get_ifs(Socket) of {ok, IFs, IP} -> - check_ip(IFs, IP); + check_ip(Driver, IFs, IP); _ -> ?shutdown(no_node) end; @@ -428,20 +447,14 @@ get_ifs(Socket) -> Error end. -check_ip([{OwnIP, _, Netmask}|IFs], PeerIP) -> - case {mask(Netmask, PeerIP), mask(Netmask, OwnIP)} of +check_ip(Driver, [{OwnIP, _, Netmask}|IFs], PeerIP) -> + case {Driver:mask(Netmask, PeerIP), Driver:mask(Netmask, OwnIP)} of {M, M} -> true; - _ -> check_ip(IFs, PeerIP) + _ -> check_ip(Driver, IFs, PeerIP) end; -check_ip([], PeerIP) -> +check_ip(_Driver, [], PeerIP) -> {false, PeerIP}. -mask({M1,M2,M3,M4}, {IP1,IP2,IP3,IP4}) -> - {M1 band IP1, - M2 band IP2, - M3 band IP3, - M4 band IP4}. - is_node_name(Node) when is_atom(Node) -> case split_node(atom_to_list(Node), $@, []) of [_, _Host] -> true; @@ -450,8 +463,14 @@ is_node_name(Node) when is_atom(Node) -> is_node_name(_Node) -> false. -tick(Sock) -> - ?to_port(Sock,[],[force]). +tick(Driver, Socket) -> + case Driver:send(Socket, [], [force]) of + {error, closed} -> + self() ! {tcp_closed, Socket}, + {error, closed}; + R -> + R + end. getstat(Socket) -> case inet:getstat(Socket, [recv_cnt, send_cnt, send_pend]) of |