File 2386-ssh-Unified-way-of-starting-listening-sockets.patch of Package erlang
From da7902412f1e77b8241c0bacbeac2d6013e8f345 Mon Sep 17 00:00:00 2001
From: Hans Nilsson <hans@erlang.org>
Date: Thu, 2 Mar 2017 16:37:54 +0100
Subject: [PATCH 06/13] ssh: Unified way of starting listening sockets
---
lib/ssh/src/ssh.erl | 215 ++++++++++++++++++-------------------------
lib/ssh/src/ssh_acceptor.erl | 106 ++++++++++-----------
2 files changed, 136 insertions(+), 185 deletions(-)
diff --git a/lib/ssh/src/ssh.erl b/lib/ssh/src/ssh.erl
index c1be9f732..c13955679 100644
--- a/lib/ssh/src/ssh.erl
+++ b/lib/ssh/src/ssh.erl
@@ -132,13 +132,12 @@ connect(Host, Port, UserOptions, Timeout) when is_integer(Port),
SocketOpts = [{active,false} | ?GET_OPT(socket_options,Options)],
try Transport:connect(Host, Port, SocketOpts, ConnectionTimeout) of
{ok, Socket} ->
- Opts = ?PUT_INTERNAL_OPT([{user_pid,self()}, {host,Host}], Options),
+ Opts = ?PUT_INTERNAL_OPT([{user_pid,self()}, {host,fmt_host(Host)}], Options),
ssh_connection_handler:start_connection(client, Socket, Opts, Timeout);
{error, Reason} ->
{error, Reason}
catch
exit:{function_clause, _F} ->
- io:format('function_clause ~p~n',[_F]),
{error, {options, {transport, TransportOpts}}};
exit:badarg ->
{error, {options, {socket_options, SocketOpts}}}
@@ -311,16 +310,15 @@ handle_daemon_args(Host, UserOptions0) ->
%%%----------------------------------------------------------------
valid_socket_to_use(Socket, {tcp,_,_}) ->
%% Is this tcp-socket a valid socket?
- case {is_tcp_socket(Socket),
- {ok,[{active,false}]} == inet:getopts(Socket, [active])
- }
+ try {is_tcp_socket(Socket),
+ {ok,[{active,false}]} == inet:getopts(Socket, [active])
+ }
of
- {true, true} ->
- ok;
- {true, false} ->
- {error, not_passive_mode};
- _ ->
- {error, not_tcp_socket}
+ {true, true} -> ok;
+ {true, false} -> {error, not_passive_mode};
+ _ -> {error, not_tcp_socket}
+ catch
+ _:_ -> {error, bad_socket}
end;
valid_socket_to_use(_, {L4,_,_}) ->
@@ -340,13 +338,7 @@ start_daemon(_, _, {error,Error}) ->
start_daemon(socket, Socket, Options) ->
case valid_socket_to_use(Socket, ?GET_OPT(transport,Options)) of
ok ->
- try
- do_start_daemon(Socket, Options)
- catch
- throw:bad_fd -> {error,bad_fd};
- throw:bad_socket -> {error,bad_socket};
- _C:_E -> {error,{cannot_start_daemon,_C,_E}}
- end;
+ start_daemon(inet:sockname(Socket), Socket, Options);
{error,SockError} ->
{error,SockError}
end;
@@ -355,136 +347,107 @@ start_daemon(Host, Port, Options) ->
try
do_start_daemon(Host, Port, Options)
catch
- throw:bad_fd -> {error,bad_fd};
- throw:bad_socket -> {error,bad_socket};
- _C:_E -> {error,{cannot_start_daemon,_C,_E}}
+ throw:bad_fd ->
+ {error,bad_fd};
+ throw:bad_socket ->
+ {error,bad_socket};
+ error:{badmatch,{error,Error}} ->
+ {error,Error};
+ _C:_E ->
+ {error,{cannot_start_daemon,_C,_E}}
end.
+%%%----------------------------------------------------------------
+do_start_daemon({error,Error}, _, _) ->
+ {error,Error};
-do_start_daemon(Socket, Options) ->
- {ok, {IP,Port}} =
- try {ok,_} = inet:sockname(Socket)
- catch
- _:_ -> throw(bad_socket)
- end,
- Host = fmt_host(IP),
- Opts = ?PUT_INTERNAL_OPT([{connected_socket, Socket},
- {address, Host},
- {port, Port},
- {role, server}], Options),
-
- Profile = ?GET_OPT(profile, Options),
- case ssh_system_sup:system_supervisor(Host, Port, Profile) of
- undefined ->
- try sshd_sup:start_child(Opts) of
- {error, {already_started, _}} ->
- {error, eaddrinuse};
- Result = {ok,_} ->
- call_ssh_acceptor_handle_connection(Host, Port, Opts, Socket, Result);
- Result = {error, _} ->
- Result
- catch
- exit:{noproc, _} ->
- {error, ssh_not_started}
- end;
- Sup ->
- AccPid = ssh_system_sup:acceptor_supervisor(Sup),
- case ssh_acceptor_sup:start_child(AccPid, Opts) of
- {error, {already_started, _}} ->
- {error, eaddrinuse};
- {ok, _} ->
- call_ssh_acceptor_handle_connection(Host, Port, Opts, Socket, {ok,Sup});
- Other ->
- Other
- end
- end.
+do_start_daemon({ok, {IP,Port}}, Socket, Options0) ->
+ finalize_start(fmt_host(IP),
+ Port,
+ ?PUT_INTERNAL_OPT({connected_socket, Socket}, Options0),
+ fun(Opts, DefaultResult) ->
+ try ssh_acceptor:handle_established_connection(
+ ?GET_INTERNAL_OPT(address, Opts),
+ ?GET_INTERNAL_OPT(port, Opts),
+ Opts,
+ Socket)
+ of
+ {error,Error} ->
+ {error,Error};
+ _ ->
+ DefaultResult
+ catch
+ C:R ->
+ {error,{could_not_start_connection,{C,R}}}
+ end
+ end);
do_start_daemon(Host0, Port0, Options0) ->
- {Host,Port1} =
- try
- case ?GET_SOCKET_OPT(fd, Options0) of
- undefined ->
- {Host0,Port0};
- Fd when Port0==0 ->
- find_hostport(Fd)
- end
- catch
- _:_ -> throw(bad_fd)
- end,
- {Port, WaitRequestControl, Options1} =
- case Port1 of
- 0 -> %% Allocate the socket here to get the port number...
- {ok,LSock} = ssh_acceptor:callback_listen(0, Options0),
- {ok,{_,LPort}} = inet:sockname(LSock),
- {LPort,
- LSock,
- ?PUT_INTERNAL_OPT({lsocket,{LSock,self()}}, Options0)
- };
- _ ->
- {Port1, false, Options0}
- end,
+ {{Host,Port}, ListenSocket} =
+ open_listen_socket(Host0, Port0, Options0),
+
+ %% Now Host,Port is what to use for the supervisor to register its name,
+ %% and ListenSocket is for listening on connections. But it is still owned
+ %% by self()...
+
+ finalize_start(Host, Port,
+ ?PUT_INTERNAL_OPT({lsocket,{ListenSocket,self()}}, Options0),
+ fun(Opts, Result) ->
+ {_, Callback, _} = ?GET_OPT(transport, Opts),
+ receive
+ {request_control, ListenSocket, ReqPid} ->
+ ok = Callback:controlling_process(ListenSocket, ReqPid),
+ ReqPid ! {its_yours,ListenSocket},
+ Result
+ end
+ end).
+
+
+open_listen_socket(Host0, Port0, Options0) ->
+ case ?GET_SOCKET_OPT(fd, Options0) of
+ undefined ->
+ {ok,LSock} = ssh_acceptor:listen(Port0, Options0),
+ {ok,{_,LPort}} = inet:sockname(LSock),
+ {{fmt_host(Host0),LPort}, LSock};
+
+ Fd when is_integer(Fd) ->
+ %% Do gen_tcp:listen with the option {fd,Fd}:
+ {ok,LSock} = ssh_acceptor:listen(0, Options0),
+ {ok,{LHost,LPort}} = inet:sockname(LSock),
+ {{fmt_host(LHost),LPort}, LSock}
+ end.
+
+%%%----------------------------------------------------------------
+finalize_start(Host, Port, Options0, F) ->
Options = ?PUT_INTERNAL_OPT([{address, Host},
{port, Port},
- {role, server}], Options1),
- Profile = ?GET_OPT(profile, Options0),
+ {role, server}], Options0),
+ Profile = ?GET_OPT(profile, Options),
case ssh_system_sup:system_supervisor(Host, Port, Profile) of
undefined ->
try sshd_sup:start_child(Options) of
{error, {already_started, _}} ->
{error, eaddrinuse};
+ {error, Error} ->
+ {error, Error};
Result = {ok,_} ->
- sync_request_control(WaitRequestControl, Options),
- Result;
- Result = {error, _} ->
- Result
+ F(Options, Result)
catch
exit:{noproc, _} ->
{error, ssh_not_started}
end;
- Sup ->
+ Sup ->
AccPid = ssh_system_sup:acceptor_supervisor(Sup),
case ssh_acceptor_sup:start_child(AccPid, Options) of
{error, {already_started, _}} ->
{error, eaddrinuse};
+ {error, Error} ->
+ {error, Error};
{ok, _} ->
- sync_request_control(WaitRequestControl, Options),
- {ok, Sup};
- Other ->
- Other
+ F(Options, {ok,Sup})
end
end.
-call_ssh_acceptor_handle_connection(Host, Port, Options, Socket, DefaultResult) ->
- {_, Callback, _} = ?GET_OPT(transport, Options),
- try ssh_acceptor:handle_connection(Callback, Host, Port, Options, Socket)
- of
- {error,Error} -> {error,Error};
- _ -> DefaultResult
- catch
- C:R -> {error,{could_not_start_connection,{C,R}}}
- end.
-
-
-sync_request_control(false, _Options) ->
- ok;
-sync_request_control(LSock, Options) ->
- {_, Callback, _} = ?GET_OPT(transport, Options),
- receive
- {request_control,LSock,ReqPid} ->
- ok = Callback:controlling_process(LSock, ReqPid),
- ReqPid ! {its_yours,LSock},
- ok
- end.
-
-find_hostport(Fd) ->
- %% Using internal functions inet:open/8 and inet:close/0.
- %% Don't try this at home unless you know what you are doing!
- {ok,S} = inet:open(Fd, {0,0,0,0}, 0, [], tcp, inet, stream, inet_tcp),
- {ok, HostPort} = inet:sockname(S),
- ok = inet:close(S),
- HostPort.
-
-fmt_host({A,B,C,D}) ->
- lists:concat([A,".",B,".",C,".",D]);
-fmt_host(T={_,_,_,_,_,_,_,_}) ->
- lists:flatten(string:join([io_lib:format("~.16B",[A]) || A <- tuple_to_list(T)], ":")).
+%%%----------------------------------------------------------------
+fmt_host(IP) when is_tuple(IP) -> inet:ntoa(IP);
+fmt_host(Str) when is_list(Str) -> Str.
diff --git a/lib/ssh/src/ssh_acceptor.erl b/lib/ssh/src/ssh_acceptor.erl
index 4943f062b..f9e228021 100644
--- a/lib/ssh/src/ssh_acceptor.erl
+++ b/lib/ssh/src/ssh_acceptor.erl
@@ -27,8 +27,8 @@
%% Internal application API
-export([start_link/4,
number_of_connections/1,
- callback_listen/2,
- handle_connection/5]).
+ listen/2,
+ handle_established_connection/4]).
%% spawn export
-export([acceptor_init/5, acceptor_loop/6]).
@@ -42,40 +42,57 @@ start_link(Port, Address, Options, AcceptTimeout) ->
Args = [self(), Port, Address, Options, AcceptTimeout],
proc_lib:start_link(?MODULE, acceptor_init, Args).
+%%%----------------------------------------------------------------
+number_of_connections(SystemSup) ->
+ length([X ||
+ {R,X,supervisor,[ssh_subsystem_sup]} <- supervisor:which_children(SystemSup),
+ is_pid(X),
+ is_reference(R)
+ ]).
+
+%%%----------------------------------------------------------------
+listen(Port, Options) ->
+ {_, Callback, _} = ?GET_OPT(transport, Options),
+ SockOpts = [{active, false}, {reuseaddr,true} | ?GET_OPT(socket_options, Options)],
+ case Callback:listen(Port, SockOpts) of
+ {error, nxdomain} ->
+ Callback:listen(Port, lists:delete(inet6, SockOpts));
+ {error, enetunreach} ->
+ Callback:listen(Port, lists:delete(inet6, SockOpts));
+ {error, eafnosupport} ->
+ Callback:listen(Port, lists:delete(inet6, SockOpts));
+ Other ->
+ Other
+ end.
+
+%%%----------------------------------------------------------------
+handle_established_connection(Address, Port, Options, Socket) ->
+ {_, Callback, _} = ?GET_OPT(transport, Options),
+ handle_connection(Callback, Address, Port, Options, Socket).
+
%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------
acceptor_init(Parent, Port, Address, Opts, AcceptTimeout) ->
- {_, Callback, _} = ?GET_OPT(transport, Opts),
try
- {LSock0,SockOwner0} = ?GET_INTERNAL_OPT(lsocket, Opts),
- true = is_pid(SockOwner0),
- {ok,{_,Port}} = inet:sockname(LSock0),
- {LSock0, SockOwner0}
+ ?GET_INTERNAL_OPT(lsocket, Opts)
of
{LSock, SockOwner} ->
- %% Use existing socket
- proc_lib:init_ack(Parent, {ok, self()}),
- request_ownership(LSock, SockOwner),
- acceptor_loop(Callback, Port, Address, Opts, LSock, AcceptTimeout)
- catch
- error:{badkey,lsocket} ->
- %% Open new socket
- try
- socket_listen(Port, Opts)
- of
- {ok, ListenSocket} ->
+ case inet:sockname(LSock) of
+ {ok,{_,Port}} -> % A usable, open LSock
proc_lib:init_ack(Parent, {ok, self()}),
- acceptor_loop(Callback,
- Port, Address, Opts, ListenSocket, AcceptTimeout);
- {error,Error} ->
- proc_lib:init_ack(Parent, Error),
- {error,Error}
- catch
- _:_ ->
- {error,listen_socket_failed}
- end;
+ request_ownership(LSock, SockOwner),
+ {_, Callback, _} = ?GET_OPT(transport, Opts),
+ acceptor_loop(Callback, Port, Address, Opts, LSock, AcceptTimeout);
+ {error,_} -> % Not open, a restart
+ {ok,NewLSock} = listen(Port, Opts),
+ proc_lib:init_ack(Parent, {ok, self()}),
+ Opts1 = ?DELETE_INTERNAL_OPT(lsocket, Opts),
+ {_, Callback, _} = ?GET_OPT(transport, Opts1),
+ acceptor_loop(Callback, Port, Address, Opts1, NewLSock, AcceptTimeout)
+ end
+ catch
_:_ ->
{error,use_existing_socket_failed}
end.
@@ -87,30 +104,7 @@ request_ownership(LSock, SockOwner) ->
{its_yours,LSock} -> ok
end.
-
-socket_listen(Port0, Opts) ->
- Port = case ?GET_SOCKET_OPT(fd, Opts) of
- undefined -> Port0;
- _ -> 0
- end,
- callback_listen(Port, Opts).
-
-
-callback_listen(Port, Opts0) ->
- {_, Callback, _} = ?GET_OPT(transport, Opts0),
- Opts = ?PUT_SOCKET_OPT([{active, false}, {reuseaddr,true}], Opts0),
- SockOpts = ?GET_OPT(socket_options, Opts),
- case Callback:listen(Port, SockOpts) of
- {error, nxdomain} ->
- Callback:listen(Port, lists:delete(inet6, SockOpts));
- {error, enetunreach} ->
- Callback:listen(Port, lists:delete(inet6, SockOpts));
- {error, eafnosupport} ->
- Callback:listen(Port, lists:delete(inet6, SockOpts));
- Other ->
- Other
- end.
-
+%%%----------------------------------------------------------------
acceptor_loop(Callback, Port, Address, Opts, ListenSocket, AcceptTimeout) ->
case (catch Callback:accept(ListenSocket, AcceptTimeout)) of
{ok, Socket} ->
@@ -127,6 +121,7 @@ acceptor_loop(Callback, Port, Address, Opts, ListenSocket, AcceptTimeout) ->
ListenSocket, AcceptTimeout)
end.
+%%%----------------------------------------------------------------
handle_connection(Callback, Address, Port, Options, Socket) ->
Profile = ?GET_OPT(profile, Options),
SystemSup = ssh_system_sup:system_supervisor(Address, Port, Profile),
@@ -158,7 +153,7 @@ handle_connection(Callback, Address, Port, Options, Socket) ->
{error,max_sessions}
end.
-
+%%%----------------------------------------------------------------
handle_error(timeout) ->
ok;
@@ -185,10 +180,3 @@ handle_error(Reason) ->
error_logger:error_report(String),
exit({accept_failed, String}).
-
-number_of_connections(SystemSup) ->
- length([X ||
- {R,X,supervisor,[ssh_subsystem_sup]} <- supervisor:which_children(SystemSup),
- is_pid(X),
- is_reference(R)
- ]).
--
2.12.2
