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.