8000 otp/lib/stdlib/src/peer.erl at master · googol-lab/otp · GitHub
[go: up one dir, main page]
Skip to content

peer.erl

Latest commit

 

History

History
1050 lines (939 loc) · 43 KB

peer.erl

File metadata and controls

1050 lines (939 loc) · 43 KB
Raw
%%
%% Copyright WhatsApp Inc. and its affiliates. 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.
%% 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.
%%
%% @doc
%% Controller for additional Erlang node running on the same host,
%% or in 8000 a different container/host (e.g. Docker).
%%
%% == Terms ==
%% Origin node - Erlang VM instance that spawns additional nodes.
%% Peer node - a node spawned by the origin.
%% Control process - a process running on origin node, if it terminates,
%% peer node terminates too.
%% Control connection - a connection between origin and peer, can be
%% ether Erlang Distribution connection, or alternative one.
%%
%% I/O is forwarded from peer node to origin via control connection.
%%
%% When standard_io is used as alternative connection, peer node
%% uses standard out to multiplex console output and control sequences.
%% Characters in range of 192-255 are reserved for control sequences,
%% see encode_port_data for details. If peer node attempts to print
%% characters in this range, an controlling process on the origin
%% node may terminate with an error (because CRC check will fail).
%%
%% Alternative connection via TCP does not have that limitation, but
%% it also does not redirect console I/O from the peer node.
%% @end
-module(peer).
-author("maximfca@gmail.com").
%% API
-export([
start_link/0,
start_link/1,
start/1,
stop/1,
random_name/0,
random_name/1,
get_state/1,
call/4,
call/5,
cast/4,
send/3
]).
%% Could be gen_statem too, with peer_state, but most interactions
%% are anyway available in all states.
-behaviour(gen_server).
%% gen_server callbacks
-export([
init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
terminate/2
]).
%% Internal exports for stdin/stdout, non-distribution RPC, and tests
-export([
start/0 %% this function must be named "start", requirement for user.erl
]).
%% Origin node will listen to the specified port (port 0 is auto-select),
%% or specified IP/Port, and expect peer node to connect to this port.
-type connection() ::
Port :: 0..65535 |
{inet:ip_address(), 0..65535} |
standard_io.
%% Specification for boot waiting
-type wait_boot() ::
timeout() | %% wait for node to boot (default, 15 sec),
{pid(), Tag :: term()} | %% do not wait, send {Tag, {started, node(), pid()}} to Pid when node boots
false. %% don't wait, don't notify
-type exec() ::
file:name() | %% path to "erl" (default is init:get_argument(progname))
{file:name(), [string()]}. %% SSH support: {"/usr/bin/ssh", ["account@host_b", "/usr/bin/erl"]}
%% Peer node start options
-type start_options() ::
#{
name => atom() | string(), %% node name (part before @), if not defined, peer
%% starts in non-distributed mode (requires alternative connection)
longnames => boolean(), %% long/short names (default is net_kernel:longnames(), and shortnames)
host => string(), %% force hostname (when not specified, actual peer node hostname is used)
peer_down => stop | continue | crash, %% stop (default): when peer terminates, peer control process
%% stops normally regardless of the reason.
%% continue: when peer terminates, peer control process stays up
%% saving exit reason in the state
%% crash: when peer terminates, origin process
%% terminates with underlying reason
exec => exec(), %% path to executable, or SSH/Docker support
connection => connection(), %% alternative connection specification
args => [string()], %% additional command line parameters to append
env => [{string(), string()}], %% additional environment variables
wait_boot => wait_boot(), %% default is synchronous start with 15 sec timeout
shutdown => close | %% close supervision channel
halt | %% The default... %% stop node using erlang:halt() wait default timeout for nodedown
{halt, disconnect_timeout()} | %% stop node using erlang:halt() wait timeout() for nodedown
disconnect_timeout() %% send init:stop() request and wait up to specified timeout for nodedown
}.
%% Peer node states
-type peer_state() :: booting | running | {down, Reason :: term()}.
-export_type([
start_options/0,
peer_state/0,
exec/0,
disconnect_timeout/0
]).
%% Maximum integer timeout value in a receive...
-define (MAX_INT_TIMEOUT, 4294967295).
%% Default time we wait for distributed connection to be removed,
%% when shutdown type is halt, before we disconnect from the node...
-define (DEFAULT_HALT_DISCONNECT_TIMEOUT, 5000).
%% Minimum time we wait for distributed connection to be removed,
%% before we disconnect from the node (except in the shutdown
%% close case)...
-define (MIN_DISCONNECT_TIMEOUT, 1000).
%% Socket connect timeout, for TCP connection.
-define (CONNECT_TIMEOUT, 10000).
%% Socket accept timeout, for TCP connection.
-define (ACCEPT_TIMEOUT, 10000).
%% How long to wait for graceful shutdown.
-define (SHUTDOWN_TIMEOUT, 10000).
%% Synchronous RPC timeout for alternative connection.
-define (SYNC_RPC_TIMEOUT, 5000).
%% Default timeout for peer node to boot.
-define (WAIT_BOOT_TIMEOUT, 15000).
-type disconnect_timeout() :: ?MIN_DISCONNECT_TIMEOUT..?MAX_INT_TIMEOUT | infinity.
%% @doc Creates random node name, using "peer" as prefix.
-spec random_name() -> string().
random_name() ->
random_name(?MODULE_STRING).
%% @doc Creates sufficiently random node name,
%% using OS process ID for origin VM, resulting name
%% looks like prefix-3-7161
-spec random_name(Prefix :: string() | atom()) -> string().
random_name(Prefix) ->
OsPid = os:getpid(),
Uniq = erlang:unique_integer([positive]),
lists:concat([Prefix, "-", Uniq, "-", OsPid]).
%% @doc Starts a distributed node with random name, on this host,
%% and waits for that node to boot. Returns full node name,
%% registers local process with the same name as peer node.
-spec start_link() -> {ok, node()} | {error, Reason :: term()}.
start_link() ->
start_link(#{name => random_name()}).
%% @doc Starts peer node, linked to the calling process.
%% Accepts additional command line arguments and
%% other important options.
-spec start_link(start_options()) -> {ok, pid()} | {ok, pid(), node()} | {error, Reason}
when Reason :: term().
start_link(Options) ->
start_it(Options, start_link).
%% @doc Starts peer node, not linked to the calling process.
-spec start(start_options()) -> {ok, pid()} | {ok, pid(), node()} | {error, Reason}
when Reason :: term().
start(Options) ->
start_it(Options, start).
%% @doc Stops controlling process, shutting down peer node synchronously
-spec stop(gen_statem:server_ref()) -> ok.
stop(Dest) ->
gen_server:stop(Dest).
%% @doc returns peer node state.
-spec get_state(Dest :: gen_statem:server_ref()) -> peer_state().
get_state(Dest) ->
gen_server:call(Dest, get_state).
%% @doc Calls M:F(A) remotely, via alternative connection, with default 5 seconds timeout
-spec call(Dest :: gen_statem:server_ref(), Module :: module(), Function :: atom(),
Args :: [term()]) -> Result :: term().
call(Dest, M, F, A) ->
call(Dest, M, F, A, ?SYNC_RPC_TIMEOUT).
%% @doc Call M:F(A) remotely, timeout is explicitly specified
-spec call(Dest :: gen_statem:server_ref(), Module :: module(), Function :: atom(),
Args :: [term()], Timeout :: timeout()) -> Result :: term().
call(Dest, M, F, A, Timeout) ->
case gen_server:call(Dest, {call, M, F, A}, Timeout) of
{ok, Reply} ->
Reply;
{Class, {Reason, Stack}} ->
erlang:raise(Class, Reason, Stack);
{error, Reason} ->
erlang:error(Reason)
end.
%% @doc Cast M:F(A) remotely, don't care about the result
-spec cast(Dest :: gen_statem:server_ref(), Module :: module(), Function :: atom(), Args :: [term()]) -> ok.
cast(Dest, M, F, A) ->
gen_server:cast(Dest, {cast, M, F, A}).
%% @doc Sends a message to pid or named process on the peer node
%% using alternative connection. No delivery guarantee.
-spec send(Dest :: gen_statem:server_ref(), To :: pid() | atom(), Message :: term()) -> ok.
send(Dest, To, Message) ->
gen_server:cast(Dest, {send, To, Message}).
%%--------------------------------------------------------------------
%%% gen_server callbacks
-record(peer_state, {
options :: start_options(),
%% full node name, can be 'undefined'
node :: atom(),
%% debugging information: executable and arguments used to
%% start the peer
exec :: file:name(),
args :: [string()],
%% alternative connection socket/port
connection :: undefined | port() | gen_tcp:socket(),
%% listening socket, while waiting for network alternative connection
listen_socket :: undefined | gen_tcp:socket(),
%% accumulator for RPC over standard_io
stdio = <<>> :: binary(),
%% peer state
peer_state = booting :: peer_state(),
%% pid/ref saved for gen:reply() when node is booted, or false
notify = false :: false | {pid(), reference()},
%% counter (reference) for calls.
%% it is not possible to use erlang reference, or pid,
%% because it changes when node becomes distributed dynamically.
seq = 0 :: non_neg_integer(),
%% outstanding calls
outstanding = #{} :: #{non_neg_integer() => {reference(), pid()}}
}).
-type state() :: #peer_state{}.
-spec init([Name :: atom(), ... ]) -> {ok, state()}.
init([Notify, Options]) ->
process_flag(trap_exit, true), %% need this to ensure terminate/2 is called
{ListenSocket, Listen} = maybe_listen(Options),
{Exec, Args} = command_line(Listen, Options),
Env = maps:get(env, Options, []),
%% close port if running detached
Conn =
case maps:find(connection, Options) of
{ok, standard_io} ->
%% Cannot detach a peer that uses stdio. Request exit_status.
open_port({spawn_executable, Exec}, [{args, Args}, {env, Env}, hide, binary, exit_status]);
_ ->
Port = open_port({spawn_executable, Exec}, [{args, Args}, {env, Env}, hide, binary]),
%% peer can close the port before we get here which will cause
%% port_close to throw. Catch this and ignore.
catch erlang:port_close(Port),
receive {'EXIT', Port, _} -> undefined end
end,
%% Remove the default 'halt' shutdown option if present; the default is
%% defined in terminate()...
SaveOptions = case maps:find(shutdown, Options) of
{ok, halt} ->
maps:remove(shutdown, Options);
_ ->
Options
end,
State = #peer_state{options = SaveOptions, notify = Notify, args = Args, exec = Exec},
%% accept TCP connection if requested
if ListenSocket =:= undefined ->
{ok, State#peer_state{connection = Conn}};
true ->
_ = prim_inet:async_accept(ListenSocket, ?ACCEPT_TIMEOUT),
{ok, State#peer_state{listen_socket = ListenSocket}}
end.
%% not connected: no alternative connection available
handle_call({call, _M, _F, _A}, _From, #peer_state{connection = undefined} = State) ->
{reply, {error, noconnection}, State};
handle_call({call, M, F, A}, From,
#peer_state{connection = Port, options = #{connection := standard_io},
outstanding = Out, seq = Seq} = State) ->
origin_to_peer(port, Port, {call, Seq, M, F, A}),
{noreply, State#peer_state{outstanding = Out#{Seq => From}, seq = Seq + 1}};
handle_call({call, M, F, A}, From,
#peer_state{connection = Socket, outstanding = Out, seq = Seq} = State) ->
origin_to_peer(tcp, Socket, {call, Seq, M, F, A}),
{noreply, State#peer_state{outstanding = Out#{Seq => From}, seq = Seq + 1}};
handle_call(get_node, _From, #peer_state{node = Node} = State) ->
{reply, Node, State};
handle_call(get_state, _From, #peer_state{peer_state = PeerState} = State) ->
{reply, PeerState, State};
handle_call(group_leader, _From, State) ->
{reply, group_leader(), State}.
handle_cast({cast, _M, _F, _A}, #peer_state{connection = undefined} = State) ->
{noreply, State};
handle_cast({cast, M, F, A},
#peer_state{connection = Port, options = #{connection := standard_io}} = State) ->
origin_to_peer(port, Port, {cast, M, F, A}),
{noreply, State};
handle_cast({cast, M, F, A}, #peer_state{connection = Socket} = State) ->
origin_to_peer(tcp, Socket, {cast, M, F, A}),
{noreply, State};
handle_cast({send, _Dest, _Message}, #peer_state{connection = undefined} = State) ->
{noreply, State};
handle_cast({send, Dest, Message},
#peer_state{connection = Port, options = #{connection := standard_io}} = State) ->
origin_to_peer(port, Port, {message, Dest, Message}),
{noreply, State};
handle_cast({send, Dest, Message}, #peer_state{connection = Socket} = State) ->
origin_to_peer(tcp, Socket, {message, Dest, Message}),
{noreply, State}.
%%--------------------------------------------------------------------
%% alternative connections handling
%% alternative communications - request or response from peer
handle_info({tcp, Socket, SocketData}, #peer_state{connection = Socket} = State) ->
ok = inet:setopts(Socket, [{active, once}]),
{noreply, handle_alternative_data(tcp, binary_to_term(SocketData), State)};
%% standard_io
handle_info({Port, {data, PortData}}, #peer_state{connection = Port, stdio = PrevBin} = State) ->
{Str, NewBin} = decode_port_data(PortData, <<>>, PrevBin),
Str =/= <<>> andalso io:put_chars(Str),
{noreply, handle_port_binary(NewBin, State)};
%% booting: accepted TCP connection from the peer, but it is not yet
%% complete handshake
handle_info({inet_async, LSock, _Ref, {ok, CliSocket}},
#peer_state{listen_socket = LSock} = State) ->
true = inet_db:register_socket(CliSocket, inet_tcp),
ok = inet:setopts(CliSocket, [{active, once}]),
catch gen_tcp:close(LSock),
{noreply, State#peer_state{connection = CliSocket, listen_socket = undefined}};
handle_info({inet_async, LSock, _Ref, {error, Reason}},
#peer_state{listen_socket = LSock} = State) ->
%% failed to accept a TCP connection
catch gen_tcp:close(LSock),
%% stop unconditionally, it is essentially a part of gen_server:init callback
{stop, {inet_async, Reason}, State#peer_state{connection = undefined, listen_socket = undefined}};
%% booting: peer notifies via Erlang distribution
handle_info({started, Node}, State)->
true = erlang:monitor_node(Node, true),
{noreply, boot_complete(Node, started, State)};
%% nodedown: no-alternative dist-connected peer node is down
handle_info({nodedown, Node}, #peer_state{connection = undefined} = State) ->
maybe_stop({nodedown, Node}, State);
%% external program exited, returned Status code
handle_info({Port, {exit_status, Status}}, #peer_state{connection = Port} = State) ->
catch erlang:port_close(Port),
maybe_stop({exit_status, Status}, State);
%% port terminated: cannot proceed, stop the server
handle_info({'EXIT', Port, Reason}, #peer_state{connection = Port} = State) ->
catch erlang: 10000 port_close(Port),
maybe_stop(Reason, State);
handle_info({tcp_closed, Sock}, #peer_state{connection = Sock} = State) ->
%% TCP connection closed, no i/o port - assume node is stopped
catch gen_tcp:close(Sock),
maybe_stop(normal, State#peer_state{connection = undefined}).
%%--------------------------------------------------------------------
%% cleanup/termination
-spec terminate(Reason :: term(), state()) -> ok.
terminate(_Reason, #peer_state{connection = Port, options = Options, node = Node}) ->
case {maps:get(shutdown, Options, {halt, ?DEFAULT_HALT_DISCONNECT_TIMEOUT}),
maps:find(connection, Options)} of
{close, {ok, standard_io}} ->
Port /= undefined andalso (catch erlang:port_close(Port));
{close, {ok, _TCP}} ->
Port /= undefined andalso (catch gen_tcp:close(Port));
{close, error} ->
_ = erlang:disconnect_node(Node);
{{halt, Timeout}, {ok, standard_io}} ->
Port /= undefined andalso (catch erlang:port_close(Port)),
wait_disconnected(Node, {timeout, Timeout});
{{halt, Timeout}, {ok, _TCP}} ->
Port /= undefined andalso (catch gen_tcp:close(Port)),
wait_disconnected(Node, {timeout, Timeout});
{{halt, Timeout}, error} ->
try
_ = erpc:call(Node, erlang, halt, [], Timeout),
ok
catch
error:{erpc,noconnection} -> ok;
_:_ -> force_disconnect_node(Node)
end;
{Shutdown, error} ->
Timeout = shutdown(dist, undefined, Node, Shutdown),
wait_disconnected(Node, {timeout, Timeout});
{Shutdown, {ok, standard_io}} ->
Timeout = shutdown(port, Port, Node, Shutdown),
Deadline = deadline(Timeout),
receive {'EXIT', Port, _Reason2} -> ok after Timeout -> ok end,
catch erlang:port_close(Port),
wait_disconnected(Node, Deadline);
{Shutdown, {ok, _TCP}} ->
Timeout = shutdown(tcp, Port, Node, Shutdown),
Deadline = deadline(Timeout),
receive {tcp_closed, Port} -> ok after Timeout -> ok end,
catch catch gen_tcp:close(Port),
wait_disconnected(Node, Deadline)
end,
ok.
%%--------------------------------------------------------------------
%% Internal implementation
deadline(infinity) ->
{timeout, infinity};
deadline(Timeout) when is_integer(Timeout) ->
{deadline, erlang:monotonic_time(millisecond) + Timeout}.
wait_disconnected(Node, WaitUntil) ->
%% Should only be called just before we are exiting the caller, so
%% we do not bother disabling nodes monitoring if we enable it and
%% do not flush any nodeup/nodedown messages that we got due to the
%% nodes monitoring...
case lists:member(Node, nodes(connected)) of
false ->
ok;
true ->
_ = net_kernel:monitor_nodes(true, [{node_type, all}]),
%% Need to check connected nodes list again, since it
%% might have disconnected before we enabled nodes
%% monitoring...
case lists:member(Node, nodes(connected)) of
false ->
ok;
true ->
Tmo = case WaitUntil of
{timeout, T} ->
T;
{deadline, T} ->
TL = T - erlang:monotonic_time(millisecond),
if TL < 0 -> 0;
true -> TL
end
end,
receive {nodedown, Node, _} -> ok
after Tmo -> force_disconnect_node(Node)
end
end
end.
force_disconnect_node(Node) ->
_ = erlang:disconnect_node(Node),
logger:warning("peer:stop() timed out waiting for disconnect from "
"node ~p. The connection was forcefully taken down.",
[Node]).
%% This hack is a temporary workaround for test coverage reports
shutdown(_Type, _Port, Node, Timeout) when is_integer(Timeout); Timeout =:= infinity ->
erpc:cast(Node, init, stop, []),
Timeout;
shutdown(dist, undefined, Node, {Timeout, CoverNode}) when is_integer(Timeout); Timeout =:= infinity ->
rpc:call(CoverNode, cover, flush, [Node]),
erpc:cast(Node, init, stop, []),
Timeout;
shutdown(Type, Port, Node, {Timeout, CoverNode}) when is_integer(Timeout); Timeout =:= infinity ->
rpc:call(CoverNode, cover, flush, [Node]),
Port /= undefined andalso origin_to_peer(Type, Port, {cast, init, stop, []}),
Timeout.
%% Verify options correctness (Dialyzer also does the job, but slightly less convenient)
verify_args(Options) ->
%% verify that "Args" is valid - common problem is when Args aren't strings
Args = maps:get(args, Options, []),
is_list(Args) orelse error({invalid_arg, Args}),
[error({invalid_arg, Arg}) || Arg <- Args, not io_lib:char_list(Arg)],
%% alternative connection must be requested for non-distributed node,
%% or a distributed node when origin is not alive
is_map_key(connection, Options) orelse
(is_map_key(name, Options) andalso erlang:is_alive()) orelse error(not_alive),
%% exec must be a string, or a tuple of string(), [string()]
case maps:find(exec, Options) of
{ok, {Exec, Strs}} ->
io_lib:char_list(Exec) orelse error({exec, Exec}),
[error({exec, Str}) || Str <- Strs, not io_lib:char_list(Str)],
ok;
{ok, Exec} when is_list(Exec) ->
io_lib:char_list(Exec) orelse error({exec, Exec}),
ok;
error ->
ok;
{ok, Err} ->
error({exec, Err})
end,
case maps:find(shutdown, Options) of
{ok, close} ->
ok;
{ok, halt} ->
ok;
{ok, {halt, Tmo}} when (is_integer(Tmo)
andalso ?MIN_DISCONNECT_TIMEOUT =< Tmo
andalso Tmo =< ?MAX_INT_TIMEOUT)
orelse Tmo == infinity ->
ok;
{ok, Tmo} when (is_integer(Tmo)
andalso ?MIN_DISCONNECT_TIMEOUT =< Tmo
andalso Tmo =< ?MAX_INT_TIMEOUT)
orelse Tmo == infinity ->
ok;
{ok, {Tmo, Node}} when ((is_integer(Tmo)
andalso ?MIN_DISCONNECT_TIMEOUT =< Tmo
andalso Tmo =< ?MAX_INT_TIMEOUT)
orelse Tmo == infinity)
andalso is_atom(Node) ->
ok;
error ->
ok;
{ok, Err2} ->
error({shutdown, Err2})
end.
make_notify_ref(infinity) ->
{self(), make_ref()};
make_notify_ref(WaitBoot) when is_integer(WaitBoot) ->
{self(), make_ref()};
make_notify_ref({ReplyTo, Tag}) when is_pid(ReplyTo) ->
{ReplyTo, Tag};
make_notify_ref(false) ->
false.
start_it(Options, StartFun) ->
verify_args(Options),
WaitBoot = maps:get(wait_boot, Options, ?WAIT_BOOT_TIMEOUT),
Notify = make_notify_ref(WaitBoot),
case gen_server:StartFun(?MODULE, [Notify, Options], []) of
{ok, Pid} when WaitBoot =:= infinity; is_integer(WaitBoot) ->
{_, Ref} = Notify,
Mref = erlang:monitor(process, Pid),
receive
{Ref, {started, NodeName, Pid}} ->
erlang:demonitor(Mref, [flush]),
{ok, Pid, NodeName};
{Ref, {boot_failed, Reason, Pid}} ->
erlang:demonitor(Mref, [flush]),
erlang:exit({boot_failed, Reason});
{'DOWN', Mref, _, _, Reason} ->
erlang:exit(Reason)
after
WaitBoot ->
_ = gen_server:stop(Pid),
erlang:demonitor(Mref, [flush]),
erlang:exit(timeout)
end;
{ok, Pid} when is_map_key(host, Options) ->
{ok, Pid, node_name(Options)};
{ok, Pid} ->
{ok, Pid};
Error ->
Error
end.
node_name(#{name := Name, host := Host}) ->
list_to_atom(lists:concat([Name, "@", Host]));
node_name(_Options) ->
undefined.
%% Lost control connection to the peer while the node was
%% booting, this generally means a crash
maybe_stop(Reason, #peer_state{peer_state = booting} = State) ->
_ = boot_complete(Reason, boot_failed, State),
maybe_stop(Reason, State#peer_state{peer_state = {down, Reason}});
%%
maybe_stop(Reason, #peer_state{options = #{peer_down := crash}} = State) ->
{stop, Reason, State#peer_state{peer_state = {down, Reason}, connection = undefined}};
%% if state was already down, keep the original reason
maybe_stop(_Reason, #peer_state{options = #{peer_down := continue}, peer_state = {down, _}} = State) ->
{noreply, State};
%% continue working setting peer state to down
maybe_stop(Reason, #peer_state{options = #{peer_down := continue}} = State) ->
{noreply, State#peer_state{peer_state = {down, Reason}}};
%% default: ignore Reason and shut down normally
maybe_stop(Reason, State) ->
{stop, normal, State#peer_state{peer_state = {down, Reason}}}.
%% i/o protocol from origin:
%% * {io_reply, ...}
%% * {message, To, Content}
%% * {call, From, M, F, A}
%%
%% i/o port protocol, from peer:
%% * {io_request, From, ReplyAs, Request}
%% * {message, To, Content}
%% * {reply, From, ok | throw | error | exit | crash, Result | {Reason, Stack}}
%% Handles bytes coming from alternative connection, forwarding as needed.
handle_alternative_data(Kind, {io_request, From, FromRef, IoReq}, #peer_state{connection = Conn} = State) ->
%% TODO: make i/o completely async
Reply = {io_reply, From, FromRef, forward_request(IoReq)},
origin_to_peer(Kind, Conn, Reply),
State;
handle_alternative_data(_Kind, {message, To, Content}, State) ->
To ! Content,
State;
handle_alternative_data(_Kind, {reply, Seq, Class, Result}, #peer_state{outstanding = Out} = State) ->
{From, NewOut} = maps:take(Seq, Out),
gen:reply(From, {Class, Result}),
State#peer_state{outstanding = NewOut};
handle_alternative_data(_Kind, {started, NodeName}, State)->
boot_complete(NodeName, started, State).
forward_request(Req) ->
GL = group_leader(),
MRef = erlang:monitor(process, GL),
GL ! {io_request,self(), MRef, Req},
receive
{io_reply, MRef, Reply} ->
erlang:demonitor(MRef, [flush]),
Reply;
{'DOWN', MRef, _, _, _} ->
{error, terminated}
end.
%% generic primitive to send data from origin to peer via alternative connection
origin_to_peer(tcp, Sock, Term) ->
ok = gen_tcp:send(Sock, term_to_binary(Term));
origin_to_peer(port, Port, Term) ->
true = erlang:port_command(Port, encode_port_data(term_to_binary(Term))).
%% generic primitive to send data from peer to origin
peer_to_origin(tcp, Sock, Term) ->
ok = gen_tcp:send(Sock, term_to_binary(Term));
peer_to_origin(port, Port, Term) ->
%% converts Erlang term to terminal codes
%% Every binary byte is converted into two 4-bit sequences.
Bytes = term_to_binary(Term),
true = erlang:port_command(Port, encode_port_data(Bytes)).
%% convert/escape Erlang term into terminal codes.
%% Protocol consists of Erlang terms serialised into
%% External Term Format (ETF) via term_to_binary. Then
%% every byte of the resulting binary is split into two
%% nibbles (4-bit sequences), which are encoded into two
%% characters.
%% Control characters must have first two and last two bits
%% set, so the byte looks this way: 11xxxx11.
%% Example encoding, hexadecimal 16#0F will be encoded as
%% 11000011 11111111 (decimal 195 255).
encode_port_data(Bytes) ->
Size = byte_size(Bytes),
Crc = erlang:crc32(Bytes),
Total = <<Size:32, Bytes/binary, Crc:32>>,
<<<<3:2, Upper:4, 3:2, 3:2, Lower:4, 3:2>> || <<Upper:4, Lower:4>> <= Total>>.
%% convert terminal codes to Erlang term, printing everything that
%% was detected as text
decode_port_data(<<>>, Str, Bin) ->
{Str, Bin};
decode_port_data(<<3:2, Quad:4, 3:2, Rest/binary>>, Str, Bin) ->
decode_port_data(Rest, Str, <<Bin/bitstring, Quad:4>>);
decode_port_data(<<Char:8, Rest/binary>>, Str, Bin) ->
decode_port_data(Rest, <<Str/binary, Char>>, Bin).
%% recursively process buffers, potentially changing the state
handle_port_binary(<<Size:32, Payload:Size/binary, Crc:32, Rest/binary>>, State) ->
Crc = erlang:crc32(Payload),
Term = binary_to_term(Payload),
NewState = handle_alternative_data(port, Term, State),
handle_port_binary(Rest, NewState);
handle_port_binary(NewBin, State) ->
State#peer_state{stdio = NewBin}.
boot_complete(Node, _Result, #peer_state{notify = false} = State) ->
State#peer_state{peer_state = running, node = Node};
boot_complete(Node, Result, #peer_state{notify = {ReplyTo, Tag}} = State) ->
ReplyTo ! {Tag, {Result, Node, self()}},
State#peer_state{peer_state = running, node = Node}.
%% check if TCP connection is enabled, and starts listener
maybe_listen(#{connection := Port}) when is_integer(Port) ->
{ok, LSock} = gen_tcp:listen(Port, [binary, {reuseaddr, true}, {packet, 4}]),
{ok, WaitPort} = inet:port(LSock),
%% try guessing a local IP address
{ok, Ifs} = inet:getifaddrs(),
%% next incantation gets all IP addresses of all local interfaces that are up
LocalUp = lists:append(
[proplists:get_all_values(addr, Opts)
|| {_, Opts} <- Ifs, lists:member(up, proplists:get_value(flags, Opts, []))]),
%% filter invalid addresses
Local = prefer_localhost([Valid || Valid <- LocalUp, is_list(inet:ntoa(Valid))], [], []),
{LSock, {Local, WaitPort}};
maybe_listen(#{connection := {Ip, Port}}) when is_integer(Port) ->
{ok, LSock} = gen_tcp:listen(Port, [binary, {reuseaddr, true}, {packet, 4}, {ip, Ip}]),
WaitPort = if Port =:= 0 -> {ok, Dyn} = inet:port(LSock), Dyn; true -> Port end,
{LSock, {[Ip], WaitPort}};
maybe_listen(_Options) ->
{undefined, undefined}.
%% prefer localhost, IPv6 localhost, then everything else
prefer_localhost([], Preferred, Other) ->
Preferred ++ Other;
prefer_localhost([{127, _, _, _} = Local | Tail], Preferred, Other) ->
prefer_localhost(Tail, [Local | Preferred], Other);
prefer_localhost([{0, 0, 0, 0, 0, 0, 0, 1} = Local | Tail], Preferred, Other) ->
prefer_localhost(Tail, [Local | Preferred], Other);
prefer_localhost([Local | Tail], Preferred, Other) ->
prefer_localhost(Tail, Preferred, [Local | Other]).
name_arg(error, error, _) ->
[]; %% no name, no host - starting node that is not distributed
name_arg(error, {ok, Host}, LongOrShort) ->
%% interesting fallback: host is set, name is not. Do what predecessor did,
%% take the current node name and use it...
[Name, _] = string:lexemes(atom_to_list(node()), "@"),
name_arg(Name ++ "@" ++ Host, error, LongOrShort);
name_arg({ok, Name}, Host, LongOrShort) ->
name_arg(Name, Host, LongOrShort); %% unpack node name
name_arg(Name, Host, LongOrShort) when is_atom(Name) ->
name_arg(atom_to_list(Name), Host, LongOrShort); %% convert atom to list for command line
name_arg(Name, Host, {ok, ignored}) ->
name_arg(Name, Host, {ok, false}); %% fallback to shortnames when origin is not distributed
name_arg(Name, Host, error) ->
name_arg(Name, Host, {ok, net_kernel:longnames()}); %% no longnames present it start options
name_arg(Name, {ok, Host}, LongOrShort) ->
name_arg(Name ++ "@" ++ Host, error, LongOrShort); %% merge host part
%% only these are actual name creation clauses
name_arg(Name, error, {ok, true}) ->
["-name", Name];
name_arg(Name, error, {ok, false}) ->
["-sname", Name].
command_line(Listen, Options) ->
%% Node name/sname
NameArg = name_arg(maps:find(name, Options), maps:find(host, Options), maps:find(longnames, Options)),
%% additional command line args
CmdOpts = maps:get(args, Options, []),
%% start command
StartCmd =
case Listen of
undefined when map_get(connection, Options) =:= standard_io ->
["-user", atom_to_list(?MODULE)];
undefined ->
Self = base64:encode_to_string(term_to_binary(self())),
["-detached", "-noinput", "-user", atom_to_list(?MODULE), "-origin", Self];
{Ips, Port} ->
IpStr = lists:concat(lists:join(",", [inet:ntoa(Ip) || Ip <- Ips])),
["-detached", "-noinput", "-user", atom_to_list(?MODULE), "-origin", IpStr, integer_to_list(Port)]
end,
%% build command line
{Exec, PreArgs} = exec(Options),
{Exec, PreArgs ++ NameArg ++ StartCmd ++ CmdOpts}.
exec(#{exec := Prog}) when is_list(Prog) ->
{Prog, []};
exec(#{exec := {Prog, Args}}) when is_list(Prog), is_list(Args) ->
{Prog, Args};
exec(Options) when not is_map_key(exec, Options) ->
case init:get_argument(progname) of
{ok, [[Prog]]} ->
case os:find_executable(Prog) of
Exec when is_list(Exec) ->
{Exec, []};
false ->
maybe_otp_test_suite(Prog)
end;
_ ->
default_erts()
end.
maybe_otp_test_suite(Prog) ->
case string:split(Prog, "cerl ") of
[CerlPath, Args] ->
%% This is a hack to handle the 'cerl' script used
%% by the Erlang/OTP test suites. When 'cerl'
%% starts the runtime system, it typically sets
%% 'progname' to the path of the 'cerl' script,
%% followed by an argument. For example:
%%
%% /<full_path_to>/cerl -debug
%% /<full_path_to>/cerl -asan
%% /<full_path_to>/cerl -gcov
%%
%% We should find a better way to handle this, for
%% example by passing the emulator type and flavor
%% using the '-emu_type' and '-emu_flavor'
%% options. However, this is not without
%% complications as those options do not really
%% work for an installed system. Also, it is
%% probably a good idea to stop using 'slave'
%% before attempting to do this.
{filename:join(CerlPath, "cerl"), parse_args(Args)};
_ ->
default_erts()
end.
%% Split command line string into a list of arguments.
-spec parse_args(string()) -> [string()].
parse_args([]) ->
[];
parse_args([Deep | _] = AlreadyParsed) when is_list(Deep) ->
AlreadyParsed;
parse_args(CmdLine) ->
%% following regex splits command line, preserving quoted arguments, into argv[] list
Re = <<"((?:\"[^\"\\\\]*(?:\\\\[\\S\\s][^\"\\\\]*)*\"|'[^'\\\\]*(?:\\\\[\\S\\s][^'\\\\]*)*'|\\/[^\\/\\\\]*(?:\\\\[\\S\\s][^\\/\\\\]*)*\\/[gimy]*(?=\\s|$)|(?:\\\\\\s|\\S))+)(?=\\s|$)">>,
{match, Args} = re:run(CmdLine, Re, [{capture, all_but_first, list}, global]),
%% unquote arguments. It is possible to change regex capture groups to avoid extra processing.
[unquote(Arg) || [Arg] <- Args].
unquote([Q | Arg]) when Q =:= $\"; Q =:= $\' ->
case lists:last(Arg) of
Q -> lists:droplast(Arg);
_ -> [Q | Arg]
end;
unquote(Arg) ->
Arg.
%% if progname is not known, use `erlexec` from the same ERTS version we're currently running
%% BINDIR environment variable is already set
%% EMU variable it also set
default_erts() ->
Root = code:root_dir(),
Erts = filename:join(Root, lists:concat(["erts-", erlang:system_info(version)])),
BinDir = filename:join(Erts, "bin"),
{filename:join(BinDir, "erlexec"), []}.
%%--------------------------------------------------------------------
%% peer node implementation
notify_when_started(Kind, Port) ->
init:notify_when_started(self()) =:= started andalso
notify_started(Kind, Port),
ok.
notify_started(dist, Process) ->
Process ! {started, node()},
ok;
notify_started(Kind, Port) ->
peer_to_origin(Kind, Port, {started, node()}).
%% I/O redirection: peer side
-spec start() -> pid().
start() ->
case init:get_argument(origin) of
{ok, [[IpStr, PortString]]} ->
%% enter this clause when -origin IpList Port is specified in the command line.
Port = list_to_integer(PortString),
Ips = [begin {ok, Addr} = inet:parse_address(Ip), Addr end || Ip <- string:lexemes(IpStr, ",")],
spawn(fun () -> tcp_init(Ips, Port) end);
{ok, [[Base64EncProc]]} ->
%% No alternative connection, but have "-origin Base64EncProc"
OriginProcess = binary_to_term(base64:decode(Base64EncProc)),
%% setup 'user' process, I/O redirection: ask controlling process
%% who is the group leader.
GroupLeader = gen_server:call(OriginProcess, group_leader),
RelayPid = spawn(fun () -> relay(GroupLeader) end),
register(user, RelayPid),
spawn(
fun () ->
MRef = monitor(process, OriginProcess),
notify_when_started(dist, OriginProcess),
origin_link(MRef, OriginProcess)
end),
%% return RelayPid for user_sup to link to
RelayPid;
error ->
%% no -origin specified, meaning that standard I/O is used for alternative
spawn(fun io_server/0)
end.
relay(GroupLeader) ->
receive
IO ->
GroupLeader ! IO,
relay(GroupLeader)
end.
origin_link(MRef, Origin) ->
receive
{'DOWN', MRef, process, Origin, _Reason} ->
erlang:halt();
{init, started} ->
notify_started(dist, Origin),
origin_link(MRef, Origin)
end.
io_server() ->
process_flag(trap_exit, true),
Port = erlang:open_port({fd, 0, 1}, [eof, binary]),
register(user, self()),
group_leader(self(), self()),
notify_when_started(port, Port),
io_server_loop(port, Port, #{}, #{}, <<>>).
tcp_init(IpList, Port) ->
try
Sock = loop_connect(IpList, Port),
register(user, self()),
erlang:group_leader(self(), self()),
notify_when_started(tcp, Sock),
io_server_loop(tcp, Sock, #{}, #{}, undefined)
catch
Class:Reason:Stack ->
io:format(standard_io, "TCP connection failed: ~s:~p~n~120p~n", [Class, Reason, Stack]),
erlang:halt(1)
end.
loop_connect([], _Port) ->
error(noconnection);
loop_connect([Ip | More], Port) ->
case gen_tcp:connect(Ip, Port, [binary, {packet, 4}], ?CONNECT_TIMEOUT) of
{ok, Sock} ->
Sock;
_Error ->
loop_connect(More, Port)
end.
%% Message protocol between peers
io_server_loop(Kind, Port, Refs, Out, PortBuf) ->
receive
{io_request, From, ReplyAs, Request} when is_pid(From) ->
%% i/o request from this node, forward it to origin
peer_to_origin(Kind, Port, {io_request, From, ReplyAs, Request}),
io_server_loop(Kind, Port, Refs, Out, PortBuf);
{Port, {data, Bytes}} when Kind =:= port ->
{_Str, NewBin} = decode_port_data(Bytes, <<>>, PortBuf),
{NewRefs, NewOut, NewBuf} = handle_port_alternative(NewBin, Refs, Out),
io_server_loop(Kind, Port, NewRefs, NewOut, NewBuf);
{Port, eof} when Kind =:= port ->
%% stdin closed, if there is no active alternative, stop the node
erlang:halt(1);
{'EXIT', Port, badsig} when Kind =:= port ->
%% ignore badsig (what is it?)
io_server_loop(Kind, Port, Refs, Out, PortBuf);
{'EXIT', Port, _Reason} when Kind =:= port ->
%% stdin closed, if there is no active alternative, stop the node
erlang:halt(1);
{tcp, Port, Data} when Kind =:= tcp ->
ok = inet:setopts(Port, [{active, once}]), %% flow control
{NewRefs, NewOut} = handle_peer_alternative(binary_to_term(Data), Refs, Out),
io_server_loop(Kind, Port, NewRefs, NewOut, PortBuf);
{tcp_closed, Port} when Kind =:= tcp ->
%% TCP connection closed, time to shut down
erlang:halt(1);
{reply, Seq, Class, Reply} when is_integer(Seq), is_map_key(Seq, Out) ->
%% stdin/stdout RPC
{CallerRef, Out2} = maps:take(Seq, Out),
Refs2 = maps:remove(CallerRef, Refs),
erlang:demonitor(CallerRef, [flush]),
peer_to_origin(Kind, Port, {reply, Seq, Class, Reply}),
io_server_loop(Kind, Port, Refs2, Out2, PortBuf);
%% stdin/stdout message forwarding
{message, To, Content} ->
peer_to_origin(Kind, Port, {message, To, Content}),
io_server_loop(Kind, Port, Refs, Out, PortBuf);
{'DOWN', CallerRef, _, _, Reason} ->
%% this is really not expected to happen, because "do_call"
View remainder of file in raw view
0