From 28dad5d7a91a57d7ac06d8337b6e0d116531d468 Mon Sep 17 00:00:00 2001 From: Andrew Mayorov Date: Tue, 18 Jul 2023 23:11:04 +0200 Subject: [PATCH 1/4] feat(index): add topic index facility Somewhat similar to `emqx_trie` in design and logic, yet built on top of a single, potentially pre-existing table. --- apps/emqx/src/emqx_topic_index.erl | 156 ++++++++++++++++++++++ apps/emqx/test/emqx_topic_index_SUITE.erl | 143 ++++++++++++++++++++ 2 files changed, 299 insertions(+) create mode 100644 apps/emqx/src/emqx_topic_index.erl create mode 100644 apps/emqx/test/emqx_topic_index_SUITE.erl diff --git a/apps/emqx/src/emqx_topic_index.erl b/apps/emqx/src/emqx_topic_index.erl new file mode 100644 index 000000000..9f0b5fba1 --- /dev/null +++ b/apps/emqx/src/emqx_topic_index.erl @@ -0,0 +1,156 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%%-------------------------------------------------------------------- + +%% @doc Topic index for matching topics to topic filters. +%% +%% Works on top of ETS ordered_set table. Keys are parsed topic filters +%% with record ID appended to the end, wrapped in a tuple to disambiguate from +%% topic filter words. Existing table may be used if existing keys will not +%% collide with index keys. +%% +%% Designed to effectively answer questions like: +%% 1. Does any topic filter match given topic? +%% 2. Which records are associated with topic filters matching given topic? +%% +%% Questions like these are _only slightly_ less effective: +%% 1. Which topic filters match given topic? +%% 2. Which record IDs are associated with topic filters matching given topic? + +-module(emqx_topic_index). + +-export([new/0]). +-export([insert/4]). +-export([delete/3]). +-export([match/2]). +-export([matches/2]). + +-export([get_id/1]). +-export([get_topic/1]). +-export([get_record/2]). + +-type key(ID) :: [binary() | '+' | '#' | {ID}]. +-type match(ID) :: key(ID). + +new() -> + ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]). + +insert(Filter, ID, Record, Tab) -> + ets:insert(Tab, {emqx_topic:words(Filter) ++ [{ID}], Record}). + +delete(Filter, ID, Tab) -> + ets:delete(Tab, emqx_topic:words(Filter) ++ [{ID}]). + +-spec match(emqx_types:topic(), ets:table()) -> match(_ID) | false. +match(Topic, Tab) -> + {Words, RPrefix} = match_init(Topic), + match(Words, RPrefix, Tab). + +match(Words, RPrefix, Tab) -> + Prefix = lists:reverse(RPrefix), + K = ets:next(Tab, Prefix), + case match_filter(Prefix, K, Words =/= []) of + true -> + K; + stop -> + false; + Matched -> + match_rest(Matched, Words, RPrefix, Tab) + end. + +match_rest(false, [W | Rest], RPrefix, Tab) -> + match(Rest, [W | RPrefix], Tab); +match_rest(plus, [W | Rest], RPrefix, Tab) -> + case match(Rest, ['+' | RPrefix], Tab) of + Match when is_list(Match) -> + Match; + false -> + match(Rest, [W | RPrefix], Tab) + end; +match_rest(_, [], _RPrefix, _Tab) -> + false. + +-spec matches(emqx_types:topic(), ets:table()) -> [match(_ID)]. +matches(Topic, Tab) -> + {Words, RPrefix} = match_init(Topic), + matches(Words, RPrefix, Tab). + +matches(Words, RPrefix, Tab) -> + Prefix = lists:reverse(RPrefix), + matches(ets:next(Tab, Prefix), Prefix, Words, RPrefix, Tab). + +matches(K, Prefix, Words, RPrefix, Tab) -> + case match_filter(Prefix, K, Words =/= []) of + true -> + [K | matches(ets:next(Tab, K), Prefix, Words, RPrefix, Tab)]; + stop -> + []; + Matched -> + matches_rest(Matched, Words, RPrefix, Tab) + end. + +matches_rest(false, [W | Rest], RPrefix, Tab) -> + matches(Rest, [W | RPrefix], Tab); +matches_rest(plus, [W | Rest], RPrefix, Tab) -> + matches(Rest, ['+' | RPrefix], Tab) ++ matches(Rest, [W | RPrefix], Tab); +matches_rest(_, [], _RPrefix, _Tab) -> + []. + +match_filter([], [{_ID}], _IsPrefix = false) -> + % NOTE: exact match is `true` only if we match whole topic, not prefix + true; +match_filter([], ['#', {_ID}], _IsPrefix) -> + % NOTE: naturally, '#' < '+', so this is already optimal for `match/2` + true; +match_filter([], ['+' | _], _) -> + plus; +match_filter([], [_H | _], _) -> + false; +match_filter([H | T1], [H | T2], IsPrefix) -> + match_filter(T1, T2, IsPrefix); +match_filter([H1 | _], [H2 | _], _) when H2 > H1 -> + % NOTE: we're strictly past the prefix, no need to continue + stop; +match_filter(_, '$end_of_table', _) -> + stop. + +match_init(Topic) -> + case emqx_topic:words(Topic) of + [W = <<"$", _/bytes>> | Rest] -> + % NOTE + % This will effectively skip attempts to match special topics to `#` or `+/...`. + {Rest, [W]}; + Words -> + {Words, []} + end. + +-spec get_id(match(ID)) -> ID. +get_id([{ID}]) -> + ID; +get_id([_ | Rest]) -> + get_id(Rest). + +-spec get_topic(match(_ID)) -> emqx_types:topic(). +get_topic(K) -> + emqx_topic:join(cut_topic(K)). + +cut_topic([{_ID}]) -> + []; +cut_topic([W | Rest]) -> + [W | cut_topic(Rest)]. + +-spec get_record(match(_ID), ets:table()) -> _Record. +get_record(K, Tab) -> + ets:lookup_element(Tab, K, 2). diff --git a/apps/emqx/test/emqx_topic_index_SUITE.erl b/apps/emqx/test/emqx_topic_index_SUITE.erl new file mode 100644 index 000000000..98bfe48a1 --- /dev/null +++ b/apps/emqx/test/emqx_topic_index_SUITE.erl @@ -0,0 +1,143 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%%-------------------------------------------------------------------- + +-module(emqx_topic_index_SUITE). + +-compile(export_all). +-compile(nowarn_export_all). + +-include_lib("eunit/include/eunit.hrl"). + +all() -> + emqx_common_test_helpers:all(?MODULE). + +t_insert(_) -> + Tab = emqx_topic_index:new(), + true = emqx_topic_index:insert(<<"sensor/1/metric/2">>, t_insert_1, <<>>, Tab), + true = emqx_topic_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab), + true = emqx_topic_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab), + ?assertEqual(<<"sensor/#">>, topic(match(<<"sensor">>, Tab))), + ?assertEqual(t_insert_3, id(match(<<"sensor">>, Tab))). + +t_match(_) -> + Tab = emqx_topic_index:new(), + true = emqx_topic_index:insert(<<"sensor/1/metric/2">>, t_match_1, <<>>, Tab), + true = emqx_topic_index:insert(<<"sensor/+/#">>, t_match_2, <<>>, Tab), + true = emqx_topic_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab), + ?assertMatch( + [<<"sensor/#">>, <<"sensor/+/#">>], + [topic(M) || M <- matches(<<"sensor/1">>, Tab)] + ). + +t_match2(_) -> + Tab = emqx_topic_index:new(), + true = emqx_topic_index:insert(<<"#">>, t_match2_1, <<>>, Tab), + true = emqx_topic_index:insert(<<"+/#">>, t_match2_2, <<>>, Tab), + true = emqx_topic_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab), + ?assertEqual( + [<<"#">>, <<"+/#">>, <<"+/+/#">>], + [topic(M) || M <- matches(<<"a/b/c">>, Tab)] + ), + ?assertEqual( + false, + emqx_topic_index:match(<<"$SYS/broker/zenmq">>, Tab) + ). + +t_match3(_) -> + Tab = emqx_topic_index:new(), + Records = [ + {<<"d/#">>, t_match3_1}, + {<<"a/b/+">>, t_match3_2}, + {<<"a/#">>, t_match3_3}, + {<<"#">>, t_match3_4}, + {<<"$SYS/#">>, t_match3_sys} + ], + lists:foreach( + fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end, + Records + ), + Matched = matches(<<"a/b/c">>, Tab), + case length(Matched) of + 3 -> ok; + _ -> error({unexpected, Matched}) + end, + ?assertEqual( + t_match3_sys, + id(match(<<"$SYS/a/b/c">>, Tab)) + ). + +t_match4(_) -> + Tab = emqx_topic_index:new(), + Records = [{<<"/#">>, t_match4_1}, {<<"/+">>, t_match4_2}, {<<"/+/a/b/c">>, t_match4_3}], + lists:foreach( + fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end, + Records + ), + ?assertEqual( + [<<"/#">>, <<"/+">>], + [topic(M) || M <- matches(<<"/">>, Tab)] + ), + ?assertEqual( + [<<"/#">>, <<"/+/a/b/c">>], + [topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)] + ). + +t_match5(_) -> + Tab = emqx_topic_index:new(), + T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>, + Records = [ + {<<"#">>, t_match5_1}, + {<>, t_match5_2}, + {<>, t_match5_3} + ], + lists:foreach( + fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end, + Records + ), + ?assertEqual( + [<<"#">>, <>], + [topic(M) || M <- matches(T, Tab)] + ), + ?assertEqual( + [<<"#">>, <>, <>], + [topic(M) || M <- matches(<>, Tab)] + ). + +t_match6(_) -> + Tab = emqx_topic_index:new(), + T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>, + W = <<"+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/#">>, + emqx_topic_index:insert(W, ID = t_match6, <<>>, Tab), + ?assertEqual(ID, id(match(T, Tab))). + +t_match7(_) -> + Tab = emqx_topic_index:new(), + T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>, + W = <<"a/+/c/+/e/+/g/+/i/+/k/+/m/+/o/+/q/+/s/+/u/+/w/+/y/+/#">>, + emqx_topic_index:insert(W, t_match7, <<>>, Tab), + ?assertEqual(W, topic(match(T, Tab))). + +match(T, Tab) -> + emqx_topic_index:match(T, Tab). + +matches(T, Tab) -> + lists:sort(emqx_topic_index:matches(T, Tab)). + +id(Match) -> + emqx_topic_index:get_id(Match). + +topic(Match) -> + emqx_topic_index:get_topic(Match). From b821bdee00aea7cb1014f57388a643c365b8ad25 Mon Sep 17 00:00:00 2001 From: Andrew Mayorov Date: Thu, 20 Jul 2023 14:49:10 +0200 Subject: [PATCH 2/4] perf(ruleeng): employ `emqx_topic_index` to speed up topic matching --- apps/emqx_rule_engine/include/rule_engine.hrl | 1 + .../emqx_rule_engine/src/emqx_rule_engine.erl | 68 +++++++++++++------ .../src/emqx_rule_engine_app.erl | 1 + 3 files changed, 49 insertions(+), 21 deletions(-) diff --git a/apps/emqx_rule_engine/include/rule_engine.hrl b/apps/emqx_rule_engine/include/rule_engine.hrl index b2a6a549e..7df5d9941 100644 --- a/apps/emqx_rule_engine/include/rule_engine.hrl +++ b/apps/emqx_rule_engine/include/rule_engine.hrl @@ -109,6 +109,7 @@ %% Tables -define(RULE_TAB, emqx_rule_engine). +-define(RULE_TOPIC_INDEX, emqx_rule_engine_topic_index). %% Allowed sql function provider modules -define(DEFAULT_SQL_FUNC_PROVIDER, emqx_rule_funcs). diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine.erl b/apps/emqx_rule_engine/src/emqx_rule_engine.erl index 66c82d3a1..9d2d918ae 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine.erl +++ b/apps/emqx_rule_engine/src/emqx_rule_engine.erl @@ -176,7 +176,7 @@ create_rule(Params) -> create_rule(Params = #{id := RuleId}, CreatedAt) when is_binary(RuleId) -> case get_rule(RuleId) of - not_found -> parse_and_insert(Params, CreatedAt); + not_found -> with_parsed_rule(Params, CreatedAt, fun insert_rule/1); {ok, _} -> {error, already_exists} end. @@ -185,18 +185,27 @@ update_rule(Params = #{id := RuleId}) when is_binary(RuleId) -> case get_rule(RuleId) of not_found -> {error, not_found}; - {ok, #{created_at := CreatedAt}} -> - parse_and_insert(Params, CreatedAt) + {ok, RulePrev = #{created_at := CreatedAt}} -> + with_parsed_rule(Params, CreatedAt, fun(Rule) -> update_rule(Rule, RulePrev) end) end. -spec delete_rule(RuleId :: rule_id()) -> ok. delete_rule(RuleId) when is_binary(RuleId) -> - gen_server:call(?RULE_ENGINE, {delete_rule, RuleId}, ?T_CALL). + case get_rule(RuleId) of + not_found -> + ok; + {ok, Rule} -> + gen_server:call(?RULE_ENGINE, {delete_rule, Rule}, ?T_CALL) + end. -spec insert_rule(Rule :: rule()) -> ok. insert_rule(Rule) -> gen_server:call(?RULE_ENGINE, {insert_rule, Rule}, ?T_CALL). +-spec update_rule(Rule :: rule(), RulePrev :: rule()) -> ok. +update_rule(Rule, RulePrev) -> + gen_server:call(?RULE_ENGINE, {update_rule, Rule, RulePrev}, ?T_CALL). + %%---------------------------------------------------------------------------------------- %% Rule Management %%---------------------------------------------------------------------------------------- @@ -216,9 +225,8 @@ get_rules_ordered_by_ts() -> -spec get_rules_for_topic(Topic :: binary()) -> [rule()]. get_rules_for_topic(Topic) -> [ - Rule - || Rule = #{from := From} <- get_rules(), - emqx_topic:match_any(Topic, From) + emqx_topic_index:get_record(M, ?RULE_TOPIC_INDEX) + || M <- emqx_topic_index:matches(Topic, ?RULE_TOPIC_INDEX) ]. -spec get_rules_with_same_event(Topic :: binary()) -> [rule()]. @@ -411,10 +419,17 @@ init([]) -> {ok, #{}}. handle_call({insert_rule, Rule}, _From, State) -> - do_insert_rule(Rule), + ok = do_insert_rule(Rule), + ok = do_update_rule_index(Rule), + {reply, ok, State}; +handle_call({update_rule, Rule, RulePrev}, _From, State) -> + ok = do_delete_rule_index(RulePrev), + ok = do_insert_rule(Rule), + ok = do_update_rule_index(Rule), {reply, ok, State}; handle_call({delete_rule, Rule}, _From, State) -> - do_delete_rule(Rule), + ok = do_delete_rule_index(Rule), + ok = do_delete_rule(Rule), {reply, ok, State}; handle_call(Req, _From, State) -> ?SLOG(error, #{msg => "unexpected_call", request => Req}), @@ -438,7 +453,7 @@ code_change(_OldVsn, State, _Extra) -> %% Internal Functions %%---------------------------------------------------------------------------------------- -parse_and_insert(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt) -> +with_parsed_rule(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt, Fun) -> case emqx_rule_sqlparser:parse(Sql) of {ok, Select} -> Rule = #{ @@ -459,7 +474,7 @@ parse_and_insert(Params = #{id := RuleId, sql := Sql, actions := Actions}, Creat conditions => emqx_rule_sqlparser:select_where(Select) %% -- calculated fields end }, - ok = insert_rule(Rule), + ok = Fun(Rule), {ok, Rule}; {error, Reason} -> {error, Reason} @@ -471,16 +486,27 @@ do_insert_rule(#{id := Id} = Rule) -> true = ets:insert(?RULE_TAB, {Id, maps:remove(id, Rule)}), ok. -do_delete_rule(RuleId) -> - case get_rule(RuleId) of - {ok, Rule} -> - ok = unload_hooks_for_rule(Rule), - ok = clear_metrics_for_rule(RuleId), - true = ets:delete(?RULE_TAB, RuleId), - ok; - not_found -> - ok - end. +do_delete_rule(#{id := Id} = Rule) -> + ok = unload_hooks_for_rule(Rule), + ok = clear_metrics_for_rule(Id), + true = ets:delete(?RULE_TAB, Id), + ok. + +do_update_rule_index(#{id := Id, from := From} = Rule) -> + ok = lists:foreach( + fun(Topic) -> + true = emqx_topic_index:insert(Topic, Id, Rule, ?RULE_TOPIC_INDEX) + end, + From + ). + +do_delete_rule_index(#{id := Id, from := From}) -> + ok = lists:foreach( + fun(Topic) -> + true = emqx_topic_index:delete(Topic, Id, ?RULE_TOPIC_INDEX) + end, + From + ). parse_actions(Actions) -> [do_parse_action(Act) || Act <- Actions]. diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine_app.erl b/apps/emqx_rule_engine/src/emqx_rule_engine_app.erl index d8b031bdd..28515cb1a 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine_app.erl +++ b/apps/emqx_rule_engine/src/emqx_rule_engine_app.erl @@ -26,6 +26,7 @@ start(_Type, _Args) -> _ = ets:new(?RULE_TAB, [named_table, public, ordered_set, {read_concurrency, true}]), + _ = ets:new(?RULE_TOPIC_INDEX, [named_table, public, ordered_set, {read_concurrency, true}]), ok = emqx_rule_events:reload(), SupRet = emqx_rule_engine_sup:start_link(), ok = emqx_rule_engine:load_rules(), From 4e4b1ac11570b862835203b722c988f1dde6238f Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 21 Jul 2023 17:58:54 +0800 Subject: [PATCH 3/4] refactor: module move to app emqx_rule_engine - Rename to emqx_rule_index.erl - Remove test funcs from src -> test dir --- .../emqx_rule_engine/src/emqx_rule_engine.erl | 11 ++- .../src/emqx_rule_index.erl} | 26 +---- .../test/emqx_rule_index_SUITE.erl} | 97 +++++++++++-------- changes/ce/perf-11282.en.md | 1 + 4 files changed, 69 insertions(+), 66 deletions(-) rename apps/{emqx/src/emqx_topic_index.erl => emqx_rule_engine/src/emqx_rule_index.erl} (90%) rename apps/{emqx/test/emqx_topic_index_SUITE.erl => emqx_rule_engine/test/emqx_rule_index_SUITE.erl} (52%) create mode 100644 changes/ce/perf-11282.en.md diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine.erl b/apps/emqx_rule_engine/src/emqx_rule_engine.erl index 9d2d918ae..d92931d77 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine.erl +++ b/apps/emqx_rule_engine/src/emqx_rule_engine.erl @@ -225,12 +225,13 @@ get_rules_ordered_by_ts() -> -spec get_rules_for_topic(Topic :: binary()) -> [rule()]. get_rules_for_topic(Topic) -> [ - emqx_topic_index:get_record(M, ?RULE_TOPIC_INDEX) - || M <- emqx_topic_index:matches(Topic, ?RULE_TOPIC_INDEX) + emqx_rule_index:get_record(M, ?RULE_TOPIC_INDEX) + || M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX) ]. -spec get_rules_with_same_event(Topic :: binary()) -> [rule()]. get_rules_with_same_event(Topic) -> + %% TODO: event matching index not implemented yet EventName = emqx_rule_events:event_name(Topic), [ Rule @@ -240,6 +241,7 @@ get_rules_with_same_event(Topic) -> -spec get_rule_ids_by_action(action_name()) -> [rule_id()]. get_rule_ids_by_action(BridgeId) when is_binary(BridgeId) -> + %% TODO: bridge matching index not implemented yet [ Id || #{actions := Acts, id := Id, from := Froms} <- get_rules(), @@ -247,6 +249,7 @@ get_rule_ids_by_action(BridgeId) when is_binary(BridgeId) -> references_ingress_bridge(Froms, BridgeId) ]; get_rule_ids_by_action(#{function := FuncName}) when is_binary(FuncName) -> + %% TODO: action id matching index not implemented yet {Mod, Fun} = case string:split(FuncName, ":", leading) of [M, F] -> {binary_to_module(M), F}; @@ -495,7 +498,7 @@ do_delete_rule(#{id := Id} = Rule) -> do_update_rule_index(#{id := Id, from := From} = Rule) -> ok = lists:foreach( fun(Topic) -> - true = emqx_topic_index:insert(Topic, Id, Rule, ?RULE_TOPIC_INDEX) + true = emqx_rule_index:insert(Topic, Id, Rule, ?RULE_TOPIC_INDEX) end, From ). @@ -503,7 +506,7 @@ do_update_rule_index(#{id := Id, from := From} = Rule) -> do_delete_rule_index(#{id := Id, from := From}) -> ok = lists:foreach( fun(Topic) -> - true = emqx_topic_index:delete(Topic, Id, ?RULE_TOPIC_INDEX) + true = emqx_rule_index:delete(Topic, Id, ?RULE_TOPIC_INDEX) end, From ). diff --git a/apps/emqx/src/emqx_topic_index.erl b/apps/emqx_rule_engine/src/emqx_rule_index.erl similarity index 90% rename from apps/emqx/src/emqx_topic_index.erl rename to apps/emqx_rule_engine/src/emqx_rule_index.erl index 9f0b5fba1..9c16bf3ad 100644 --- a/apps/emqx/src/emqx_topic_index.erl +++ b/apps/emqx_rule_engine/src/emqx_rule_index.erl @@ -29,25 +29,24 @@ %% 1. Which topic filters match given topic? %% 2. Which record IDs are associated with topic filters matching given topic? --module(emqx_topic_index). +-module(emqx_rule_index). --export([new/0]). -export([insert/4]). -export([delete/3]). -export([match/2]). -export([matches/2]). --export([get_id/1]). --export([get_topic/1]). -export([get_record/2]). -type key(ID) :: [binary() | '+' | '#' | {ID}]. -type match(ID) :: key(ID). -new() -> - ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]). +-ifdef(TEST). +-export_type([match/1]). +-endif. insert(Filter, ID, Record, Tab) -> + %% TODO: topic compact. see also in emqx_trie.erl ets:insert(Tab, {emqx_topic:words(Filter) ++ [{ID}], Record}). delete(Filter, ID, Tab) -> @@ -136,21 +135,6 @@ match_init(Topic) -> {Words, []} end. --spec get_id(match(ID)) -> ID. -get_id([{ID}]) -> - ID; -get_id([_ | Rest]) -> - get_id(Rest). - --spec get_topic(match(_ID)) -> emqx_types:topic(). -get_topic(K) -> - emqx_topic:join(cut_topic(K)). - -cut_topic([{_ID}]) -> - []; -cut_topic([W | Rest]) -> - [W | cut_topic(Rest)]. - -spec get_record(match(_ID), ets:table()) -> _Record. get_record(K, Tab) -> ets:lookup_element(Tab, K, 2). diff --git a/apps/emqx/test/emqx_topic_index_SUITE.erl b/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl similarity index 52% rename from apps/emqx/test/emqx_topic_index_SUITE.erl rename to apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl index 98bfe48a1..c4b1b4848 100644 --- a/apps/emqx/test/emqx_topic_index_SUITE.erl +++ b/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl @@ -14,7 +14,7 @@ %% limitations under the License. %%-------------------------------------------------------------------- --module(emqx_topic_index_SUITE). +-module(emqx_rule_index_SUITE). -compile(export_all). -compile(nowarn_export_all). @@ -25,39 +25,39 @@ all() -> emqx_common_test_helpers:all(?MODULE). t_insert(_) -> - Tab = emqx_topic_index:new(), - true = emqx_topic_index:insert(<<"sensor/1/metric/2">>, t_insert_1, <<>>, Tab), - true = emqx_topic_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab), - true = emqx_topic_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab), - ?assertEqual(<<"sensor/#">>, topic(match(<<"sensor">>, Tab))), - ?assertEqual(t_insert_3, id(match(<<"sensor">>, Tab))). + Tab = new(), + true = emqx_rule_index:insert(<<"sensor/1/metric/2">>, t_insert_1, <<>>, Tab), + true = emqx_rule_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab), + true = emqx_rule_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab), + ?assertEqual(<<"sensor/#">>, get_topic(match(<<"sensor">>, Tab))), + ?assertEqual(t_insert_3, get_id(match(<<"sensor">>, Tab))). t_match(_) -> - Tab = emqx_topic_index:new(), - true = emqx_topic_index:insert(<<"sensor/1/metric/2">>, t_match_1, <<>>, Tab), - true = emqx_topic_index:insert(<<"sensor/+/#">>, t_match_2, <<>>, Tab), - true = emqx_topic_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab), + Tab = new(), + true = emqx_rule_index:insert(<<"sensor/1/metric/2">>, t_match_1, <<>>, Tab), + true = emqx_rule_index:insert(<<"sensor/+/#">>, t_match_2, <<>>, Tab), + true = emqx_rule_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab), ?assertMatch( [<<"sensor/#">>, <<"sensor/+/#">>], - [topic(M) || M <- matches(<<"sensor/1">>, Tab)] + [get_topic(M) || M <- matches(<<"sensor/1">>, Tab)] ). t_match2(_) -> - Tab = emqx_topic_index:new(), - true = emqx_topic_index:insert(<<"#">>, t_match2_1, <<>>, Tab), - true = emqx_topic_index:insert(<<"+/#">>, t_match2_2, <<>>, Tab), - true = emqx_topic_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab), + Tab = new(), + true = emqx_rule_index:insert(<<"#">>, t_match2_1, <<>>, Tab), + true = emqx_rule_index:insert(<<"+/#">>, t_match2_2, <<>>, Tab), + true = emqx_rule_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab), ?assertEqual( [<<"#">>, <<"+/#">>, <<"+/+/#">>], - [topic(M) || M <- matches(<<"a/b/c">>, Tab)] + [get_topic(M) || M <- matches(<<"a/b/c">>, Tab)] ), ?assertEqual( false, - emqx_topic_index:match(<<"$SYS/broker/zenmq">>, Tab) + emqx_rule_index:match(<<"$SYS/broker/zenmq">>, Tab) ). t_match3(_) -> - Tab = emqx_topic_index:new(), + Tab = new(), Records = [ {<<"d/#">>, t_match3_1}, {<<"a/b/+">>, t_match3_2}, @@ -66,7 +66,7 @@ t_match3(_) -> {<<"$SYS/#">>, t_match3_sys} ], lists:foreach( - fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end, + fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end, Records ), Matched = matches(<<"a/b/c">>, Tab), @@ -76,27 +76,27 @@ t_match3(_) -> end, ?assertEqual( t_match3_sys, - id(match(<<"$SYS/a/b/c">>, Tab)) + get_id(match(<<"$SYS/a/b/c">>, Tab)) ). t_match4(_) -> - Tab = emqx_topic_index:new(), + Tab = new(), Records = [{<<"/#">>, t_match4_1}, {<<"/+">>, t_match4_2}, {<<"/+/a/b/c">>, t_match4_3}], lists:foreach( - fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end, + fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end, Records ), ?assertEqual( [<<"/#">>, <<"/+">>], - [topic(M) || M <- matches(<<"/">>, Tab)] + [get_topic(M) || M <- matches(<<"/">>, Tab)] ), ?assertEqual( [<<"/#">>, <<"/+/a/b/c">>], - [topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)] + [get_topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)] ). t_match5(_) -> - Tab = emqx_topic_index:new(), + Tab = new(), T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>, Records = [ {<<"#">>, t_match5_1}, @@ -104,40 +104,55 @@ t_match5(_) -> {<>, t_match5_3} ], lists:foreach( - fun({Topic, ID}) -> emqx_topic_index:insert(Topic, ID, <<>>, Tab) end, + fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end, Records ), ?assertEqual( [<<"#">>, <>], - [topic(M) || M <- matches(T, Tab)] + [get_topic(M) || M <- matches(T, Tab)] ), ?assertEqual( [<<"#">>, <>, <>], - [topic(M) || M <- matches(<>, Tab)] + [get_topic(M) || M <- matches(<>, Tab)] ). t_match6(_) -> - Tab = emqx_topic_index:new(), + Tab = new(), T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>, W = <<"+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/#">>, - emqx_topic_index:insert(W, ID = t_match6, <<>>, Tab), - ?assertEqual(ID, id(match(T, Tab))). + emqx_rule_index:insert(W, ID = t_match6, <<>>, Tab), + ?assertEqual(ID, get_id(match(T, Tab))). t_match7(_) -> - Tab = emqx_topic_index:new(), + Tab = new(), T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>, W = <<"a/+/c/+/e/+/g/+/i/+/k/+/m/+/o/+/q/+/s/+/u/+/w/+/y/+/#">>, - emqx_topic_index:insert(W, t_match7, <<>>, Tab), - ?assertEqual(W, topic(match(T, Tab))). + emqx_rule_index:insert(W, t_match7, <<>>, Tab), + ?assertEqual(W, get_topic(match(T, Tab))). + +new() -> + ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]). match(T, Tab) -> - emqx_topic_index:match(T, Tab). + emqx_rule_index:match(T, Tab). matches(T, Tab) -> - lists:sort(emqx_topic_index:matches(T, Tab)). + lists:sort(emqx_rule_index:matches(T, Tab)). -id(Match) -> - emqx_topic_index:get_id(Match). +-spec get_id(emqx_rule_index:match(ID)) -> ID. +get_id([{ID}]) -> + ID; +get_id([_ | Rest]) -> + get_id(Rest). -topic(Match) -> - emqx_topic_index:get_topic(Match). +-spec get_topic(emqx_rule_index:match(_ID)) -> emqx_types:topic(). +get_topic(K) -> + emqx_topic:join(cut_topic(K)). + +cut_topic(K) -> + cut_topic(K, []). + +cut_topic([{_ID}], Acc) -> + lists:reverse(Acc); +cut_topic([W | Rest], Acc) -> + cut_topic(Rest, [W | Acc]). diff --git a/changes/ce/perf-11282.en.md b/changes/ce/perf-11282.en.md new file mode 100644 index 000000000..107889957 --- /dev/null +++ b/changes/ce/perf-11282.en.md @@ -0,0 +1 @@ +Added indexing to the rule engine's topic matching to improve rule search performance. From c393c2e091aca847a9473ff848cf8f8d15f6c96f Mon Sep 17 00:00:00 2001 From: JimMoen Date: Fri, 21 Jul 2023 19:42:02 +0800 Subject: [PATCH 4/4] test: ets table cleanup after cases --- .../test/emqx_rule_index_SUITE.erl | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl b/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl index c4b1b4848..cf4b67cd4 100644 --- a/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl +++ b/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl @@ -30,7 +30,8 @@ t_insert(_) -> true = emqx_rule_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab), true = emqx_rule_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab), ?assertEqual(<<"sensor/#">>, get_topic(match(<<"sensor">>, Tab))), - ?assertEqual(t_insert_3, get_id(match(<<"sensor">>, Tab))). + ?assertEqual(t_insert_3, get_id(match(<<"sensor">>, Tab))), + true = ets:delete(Tab). t_match(_) -> Tab = new(), @@ -40,7 +41,8 @@ t_match(_) -> ?assertMatch( [<<"sensor/#">>, <<"sensor/+/#">>], [get_topic(M) || M <- matches(<<"sensor/1">>, Tab)] - ). + ), + true = ets:delete(Tab). t_match2(_) -> Tab = new(), @@ -54,7 +56,8 @@ t_match2(_) -> ?assertEqual( false, emqx_rule_index:match(<<"$SYS/broker/zenmq">>, Tab) - ). + ), + true = ets:delete(Tab). t_match3(_) -> Tab = new(), @@ -77,7 +80,8 @@ t_match3(_) -> ?assertEqual( t_match3_sys, get_id(match(<<"$SYS/a/b/c">>, Tab)) - ). + ), + true = ets:delete(Tab). t_match4(_) -> Tab = new(), @@ -93,7 +97,8 @@ t_match4(_) -> ?assertEqual( [<<"/#">>, <<"/+/a/b/c">>], [get_topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)] - ). + ), + true = ets:delete(Tab). t_match5(_) -> Tab = new(), @@ -114,21 +119,24 @@ t_match5(_) -> ?assertEqual( [<<"#">>, <>, <>], [get_topic(M) || M <- matches(<>, Tab)] - ). + ), + true = ets:delete(Tab). t_match6(_) -> Tab = new(), T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>, W = <<"+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/#">>, emqx_rule_index:insert(W, ID = t_match6, <<>>, Tab), - ?assertEqual(ID, get_id(match(T, Tab))). + ?assertEqual(ID, get_id(match(T, Tab))), + true = ets:delete(Tab). t_match7(_) -> Tab = new(), T = <<"a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z">>, W = <<"a/+/c/+/e/+/g/+/i/+/k/+/m/+/o/+/q/+/s/+/u/+/w/+/y/+/#">>, emqx_rule_index:insert(W, t_match7, <<>>, Tab), - ?assertEqual(W, get_topic(match(T, Tab))). + ?assertEqual(W, get_topic(match(T, Tab))), + true = ets:delete(Tab). new() -> ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]).