From 951a96457b179d2c90e8423b53676db440cd3f65 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Thu, 27 Jul 2023 13:42:43 +0800 Subject: [PATCH] Revert "feat(index): add topic index facility " --- apps/emqx_rule_engine/include/rule_engine.hrl | 1 - .../emqx_rule_engine/src/emqx_rule_engine.erl | 71 +++----- .../src/emqx_rule_engine_app.erl | 1 - apps/emqx_rule_engine/src/emqx_rule_index.erl | 140 --------------- .../test/emqx_rule_index_SUITE.erl | 166 ------------------ changes/ce/perf-11282.en.md | 1 - 6 files changed, 21 insertions(+), 359 deletions(-) delete mode 100644 apps/emqx_rule_engine/src/emqx_rule_index.erl delete mode 100644 apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl delete mode 100644 changes/ce/perf-11282.en.md diff --git a/apps/emqx_rule_engine/include/rule_engine.hrl b/apps/emqx_rule_engine/include/rule_engine.hrl index 7df5d9941..b2a6a549e 100644 --- a/apps/emqx_rule_engine/include/rule_engine.hrl +++ b/apps/emqx_rule_engine/include/rule_engine.hrl @@ -109,7 +109,6 @@ %% 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 d92931d77..66c82d3a1 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 -> with_parsed_rule(Params, CreatedAt, fun insert_rule/1); + not_found -> parse_and_insert(Params, CreatedAt); {ok, _} -> {error, already_exists} end. @@ -185,27 +185,18 @@ update_rule(Params = #{id := RuleId}) when is_binary(RuleId) -> case get_rule(RuleId) of not_found -> {error, not_found}; - {ok, RulePrev = #{created_at := CreatedAt}} -> - with_parsed_rule(Params, CreatedAt, fun(Rule) -> update_rule(Rule, RulePrev) end) + {ok, #{created_at := CreatedAt}} -> + parse_and_insert(Params, CreatedAt) end. -spec delete_rule(RuleId :: rule_id()) -> ok. delete_rule(RuleId) when is_binary(RuleId) -> - case get_rule(RuleId) of - not_found -> - ok; - {ok, Rule} -> - gen_server:call(?RULE_ENGINE, {delete_rule, Rule}, ?T_CALL) - end. + gen_server:call(?RULE_ENGINE, {delete_rule, RuleId}, ?T_CALL). -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 %%---------------------------------------------------------------------------------------- @@ -225,13 +216,13 @@ get_rules_ordered_by_ts() -> -spec get_rules_for_topic(Topic :: binary()) -> [rule()]. get_rules_for_topic(Topic) -> [ - emqx_rule_index:get_record(M, ?RULE_TOPIC_INDEX) - || M <- emqx_rule_index:matches(Topic, ?RULE_TOPIC_INDEX) + Rule + || Rule = #{from := From} <- get_rules(), + emqx_topic:match_any(Topic, From) ]. -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 @@ -241,7 +232,6 @@ 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(), @@ -249,7 +239,6 @@ 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}; @@ -422,17 +411,10 @@ init([]) -> {ok, #{}}. handle_call({insert_rule, Rule}, _From, State) -> - 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), + do_insert_rule(Rule), {reply, ok, State}; handle_call({delete_rule, Rule}, _From, State) -> - ok = do_delete_rule_index(Rule), - ok = do_delete_rule(Rule), + do_delete_rule(Rule), {reply, ok, State}; handle_call(Req, _From, State) -> ?SLOG(error, #{msg => "unexpected_call", request => Req}), @@ -456,7 +438,7 @@ code_change(_OldVsn, State, _Extra) -> %% Internal Functions %%---------------------------------------------------------------------------------------- -with_parsed_rule(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt, Fun) -> +parse_and_insert(Params = #{id := RuleId, sql := Sql, actions := Actions}, CreatedAt) -> case emqx_rule_sqlparser:parse(Sql) of {ok, Select} -> Rule = #{ @@ -477,7 +459,7 @@ with_parsed_rule(Params = #{id := RuleId, sql := Sql, actions := Actions}, Creat conditions => emqx_rule_sqlparser:select_where(Select) %% -- calculated fields end }, - ok = Fun(Rule), + ok = insert_rule(Rule), {ok, Rule}; {error, Reason} -> {error, Reason} @@ -489,27 +471,16 @@ do_insert_rule(#{id := Id} = Rule) -> true = ets:insert(?RULE_TAB, {Id, maps:remove(id, Rule)}), ok. -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_rule_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_rule_index:delete(Topic, Id, ?RULE_TOPIC_INDEX) - end, - From - ). +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. 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 28515cb1a..d8b031bdd 100644 --- a/apps/emqx_rule_engine/src/emqx_rule_engine_app.erl +++ b/apps/emqx_rule_engine/src/emqx_rule_engine_app.erl @@ -26,7 +26,6 @@ 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(), diff --git a/apps/emqx_rule_engine/src/emqx_rule_index.erl b/apps/emqx_rule_engine/src/emqx_rule_index.erl deleted file mode 100644 index 9c16bf3ad..000000000 --- a/apps/emqx_rule_engine/src/emqx_rule_index.erl +++ /dev/null @@ -1,140 +0,0 @@ -%%-------------------------------------------------------------------- -%% 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_rule_index). - --export([insert/4]). --export([delete/3]). --export([match/2]). --export([matches/2]). - --export([get_record/2]). - --type key(ID) :: [binary() | '+' | '#' | {ID}]. --type match(ID) :: key(ID). - --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) -> - 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_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_index_SUITE.erl b/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl deleted file mode 100644 index cf4b67cd4..000000000 --- a/apps/emqx_rule_engine/test/emqx_rule_index_SUITE.erl +++ /dev/null @@ -1,166 +0,0 @@ -%%-------------------------------------------------------------------- -%% 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_rule_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 = 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))), - true = ets:delete(Tab). - -t_match(_) -> - 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/+/#">>], - [get_topic(M) || M <- matches(<<"sensor/1">>, Tab)] - ), - true = ets:delete(Tab). - -t_match2(_) -> - 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( - [<<"#">>, <<"+/#">>, <<"+/+/#">>], - [get_topic(M) || M <- matches(<<"a/b/c">>, Tab)] - ), - ?assertEqual( - false, - emqx_rule_index:match(<<"$SYS/broker/zenmq">>, Tab) - ), - true = ets:delete(Tab). - -t_match3(_) -> - Tab = 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_rule_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, - get_id(match(<<"$SYS/a/b/c">>, Tab)) - ), - true = ets:delete(Tab). - -t_match4(_) -> - Tab = new(), - Records = [{<<"/#">>, t_match4_1}, {<<"/+">>, t_match4_2}, {<<"/+/a/b/c">>, t_match4_3}], - lists:foreach( - fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end, - Records - ), - ?assertEqual( - [<<"/#">>, <<"/+">>], - [get_topic(M) || M <- matches(<<"/">>, Tab)] - ), - ?assertEqual( - [<<"/#">>, <<"/+/a/b/c">>], - [get_topic(M) || M <- matches(<<"/0/a/b/c">>, Tab)] - ), - true = ets:delete(Tab). - -t_match5(_) -> - 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}, - {<>, t_match5_2}, - {<>, t_match5_3} - ], - lists:foreach( - fun({Topic, ID}) -> emqx_rule_index:insert(Topic, ID, <<>>, Tab) end, - Records - ), - ?assertEqual( - [<<"#">>, <>], - [get_topic(M) || M <- matches(T, Tab)] - ), - ?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))), - 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))), - true = ets:delete(Tab). - -new() -> - 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)). - --spec get_id(emqx_rule_index:match(ID)) -> ID. -get_id([{ID}]) -> - ID; -get_id([_ | Rest]) -> - get_id(Rest). - --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 deleted file mode 100644 index 107889957..000000000 --- a/changes/ce/perf-11282.en.md +++ /dev/null @@ -1 +0,0 @@ -Added indexing to the rule engine's topic matching to improve rule search performance.