yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #11183 from keynslug/perf/EMQX-9591/routing 

perf(broker): avoid usorting aggregated routes in a loop
This commit is contained in:
Andrew Mayorov 2023-07-03 19:16:25 +02:00 committed by GitHub
parent 1cc3681d23 0f25df3aa8
commit a127fd6571
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 12 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
apps/emqx/src/emqx_broker.erl
View File

@ -285,16 +285,16 @@ aggre([#route{topic = To, dest = Node}]) when is_atom(Node) ->
aggre([#route{topic = To, dest = {Group, _Node}}]) -> aggre([#route{topic = To, dest = {Group, _Node}}]) ->
[{To, Group}]; [{To, Group}];
aggre(Routes) -> aggre(Routes) ->
lists:foldl( aggre(Routes, false, []).
fun
(#route{topic = To, dest = Node}, Acc) when is_atom(Node) -> aggre([#route{topic = To, dest = Node} | Rest], Dedup, Acc) when is_atom(Node) ->
[{To, Node} | Acc]; aggre(Rest, Dedup, [{To, Node} | Acc]);
(#route{topic = To, dest = {Group, _Node}}, Acc) -> aggre([#route{topic = To, dest = {Group, _Node}} | Rest], _Dedup, Acc) ->
lists:usort([{To, Group} | Acc]) aggre(Rest, true, [{To, Group} | Acc]);
end, aggre([], false, Acc) ->
[], Acc;
Routes aggre([], true, Acc) ->
). lists:usort(Acc).
%% @doc Forward message to another node. %% @doc Forward message to another node.
-spec forward(node(), emqx_types:topic(), emqx_types:delivery(), RpcMode :: sync | async) -> -spec forward(node(), emqx_types:topic(), emqx_types:delivery(), RpcMode :: sync | async) ->

4
apps/emqx/src/emqx_session_router.erl
View File

@ -243,8 +243,8 @@ handle_call(Req, _From, State) ->
handle_cast({delete_routes, SessionID, Subscriptions}, State) -> handle_cast({delete_routes, SessionID, Subscriptions}, State) ->
%% TODO: Make a batch for deleting all routes. %% TODO: Make a batch for deleting all routes.
Fun = fun({Topic, _}) -> do_delete_route(Topic, SessionID) end, Fun = fun(Topic, _) -> do_delete_route(Topic, SessionID) end,
ok = lists:foreach(Fun, maps:to_list(Subscriptions)), ok = maps:foreach(Fun, Subscriptions),
{noreply, State}; {noreply, State};
handle_cast({resume_end, SessionID, Pid}, State) -> handle_cast({resume_end, SessionID, Pid}, State) ->
case emqx_utils_ets:lookup_value(?SESSION_INIT_TAB, SessionID) of case emqx_utils_ets:lookup_value(?SESSION_INIT_TAB, SessionID) of