Refactor and optimize persistent session #12251

ieQu1 · 2024-01-03T09:34:38Z

Fixes https://emqx.atlassian.net/browse/EMQX-11556

Release version: v/e5.?

Summary

Refactor code to make more invariants correct by construction.

Hide Mnesia tables behind a CRUD module. Introduce dirty session state to avoid frequent mria transactions
Factor out inflight buffer to a different module
Use separate tracks of PacketIds for QoS1 and QoS2 messages
Avoid storing pubranges, allow only one pubrange per stream
Better logical separation of concerns

Performance test

Fan-out to wildcard subscribers (message size = 0)

emqttb --pushgw --loiter 10min \
               @pub --metadata -N 1000 --qos 1 --topic foo/%n --pubinterval 100ms --size 0 \
               @sub --qos 1 --parse-metadata --topic foo/# -N 10 --expiry 30 \
               @a -a pub/pubinterval -V 10000000 --Ti 0.1 --Kp 0.0009 --pvar [scenarios,sub,{},metrics,e2e_latency] --setpoint 1000000 --min 1000 --max 100000000 \
               @g --inflight 1000 &

https://snapshots.raintank.io/dashboard/snapshot/ccmEzWR08hLfU9zdxQL9iEqG4gD6eqKC

Storage inefficiency factor:

tar cf  data.tar data
tar czf data.tar.xz data
echo "$(stat --format %s data.tar) / $(stat --format %s data.tar.xz)" | bc -l

#+RESULTS:
: 7.470356703038198

Fan-out to wildcard subscribers (message size = 1k)

https://snapshots.raintank.io/dashboard/snapshot/fg0qoTi8v3vzvzoTCcw7uGGFdJnI1dGa

PR Checklist

Please convert it to a draft if any of the following conditions are not met. Reviewers may skip over until all the items are checked:

Added tests for the changes
Added property-based tests for code which performs user input validation
Changed lines covered in coverage report
Change log has been added to changes/(ce|ee)/(feat|perf|fix|breaking)-<PR-id>.en.md files
For internal contributor: there is a jira ticket to track this change
Created PR to emqx-docs if documentation update is required, or link to a follow-up jira ticket
Schema changes are backward compatible

Checklist for CI (.github/workflows) changes

If changed package build workflow, pass this action (manual trigger)
Change log has been added to changes/ dir for user-facing artifacts update

apps/emqx/src/emqx_persistent_session_ds_state.erl

ieQu1 · 2024-01-08T20:30:03Z

apps/emqx/src/emqx_persistent_session_ds_stream_scheduler.erl

+
+ensure_iterator(TopicFilter, StartTime, SubId, {{RankX, RankY}, Stream}, S) ->
+    %% TODO: hash collisions
+    Key = {SubId, erlang:phash2(Stream)},


This is the same problem as Thales was facing... I need to come up with an idea how to store the streams without worrying about collisions.

Shouldn't we use the stream itself as the key? That was what I was doing for the DS cache. 🙈

RocksDB doesn't like maps in the keys; all match expressions will fail with badarg.

Right. The binary representation of the stream term then?

It's kinda fat though 🤔

apps/emqx/src/emqx_persistent_session_ds.hrl

thalesmg · 2024-01-11T13:12:47Z

apps/emqx/src/emqx_persistent_session_ds_stream_scheduler.erl

+
+ensure_iterator(TopicFilter, StartTime, SubId, {{RankX, RankY}, Stream}, S) ->
+    %% TODO: hash collisions
+    Key = {SubId, erlang:phash2(Stream)},


Shouldn't we use the stream itself as the key? That was what I was doing for the DS cache. 🙈

apps/emqx/src/emqx_persistent_session_ds_gc_worker.erl

apps/emqx/src/emqx_persistent_session_ds_state.erl

thalesmg · 2024-01-11T18:20:22Z

apps/emqx/src/emqx_persistent_session_ds_state.erl

+-define(created_at, created_at).
+-define(last_alive_at, last_alive_at).
+-define(conninfo, conninfo).
+%% Unique integer used to create unique identities
+-define(last_id, last_id).


Q: shouldn't these use integer values to reduce the storage burden when persisting?

Translation of a map from/to a compact representation will be done here: https://github.com/emqx/emqx/pull/12251/files#diff-4f926b90b373b839679213fbdb3c30c784f5575bc3d4b721afce21466dded77eR394 (I haven't implemented any encoder callbacks yet).

I found that integer-keyed maps are rather unpleasant when debugging, I'd like to avoid them when possible.

apps/emqx/src/emqx_persistent_session_ds_state.erl

apps/emqx/src/emqx_persistent_session_ds_stream_scheduler.erl

ieQu1 · 2024-01-12T01:48:43Z

apps/emqx_exhook/test/emqx_exhook_SUITE.erl

@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2020-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
+%% Copyright (c) 2020-2024 EMQ Technologies Co., Ltd. All Rights Reserved.


Auto-update copyright hook...
This change is accidental, but I guess it won't hurt.

apps/emqx/test/emqx_persistent_messages_SUITE.erl

ieQu1 · 2024-01-12T01:54:23Z

apps/emqx/src/emqx_schema.erl

@@ -1777,7 +1777,7 @@ fields("session_persistence") ->
            sc(
                pos_integer(),
                #{
-                    default => 1000,
+                    default => 100,


Since the introduction of the mqueue buffer for persistent sessions, it's no longer necessary to use huge batches.

apps/emqx/src/emqx_persistent_session_ds.erl

apps/emqx/src/emqx_persistent_session_ds_state.erl

changes/ce/feat-12251.en.md

apps/emqx/src/emqx_topic_gbt.erl

keynslug · 2024-01-12T13:06:09Z

apps/emqx/src/emqx_session.erl

-    case enrich_deliver(ClientInfo, D, UpgradeQoS, Session) of
-        [] ->
-            enrich_delivers(ClientInfo, Rest, UpgradeQoS, Session);
-        Msg ->
-            [Msg | enrich_delivers(ClientInfo, Rest, UpgradeQoS, Session)]
-    end.
+    enrich_deliver(ClientInfo, D, UpgradeQoS, Session) ++
+        enrich_delivers(ClientInfo, Rest, UpgradeQoS, Session).


What's the point of this change? It seems to just introduce a couple of extra heap allocations per each message.

++ is a BIF though, and we can get rid of the pattern-matching. Not sure what is more expensive in terms of micro-optimizations, but always returning a list is at least nicer from the typing perspective.

apps/emqx/src/emqx_persistent_session_ds.erl

keynslug · 2024-01-15T15:15:16Z

apps/emqx/src/emqx_persistent_session_ds.erl

+-spec update_seqno(puback | pubrec | pubcomp, emqx_types:packet_id(), session()) ->
+    {ok, emqx_types:message(), session()} | {error, _}.
+update_seqno(Track, PacketId, Session = #{id := SessionId, s := S}) ->
+    SeqNo = packet_id_to_seqno(PacketId, S),


Q: Should we reject things like PUBREC(PacketID = 42) (i.e. packet id from different track)?

apps/emqx/src/emqx_persistent_session_ds.erl

keynslug · 2024-01-15T15:31:40Z

apps/emqx/src/emqx_persistent_session_ds_state.erl

+%%
+
+transaction(Fun) ->
+    mria:async_dirty(?DS_MRIA_SHARD, Fun).


This looks risky. For consistency to be preserved the order of operations in Fun should be correct. Right now it almost certainly isn't (e.g. because not-yet-persisted state is kept in a map, streams' iterators are updated in unspecified order, etc.), so the consistency looks easy to break.

Agree that this part is a bit sketchy. However, session must NOT rely on mnesia_locker. If we need a locking mechanism to prevent multiple versions of session from clobbering each others' persistent data, it should be explicit.

It's not about concurrent updates, it's more about atomicity: async_dirty is not atomic in presence of failures I believe. It could have been much less of a problem if the operations were issued in the consistency-preserving order, but they are not.

If we intend to move the sessions to DS, we cannot rely on the atomicity of updates either. What would be the consistency preserving order here?

we cannot rely on the atomicity of updates either

Why? There are usable transactions both in rocksdb and fdb.

What would be the consistency preserving order here?

Hard to define simply enough. There might be none actually, given that more pieces of state are now tracked in separate records. Persisting ?dup markers first, then streams in the order of their respective seqno, then ?commited markers could be a good start, though I'm still unsure if holes during replay would be impossible (probably still possible), and when to persist ranks (is it reconstructible from list of ifss?).

keynslug · 2024-01-15T15:33:02Z

apps/emqx/src/emqx_persistent_session_ds.erl

+    ),
+    {ok, [], Session};
+handle_timeout(_ClientInfo, ?TIMER_BUMP_LAST_ALIVE_AT, Session0 = #{s := S0}) ->
+    S = emqx_persistent_session_ds_state:commit(bump_last_alive(S0)),


I'd expects users will want this configurable, especially when QoS2 flows are involved.

Not sure if I understand the remark. bump_interval() function inside bump_last_alive() reads the configuration. ?TIMER_BUMP_LAST_ALIVE_AT is the id of the timer event.

Sorry, I was thinking about the configurability of the circumstances under which the session state is persisted.

I specifically tried to avoid adding new config parameters in this PR, since I find constant adding and deprecation of config parameters quite annoying.

If there is real-life need to fine tune these parameters separately, we can always add a new timer/new config.

apps/emqx/src/emqx_persistent_session_ds.erl

ieQu1 · 2024-01-25T01:11:08Z

apps/emqx/integration_test/emqx_persistent_session_ds_SUITE.erl

 %%------------------------------------------------------------------------------
 %% Testcases
 %%------------------------------------------------------------------------------

-t_non_persistent_session_subscription(_Config) ->


I removed this testcase because it tests in-memory session. Not sure if it's relevant for this testsuite. Also, testcases that verify absence of certain trace events are unreliable on their own: if a trace event is renamed or removed, the test will always pass.

ieQu1 force-pushed the dev/refactor-sessds branch from ac74c03 to e1a2514 Compare January 8, 2024 11:57

ieQu1 commented Jan 8, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds_state.erl Outdated Show resolved Hide resolved

ieQu1 commented Jan 8, 2024

View reviewed changes

ieQu1 commented Jan 9, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds.hrl Outdated Show resolved Hide resolved

ieQu1 force-pushed the dev/refactor-sessds branch from e1a2514 to fe4bb00 Compare January 10, 2024 23:20

thalesmg reviewed Jan 11, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds_state.erl Show resolved Hide resolved

thalesmg reviewed Jan 11, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds_stream_scheduler.erl Outdated Show resolved Hide resolved

apps/emqx/src/emqx_persistent_session_ds_stream_scheduler.erl Show resolved Hide resolved

ieQu1 force-pushed the dev/refactor-sessds branch from 1eaa2af to 334c4e7 Compare January 12, 2024 01:43

ieQu1 commented Jan 12, 2024

View reviewed changes

apps/emqx/test/emqx_persistent_messages_SUITE.erl Show resolved Hide resolved

ieQu1 commented Jan 12, 2024

View reviewed changes

apps/emqx/test/emqx_persistent_messages_SUITE.erl Show resolved Hide resolved

ieQu1 commented Jan 12, 2024

View reviewed changes

ieQu1 marked this pull request as ready for review January 12, 2024 10:25

ieQu1 requested review from keynslug, a team, JimMoen, HJianBo and lafirest as code owners January 12, 2024 10:25

ieQu1 force-pushed the dev/refactor-sessds branch from 14f60c6 to 90ba95c Compare January 12, 2024 10:27

savonarola reviewed Jan 12, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds.erl Outdated Show resolved Hide resolved

ieQu1 force-pushed the dev/refactor-sessds branch 2 times, most recently from d732d54 to 10cb2f0 Compare January 12, 2024 11:24

ieQu1 changed the title ~~Refactor persistent session~~ Refactor and optimize persistent session Jan 12, 2024

savonarola reviewed Jan 12, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds.erl Outdated Show resolved Hide resolved

savonarola reviewed Jan 12, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds.erl Show resolved Hide resolved

savonarola reviewed Jan 12, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds_state.erl Outdated Show resolved Hide resolved

ieQu1 force-pushed the dev/refactor-sessds branch 2 times, most recently from 25e93ff to d725db4 Compare January 12, 2024 21:50

keynslug reviewed Jan 15, 2024

View reviewed changes

thalesmg previously approved these changes Jan 24, 2024

View reviewed changes

apps/emqx/src/emqx_persistent_session_ds.erl Show resolved Hide resolved

apps/emqx/src/emqx_persistent_session_ds.erl Outdated Show resolved Hide resolved

ieQu1 dismissed thalesmg’s stale review via e1f7d02 January 25, 2024 01:08

ieQu1 force-pushed the dev/refactor-sessds branch from 2681866 to e1f7d02 Compare January 25, 2024 01:08

ieQu1 commented Jan 25, 2024

View reviewed changes

thalesmg previously approved these changes Jan 26, 2024

View reviewed changes

ieQu1 added 19 commits January 26, 2024 17:49

refactor(ds): Create a CRUD module for the persistent session

2d08aa8

fix(sessds): Refactor emqx_persistent_session_ds to use CRUD module

8e8d3af

test(sessds): Add property-based tests for seqno generator

3fb2064

feat(sessds): Index streams by a unique subid

82ef349

refactor(sessds): Factor out stream scheduler into its own module

4f4831f

refactor(sessds): Simplify data structure of ds_state pmap datatype

1b4f69b

refactor(sessds): Simplify representation of QoS tracks

978a3bf

fix(sessds): Store the QoS as the MSB of the packet ID

cff6c15

fix(sessds): Delete the routes when the session expires

963df8f

fix(sessds): Fix session garbage collection after the refactoring

893656f

test(sessds): Create a property-based test for the session state

e7b03cd

test(sessds): Fix failing tests

3985762

fix(sessds): Apply review remarks

f5b9bd3

fix(sessds): Use mria:async_dirty instead of transaction

ebe2339

test(sessds): Don't trap exits in the test

974760d

refactor(sessds): #ifs -> #srs

2d23212

test(sessds): Fix unstable tests

3c451c6

fix(sessds): Improve comments

eec56b0

fix(sessds): Stricter checks for PacketIds

2c6a776

ieQu1 dismissed thalesmg’s stale review via 2c6a776 January 26, 2024 16:50

ieQu1 force-pushed the dev/refactor-sessds branch from e1f7d02 to 2c6a776 Compare January 26, 2024 16:50

thalesmg approved these changes Jan 26, 2024

View reviewed changes

ieQu1 merged commit 9e0bea0 into emqx:master Jan 26, 2024
162 of 164 checks passed

ieQu1 mentioned this pull request Feb 2, 2024

fix(sessds): drop streams related to deleted subscriptions #12127

Closed

7 tasks

BrewTestBot mentioned this pull request Mar 28, 2024

emqx 5.6.0 Homebrew/homebrew-core#167392

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor and optimize persistent session #12251

Refactor and optimize persistent session #12251

ieQu1 commented Jan 3, 2024 •

edited

Loading

ieQu1 Jan 8, 2024

thalesmg Jan 11, 2024

ieQu1 Jan 12, 2024 •

edited

Loading

thalesmg Jan 16, 2024

ieQu1 Jan 24, 2024

thalesmg Jan 11, 2024

thalesmg Jan 11, 2024

ieQu1 Jan 12, 2024 •

edited

Loading

ieQu1 Jan 12, 2024

ieQu1 Jan 12, 2024

keynslug Jan 12, 2024

ieQu1 Jan 15, 2024 •

edited

Loading

keynslug Jan 15, 2024

keynslug Jan 15, 2024

ieQu1 Jan 16, 2024

keynslug Jan 16, 2024

ieQu1 Jan 16, 2024

keynslug Jan 16, 2024

keynslug Jan 15, 2024

ieQu1 Jan 15, 2024

keynslug Jan 16, 2024

ieQu1 Jan 16, 2024

ieQu1 Jan 25, 2024

Refactor and optimize persistent session #12251

Refactor and optimize persistent session #12251

Conversation

ieQu1 commented Jan 3, 2024 • edited Loading

Summary

Performance test

Fan-out to wildcard subscribers (message size = 0)

Fan-out to wildcard subscribers (message size = 1k)

PR Checklist

Checklist for CI (.github/workflows) changes

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ieQu1 Jan 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ieQu1 Jan 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ieQu1 Jan 15, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ieQu1 commented Jan 3, 2024 •

edited

Loading

ieQu1 Jan 12, 2024 •

edited

Loading

ieQu1 Jan 12, 2024 •

edited

Loading

ieQu1 Jan 15, 2024 •

edited

Loading