mirror of
https://github.com/fluxerapp/fluxer
synced 2026-10-18 05:25:40 +09:00
perf(gateway): skip materialising member list subscribers (#2184)
This commit is contained in:
1 parent
57ec484626
commit
5aac331368
3 files changed
+91
-21
No files matched your search
@@ -14,10 +14,12 @@
|
||||
retain_only_session_list/3,
|
||||
unsubscribe_session/2,
|
||||
get_list_subs/2,
|
||||
has_subs/2,
|
||||
get_session_ranges/3,
|
||||
list_ids/1,
|
||||
fold_list_subs/4,
|
||||
fold_lists/3,
|
||||
fold_list_ids/3,
|
||||
foreach_list/2,
|
||||
is_subscribed/3
|
||||
]).
|
||||
@@ -121,6 +123,13 @@ lookup_list(Tab, Key) ->
|
||||
get_list_subs(ListId, Tab) ->
|
||||
collect_list(Tab, ets:next(Tab, {ListId, <<>>}), ListId, #{}).
|
||||
|
||||
-spec has_subs(binary(), ets:table()) -> boolean().
|
||||
has_subs(ListId, Tab) ->
|
||||
case ets:next(Tab, {ListId, <<>>}) of
|
||||
{ListId, SessionId} when is_binary(SessionId) -> true;
|
||||
_ -> false
|
||||
end.
|
||||
|
||||
-spec get_session_ranges(binary(), binary(), ets:table()) -> [range()].
|
||||
get_session_ranges(SessionId, ListId, Tab) ->
|
||||
case ets:lookup(Tab, {ListId, SessionId}) of
|
||||
@@ -142,6 +151,10 @@ fold_list_subs(ListId, Tab, Fun, Acc0) ->
|
||||
fold_lists(Fun, Acc0, Tab) ->
|
||||
fold_lists_loop(Fun, Acc0, Tab, ets:next(Tab, {<<>>, <<>>})).
|
||||
|
||||
-spec fold_list_ids(fun((binary(), Acc) -> Acc), Acc, ets:table()) -> Acc when Acc :: term().
|
||||
fold_list_ids(Fun, Acc0, Tab) ->
|
||||
fold_list_ids_loop(Fun, Acc0, Tab, ets:next(Tab, {<<>>, <<>>})).
|
||||
|
||||
-spec foreach_list(fun((binary(), list_subs()) -> term()), ets:table()) -> ok.
|
||||
foreach_list(Fun, Tab) ->
|
||||
_ = fold_lists(
|
||||
@@ -201,6 +214,17 @@ fold_lists_loop(Fun, Acc, Tab, {ListId, _} = Key) when is_binary(ListId) ->
|
||||
NextListKey = skip_past_list(Tab, Key, ListId),
|
||||
fold_lists_loop(Fun, Acc2, Tab, NextListKey).
|
||||
|
||||
-spec fold_list_ids_loop(fun((binary(), Acc) -> Acc), Acc, ets:table(), term()) -> Acc when
|
||||
Acc :: term().
|
||||
fold_list_ids_loop(_Fun, Acc, _Tab, '$end_of_table') ->
|
||||
Acc;
|
||||
fold_list_ids_loop(_Fun, Acc, _Tab, {idx_tab, _}) ->
|
||||
Acc;
|
||||
fold_list_ids_loop(Fun, Acc, Tab, {ListId, _} = Key) when is_binary(ListId) ->
|
||||
Acc2 = Fun(ListId, Acc),
|
||||
NextListKey = skip_past_list(Tab, Key, ListId),
|
||||
fold_list_ids_loop(Fun, Acc2, Tab, NextListKey).
|
||||
|
||||
-spec skip_past_list(ets:table(), term(), binary()) -> term().
|
||||
skip_past_list(Tab, Key, ListId) ->
|
||||
case ets:next(Tab, Key) of
|
||||
|
||||
@@ -180,10 +180,10 @@ broadcast_channel_engine_connection_change(UserId, State, SubsTab) ->
|
||||
guild_state().
|
||||
fold_connection_change_lists(GuildId, UserId, State, SubsTab) ->
|
||||
Sessions = maps:get(sessions, State, #{}),
|
||||
guild_member_list_subs:fold_lists(
|
||||
fun(ListId, ListSubs, AccState) ->
|
||||
guild_member_list_subs:fold_list_ids(
|
||||
fun(ListId, AccState) ->
|
||||
sync_connection_change_for_list(
|
||||
GuildId, UserId, ListId, ListSubs, Sessions, AccState
|
||||
GuildId, UserId, ListId, SubsTab, Sessions, AccState
|
||||
)
|
||||
end,
|
||||
State,
|
||||
@@ -200,10 +200,10 @@ fold_connection_change_lists(GuildId, UserId, State, SubsTab) ->
|
||||
dispatch_user_change_to_subscribed_lists(
|
||||
UserId, OldMember, NewMember, SubsTab, State
|
||||
) ->
|
||||
guild_member_list_subs:fold_lists(
|
||||
fun(ListId, ListSubs, AccState) ->
|
||||
guild_member_list_subs:fold_list_ids(
|
||||
fun(ListId, AccState) ->
|
||||
dispatch_user_change_to_list(
|
||||
UserId, OldMember, NewMember, ListId, ListSubs, AccState
|
||||
UserId, OldMember, NewMember, ListId, SubsTab, AccState
|
||||
)
|
||||
end,
|
||||
State,
|
||||
@@ -215,14 +215,14 @@ dispatch_user_change_to_subscribed_lists(
|
||||
map() | undefined,
|
||||
map() | undefined,
|
||||
list_id(),
|
||||
guild_member_list_subs:list_subs(),
|
||||
ets:table(),
|
||||
guild_state()
|
||||
) -> guild_state().
|
||||
dispatch_user_change_to_list(
|
||||
UserId, OldMember, NewMember, ListId, ListSubs, State
|
||||
UserId, OldMember, NewMember, ListId, SubsTab, State
|
||||
) ->
|
||||
State1 = apply_user_change_to_channel_store(UserId, ListId, OldMember, NewMember, State),
|
||||
queue_list_sync_if_subscribed(ListId, ListSubs, State1).
|
||||
queue_list_sync_if_subscribed(ListId, SubsTab, State1).
|
||||
|
||||
-spec apply_user_change_to_channel_store(
|
||||
user_id(), list_id(), map() | undefined, map() | undefined, guild_state()
|
||||
@@ -253,14 +253,14 @@ apply_channel_member_change(_UserId, _ListId, _OldMember, _NewMember, _State) ->
|
||||
integer(),
|
||||
user_id(),
|
||||
list_id(),
|
||||
guild_member_list_subs:list_subs(),
|
||||
ets:table(),
|
||||
map(),
|
||||
guild_state()
|
||||
) -> guild_state().
|
||||
sync_connection_change_for_list(_GuildId, _UserId, ListId, ListSubs, _Sessions, State) ->
|
||||
sync_connection_change_for_list(_GuildId, _UserId, ListId, SubsTab, _Sessions, State) ->
|
||||
case guild_member_list_channel_engine:is_engine_list(ListId, State) of
|
||||
true ->
|
||||
queue_list_sync_if_subscribed(ListId, ListSubs, State);
|
||||
queue_list_sync_if_subscribed(ListId, SubsTab, State);
|
||||
false ->
|
||||
State
|
||||
end.
|
||||
@@ -295,13 +295,11 @@ rebuild_channel_store(ListId, State) ->
|
||||
false -> State
|
||||
end.
|
||||
|
||||
-spec queue_list_sync_if_subscribed(
|
||||
list_id(), guild_member_list_subs:list_subs(), guild_state()
|
||||
) -> guild_state().
|
||||
queue_list_sync_if_subscribed(ListId, ListSubs, State) ->
|
||||
case map_size(ListSubs) of
|
||||
0 -> State;
|
||||
_ -> guild_member_list_sync_batch:queue_list_sync(ListId, State)
|
||||
-spec queue_list_sync_if_subscribed(list_id(), ets:table(), guild_state()) -> guild_state().
|
||||
queue_list_sync_if_subscribed(ListId, SubsTab, State) ->
|
||||
case guild_member_list_subs:has_subs(ListId, SubsTab) of
|
||||
false -> State;
|
||||
true -> guild_member_list_sync_batch:queue_list_sync(ListId, State)
|
||||
end.
|
||||
|
||||
-spec flush_pending_member_list_syncs(guild_state()) -> guild_state().
|
||||
@@ -319,9 +317,11 @@ apply_user_change_to_channel_store_updates_engine_test() ->
|
||||
?assert(maps:is_key(<<"123">>, maps:get(channel_member_list_engines, State1))).
|
||||
|
||||
sync_connection_change_ignores_zero_list_test() ->
|
||||
SubsTab = guild_member_list_subs:new(),
|
||||
?assertEqual(
|
||||
#{},
|
||||
sync_connection_change_for_list(1, 2, <<"0">>, #{}, #{}, #{})
|
||||
).
|
||||
sync_connection_change_for_list(1, 2, <<"0">>, SubsTab, #{}, #{})
|
||||
),
|
||||
guild_member_list_subs:destroy(SubsTab).
|
||||
|
||||
-endif.
|
||||
Reference in new issue
Block a user