socket: distinguish eagain from closed, and retry transparently via write-select

CHANGELOG.md

@@ -6,6 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [0.7.0-beta.1] - Unreleased
 
+### Added
+- Added `enif_select_write` and `ERL_NIF_SELECT_CANCEL` (both previously declared but not
+  implemented), and reworked `enif_select`/`enif_select_read`/`enif_select_write` so a resource
+  can hold independent pending read and write selects (each with its own ref/message/pid) on the
+  same event at the same time
+- Added `socket:nif_select_write/2`; sockets now keep one selecting process per direction, so a
+  process blocked in `socket:send/2` no longer disturbs another process blocked in `socket:recv/2,3`
+
 ### Fixed
 - Fixed maps decoded by `binary_to_term/1,2` and received over distribution keeping the
   encoder's key order, which broke `==` and made `=>` updates on an existing key produce
@@ -23,6 +31,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed the ESP32 UART driver blocking the scheduler, until the task watchdog reset the chip,
   when a read had already taken the bytes of a pending data event, or when `uart:read/2` counted
   bytes that the other scheduler then took
+- `socket:send/2,3` now correctly distinguishes transient send backpressure (lwIP `ERR_MEM` / BSD
+  `EAGAIN`|`EWOULDBLOCK`) from a closed connection at the NIF level, returning `{error, closed}`
+  (rather than `{ok, Data}`) when the peer has closed the connection, and waits for the socket to
+  become writable again and retries internally, so callers get `ok | {error, Reason}` (or
+  `{error, {Reason, RestData}}` when an error follows a partial send, as in OTP) and no longer
+  need to implement their own retry/backoff for partial sends or backpressure; `ssl:send/2`,
+  `ssl:recv/2` and the TLS handshake/close-notify loops likewise wait for write-readiness instead
+  of busy-looping on `want_write`
+- Fixed Erlang distribution over sockets (`socket_dist_controller`) silently ignoring
+  `socket:send/2` errors on `tick` and outgoing distribution data; the connection is now
+  terminated instead of continuing to run against a broken socket
 
 ## [0.7.0-beta.0] - 2026-09-28
 

examples/erlang/rp2/picow_tcp_server.erl

@@ -78,8 +78,6 @@ echo(ConnectedSocket) ->
             case socket:send(ConnectedSocket, Data) of
                 ok ->
                     io:format("All data was sent~n");
-                {ok, Rest} ->
-                    io:format("Some data was sent.  Remaining: ~p~n", [Rest]);
                 {error, Reason} ->
                     io:format("An error occurred sending data: ~p~n", [Reason])
             end,

examples/erlang/tcp_socket_server.erl

@@ -61,8 +61,6 @@ echo(ConnectedSocket) ->
             case socket:send(ConnectedSocket, Data) of
                 ok ->
                     io:format("All data was sent~n");
-                {ok, Rest} ->
-                    io:format("Some data was sent.  Remaining: ~p~n", [Rest]);
                 {error, Reason} ->
                     io:format("An error occurred sending data: ~p~n", [Reason])
             end,

examples/erlang/udp_socket_client.erl

@@ -37,10 +37,6 @@ loop(Socket) ->
             io:format("Send packet ~p to ~p.~n", [
                 Data, Dest
             ]);
-        {ok, Rest} ->
-            io:format("Send packet ~p to ~p.  Remaining: ~p~n", [
-                Data, Dest, Rest
-            ]);
         {error, Reason} ->
             io:format("An error occurred sending a packet: ~p~n", [Reason])
     end,

libs/avm_network/src/epmd.erl

@@ -157,7 +157,7 @@ process_req(
 ) ->
     case socket:recv(Socket, 1, nowait) of
         {select, {select_info, recv, Ref}} ->
-            socket:send(Socket, <<?ALIVE2_X_RESP, 0, Creation:32>>),
+            ok = socket:send(Socket, <<?ALIVE2_X_RESP, 0, Creation:32>>),
             State#state{
                 recv_selects = [{Ref, client} | RecvSelects],
                 clients = [

libs/estdlib/src/gen_tcp_socket.erl

@@ -263,7 +263,13 @@ handle_call({send, Packet}, From, State) ->
         {send, Packet}, From, State
     ]),
     ?LOG_INFO("Sending packet"),
-    {reply, socket:send(State#state.socket, Packet), State};
+    Reply =
+        case socket:send(State#state.socket, Packet) of
+            % gen_tcp:send/2 does not report the data that was not sent
+            {error, {Reason, _RestData}} -> {error, Reason};
+            Result -> Result
+        end,
+    {reply, Reply, State};
 handle_call({recv, Length, Timeout}, From, State) ->
     ?LOG_DEBUG("handle_call [~p], ~p, ~p]", [
         {recv, Length, Timeout}, From, State
@@ -379,18 +385,18 @@ handle_info({'$socket', Socket, abort, {Ref, closed}}, State) ->
     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),
+            socket:nif_select_read_stop(Socket),
             {noreply, State};
         {accept, From, _AcceptingProc, _Timeout} ->
-            socket:nif_select_stop(Socket),
+            socket:nif_select_read_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),
+            socket:nif_select_read_stop(Socket),
             gen_server:reply(From, {error, closed}),
             {noreply, State#state{
                 pending_selects = maps:remove(Ref, State#state.pending_selects)
@@ -465,7 +471,7 @@ handle_accept(State, From, AcceptingProc) ->
         {error, _Reason} = Error ->
             ?LOG_ERROR("Failed to accept connection: ~p", [Error]),
             %% CAUTION: internal API
-            socket:nif_select_stop(Socket),
+            socket:nif_select_read_stop(Socket),
             gen_server:reply(From, Error),
             State
     end.
@@ -510,15 +516,15 @@ handle_active_recv(State) ->
         {error, closed} ->
             ?LOG_INFO("Socket was closed by peer"),
             %% CAUTION: internal API
-            socket:nif_select_stop(Socket),
+            socket:nif_select_read_stop(Socket),
             %% peer has closed the connection
             WrappedSocket = {?GEN_TCP_MONIKER, self(), ?MODULE},
             State#state.controlling_process ! {tcp_closed, WrappedSocket},
             State;
         {error, _} = Error ->
             ?LOG_ERROR("Error on receive on pending select Error=~p", [Error]),
             %% CAUTION: internal API
-            socket:nif_select_stop(Socket),
+            socket:nif_select_read_stop(Socket),
             %% TODO is this right?  I don't think active receivers would know
             %% what to do with this message.  Maybe log it instead?
             State#state.controlling_process ! {tcp_error, WrappedSocket, Error},
@@ -547,7 +553,7 @@ handle_passive_recv(State, From, Length, _Timeout) ->
             State;
         {error, _} = Error ->
             ?LOG_ERROR("Error on receive from socket:nif_recv Error=~p", [Error]),
-            socket:nif_select_stop(Socket),
+            socket:nif_select_read_stop(Socket),
             ?LOG_INFO("unable to receive on pending select~n"),
             gen_server:reply(From, Error),
             State

libs/estdlib/src/gen_udp_socket.erl

@@ -335,7 +335,7 @@ handle_active_recvfrom(State) ->
             State;
         {error, _} = E ->
             %% CAUTION: internal API
-            socket:nif_select_stop(Socket),
+            socket:nif_select_read_stop(Socket),
             ?LOG_INFO("unable to receive on pending select~n"),
             State#state.controlling_process ! {udp_error, WrappedSocket, E},
             State
@@ -353,7 +353,7 @@ handle_passive_recvfrom(State, From, Length, _Timeout) ->
             gen_server:reply(From, {ok, {Addr, Port, BinaryOrList}}),
             State;
         {error, _} = E ->
-            socket:nif_select_stop(Socket),
+            socket:nif_select_read_stop(Socket),
             ?LOG_INFO("unable to receive on pending select~n"),
             gen_server:reply(From, E),
             State

libs/estdlib/src/socket.erl

@@ -46,10 +46,12 @@
 %% internal nifs
 -export([
     nif_select_read/2,
+    nif_select_write/2,
     nif_accept/1,
     nif_recv/2,
     nif_recvfrom/2,
     nif_select_stop/1,
+    nif_select_read_stop/1,
     nif_send/2,
     nif_sendto/3
 ]).
@@ -269,7 +271,7 @@ accept0(Socket, Timeout) ->
                 {'$socket', Socket, select, Ref} ->
                     case ?MODULE:nif_accept(Socket) of
                         {error, _} = E ->
-                            ?MODULE:nif_select_stop(Socket),
+                            ?MODULE:nif_select_read_stop(Socket),
                             E;
                         {ok, _Socket} = Reply ->
                             Reply
@@ -382,7 +384,7 @@ recv0(Socket, Length, Timeout) ->
                 {'$socket', Socket, select, Ref} ->
                     case ?MODULE:nif_recv(Socket, Length) of
                         {error, _} = E ->
-                            ?MODULE:nif_select_stop(Socket),
+                            ?MODULE:nif_select_read_stop(Socket),
                             E;
                         {ok, Data} ->
                             {ok, Data}
@@ -433,7 +435,7 @@ recv0_r(Socket, Length, Timeout, EndQuery, Acc) ->
                 {'$socket', Socket, select, Ref} ->
                     case ?MODULE:nif_recv(Socket, Length) of
                         {error, _} = E ->
-                            ?MODULE:nif_select_stop(Socket),
+                            ?MODULE:nif_select_read_stop(Socket),
                             E;
                         {ok, Data} ->
                             NewAcc = [Data | Acc],
@@ -537,7 +539,7 @@ recvfrom0(Socket, Length, Timeout) ->
                 {'$socket', Socket, select, Ref} ->
                     case ?MODULE:nif_recvfrom(Socket, Length) of
                         {error, _} = E ->
-                            ?MODULE:nif_select_stop(Socket),
+                            ?MODULE:nif_select_read_stop(Socket),
                             E;
                         {ok, {_Address, _Data}} = Reply ->
                             Reply
@@ -571,49 +573,61 @@ recvfrom0_nowait(Socket, Length, Ref) ->
 %%-----------------------------------------------------------------------------
 %% @param   Socket the socket
 %% @param   Data the data to send
-%% @returns `{ok, Rest}' if successful; `{error, Reason}', otherwise.
+%% @returns `ok' if successful; `{error, Reason}', otherwise, or
+%%          `{error, {Reason, RestData}}' if an error occurred after some of
+%%          the data was already sent, `RestData' being the data that was not
+%%          sent.
 %% @doc     Send data on the specified socket.
 %%
 %%          Note that this function will block until data is sent
 %%          on the socket.  The data may not have been received by the
 %%          intended recipient, and the data may not even have been sent
 %%          over the network.
 %%
+%%          Partial sends and transient send backpressure (e.g. a full TCP
+%%          send buffer) are retried internally, waiting for the socket to
+%%          become writable again, until all data has been accepted by the
+%%          stack or an error occurs. Callers therefore do not need to
+%%          implement their own retry logic for `eagain'-like conditions.
+%%
 %% Example:
 %%
 %%          `ok = socket:send(ConnectedSocket, Data)'
 %% @end
 %%-----------------------------------------------------------------------------
 -spec send(Socket :: socket(), Data :: iodata()) ->
-    ok | {ok, Rest :: binary()} | {error, Reason :: term()}.
+    ok | {error, Reason :: term()} | {error, {Reason :: term(), RestData :: binary()}}.
 send(Socket, Data) when is_binary(Data) ->
-    ?MODULE:nif_send(Socket, Data);
+    send_all(Socket, Data);
 send(Socket, Data) ->
-    ?MODULE:nif_send(Socket, erlang:iolist_to_binary(Data)).
+    send_all(Socket, erlang:iolist_to_binary(Data)).
 
 %%-----------------------------------------------------------------------------
 %% @param   Socket the socket
 %% @param   Data the data to send
 %% @param   Dest the destination to which to send the data
-%% @returns `{ok, Rest}' if successful; `{error, Reason}', otherwise.
+%% @returns `ok' if successful; `{error, Reason}', otherwise.
 %% @doc     Send data on the specified socket to the specified destination.
 %%
 %%          Note that this function will block until data is sent
 %%          on the socket.  The data may not have been received by the
 %%          intended recipient, and the data may not even have been sent
 %%          over the network.
 %%
+%%          Transient send backpressure is retried internally, waiting for
+%%          the socket to become writable again, as with `send/2'.
+%%
 %% Example:
 %%
 %%          `ok = socket:sendto(ConnectedSocket, Data, Dest)'
 %% @end
 %%-----------------------------------------------------------------------------
 -spec sendto(Socket :: socket(), Data :: iodata(), Dest :: sockaddr()) ->
-    ok | {ok, Rest :: binary()} | {error, Reason :: term()}.
+    ok | {error, Reason :: term()}.
 sendto(Socket, Data, Dest) when is_binary(Data) ->
-    ?MODULE:nif_sendto(Socket, Data, Dest);
+    sendto_all(Socket, Data, Dest);
 sendto(Socket, Data, Dest) ->
-    ?MODULE:nif_sendto(Socket, erlang:iolist_to_binary(Data), Dest).
+    sendto_all(Socket, erlang:iolist_to_binary(Data), Dest).
 
 %%-----------------------------------------------------------------------------
 %% @param   Socket the socket
@@ -708,6 +722,10 @@ shutdown(_Socket, _How) ->
 nif_select_read(_Socket, _Ref) ->
     erlang:nif_error(undefined).
 
+%% @private
+nif_select_write(_Socket, _Ref) ->
+    erlang:nif_error(undefined).
+
 %% @private
 nif_accept(_Socket) ->
     erlang:nif_error(undefined).
@@ -724,10 +742,83 @@ nif_recvfrom(_Socket, _Length) ->
 nif_select_stop(_Socket) ->
     erlang:nif_error(undefined).
 
+%% @private
+%% Like nif_select_stop/1, but if another process is selecting for write on the
+%% socket, only the read direction is stopped.
+nif_select_read_stop(_Socket) ->
+    erlang:nif_error(undefined).
+
 %% @private
 nif_send(_Socket, _Data) ->
     erlang:nif_error(undefined).
 
 %% @private
 nif_sendto(_Socket, _Data, _Dest) ->
     erlang:nif_error(undefined).
+
+%% @private
+%% Send all of Data on Socket, retrying partial sends and transient send
+%% backpressure ({error, eagain}) by waiting for the socket to become
+%% writable again, until all data has been accepted or a real error occurs.
+%% As in OTP, an error that follows a partial send is reported as
+%% {error, {Reason, RestData}}, and as {error, Reason} if nothing was accepted.
+send_all(Socket, Data) ->
+    send_all(Socket, Data, false).
+
+%% @private
+send_all(Socket, Data, Partial) ->
+    case ?MODULE:nif_send(Socket, Data) of
+        ok ->
+            ok;
+        {ok, Rest} ->
+            send_all(Socket, Rest, true);
+        {error, eagain} ->
+            case wait_write(Socket) of
+                ok ->
+                    send_all(Socket, Data, Partial);
+                {error, Reason} ->
+                    send_error(Reason, Data, Partial)
+            end;
+        {error, Reason} ->
+            send_error(Reason, Data, Partial)
+    end.
+
+%% @private
+send_error(Reason, RestData, true) ->
+    {error, {Reason, RestData}};
+send_error(Reason, _RestData, false) ->
+    {error, Reason}.
+
+%% @private
+sendto_all(Socket, Data, Dest) ->
+    case ?MODULE:nif_sendto(Socket, Data, Dest) of
+        ok ->
+            ok;
+        {ok, Rest} ->
+            sendto_all(Socket, Rest, Dest);
+        {error, eagain} ->
+            case wait_write(Socket) of
+                ok ->
+                    sendto_all(Socket, Data, Dest);
+                {error, _Reason} = Error ->
+                    Error
+            end;
+        {error, _Reason} = Error ->
+            Error
+    end.
+
+%% @private
+%% Block the calling process until Socket is writable (or closed).
+wait_write(Socket) ->
+    Ref = erlang:make_ref(),
+    case ?MODULE:nif_select_write(Socket, Ref) of
+        ok ->
+            receive
+                {'$socket', Socket, select, Ref} ->
+                    ok;
+                {'$socket', Socket, abort, {Ref, Reason}} ->
+                    {error, Reason}
+            end;
+        {error, _Reason} = Error ->
+            Error
+    end.