diff --git a/CHANGELOG.md b/CHANGELOG.md index c5d8d94de8..0a314c9569 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/examples/erlang/rp2/picow_tcp_server.erl b/examples/erlang/rp2/picow_tcp_server.erl index 053bf170c2..79ff6923a8 100644 --- a/examples/erlang/rp2/picow_tcp_server.erl +++ b/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, diff --git a/examples/erlang/tcp_socket_server.erl b/examples/erlang/tcp_socket_server.erl index 96fdb53a1e..029e9b6f9f 100644 --- a/examples/erlang/tcp_socket_server.erl +++ b/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, diff --git a/examples/erlang/udp_socket_client.erl b/examples/erlang/udp_socket_client.erl index 3389d31b11..a4b6df97ad 100644 --- a/examples/erlang/udp_socket_client.erl +++ b/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, diff --git a/libs/avm_network/src/epmd.erl b/libs/avm_network/src/epmd.erl index b012fe1eb7..fc9cc839cb 100644 --- a/libs/avm_network/src/epmd.erl +++ b/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, <>), + ok = socket:send(Socket, <>), State#state{ recv_selects = [{Ref, client} | RecvSelects], clients = [ diff --git a/libs/estdlib/src/gen_tcp_socket.erl b/libs/estdlib/src/gen_tcp_socket.erl index b17e5059d5..4ee3eb0e39 100644 --- a/libs/estdlib/src/gen_tcp_socket.erl +++ b/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,10 +385,10 @@ 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 -> @@ -390,7 +396,7 @@ handle_info({'$socket', Socket, abort, {Ref, closed}}, State) -> 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,7 +516,7 @@ 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}, @@ -518,7 +524,7 @@ handle_active_recv(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 diff --git a/libs/estdlib/src/gen_udp_socket.erl b/libs/estdlib/src/gen_udp_socket.erl index c3a561e53e..80883f8508 100644 --- a/libs/estdlib/src/gen_udp_socket.erl +++ b/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 diff --git a/libs/estdlib/src/socket.erl b/libs/estdlib/src/socket.erl index 810cf563c2..7709af80b7 100644 --- a/libs/estdlib/src/socket.erl +++ b/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,7 +573,10 @@ 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 @@ -579,23 +584,29 @@ recvfrom0_nowait(Socket, Length, Ref) -> %% 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 @@ -603,17 +614,20 @@ send(Socket, Data) -> %% 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,6 +742,12 @@ 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). @@ -731,3 +755,70 @@ nif_send(_Socket, _Data) -> %% @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. diff --git a/libs/estdlib/src/socket_dist_controller.erl b/libs/estdlib/src/socket_dist_controller.erl index d97f7eb59b..3e66782025 100644 --- a/libs/estdlib/src/socket_dist_controller.erl +++ b/libs/estdlib/src/socket_dist_controller.erl @@ -142,8 +142,12 @@ handle_call(getstat, _From, #state{received = Received, sent = Sent} = State) -> {reply, {ok, Received, Sent, 0}, State}. handle_cast(tick, #state{socket = Socket, sent = Sent} = State) -> - socket:send(Socket, <<0:32>>), - {noreply, State#state{sent = Sent + 1}}; + case socket:send(Socket, <<0:32>>) of + ok -> + {noreply, State#state{sent = Sent + 1}}; + {error, Reason} -> + {stop, {send_error, Reason}, State} + end; handle_cast({handshake_complete, DHandle}, State0) -> ok = erlang:dist_ctrl_get_data_notification(DHandle), % We now need to get messages when data is coming. @@ -154,8 +158,10 @@ handle_cast({handshake_complete, DHandle}, State0) -> end. handle_info(dist_data, State0) -> - State1 = send_data_loop(State0), - {noreply, State1}; + case send_data_loop(State0) of + {ok, State1} -> {noreply, State1}; + {stop, Reason, State1} -> {stop, Reason, State1} + end; handle_info( {'$socket', Socket, select, SelectHandle}, #state{socket = Socket, select_handle = SelectHandle} = State0 @@ -209,10 +215,14 @@ send_data_loop(#state{dhandle = DHandle, socket = Socket, sent = Sent} = State) case erlang:dist_ctrl_get_data(DHandle) of none -> ok = erlang:dist_ctrl_get_data_notification(DHandle), - State; + {ok, State}; Data -> DataBin = ?PRE_PROCESS(Data), DataSize = byte_size(DataBin), - socket:send(Socket, <>), - send_data_loop(State#state{sent = Sent + 1}) + case socket:send(Socket, <>) of + ok -> + send_data_loop(State#state{sent = Sent + 1}); + {error, Reason} -> + {stop, {send_error, Reason}, State} + end end. diff --git a/libs/estdlib/src/ssl.erl b/libs/estdlib/src/ssl.erl index 46385b0662..90ffc39506 100644 --- a/libs/estdlib/src/ssl.erl +++ b/libs/estdlib/src/ssl.erl @@ -200,8 +200,13 @@ handshake_loop(SSLContext, Socket) -> Error end; want_write -> - % We're currrently missing non-blocking writes - handshake_loop(SSLContext, Socket); + case wait_write(Socket) of + ok -> + handshake_loop(SSLContext, Socket); + {error, _Reason} = Error -> + socket:close(Socket), + Error + end; {error, _Reason} = Error -> socket:close(Socket), Error @@ -253,8 +258,13 @@ close_notify_loop(SSLContext, Socket) -> Error end; want_write -> - % We're currrently missing non-blocking writes - close_notify_loop(SSLContext, Socket); + case wait_write(Socket) of + ok -> + close_notify_loop(SSLContext, Socket); + {error, _Reason} = Error -> + socket:close(Socket), + Error + end; {error, _Reason} = Error -> socket:close(Socket), Error @@ -283,8 +293,12 @@ send({SSLContext, Socket} = SSLSocket, Binary) -> Error end; want_write -> - % We're currrently missing non-blocking writes - send(SSLSocket, Binary); + case wait_write(Socket) of + ok -> + send(SSLSocket, Binary); + {error, _Reason} = Error -> + Error + end; {error, _Reason} = Error -> Error end. @@ -318,8 +332,27 @@ recv0({SSLContext, Socket} = SSLSocket, Length, Remaining, Acc) -> Error end; want_write -> - % We're currrently missing non-blocking writes - recv0(SSLSocket, Length, Remaining, Acc); + case wait_write(Socket) of + ok -> + recv0(SSLSocket, Length, Remaining, Acc); + {error, _Reason} = Error -> + Error + end; + {error, _Reason} = Error -> + Error + end. + +%% @private +wait_write(Socket) -> + Ref = erlang:make_ref(), + case socket: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. diff --git a/src/libAtomVM/erl_nif.h b/src/libAtomVM/erl_nif.h index 73ab4e9e2b..90c90d8ce6 100644 --- a/src/libAtomVM/erl_nif.h +++ b/src/libAtomVM/erl_nif.h @@ -106,26 +106,26 @@ typedef enum /** * @brief enif_select mode flags - * @details ERL_NIF_SELECT_CANCEL which was introduced with OTP-22, is unimplemented. + * @details `ERL_NIF_SELECT_CANCEL` must be combined with `ERL_NIF_SELECT_READ` + * and/or `ERL_NIF_SELECT_WRITE`, and cannot be combined with `ERL_NIF_SELECT_STOP`. */ enum ErlNifSelectFlags { ERL_NIF_SELECT_READ = 1, ERL_NIF_SELECT_WRITE = 2, ERL_NIF_SELECT_STOP = 4, - // ERL_NIF_SELECT_CANCEL = 8, + ERL_NIF_SELECT_CANCEL = 8, }; /** * @brief enif_select result flags - * @details ERL_NIF_SELECT_CANCEL which was introduced with OTP-22, is unimplemented. */ enum { ERL_NIF_SELECT_STOP_CALLED = 1, ERL_NIF_SELECT_STOP_SCHEDULED = 2, - // ERL_NIF_SELECT_READ_CANCELLED = 4, - // ERL_NIF_SELECT_WRITE_CANCELLED = 8, + ERL_NIF_SELECT_READ_CANCELLED = 4, + ERL_NIF_SELECT_WRITE_CANCELLED = 8, ERL_NIF_SELECT_INVALID_EVENT = -1, ERL_NIF_SELECT_FAILED = -2, @@ -257,12 +257,15 @@ ERL_NIF_TERM enif_make_resource_binary(ErlNifEnv *env, void *obj, const void *da * @param env current environment * @param event event object (typically a file descriptor) * @param mode select mode (`ERL_NIF_SELECT_READ` and/or `ERL_NIF_SELECT_WRITE`) - * optionally with `ERL_NIF_SELECT_CANCEL` to cancel, or `ERL_NIF_SELECT_STOP` - * to stop. + * optionally with `ERL_NIF_SELECT_CANCEL` to cancel a pending select of the + * given direction(s) while keeping the event, or `ERL_NIF_SELECT_STOP` to stop. * @param obj resource object working as a container of the event object. * @param pid process id to send a message to or NULL to use the current process (from `env`) * @param ref reference object used in sent messages or `undefined` atom. - * @return a negative value on failure, 0 or flags on success. + * @return a negative value on failure, 0 or flags on success. With + * `ERL_NIF_SELECT_CANCEL`, `ERL_NIF_SELECT_READ_CANCELLED` and/or + * `ERL_NIF_SELECT_WRITE_CANCELLED` are set for each pending select that was + * cancelled. */ int enif_select(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSelectFlags mode, void *obj, const ErlNifPid *pid, ERL_NIF_TERM ref); diff --git a/src/libAtomVM/otp_socket.c b/src/libAtomVM/otp_socket.c index a561e0392d..4b2a3b1871 100644 --- a/src/libAtomVM/otp_socket.c +++ b/src/libAtomVM/otp_socket.c @@ -113,6 +113,7 @@ enum SocketState SocketStateSelectingRead = 1 << 3, SocketStateConnected = 1 << 4, SocketStateListening = 1 << 5, + SocketStateSelectingWrite = 1 << 6, // Actual states SocketStateUDPIdle = SocketStateUDP, @@ -120,6 +121,8 @@ enum SocketState SocketStateTCPNew = SocketStateTCP, SocketStateTCPConnected = SocketStateTCP | SocketStateConnected, SocketStateTCPSelectingRead = SocketStateTCPConnected | SocketStateSelectingRead, + SocketStateTCPSelectingWrite = SocketStateTCPConnected | SocketStateSelectingWrite, + SocketStateTCPSelectingReadWrite = SocketStateTCPConnected | SocketStateSelectingRead | SocketStateSelectingWrite, SocketStateTCPListening = SocketStateTCP | SocketStateListening, SocketStateTCPSelectingAccept = SocketStateTCPListening | SocketStateSelectingRead, }; @@ -146,19 +149,30 @@ struct UDPReceivedItem }; static err_t tcp_recv_cb(void *arg, struct tcp_pcb *tpcb, struct pbuf *p, err_t err); +static err_t tcp_sent_cb(void *arg, struct tcp_pcb *tpcb, u16_t len); +static err_t tcp_poll_cb(void *arg, struct tcp_pcb *tpcb); static void udp_recv_cb(void *arg, struct udp_pcb *pcb, struct pbuf *p, const ip_addr_t *addr, u16_t port); static void tcp_err_cb(void *arg, err_t err); #endif +// A process selecting a socket in one direction. Like in resources.c, a socket +// has at most one owner per direction: selecting again in the same direction +// replaces the previous owner, while the other direction is left untouched. +struct SocketSelect +{ + int32_t process_id; + ErlNifMonitor monitor; + uint64_t ref_ticks; +}; + #if OTP_SOCKET_BSD struct SocketResource { int fd; uint64_t socket_ref_ticks; - uint64_t select_ref_ticks; - int32_t selecting_process_id; - ErlNifMonitor selecting_process_monitor; + struct SocketSelect read_select; + struct SocketSelect write_select; size_t buf_size; #ifndef AVM_NO_SMP RWLock *socket_lock; @@ -174,9 +188,9 @@ struct SocketResource struct udp_pcb *udp_pcb; }; uint64_t socket_ref_ticks; - uint64_t select_ref_ticks; - int32_t selecting_process_id; // trapped or selecting - ErlNifMonitor selecting_process_monitor; + struct SocketSelect read_select; + // Also used to trap the process waiting for a connect to complete. + struct SocketSelect write_select; bool linger_on; int linger_sec; size_t pos; @@ -240,6 +254,12 @@ static const AtomStringIntPair otp_socket_setopt_level_table[] = { #define DEFAULT_BUFFER_SIZE 512 +#if OTP_SOCKET_LWIP +// Interval of the lwIP tcp_poll callback, in units of 500 ms (TCP_SLOW_INTERVAL). +// The callback is the fallback wakeup of processes selecting for write. +#define SOCKET_TCP_POLL_INTERVAL 1 +#endif + #ifndef MIN #define MIN(A, B) (((A) < (B)) ? (A) : (B)) #endif @@ -247,12 +267,35 @@ static const AtomStringIntPair otp_socket_setopt_level_table[] = { static ErlNifResourceType *socket_resource_type; #define SOCKET_MAKE_SELECT_NOTIFICATION_SIZE (TUPLE_SIZE(4) + TERM_BOXED_REFERENCE_SHORT_SIZE + TUPLE_SIZE(2) + TERM_BOXED_REFERENCE_SHORT_SIZE + TERM_BOXED_REFERENCE_RESOURCE_SIZE) -static term socket_make_select_notification(struct SocketResource *rsrc_obj, Heap *heap); +static term socket_make_select_notification(struct SocketResource *rsrc_obj, Heap *heap, enum ErlNifSelectFlags direction); // // resource operations // +#if OTP_SOCKET_LWIP +// Detach every callback (and the callback argument) from a TCP pcb that is +// about to be closed or aborted. lwIP keeps a closed pcb alive until the +// connection is fully shut down (e.g. in FIN_WAIT_1) and keeps calling its +// callbacks (poll, sent, err, ...) with the old argument, which would point to +// a socket resource that may already be gone. +// Must be called with the lwIP lock held. +static void socket_tcp_detach_callbacks(struct tcp_pcb *pcb, enum SocketState socket_state) +{ + tcp_arg(pcb, NULL); + if ((socket_state & SocketStateTCPListening) == SocketStateTCPListening) { + // lwIP only accepts the accept callback on listening pcbs and asserts + // when recv, sent, poll or err are set on them. + tcp_accept(pcb, NULL); + } else { + tcp_recv(pcb, NULL); + tcp_sent(pcb, NULL); + tcp_poll(pcb, NULL, 0); + tcp_err(pcb, NULL); + } +} +#endif + static void socket_dtor(ErlNifEnv *caller_env, void *obj) { UNUSED(caller_env); @@ -273,6 +316,7 @@ static void socket_dtor(ErlNifEnv *caller_env, void *obj) } else if (rsrc_obj->socket_state & SocketStateTCP) { // Try to nicely close the connection here // NOTE: we can't handle linger if the socket is gone. + socket_tcp_detach_callbacks(rsrc_obj->tcp_pcb, rsrc_obj->socket_state); if (UNLIKELY(tcp_close(rsrc_obj->tcp_pcb) != ERR_OK)) { // The resource will be gone, there is not much we can do here. if (!(rsrc_obj->socket_state & SocketStateTCPListening)) { @@ -290,22 +334,56 @@ static void socket_dtor(ErlNifEnv *caller_env, void *obj) #endif } +static void socket_select_init(struct SocketSelect *select) +{ + select->process_id = INVALID_PROCESS_ID; + select->monitor.ref_ticks = 0; + select->monitor.resource_type = NULL; + select->ref_ticks = 0; +} + +// Must be used for every socket resource, whatever way it was created, so that +// a new field cannot be left uninitialized on one of the creation paths. +static void socket_init_select_state(struct SocketResource *rsrc_obj) +{ + socket_select_init(&rsrc_obj->read_select); + socket_select_init(&rsrc_obj->write_select); +} + +static struct SocketSelect *socket_get_select(struct SocketResource *rsrc_obj, enum ErlNifSelectFlags direction) +{ + return (direction == ERL_NIF_SELECT_WRITE) ? &rsrc_obj->write_select : &rsrc_obj->read_select; +} + +// Forget the owner of a direction, demonitoring it. +// If the monitor already fired, socket_down will balance the refcount. +static void socket_select_disown(ErlNifEnv *env, struct SocketResource *rsrc_obj, struct SocketSelect *select) +{ + if (select->process_id == INVALID_PROCESS_ID) { + return; + } + // An owner can have no monitor (a process trapped waiting for a connect). + if (select->monitor.resource_type != NULL) { + // demonitor can fail if process is gone. + if (LIKELY(enif_demonitor_process(env, rsrc_obj, &select->monitor) == 0)) { + // decrement ref count as we are demonitoring + refc_binary_decrement_refcount(refc_binary_from_data(rsrc_obj), env->global); + } + select->monitor.resource_type = NULL; + } + select->process_id = INVALID_PROCESS_ID; +} + #if OTP_SOCKET_BSD static void socket_stop(ErlNifEnv *caller_env, void *obj, ErlNifEvent event, int is_direct_call) { - UNUSED(caller_env); UNUSED(event); UNUSED(is_direct_call); struct SocketResource *rsrc_obj = (struct SocketResource *) obj; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - if (LIKELY(enif_demonitor_process(caller_env, rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); - refc_binary_decrement_refcount(rsrc_refc, caller_env->global); - } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; - } + socket_select_disown(caller_env, rsrc_obj, &rsrc_obj->read_select); + socket_select_disown(caller_env, rsrc_obj, &rsrc_obj->write_select); TRACE("socket_stop called on fd=%i\n", rsrc_obj->fd); } @@ -313,9 +391,7 @@ static void socket_stop(ErlNifEnv *caller_env, void *obj, ErlNifEvent event, int static void socket_down(ErlNifEnv *caller_env, void *obj, ErlNifPid *pid, ErlNifMonitor *mon) { - UNUSED(caller_env); UNUSED(pid); - UNUSED(mon); struct SocketResource *rsrc_obj = (struct SocketResource *) obj; @@ -328,34 +404,39 @@ static void socket_down(ErlNifEnv *caller_env, void *obj, ErlNifPid *pid, ErlNif struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); SMP_RWLOCK_WRLOCK(rsrc_obj->socket_lock); - if (rsrc_obj->selecting_process_id == INVALID_PROCESS_ID) { + // Only the direction owned by the dead process is cleared. The monitor + // identifies it: if none matches, the owner was already replaced or + // removed and only the reference held for the monitor is left to drop. + enum ErlNifSelectFlags direction; + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID && rsrc_obj->read_select.monitor.ref_ticks == mon->ref_ticks) { + direction = ERL_NIF_SELECT_READ; + } else if (rsrc_obj->write_select.process_id != INVALID_PROCESS_ID && rsrc_obj->write_select.monitor.ref_ticks == mon->ref_ticks) { + direction = ERL_NIF_SELECT_WRITE; + } else { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + refc_binary_decrement_refcount(rsrc_refc, caller_env->global); return; } + // Monitor fired, so make sure we don't try to demonitor it later + // as it could crash trying to reacquire lock on process table. + struct SocketSelect *select = socket_get_select(rsrc_obj, direction); + select->process_id = INVALID_PROCESS_ID; + select->monitor.resource_type = NULL; + select->ref_ticks = 0; + #if OTP_SOCKET_BSD - // Monitor fired, so make sure we don't try to demonitor in select_stop - // as it could crash trying to reacquire lock on process table // enif_select can decrement ref count but it's at least 2 in this case (1 for monitor and 1 for select) - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; - enif_select(caller_env, rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()); + enif_select(caller_env, rsrc_obj->fd, direction | ERL_NIF_SELECT_CANCEL, rsrc_obj, NULL, term_nil()); + if (rsrc_obj->read_select.process_id == INVALID_PROCESS_ID && rsrc_obj->write_select.process_id == INVALID_PROCESS_ID) { + // No owner left on the other direction, release the select event. + enif_select(caller_env, rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()); + } #elif OTP_SOCKET_LWIP - // Monitor can be called when we're selecting, accepting or connecting. + // The connection is not aborted: like on BSD sockets, it is up to the + // other owner or to the resource destructor to close it. LWIP_BEGIN(); - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; - if (rsrc_obj->socket_state & SocketStateTCP) { - if (rsrc_obj->socket_state & SocketStateTCPListening) { - (void) tcp_close(rsrc_obj->tcp_pcb); - } else { - tcp_abort(rsrc_obj->tcp_pcb); - } - rsrc_obj->tcp_pcb = NULL; - rsrc_obj->socket_state = SocketStateClosed; - } else if (rsrc_obj->socket_state & SocketStateUDP) { - udp_remove(rsrc_obj->udp_pcb); - rsrc_obj->udp_pcb = NULL; - rsrc_obj->socket_state = SocketStateClosed; - } + rsrc_obj->socket_state &= (direction == ERL_NIF_SELECT_WRITE) ? ~SocketStateSelectingWrite : ~SocketStateSelectingRead; LWIP_END(); #endif @@ -377,7 +458,7 @@ static const ErlNifResourceTypeInit SocketResourceTypeInit = { }; // Make a notification message, using SOCKET_MAKE_SELECT_NOTIFICATION_SIZE on heap -static term socket_make_select_notification(struct SocketResource *rsrc_obj, Heap *heap) +static term socket_make_select_notification(struct SocketResource *rsrc_obj, Heap *heap, enum ErlNifSelectFlags direction) { term notification = term_alloc_tuple(4, heap); term_put_tuple_element(notification, 0, DOLLAR_SOCKET_ATOM); @@ -392,11 +473,12 @@ static term socket_make_select_notification(struct SocketResource *rsrc_obj, Hea term_put_tuple_element(socket_tuple, 1, socket_ref); term_put_tuple_element(notification, 1, socket_tuple); term_put_tuple_element(notification, 2, SELECT_ATOM); + uint64_t select_ref_ticks = socket_get_select(rsrc_obj, direction)->ref_ticks; term select_ref; - if (rsrc_obj->select_ref_ticks == 0) { + if (select_ref_ticks == 0) { select_ref = UNDEFINED_ATOM; } else { - select_ref = term_from_ref_ticks(rsrc_obj->select_ref_ticks, heap); + select_ref = term_from_ref_ticks(select_ref_ticks, heap); } term_put_tuple_element(notification, 3, select_ref); return notification; @@ -404,20 +486,20 @@ static term socket_make_select_notification(struct SocketResource *rsrc_obj, Hea // select emulation for lwIP that doesn't have select. #if OTP_SOCKET_LWIP -static void select_event_send_notification_from_nif(struct SocketResource *rsrc_obj, Context *locked_ctx) +static void select_event_send_notification_from_nif(struct SocketResource *rsrc_obj, Context *locked_ctx, enum ErlNifSelectFlags direction) { BEGIN_WITH_STACK_HEAP(SOCKET_MAKE_SELECT_NOTIFICATION_SIZE, heap) - term notification = socket_make_select_notification(rsrc_obj, &heap); + term notification = socket_make_select_notification(rsrc_obj, &heap, direction); mailbox_send(locked_ctx, notification); END_WITH_STACK_HEAP(heap, locked_ctx->global) } -static void select_event_send_notification_from_handler(struct SocketResource *rsrc_obj, int32_t process_id) +static void select_event_send_notification_from_handler(struct SocketResource *rsrc_obj, int32_t process_id, enum ErlNifSelectFlags direction) { struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); GlobalContext *global = rsrc_refc->resource_type->global; BEGIN_WITH_STACK_HEAP(SOCKET_MAKE_SELECT_NOTIFICATION_SIZE, heap) - term notification = socket_make_select_notification(rsrc_obj, &heap); + term notification = socket_make_select_notification(rsrc_obj, &heap, direction); globalcontext_send_message(global, process_id, notification); END_WITH_STACK_HEAP(heap, global) } @@ -587,7 +669,7 @@ static term nif_socket_open(Context *ctx, int argc, term argv[]) return make_errno_tuple(ctx); } else { TRACE("nif_socket_open: Created socket fd=%i\n", rsrc_obj->fd); - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + socket_init_select_state(rsrc_obj); if (type != SOCK_STREAM) { // TCP sockets are made non-blocking after connect, for now. @@ -618,7 +700,7 @@ static term nif_socket_open(Context *ctx, int argc, term argv[]) enif_release_resource(rsrc_obj); RAISE_ERROR(OUT_OF_MEMORY_ATOM); } else { - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + socket_init_select_state(rsrc_obj); rsrc_obj->linger_on = false; rsrc_obj->linger_sec = 0; rsrc_obj->pos = 0; @@ -628,6 +710,8 @@ static term nif_socket_open(Context *ctx, int argc, term argv[]) tcp_arg(rsrc_obj->tcp_pcb, rsrc_obj); tcp_err(rsrc_obj->tcp_pcb, tcp_err_cb); tcp_recv(rsrc_obj->tcp_pcb, tcp_recv_cb); + tcp_sent(rsrc_obj->tcp_pcb, tcp_sent_cb); + tcp_poll(rsrc_obj->tcp_pcb, tcp_poll_cb, SOCKET_TCP_POLL_INTERVAL); LWIP_END(); } else { LWIP_BEGIN(); @@ -692,7 +776,7 @@ bool term_is_otp_socket(term socket_term) // close // -static int send_closed_notification(Context *ctx, term socket_term, int32_t selecting_process_id, struct SocketResource *rsrc_obj) +static int send_closed_notification(Context *ctx, term socket_term, int32_t selecting_process_id, struct SocketResource *rsrc_obj, enum ErlNifSelectFlags direction) { // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid if (UNLIKELY(memory_ensure_free_with_roots(ctx, TUPLE_SIZE(4) + TUPLE_SIZE(2) + TERM_BOXED_REFERENCE_SHORT_SIZE, 1, &socket_term, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) { @@ -706,7 +790,8 @@ static int send_closed_notification(Context *ctx, term socket_term, int32_t sele term_put_tuple_element(socket_tuple, 2, ABORT_ATOM); term error_tuple = term_alloc_tuple(2, &ctx->heap); - term ref = (rsrc_obj->select_ref_ticks == 0) ? UNDEFINED_ATOM : term_from_ref_ticks(rsrc_obj->select_ref_ticks, &ctx->heap); + uint64_t select_ref_ticks = socket_get_select(rsrc_obj, direction)->ref_ticks; + term ref = (select_ref_ticks == 0) ? UNDEFINED_ATOM : term_from_ref_ticks(select_ref_ticks, &ctx->heap); term_put_tuple_element(error_tuple, 0, ref); term_put_tuple_element(error_tuple, 1, CLOSED_ATOM); term_put_tuple_element(socket_tuple, 3, error_tuple); @@ -748,29 +833,51 @@ static term nif_socket_close(Context *ctx, int argc, term argv[]) // process, as documented in specification of the abort message. // So we handle closing a socket while another process is selecting - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - // Save process id as socket_stop may be called by enif_select. - int32_t selecting_process_id = rsrc_obj->selecting_process_id; + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID || rsrc_obj->write_select.process_id != INVALID_PROCESS_ID) { + // Save the owners as socket_stop may be called by enif_select. + int32_t read_process_id = rsrc_obj->read_select.process_id; + int32_t write_process_id = rsrc_obj->write_select.process_id; // Another process is selecting, therefore ref_count >= 2 // 1. this caller's context heap (parameter to close) // 2. select + // Cancel each direction first: only a select that is still pending + // needs to be told that the socket is gone. A select that already + // fired has delivered its notification. + ErlNifEnv *env = erl_nif_env_from_context(ctx); + bool read_pending = false; + bool write_pending = false; + if (read_process_id != INVALID_PROCESS_ID) { + int cancel_res = enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, rsrc_obj, NULL, term_nil()); + read_pending = cancel_res > 0 && (cancel_res & ERL_NIF_SELECT_READ_CANCELLED); + } + if (write_process_id != INVALID_PROCESS_ID) { + int cancel_res = enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_WRITE | ERL_NIF_SELECT_CANCEL, rsrc_obj, NULL, term_nil()); + write_pending = cancel_res > 0 && (cancel_res & ERL_NIF_SELECT_WRITE_CANCELLED); + } + // Stop selecting. - int stop_res = enif_select(erl_nif_env_from_context(ctx), rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()); + int stop_res = enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()); if (UNLIKELY(stop_res < 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(BADARG_ATOM); } - // If there is a selecting process who may be waiting on select, - // send a closed notification to it, so that it can break - // out of its receive statement. + // If there are selecting processes who may be waiting on select, + // send a closed notification to each of them, so that they can + // break out of their receive statements. // // When using asynchronous API, the selecting process can be the // calling process. In this case we don't send any notification. // - if (selecting_process_id != ctx->process_id) { - // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid - if (UNLIKELY(send_closed_notification(ctx, argv[0], selecting_process_id, rsrc_obj) < 0)) { + // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid + if (read_pending && read_process_id != ctx->process_id) { + if (UNLIKELY(send_closed_notification(ctx, argv[0], read_process_id, rsrc_obj, ERL_NIF_SELECT_READ) < 0)) { + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + RAISE_ERROR(OUT_OF_MEMORY_ATOM); + } + } + if (write_pending && write_process_id != ctx->process_id) { + if (UNLIKELY(send_closed_notification(ctx, argv[0], write_process_id, rsrc_obj, ERL_NIF_SELECT_WRITE) < 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(OUT_OF_MEMORY_ATOM); } @@ -789,12 +896,21 @@ static term nif_socket_close(Context *ctx, int argc, term argv[]) TRACE("Double close on socket fd %i\n", rsrc_obj->fd); } #elif OTP_SOCKET_LWIP - // If the socket is being selected by another process, send a closed notification. - if (rsrc_obj->socket_state & SocketStateSelectingRead - && rsrc_obj->selecting_process_id != INVALID_PROCESS_ID - && rsrc_obj->selecting_process_id != ctx->process_id) { - // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid - if (UNLIKELY(send_closed_notification(ctx, argv[0], rsrc_obj->selecting_process_id, rsrc_obj) < 0)) { + // If the socket is being selected by other processes, send a closed + // notification to the owner of each direction being selected. + // send a {'$socket', Socket, abort, {Ref | undefined, closed}} message to the pid + if ((rsrc_obj->socket_state & SocketStateSelectingRead) + && rsrc_obj->read_select.process_id != INVALID_PROCESS_ID + && rsrc_obj->read_select.process_id != ctx->process_id) { + if (UNLIKELY(send_closed_notification(ctx, argv[0], rsrc_obj->read_select.process_id, rsrc_obj, ERL_NIF_SELECT_READ) < 0)) { + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + RAISE_ERROR(OUT_OF_MEMORY_ATOM); + } + } + if ((rsrc_obj->socket_state & SocketStateSelectingWrite) + && rsrc_obj->write_select.process_id != INVALID_PROCESS_ID + && rsrc_obj->write_select.process_id != ctx->process_id) { + if (UNLIKELY(send_closed_notification(ctx, argv[0], rsrc_obj->write_select.process_id, rsrc_obj, ERL_NIF_SELECT_WRITE) < 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(OUT_OF_MEMORY_ATOM); } @@ -804,7 +920,7 @@ static term nif_socket_close(Context *ctx, int argc, term argv[]) } else { if (rsrc_obj->socket_state & SocketStateTCP) { LWIP_BEGIN(); - tcp_arg(rsrc_obj->tcp_pcb, NULL); + socket_tcp_detach_callbacks(rsrc_obj->tcp_pcb, rsrc_obj->socket_state); err_t err = tcp_close(rsrc_obj->tcp_pcb); LWIP_END(); if (UNLIKELY(err != ERR_OK)) { @@ -848,7 +964,7 @@ static struct SocketResource *make_accepted_socket_resource(struct tcp_pcb *newp } conn_rsrc_obj->socket_state = SocketStateTCPConnected; conn_rsrc_obj->tcp_pcb = newpcb; - conn_rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + socket_init_select_state(conn_rsrc_obj); conn_rsrc_obj->pos = 0; conn_rsrc_obj->linger_on = false; conn_rsrc_obj->linger_sec = 0; @@ -864,6 +980,8 @@ static struct SocketResource *make_accepted_socket_resource(struct tcp_pcb *newp tcp_arg(newpcb, conn_rsrc_obj); tcp_recv(newpcb, tcp_recv_cb); + tcp_sent(newpcb, tcp_sent_cb); + tcp_poll(newpcb, tcp_poll_cb, SOCKET_TCP_POLL_INTERVAL); return conn_rsrc_obj; } @@ -878,8 +996,8 @@ static void tcp_accept_handler(struct LWIPEvent *event) if (rsrc_obj->socket_state & SocketStateSelectingRead) { // Clear flag to avoid sending a message again. rsrc_obj->socket_state &= ~SocketStateSelectingRead; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->selecting_process_id); + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID) { + select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->read_select.process_id, ERL_NIF_SELECT_READ); } // otherwise, selecting process died but we can just wait for monitor to handle it } } @@ -889,7 +1007,7 @@ static err_t tcp_accept_cb(void *arg, struct tcp_pcb *newpcb, err_t err) UNUSED(err); struct SocketResource *rsrc_obj = (struct SocketResource *) arg; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID) { if (newpcb != NULL) { // Because data can come in, we need to immediately set the receive callback // However, we cannot allocate the resource because we can't call malloc from ISR. @@ -941,8 +1059,8 @@ static void tcp_recv_handler(struct LWIPEvent *event) if (rsrc_obj->socket_state & SocketStateSelectingRead) { // Clear flag to avoid sending a message again. rsrc_obj->socket_state &= ~SocketStateSelectingRead; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->selecting_process_id); + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID) { + select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->read_select.process_id, ERL_NIF_SELECT_READ); } // otherwise, selecting process died but we can just wait for monitor to handle it } } @@ -963,6 +1081,73 @@ static err_t tcp_recv_cb(void *arg, struct tcp_pcb *tpcb, struct pbuf *p, err_t return ERR_OK; } +// Returns true if the lwIP TCP send buffer has room for more data, i.e. a +// subsequent tcp_write/tcp_output is likely to succeed rather than return +// ERR_MEM. +// Must be called with the lwIP lock held. +static bool lwip_tcp_can_send(const struct SocketResource *rsrc_obj) +{ + return rsrc_obj->tcp_pcb != NULL && tcp_sndbuf(rsrc_obj->tcp_pcb) > 0 && tcp_sndqueuelen(rsrc_obj->tcp_pcb) < TCP_SND_QUEUELEN; +} + +static void tcp_sent_handler(struct LWIPEvent *event) +{ + struct SocketResource *rsrc_obj = event->tcp_sent.rsrc_obj; + + // Send notification if we are selecting for write and the send buffer + // has room again. + if (!(rsrc_obj->socket_state & SocketStateSelectingWrite)) { + return; + } + LWIP_BEGIN(); + bool can_send = lwip_tcp_can_send(rsrc_obj); + LWIP_END(); + if (can_send) { + // Clear flag to avoid sending a message again. + rsrc_obj->socket_state &= ~SocketStateSelectingWrite; + if (rsrc_obj->write_select.process_id != INVALID_PROCESS_ID) { + select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->write_select.process_id, ERL_NIF_SELECT_WRITE); + } // otherwise, selecting process died but we can just wait for monitor to handle it + } +} + +// Called by lwIP when previously written data has been acknowledged by the +// peer, freeing up room in the send buffer. +static err_t tcp_sent_cb(void *arg, struct tcp_pcb *tpcb, u16_t len) +{ + UNUSED(tpcb); + UNUSED(len); + + // Only post an event when a process is waiting for write-readiness: the + // event queue is small and shared with the other socket events, and every + // post can wake the scheduler. + struct SocketResource *rsrc_obj = (struct SocketResource *) arg; + if (LIKELY(rsrc_obj) && (rsrc_obj->socket_state & SocketStateSelectingWrite)) { + struct LWIPEvent event; + event.handler = tcp_sent_handler; + event.tcp_sent.rsrc_obj = rsrc_obj; + otp_socket_lwip_enqueue(&event); + } + return ERR_OK; +} + +// Periodic poll callback, used as a fallback wakeup for write-readiness: some +// lwIP send failure paths (e.g. hitting TCP_SND_QUEUELEN without any bytes +// being freshly acknowledged) would otherwise never trigger tcp_sent_cb. +static err_t tcp_poll_cb(void *arg, struct tcp_pcb *tpcb) +{ + UNUSED(tpcb); + + struct SocketResource *rsrc_obj = (struct SocketResource *) arg; + if (LIKELY(rsrc_obj) && (rsrc_obj->socket_state & SocketStateSelectingWrite)) { + struct LWIPEvent event; + event.handler = tcp_sent_handler; + event.tcp_sent.rsrc_obj = rsrc_obj; + otp_socket_lwip_enqueue(&event); + } + return ERR_OK; +} + static void udp_recv_handler(struct LWIPEvent *event) { struct SocketResource *rsrc_obj = event->udp_recv.rsrc_obj; @@ -976,8 +1161,8 @@ static void udp_recv_handler(struct LWIPEvent *event) if (rsrc_obj->socket_state & SocketStateSelectingRead) { // Clear flag to avoid sending a message again. rsrc_obj->socket_state &= ~SocketStateSelectingRead; - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->selecting_process_id); + if (rsrc_obj->read_select.process_id != INVALID_PROCESS_ID) { + select_event_send_notification_from_handler(rsrc_obj, rsrc_obj->read_select.process_id, ERL_NIF_SELECT_READ); } // otherwise, selecting process died } } @@ -1021,9 +1206,9 @@ int tcpip_try_callback(tcpip_callback_fn function, void *ctx) #endif -static term nif_socket_select_read(Context *ctx, int argc, term argv[]) +static term nif_socket_select(Context *ctx, int argc, term argv[], enum ErlNifSelectFlags direction) { - TRACE("nif_socket_select_read\n"); + TRACE("nif_socket_select direction=%i\n", (int) direction); UNUSED(argc); @@ -1046,46 +1231,42 @@ static term nif_socket_select_read(Context *ctx, int argc, term argv[]) SMP_RWLOCK_WRLOCK(rsrc_obj->socket_lock); ErlNifEnv *env = erl_nif_env_from_context(ctx); - if (rsrc_obj->selecting_process_id != ctx->process_id && rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - // demonitor can fail if process is gone. - if (LIKELY(enif_demonitor_process(env, rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - // decrement ref count as we are demonitoring - refc_binary_decrement_refcount(rsrc_refc, ctx->global); - } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; - } - // Monitor first as select is less likely to fail and it's less expensive to demonitor - // if select fails than to stop select if monitor fails - if (rsrc_obj->selecting_process_id != ctx->process_id) { - if (UNLIKELY(enif_monitor_process(env, rsrc_obj, &ctx->process_id, &rsrc_obj->selecting_process_monitor) != 0)) { + // Only the requested direction is touched, the other one keeps its owner. + // Another process selecting the same direction is replaced, not queued. + struct SocketSelect *select = socket_get_select(rsrc_obj, direction); + if (select->process_id != ctx->process_id) { + socket_select_disown(env, rsrc_obj, select); + // Monitor first as select is less likely to fail and it's less expensive to demonitor + // if select fails than to stop select if monitor fails + if (UNLIKELY(enif_monitor_process(env, rsrc_obj, &ctx->process_id, &select->monitor) != 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(NOPROC_ATOM); } // increment ref count so the resource doesn't go away until monitor is fired refc_binary_increment_refcount(rsrc_refc); - rsrc_obj->selecting_process_id = ctx->process_id; + select->process_id = ctx->process_id; } - rsrc_obj->select_ref_ticks = (select_ref_term == UNDEFINED_ATOM) ? 0 : term_to_ref_ticks(select_ref_term); + select->ref_ticks = (select_ref_term == UNDEFINED_ATOM) ? 0 : term_to_ref_ticks(select_ref_term); #if OTP_SOCKET_BSD TRACE("rsrc_obj->fd=%i\n", (int) rsrc_obj->fd); // The socket may be closed here. if (rsrc_obj->fd == CLOSED_FD) { - send_closed_notification(ctx, argv[0], ctx->process_id, rsrc_obj); + send_closed_notification(ctx, argv[0], ctx->process_id, rsrc_obj, direction); } else { if (UNLIKELY(memory_ensure_free_with_roots(ctx, SOCKET_MAKE_SELECT_NOTIFICATION_SIZE, 2, argv, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) { AVM_LOGW(TAG, "Failed to allocate memory: %s:%i.", __FILE__, __LINE__); SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(OUT_OF_MEMORY_ATOM); } - term notification = socket_make_select_notification(rsrc_obj, &ctx->heap); - if (UNLIKELY(enif_select_read(erl_nif_env_from_context(ctx), rsrc_obj->fd, rsrc_obj, &ctx->process_id, notification, NULL) < 0)) { - if (LIKELY(enif_demonitor_process(env, rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - refc_binary_decrement_refcount(rsrc_refc, ctx->global); - } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + term notification = socket_make_select_notification(rsrc_obj, &ctx->heap, direction); + int select_result = (direction == ERL_NIF_SELECT_WRITE) + ? enif_select_write(env, rsrc_obj->fd, rsrc_obj, &ctx->process_id, notification, NULL) + : enif_select_read(env, rsrc_obj->fd, rsrc_obj, &ctx->process_id, notification, NULL); + if (UNLIKELY(select_result < 0)) { + socket_select_disown(env, rsrc_obj, select); SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(BADARG_ATOM); } @@ -1093,50 +1274,75 @@ static term nif_socket_select_read(Context *ctx, int argc, term argv[]) #elif OTP_SOCKET_LWIP LWIP_BEGIN(); - switch (rsrc_obj->socket_state) { - case SocketStateTCPListening: { - if (!list_is_empty(&rsrc_obj->received_list)) { - // Send (or resend) notification - select_event_send_notification_from_nif(rsrc_obj, ctx); - } else { - // Set flag to send it when a packet will arrive. - rsrc_obj->socket_state = SocketStateTCPSelectingAccept; - } - } break; - case SocketStateTCPSelectingAccept: - // noop - break; - case SocketStateTCPConnected: { - if (!list_is_empty(&rsrc_obj->received_list)) { + if (direction == ERL_NIF_SELECT_WRITE) { + if ((rsrc_obj->socket_state & SocketStateTCPConnected) == SocketStateTCPConnected) { + if (lwip_tcp_can_send(rsrc_obj)) { + // Clear the flag, in case a previous write select was still + // pending (replaced by this one), to avoid sending a second + // notification from tcp_sent_handler. + rsrc_obj->socket_state &= ~SocketStateSelectingWrite; // Send (or resend) notification - select_event_send_notification_from_nif(rsrc_obj, ctx); + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_WRITE); } else { - // Set flag to send it when a packet will arrive. - rsrc_obj->socket_state = SocketStateTCPSelectingRead; - } - } break; - case SocketStateTCPSelectingRead: - // noop - break; - case SocketStateUDPIdle: { - if (!list_is_empty(&rsrc_obj->received_list)) { - // Send (or resend) notification - select_event_send_notification_from_nif(rsrc_obj, ctx); - } else { - rsrc_obj->socket_state = SocketStateUDPSelectingRead; - } - } break; - case SocketStateUDPSelectingRead: - // noop - break; - default: - if (LIKELY(enif_demonitor_process(env, rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - refc_binary_decrement_refcount(rsrc_refc, ctx->global); + // Set flag to send it when the send buffer has room again. + rsrc_obj->socket_state |= SocketStateSelectingWrite; } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + } else if (rsrc_obj->socket_state & SocketStateUDP) { + // UDP datagrams are never buffered by lwIP in a way that blocks + // the caller, so writability is immediate. + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_WRITE); + } else { + socket_select_disown(env, rsrc_obj, select); LWIP_END(); SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(BADARG_ATOM); + } + } else { + switch (rsrc_obj->socket_state) { + case SocketStateTCPListening: { + if (!list_is_empty(&rsrc_obj->received_list)) { + // Send (or resend) notification + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_READ); + } else { + // Set flag to send it when a packet will arrive. + rsrc_obj->socket_state = SocketStateTCPSelectingAccept; + } + } break; + case SocketStateTCPSelectingAccept: + // noop + break; + case SocketStateTCPConnected: + case SocketStateTCPSelectingWrite: { + if (!list_is_empty(&rsrc_obj->received_list)) { + // Send (or resend) notification + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_READ); + } else { + // Set flag to send it when a packet will arrive, preserving + // a pending write select on the same socket. + rsrc_obj->socket_state |= SocketStateSelectingRead; + } + } break; + case SocketStateTCPSelectingRead: + case SocketStateTCPSelectingReadWrite: + // noop + break; + case SocketStateUDPIdle: { + if (!list_is_empty(&rsrc_obj->received_list)) { + // Send (or resend) notification + select_event_send_notification_from_nif(rsrc_obj, ctx, ERL_NIF_SELECT_READ); + } else { + rsrc_obj->socket_state = SocketStateUDPSelectingRead; + } + } break; + case SocketStateUDPSelectingRead: + // noop + break; + default: + socket_select_disown(env, rsrc_obj, select); + LWIP_END(); + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + RAISE_ERROR(BADARG_ATOM); + } } LWIP_END(); #endif @@ -1145,36 +1351,61 @@ static term nif_socket_select_read(Context *ctx, int argc, term argv[]) return OK_ATOM; } -static term nif_socket_select_stop(Context *ctx, int argc, term argv[]) +static term nif_socket_select_read(Context *ctx, int argc, term argv[]) { - TRACE("nif_socket_stop\n"); - UNUSED(argc); + return nif_socket_select(ctx, argc, argv, ERL_NIF_SELECT_READ); +} - VALIDATE_VALUE(argv[0], term_is_otp_socket); +static term nif_socket_select_write(Context *ctx, int argc, term argv[]) +{ + return nif_socket_select(ctx, argc, argv, ERL_NIF_SELECT_WRITE); +} + +// Stop selecting on a socket. When read_only is true and another process is +// selecting for write on the same socket, only the read direction is stopped, +// so that cleaning up after a failed read does not disown a blocked writer. +static term nif_socket_select_stop_internal(Context *ctx, term socket, bool read_only) +{ + VALIDATE_VALUE(socket, term_is_otp_socket); struct SocketResource *rsrc_obj; - if (UNLIKELY(!term_to_otp_socket(argv[0], &rsrc_obj, ctx))) { + if (UNLIKELY(!term_to_otp_socket(socket, &rsrc_obj, ctx))) { RAISE_ERROR(BADARG_ATOM); } + ErlNifEnv *env = erl_nif_env_from_context(ctx); // Avoid the race condition with select object here. SMP_RWLOCK_WRLOCK(rsrc_obj->socket_lock); - if (rsrc_obj->selecting_process_id != INVALID_PROCESS_ID) { - if (LIKELY(enif_demonitor_process(erl_nif_env_from_context(ctx), rsrc_obj, &rsrc_obj->selecting_process_monitor) == 0)) { - struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); - refc_binary_decrement_refcount(rsrc_refc, ctx->global); + if (read_only && rsrc_obj->write_select.process_id != INVALID_PROCESS_ID) { + // The write direction has an owner: stop the read direction only. + socket_select_disown(env, rsrc_obj, &rsrc_obj->read_select); + rsrc_obj->read_select.ref_ticks = 0; +#if OTP_SOCKET_BSD + if (UNLIKELY(enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, rsrc_obj, NULL, term_nil()) < 0)) { + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + RAISE_ERROR(BADARG_ATOM); } - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; +#elif OTP_SOCKET_LWIP + LWIP_BEGIN(); + rsrc_obj->socket_state &= ~SocketStateSelectingRead; + LWIP_END(); +#endif + SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); + return OK_ATOM; } + + // Stop both directions. + socket_select_disown(env, rsrc_obj, &rsrc_obj->read_select); + socket_select_disown(env, rsrc_obj, &rsrc_obj->write_select); + rsrc_obj->read_select.ref_ticks = 0; + rsrc_obj->write_select.ref_ticks = 0; #if OTP_SOCKET_BSD - if (UNLIKELY(enif_select(erl_nif_env_from_context(ctx), rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()) < 0)) { + if (UNLIKELY(enif_select(env, rsrc_obj->fd, ERL_NIF_SELECT_STOP, rsrc_obj, NULL, term_nil()) < 0)) { SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); RAISE_ERROR(BADARG_ATOM); } #elif OTP_SOCKET_LWIP LWIP_BEGIN(); - if (rsrc_obj->socket_state & SocketStateSelectingRead) { - rsrc_obj->socket_state &= ~SocketStateSelectingRead; - } + rsrc_obj->socket_state &= ~(SocketStateSelectingRead | SocketStateSelectingWrite); LWIP_END(); #endif SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); @@ -1182,6 +1413,22 @@ static term nif_socket_select_stop(Context *ctx, int argc, term argv[]) return OK_ATOM; } +static term nif_socket_select_stop(Context *ctx, int argc, term argv[]) +{ + TRACE("nif_socket_select_stop\n"); + UNUSED(argc); + + return nif_socket_select_stop_internal(ctx, argv[0], false); +} + +static term nif_socket_select_read_stop(Context *ctx, int argc, term argv[]) +{ + TRACE("nif_socket_select_read_stop\n"); + UNUSED(argc); + + return nif_socket_select_stop_internal(ctx, argv[0], true); +} + // // getopt // @@ -1895,7 +2142,7 @@ static term nif_socket_accept(Context *ctx, int argc, term argv[]) } struct SocketResource *conn_rsrc_obj = enif_alloc_resource(socket_resource_type, sizeof(struct SocketResource)); conn_rsrc_obj->fd = fd; - conn_rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + socket_init_select_state(conn_rsrc_obj); conn_rsrc_obj->buf_size = DEFAULT_BUFFER_SIZE; #ifndef AVM_NO_SMP conn_rsrc_obj->socket_lock = smp_rwlock_create(); @@ -2491,21 +2738,33 @@ static ssize_t do_socket_send(struct SocketResource *rsrc_obj, const uint8_t *bu pbuf_free(p); } else { // If the socket is connection-mode, dest_addr shall be ignored. - // Because we are copying, we cannot really send tcp_sndbuf(rsrc_obj->tcp_pcb) at once + // Because we are copying, we cannot really send tcp_sndbuf(rsrc_obj->tcp_pcb) at once. + // tcp_write fails with ERR_MEM when the data exceeds the send buffer, so + // cap the write at the available space, consistently with + // lwip_tcp_can_send: a partial write is returned as {ok, Rest}. size_t buflen = MIN(len, TCP_MSS); + if (len > 0) { + size_t sndbuf = tcp_sndbuf(rsrc_obj->tcp_pcb); + if (sndbuf == 0) { + LWIP_END(); + return SocketWouldBlock; + } + buflen = MIN(buflen, sndbuf); + } int flags = TCP_WRITE_FLAG_COPY; if (buflen < len) { flags |= TCP_WRITE_FLAG_MORE; } err = tcp_write(rsrc_obj->tcp_pcb, buf, buflen, flags); if (err == ERR_MEM) { - sent_data = 0; - } else { + // Send buffer or queue full: transient, the caller can retry. + LWIP_END(); + return SocketWouldBlock; + } + if (err == ERR_OK) { + err = tcp_output(rsrc_obj->tcp_pcb); if (err == ERR_OK) { - err = tcp_output(rsrc_obj->tcp_pcb); - if (err == ERR_OK) { - sent_data = buflen; - } + sent_data = buflen; } } } @@ -2513,9 +2772,6 @@ static ssize_t do_socket_send(struct SocketResource *rsrc_obj, const uint8_t *bu if (err == ERR_CLSD) { return SocketClosed; } - if (sent_data == 0) { - return SocketWouldBlock; - } if (err != ERR_OK) { return SocketOtherError; } @@ -2577,6 +2833,13 @@ static term nif_socket_send_internal(Context *ctx, int argc, term argv[], bool i // {ok, RestData} | {error, Reason} + // Transient send backpressure (lwIP ERR_MEM / BSD EAGAIN|EWOULDBLOCK) is + // reported as {error, eagain} so callers can retry rather than mistaking it + // for a closed connection. + if (sent_data == SocketWouldBlock) { + return make_error_tuple(posix_errno_to_term(EAGAIN, global), ctx); + } + size_t rest_len = len - sent_data; if (rest_len == 0) { return OK_ATOM; @@ -2592,12 +2855,10 @@ static term nif_socket_send_internal(Context *ctx, int argc, term argv[], bool i return port_create_tuple2(ctx, OK_ATOM, rest); } else if (sent_data == 0) { - if (UNLIKELY(memory_ensure_free_with_roots(ctx, TUPLE_SIZE(2), 1, &data, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) { - AVM_LOGW(TAG, "Failed to allocate memory: %s:%i.", __FILE__, __LINE__); - RAISE_ERROR(OUT_OF_MEMORY_ATOM); - } - - return port_create_tuple2(ctx, OK_ATOM, data); + // do_socket_send only returns 0 for a closed connection (SocketClosed: + // lwIP ERR_CLSD, BSD EBADF|ECONNRESET); an empty payload already + // returned ok above via the rest_len == 0 check. Report it as such. + return make_error_tuple(CLOSED_ATOM, ctx); } else { TRACE("Unable to send data: res=%zi.\n", sent_data); return make_error_tuple(CLOSED_ATOM, ctx); @@ -2654,8 +2915,9 @@ static err_t tcp_connected_cb(void *arg, struct tcp_pcb *tpcb, err_t err) struct SocketResource *rsrc_obj = (struct SocketResource *) arg; struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); GlobalContext *global = rsrc_refc->resource_type->global; - int32_t target_pid = rsrc_obj->selecting_process_id; - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + // The process waiting for the connect to complete is trapped on the write slot. + int32_t target_pid = rsrc_obj->write_select.process_id; + rsrc_obj->write_select.process_id = INVALID_PROCESS_ID; rsrc_obj->socket_state = SocketStateTCPConnected; if (target_pid != INVALID_PROCESS_ID) { if (err == ERR_OK) { @@ -2680,10 +2942,19 @@ static void tcp_err_cb(void *arg, err_t err) UNUSED(err); struct SocketResource *rsrc_obj = (struct SocketResource *) arg; + if (UNLIKELY(rsrc_obj == NULL)) { + // Socket was closed. + return; + } struct RefcBinary *rsrc_refc = refc_binary_from_data(rsrc_obj); GlobalContext *global = rsrc_refc->resource_type->global; - int32_t target_pid = rsrc_obj->selecting_process_id; - rsrc_obj->selecting_process_id = INVALID_PROCESS_ID; + // Only a process trapped waiting for a connect is answered: once connected + // the write slot belongs to a process selecting for write, which is not trapped. + int32_t target_pid = INVALID_PROCESS_ID; + if (rsrc_obj->socket_state == SocketStateTCPNew) { + target_pid = rsrc_obj->write_select.process_id; + rsrc_obj->write_select.process_id = INVALID_PROCESS_ID; + } rsrc_obj->socket_state = SocketStateTCPConnected; if (target_pid != INVALID_PROCESS_ID) { struct LWIPEvent event; @@ -2804,7 +3075,7 @@ static term nif_socket_connect(Context *ctx, int argc, term argv[]) SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); return OK_ATOM; } else { - rsrc_obj->selecting_process_id = ctx->process_id; + rsrc_obj->write_select.process_id = ctx->process_id; // Trap caller waiting for completion context_update_flags(ctx, ~NoFlags, Trap); SMP_RWLOCK_UNLOCK(rsrc_obj->socket_lock); @@ -2877,6 +3148,13 @@ static term nif_socket_shutdown(Context *ctx, int argc, term argv[]) if (rsrc_obj->socket_state & SocketStateTCP) { err_t res = tcp_shutdown(rsrc_obj->tcp_pcb, how != SHUT_WR, how != SHUT_RD); if (how == SHUT_RDWR && res == ERR_OK) { + // tcp_shutdown(1, 1) closes the pcb, but lwIP keeps it (e.g. in + // FIN_WAIT) and its callbacks running with our argument. We will + // lose the pointer to the pcb, so detach them now. This is done + // after tcp_shutdown succeeded so that a failed shutdown leaves the + // socket intact, and under the lwIP lock so no callback can run in + // between. + socket_tcp_detach_callbacks(rsrc_obj->tcp_pcb, rsrc_obj->socket_state); rsrc_obj->tcp_pcb = NULL; } if (res != ERR_OK) { @@ -2906,6 +3184,10 @@ static const struct Nif socket_select_stop_nif = { .base.type = NIFFunctionType, .nif_ptr = nif_socket_select_stop }; +static const struct Nif socket_select_read_stop_nif = { + .base.type = NIFFunctionType, + .nif_ptr = nif_socket_select_read_stop +}; static const struct Nif socket_getopt_nif = { .base.type = NIFFunctionType, .nif_ptr = nif_socket_getopt @@ -2934,6 +3216,10 @@ static const struct Nif socket_select_read_nif = { .base.type = NIFFunctionType, .nif_ptr = nif_socket_select_read }; +static const struct Nif socket_select_write_nif = { + .base.type = NIFFunctionType, + .nif_ptr = nif_socket_select_write +}; static const struct Nif socket_accept_nif = { .base.type = NIFFunctionType, .nif_ptr = nif_socket_accept @@ -2980,6 +3266,10 @@ const struct Nif *otp_socket_nif_get_nif(const char *nifname) TRACE("Resolved platform nif %s ...\n", nifname); return &socket_select_stop_nif; } + if (strcmp("nif_select_read_stop/1", rest) == 0) { + TRACE("Resolved platform nif %s ...\n", nifname); + return &socket_select_read_stop_nif; + } if (strcmp("getopt/2", rest) == 0) { TRACE("Resolved platform nif %s ...\n", nifname); return &socket_getopt_nif; @@ -3008,6 +3298,10 @@ const struct Nif *otp_socket_nif_get_nif(const char *nifname) TRACE("Resolved platform nif %s ...\n", nifname); return &socket_select_read_nif; } + if (strcmp("nif_select_write/2", rest) == 0) { + TRACE("Resolved platform nif %s ...\n", nifname); + return &socket_select_write_nif; + } if (strcmp("nif_accept/1", rest) == 0) { TRACE("Resolved platform nif %s ...\n", nifname); return &socket_accept_nif; diff --git a/src/libAtomVM/otp_socket.h b/src/libAtomVM/otp_socket.h index cfd9626531..ca8cd43f02 100644 --- a/src/libAtomVM/otp_socket.h +++ b/src/libAtomVM/otp_socket.h @@ -111,6 +111,10 @@ struct LWIPEvent err_t err; } tcp_recv; struct + { + struct SocketResource *rsrc_obj; + } tcp_sent; + struct { struct SocketResource *rsrc_obj; struct pbuf *buf; diff --git a/src/libAtomVM/resources.c b/src/libAtomVM/resources.c index 87ca0b5eaa..cf2e90d824 100644 --- a/src/libAtomVM/resources.c +++ b/src/libAtomVM/resources.c @@ -169,16 +169,17 @@ static int enif_select_common(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSele resource->resource_type->stop(env, obj, event, true); } refc_binary_decrement_refcount(resource, global); - enif_select_event_message_dispose(select_event->message, global, false); + enif_select_event_message_dispose(select_event->read_message, global, false); + enif_select_event_message_dispose(select_event->write_message, global, false); free((void *) select_event); return ERL_NIF_SELECT_STOP_CALLED; } // We cannot call stop now because scheduler loop unlocks after building // the select set but before calling select (or equivalent) // So instead we flag the event. - select_event->close = 1; - select_event->read = 0; - select_event->write = 0; + select_event->close = true; + select_event->read = false; + select_event->write = false; synclist_unlock(&global->select_events); // Platform loop should check close flag after unregister is called if (was_read) { @@ -189,6 +190,42 @@ static int enif_select_common(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSele } return ERL_NIF_SELECT_STOP_SCHEDULED; } + if (mode & ERL_NIF_SELECT_CANCEL) { + // Cancel only the requested directions that are pending. The event + // (and the resource refcount it holds) is kept, so the other direction + // stays armed and the caller can still select again or stop later. + int result = 0; + bool cancel_read = false; + bool cancel_write = false; + if (select_event != NULL) { + if ((mode & ERL_NIF_SELECT_READ) && select_event->read) { + select_event->read = false; + enif_select_event_message_dispose(select_event->read_message, global, false); + select_event->read_message = NULL; + select_event->read_ref_ticks = 0; + select_event->read_local_pid = INVALID_PROCESS_ID; + cancel_read = true; + result |= ERL_NIF_SELECT_READ_CANCELLED; + } + if ((mode & ERL_NIF_SELECT_WRITE) && select_event->write) { + select_event->write = false; + enif_select_event_message_dispose(select_event->write_message, global, false); + select_event->write_message = NULL; + select_event->write_ref_ticks = 0; + select_event->write_local_pid = INVALID_PROCESS_ID; + cancel_write = true; + result |= ERL_NIF_SELECT_WRITE_CANCELLED; + } + } + synclist_unlock(&global->select_events); + if (cancel_read) { + sys_unregister_select_event(global, event, false); + } + if (cancel_write) { + sys_unregister_select_event(global, event, true); + } + return result; + } // Create new event if it doesn't exist. if (select_event == NULL) { select_event = malloc(sizeof(struct SelectEvent)); @@ -197,34 +234,45 @@ static int enif_select_common(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSele } select_event->event = event; select_event->resource = resource; - select_event->message = NULL; - select_event->ref_ticks = 0; + select_event->read = false; + select_event->write = false; + select_event->close = false; + select_event->read_message = NULL; + select_event->write_message = NULL; + select_event->read_ref_ticks = 0; + select_event->write_ref_ticks = 0; + select_event->read_local_pid = INVALID_PROCESS_ID; + select_event->write_local_pid = INVALID_PROCESS_ID; // Resource is used in select_event, so we increase refcount. refc_binary_increment_refcount(resource); list_init(&select_event->head); list_append(select_events, &select_event->head); } - // Second read or second write overwrite ref/message & pid. - enif_select_event_message_dispose(select_event->message, global, false); - select_event->message = message; - if (message) { - select_event->ref_ticks = 0; - } else { - if (ref == UNDEFINED_ATOM) { - select_event->ref_ticks = 0; - } else { - select_event->ref_ticks = term_to_ref_ticks(ref); - } - } - select_event->local_pid = *pid; - select_event->read = mode & ERL_NIF_SELECT_READ; - select_event->write = mode & ERL_NIF_SELECT_WRITE; - select_event->close = 0; + // Second read or second write overwrites only that direction, preserving + // the other direction when both read and write are selected on the same + // event (e.g. a socket with a pending recv select and a pending send + // select at the same time). + uint64_t ref_ticks = (!message && term_is_local_reference(ref)) ? term_to_ref_ticks(ref) : 0; + if (mode & ERL_NIF_SELECT_READ) { + enif_select_event_message_dispose(select_event->read_message, global, false); + select_event->read_message = message; + select_event->read_ref_ticks = message ? 0 : ref_ticks; + select_event->read_local_pid = *pid; + select_event->read = true; + } + if (mode & ERL_NIF_SELECT_WRITE) { + enif_select_event_message_dispose(select_event->write_message, global, false); + select_event->write_message = message; + select_event->write_ref_ticks = message ? 0 : ref_ticks; + select_event->write_local_pid = *pid; + select_event->write = true; + } + select_event->close = false; synclist_unlock(&global->select_events); - if (select_event->read) { + if (mode & ERL_NIF_SELECT_READ) { sys_register_select_event(global, event, false); } - if (select_event->write) { + if (mode & ERL_NIF_SELECT_WRITE) { sys_register_select_event(global, event, true); } return 0; @@ -235,6 +283,13 @@ int enif_select(ErlNifEnv *env, ErlNifEvent event, enum ErlNifSelectFlags mode, if (!(mode & (ERL_NIF_SELECT_STOP | ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE))) { return ERL_NIF_SELECT_BADARG; } + if (mode & ERL_NIF_SELECT_CANCEL) { + // Cancel applies to read and/or write, and cannot be combined with stop. + if ((mode & ERL_NIF_SELECT_STOP) || !(mode & (ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE))) { + return ERL_NIF_SELECT_BADARG; + } + return enif_select_common(env, event, mode, obj, pid, ref, NULL); + } if (UNLIKELY(mode & (ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE) && !term_is_local_reference(ref) && ref != UNDEFINED_ATOM)) { return ERL_NIF_SELECT_BADARG; } @@ -252,7 +307,17 @@ int enif_select_read(ErlNifEnv *env, ErlNifEvent event, void *obj, const ErlNifP } Message *message = mailbox_message_create_normal_message_from_term(msg); enum ErlNifSelectFlags mode = ERL_NIF_SELECT_READ; - return enif_select_common(env, event, mode, obj, pid, term_nil(), message); + return enif_select_common(env, event, mode, obj, pid, UNDEFINED_ATOM, message); +} + +int enif_select_write(ErlNifEnv *env, ErlNifEvent event, void *obj, const ErlNifPid *pid, ERL_NIF_TERM msg, ErlNifEnv *msg_env) +{ + if (UNLIKELY(msg_env != NULL)) { + return ERL_NIF_SELECT_BADARG; + } + Message *message = mailbox_message_create_normal_message_from_term(msg); + enum ErlNifSelectFlags mode = ERL_NIF_SELECT_WRITE; + return enif_select_common(env, event, mode, obj, pid, UNDEFINED_ATOM, message); } term select_event_make_notification(void *rsrc_obj, uint64_t ref_ticks, bool is_write, Heap *heap) @@ -273,32 +338,36 @@ term select_event_make_notification(void *rsrc_obj, uint64_t ref_ticks, bool is_ static void select_event_send_notification(struct SelectEvent *select_event, bool is_write, GlobalContext *global) { - if (select_event->message) { + Message **message = is_write ? &select_event->write_message : &select_event->read_message; + int32_t local_pid = is_write ? select_event->write_local_pid : select_event->read_local_pid; + uint64_t ref_ticks = is_write ? select_event->write_ref_ticks : select_event->read_ref_ticks; + + if (*message) { enum SendMessageResult result; #ifdef AVM_SELECT_IN_TASK - result = globalcontext_post_message_from_task(global, select_event->local_pid, select_event->message); + result = globalcontext_post_message_from_task(global, local_pid, *message); #else - result = globalcontext_post_message(global, select_event->local_pid, select_event->message); + result = globalcontext_post_message(global, local_pid, *message); #endif if (result == SEND_MESSAGE_OK) { // Ownership was properly transfered. // Otherwise, it will be destroyed when we have a context (when enif_select is called with stop for example) - select_event->message = NULL; + *message = NULL; } } else { BEGIN_WITH_STACK_HEAP(SELECT_EVENT_NOTIFICATION_SIZE, heap) - term notification = select_event_make_notification(select_event->resource->data, select_event->ref_ticks, is_write, &heap); + term notification = select_event_make_notification(select_event->resource->data, ref_ticks, is_write, &heap); #ifdef AVM_SELECT_IN_TASK - globalcontext_send_message_from_task(global, select_event->local_pid, NormalMessage, notification); + globalcontext_send_message_from_task(global, local_pid, NormalMessage, notification); #else - globalcontext_send_message(global, select_event->local_pid, notification); + globalcontext_send_message(global, local_pid, notification); #endif END_WITH_STACK_HEAP(heap, global) } if (is_write) { - select_event->write = 0; + select_event->write = false; } else { - select_event->read = 0; + select_event->read = false; } sys_unregister_select_event(global, select_event->event, is_write); } @@ -342,7 +411,8 @@ static inline void select_event_destroy(struct SelectEvent *select_event, Global #else refc_binary_decrement_refcount(select_event->resource, global); #endif - enif_select_event_message_dispose(select_event->message, global, true); + enif_select_event_message_dispose(select_event->read_message, global, true); + enif_select_event_message_dispose(select_event->write_message, global, true); free((void *) select_event); } diff --git a/src/libAtomVM/resources.h b/src/libAtomVM/resources.h index 58c97cf9b6..654eaa8acc 100644 --- a/src/libAtomVM/resources.h +++ b/src/libAtomVM/resources.h @@ -92,9 +92,12 @@ struct SelectEvent bool read; bool write; bool close; - int32_t local_pid; - uint64_t ref_ticks; - Message *message; + int32_t read_local_pid; + int32_t write_local_pid; + uint64_t read_ref_ticks; + uint64_t write_ref_ticks; + Message *read_message; + Message *write_message; }; /** diff --git a/tests/libs/estdlib/test_tcp_socket.erl b/tests/libs/estdlib/test_tcp_socket.erl index 4ceed655b9..2897214f17 100644 --- a/tests/libs/estdlib/test_tcp_socket.erl +++ b/tests/libs/estdlib/test_tcp_socket.erl @@ -31,15 +31,23 @@ test() -> ok = test_recv_nowait(), ok = test_accept_nowait(), ok = test_setopt_getopt(), + ok = test_close_accepted_socket_while_selecting(), + ok = test_send_empty(), case erlang:system_info(machine) of "ATOM" -> - ok = test_abandon_select(); + ok = test_abandon_select(), + ok = test_send_backpressure(), + ok = test_close_with_reader_and_writer(), + ok = test_writer_killed_reader_survives(), + ok = test_reader_cleanup_keeps_writer(), + ok = test_send_error_after_partial_send(); "BEAM" -> ok end, ok. -define(PACKET_SIZE, 7). +-define(LARGE_PAYLOAD_SIZE, (64 * 1024 * 1024)). test_echo_server() -> etest:flush_msg_queue(), @@ -570,5 +578,415 @@ test_abandon_select() -> erlang:garbage_collect(), ok. +%% +%% test_send_backpressure +%% +%% Exercises the write-select mechanism used internally by socket:send/2 to +%% wait for transient send backpressure (a full TCP send buffer) to clear, +%% instead of leaking {error, eagain} or {ok, Rest} to the caller. +%% + +test_send_backpressure() -> + etest:flush_msg_queue(), + + {ok, ListenSocket} = socket:open(inet, stream, tcp), + ok = socket:setopt(ListenSocket, {socket, reuseaddr}, true), + ok = socket:setopt(ListenSocket, {socket, linger}, #{onoff => true, linger => 0}), + ok = socket:bind(ListenSocket, #{family => inet, addr => loopback, port => 0}), + ok = socket:listen(ListenSocket), + {ok, #{port := Port}} = socket:sockname(ListenSocket), + + Self = self(), + Acceptor = spawn_link(fun() -> backpressure_acceptor(Self, ListenSocket) end), + + {ok, ClientSocket} = socket:open(inet, stream, tcp), + ok = try_connect(ClientSocket, Port, 10), + + ok = + receive + {server_socket, _ServerSocket} -> ok + after 5000 -> + error({timeout, waiting_for_server_socket}) + end, + + %% Fill the client's send buffer (the server never reads) using the raw + %% nif_send/2 directly, bypassing socket:send/2's automatic retry, until + %% we either observe a real {error, eagain} or give up after a generous + %% number of attempts. Different platforms/kernels size their socket + %% buffers differently, so we tolerate never observing backpressure + %% rather than failing the test outright. + Chunk = binary:copy(<<0>>, 65536), + {TotalSent, EAgainObserved} = fill_send_buffer(ClientSocket, Chunk, 0, 256), + + ok = + case EAgainObserved andalso TotalSent > 0 of + true -> + %% socket:nif_select_write/2 should let us wait until the + %% socket becomes writable again. Depending on how the + %% platform/kernel sizes and accounts for socket buffers, a + %% small partial drain on the peer may not be enough to + %% cross the low-water mark for writability, so we have the + %% acceptor drain everything (as a real reader normally + %% would) to reliably free up space. + Ref = erlang:make_ref(), + ok = socket:nif_select_write(ClientSocket, Ref), + + Acceptor ! {drain_all, self()}, + ok = + receive + {'$socket', ClientSocket, select, Ref} -> + ok + after 30000 -> + error({timeout, waiting_for_select_write}) + end, + + %% socket:send/2 should now transparently retry (internally + %% waiting for write-readiness as needed) and complete + %% successfully instead of returning {error, eagain} or + %% {ok, Rest}. + ok = socket:send(ClientSocket, Chunk), + ok = socket:close(ClientSocket), + ok = + receive + {drained_all, N} when is_integer(N) -> ok + after 30000 -> error({timeout, waiting_for_drain_all}) + end, + ok; + false -> + %% We never managed to fill the send buffer; nothing more to + %% verify on this platform. + Acceptor ! {drain_all, self()}, + ok = socket:close(ClientSocket), + ok = + receive + {drained_all, N} when is_integer(N) -> ok + after 30000 -> error({timeout, waiting_for_drain_all}) + end, + ok + end, + + ok. + +%% @private +%% Accepts a single connection and gives control of when to start reading +%% on it to the test process, so the test can deliberately stall the +%% receiver in order to build up send backpressure on the client side. +backpressure_acceptor(Owner, ListenSocket) -> + {ok, ServerSocket} = socket:accept(ListenSocket), + Owner ! {server_socket, ServerSocket}, + backpressure_acceptor_loop(Owner, ServerSocket), + ok = socket:close(ListenSocket). + +backpressure_acceptor_loop(Owner, ServerSocket) -> + receive + {drain_all, Owner} -> + Total = recv_until_closed(ServerSocket, 0), + Owner ! {drained_all, Total}, + ok = socket:close(ServerSocket) + after 60000 -> + ok = socket:close(ServerSocket) + end. + +recv_until_closed(Socket, Acc) -> + case socket:recv(Socket, 0, 30000) of + {ok, Data} -> + recv_until_closed(Socket, Acc + byte_size(Data)); + {error, closed} -> + Acc; + {error, timeout} -> + Acc + end. + +%% @private +%% Repeatedly calls the raw nif_send/2 (bypassing socket:send/2's automatic +%% retry) with the same chunk of data, until either an {error, eagain} is +%% observed (returns {TotalSent, true}) or MaxAttempts is reached without +%% ever observing backpressure (returns {TotalSent, false}). +fill_send_buffer(_Socket, _Chunk, TotalSent, 0) -> + {TotalSent, false}; +fill_send_buffer(Socket, Chunk, TotalSent, AttemptsLeft) -> + case socket:nif_send(Socket, Chunk) of + ok -> + fill_send_buffer(Socket, Chunk, TotalSent + byte_size(Chunk), AttemptsLeft - 1); + {ok, Rest} -> + Sent = byte_size(Chunk) - byte_size(Rest), + fill_send_buffer(Socket, Chunk, TotalSent + Sent, AttemptsLeft - 1); + {error, eagain} -> + {TotalSent, true}; + {error, Reason} -> + error({unexpected_send_error, Reason}) + end. + +%% +%% test_close_accepted_socket_while_selecting +%% +%% A process selects on an accepted socket and another process closes it. The +%% selecting state of a socket must be initialized whatever way the socket was +%% created, including by accept. +%% + +test_close_accepted_socket_while_selecting() -> + etest:flush_msg_queue(), + + {ok, ListenSocket} = socket:open(inet, stream, tcp), + ok = socket:setopt(ListenSocket, {socket, reuseaddr}, true), + ok = socket:bind(ListenSocket, #{family => inet, addr => loopback, port => 0}), + ok = socket:listen(ListenSocket), + {ok, #{port := Port}} = socket:sockname(ListenSocket), + + {ok, ClientSocket} = socket:open(inet, stream, tcp), + ok = try_connect(ClientSocket, Port, 10), + {ok, ServerSocket} = socket:accept(ListenSocket), + + Self = self(), + Reader = spawn_link(fun() -> + Self ! {recv_result, self(), socket:recv(ServerSocket, 0, 30000)} + end), + timer:sleep(200), + ok = socket:close(ServerSocket), + ok = expect_result(recv_result, Reader, {error, closed}), + + ok = socket:close(ClientSocket), + ok = socket:close(ListenSocket), + ok. + +%% +%% test_close_with_reader_and_writer +%% +%% A socket has one selecting process per direction: a process blocked in +%% recv and another one blocked in send, against a peer that does not read. +%% Closing the socket from a third process must abort both of them. +%% + +test_close_with_reader_and_writer() -> + etest:flush_msg_queue(), + case setup_full_send_buffer() of + {ok, Setup = #{client := Client}} -> + {Reader, Writer} = start_reader_and_writer(Client), + ok = socket:close(Client), + ok = expect_result(recv_result, Reader, {error, closed}), + ok = expect_send_closed(Writer), + teardown_full_send_buffer(Setup); + no_backpressure -> + ok + end. + +%% +%% test_writer_killed_reader_survives +%% +%% When the process blocked in send dies, only its direction is released: the +%% process blocked in recv must still receive data sent afterwards. +%% + +test_writer_killed_reader_survives() -> + etest:flush_msg_queue(), + case setup_full_send_buffer() of + {ok, Setup = #{client := Client, server := Server}} -> + {Reader, Writer} = start_reader_and_writer(Client), + MonitorRef = erlang:monitor(process, Writer), + exit(Writer, kill), + ok = + receive + {'DOWN', MonitorRef, process, Writer, killed} -> ok + after 5000 -> error({timeout, waiting_for_writer_down}) + end, + %% Give the dead writer's monitor time to fire. + timer:sleep(200), + ok = socket:send(Server, <<"hello">>), + ok = expect_result(recv_result, Reader, {ok, <<"hello">>}), + ok = socket:close(Client), + teardown_full_send_buffer(Setup); + no_backpressure -> + ok + end. + +%% +%% test_send_empty +%% +%% Sending an empty payload goes through the NIF as any other payload: it +%% succeeds on a connected socket and fails on a closed one. +%% + +test_send_empty() -> + etest:flush_msg_queue(), + + {ListenSocket, Port} = start_echo_server(0), + {ok, Socket} = socket:open(inet, stream, tcp), + ok = try_connect(Socket, Port, 10), + ok = socket:send(Socket, <<>>), + ok = socket:send(Socket, [<<>>]), + ok = socket:send(Socket, <<"echo:01">>), + {ok, <<"echo:01">>} = socket:recv(Socket, ?PACKET_SIZE, 5000), + ok = close_client_socket(Socket), + {error, _} = socket:send(Socket, <<>>), + {error, _} = socket:send(Socket, <<1>>), + + ok = close_listen_socket(ListenSocket), + ok. + +%% +%% test_reader_cleanup_keeps_writer +%% +%% A process blocked in recv and another one blocked in send, against a peer +%% that does not read. When the peer half-closes (shutdown write), the reader +%% gets {error, closed} and cleans up its select. This cleanup must not disown +%% the writer: when the peer then reads everything, the send completes. +%% + +test_reader_cleanup_keeps_writer() -> + etest:flush_msg_queue(), + case setup_full_send_buffer() of + {ok, Setup = #{client := Client, server := Server, sent := Sent}} -> + {Reader, Writer} = start_reader_and_writer(Client), + ok = socket:shutdown(Server, write), + ok = expect_result(recv_result, Reader, {error, closed}), + %% The writer is still blocked: read everything on the peer. + ok = socket:setopt(Server, {otp, rcvbuf}, 65536), + Total = Sent + ?LARGE_PAYLOAD_SIZE, + ok = recv_at_least(Server, Total), + ok = expect_result(send_result, Writer, ok), + ok = socket:close(Client), + teardown_full_send_buffer(Setup); + no_backpressure -> + ok + end. + +%% +%% test_send_error_after_partial_send +%% +%% As in OTP, an error following a partial send is reported with the data that +%% was not sent: {error, {Reason, RestData}}. Without any data accepted, it is +%% a plain {error, Reason}. +%% + +test_send_error_after_partial_send() -> + etest:flush_msg_queue(), + case setup_full_send_buffer() of + {ok, Setup = #{client := Client, server := Server}} -> + Self = self(), + Data = binary:copy(<<0>>, ?LARGE_PAYLOAD_SIZE), + Writer = spawn(fun() -> + Self ! {send_result, self(), socket:send(Client, Data)} + end), + timer:sleep(100), + %% Read some data on the peer so that the writer makes progress, + %% then let it block again. + ok = socket:setopt(Server, {otp, rcvbuf}, 65536), + ok = recv_at_least(Server, 1024 * 1024), + timer:sleep(200), + ok = socket:close(Client), + receive + {send_result, Writer, {error, {_Reason, Rest}}} when + is_binary(Rest), byte_size(Rest) > 0, byte_size(Rest) < ?LARGE_PAYLOAD_SIZE + -> + ok; + {send_result, Writer, Other} -> + error({unexpected_send_result, Other}) + after 5000 -> + error({timeout, send_result}) + end, + teardown_full_send_buffer(Setup); + no_backpressure -> + ok + end. + +%% @private +%% Receives at least Remaining bytes (a recv can return more than needed). +recv_at_least(_Socket, Remaining) when Remaining =< 0 -> + ok; +recv_at_least(Socket, Remaining) -> + case socket:recv(Socket, 0, 10000) of + {ok, Data} -> + recv_at_least(Socket, Remaining - byte_size(Data)); + {error, Reason} -> + error({recv_failed, Reason, Remaining}) + end. + +%% @private +%% Starts a process blocked in recv on Client, and then a process blocked in +%% send on it (the peer of Client does not read). +start_reader_and_writer(Client) -> + Self = self(), + Reader = spawn(fun() -> + Self ! {recv_result, self(), socket:recv(Client, 0, 30000)} + end), + timer:sleep(100), + %% Much more than any socket buffer can hold (buffers can grow after the + %% first eagain), so that the send cannot complete while the peer is not + %% reading. + Data = binary:copy(<<0>>, ?LARGE_PAYLOAD_SIZE), + Writer = spawn(fun() -> + Self ! {send_result, self(), socket:send(Client, Data)} + end), + timer:sleep(200), + {Reader, Writer}. + +%% @private +%% The writer may or may not have sent part of its data before being aborted, +%% so the error is either {error, closed} or {error, {closed, RestData}}. +expect_send_closed(Writer) -> + receive + {send_result, Writer, {error, closed}} -> ok; + {send_result, Writer, {error, {closed, Rest}}} when is_binary(Rest) -> ok; + {send_result, Writer, Other} -> error({unexpected_result, send_result, Other}) + after 5000 -> + error({timeout, send_result}) + end. + +%% @private +expect_result(Tag, Pid, Expected) -> + receive + {Tag, Pid, Expected} -> ok; + {Tag, Pid, Other} -> error({unexpected_result, Tag, Other}) + after 5000 -> + error({timeout, Tag}) + end. + +%% @private +%% Connects a client to a server that never reads, and fills the client send +%% buffer. Returns no_backpressure if the platform never reported eagain. +setup_full_send_buffer() -> + {ok, ListenSocket} = socket:open(inet, stream, tcp), + ok = socket:setopt(ListenSocket, {socket, reuseaddr}, true), + ok = socket:setopt(ListenSocket, {socket, linger}, #{onoff => true, linger => 0}), + ok = socket:bind(ListenSocket, #{family => inet, addr => loopback, port => 0}), + ok = socket:listen(ListenSocket), + {ok, #{port := Port}} = socket:sockname(ListenSocket), + + Self = self(), + Acceptor = spawn_link(fun() -> backpressure_acceptor(Self, ListenSocket) end), + + {ok, Client} = socket:open(inet, stream, tcp), + ok = try_connect(Client, Port, 10), + Server = + receive + {server_socket, ServerSocket} -> ServerSocket + after 5000 -> + error({timeout, waiting_for_server_socket}) + end, + + Chunk = binary:copy(<<0>>, 65536), + Setup = #{client => Client, server => Server, acceptor => Acceptor}, + case fill_send_buffer(Client, Chunk, 0, 256) of + {TotalSent, true} -> + {ok, Setup#{sent => TotalSent}}; + {_TotalSent, false} -> + teardown_full_send_buffer(Setup), + no_backpressure + end. + +%% @private +%% Lets the acceptor drain and close its side. The client socket must be +%% closed by the caller, or is closed here if it was not already. +teardown_full_send_buffer(#{client := Client, acceptor := Acceptor}) -> + _ = socket:close(Client), + Acceptor ! {drain_all, self()}, + ok = + receive + {drained_all, N} when is_integer(N) -> ok + after 30000 -> error({timeout, waiting_for_drain_all}) + end, + ok. + id(X) -> X. diff --git a/tests/libs/estdlib/test_udp_socket.erl b/tests/libs/estdlib/test_udp_socket.erl index ff57169669..fea90f1c93 100644 --- a/tests/libs/estdlib/test_udp_socket.erl +++ b/tests/libs/estdlib/test_udp_socket.erl @@ -24,6 +24,7 @@ test() -> ok = test_echo(), + ok = test_send_empty(), ok = test_buf_size(), ok = test_timeout(), ok = test_nowait(), @@ -50,6 +51,18 @@ test() -> -define(PACKET_SIZE, 7). +%% An empty datagram is a valid datagram: sendto/3 must hand it to the socket. +test_send_empty() -> + {ok, Receiver} = socket:open(inet, dgram, udp), + ok = socket:bind(Receiver, #{family => inet, addr => loopback, port => 0}), + {ok, #{port := Port}} = socket:sockname(Receiver), + {ok, Sender} = socket:open(inet, dgram, udp), + ok = socket:sendto(Sender, <<>>, #{family => inet, addr => loopback, port => Port}), + ok = socket:close(Sender), + {error, _} = socket:sendto(Sender, <<>>, #{family => inet, addr => loopback, port => Port}), + ok = socket:close(Receiver), + ok. + start_echo_server(Port) -> {ok, Socket} = socket:open(inet, dgram, udp), diff --git a/tests/test-enif.c b/tests/test-enif.c index 200b1d4534..1447e2a8c7 100644 --- a/tests/test-enif.c +++ b/tests/test-enif.c @@ -39,6 +39,15 @@ #include "sys.h" #include "utils.h" +// Evaluate an expression with side effects outside of assert(), so that it is +// still evaluated when assertions are disabled (NDEBUG). +#define VERIFY(expr) \ + do { \ + bool verify_ok = (expr); \ + assert(verify_ok); \ + UNUSED(verify_ok); \ + } while (0) + static uint32_t cb_read_resource = 0; static int32_t down_pid = 0; static ErlNifMonitor down_mon = { NULL, 0 }; @@ -761,9 +770,12 @@ void test_select_event_armed_during_listener_only_rebuild(void) select_event->read = true; select_event->write = false; select_event->close = false; - select_event->local_pid = ctx->process_id; - select_event->ref_ticks = 0; - select_event->message = NULL; + select_event->read_local_pid = ctx->process_id; + select_event->write_local_pid = INVALID_PROCESS_ID; + select_event->read_ref_ticks = 0; + select_event->write_ref_ticks = 0; + select_event->read_message = NULL; + select_event->write_message = NULL; refc_binary_increment_refcount(select_event->resource); list_init(&select_event->head); struct ListHead *select_events = synclist_wrlock(&glb->select_events); @@ -796,6 +808,332 @@ void test_select_event_armed_during_listener_only_rebuild(void) } #endif +// enif_select_read/enif_select_write should track independent messages, refs +// and target pids for the same event, so a resource can have a pending read +// select and a pending write select at the same time (e.g. a socket with a +// permanent recv select and a transient send-backpressure select). +void test_resource_select_read_write_independent_messages(void) +{ + GlobalContext *glb = globalcontext_new(); + Context *read_ctx = context_new(glb); + Context *write_ctx = context_new(glb); + ErlNifEnv *env = erl_nif_env_from_context(read_ctx); + + ErlNifResourceTypeInit init; + init.members = 1; + init.dtor = resource_dtor; + ErlNifResourceFlags flags; + cb_read_resource = 0; + dtor_call_count = 0; + + ErlNifResourceType *resource_type = enif_init_resource_type(env, "test_select_resource", &init, ERL_NIF_RT_CREATE, &flags); + assert(resource_type != NULL); + + void *ptr = enif_alloc_resource(resource_type, sizeof(uint32_t)); + uint32_t *resource = (uint32_t *) ptr; + *resource = 99; + + int pipefds[2]; + VERIFY(pipe(pipefds) == 0); + ErlNifEvent event = (ErlNifEvent) pipefds[0]; + + ErlNifPid read_pid = read_ctx->process_id; + ErlNifPid write_pid = write_ctx->process_id; + + term read_msg = term_from_int(11); + term write_msg = term_from_int(22); + + int r = enif_select_read(env, event, ptr, &read_pid, read_msg, NULL); + assert(r == 0); + r = enif_select_write(env, event, ptr, &write_pid, write_msg, NULL); + assert(r == 0); + + // Firing both read and write readiness must deliver the read message to + // read_ctx and the write message to write_ctx, without either clobbering + // the other (this is the bug being tested: a shared message/ref/pid field + // would either lose one notification or send it to the wrong process). + bool notified = select_event_notify(event, true, true, glb); + assert(notified); + + VERIFY(mailbox_process_outer_list(read_ctx) == NULL); + assert(mailbox_has_next(&read_ctx->mailbox)); + term read_received; + VERIFY(mailbox_peek(read_ctx, &read_received)); + assert(read_received == read_msg); + mailbox_remove_message(&read_ctx->mailbox, &read_ctx->heap); + assert(!mailbox_has_next(&read_ctx->mailbox)); + + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + assert(mailbox_has_next(&write_ctx->mailbox)); + term write_received; + VERIFY(mailbox_peek(write_ctx, &write_received)); + assert(write_received == write_msg); + mailbox_remove_message(&write_ctx->mailbox, &write_ctx->heap); + assert(!mailbox_has_next(&write_ctx->mailbox)); + + // Once a direction fires, only that direction is consumed; the other + // stays armed until it is separately triggered or explicitly stopped. + // Re-arm both to test the ref-ticks (no custom message) path together. + r = enif_select_read(env, event, ptr, &read_pid, read_msg, NULL); + assert(r == 0); + r = enif_select_write(env, event, ptr, &write_pid, write_msg, NULL); + assert(r == 0); + + // Re-selecting read must not disturb the pending write selection (and + // vice-versa): overwrite only the read side with a new pid/message. + Context *other_read_ctx = context_new(glb); + ErlNifPid other_read_pid = other_read_ctx->process_id; + term other_read_msg = term_from_int(33); + r = enif_select_read(env, event, ptr, &other_read_pid, other_read_msg, NULL); + assert(r == 0); + + notified = select_event_notify(event, true, true, glb); + assert(notified); + + // Original read_ctx should get nothing new (its select was overwritten). + VERIFY(mailbox_process_outer_list(read_ctx) == NULL); + assert(!mailbox_has_next(&read_ctx->mailbox)); + + VERIFY(mailbox_process_outer_list(other_read_ctx) == NULL); + assert(mailbox_has_next(&other_read_ctx->mailbox)); + term other_read_received; + VERIFY(mailbox_peek(other_read_ctx, &other_read_received)); + assert(other_read_received == other_read_msg); + + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + assert(mailbox_has_next(&write_ctx->mailbox)); + term write_received_again; + VERIFY(mailbox_peek(write_ctx, &write_received_again)); + assert(write_received_again == write_msg); + mailbox_remove_message(&write_ctx->mailbox, &write_ctx->heap); + + // Both directions have now fired and are no longer armed, so stopping the + // select should immediately release the extra refcount it was holding. + int stop_result = enif_select(env, event, ERL_NIF_SELECT_STOP, ptr, NULL, term_nil()); + assert(stop_result == ERL_NIF_SELECT_STOP_CALLED); + + int release_result = enif_release_resource(ptr); + assert(release_result); + + scheduler_terminate(read_ctx); + scheduler_terminate(write_ctx); + scheduler_terminate(other_read_ctx); + + close(pipefds[0]); + close(pipefds[1]); + + globalcontext_destroy(glb); +} + +// The generic enif_select/2 API (no custom message, just a reference) must +// also track read and write refs independently when both directions are +// selected on the same event, e.g. a single ref selecting both directions, +// followed by a read-only re-select that must not clear the write ref. +void test_resource_select_read_write_independent_refs(void) +{ + GlobalContext *glb = globalcontext_new(); + Context *ctx = context_new(glb); + ErlNifEnv *env = erl_nif_env_from_context(ctx); + + ErlNifResourceTypeInit init; + init.members = 1; + init.dtor = resource_dtor; + ErlNifResourceFlags flags; + cb_read_resource = 0; + + ErlNifResourceType *resource_type = enif_init_resource_type(env, "test_select_resource_refs", &init, ERL_NIF_RT_CREATE, &flags); + assert(resource_type != NULL); + + void *ptr = enif_alloc_resource(resource_type, sizeof(uint32_t)); + uint32_t *resource = (uint32_t *) ptr; + *resource = 7; + + int pipefds[2]; + VERIFY(pipe(pipefds) == 0); + ErlNifEvent event = (ErlNifEvent) pipefds[0]; + + ErlNifPid pid = ctx->process_id; + + enum MemoryGCResult ensure_free_result = memory_erl_nif_env_ensure_free(env, TERM_BOXED_REFERENCE_SHORT_SIZE); + assert(ensure_free_result == MEMORY_GC_OK); + UNUSED(ensure_free_result); + term write_ref = term_from_ref_ticks(globalcontext_get_ref_ticks(glb), &env->heap); + + // Select for write with an explicit ref and no custom message. + int r = enif_select(env, event, ERL_NIF_SELECT_WRITE, ptr, &pid, write_ref); + assert(r == 0); + + // Now select read only, with UNDEFINED_ATOM (no ref requested). This must + // not disturb the write ref registered above. + r = enif_select(env, event, ERL_NIF_SELECT_READ, ptr, &pid, UNDEFINED_ATOM); + assert(r == 0); + + bool notified = select_event_notify(event, true, true, glb); + assert(notified); + + // Two notifications should have been queued: {select, Resource, undefined, ready_input} + // and {select, Resource, WriteRef, ready_output}. + VERIFY(mailbox_process_outer_list(ctx) == NULL); + assert(mailbox_has_next(&ctx->mailbox)); + + bool saw_read = false; + bool saw_write = false; + for (int i = 0; i < 2; i++) { + term msg; + VERIFY(mailbox_peek(ctx, &msg)); + assert(term_is_tuple(msg)); + assert(term_get_tuple_arity(msg) == 4); + assert(term_get_tuple_element(msg, 0) == SELECT_ATOM); + term ref_or_undefined = term_get_tuple_element(msg, 2); + term kind = term_get_tuple_element(msg, 3); + if (kind == READY_INPUT_ATOM) { + assert(ref_or_undefined == UNDEFINED_ATOM); + saw_read = true; + } else { + assert(kind == READY_OUTPUT_ATOM); + assert(term_is_reference(ref_or_undefined)); + saw_write = true; + } + if (i == 0) { + assert(mailbox_has_next(&ctx->mailbox)); + mailbox_next(&ctx->mailbox); + } + } + assert(saw_read); + assert(saw_write); + mailbox_remove_message(&ctx->mailbox, &ctx->heap); + mailbox_remove_message(&ctx->mailbox, &ctx->heap); + assert(!mailbox_has_next(&ctx->mailbox)); + + // Both directions have now fired and are no longer armed, so stopping the + // select should immediately release the extra refcount it was holding. + int stop_result = enif_select(env, event, ERL_NIF_SELECT_STOP, ptr, NULL, term_nil()); + assert(stop_result == ERL_NIF_SELECT_STOP_CALLED); + + int release_result = enif_release_resource(ptr); + assert(release_result); + + scheduler_terminate(ctx); + + close(pipefds[0]); + close(pipefds[1]); + +#ifdef AVM_TASK_DRIVER_ENABLED + // Notifications built without a custom message (the ref-only path + // exercised above) release their transient resource reference through + // the task-driver refc queue rather than synchronously; drain it so the + // resource is properly freed instead of merely being reported as a + // (harmless) dangling resource on shutdown. + globalcontext_process_task_driver_queues(glb); +#endif + + globalcontext_destroy(glb); +} + +// ERL_NIF_SELECT_CANCEL must only cancel the requested direction(s), dispose +// their pending messages, and keep the other direction armed. +void test_resource_select_cancel(void) +{ + GlobalContext *glb = globalcontext_new(); + Context *read_ctx = context_new(glb); + Context *write_ctx = context_new(glb); + ErlNifEnv *env = erl_nif_env_from_context(read_ctx); + + ErlNifResourceTypeInit init; + init.members = 1; + init.dtor = resource_dtor; + ErlNifResourceFlags flags; + cb_read_resource = 0; + dtor_call_count = 0; + + ErlNifResourceType *resource_type = enif_init_resource_type(env, "test_select_cancel_resource", &init, ERL_NIF_RT_CREATE, &flags); + assert(resource_type != NULL); + + void *ptr = enif_alloc_resource(resource_type, sizeof(uint32_t)); + + int pipefds[2]; + VERIFY(pipe(pipefds) == 0); + ErlNifEvent event = (ErlNifEvent) pipefds[0]; + + ErlNifPid read_pid = read_ctx->process_id; + ErlNifPid write_pid = write_ctx->process_id; + term read_msg = term_from_int(11); + term write_msg = term_from_int(22); + + // Cancelling an event that was never selected is not an error. + int r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == 0); + + // Invalid combinations. + r = enif_select(env, event, ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == ERL_NIF_SELECT_BADARG); + r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_STOP | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == ERL_NIF_SELECT_BADARG); + + VERIFY(enif_select_read(env, event, ptr, &read_pid, read_msg, NULL) == 0); + VERIFY(enif_select_write(env, event, ptr, &write_pid, write_msg, NULL) == 0); + + // Cancel read only: write stays armed. + r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == ERL_NIF_SELECT_READ_CANCELLED); + // Cancelling again has nothing to cancel. + r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == 0); + + VERIFY(select_event_notify(event, true, true, glb)); + VERIFY(mailbox_process_outer_list(read_ctx) == NULL); + assert(!mailbox_has_next(&read_ctx->mailbox)); + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + term received; + VERIFY(mailbox_peek(write_ctx, &received)); + assert(received == write_msg); + mailbox_remove_message(&write_ctx->mailbox, &write_ctx->heap); + + // Cancel write only: read (armed again) stays armed. + VERIFY(enif_select_read(env, event, ptr, &read_pid, read_msg, NULL) == 0); + VERIFY(enif_select_write(env, event, ptr, &write_pid, write_msg, NULL) == 0); + r = enif_select(env, event, ERL_NIF_SELECT_WRITE | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == ERL_NIF_SELECT_WRITE_CANCELLED); + + VERIFY(select_event_notify(event, true, true, glb)); + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + assert(!mailbox_has_next(&write_ctx->mailbox)); + VERIFY(mailbox_process_outer_list(read_ctx) == NULL); + VERIFY(mailbox_peek(read_ctx, &received)); + assert(received == read_msg); + mailbox_remove_message(&read_ctx->mailbox, &read_ctx->heap); + + // Cancel both at once: nothing is delivered afterwards. + VERIFY(enif_select_read(env, event, ptr, &read_pid, read_msg, NULL) == 0); + VERIFY(enif_select_write(env, event, ptr, &write_pid, write_msg, NULL) == 0); + r = enif_select(env, event, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_WRITE | ERL_NIF_SELECT_CANCEL, ptr, NULL, term_nil()); + assert(r == (ERL_NIF_SELECT_READ_CANCELLED | ERL_NIF_SELECT_WRITE_CANCELLED)); + VERIFY(!select_event_notify(event, true, true, glb)); + assert(!mailbox_has_next(&read_ctx->mailbox)); + assert(!mailbox_has_next(&write_ctx->mailbox)); + + // The event is kept after cancelling, so it can be selected again. + VERIFY(enif_select_write(env, event, ptr, &write_pid, write_msg, NULL) == 0); + VERIFY(select_event_notify(event, false, true, glb)); + VERIFY(mailbox_process_outer_list(write_ctx) == NULL); + VERIFY(mailbox_peek(write_ctx, &received)); + assert(received == write_msg); + mailbox_remove_message(&write_ctx->mailbox, &write_ctx->heap); + + // Nothing is armed anymore, so stop is immediate. + VERIFY(enif_select(env, event, ERL_NIF_SELECT_STOP, ptr, NULL, term_nil()) == ERL_NIF_SELECT_STOP_CALLED); + + VERIFY(enif_release_resource(ptr)); + + scheduler_terminate(read_ctx); + scheduler_terminate(write_ctx); + + close(pipefds[0]); + close(pipefds[1]); + + globalcontext_destroy(glb); +} + int main(int argc, char **argv) { UNUSED(argc); @@ -814,6 +1152,9 @@ int main(int argc, char **argv) #ifndef HAVE_KQUEUE test_select_event_armed_during_listener_only_rebuild(); #endif + test_resource_select_read_write_independent_messages(); + test_resource_select_read_write_independent_refs(); + test_resource_select_cancel(); return EXIT_SUCCESS; }