[zmstone]

Add prim_inet:send_async/2,3 for TCP sockets of the inet backend (ports of the inet driver). The function gives the data to the inet driver and returns at once. The result of the send arrives later in the mailbox of the caller.

Why

gen_tcp:send/2 gives the data to the inet driver and then waits for the reply of the driver. Each outgoing packet costs the caller a round trip through the port, an inet_db:lookup_socket call, a port monitor, a term_to_binary/2 call and a timed receive.

EMQX 4.4 sent with erlang:port_command/3 and read the reply from the mailbox later. Since OTP 26 (OTP-18520), the driver requires a caller tag in front of the data, and a bare port_command fails with EINVAL. esockd therefore made its asynchronous send synchronous on OTP 26 and later.

In an EMQX benchmark (1:1 publish/subscribe, QoS 0, 1 KB payload, 20k messages/s, 40k connections, 8 schedulers, gen_tcp listener), the asynchronous send lowered broker CPU per message from 256.5 µs to 225.5 µs (mean of 3 runs each, -12.1%).

Change

erts/preloaded/src/prim_inet.erl, next to send/3, and the rebuilt preloaded prim_inet.beam. No driver change and no public API change: the function is in prim_inet because it works only on inet driver ports, takes erlang:port_command/3 options and returns the driver's own reply message.

• send_async(Port, Data) and send_async(Port, Data, Options), with Options passed to erlang:port_command/3 ([], [nosuspend]).
• Return: ok, {error, busy} (port busy and nosuspend given; nothing is queued), or {error, closed} (the port is gone). A driver error arrives in the reply message, not as a return value.
• A socket of the socket backend is not a port and does not match the function (function_clause).
• One reply per ok, in send order: {inet_reply, Socket, ok | {error, Reason}, send_async}. When the driver queue is above high_watermark, the reply is delayed until the queue drains below low_watermark.
• A port that dies before it replies sends no reply. The owner receives the exit signal of the port link, or {tcp_closed, Socket} in active mode.

The function prepends a constant tag, term_to_binary(send_async), pre-encoded as a literal binary, instead of a reference for each call. The driver only copies the tag and echoes it in the reply. So a send costs no port monitor and no term_to_binary/2 call.

A zero-length tag was rejected. With a zero-length tag the driver drops the whole reply, including the {error, timeout} reply of send_timeout.

Tests

New group t_send_async in gen_tcp_api_SUITE (under each backend group). The cases open sockets with gen_tcp and send with prim_inet:send_async/2,3:

• t_send_async_basic: ok reply, data received by the peer, iodata accepted, no other messages.
• t_send_async_busy: {error, busy} with nosuspend; the reply of a send queued above the high watermark arrives after the peer reads the data.
• t_send_async_send_timeout: send_timeout arrives as an {error, timeout} reply.
• t_send_async_send_timeout_close: after send_timeout_close, the next send gets an error reply (enotconn).
• t_send_async_port_death: a killed port with a pending reply sends no reply; the owner gets {'EXIT', Socket, killed}.

On the socket backend each case skips.

Run on OTP 28.4.1-7 with the new prim_inet loaded at node start (erlang:load_module/2, before ct_run): gen_tcp_api_SUITE 31 ok, 0 failed, 2 skipped (environment: no link-local IPv6 address, util NIF). With -kernel inet_backend socket: the five new cases skip as inet-only testcase.

Release

This PR targets the 28.4.1 line (emqx-OTP-28.4.1), which EMQX 6.3 uses. The port to the 28.5 line (EMQX 7.0) is for the maintainer of that line.

The EMQX changes that use the function: emqx/esockd#236 (esockd_transport:async_send/3) and emqx/emqx#19627 (emqx_connection).

[zmstone]

ci failure is unrelated.

[keynslug]

I think it makes more sense to put directly in prim_inet as it: 1) does not work for socket-backed sockets, 2) skips implementation module lookups, 3) and exposes port-specific options.

[keynslug]

Nit. inet:posix() is probably impossible to get here?