File 1691-peer-peer-node-implementation.patch of Package erlang

Overview Repositories Revisions Requests Users Attributes Meta

File 1691-peer-peer-node-implementation.patch of Package erlang (Revision 57cf7c87cb80e2aaaf3f20327bf57529)

Currently displaying revision 57cf7c87cb80e2aaaf3f20327bf57529 , Show latest

From 8ebb2cd13b1897e6ab57f1e2b9ae1f4a5f92461b Mon Sep 17 00:00:00 2001
From: Maxim Fedorov <dane@whatsapp.com>
Date: Tue, 15 Jun 2021 11:08:51 +0300
Subject: [PATCH 1/7] peer: peer node implementation

Add a new module 'peer' implementing a linked peer node.
It is designed to replace 'slave', 'ct_slave', 'loose_node',
and a few more implementations of the same concept.
---
 bootstrap/lib/stdlib/ebin/stdlib.app |   1 +
 erts/doc/src/erl_cmd.xml             |   2 +-
 lib/stdlib/doc/src/Makefile          |   1 +
 lib/stdlib/doc/src/peer.xml          | 539 ++++++++++++++++
 lib/stdlib/doc/src/ref_man.xml       |   1 +
 lib/stdlib/doc/src/specs.xml         |   1 +
 lib/stdlib/src/Makefile              |   1 +
 lib/stdlib/src/peer.erl              | 918 +++++++++++++++++++++++++++
 lib/stdlib/src/stdlib.app.src        |   1 +
 lib/stdlib/test/Makefile             |   1 +
 lib/stdlib/test/peer_SUITE.erl       | 463 ++++++++++++++
 11 files changed, 1928 insertions(+), 1 deletion(-)
 create mode 100644 lib/stdlib/doc/src/peer.xml
 create mode 100644 lib/stdlib/src/peer.erl
 create mode 100644 lib/stdlib/test/peer_SUITE.erl

diff --git a/bootstrap/lib/stdlib/ebin/stdlib.app b/bootstrap/lib/stdlib/ebin/stdlib.app
index 322c34bab2..04eadf5b70 100644
--- a/bootstrap/lib/stdlib/ebin/stdlib.app
+++ b/bootstrap/lib/stdlib/ebin/stdlib.app
@@ -81,6 +81,7 @@
 	 orddict,
 	 ordsets,
 	 otp_internal,
+	 peer,
 	 pool,
 	 proc_lib,
 	 proplists,
diff --git a/erts/doc/src/erl_cmd.xml b/erts/doc/src/erl_cmd.xml
index 7ef3ca88f9..da3d6cb3b8 100644
--- a/erts/doc/src/erl_cmd.xml
+++ b/erts/doc/src/erl_cmd.xml
@@ -400,7 +400,7 @@ $ <input>erl \
 the name supplied together with flag <c><![CDATA[-sname]]></c> or
 <c><![CDATA[-name]]></c>.
 </item>
- <tag><c><![CDATA[-init_debug]]></c></tag>
+ <tag><marker id="init_debug"/><c><![CDATA[-init_debug]]></c></tag>
 <item>
 Makes <c><![CDATA[init]]></c> write some debug information while
 interpreting the boot script.
diff --git a/lib/stdlib/doc/src/Makefile b/lib/stdlib/doc/src/Makefile
index fbec85ec53..5e8fc7b90b 100644
--- a/lib/stdlib/doc/src/Makefile
+++ b/lib/stdlib/doc/src/Makefile
@@ -74,6 +74,7 @@ XML_REF3_FILES = \
 	ms_transform.xml \
 	orddict.xml \
 	ordsets.xml \
+	peer.xml \
 	pool.xml \
 	proc_lib.xml \
 	proplists.xml \
diff --git a/lib/stdlib/doc/src/peer.xml b/lib/stdlib/doc/src/peer.xml
new file mode 100644
index 0000000000..d67385157f
--- /dev/null
+++ b/lib/stdlib/doc/src/peer.xml
@@ -0,0 +1,539 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<!DOCTYPE erlref SYSTEM "erlref.dtd">
+
+
+
+<erlref>
+ <header>
+ <copyright>
+ <year>2021</year><year>2021</year>
+ <holder>Maxim Fedorov, WhatsApp Inc.</holder>
+ </copyright>
+ <legalnotice>
+ 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.
+
+ </legalnotice>
+
+ <title>peer</title>
+ <prepared>maximfca@gmail.com</prepared>
+ <docno></docno>
+ <date></date>
+ <rev></rev>
+ <file>peer.xml</file>
+ </header>
+ <module since="OTP 25.0">peer</module>
+ <modulesummary>Start and control linked Erlang nodes.
+ </modulesummary>
+ <description>
+ 
+ This module provides functions for starting linked Erlang nodes.
+ The node spawning new nodes is called origin, and newly started
+ nodes are peer nodes, or peers. A peer node automatically
+ terminates when it loses the control connection to the origin. This
+ connection could be an Erlang distribution connection, or an alternative -
+ TCP or standard I/O. The alternative connection provides a way to execute
+ remote procedure calls even when Erlang Distribution is not available,
+ allowing to test the distribution itself.
+ 
+
+ 
+ Peer node terminal input/output is relayed through the origin.
+ If a standard I/O alternative connection is requested, console output
+ also goes via the origin, allowing debugging of node startup and boot
+ script execution (see <seecom marker="erts:erl#init_debug">
+ <c>-init_debug</c></seecom>). File I/O is not redirected, contrary to
+ <seeerl marker="slave"><c>slave(3)</c></seeerl> behaviour.
+ 
+
+ 
+ The peer node can start on the same or a different host (via <c>ssh</c>)
+ or in a separate container (for example Docker).
+ When the peer starts on the same host as the origin, it inherits
+ the current directory and environment variables from the origin.
+ 
+
+ <note>
+ 
+ This module is designed to facilitate multi-node testing with Common Test.
+ Use the <c>?CT_PEER()</c> macro to start a linked peer node according to
+ Common Test conventions: crash dumps written to specific location, node
+ name prefixed with module name, calling function, and origin OS process
+ ID). Use <seemfa marker="#random_name/1"><c>random_name/1</c></seemfa> to
+ create sufficiently unique node names if you need more control.
+ 
+ 
+ A peer node started without alternative connection behaves similarly
+ to <seeerl marker="slave"><c>slave(3)</c></seeerl>. When an alternative
+ connection is requested, the behaviour is similar to
+ <c>test_server:start_node(Name, peer, Args).</c>
+ </note>
+
+ </description>
+
+ <section>
+ <title>Example</title>
+ 
+ The following example implements a test suite starting extra Erlang nodes.
+ It employs a number of techniques to speed up testing and reliably shut
+ down peer nodes:
+ 
+ <list>
+ <item>peers start linked to test runner process. If the test case fails,
+ the peer node is stopped automatically, leaving no rogue nodes running in
+ the background</item>
+ <item>arguments used to start the peer are saved in the control process
+ state for manual analysis. If the test case fails, the CRASH REPORT contains
+ these arguments</item>
+ <item>multiple test cases can run concurrently speeding up overall testing
+ process, peer node names are unique even when there are multiple instances
+ of the same test suite running in parallel</item>
+ </list>
+ <code type="erl">
+ -module(my_SUITE).
+ -behaviour(ct_suite).
+ -export([all/0, groups/0]).
+ -export([basic/1, args/1, named/1, restart_node/1, multi_node/1]).
+
+ -include_lib("common_test/include/ct.hrl").
+
+ groups() ->
+ [{quick, [parallel],
+ [basic, args, named, restart_node, multi_node]}].
+
+ all() ->
+ [{group, quick}].
+
+ basic(Config) when is_list(Config) ->
+ {ok, Peer, _Node} = ?CT_PEER(),
+ peer:stop(Peer).
+
+ args(Config) when is_list(Config) ->
+ %% specify additional arguments to the new node
+ {ok, Peer, _Node} = ?CT_PEER(["-emu_flavor", "smp"]),
+ peer:stop(Peer).
+
+ named(Config) when is_list(Config) ->
+ %% pass test case name down to function starting nodes
+ Peer = start_node_impl(named_test),
+ peer:stop(Peer).
+
+ start_node_impl(ActualTestCase) ->
+ {ok, Peer, Node} = ?CT_PEER(#{name => ?CT_PEER_NAME(ActualTestCase)}),
+ %% extra setup needed for multiple test cases
+ ok = rpc:call(Node, application, set_env, [kernel, key, value]),
+ Peer.
+
+ restart_node(Config) when is_list(Config) ->
+ Name = ?CT_PEER_NAME(),
+ {ok, Peer, Node} = ?CT_PEER(#{name => Name}),
+ peer:stop(Peer),
+ %% restart the node with the same name as before
+ {ok, Peer2, Node} = ?CT_PEER(#{name => Name, args => ["+fnl"]}),
+ peer:stop(Peer2).
+ </code>
+
+ 
+ The next example demonstrates how to start multiple nodes concurrently:
+ 
+ <code type="erl">
+ multi_node(Config) when is_list(Config) ->
+ Peers = [?CT_PEER(#{wait_boot => {self(), tag}})
+ || _ &lt;- lists:seq(1, 4)],
+ %% wait for all nodes to complete boot process, get their names:
+ _Nodes = [receive {tag, {started, Node, Peer}} -> Node end
+ || {ok, Peer} &lt;- Peers],
+ [peer:stop(Peer) || {ok, Peer} &lt;- Peers].
+ </code>
+
+ 
+ Start a peer on a different host. Requires <c>ssh</c> key-based
+ authentication set up, allowing "another_host" connection without password
+ prompt.
+ 
+ <code type="erl">
+ Ssh = os:find_executable("ssh"),
+ peer:start_link(#{exec => {Ssh, ["another_host", "erl"]},
+ connection => standard_io}),
+ </code>
+
+ 
+ The following Common Test case demonstrates Docker integration, starting two
+ containers with hostnames "one" and "two". In this example Erlang nodes
+ running inside containers form an Erlang cluster.
+ 
+ <code type="erl">
+ docker(Config) when is_list(Config) ->
+ Docker = os:find_executable("docker"),
+ PrivDir = proplists:get_value(priv_dir, Config),
+ build_release(PrivDir),
+ build_image(PrivDir),
+
+ %% start two Docker containers
+ {ok, Peer, Node} = peer:start_link(#{name => lambda,
+ connection => standard_io,
+ exec => {Docker, ["run", "-h", "one", "-i", "lambda"]}}),
+ {ok, Peer2, Node2} = peer:start_link(#{name => lambda,
+ connection => standard_io,
+ exec => {Docker, ["run", "-h", "two", "-i", "lambda"]}}),
+
+ %% find IP address of the second node using alternative connection RPC
+ {ok, Ips} = peer:call(Peer2, inet, getifaddrs, []),
+ {"eth0", Eth0} = lists:keyfind("eth0", 1, Ips),
+ {addr, Ip} = lists:keyfind(addr, 1, Eth0),
+
+ %% make first node to discover second one
+ ok = peer:call(Peer, inet_db, set_lookup, [[file]]),
+ ok = peer:call(Peer, inet_db, add_host, [Ip, ["two"]]),
+
+ %% join a cluster
+ true = peer:call(Peer, net_kernel, connect_node, [Node2]),
+ %% verify that second peer node has only the first node visible
+ [Node] = peer:call(Peer2, erlang, nodes, []),
+
+ %% stop peers, causing containers to also stop
+ peer:stop(Peer2),
+ peer:stop(Peer).
+
+ build_release(Dir) ->
+ %% load sasl.app file, otherwise application:get_key will fail
+ application:load(sasl),
+ %% create *.rel - release file
+ RelFile = filename:join(Dir, "lambda.rel"),
+ Release = {release, {"lambda", "1.0.0"},
+ {erts, erlang:system_info(version)},
+ [{App, begin {ok, Vsn} = application:get_key(App, vsn), Vsn end}
+ || App &lt;- [kernel, stdlib, sasl]]},
+ ok = file:write_file(RelFile, list_to_binary(lists:flatten(
+ io_lib:format("~tp.", [Release])))),
+ RelFileNoExt = filename:join(Dir, "lambda"),
+
+ %% create boot script
+ {ok, systools_make, []} = systools:make_script(RelFileNoExt,
+ [silent, {outdir, Dir}]),
+ %% package release into *.tar.gz
+ ok = systools:make_tar(RelFileNoExt, [{erts, code:root_dir()}]).
+
+ build_image(Dir) ->
+ %% Create Dockerfile example, working only for Ubuntu 20.04
+ %% Expose port 4445, and make Erlang distribution to listen
+ %% on this port, and connect to it without EPMD
+ %% Set cookie on both nodes to be the same.
+ BuildScript = filename:join(Dir, "Dockerfile"),
+ Dockerfile =
+ "FROM ubuntu:20.04 as runner\n"
+ "EXPOSE 4445\n"
+ "WORKDIR /opt/lambda\n"
+ "COPY lambda.tar.gz /tmp\n"
+ "RUN tar -zxvf /tmp/lambda.tar.gz -C /opt/lambda\n"
+ "ENTRYPOINT [\"/opt/lambda/erts-" ++ erlang:system_info(version) ++
+ "/bin/dyn_erl\", \"-boot\", \"/opt/lambda/releases/1.0.0/start\","
+ " \"-kernel\", \"inet_dist_listen_min\", \"4445\","
+ " \"-erl_epmd_port\", \"4445\","
+ " \"-setcookie\", \"secret\"]\n",
+ ok = file:write_file(BuildScript, Dockerfile),
+ os:cmd("docker build -t lambda " ++ Dir).
+ </code>
+ </section>
+
+ <datatypes>
+ <datatype>
+ <name name="start_options"/>
+ <desc>
+ 
+ Options that can be used when starting
+ a <c>peer</c> node through <seemfa marker="#start/1"><c>start/1</c></seemfa>
+ and <seemfa marker="#start_link/0"><c>start_link/0,1</c></seemfa>.
+ 
+ <taglist>
+ <tag><c>name</c></tag>
+ <item>
+ 
+ Node name (the part before "@"). When <c>name</c> is not specified, but <c>host</c>
+ is, <c>peer</c> follows compatibility behaviour and uses the origin node name.
+ 
+ </item>
+ <tag><c>host</c></tag>
+ <item>
+ 
+ Enforces a specific host name. Can be used to override the default
+ behaviour and start "node@localhost" instead of "node@realhostname".
+ 
+ </item>
+ <tag><c>longnames</c></tag>
+ <item>
+ 
+ Use long names to start a node. Default is taken from the origin
+ using <c>net_kernel:longnames()</c>. If the origin is not distributed,
+ short names is the default.
+ 
+ </item>
+ <tag><c>peer_down</c></tag>
+ <item>
+ 
+ Defines the peer control process behaviour when the control connection is
+ closed from the peer node side (for example when the peer crashes or dumps core).
+ When set to <c>stop</c> (default), a lost control connection causes
+ the control process to exit normally. Setting <c>peer_down</c> to <c>continue</c>
+ keeps the control process running, and <c>crash</c> will cause
+ the controlling process to exit abnormally.
+ 
+ </item>
+ <tag><c>exec</c></tag>
+ <item>
+ 
+ Alternative mechanism to start peer nodes with, for example, ssh instead of the
+ default bash.
+ 
+ </item>
+ <tag><c>connection</c></tag>
+ <item>
+ Alternative connection specification. See the
+ <seetype marker="#connection"><c>connection</c> datatype</seetype>.
+ </item>
+ <tag><c>args</c></tag>
+ <item>
+ Extra command line arguments to append to the "erl" command. Arguments are
+ passed as is, no escaping or quoting is needed or accepted.
+ </item>
+ <tag><c>env</c></tag>
+ <item>
+ 
+ List of environment variables with their values. This list is applied
+ to a locally started executable. If you need to change the environment of
+ the remote peer, adjust <c>args</c> to contain
+ <c>-env ENV_KEY ENV_VALUE</c>.
+ 
+ </item>
+ <tag><c>wait_boot</c></tag>
+ <item>
+ Specifies the start/start_link timeout.
+ See <seetype marker="#wait_boot"><c>wait_boot</c> datatype</seetype>.
+ 
+ </item>
+ <tag><c>shutdown</c></tag>
+ <item>
+ Specifies the peer node stopping behaviour. See
+ <seemfa marker="#stop/1"><c>stop()</c></seemfa>.
+ </item>
+ </taglist>
+ </desc>
+ </datatype>
+ <datatype>
+ <name name="peer_state"/>
+ <desc>Peer node state.</desc>
+ </datatype>
+ <datatype>
+ <name name="connection"/>
+ <desc>Alternative connection between the origin and the peer. When the
+ connection closes, the peer node terminates automatically. If
+ the <c>peer_down</c> startup flag is set to <c>crash</c>, the controlling
+ process on the origin node exits with corresponding reason, effectively
+ providing a two-way link. 
+ When <c>connection</c> is set to a port number, the origin starts listening on
+ the requested TCP port, and the peer node connects to the port. When it is set to
+ an <c>{IP, Port}</c> tuple, the origin listens only on the specified IP. The port
+ number can be set to 0 for automatic selection.
+ 
+ Using the <c>standard_io</c> alternative connection starts the peer attached to
+ the origin (other connections use <c>-detached</c> flag to erl). In this mode
+ peer and origin communicate via stdin/stdout.
+ 
+ </desc>
+ </datatype>
+ <datatype>
+ <name name="exec"/>
+ <desc>
+ 
+ Overrides executable to start peer nodes with. By default it is
+ the path to "erl", taken from <c>init:get_argument(progname)</c>.
+ If <c>progname</c> is not known, <c>peer</c> makes best guess given the current
+ ERTS version.
+ 
+ 
+ When a tuple is passed, the first element is the path to executable,
+ and the second element is prepended to the final command line. This can be used
+ to start peers on a remote host or in a Docker container. See the examples
+ above.
+ 
+ 
+ This option is useful for testing backwards compatibility with previous releases,
+ installed at specific paths, or when the Erlang installation location
+ is missing from the <c>PATH</c>.
+ 
+ </desc>
+ </datatype>
+ <datatype>
+ <name name="wait_boot"/>
+ <desc>Specifies start/start_link timeout in milliseconds. Can be set to
+ <c>false</c>, allowing the peer to start asynchronously. If <c>{Pid, Tag}</c>
+ is specified instead of a timeout, the peer will send <c>Tag</c> to the
+ requested process.</desc>
+ </datatype>
+ </datatypes>
+
+ <funcs>
+
+ <func>
+ <name name="call" arity="4" since="OTP 25.0"/>
+ <name name="call" arity="5" since="OTP 25.0"/>
+ <fsummary>Evaluates a function call on a peer node.</fsummary>
+ <desc>
+ 
+ Uses the alternative connection to
+ evaluate <c>apply(<anno>Module</anno>, <anno>Function</anno>,
+ <anno>Args</anno>)</c> on the peer node and returns
+ the corresponding value <c><anno>Result</anno></c>.
+ <c><anno>Timeout</anno></c> is an integer representing
+ the timeout in milliseconds or the atom <c>infinity</c>
+ which prevents the operation from ever timing out.
+ 
+ 
+ When an alternative connection is not requested, this
+ function will raise <c>exit</c> signal with the <c>noconnection</c>
+ reason. Use <seeerl marker="kernel:erpc"><c>erpc</c></seeerl> module
+ to communicate over Erlang distribution.
+ 
+ </desc>
+ </func>
+
+ <func>
+ <name name="cast" arity="4" since="OTP 25.0"/>
+ <fsummary>Evaluates a function call on a peer node ignoring the result.</fsummary>
+ <desc>
+ 
+ Uses the alternative connection to
+ evaluate <c>apply(<anno>Module</anno>, <anno>Function</anno>,
+ <anno>Args</anno>)</c> on the peer node. No response is delivered to the
+ calling process.
+ 
+ 
+ <c>peer:cast/4</c> fails silently when the alternative connection is not
+ configured. Use <seeerl marker="kernel:erpc"><c>erpc</c></seeerl> module
+ to communicate over Erlang distribution.
+ 
+ </desc>
+ </func>
+
+ <func>
+ <name name="send" arity="3" since="OTP 25.0"/>
+ <fsummary>Sends a message to a process on the peer node.</fsummary>
+ <desc>
+ 
+ Uses the alternative connection to send <anno>Message</anno> to a process on the
+ the peer node. Silently fails if no alternative connection is configured.
+ The process can be referenced by process ID or registered name.
+ 
+ </desc>
+ </func>
+
+ <func>
+ <name name="get_state" arity="1" since="OTP 25.0"/>
+ <fsummary>Returns peer node state.</fsummary>
+ <desc>
+ Returns the peer node state. Th initial state is <c>booting</c>; the node stays in that
+ state until then boot script is complete, and then the node progresses to <c>running</c>.
+ If the node stops (gracefully or not), the state changes to <c>down</c>.
+ 
+ </desc>
+ </func>
+
+ <func>
+ <name name="random_name" arity="0" since="OTP 25.0"/>
+ <fsummary>Creates a sufficiently unique node name.</fsummary>
+ <desc>
+ 
+ The same as <seemfa marker="#random_name/1"><c>random_name(peer)</c></seemfa>.
+ 
+ </desc>
+ </func>
+
+ <func>
+ <name name="random_name" arity="1" since="OTP 25.0"/>
+ <fsummary>Creates a sufficiently unique node name given a prefix.</fsummary>
+ <desc>
+ 
+ Creates a sufficiently unique node name for the current host,
+ combining a prefix, a unique number, and the current OS process ID.
+ 
+ <note>
+ 
+ Use the <c>?CT_PEER(["erl_arg1"])</c> macro provided by Common Test
+ <c>-include_lib("common_test/include/ct.hrl")</c> for convenience.
+ It starts a new peer using Erlang distribution as the control channel,
+ supplies thes calling module's code path to the peer, and uses the calling
+ function name for the name prefix.
+ 
+ </note>
+ </desc>
+ </func>
+
+ <func>
+ <name name="start" arity="1" since="OTP 25.0"/>
+ <fsummary>Starts a peer node.</fsummary>
+ <desc>
+ 
+ Starts a peer node with the specified
+ <seetype marker="#start_options"><c>start_options()</c></seetype>.
+ Returns the controlling process and the full peer node name, unless
+ <c>wait_boot</c> is not requested and the host name is not known in advance.
+ 
+ </desc>
+ </func>
+
+ <func>
+ <name name="start_link" arity="0" since="OTP 25.0"/>
+ <fsummary>Starts a peer node, and links controlling process to caller process.</fsummary>
+ <desc>
+ 
+ The same as
+ <seemfa marker="#start_link/1"><c>start_link(#{name => random_name()})</c></seemfa>.
+ 
+ </desc>
+ </func>
+
+ <func>
+ <name name="start_link" arity="1" since="OTP 25.0"/>
+ <fsummary>Starts a peer node, and links controlling process to caller process.</fsummary>
+ <desc>
+ Starts a peer node in the same way as <seemfa marker="#start/1"><c>start/1</c></seemfa>,
+ except that the peer node is linked to the currently
+ executing process. If that process terminates, the peer node
+ also terminates.
+ 
+ Accepts <seetype marker="#start_options"><c>start_options()</c></seetype>.
+ Returns the controlling process and the full peer node name, unless <c>wait_boot</c> is not
+ requested and host name is not known in advance.
+ 
+ 
+ When the <c>standard_io</c> alternative connection is requested, and <c>wait_boot</c> is
+ not set to <c>false</c>, a failed peer boot sequence causes the caller to exit with
+ the <c>{boot_failed, {exit_status, ExitCode}}</c> reason.
+ 
+ </desc>
+ </func>
+
+ <func>
+ <name name="stop" arity="1" since="OTP 25.0"/>
+ <fsummary>Stop controlling process and terminate peer node.</fsummary>
+ <desc>
+ Depending on the <c>shutdown</c> option used to start the peer,
+ stopping may be graceful (via <c>init:stop()</c>) or less so,
+ by just closing the control connection, which triggers <c>erlang:halt()</c>
+ executed by the peer.
+ A graceful shutdown can be slower, or even block indefinitely.
+ </desc>
+ </func>
+
+ </funcs>
+</erlref>
+
diff --git a/lib/stdlib/doc/src/ref_man.xml b/lib/stdlib/doc/src/ref_man.xml
index ebb198bf2b..cf4333c205 100644
--- a/lib/stdlib/doc/src/ref_man.xml
+++ b/lib/stdlib/doc/src/ref_man.xml
@@ -74,6 +74,7 @@
 <xi:include href="ms_transform.xml"/>
 <xi:include href="orddict.xml"/>
 <xi:include href="ordsets.xml"/>
+ <xi:include href="peer.xml"/>
 <xi:include href="pool.xml"/>
 <xi:include href="proc_lib.xml"/>
 <xi:include href="proplists.xml"/>
diff --git a/lib/stdlib/doc/src/specs.xml b/lib/stdlib/doc/src/specs.xml
index 75266530f2..5f7e3555d8 100644
--- a/lib/stdlib/doc/src/specs.xml
+++ b/lib/stdlib/doc/src/specs.xml
@@ -41,6 +41,7 @@
 <xi:include href="../specs/specs_ms_transform.xml"/>
 <xi:include href="../specs/specs_orddict.xml"/>
 <xi:include href="../specs/specs_ordsets.xml"/>
+ <xi:include href="../specs/specs_peer.xml"/>
 <xi:include href="../specs/specs_pool.xml"/>
 <xi:include href="../specs/specs_proc_lib.xml"/>
 <xi:include href="../specs/specs_proplists.xml"/>
diff --git a/lib/stdlib/src/Makefile b/lib/stdlib/src/Makefile
index 8a45b8e249..9f04fef6ef 100644
--- a/lib/stdlib/src/Makefile
+++ b/lib/stdlib/src/Makefile
@@ -102,6 +102,7 @@ MODULES= \
 	orddict \
 	ordsets \
 	re \
+	peer \
 	pool \
 	proc_lib \
 	proplists \
diff --git a/lib/stdlib/src/peer.erl b/lib/stdlib/src/peer.erl
new file mode 100644
index 0000000000..a6407590cc
--- /dev/null
+++ b/lib/stdlib/src/peer.erl
@@ -0,0 +1,918 @@
+%% @doc
+%% Controller for additional Erlang node running on the same host,
+%% or in 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 => brutal_kill | %% halt the peer brutally (default)
+ timeout() %% send init:stop() request and wait up to specified timeout
+ }.
+
+%% Peer node states
+-type peer_state() :: booting | running | {down, Reason :: term()}.
+
+-export_type([
+ start_options/0,
+ peer_state/0,
+ exec/0
+ ]).
+
+%% 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).
+
+%% @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_link).
+
+%% @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,
+
+ State = #peer_state{options = Options, 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: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, brutal_kill), maps:find(connection, Options)} of
+ {brutal_kill, {ok, standard_io}} ->
+ Port /= undefined andalso (catch erlang:port_close(Port));
+ {brutal_kill, {ok, _TCP}} ->
+ Port /= undefined andalso (catch gen_tcp:close(Port));
+ {brutal_kill, error} ->
+ net_kernel:disconnect(Node);
+ {Shutdown, error} ->
+ Timeout = shutdown(dist, undefined, Node, Shutdown),
+ receive {nodedown, Node} -> ok after Timeout -> ok end;
+ {Shutdown, {ok, standard_io}} ->
+ Timeout = shutdown(port, Port, Node, Shutdown),
+ receive {'EXIT', Port, _Reason2} -> ok after Timeout -> ok end,
+ catch erlang:port_close(Port);
+ {Shutdown, {ok, _TCP}} ->
+ Timeout = shutdown(tcp, Port, Node, Shutdown),
+ receive {tcp_closed, Port} -> ok after Timeout -> ok end,
+ catch catch gen_tcp:close(Port)
+ end,
+ ok.
+
+%%--------------------------------------------------------------------
+%% Internal implementation
+
+%% This hack is a temporary workaround for test coverage reports
+shutdown(_Type, _Port, _Node, Timeout) when is_integer(Timeout); timeout =:= infinity ->
+ 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.
+
+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"
+ %% catches all exceptions
+ {Seq, Refs3} = maps:take(CallerRef, Refs),
+ {CallerRef, Out3} = maps:take(Seq, Out),
+ peer_to_origin(Kind, Port, {reply, Seq, crash, Reason}),
+ io_server_loop(Kind, Port, Refs3, Out3, PortBuf);
+ {init, started} ->
+ notify_started(Kind, Port),
+ io_server_loop(Kind, Port, Refs, Out, PortBuf);
+ _Other ->
+ %% below, what is it?
+ io_server_loop(Kind, Port, Refs, Out, PortBuf)
+ end.
+
+handle_peer_alternative({io_reply, From, FromRef, Reply}, Refs, Out) ->
+ From ! {io_reply, FromRef, Reply},
+ {Refs, Out};
+handle_peer_alternative({call, Seq, M, F, A}, Refs, Out) ->
+ CallerRef = do_call(Seq, M, F, A),
+ {Refs#{CallerRef => Seq}, Out#{Seq => CallerRef}};
+handle_peer_alternative({cast, M, F, A}, Refs, Out) ->
+ %% spawn a separate process to avoid blocking further RPC
+ spawn(fun () -> erlang:apply(M, F, A) end),
+ {Refs, Out};
+handle_peer_alternative({message, Dest, Message}, Refs, Out) ->
+ Dest ! Message,
+ {Refs, Out}.
+
+%% single port input message may contain multiple messages
+handle_port_alternative(<<Size:32, Payload:Size/binary, Crc:32, Rest/binary>>, Refs, Out) ->
+ Crc = erlang:crc32(Payload), %% assert
+ {NewRefs, NewOut} = handle_peer_alternative(binary_to_term(Payload), Refs, Out),
+ handle_port_alternative(Rest, NewRefs, NewOut);
+handle_port_alternative(Rest, Refs, Out) ->
+ {Refs, Out, Rest}.
+
+do_call(Seq, M, F, A) ->
+ Proxy = self(),
+ {_, CallerRef} =
+ spawn_monitor(
+ fun () ->
+ %% catch all errors, otherwise emulator will log
+ %% ERROR REPORT when it is not expected
+ try
+ Proxy ! {reply, Seq, ok, erlang:apply(M, F, A)}
+ catch
+ Class:Reason:Stack ->
+ Proxy ! {reply, Seq, Class, {Reason, Stack}}
+ end
+ end),
+ CallerRef.
diff --git a/lib/stdlib/src/stdlib.app.src b/lib/stdlib/src/stdlib.app.src
index f6a908a819..7778c8deb1 100644
--- a/lib/stdlib/src/stdlib.app.src
+++ b/lib/stdlib/src/stdlib.app.src
@@ -81,6 +81,7 @@
 	 orddict,
 	 ordsets,
 	 otp_internal,
+	 peer,
 	 pool,
 	 proc_lib,
 	 proplists,
diff --git a/lib/stdlib/test/Makefile b/lib/stdlib/test/Makefile
index bda16c263d..7a92c7b621 100644
--- a/lib/stdlib/test/Makefile
+++ b/lib/stdlib/test/Makefile
@@ -51,6 +51,7 @@ MODULES= \
 	log_mf_h_SUITE \
 	math_SUITE \
 	ms_transform_SUITE \
+	peer_SUITE \
 	pool_SUITE \
 	proc_lib_SUITE \
 	proplists_SUITE \
diff --git a/lib/stdlib/test/peer_SUITE.erl b/lib/stdlib/test/peer_SUITE.erl
new file mode 100644
index 0000000000..772f57c8ea
--- /dev/null
+++ b/lib/stdlib/test/peer_SUITE.erl
@@ -0,0 +1,463 @@
+%%% @doc
+%%% Smoke tests for peer node controller
+%%% @end
+-module(peer_SUITE).
+-author("maximfca@gmail.com").
+
+%% Common Test headers
+-include_lib("stdlib/include/assert.hrl").
+-include_lib("common_test/include/ct.hrl").
+
+%% Test server callbacks
+-export([
+ suite/0,
+ all/0,
+ groups/0,
+ init_per_group/2,
+ end_per_group/2
+]).
+
+%% Test cases
+-export([
+ dist/0, dist/1,
+ peer_down_crash/0, peer_down_crash/1,
+ peer_down_continue/0, peer_down_continue/1,
+ peer_down_boot/0, peer_down_boot/1,
+ dist_io_redirect/0, dist_io_redirect/1,
+ dist_localhost/0, dist_localhost/1,
+ errors/0, errors/1,
+ basic/0, basic/1,
+ peer_states/0, peer_states/1,
+ cast/0, cast/1,
+ detached/0, detached/1,
+ dyn_peer/0, dyn_peer/1,
+ stop_peer/0, stop_peer/1,
+ init_debug/0, init_debug/1,
+ io_redirect/0, io_redirect/1,
+ multi_node/0, multi_node/1,
+ dist_up_down/0, dist_up_down/1,
+ duplicate_name/0, duplicate_name/1,
+ old_release/0, old_release/1,
+ ssh/0, ssh/1,
+ docker/0, docker/1
+]).
+
+suite() ->
+ [{timetrap, {minutes, 1}}].
+
+alternative() ->
+ [basic, peer_states, cast, detached, dyn_peer, stop_peer, io_redirect,
+ multi_node, duplicate_name].
+
+groups() ->
+ [
+ {dist, [parallel], [errors, dist, peer_down_crash, peer_down_continue, peer_down_boot,
+ dist_io_redirect, dist_up_down, dist_localhost]},
+ {tcp, [parallel], alternative()},
+ {standard_io, [parallel], [init_debug | alternative()]},
+ {compatibility, [parallel], [old_release]},
+ {remote, [parallel], [ssh]}
+ ].
+
+all() ->
+ [{group, dist}, {group, tcp}, {group, standard_io}, {group, compatibility}, {group, remote}].
+
+init_per_group(remote, Config) ->
+ %% check that SSH can connect to localhost, skip the test if not
+ try os:cmd("ssh localhost echo ok") of
+ "ok\n" -> Config;
+ _ -> {skip, "'ssh localhost echo ok' did not return ok"}
+ catch
+ Class:Reason ->
+ SkipReason = io_lib:format("'ssh localhost echo ok' failed with ~s:~p", [Class, Reason]),
+ {skip, SkipReason}
+ end;
+init_per_group(dist, Config) ->
+ case erlang:is_alive() of
+ true -> Config;
+ false -> {skip, "origin is not distributed"}
+ end;
+init_per_group(tcp, Config) ->
+ [{connection, 0} | Config];
+init_per_group(standard_io, Config) ->
+ [{connection, standard_io} | Config];
+init_per_group(_Group, Config) ->
+ Config.
+
+end_per_group(_TestCase, Config) ->
+ proplists:delete(connection, Config).
+
+%% -------------------------------------------------------------------
+%% Distribution-enabled cases
+
+errors() ->
+ [{doc, "Verify that invalid command line is rejected immediately"}].
+
+errors(Config) when is_list(Config) ->
+ ?assertException(error, {invalid_arg, atom}, peer:start(#{args => atom})),
+ ?assertException(error, {invalid_arg, $s}, peer:start(#{args => "string"})),
+ ?assertException(error, {invalid_arg, ["2"]}, peer:start(#{args => ["1", ["2"]]})),
+ ?assertException(error, {exec, atom}, peer:start(#{exec => atom, name => name})),
+ ?assertException(error, {exec, atom}, peer:start(#{exec => {atom, []}, name => name})),
+ ?assertException(error, {exec, atom}, peer:start(#{exec => {"erl", [atom]}, name => name})),
+ ?assertException(error, not_alive, peer:start(#{})). %% peer node won't start - it's not alive
+
+dist() ->
+ [{doc, "Classic behaviour: detached peer with no alternative, connects via dist"}].
+
+%% Classic 'slave': start new node locally, with random name, ask peer control process to exit normally
+dist(Config) when is_list(Config) ->
+ {ok, Peer, Node} = peer:start_link(),
+ %% distribution is expected to be connected
+ ?assertEqual(Node, erpc:call(Node, erlang, node, [])),
+ %% but not alternative...
+ ?assertException(error, noconnection, peer:call(Peer, erlang, node, [])),
+ ?assertEqual(true, net_kernel:disconnect(Node)),
+ %% ^^^ this makes the node go down, but we also need to ensure that net_kernel
+ %% finished processing
+ ct:sleep(500),
+ _ = sys:get_state(net_kernel),
+ ?assertException(exit, {noproc, _}, peer:get_state(Peer)).
+
+peer_down_crash() ->
+ [{doc, "Tests peer_down handling when crash mode is requested"}].
+
+peer_down_crash(Config) when is_list(Config) ->
+ %% two-way link: "crash" mode
+ {ok, Peer, Node} = peer:start_link(#{name => peer:random_name(?FUNCTION_NAME), peer_down => crash}),
+ %% verify node started locally
+ ?assertEqual(Node, erpc:call(Node, erlang, node, [])),
+ %% verify there is no alternative connection
+ ?assertException(error, noconnection, peer:call(Peer, erlang, node, [])),
+ %% unlink and monitor
+ unlink(Peer),
+ MRef = monitor(process, Peer),
+ ?assertEqual(true, net_kernel:disconnect(Node)),
+ %% ^^^ this makes the node go down
+ %% since two-way link is requested, it triggers peer to stop
+ receive
+ {'DOWN', MRef, process, Peer, {nodedown, Node}} ->
+ ok
+ after 2000 ->
+ link(Peer),
+ {fail, disconnect_timeout}
+ end.
+
+peer_down_continue() ->
+ [{doc, "Tests peer_down handling for continue setting"}].
+
+peer_down_continue(Config) when is_list(Config) ->
+ {ok, Peer, Node} = peer:start_link(#{name => peer:random_name(?FUNCTION_NAME), peer_down => continue}),
+ ?assertEqual(ok, erpc:cast(Node, erlang, halt, [])),
+ ct:sleep(500),
+ sys:replace_state(net_kernel, fun(S) -> sys:get_state(Peer), S end),
+ ?assertMatch({down, _}, peer:get_state(Peer)),
+ peer:stop(Peer).
+
+peer_down_boot() ->
+ [{doc, "Tests that peer node failing boot fails start_link correctly"}].
+
+peer_down_boot(Config) when is_list(Config) ->
+ ?assertException(exit, {boot_failed, {exit_status, 1}},
+ peer:start_link(#{connection => standard_io, name => peer:random_name(), args => ["-no_epmd"]})).
+
+dist_io_redirect() ->
+ [{doc, "Tests i/o redirection working for dist"}].
+
+dist_io_redirect(Config) when is_list(Config) ->
+ %% Common Test changes group leader process to capture output.
+ %% 'peer' relays output via control process group leader.
+ ct:capture_start(),
+ {ok, Peer, Node} = peer:start_link(#{name => peer:random_name(?FUNCTION_NAME),
+ args => ["-eval", "io:format(\"out\")."]}),
+ %% RPC is smart enough to set the group leader, so force 'user' output
+ %% to check that peer node redirects 'user' to the current process group
+ %% leader.
+ ?assertEqual(ok, erpc:call(Node, io, format, [user, "STRONGFLOUR.", []])),
+ %% verify that 'send' is ignored when no alternative connection is done
+ ?assertEqual(ok, peer:send(Peer, init, {stop, stop})),
+ %% check that RPC sets the group leader, even via 'apply'
+ ?assertEqual(ok, erpc:call(Node, erlang, apply, [io, format, ["second."]])),
+ %% 'eval' at the end may be quite slow, so have to wait here
+ ct:sleep(500),
+ peer:stop(Peer),
+ ct:capture_stop(),
+ Texts = ct:capture_get(),
+ %% order is not guaranteed, so sort explicitly
+ ?assertEqual(lists:sort(["STRONGFLOUR.", "second.", "out"]), lists:sort(Texts)).
+
+dist_up_down() ->
+ [{doc, "Test that Erlang distribution can go up and down (with TCP alternative)"}].
+
+dist_up_down(Config) when is_list(Config) ->
+ %% skip establishing full mesh, for it makes 'global' hang
+ %% TODO: fix 'global.erl' locker process so it does not hang
+ {ok, Peer, Node} = peer:start_link(#{name => peer:random_name(?FUNCTION_NAME),
+ connection => {{127, 0, 0, 1}, 0}, args => ["-connect_all", "false"]}),
+ ?assertEqual(true, net_kernel:connect_node(Node)),
+ ?assertEqual(true, net_kernel:disconnect(Node)),
+ ?assertEqual(true, net_kernel:connect_node(Node)),
+ peer:stop(Peer).
+
+dist_localhost() ->
+ [{doc, "Test that localhost and gethostname operate together"}].
+
+dist_localhost(Config) when is_list(Config) ->
+ {ok, Peer, Node} = peer:start_link(#{name => peer:random_name(?FUNCTION_NAME), host => "localhost"}),
+ ?assertMatch([_, "localhost"], string:lexemes(atom_to_list(Node), "@")),
+ %% start second peer, ensure they see each other
+ {ok, Host} = inet:gethostname(),
+ {ok, Peer2, Node2} = peer:start_link(#{name => peer:random_name(?FUNCTION_NAME), host => Host}),
+ true = erpc:call(Node, net_kernel, connect_node, [Node2]),
+ peer:stop(Peer),
+ peer:stop(Peer2).
+
+%% -------------------------------------------------------------------
+%% alternative connection cases
+
+%% Runs in the peer node context, forward a message from peer node to origin
+%% node via alternative connection.
+-spec forward(Dest :: pid() | atom(), Message :: term()) -> term().
+forward(Dest, Message) ->
+ group_leader() ! {message, Dest, Message}.
+
+basic() ->
+ [{doc, "Tests peer node start, and do some RPC via stdin/stdout"}].
+
+basic(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ {ok, Peer, _Node} = peer:start_link(#{connection => Conn}),
+ %% test the alternative connection
+ ?assertEqual('nonode@nohost', peer:call(Peer, erlang, node, [])),
+ ?assertException(throw, ball, peer:call(Peer, erlang, throw, [ball])),
+ %% setup code path to this module (needed to "fancy RPC")
+ Path = filename:dirname(code:which(?MODULE)),
+ ?assertEqual(true, peer:call(Peer, code, add_path, [Path])),
+ %% fancy RPC via message exchange (uses forwarding from the peer)
+ Control = self(),
+ RFun = fun() -> receive do -> forward(Control, done) end end,
+ RemotePid = peer:call(Peer, erlang, spawn, [RFun]),
+ peer:send(Peer, RemotePid, do),
+ %% wait back from that process
+ receive done -> ok end,
+ %% shutdown the node
+ ?assertEqual(ok, peer:stop(Peer)),
+ ?assertNot(is_process_alive(Peer)).
+
+peer_states() ->
+ [{doc, "Tests peer node states"}].
+
+peer_states(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ {ok, Peer} = peer:start_link(#{connection => Conn, wait_boot => {self(), ?FUNCTION_NAME},
+ peer_down => continue}),
+ ?assertEqual(booting, peer:get_state(Peer)),
+ %% running
+ receive {?FUNCTION_NAME, {started, _Node, Peer}} -> ok end,
+ ?assertEqual(running, peer:get_state(Peer)),
+ peer:cast(Peer, erlang, halt, []),
+ ct:sleep(1000), %% source of flakiness, should switch to some better notification mechanism
+ %% down
+ ?assertMatch({down, _}, peer:get_state(Peer)),
+ peer:stop(Peer).
+
+cast() ->
+ [{doc, "Tests casts via alternative connections"}].
+
+cast(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ {ok, Peer, _Node} = peer:start_link(#{connection => Conn}),
+ %% cast RPC
+ ?assertEqual(undefined, peer:call(Peer, application, get_env, [kernel, foo])),
+ peer:cast(Peer, application, set_env, [kernel, foo, bar]),
+ %% this is only to ensure application_controller completed processing
+ peer:call(Peer, sys, get_state, [application_controller]),
+ ?assertEqual({ok, bar}, peer:call(Peer, application, get_env, [kernel, foo])),
+ peer:stop(Peer).
+
+detached() ->
+ [{doc, "Tests detached node (RPC via alternative connection)"}].
+
+detached(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ {ok, Peer, _Node} = peer:start_link(#{connection => Conn}),
+ ?assertEqual('nonode@nohost', peer:call(Peer, erlang, node, [])),
+ %% check exceptions
+ ?assertException(throw, ball, peer:call(Peer, erlang, throw, [ball])),
+ %% check tcp forwarding
+ Path = filename:dirname(code:which(?MODULE)),
+ ?assertEqual(true, peer:call(Peer, code, add_path, [Path])),
+ %% fancy RPC via message exchange (uses forwarding from the peer)
+ Control = self(),
+ RFun = fun() -> receive do -> forward(Control, done) end end,
+ RemotePid = peer:call(Peer, erlang, spawn, [RFun]),
+ %% test that sending message over alternative TCP connection works
+ peer:send(Peer, RemotePid, do),
+ %% wait back from that process
+ receive done -> ok end,
+ %% logging via TCP
+ ct:capture_start(),
+ peer:call(Peer, io, format, ["one."]),
+ peer:call(Peer, erlang, apply, [io, format, ["two."]]),
+ peer:stop(Peer),
+ ct:capture_stop(),
+ Texts = ct:capture_get(),
+ %% just stop
+ ?assertEqual(["one.", "two."], Texts).
+
+dyn_peer() ->
+ [{doc, "Origin is not distributed, and peer becomes distributed dynamically"}].
+
+dyn_peer(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ {ok, Peer, 'nonode@nohost'} = peer:start_link(#{connection => Conn}), %% start not distributed
+ Node = list_to_atom(lists:concat([peer:random_name(?FUNCTION_NAME), "@forced.host"])),
+ {ok, _} = peer:call(Peer, net_kernel, start, [[Node, longnames]]),
+ ?assertEqual(Node, peer:call(Peer, erlang, node, [])),
+ peer:stop(Peer).
+
+stop_peer() ->
+ [{doc, "Test that peer shuts down even when node sleeps, but control connection closed"}].
+
+stop_peer(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ {ok, Peer, _Node} = peer:start_link(#{name => peer:random_name(?FUNCTION_NAME),
+ connection => Conn, args => ["-eval", "timer:sleep(60000)."]}),
+ %% shutdown node
+ peer:stop(Peer).
+
+init_debug() ->
+ [{doc, "Test that debug messages in init work"}].
+
+init_debug(Config) when is_list(Config) ->
+ ct:capture_start(),
+ {ok, Peer, _Node} = peer:start_link(#{name => peer:random_name(?FUNCTION_NAME), shutdown => 1000,
+ connection => standard_io, args => ["-init_debug"]}),
+ ct:sleep(200), %% without this sleep, peer is not fast enough to print
+ peer:stop(Peer),
+ ct:capture_stop(),
+ Texts = lists:append([string:trim(Str, trailing, "\r\n") || Str <- ct:capture_get()]),
+ %% every boot script starts with this
+ Expected = "{progress,preloaded}",
+ Actual = lists:sublist(Texts, 1, length(Expected)),
+ ?assertEqual(Expected, Actual).
+
+io_redirect() ->
+ [{doc, "Tests i/o redirection working for std"}].
+
+io_redirect(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ {ok, Peer, _Node} = peer:start_link(#{connection => Conn}),
+ ct:capture_start(),
+ peer:call(Peer, io, format, [user, "test.", []]),
+ peer:call(Peer, erlang, apply, [io, format, ["second."]]),
+ %% ensure no dist connection is set up
+ ?assertNot(lists:member(Peer, nodes()), {dist_connected, Peer, nodes()}),
+ ct:capture_stop(),
+ Texts = ct:capture_get(),
+ peer:stop(Peer),
+ ?assertEqual(["test.", "second."], Texts).
+
+multi_node() ->
+ [{doc, "Tests several nodes starting concurrently"}].
+
+multi_node(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ Peers = [
+ peer:start_link(#{name => peer:random_name(?FUNCTION_NAME),
+ wait_boot => {self(), tag}, connection => Conn})
+ || _ <- lists:seq(1, 4)],
+ Nodes = [receive {tag, {started, Node, Peer}} -> Node end || {ok, Peer} <- Peers],
+ ?assertEqual(4, length(Nodes)),
+ [?assertEqual(ok, peer:stop(Peer)) || {ok, Peer} <- Peers].
+
+duplicate_name() ->
+ [{doc, "Tests that a node with the same name fails to start"}].
+
+duplicate_name(Config) when is_list(Config) ->
+ Conn = proplists:get_value(connection, Config),
+ {ok, Peer, _Node} = peer:start_link(#{connection => Conn, name => ?FUNCTION_NAME, register => false}),
+ ?assertException(exit, _, peer:start_link(#{connection => standard_io, name => ?FUNCTION_NAME})),
+ peer:stop(Peer).
+
+%% -------------------------------------------------------------------
+%% Compatibility: old releases
+old_release() ->
+ [{doc, "Verity running with previous OTP release"}].
+
+old_release(Config) when is_list(Config) ->
+ PrivDir = proplists:get_value(priv_dir, Config),
+ %% don't expect OTP 10 to be available
+ ?assertEqual(not_available, ?CT_PEER([], "10", PrivDir)),
+ PrevRel = integer_to_list(list_to_integer(erlang:system_info(otp_release)) - 2),
+ case ?CT_PEER([], PrevRel, PrivDir) of
+ not_available ->
+ {skip, "OTP " ++ PrevRel ++ " not found"};
+ {ok, Peer, Node} ->
+ ?assertEqual(PrevRel, rpc:call(Node, erlang, system_info, [otp_release])),
+ peer:stop(Peer)
+ end.
+
+%% -------------------------------------------------------------------
+%% SSH/Docker cases
+
+ssh() ->
+ [{doc, "Tests ssh (localhost) node support"}].
+
+ssh(Config) when is_list(Config) ->
+ %% figure out path to 'erl' locally ('erl' may not be in path via SSH)
+ case {os:find_executable("erl"), os:find_executable("ssh")} of
+ {false, _} ->
+ {skip, "erl not found"};
+ {ErlPath, SshPath} ->
+ Name = peer:random_name(?FUNCTION_NAME),
+ {ok, Peer, _Node} = peer:start_link(#{exec => {SshPath, ["localhost", ErlPath]},
+ connection => standard_io, name => Name,
+ args => ["-env", "BINDIR", os:getenv("BINDIR")], host => "localhost"}),
+ %% TODO: how to check it really goes over SSH?
+ %% ssh-ed node is not distributed
+ ?assertEqual(list_to_atom(Name ++ "@localhost"), peer:call(Peer, erlang, node, [])),
+ peer:stop(Peer)
+ end.
+
+docker() ->
+ [{doc, "Tests starting peer node in Docker container"}, {timetrap, {seconds, 60}}].
+
+build_release(Dir) ->
+ application:load(sasl), %% otherwise application:get_key will fail
+ %% build release (tarball)
+ RelFile = filename:join(Dir, "lambda.rel"),
+ Release = {release, {"lambda", "1.0.0"}, {erts, erlang:system_info(version)},
+ [{App, begin {ok, Vsn} = application:get_key(App, vsn), Vsn end} || App <- [kernel, stdlib, sasl]]},
+ ok = file:write_file(RelFile, list_to_binary(lists:flatten(io_lib:format("~tp.", [Release])))),
+ RelFileNoExt = filename:join(Dir, "lambda"),
+ {ok, systools_make, []} = systools:make_script(RelFileNoExt, [silent, {outdir, Dir}]),
+ ok = systools:make_tar(RelFileNoExt, [{erts, code:root_dir()}]).
+
+build_image(Dir) ->
+ %% build docker image
+ BuildScript = filename:join(Dir, "Dockerfile"),
+ Dockerfile =
+ "FROM ubuntu:20.04 as runner\n"
+ "WORKDIR /opt/lambda\n"
+ "COPY lambda.tar.gz /tmp\n"
+ "RUN tar -zxvf /tmp/lambda.tar.gz -C /opt/lambda\n"
+ "ENTRYPOINT [\"/opt/lambda/erts-" ++ erlang:system_info(version) ++ "/bin/dyn_erl\", \"-boot\", \"/opt/lambda/releases/1.0.0/start\"]\n",
+ ok = file:write_file(BuildScript, Dockerfile),
+ Output = os:cmd("docker build -t lambda " ++ Dir),
+ ct:pal("Build result: ~s~n", [Output]).
+
+docker(Config) when is_list(Config) ->
+ case os:find_executable("docker") of
+ false ->
+ {skip, "Docker not found"};
+ Docker ->
+ PrivDir = proplists:get_value(priv_dir, Config),
+ build_release(PrivDir),
+ build_image(PrivDir),
+
+ {ok, Peer, Node} = peer:start_link(#{name => ?CT_PEER_NAME(),
+ exec => {Docker, ["run", "-i", "lambda"]}, connection => standard_io}),
+ ?assertEqual(Node, peer:call(Peer, erlang, node, [])),
+ peer:stop(Peer)
+ end.
-- 
2.31.1

Places

File 1691-peer-peer-node-implementation.patch of Package erlang (Revision 57cf7c87cb80e2aaaf3f20327bf57529)

Places