Add erlang-based epmd server

.github/workflows/run-tests-with-beam.yaml

@@ -142,7 +142,7 @@ jobs:
       working-directory: build
       run: |
         export PATH="${{ matrix.path_prefix }}$PATH"
-        erl -pa tests/libs/estdlib/ -pa tests/libs/estdlib/beams/ -pa libs/etest/src/beams -s tests -s init stop -noshell
+        erl -pa tests/libs/estdlib/ -pa tests/libs/estdlib/beams/ -pa libs/etest/src/beams -pa libs/eavmlib/src/beams -s tests -s init stop -noshell
 
     # Test
     - name: "Run tests/libs/etest/test_eunit with OTP eunit"

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added `atomvm:subprocess/4` to perform pipe/fork/execve on POSIX platforms
 - Added `externalterm_to_term_with_roots` to efficiently preserve roots when allocating memory for external terms.
 - Added `erl_epmd` client implementation to epmd using `socket` module
+- Added support for socket asynchronous API for `recv`, `recvfrom` and `accept`.
 
 ### Changed
 

libs/eavmlib/src/CMakeLists.txt

@@ -28,6 +28,7 @@ set(ERLANG_MODULES
     avm_pubsub
     console
     emscripten
+    epmd
     esp
     esp_adc
     gpio

libs/eavmlib/src/epmd.erl

@@ -0,0 +1,197 @@
+%
+% This file is part of AtomVM.
+%
+% Copyright 2025 Paul Guyot <[email protected]>
+%
+% 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.
+%
+% SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later
+%
+
+-module(epmd).
+
+-behaviour(gen_server).
+
+-export([start_link/1]).
+
+% gen_server API
+-export([
+    init/1,
+    handle_call/3,
+    handle_cast/2,
+    handle_info/2,
+    terminate/2
+]).
+
+-define(NAMES_REQ, 110).
+-define(ALIVE2_X_RESP, 118).
+-define(PORT2_RESP, 119).
+-define(ALIVE2_REQ, 120).
+-define(ALIVE2_RESP, 121).
+-define(PORT_PLEASE2_REQ, 122).
+
+-define(EPMD_DEFAULT_PORT, 4369).
+-type epmd_config_option() :: {port, non_neg_integer()}.
+-type epmd_config() :: [epmd_config_option()].
+
+-spec start_link(Config :: epmd_config()) -> {ok, pid()} | {error, Reason :: term()}.
+start_link(Config) ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, Config, []).
+
+%%
+%% gen_server callbacks
+%%
+
+-record(state, {
+    socket :: any(),
+    port :: non_neg_integer(),
+    accept_handle :: undefined | reference(),
+    recv_selects :: [tuple()],
+    clients :: [{binary(), non_neg_integer(), reference(), binary()}],
+    creation :: non_neg_integer()
+}).
+
+%% @hidden
+init(Config) ->
+    Port = proplists:get_value(port, Config, ?EPMD_DEFAULT_PORT),
+    {ok, Socket} = socket:open(inet, stream, tcp),
+    ok = socket:setopt(Socket, {socket, reuseaddr}, true),
+    ok = socket:bind(Socket, #{
+        family => inet,
+        port => Port,
+        addr => {0, 0, 0, 0}
+    }),
+    ok = socket:listen(Socket),
+    RandCreation = 42,
+    State0 = #state{
+        socket = Socket, port = Port, recv_selects = [], clients = [], creation = RandCreation
+    },
+    State1 = accept(State0),
+    {ok, State1}.
+
+%% @hidden
+handle_call(_Msg, _From, State) ->
+    {noreply, State}.
+
+%% @hidden
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+%% @hidden
+handle_info(
+    {'$socket', _Socket, abort, {Ref, closed}},
+    #state{clients = Clients0, recv_selects = RecvSelects0} = State
+) ->
+    Clients1 = lists:keydelete(Ref, 3, Clients0),
+    RecvSelects1 = lists:keydelete(Ref, 1, RecvSelects0),
+    {noreply, State#state{clients = Clients1, recv_selects = RecvSelects1}};
+handle_info({'$socket', Socket, select, Ref}, #state{socket = Socket, accept_handle = Ref} = State) ->
+    NewState = accept(State),
+    {noreply, NewState};
+handle_info(
+    {'$socket', Socket, select, Ref},
+    #state{clients = Clients0, recv_selects = RecvSelects0} = State
+) ->
+    NewState =
+        case lists:keyfind(Ref, 1, RecvSelects0) of
+            {Ref, client} ->
+                socket:close(Socket),
+                Clients1 = lists:keydelete(Ref, 3, Clients0),
+                RecvSelects1 = lists:keydelete(Ref, 1, RecvSelects0),
+                State#state{clients = Clients1, recv_selects = RecvSelects1};
+            {Ref, req_size} ->
+                RecvSelects1 = lists:keydelete(Ref, 1, RecvSelects0),
+                client_socket_recv_req_size(Socket, State#state{recv_selects = RecvSelects1});
+            {Ref, req, Size} ->
+                RecvSelects1 = lists:keydelete(Ref, 1, RecvSelects0),
+                client_socket_recv_req(Socket, Size, State#state{recv_selects = RecvSelects1});
+            false ->
+                State
+        end,
+    {noreply, NewState}.
+
+%% @hidden
+terminate(_Reason, _State) ->
+    ok.
+
+accept(#state{socket = Socket} = State) ->
+    case socket:accept(Socket, nowait) of
+        {select, {select_info, accept, Ref}} ->
+            State#state{accept_handle = Ref};
+        {ok, ClientSocket} ->
+            State1 = client_socket_recv_req_size(ClientSocket, State),
+            accept(State1)
+    end.
+
+client_socket_recv_req_size(Socket, #state{recv_selects = RecvSelects} = State) ->
+    case socket:recv(Socket, 2, nowait) of
+        {select, {select_info, recv, Ref}} ->
+            State#state{recv_selects = [{Ref, req_size} | RecvSelects]};
+        {ok, <<Size:16>>} ->
+            client_socket_recv_req(Socket, Size, State)
+    end.
+
+client_socket_recv_req(Socket, Size, #state{recv_selects = RecvSelects} = State) ->
+    case socket:recv(Socket, Size, nowait) of
+        {select, {select_info, recv, Ref}} ->
+            State#state{recv_selects = [{Ref, req, Size} | RecvSelects]};
+        {ok, Data} ->
+            process_req(Data, Socket, State)
+    end.
+
+process_req(
+    <<?ALIVE2_REQ, Port:16, NodeType, Protocol, HighestVersion:16, LowestVersion:16, NameLen:16,
+        Name:NameLen/binary, ExtraLen:16, ExtraData:ExtraLen/binary>>,
+    Socket,
+    #state{clients = Clients, recv_selects = RecvSelects, creation = Creation} = State
+) ->
+    case socket:recv(Socket, 1, nowait) of
+        {select, {select_info, recv, Ref}} ->
+            socket:send(Socket, <<?ALIVE2_X_RESP, 0, Creation:32>>),
+            State#state{
+                recv_selects = [{Ref, client} | RecvSelects],
+                clients = [
+                    {Name, Port, Ref,
+                        <<Port:16, NodeType, Protocol, HighestVersion:16, LowestVersion:16,
+                            NameLen:16, Name:NameLen/binary, ExtraLen:16,
+                            ExtraData:ExtraLen/binary>>}
+                    | Clients
+                ],
+                creation = (Creation + 1) rem 16#ffffffff
+            };
+        {ok, <<_Byte>>} ->
+            socket:close(Socket),
+            State;
+        {error, closed} ->
+            State
+    end;
+process_req(<<?PORT_PLEASE2_REQ, Name/binary>>, Socket, #state{clients = Clients} = State) ->
+    case lists:keyfind(Name, 1, Clients) of
+        false ->
+            ok = socket:send(Socket, <<?PORT2_RESP, 1>>);
+        {Name, _Port, _Ref, Data} ->
+            ok = socket:send(Socket, <<?PORT2_RESP, 0, Data/binary>>)
+    end,
+    socket:close(Socket),
+    State;
+process_req(<<?NAMES_REQ>>, Socket, #state{clients = Clients, port = Port} = State) ->
+    ok = socket:send(Socket, <<Port:32>>),
+    lists:foreach(
+        fun({NodeName, NodePort, _Ref, _Data}) ->
+            Line = iolist_to_binary(io_lib:format("name ~ts at port ~p~n", [NodeName, NodePort])),
+            ok = socket:send(Socket, Line)
+        end,
+        Clients
+    ),
+    socket:close(Socket),
+    State.

libs/estdlib/src/gen_tcp_socket.erl

@@ -339,10 +339,7 @@ handle_cast(_Request, State) ->
     {noreply, State}.
 
 %% @hidden
-handle_info({select, _Socket, Ref, ready_input}, State) ->
-    ?LOG_DEBUG("handle_info [~p], ~p]", [
-        {select, _Socket, Ref, ready_input}, State
-    ]),
+handle_info({'$socket', _Socket, select, Ref}, State) ->
     %% TODO cancel timer
     case maps:get(Ref, State#state.pending_selects, undefined) of
         undefined ->
@@ -366,6 +363,28 @@ handle_info({select, _Socket, Ref, ready_input}, State) ->
                 pending_selects = maps:remove(Ref, State#state.pending_selects)
             }}
     end;
+handle_info({'$socket', Socket, abort, {Ref, closed}}, State) ->
+    %% TODO cancel timer
+    case maps:get(Ref, State#state.pending_selects, undefined) of
+        undefined ->
+            ?LOG_WARNING("Unable to find select ref ~p in pending selects", [Ref]),
+            socket:nif_select_stop(Socket),
+            {noreply, State};
+        {accept, From, _AcceptingProc, _Timeout} ->
+            socket:nif_select_stop(Socket),
+            gen_server:reply(From, {error, closed}),
+            {noreply, State};
+        active ->
+            WrappedSocket = {?GEN_TCP_MONIKER, self(), ?MODULE},
+            State#state.controlling_process ! {tcp_closed, WrappedSocket},
+            {noreply, State};
+        {passive, From, _Length, _Timeout} ->
+            socket:nif_select_stop(Socket),
+            gen_server:reply(From, {error, closed}),
+            {noreply, State#state{
+                pending_selects = maps:remove(Ref, State#state.pending_selects)
+            }}
+    end;
 handle_info({timeout, Ref, From}, State) ->
     ?LOG_DEBUG("handle_info [~p], ~p]", [
         {timeout, Ref, From}, State

libs/estdlib/src/gen_udp_socket.erl

@@ -242,7 +242,7 @@ handle_cast(_Request, State) ->
     {noreply, State}.
 
 %% @hidden
-handle_info({select, _Socket, Ref, ready_input}, State) ->
+handle_info({'$socket', _Socket, select, Ref}, State) ->
     case maps:get(Ref, State#state.pending_selects, undefined) of
         undefined ->
             ?LOG_INFO("Unable to find select ref ~p in pending selects", [Ref]),