feat(socket): implement `esockd_socket` connection module #15451

keynslug · 2025-06-30T16:22:09Z

Part of EMQX-14333.

Release version: 6.0.0

Summary

This PR introduces tcp_backend listener option, exclusively to TCP listeners, and employs emqx/esockd#204 with a separate connection module named emqx_socket_connection when tcp_backend is socket. This shows consistently better wide-fanout performance, especially when combined with async shard dispatch introduced in #15421, see report.

This new connection module is identical to emqx_connection module in most of aspects, but made separate primarily because of:

Socket-based connections use substantially different connection loop, and integrating it into emqx_connection might have made the latter too bloated, harder to test and reason about.
Combined module would have larger footprint, which would lead to a bunch of "micro-deoptimizations" affecting every listener, admittedly pretty minor ones.

Obvious downside is the need to keep changes to connection modules in sync. This requires additional development processes, or perhaps some CI magic, otherwise code will eventually diverge. Ideas are welcome.

See individual commits for details.

Notes:

Commit ed96d99 is part of iovec() experiments that I decided to include in this PR, because it shows small consistent improvements in average message processing latency (~2% less) and VM reductions (~3.5% less).
While TLS connections are currently not supported under esockd_socket listeners, there's ongoing work in Erlang/OTP to support familiar SSL API with socket sockets.

PR Checklist

For internal contributor: there is a jira ticket to track this change
The changes are covered with new or existing tests
Change log for changes visible by users has been added to changes/ee/(feat|perf|fix|breaking)-<PR-id>.en.md files
Schema changes are backward compatible

codecov-commenter · 2025-07-09T16:50:22Z

Codecov Report

Attention: Patch coverage is 76.65370% with 120 lines in your changes missing coverage. Please review.

Project coverage is 84.47%. Comparing base (22d202c) to head (80429fa).
Report is 2 commits behind head on release-60.

Files with missing lines	Patch %	Lines
apps/emqx/src/emqx_socket_connection.erl	72.40%	117 Missing ⚠️
apps/emqx/src/emqx_congestion.erl	96.87%	1 Missing ⚠️
apps/emqx/src/emqx_listeners.erl	90.90%	1 Missing ⚠️
apps/emqx/src/emqx_schema.erl	50.00%	1 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff               @@
##           release-60   #15451      +/-   ##
==============================================
- Coverage       84.51%   84.47%   -0.04%     
==============================================
  Files            1113     1114       +1     
  Lines           78238    78682     +444     
==============================================
+ Hits            66120    66466     +346     
- Misses          12118    12216      +98

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

qzhuyan · 2025-07-15T09:40:02Z

rel/i18n/emqx_schema.hocon

+- `gen_tcp`: Standard backend, in use since EMQX 5.0 release.
+
+- `socket`: Experimental backend, looking to improve message latency and compute resource usage.
+    Note that some `tcp_options` settings will have no effect when using this backend, e.g.: `high_watermark` and `send_timeout_close`."""


not suggested for IPv6 as otp doc said it is not fully tested.

Well, it says "experimental". Note that the same doc also does not say the same about {inet_backend, socket} in gen_tcp, which suggests that relevant API surface is well-tested.

thalesmg

🌶️
(still reading)

thalesmg · 2025-07-15T13:50:19Z

apps/emqx/src/emqx_socket_connection.erl

+-endif.
+
+-elvis([{elvis_style, used_ignored_variable, disable}]).
+-elvis([{elvis_style, invalid_dynamic_call, #{ignore => [emqx_connection]}}]).


Stale directive? Or perhaps:

Suggested change

-elvis([{elvis_style, invalid_dynamic_call, #{ignore => [emqx_connection]}}]).

-elvis([{elvis_style, invalid_dynamic_call, #{ignore => [emqx_socket_connection]}}]).

?

apps/emqx/src/emqx_socket_connection.erl

thalesmg · 2025-07-15T14:06:33Z

apps/emqx/src/emqx_socket_connection.erl

+handle_msg({'$socket', _Socket, abort, {_Handle, Reason}}, State) ->
+    handle_info({sock_error, Reason}, State);


Q: could this be a stale message if _Handle in #congested.handle is different?

Hmm, IIUC any abort basically means someone closed the socket / there was a socket error. Maybe I'm missing something?

I was asking if it is possible that a stale abort arrives here, but is no longer relevant, so it wouldn't be necessary to log it? e.g. the handle changed, and the Handle in this message has been abandoned.

sock_error later forces the connection process to close and abandon the socket. But in theory we may get 2 aborts per each outstanding select (one for recv, one for send). So, spot on, I'll double-check.

thalesmg

🔥

thalesmg · 2025-07-15T16:19:29Z

apps/emqx/src/emqx_socket_connection.erl

+        emqx_congestion:cancel_alarms(?MODULE, State),
+        emqx_channel:terminate(Reason, Channel1),
+        close_socket_ok(State),
+        ?TRACE("SOCKET", "emqx_connection_terminated", #{reason => Reason})


Suggested change

?TRACE("SOCKET", "emqx_connection_terminated", #{reason => Reason})

?TRACE("SOCKET", "emqx_socket_connection_terminated", #{reason => Reason})

?

thalesmg · 2025-07-15T16:20:10Z

apps/emqx/src/emqx_socket_connection.erl

+        ?TRACE("SOCKET", "emqx_connection_terminated", #{reason => Reason})
+    catch
+        E:C:S ->
+            ?tp(warning, unclean_terminate, #{exception => E, context => C, stacktrace => S})


Nit: maybe add the connection module as metadata, to differentiate between this module and emqx_connection?

Though, seeing how many tracepoints and logs would need to change, maybe it could be added to the process logger metadata. 🤔

Good point, did the latter.

thalesmg · 2025-07-15T16:26:47Z

apps/emqx/src/emqx_socket_connection.erl

+
+-compile({inline, [request_more_data/4]}).
+request_more_data(Socket, More, Acc, State) ->
+    %% TODO: `{otp, select_read}`.


Q: should this be addressed here? Or maybe in a follow up?

I was thinking about a followup. It's not entirely clear this will prove useful, and at very least need experimentation.

thalesmg · 2025-07-15T16:40:26Z

apps/emqx/src/emqx_socket_connection.erl

+%%--------------------------------------------------------------------
+%% Send data
+
+-spec send(non_neg_integer(), iodata(), state()) -> {ok, state()}.


Nit: typespec does not cover {ok, {sock_error, _}, state()}.

thalesmg · 2025-07-15T16:43:48Z

apps/emqx/src/emqx_socket_connection.erl

+            {ok, queue_send(Handle, Rest, State)};
+        {select, _Info} ->
+            %% Totally congested, keep the deadline.
+            NState = State#state{sockstate = SS#congested{sendq = IoData}},


Q: don't we need to update/set #congested.handle to the freshly created handle here?

Good find! That's most likely a bug.

qzhuyan · 2025-07-22T10:45:39Z

apps/emqx/src/emqx_frame.erl


 -type parse_result() ::
-    {more, parse_state()}
+    %% Need more bytes out of stream, `0` means it's unclear how much more.


it is counter-intuitive, maybe error prone.

More = 0, is nomore to me.

I prefer type unknown | non_neg_integer()

While I agree about explicitness / error-proneness, unknown and 0 have no practical difference for callers, at least for now: i.e. there's no such state as "we don't need to read byte stream anymore". And using 0 avoids an extra mapping step in emqx_socket_connection.

I prefer type unknown | non_neg_integer()

Perhaps you meant unknown | pos_integer()?

Addressed in 566e73a.

qzhuyan · 2025-07-22T11:00:18Z

apps/emqx/test/emqx_client_SUITE.erl

-    ok = emqtt:publish(C2, nth(2, ?TOPICS), <<"qos 0">>, 0),
-    {ok, _} = emqtt:publish(C2, nth(3, ?TOPICS), <<"qos 1">>, 1),
-    {ok, _} = emqtt:publish(C2, nth(4, ?TOPICS), <<"qos 2">>, 2),
+    ok = emqtt:publish(C2, <<"TopicA/B">>, <<"qos 0">>, 0),


nit, honestly I think old code is more readable

Interesting! To me, it was always an additional cognitive load to understand how nth(3, ?TOPICS) and nth(4, ?TOPICS) relate to nth(6, ?WILD_TOPICS) in one testcase, or why nth(7, ?WILD_TOPICS) and nth(1, ?WILD_TOPICS) were specifically chosen for "overlapping subscriptions" testcase, etc.

zmstone · 2025-07-22T11:04:53Z

apps/emqx/src/emqx_listeners.erl

+        _Different ->
+            %% TODO
+            %% Again, we're not strictly required to drop live connections in this case.
+            {error, not_supported}


maybe add some description to the config schema desc?
e.g. when choose to use socket (instead of gen_tcp), the socket options cannot be updated from Dashboard or API.

Most of TCP options can be updated I think, it's tcp_backend that cannot be changed without running restart_listener/4. I'll clarify the latter, unless I'm missing something.

zmstone · 2025-07-22T11:13:13Z

apps/emqx/src/emqx_socket_connection.erl

+%% This module interacts with the transport layer of MQTT
+%% Transport: esockd_socket.
+%%
+%% NOTE


nit: add same note to emqx_connection too ?

zmstone · 2025-07-22T11:47:47Z

apps/emqx/src/emqx_socket_connection.erl

+-compile({inline, [request_more_data/4]}).
+request_more_data(Socket, More, Acc, State) ->
+    %% TODO: `{otp, select_read}`.
+    case sock_async_recv(Socket, More) of


will this return {complete, ComplteInfo}?

Only when running on Windows I believe. What do you suggest to do about it?

maybe do not allow this backend on Window when checking schema.

apps/emqx/src/emqx_socket_connection.erl

zmstone · 2025-07-22T12:06:23Z

apps/emqx/src/emqx_socket_connection.erl

+        {ok, Data} ->
+            NState = start_idle_timer(State),
+            handle_recv({recv_more, Data}, Parent, NState);
+        {select, _SelectInfo} ->


so there is no need to kee SelectInfo for the next call ?

Currently, socket keeps no special context in recv's select_info()s, it's just {select_info, recv, Handle}. Supposedly, one can just socket:recv(Socket, 0, ...) again once requested number of bytes has arrived, signaled through {$socket, Socket, select, ...}.

Connection loop needs to disambiguate recv and send "selects" though, and for that we only keep send handles (in #congested{} structure) but not recv handles.

This is useful for `socket`-based connections to request subsequent asynchronous socket reads.

Behavior is slightly different from `emqx_connection`: there's no equivalent to `high_watermark`, so connection is never blocked on send.

…rror Effectively mirroring `emqx_socket_connection` behavior.

Co-authored-by: zmstone <zmstone@gmail.com>

emqxqa · 2025-07-28T08:24:37Z

TestExecution

keynslug force-pushed the perf/EMQX-14333/fanout-part-iv branch 5 times, most recently from 5483eef to 23a9313 Compare July 2, 2025 18:47

keynslug force-pushed the perf/EMQX-14333/fanout-part-iv branch 3 times, most recently from d8f2480 to c25571c Compare July 9, 2025 16:20

keynslug force-pushed the perf/EMQX-14333/fanout-part-iv branch from c25571c to 5392456 Compare July 10, 2025 17:38

keynslug marked this pull request as ready for review July 10, 2025 18:42

keynslug requested a review from a team as a code owner July 10, 2025 18:42

qzhuyan reviewed Jul 15, 2025

View reviewed changes

thalesmg reviewed Jul 15, 2025

View reviewed changes

thalesmg previously approved these changes Jul 15, 2025

View reviewed changes

keynslug dismissed thalesmg’s stale review via 73cc032 July 18, 2025 15:16

keynslug force-pushed the perf/EMQX-14333/fanout-part-iv branch 3 times, most recently from 8cefbac to b072f42 Compare July 18, 2025 16:14

qzhuyan previously approved these changes Jul 22, 2025

View reviewed changes

keynslug dismissed qzhuyan’s stale review via 566e73a July 22, 2025 12:01

zmstone reviewed Jul 22, 2025

View reviewed changes

keynslug requested a review from zmstone July 23, 2025 10:53

zmstone previously approved these changes Jul 23, 2025

View reviewed changes

keynslug added 5 commits July 23, 2025 16:18

feat(parser): tell how much more bytes are expected when incomplete

0630af6

This is useful for `socket`-based connections to request subsequent asynchronous socket reads.

chore: update esockd to 5.15.0 with esockd_socket backend

7cf8259

feat(socket): implement esockd_socket connection module

be4b994

test(emqx): make peeking into connection state module-aware

28bc810

test(emqx): adapt PP2 mocks to esockd_socket 38BA listeners

3a6b020

keynslug and others added 24 commits July 23, 2025 16:19

test(emqx): cleanup emqx_client_SUITE and unbreak 2 testcases

0761af5

feat(congestion): make module generic and connmod-aware

1f9dc79

feat(socket): use async send APIs and respect send_timeout option

280ac5a

Behavior is slightly different from `emqx_connection`: there's no equivalent to `high_watermark`, so connection is never blocked on send.

chore(conn): make esockd_transport connection emit send_timeout e…

a43ee50

…rror Effectively mirroring `emqx_socket_connection` behavior.

test(conn): verify congestion and send timeout logic

63f64ec

feat(emqx): support esockd_socket backend through listener config

e1312b6

test(conn): verify connmods work well with socket close and keepalive

960f9fb

perf(frame): optimize away and inline few serialization routines

f69c877

chore(socket): fix dialyzer complaints

13a081b

chore: add changelog entry

3a3ce16

chore(socket): drop unnecessary directive

242148f

fix(socket): avoid closing already closed socket

6dcae1c

fix(socket): anticipate > 1 socket aborts on socket close

3a81374

chore(socket): correct typespec

3430943

fix(socket): preserve select handle + handle decongestion correctly

b3d1f5e

test(client): add congested-then-decongested client testcase

a724a65

chore(socket): annotate log events with connmod for observability

8bb14d9

chore(frame): avoid using 0 to signal "some more" bytes expected

7b8dcbc

chore(socket): drop dead copy-paste artifacts

e66c98c

Co-authored-by: zmstone <zmstone@gmail.com>

chore(conn): mention related emqx_socket_connection in header

9a26fb3

chore(schema): mention tcp_backend change requires listener restart

8cf1ee0

fix(schema): do not announce socket TCP backend under Windows

b255ddd

chore(socket): address dialyzer concerns

367fd4b

test(telemetry): fix testcase timeout due to dirty server state

80429fa

keynslug dismissed zmstone’s stale review via 80429fa July 23, 2025 14:20

keynslug force-pushed the perf/EMQX-14333/fanout-part-iv branch from b2eca06 to 80429fa Compare July 23, 2025 14:20

zmstone approved these changes Jul 23, 2025

View reviewed changes

keynslug merged commit c9cd26d into emqx:release-60 Jul 24, 2025
436 of 438 checks passed

keynslug deleted the perf/EMQX-14333/fanout-part-iv branch July 24, 2025 08:06

	-elvis([{elvis_style, invalid_dynamic_call, #{ignore => [emqx_connection]}}]).
	-elvis([{elvis_style, invalid_dynamic_call, #{ignore => [emqx_socket_connection]}}]).

		handle_msg({'$socket', _Socket, abort, {_Handle, Reason}}, State) ->
		handle_info({sock_error, Reason}, State);

	?TRACE("SOCKET", "emqx_connection_terminated", #{reason => Reason})
	?TRACE("SOCKET", "emqx_socket_connection_terminated", #{reason => Reason})

feat(socket): implement esockd_socket connection module #15451

feat(socket): implement esockd_socket connection module #15451

Uh oh!

Conversation

Uh oh!

Summary

PR Checklist

Uh oh!

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

feat(socket): implement `esockd_socket` connection module #15451

feat(socket): implement `esockd_socket` connection module #15451