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

refactor: module move to app emqx_rule_engine 

- Rename to emqx_rule_index.erl
- Remove test funcs from src -> test dir
This commit is contained in:
JimMoen 2023-07-21 17:58:54 +08:00
parent b821bdee00
commit 4e4b1ac115
No known key found for this signature in database
GPG Key ID: 87A520B4F76BA86D
4 changed files with 69 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
apps/emqx_rule_engine/src/emqx_rule_engine.erl
View File

@ -225,12 +225,13 @@ get_rules_ordered_by_ts() ->
-spec get_rules_for_topic(Topic :: binary()) -> [rule()]. -spec get_rules_for_topic(Topic :: binary()) -> [rule()].
get_rules_for_topic(Topic) -> get_rules_for_topic(Topic) ->
[ [
emqx_topic_index:get_record(M, ?RULE_TOPIC_INDEX) emqx_rule_index:get_record(M, ?RULE_TOPIC_INDEX)
|| M <- emqx_topic_index:matches(Topic, ?RULE_TOPIC_INDEX) || M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX)
]. ].
-spec get_rules_with_same_event(Topic :: binary()) -> [rule()]. -spec get_rules_with_same_event(Topic :: binary()) -> [rule()].
get_rules_with_same_event(Topic) -> get_rules_with_same_event(Topic) ->
%% TODO: event matching index not implemented yet
EventName = emqx_rule_events:event_name(Topic), EventName = emqx_rule_events:event_name(Topic),
[ [
Rule Rule
@ -240,6 +241,7 @@ get_rules_with_same_event(Topic) ->
-spec get_rule_ids_by_action(action_name()) -> [rule_id()]. -spec get_rule_ids_by_action(action_name()) -> [rule_id()].
get_rule_ids_by_action(BridgeId) when is_binary(BridgeId) -> get_rule_ids_by_action(BridgeId) when is_binary(BridgeId) ->
%% TODO: bridge matching index not implemented yet
[ [
Id Id
|| #{actions := Acts, id := Id, from := Froms} <- get_rules(), || #{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) references_ingress_bridge(Froms, BridgeId)
]; ];
get_rule_ids_by_action(#{function := FuncName}) when is_binary(FuncName) -> get_rule_ids_by_action(#{function := FuncName}) when is_binary(FuncName) ->
%% TODO: action id matching index not implemented yet
{Mod, Fun} = {Mod, Fun} =
case string:split(FuncName, ":", leading) of case string:split(FuncName, ":", leading) of
[M, F] -> {binary_to_module(M), F}; [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) -> do_update_rule_index(#{id := Id, from := From} = Rule) ->
ok = lists:foreach( ok = lists:foreach(
fun(Topic) -> 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, end,
From From
). ).
@ -503,7 +506,7 @@ do_update_rule_index(#{id := Id, from := From} = Rule) ->
do_delete_rule_index(#{id := Id, from := From}) -> do_delete_rule_index(#{id := Id, from := From}) ->
ok = lists:foreach( ok = lists:foreach(
fun(Topic) -> fun(Topic) ->
true = emqx_topic_index:delete(Topic, Id, ?RULE_TOPIC_INDEX) true = emqx_rule_index:delete(Topic, Id, ?RULE_TOPIC_INDEX)
end, end,
From From
). ).

26
apps/emqx/src/emqx_topic_index.erl → apps/emqx_rule_engine/src/emqx_rule_index.erl
View File

@ -29,25 +29,24 @@
%% 1. Which topic filters match given topic? %% 1. Which topic filters match given topic?
%% 2. Which record IDs are associated with topic filters matching 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([insert/4]).
-export([delete/3]). -export([delete/3]).
-export([match/2]). -export([match/2]).
-export([matches/2]). -export([matches/2]).
-export([get_id/1]).
-export([get_topic/1]).
-export([get_record/2]). -export([get_record/2]).
-type key(ID) :: [binary() | '+' | '#' | {ID}]. -type key(ID) :: [binary() | '+' | '#' | {ID}].
-type match(ID) :: key(ID). -type match(ID) :: key(ID).
new() -> -ifdef(TEST).
ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]). -export_type([match/1]).
-endif.
insert(Filter, ID, Record, Tab) -> insert(Filter, ID, Record, Tab) ->
%% TODO: topic compact. see also in emqx_trie.erl
ets:insert(Tab, {emqx_topic:words(Filter) ++ [{ID}], Record}). ets:insert(Tab, {emqx_topic:words(Filter) ++ [{ID}], Record}).
delete(Filter, ID, Tab) -> delete(Filter, ID, Tab) ->
@ -136,21 +135,6 @@ match_init(Topic) ->
{Words, []} {Words, []}
end. 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. -spec get_record(match(_ID), ets:table()) -> _Record.
get_record(K, Tab) -> get_record(K, Tab) ->
ets:lookup_element(Tab, K, 2). ets:lookup_element(Tab, K, 2).

97
apps/emqx/test/emqx_topic_index_SUITE.erl → apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl
View File

@ -14,7 +14,7 @@
%% limitations under the License. %% limitations under the License.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
-module(emqx_topic_index_SUITE). -module(emqx_rule_index_SUITE).
-compile(export_all). -compile(export_all).
-compile(nowarn_export_all). -compile(nowarn_export_all).
@ -25,39 +25,39 @@ all() ->
emqx_common_test_helpers:all(?MODULE). emqx_common_test_helpers:all(?MODULE).
t_insert(_) -> t_insert(_) ->
Tab = emqx_topic_index:new(), Tab = new(),
true = emqx_topic_index:insert(<<"sensor/1/metric/2">>, t_insert_1, <<>>, Tab), true = emqx_rule_index:insert(<<"sensor/1/metric/2">>, t_insert_1, <<>>, Tab),
true = emqx_topic_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab), true = emqx_rule_index:insert(<<"sensor/+/#">>, t_insert_2, <<>>, Tab),
true = emqx_topic_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab), true = emqx_rule_index:insert(<<"sensor/#">>, t_insert_3, <<>>, Tab),
?assertEqual(<<"sensor/#">>, topic(match(<<"sensor">>, Tab))), ?assertEqual(<<"sensor/#">>, get_topic(match(<<"sensor">>, Tab))),
?assertEqual(t_insert_3, id(match(<<"sensor">>, Tab))). ?assertEqual(t_insert_3, get_id(match(<<"sensor">>, Tab))).
t_match(_) -> t_match(_) ->
Tab = emqx_topic_index:new(), Tab = new(),
true = emqx_topic_index:insert(<<"sensor/1/metric/2">>, t_match_1, <<>>, Tab), true = emqx_rule_index:insert(<<"sensor/1/metric/2">>, t_match_1, <<>>, Tab),
true = emqx_topic_index:insert(<<"sensor/+/#">>, t_match_2, <<>>, Tab), true = emqx_rule_index:insert(<<"sensor/+/#">>, t_match_2, <<>>, Tab),
true = emqx_topic_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab), true = emqx_rule_index:insert(<<"sensor/#">>, t_match_3, <<>>, Tab),
?assertMatch( ?assertMatch(
[<<"sensor/#">>, <<"sensor/+/#">>], [<<"sensor/#">>, <<"sensor/+/#">>],
[topic(M) || M <- matches(<<"sensor/1">>, Tab)] [get_topic(M) || M <- matches(<<"sensor/1">>, Tab)]
). ).
t_match2(_) -> t_match2(_) ->
Tab = emqx_topic_index:new(), Tab = new(),
true = emqx_topic_index:insert(<<"#">>, t_match2_1, <<>>, Tab), true = emqx_rule_index:insert(<<"#">>, t_match2_1, <<>>, Tab),
true = emqx_topic_index:insert(<<"+/#">>, t_match2_2, <<>>, Tab), true = emqx_rule_index:insert(<<"+/#">>, t_match2_2, <<>>, Tab),
true = emqx_topic_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab), true = emqx_rule_index:insert(<<"+/+/#">>, t_match2_3, <<>>, Tab),
?assertEqual( ?assertEqual(
[<<"#">>, <<"+/#">>, <<"+/+/#">>], [<<"#">>, <<"+/#">>, <<"+/+/#">>],
[topic(M) || M <- matches(<<"a/b/c">>, Tab)] [get_topic(M) || M <- matches(<<"a/b/c">>, Tab)]
), ),
?assertEqual( ?assertEqual(
false, false,
emqx_topic_index:match(<<"$SYS/broker/zenmq">>, Tab) emqx_rule_index:match(<<"$SYS/broker/zenmq">>, Tab)
). ).
t_match3(_) -> t_match3(_) ->
Tab = emqx_topic_index:new(), Tab = new(),
Records = [ Records = [
{<<"d/#">>, t_match3_1}, {<<"d/#">>, t_match3_1},
{<<"a/b/+">>, t_match3_2}, {<<"a/b/+">>, t_match3_2},
@ -66,7 +66,7 @@ t_match3(_) ->
{<<"$SYS/#">>, t_match3_sys} {<<"$SYS/#">>, t_match3_sys}
], ],
lists:foreach( 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 Records
), ),
Matched = matches(<<"a/b/c">>, Tab), Matched = matches(<<"a/b/c">>, Tab),
@ -76,27 +76,27 @@ t_match3(_) ->
end, end,
?assertEqual( ?assertEqual(
t_match3_sys, t_match3_sys,
id(match(<<"$SYS/a/b/c">>, Tab)) get_id(match(<<"$SYS/a/b/c">>, Tab))
). ).
t_match4(_) -> t_match4(_) ->
Tab = emqx_topic_index:new(), Tab = new(),
Records = [{<<"/#">>, t_match4_1}, {<<"/+">>, t_match4_2}, {<<"/+/a/b/c">>, t_match4_3}], Records = [{<<"/#">>, t_match4_1}, {<<"/+">>, t_match4_2}, {<<"/+/a/b/c">>, t_match4_3}],
lists:foreach( 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 Records
), ),
?assertEqual( ?assertEqual(
[<<"/#">>, <<"/+">>], [<<"/#">>, <<"/+">>],
[topic(M) || M <- matches(<<"/">>, Tab)] [get_topic(M) || M <- matches(<<"/">>, Tab)]
), ),
?assertEqual( ?assertEqual(
[<<"/#">>, <<"/+/a/b/c">>], [<<"/#">>, <<"/+/a/b/c">>],
[topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)] [get_topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)]
). ).
t_match5(_) -> 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">>, 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 = [ Records = [
{<<"#">>, t_match5_1}, {<<"#">>, t_match5_1},
@ -104,40 +104,55 @@ t_match5(_) ->
{<<T/binary, "/+">>, t_match5_3} {<<T/binary, "/+">>, t_match5_3}
], ],
lists:foreach( 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 Records
), ),
?assertEqual( ?assertEqual(
[<<"#">>, <<T/binary, "/#">>], [<<"#">>, <<T/binary, "/#">>],
[topic(M) || M <- matches(T, Tab)] [get_topic(M) || M <- matches(T, Tab)]
), ),
?assertEqual( ?assertEqual(
[<<"#">>, <<T/binary, "/#">>, <<T/binary, "/+">>], [<<"#">>, <<T/binary, "/#">>, <<T/binary, "/+">>],
[topic(M) || M <- matches(<<T/binary, "/1">>, Tab)] [get_topic(M) || M <- matches(<<T/binary, "/1">>, Tab)]
). ).
t_match6(_) -> 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">>, 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 = <<"+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/#">>, W = <<"+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/+/#">>,
emqx_topic_index:insert(W, ID = t_match6, <<>>, Tab), emqx_rule_index:insert(W, ID = t_match6, <<>>, Tab),
?assertEqual(ID, id(match(T, Tab))). ?assertEqual(ID, get_id(match(T, Tab))).
t_match7(_) -> 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">>, 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/+/#">>, W = <<"a/+/c/+/e/+/g/+/i/+/k/+/m/+/o/+/q/+/s/+/u/+/w/+/y/+/#">>,
emqx_topic_index:insert(W, t_match7, <<>>, Tab), emqx_rule_index:insert(W, t_match7, <<>>, Tab),
?assertEqual(W, topic(match(T, Tab))). ?assertEqual(W, get_topic(match(T, Tab))).
new() ->
ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]).
match(T, Tab) -> match(T, Tab) ->
emqx_topic_index:match(T, Tab). emqx_rule_index:match(T, Tab).
matches(T, Tab) -> matches(T, Tab) ->
lists:sort(emqx_topic_index:matches(T, Tab)). lists:sort(emqx_rule_index:matches(T, Tab)).
id(Match) -> -spec get_id(emqx_rule_index:match(ID)) -> ID.
emqx_topic_index:get_id(Match). get_id([{ID}]) ->
ID;
get_id([_ | Rest]) ->
get_id(Rest).
topic(Match) -> -spec get_topic(emqx_rule_index:match(_ID)) -> emqx_types:topic().
emqx_topic_index:get_topic(Match). 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]).

1
changes/ce/perf-11282.en.md Normal file
View File

@ -0,0 +1 @@
Added indexing to the rule engine's topic matching to improve rule search performance.