diff --git a/apps/emqx_rule_engine/src/emqx_rule_engine.erl b/apps/emqx_rule_engine/src/emqx_rule_engine.erl index dd4b52d44..d92931d77 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine.erl +++ b/apps/emqx_rule_engine/src/emqx_rule_engine.erl @@ -226,7 +226,7 @@ get_rules_ordered_by_ts() -> get_rules_for_topic(Topic) -> [ emqx_rule_index:get_record(M, ?RULE_TOPIC_INDEX) - || M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX, [unique]) + || M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX) ]. -spec get_rules_with_same_event(Topic :: binary()) -> [rule()]. diff --git a/apps/emqx_rule_engine/src/emqx_rule_index.erl b/apps/emqx_rule_engine/src/emqx_rule_index.erl index 4dd395b94..9c16bf3ad 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_index.erl +++ b/apps/emqx_rule_engine/src/emqx_rule_index.erl @@ -16,29 +16,29 @@ %% @doc Topic index for matching topics to topic filters. %% -%% Works on top of ETS ordered_set table. Keys are tuples constructed from -%% parsed topic filters and record IDs, wrapped in a tuple to order them -%% strictly greater than unit tuple (`{}`). Existing table may be used if -%% existing keys will not collide with index keys. +%% 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? -%% 3. Which topic filters match given topic? -%% 4. Which record IDs 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_rule_index). -export([insert/4]). -export([delete/3]). -export([match/2]). --export([matches/3]). +-export([matches/2]). --export([get_id/1]). --export([get_topic/1]). -export([get_record/2]). --type key(ID) :: {[binary() | '+' | '#'], {ID}}. +-type key(ID) :: [binary() | '+' | '#' | {ID}]. -type match(ID) :: key(ID). -ifdef(TEST). @@ -46,10 +46,11 @@ -endif. insert(Filter, ID, Record, Tab) -> - ets:insert(Tab, {{emqx_topic:words(Filter), {ID}}, Record}). + %% TODO: topic compact. see also in emqx_trie.erl + ets:insert(Tab, {emqx_topic:words(Filter) ++ [{ID}], Record}). delete(Filter, ID, Tab) -> - ets:delete(Tab, {emqx_topic:words(Filter), {ID}}). + ets:delete(Tab, emqx_topic:words(Filter) ++ [{ID}]). -spec match(emqx_types:topic(), ets:table()) -> match(_ID) | false. match(Topic, Tab) -> @@ -58,8 +59,8 @@ match(Topic, Tab) -> match(Words, RPrefix, Tab) -> Prefix = lists:reverse(RPrefix), - K = ets:next(Tab, {Prefix, {}}), - case match_filter(Prefix, K, Words == []) of + K = ets:next(Tab, Prefix), + case match_filter(Prefix, K, Words =/= []) of true -> K; stop -> @@ -72,7 +73,7 @@ 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 = {_, _} -> + Match when is_list(Match) -> Match; false -> match(Rest, [W | RPrefix], Tab) @@ -80,86 +81,48 @@ match_rest(plus, [W | Rest], RPrefix, Tab) -> match_rest(_, [], _RPrefix, _Tab) -> false. --spec matches(emqx_types:topic(), ets:table(), _Opts :: [unique]) -> [match(_ID)]. -matches(Topic, Tab, Opts) -> +-spec matches(emqx_types:topic(), ets:table()) -> [match(_ID)]. +matches(Topic, Tab) -> {Words, RPrefix} = match_init(Topic), - AccIn = - case Opts of - [unique | _] -> #{}; - [] -> [] - end, - Matches = matches(Words, RPrefix, AccIn, Tab), - %% return rules ordered by Rule ID - case Matches of - #{} -> - maps:values(Matches); - _ -> - F = fun({_, {ID1}}, {_, {ID2}}) -> ID1 < ID2 end, - lists:sort(F, Matches) - end. + matches(Words, RPrefix, Tab). -matches(Words, RPrefix, Acc, Tab) -> +matches(Words, RPrefix, Tab) -> Prefix = lists:reverse(RPrefix), - matches(ets:next(Tab, {Prefix, {}}), Prefix, Words, RPrefix, Acc, Tab). + matches(ets:next(Tab, Prefix), Prefix, Words, RPrefix, Tab). -matches(K, Prefix, Words, RPrefix, Acc, Tab) -> - case match_filter(Prefix, K, Words == []) of +matches(K, Prefix, Words, RPrefix, Tab) -> + case match_filter(Prefix, K, Words =/= []) of true -> - matches(ets:next(Tab, K), Prefix, Words, RPrefix, match_add(K, Acc), Tab); + [K | matches(ets:next(Tab, K), Prefix, Words, RPrefix, Tab)]; stop -> - Acc; + []; Matched -> - matches_rest(Matched, Words, RPrefix, Acc, Tab) + matches_rest(Matched, Words, RPrefix, Tab) end. -matches_rest(false, [W | Rest], RPrefix, Acc, Tab) -> - matches(Rest, [W | RPrefix], Acc, Tab); -matches_rest(sharp, [W | Rest], RPrefix, Acc, Tab) -> - NAcc1 = matches([], ['#' | RPrefix], Acc, Tab), - NAcc2 = matches(Rest, ['+' | RPrefix], NAcc1, Tab), - matches(Rest, [W | RPrefix], NAcc2, Tab); -matches_rest(plus, [W | Rest], RPrefix, Acc, Tab) -> - NAcc = matches(Rest, ['+' | RPrefix], Acc, Tab), - matches(Rest, [W | RPrefix], NAcc, Tab); -matches_rest(_, [], _RPrefix, Acc, _Tab) -> - Acc. +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_add(K = {_Filter, ID}, Acc = #{}) -> - Acc#{ID => K}; -match_add(K, Acc) -> - [K | Acc]. - -match_filter(Prefix, {Filter, _ID}, NotPrefix) -> - case match_filter(Prefix, Filter) of - exact -> - % NOTE: exact match is `true` only if we match whole topic, not prefix - case NotPrefix of - true -> - true; - false -> - sharp - end; - Match -> - Match - end; -match_filter(_, '$end_of_table', _) -> - stop. - -match_filter([], []) -> - exact; -match_filter([], ['' | _]) -> - sharp; -match_filter([], ['#']) -> +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([], ['+' | _]) -> +match_filter([], ['+' | _], _) -> plus; -match_filter([], [_H | _]) -> +match_filter([], [_H | _], _) -> false; -match_filter([H | T1], [H | T2]) -> - match_filter(T1, T2); -match_filter([H1 | _], [H2 | _]) when H2 > H1 -> +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) -> @@ -172,14 +135,6 @@ match_init(Topic) -> {Words, []} end. --spec get_id(match(ID)) -> ID. -get_id({_Filter, {ID}}) -> - ID. - --spec get_topic(match(_ID)) -> emqx_types:topic(). -get_topic({Filter, _ID}) -> - emqx_topic:join(Filter). - -spec get_record(match(_ID), ets:table()) -> _Record. get_record(K, Tab) -> ets:lookup_element(Tab, K, 2). diff --git a/apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl b/apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl index a069a132c..8c3bd0ebb 100644 --- a/apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl +++ b/apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl @@ -58,7 +58,6 @@ groups() -> t_create_existing_rule, t_get_rules_for_topic, t_get_rules_for_topic_2, - t_get_rules_for_topic_3, t_get_rules_with_same_event, t_get_rule_ids_by_action, t_ensure_action_removed @@ -401,44 +400,6 @@ t_get_rules_for_topic_2(_Config) -> ]), ok. -t_get_rules_for_topic_3(_Config) -> - ok = create_rules( - [ - make_simple_rule(<<"rule-debug-5">>, <<"select * from \"simple/#\"">>), - make_simple_rule(<<"rule-debug-4">>, <<"select * from \"simple/+\"">>), - make_simple_rule(<<"rule-debug-3">>, <<"select * from \"simple/+/1\"">>), - make_simple_rule(<<"rule-debug-2">>, <<"select * from \"simple/1\"">>), - make_simple_rule( - <<"rule-debug-1">>, - <<"select * from \"simple/2\", \"simple/+\", \"simple/3\"">> - ) - ] - ), - Rules1 = get_rules_for_topic_in_e510_impl(<<"simple/1">>), - Rules2 = emqx_rule_engine:get_rules_for_topic(<<"simple/1">>), - %% assert, ensure the order of rules is the same as e5.1.0 - ?assertEqual(Rules1, Rules2), - ?assertEqual( - [<<"rule-debug-1">>, <<"rule-debug-2">>, <<"rule-debug-4">>, <<"rule-debug-5">>], - [Id || #{id := Id} <- Rules1] - ), - - ok = delete_rules_by_ids([ - <<"rule-debug-1">>, - <<"rule-debug-2">>, - <<"rule-debug-3">>, - <<"rule-debug-4">>, - <<"rule-debug-5">> - ]), - ok. - -get_rules_for_topic_in_e510_impl(Topic) -> - [ - Rule - || Rule = #{from := From} <- emqx_rule_engine:get_rules(), - emqx_topic:match_any(Topic, From) - ]. - t_get_rules_with_same_event(_Config) -> PubT = <<"simple/1">>, PubN = length(emqx_rule_engine:get_rules_with_same_event(PubT)), 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 8a65ee8e8..cf4b67cd4 100644 --- a/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl +++ b/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl @@ -19,7 +19,6 @@ -compile(export_all). -compile(nowarn_export_all). --include_lib("proper/include/proper.hrl"). -include_lib("eunit/include/eunit.hrl"). all() -> @@ -30,8 +29,8 @@ t_insert(_) -> 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/#">>, topic(match(<<"sensor">>, Tab))), - ?assertEqual(t_insert_3, id(match(<<"sensor">>, Tab))), + ?assertEqual(<<"sensor/#">>, get_topic(match(<<"sensor">>, Tab))), + ?assertEqual(t_insert_3, get_id(match(<<"sensor">>, Tab))), true = ets:delete(Tab). t_match(_) -> @@ -41,7 +40,7 @@ t_match(_) -> 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)] ), true = ets:delete(Tab). @@ -52,7 +51,7 @@ t_match2(_) -> 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, @@ -80,7 +79,7 @@ t_match3(_) -> end, ?assertEqual( t_match3_sys, - id(match(<<"$SYS/a/b/c">>, Tab)) + get_id(match(<<"$SYS/a/b/c">>, Tab)) ), true = ets:delete(Tab). @@ -93,11 +92,11 @@ t_match4(_) -> ), ?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)] ), true = ets:delete(Tab). @@ -115,11 +114,11 @@ t_match5(_) -> ), ?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)] ), true = ets:delete(Tab). @@ -128,7 +127,7 @@ t_match6(_) -> 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, id(match(T, Tab))), + ?assertEqual(ID, get_id(match(T, Tab))), true = ets:delete(Tab). t_match7(_) -> @@ -136,173 +135,32 @@ t_match7(_) -> 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, topic(match(T, Tab))), + ?assertEqual(W, get_topic(match(T, Tab))), true = ets:delete(Tab). -t_match_unique(_) -> - Tab = new(), - emqx_rule_index:insert(<<"a/b/c">>, t_match_id1, <<>>, Tab), - emqx_rule_index:insert(<<"a/b/+">>, t_match_id1, <<>>, Tab), - emqx_rule_index:insert(<<"a/b/c/+">>, t_match_id2, <<>>, Tab), - ?assertEqual( - [t_match_id1, t_match_id1], - [id(M) || M <- emqx_rule_index:matches(<<"a/b/c">>, Tab, [])] - ), - ?assertEqual( - [t_match_id1], - [id(M) || M <- emqx_rule_index:matches(<<"a/b/c">>, Tab, [unique])] - ). - -t_match_ordering(_) -> - Tab = new(), - emqx_rule_index:insert(<<"a/b/+">>, t_match_id2, <<>>, Tab), - emqx_rule_index:insert(<<"a/b/c">>, t_match_id1, <<>>, Tab), - emqx_rule_index:insert(<<"a/b/#">>, t_match_id3, <<>>, Tab), - Ids1 = [id(M) || M <- emqx_rule_index:matches(<<"a/b/c">>, Tab, [])], - Ids2 = [id(M) || M <- emqx_rule_index:matches(<<"a/b/c">>, Tab, [unique])], - ?assertEqual(Ids1, Ids2), - ?assertEqual([t_match_id1, t_match_id2, t_match_id3], Ids1). - -t_match_wildcard_edge_cases(_) -> - CommonTopics = [ - <<"a/b">>, - <<"a/b/#">>, - <<"a/b/#">>, - <<"a/b/c">>, - <<"a/b/+">>, - <<"a/b/d">>, - <<"a/+/+">>, - <<"a/+/#">> - ], - Datasets = - [ - %% Topics, TopicName, Results - {CommonTopics, <<"a/b/c">>, [2, 3, 4, 5, 7, 8]}, - {CommonTopics, <<"a/b">>, [1, 2, 3, 8]}, - {[<<"+/b/c">>, <<"/">>], <<"a/b/c">>, [1]}, - {[<<"#">>, <<"/">>], <<"a">>, [1]}, - {[<<"/">>, <<"+">>], <<"a">>, [2]} - ], - F = fun({Topics, TopicName, Expected}) -> - Tab = new(), - _ = lists:foldl( - fun(T, N) -> - emqx_rule_index:insert(T, N, <<>>, Tab), - N + 1 - end, - 1, - Topics - ), - Results = [id(M) || M <- emqx_rule_index:matches(TopicName, Tab, [unique])], - case Results == Expected of - true -> - ets:delete(Tab); - false -> - ct:pal( - "Base topics: ~p~n" - "Topic name: ~p~n" - "Index results: ~p~n" - "Expected results:: ~p~n", - [Topics, TopicName, Results, Expected] - ), - error(bad_matches) - end - end, - lists:foreach(F, Datasets). - -t_prop_matches(_) -> - ?assert( - proper:quickcheck( - topic_matches_prop(), - [{max_size, 100}, {numtests, 100}] - ) - ). - -topic_matches_prop() -> - ?FORALL( - Topics0, - list(emqx_proper_types:topic()), - begin - Tab = new(), - Topics = lists:filter(fun(Topic) -> Topic =/= <<>> end, Topics0), - lists:foldl( - fun(Topic, N) -> - true = emqx_rule_index:insert(Topic, N, <<>>, Tab), - N + 1 - end, - 1, - Topics - ), - lists:foreach( - fun(Topic0) -> - Topic = topic_filter_to_topic_name(Topic0), - Ids1 = [ - emqx_rule_index:get_id(R) - || R <- emqx_rule_index:matches(Topic, Tab, [unique]) - ], - Ids2 = topic_matches(Topic, Topics), - case Ids2 == Ids1 of - true -> - ok; - false -> - ct:pal( - "Base topics: ~p~n" - "Topic name: ~p~n" - "Index results: ~p~n" - "Topic match results:: ~p~n", - [Topics, Topic, Ids1, Ids2] - ), - error(bad_matches) - end - end, - Topics - ), - true - end - ). - -%%-------------------------------------------------------------------- -%% helpers - new() -> - ets:new(?MODULE, [public, ordered_set, {read_concurrency, true}]). + ets:new(?MODULE, [public, ordered_set, {write_concurrency, true}]). match(T, Tab) -> emqx_rule_index:match(T, Tab). matches(T, Tab) -> - lists:sort(emqx_rule_index:matches(T, Tab, [])). + lists:sort(emqx_rule_index:matches(T, Tab)). -id(Match) -> - emqx_rule_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_rule_index:get_topic(Match). +-spec get_topic(emqx_rule_index:match(_ID)) -> emqx_types:topic(). +get_topic(K) -> + emqx_topic:join(cut_topic(K)). -topic_filter_to_topic_name(Topic) when is_binary(Topic) -> - topic_filter_to_topic_name(emqx_topic:words(Topic)); -topic_filter_to_topic_name(Words) when is_list(Words) -> - topic_filter_to_topic_name(Words, []). +cut_topic(K) -> + cut_topic(K, []). -topic_filter_to_topic_name([], Acc) -> - emqx_topic:join(lists:reverse(Acc)); -topic_filter_to_topic_name(['#' | _Rest], Acc) -> - case rand:uniform(2) of - 1 -> emqx_topic:join(lists:reverse(Acc)); - _ -> emqx_topic:join(lists:reverse([<<"_sharp">> | Acc])) - end; -topic_filter_to_topic_name(['+' | Rest], Acc) -> - topic_filter_to_topic_name(Rest, [<<"_plus">> | Acc]); -topic_filter_to_topic_name([H | Rest], Acc) -> - topic_filter_to_topic_name(Rest, [H | Acc]). - -topic_matches(Topic, Topics0) -> - Topics = lists:zip(lists:seq(1, length(Topics0)), Topics0), - lists:sort( - lists:filtermap( - fun({Id, Topic0}) -> - emqx_topic:match(Topic, Topic0) andalso {true, Id} - end, - Topics - ) - ). +cut_topic([{_ID}], Acc) -> + lists:reverse(Acc); +cut_topic([W | Rest], Acc) -> + cut_topic(Rest, [W | Acc]).