From 36e7d63d66b47d9c6ba6f4de8d33a60b4ad0be31 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Fri, 7 Dec 2018 18:20:09 +0800 Subject: [PATCH] Implement subscription sharding. 1. Improve the design router, broker and shared_sub 2. New ets tables' design for subscription sharding --- src/emqx_access_control.erl | 2 +- src/emqx_acl_internal.erl | 2 +- src/emqx_broker.erl | 149 ++++++++++++++++++------------ src/emqx_broker_helper.erl | 14 ++- src/emqx_cm.erl | 6 +- src/emqx_ctl.erl | 2 +- src/emqx_hooks.erl | 2 +- src/emqx_metrics.erl | 2 +- src/emqx_router.erl | 175 +++++++++++++----------------------- src/emqx_sequence.erl | 3 +- src/emqx_shared_sub.erl | 63 ++++++++----- src/emqx_sm.erl | 8 +- src/emqx_stats.erl | 2 +- src/emqx_tables.erl | 6 +- src/emqx_zone.erl | 2 +- 15 files changed, 227 insertions(+), 211 deletions(-) diff --git a/src/emqx_access_control.erl b/src/emqx_access_control.erl index 1b9d76937..06ea86633 100644 --- a/src/emqx_access_control.erl +++ b/src/emqx_access_control.erl @@ -148,7 +148,7 @@ stop() -> %%----------------------------------------------------------------------------- init([]) -> - _ = emqx_tables:new(?TAB, [set, protected, {read_concurrency, true}]), + ok = emqx_tables:new(?TAB, [set, protected, {read_concurrency, true}]), {ok, #{}}. handle_call({register_mod, Type, Mod, Opts, Seq}, _From, State) -> diff --git a/src/emqx_acl_internal.erl b/src/emqx_acl_internal.erl index eee7e6c18..f8b995096 100644 --- a/src/emqx_acl_internal.erl +++ b/src/emqx_acl_internal.erl @@ -45,7 +45,7 @@ all_rules() -> -spec(init([File :: string()]) -> {ok, #{}}). init([File]) -> - _ = emqx_tables:new(?ACL_RULE_TAB, [set, public, {read_concurrency, true}]), + ok = emqx_tables:new(?ACL_RULE_TAB, [set, public, {read_concurrency, true}]), true = load_rules_from_file(File), {ok, #{acl_file => File}}. diff --git a/src/emqx_broker.erl b/src/emqx_broker.erl index 3a3cde1fa..0a9264489 100644 --- a/src/emqx_broker.erl +++ b/src/emqx_broker.erl @@ -23,7 +23,7 @@ -export([unsubscribe/1, unsubscribe/2]). -export([subscriber_down/1]). -export([publish/1, safe_publish/1]). --export([dispatch/2, dispatch/3]). +-export([dispatch/2]). -export([subscriptions/1, subscribers/1, subscribed/2]). -export([get_subopts/2, set_subopts/2]). -export([topics/0]). @@ -34,8 +34,6 @@ -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). --define(SHARD, 1024). --define(TIMEOUT, 60000). -define(BROKER, ?MODULE). %% ETS tables @@ -44,33 +42,36 @@ -define(SUBSCRIBER, emqx_subscriber). -define(SUBSCRIPTION, emqx_subscription). -%% Gards +%% Guards -define(is_subid(Id), (is_binary(Id) orelse is_atom(Id))). -spec(start_link(atom(), pos_integer()) -> emqx_types:startlink_ret()). start_link(Pool, Id) -> _ = create_tabs(), - gen_server:start_link({local, emqx_misc:proc_name(?BROKER, Id)}, ?MODULE, [Pool, Id], []). + Name = emqx_misc:proc_name(?BROKER, Id), + gen_server:start_link({local, Name}, ?MODULE, [Pool, Id], []). %%------------------------------------------------------------------------------ %% Create tabs %%------------------------------------------------------------------------------ +-spec(create_tabs() -> ok). create_tabs() -> TabOpts = [public, {read_concurrency, true}, {write_concurrency, true}], - %% SubId: SubId -> SubPid1, SubPid2,... - _ = emqx_tables:new(?SUBID, [bag | TabOpts]), + %% SubId: SubId -> SubPid + ok = emqx_tables:new(?SUBID, [set | TabOpts]), + %% SubOption: {SubPid, Topic} -> SubOption - _ = emqx_tables:new(?SUBOPTION, [set | TabOpts]), + ok = emqx_tables:new(?SUBOPTION, [set | TabOpts]), %% Subscription: SubPid -> Topic1, Topic2, Topic3, ... %% duplicate_bag: o(1) insert - _ = emqx_tables:new(?SUBSCRIPTION, [duplicate_bag | TabOpts]), + ok = emqx_tables:new(?SUBSCRIPTION, [duplicate_bag | TabOpts]), %% Subscriber: Topic -> SubPid1, SubPid2, SubPid3, ... %% duplicate_bag: o(1) insert - emqx_tables:new(?SUBSCRIBER, [duplicate_bag | TabOpts]). + ok = emqx_tables:new(?SUBSCRIBER, [duplicate_bag | TabOpts]). %%------------------------------------------------------------------------------ %% Subscribe API @@ -92,14 +93,23 @@ subscribe(Topic, SubId, SubOpts) when is_binary(Topic), ?is_subid(SubId), is_map case ets:member(?SUBOPTION, {SubPid, Topic}) of false -> ok = emqx_broker_helper:monitor(SubPid, SubId), - Group = maps:get(share, SubOpts, undefined), %% true = ets:insert(?SUBID, {SubId, SubPid}), true = ets:insert(?SUBSCRIPTION, {SubPid, Topic}), - %% SeqId = emqx_broker_helper:create_seq(Topic), - true = ets:insert(?SUBSCRIBER, {Topic, shared(Group, SubPid)}), - true = ets:insert(?SUBOPTION, {{SubPid, Topic}, SubOpts}), - ok = emqx_shared_sub:subscribe(Group, Topic, SubPid), - call(pick(Topic), {subscribe, Group, Topic}); + case maps:get(share, SubOpts, undefined) of + undefined -> + Shard = emqx_broker_helper:get_shard(SubPid, Topic), + case Shard of + 0 -> true = ets:insert(?SUBSCRIBER, {Topic, SubPid}); + I -> true = ets:insert(?SUBSCRIBER, {{shard, Topic, I}, SubPid}), + true = ets:insert(?SUBSCRIBER, {Topic, {shard, I}}) + end, + SubOpts1 = maps:put(shard, Shard, SubOpts), + true = ets:insert(?SUBOPTION, {{SubPid, Topic}, SubOpts1}), + call(pick({Topic, Shard}), {subscribe, Topic}); + Group -> %% Shared subscription + true = ets:insert(?SUBOPTION, {{SubPid, Topic}, SubOpts}), + emqx_shared_sub:subscribe(Group, Topic, SubPid) + end; true -> ok end. @@ -112,12 +122,21 @@ unsubscribe(Topic) when is_binary(Topic) -> SubPid = self(), case ets:lookup(?SUBOPTION, {SubPid, Topic}) of [{_, SubOpts}] -> - Group = maps:get(share, SubOpts, undefined), + _ = emqx_broker_helper:reclaim_seq(Topic), + case maps:get(share, SubOpts, undefined) of + undefined -> + case maps:get(shared, SubOpts, 0) of + 0 -> true = ets:delete_object(?SUBSCRIBER, {Topic, SubPid}), + ok = cast(pick(Topic), {unsubscribed, Topic}); + I -> true = ets:delete_object(?SUBSCRIBER, {{shard, Topic, I}, SubPid}), + ok = cast(pick({Topic, I}), {unsubscribed, Topic, I}) + end; + Group -> + ok = emqx_shared_sub:unsubscribe(Group, Topic, SubPid) + end, true = ets:delete_object(?SUBSCRIPTION, {SubPid, Topic}), - true = ets:delete_object(?SUBSCRIBER, {Topic, shared(Group, SubPid)}), - true = ets:delete(?SUBOPTION, {SubPid, Topic}), - ok = emqx_shared_sub:unsubscribe(Group, Topic, SubPid), - call(pick(Topic), {unsubscribe, Group, Topic}); + %%true = ets:delete_object(?SUBID, {SubId, SubPid}), + true = ets:delete(?SUBOPTION, {SubPid, Topic}); [] -> ok end. @@ -207,22 +226,23 @@ dispatch(Topic, Delivery = #delivery{message = Msg, results = Results}) -> emqx_hooks:run('message.dropped', [#{node => node()}, Msg]), inc_dropped_cnt(Topic), Delivery; - [SubPid] -> %% optimize? - dispatch(SubPid, Topic, Msg), + [Sub] -> %% optimize? + dispatch(Sub, Topic, Msg), Delivery#delivery{results = [{dispatch, Topic, 1}|Results]}; - SubPids -> - Count = lists:foldl(fun(SubPid, Acc) -> - dispatch(SubPid, Topic, Msg), Acc + 1 - end, 0, SubPids), + Subs -> + Count = lists:foldl( + fun(Sub, Acc) -> + dispatch(Sub, Topic, Msg), Acc + 1 + end, 0, Subs), Delivery#delivery{results = [{dispatch, Topic, Count}|Results]} end. dispatch(SubPid, Topic, Msg) when is_pid(SubPid) -> - SubPid ! {dispatch, Topic, Msg}, - true; -%% TODO: how to optimize the share sub? -dispatch({share, _Group, _SubPid}, _Topic, _Msg) -> - false. + SubPid ! {dispatch, Topic, Msg}; +dispatch({shard, I}, Topic, Msg) -> + lists:foreach(fun(SubPid) -> + SubPid ! {dispatch, Topic, Msg} + end, safe_lookup_element(?SUBSCRIBER, {share, Topic, I}, [])). inc_dropped_cnt(<<"$SYS/", _/binary>>) -> ok; @@ -240,17 +260,20 @@ subscribers(Topic) -> -spec(subscriber_down(pid()) -> true). subscriber_down(SubPid) -> lists:foreach( - fun(Sub = {_, Topic}) -> + fun(Sub = {_Pid, Topic}) -> case ets:lookup(?SUBOPTION, Sub) of [{_, SubOpts}] -> - Group = maps:get(share, SubOpts, undefined), - true = ets:delete_object(?SUBSCRIBER, {Topic, shared(Group, SubPid)}), - true = ets:delete(?SUBOPTION, Sub), - gen_server:cast(pick(Topic), {unsubscribe, Group, Topic}); + _ = emqx_broker_helper:reclaim_seq(Topic), + case maps:get(shared, SubOpts, 0) of + 0 -> true = ets:delete_object(?SUBSCRIBER, {Topic, SubPid}), + ok = cast(pick(Topic), {unsubscribed, Topic}); + I -> true = ets:delete_object(?SUBSCRIBER, {{shard, Topic, I}, SubPid}), + ok = cast(pick({Topic, I}), {unsubscribed, Topic, I}) + end; [] -> ok end end, ets:lookup(?SUBSCRIPTION, SubPid)), - ets:delete(?SUBSCRIPTION, SubPid). + true = ets:delete(?SUBSCRIPTION, SubPid). %%------------------------------------------------------------------------------ %% Management APIs @@ -305,11 +328,14 @@ safe_update_stats(Tab, Stat, MaxStat) -> end. %%------------------------------------------------------------------------------ -%% Pick and call +%% call, cast, pick %%------------------------------------------------------------------------------ call(Broker, Req) -> - gen_server:call(Broker, Req, ?TIMEOUT). + gen_server:call(Broker, Req). + +cast(Broker, Msg) -> + gen_server:cast(Broker, Msg). %% Pick a broker pick(Topic) -> @@ -320,24 +346,41 @@ pick(Topic) -> %%------------------------------------------------------------------------------ init([Pool, Id]) -> + _ = emqx_router:set_mode(protected), true = gproc_pool:connect_worker(Pool, {Pool, Id}), {ok, #{pool => Pool, id => Id}}. -handle_call({subscribe, Group, Topic}, _From, State) -> - Ok = emqx_router:add_route(Topic, dest(Group)), - {reply, Ok, State}; - -handle_call({unsubscribe, Group, Topic}, _From, State) -> - Ok = case ets:member(?SUBSCRIBER, Topic) of - false -> emqx_router:delete_route(Topic, dest(Group)); - true -> ok - end, - {reply, Ok, State}; +handle_call({subscribe, Topic}, _From, State) -> + case get(Topic) of + undefined -> + _ = put(Topic, true), + emqx_router:add_route(Topic); + true -> ok + end, + {reply, ok, State}; handle_call(Req, _From, State) -> emqx_logger:error("[Broker] unexpected call: ~p", [Req]), {reply, ignored, State}. +handle_cast({unsubscribed, Topic}, State) -> + case ets:member(?SUBSCRIBER, Topic) of + false -> + _ = erase(Topic), + emqx_router:delete_route(Topic); + true -> ok + end, + {noreply, State}; + +handle_cast({unsubscribed, Topic, I}, State) -> + case ets:member(?SUBSCRIBER, {shard, Topic, I}) of + false -> + true = ets:delete_object(?SUBSCRIBER, {Topic, {shard, I}}), + cast(pick(Topic), {unsubscribed, Topic}); + true -> ok + end, + {noreply, State}; + handle_cast(Msg, State) -> emqx_logger:error("[Broker] unexpected cast: ~p", [Msg]), {noreply, State}. @@ -356,9 +399,3 @@ code_change(_OldVsn, State, _Extra) -> %% Internal functions %%------------------------------------------------------------------------------ -dest(undefined) -> node(); -dest(Group) -> {Group, node()}. - -shared(undefined, Name) -> Name; -shared(Group, Name) -> {share, Group, Name}. - diff --git a/src/emqx_broker_helper.erl b/src/emqx_broker_helper.erl index 35fe06f0d..d3e7f9d37 100644 --- a/src/emqx_broker_helper.erl +++ b/src/emqx_broker_helper.erl @@ -20,6 +20,7 @@ -export([start_link/0]). -export([monitor/2]). +-export([get_shard/2]). -export([create_seq/1, reclaim_seq/1]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, @@ -46,6 +47,13 @@ monitor(SubPid, SubId) when is_pid(SubPid) -> error(subid_conflict) end. +-spec(get_shard(pid(), emqx_topic:topic()) -> non_neg_integer()). +get_shard(SubPid, Topic) -> + case create_seq(Topic) of + Seq when Seq =< 1024 -> 0; + _Seq -> erlang:phash2(SubPid, ets:lookup_element(?SUBSEQ, shards, 2)) + end. + -spec(create_seq(emqx_topic:topic()) -> emqx_sequence:seqid()). create_seq(Topic) -> emqx_sequence:nextval(?SUBSEQ, Topic). @@ -60,9 +68,11 @@ reclaim_seq(Topic) -> init([]) -> %% SubSeq: Topic -> SeqId - _ = emqx_sequence:create(?SUBSEQ), + ok = emqx_sequence:create(?SUBSEQ), + %% Shards: CPU * 32 + true = ets:insert(?SUBSEQ, {shards, emqx_vm:schedulers() * 32}), %% SubMon: SubPid -> SubId - _ = emqx_tables:new(?SUBMON, [set, protected, {read_concurrency, true}]), + ok = emqx_tables:new(?SUBMON, [set, protected, {read_concurrency, true}]), %% Stats timer emqx_stats:update_interval(broker_stats, fun emqx_broker:stats_fun/0), {ok, #state{pmon = emqx_pmon:new()}, hibernate}. diff --git a/src/emqx_cm.erl b/src/emqx_cm.erl index 19892b386..6756cf02b 100644 --- a/src/emqx_cm.erl +++ b/src/emqx_cm.erl @@ -125,9 +125,9 @@ notify(Msg) -> init([]) -> TabOpts = [public, set, {write_concurrency, true}], - _ = emqx_tables:new(?CONN_TAB, [{read_concurrency, true} | TabOpts]), - _ = emqx_tables:new(?CONN_ATTRS_TAB, TabOpts), - _ = emqx_tables:new(?CONN_STATS_TAB, TabOpts), + ok = emqx_tables:new(?CONN_TAB, [{read_concurrency, true} | TabOpts]), + ok = emqx_tables:new(?CONN_ATTRS_TAB, TabOpts), + ok = emqx_tables:new(?CONN_STATS_TAB, TabOpts), ok = emqx_stats:update_interval(cm_stats, fun ?MODULE:update_conn_stats/0), {ok, #{conn_pmon => emqx_pmon:new()}}. diff --git a/src/emqx_ctl.erl b/src/emqx_ctl.erl index 17166a014..c00556eb7 100644 --- a/src/emqx_ctl.erl +++ b/src/emqx_ctl.erl @@ -96,7 +96,7 @@ usage() -> %%------------------------------------------------------------------------------ init([]) -> - _ = emqx_tables:new(?TAB, [ordered_set, protected]), + ok = emqx_tables:new(?TAB, [protected, ordered_set]), {ok, #state{seq = 0}}. handle_call(Req, _From, State) -> diff --git a/src/emqx_hooks.erl b/src/emqx_hooks.erl index 073c12870..b2eb0d6f4 100644 --- a/src/emqx_hooks.erl +++ b/src/emqx_hooks.erl @@ -139,7 +139,7 @@ lookup(HookPoint) -> %%------------------------------------------------------------------------------ init([]) -> - _ = emqx_tables:new(?TAB, [{keypos, #hook.name}, {read_concurrency, true}]), + ok = emqx_tables:new(?TAB, [{keypos, #hook.name}, {read_concurrency, true}]), {ok, #{}}. handle_call({add, HookPoint, Callback = #callback{action = Action}}, _From, State) -> diff --git a/src/emqx_metrics.erl b/src/emqx_metrics.erl index caf862146..b4b3a1307 100644 --- a/src/emqx_metrics.erl +++ b/src/emqx_metrics.erl @@ -285,7 +285,7 @@ qos_sent(?QOS_2) -> init([]) -> % Create metrics table - _ = emqx_tables:new(?TAB, [set, public, {write_concurrency, true}]), + ok = emqx_tables:new(?TAB, [public, set, {write_concurrency, true}]), lists:foreach(fun new/1, ?BYTES_METRICS ++ ?PACKET_METRICS ++ ?MESSAGE_METRICS), {ok, #{}, hibernate}. diff --git a/src/emqx_router.erl b/src/emqx_router.erl index 941c004f7..313adc475 100644 --- a/src/emqx_router.erl +++ b/src/emqx_router.erl @@ -28,23 +28,22 @@ -export([start_link/2]). %% Route APIs --export([add_route/1, add_route/2, add_route/3]). +-export([add_route/1, add_route/2]). -export([get_routes/1]). --export([del_route/1, del_route/2, del_route/3]). +-export([delete_route/1, delete_route/2]). -export([has_routes/1, match_routes/1, print_routes/1]). -export([topics/0]). + +%% Mode +-export([set_mode/1, get_mode/0]). + %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -type(destination() :: node() | {binary(), node()}). --record(batch, {enabled, timer, pending}). --record(state, {pool, id, batch :: #batch{}}). - -define(ROUTE, emqx_route). --define(BATCH(Enabled), #batch{enabled = Enabled}). --define(BATCH(Enabled, Pending), #batch{enabled = Enabled, pending = Pending}). %%------------------------------------------------------------------------------ %% Mnesia bootstrap @@ -62,49 +61,66 @@ mnesia(copy) -> ok = ekka_mnesia:copy_table(?ROUTE). %%------------------------------------------------------------------------------ -%% Strat a router +%% Start a router %%------------------------------------------------------------------------------ --spec(start_link(atom(), pos_integer()) -> {ok, pid()} | ignore | {error, term()}). +-spec(start_link(atom(), pos_integer()) -> emqx_types:startlink_ret()). start_link(Pool, Id) -> - gen_server:start_link({local, emqx_misc:proc_name(?MODULE, Id)}, - ?MODULE, [Pool, Id], [{hibernate_after, 2000}]). + Name = emqx_misc:proc_name(?MODULE, Id), + gen_server:start_link({local, Name}, ?MODULE, [Pool, Id], [{hibernate_after, 1000}]). %%------------------------------------------------------------------------------ %% Route APIs %%------------------------------------------------------------------------------ --spec(add_route(emqx_topic:topic() | emqx_types:route()) -> ok). +-spec(add_route(emqx_topic:topic() | emqx_types:route()) -> ok | {error, term()}). add_route(Topic) when is_binary(Topic) -> add_route(#route{topic = Topic, dest = node()}); add_route(Route = #route{topic = Topic}) -> - cast(pick(Topic), {add_route, Route}). + case get_mode() of + protected -> do_add_route(Route); + undefined -> call(pick(Topic), {add_route, Route}) + end. --spec(add_route(emqx_topic:topic(), destination()) -> ok). +-spec(add_route(emqx_topic:topic(), destination()) -> ok | {error, term()}). add_route(Topic, Dest) when is_binary(Topic) -> add_route(#route{topic = Topic, dest = Dest}). --spec(add_route({pid(), reference()}, emqx_topic:topic(), destination()) -> ok). -add_route(From, Topic, Dest) when is_binary(Topic) -> - cast(pick(Topic), {add_route, From, #route{topic = Topic, dest = Dest}}). +%% @private +do_add_route(Route = #route{topic = Topic, dest = Dest}) -> + case lists:member(Route, get_routes(Topic)) of + true -> ok; + false -> + ok = emqx_router_helper:monitor(Dest), + case emqx_topic:wildcard(Topic) of + true -> trans(fun add_trie_route/1, [Route]); + false -> add_direct_route(Route) + end + end. -spec(get_routes(emqx_topic:topic()) -> [emqx_types:route()]). get_routes(Topic) -> ets:lookup(?ROUTE, Topic). --spec(del_route(emqx_topic:topic() | emqx_types:route()) -> ok). -del_route(Topic) when is_binary(Topic) -> - del_route(#route{topic = Topic, dest = node()}); -del_route(Route = #route{topic = Topic}) -> - cast(pick(Topic), {del_route, Route}). +-spec(delete_route(emqx_topic:topic() | emqx_types:route()) -> ok | {error, term()}). +delete_route(Topic) when is_binary(Topic) -> + delete_route(#route{topic = Topic, dest = node()}); +delete_route(Route = #route{topic = Topic}) -> + case get_mode() of + protected -> do_delete_route(Route); + undefined -> call(pick(Topic), {delete_route, Route}) + end. --spec(del_route(emqx_topic:topic(), destination()) -> ok). -del_route(Topic, Dest) when is_binary(Topic) -> - del_route(#route{topic = Topic, dest = Dest}). +-spec(delete_route(emqx_topic:topic(), destination()) -> ok | {error, term()}). +delete_route(Topic, Dest) when is_binary(Topic) -> + delete_route(#route{topic = Topic, dest = Dest}). --spec(del_route({pid(), reference()}, emqx_topic:topic(), destination()) -> ok). -del_route(From, Topic, Dest) when is_binary(Topic) -> - cast(pick(Topic), {del_route, From, #route{topic = Topic, dest = Dest}}). +%% @private +do_delete_route(Route = #route{topic = Topic}) -> + case emqx_topic:wildcard(Topic) of + true -> trans(fun del_trie_route/1, [Route]); + false -> del_direct_route(Route) + end. -spec(has_routes(emqx_topic:topic()) -> boolean()). has_routes(Topic) when is_binary(Topic) -> @@ -127,8 +143,15 @@ print_routes(Topic) -> io:format("~s -> ~s~n", [To, Dest]) end, match_routes(Topic)). -cast(Router, Msg) -> - gen_server:cast(Router, Msg). +-spec(set_mode(protected | atom()) -> any()). +set_mode(Mode) when is_atom(Mode) -> + put('$router_mode', Mode). + +-spec(get_mode() -> protected | undefined | atom()). +get_mode() -> get('$router_mode'). + +call(Router, Msg) -> + gen_server:call(Router, Msg, infinity). pick(Topic) -> gproc_pool:pick_worker(router, Topic). @@ -138,71 +161,28 @@ pick(Topic) -> %%------------------------------------------------------------------------------ init([Pool, Id]) -> - rand:seed(exsplus, erlang:timestamp()), - gproc_pool:connect_worker(Pool, {Pool, Id}), - Batch = #batch{enabled = emqx_config:get_env(route_batch_clean, false), - pending = sets:new()}, - {ok, ensure_batch_timer(#state{pool = Pool, id = Id, batch = Batch})}. + true = gproc_pool:connect_worker(Pool, {Pool, Id}), + {ok, #{pool => Pool, id => Id}}. + +handle_call({add_route, Route}, _From, State) -> + {reply, do_add_route(Route), State}; + +handle_call({delete_route, Route}, _From, State) -> + {reply, do_delete_route(Route), State}; handle_call(Req, _From, State) -> emqx_logger:error("[Router] unexpected call: ~p", [Req]), {reply, ignored, State}. -handle_cast({add_route, From, Route}, State) -> - {noreply, NewState} = handle_cast({add_route, Route}, State), - _ = gen_server:reply(From, ok), - {noreply, NewState}; - -handle_cast({add_route, Route = #route{topic = Topic, dest = Dest}}, State) -> - case lists:member(Route, get_routes(Topic)) of - true -> ok; - false -> - ok = emqx_router_helper:monitor(Dest), - case emqx_topic:wildcard(Topic) of - true -> log(trans(fun add_trie_route/1, [Route])); - false -> add_direct_route(Route) - end - end, - {noreply, State}; - -handle_cast({del_route, From, Route}, State) -> - {noreply, NewState} = handle_cast({del_route, Route}, State), - _ = gen_server:reply(From, ok), - {noreply, NewState}; - -handle_cast({del_route, Route = #route{topic = Topic, dest = Dest}}, State) when is_tuple(Dest) -> - {noreply, case emqx_topic:wildcard(Topic) of - true -> log(trans(fun del_trie_route/1, [Route])), - State; - false -> del_direct_route(Route, State) - end}; - -handle_cast({del_route, Route = #route{topic = Topic}}, State) -> - %% Confirm if there are still subscribers... - {noreply, case ets:member(emqx_subscriber, Topic) of - true -> State; - false -> - case emqx_topic:wildcard(Topic) of - true -> log(trans(fun del_trie_route/1, [Route])), - State; - false -> del_direct_route(Route, State) - end - end}; - handle_cast(Msg, State) -> emqx_logger:error("[Router] unexpected cast: ~p", [Msg]), {noreply, State}. -handle_info({timeout, _TRef, batch_delete}, State = #state{batch = Batch}) -> - _ = del_direct_routes(sets:to_list(Batch#batch.pending)), - {noreply, ensure_batch_timer(State#state{batch = ?BATCH(true, sets:new())}), hibernate}; - handle_info(Info, State) -> emqx_logger:error("[Router] unexpected info: ~p", [Info]), {noreply, State}. -terminate(_Reason, #state{pool = Pool, id = Id, batch = Batch}) -> - _ = cacel_batch_timer(Batch), +terminate(_Reason, #{pool := Pool, id := Id}) -> gproc_pool:disconnect_worker(Pool, {Pool, Id}). code_change(_OldVsn, State, _Extra) -> @@ -212,17 +192,6 @@ code_change(_OldVsn, State, _Extra) -> %% Internal functions %%------------------------------------------------------------------------------ -ensure_batch_timer(State = #state{batch = #batch{enabled = false}}) -> - State; -ensure_batch_timer(State = #state{batch = Batch}) -> - TRef = erlang:start_timer(50 + rand:uniform(50), self(), batch_delete), - State#state{batch = Batch#batch{timer = TRef}}. - -cacel_batch_timer(#batch{enabled = false}) -> - ok; -cacel_batch_timer(#batch{enabled = true, timer = TRef}) -> - catch erlang:cancel_timer(TRef). - add_direct_route(Route) -> mnesia:async_dirty(fun mnesia:write/3, [?ROUTE, Route, sticky_write]). @@ -233,25 +202,9 @@ add_trie_route(Route = #route{topic = Topic}) -> end, mnesia:write(?ROUTE, Route, sticky_write). -del_direct_route(Route, State = #state{batch = ?BATCH(false)}) -> - del_direct_route(Route), State; -del_direct_route(Route, State = #state{batch = Batch = ?BATCH(true, Pending)}) -> - State#state{batch = Batch#batch{pending = sets:add_element(Route, Pending)}}. - del_direct_route(Route) -> mnesia:async_dirty(fun mnesia:delete_object/3, [?ROUTE, Route, sticky_write]). -del_direct_routes([]) -> - ok; -del_direct_routes(Routes) -> - DelFun = fun(R = #route{topic = Topic}) -> - case ets:member(emqx_subscriber, Topic) of - true -> ok; - false -> mnesia:delete_object(?ROUTE, R, sticky_write) - end - end, - mnesia:async_dirty(fun lists:foreach/2, [DelFun, Routes]). - del_trie_route(Route = #route{topic = Topic}) -> case mnesia:wread({?ROUTE, Topic}) of [Route] -> %% Remove route and trie @@ -270,7 +223,3 @@ trans(Fun, Args) -> {aborted, Error} -> {error, Error} end. -log(ok) -> ok; -log({error, Reason}) -> - emqx_logger:error("[Router] mnesia aborted: ~p", [Reason]). - diff --git a/src/emqx_sequence.erl b/src/emqx_sequence.erl index c4fc0fd08..022531df5 100644 --- a/src/emqx_sequence.erl +++ b/src/emqx_sequence.erl @@ -25,8 +25,7 @@ %% @doc Create a sequence. -spec(create(name()) -> ok). create(Name) -> - _ = ets:new(Name, [set, public, named_table, {write_concurrency, true}]), - ok. + emqx_tables:new(Name, [public, set, {write_concurrency, true}]). %% @doc Next value of the sequence. -spec(nextval(name(), key()) -> seqid()). diff --git a/src/emqx_shared_sub.erl b/src/emqx_shared_sub.erl index ebe6d51f8..d1d0d921d 100644 --- a/src/emqx_shared_sub.erl +++ b/src/emqx_shared_sub.erl @@ -17,6 +17,7 @@ -behaviour(gen_server). -include("emqx.hrl"). +-include("emqx_mqtt.hrl"). %% Mnesia bootstrap -export([mnesia/1]). @@ -27,7 +28,8 @@ -export([start_link/0]). -export([subscribe/3, unsubscribe/3]). --export([dispatch/3, maybe_ack/1, maybe_nack_dropped/1, nack_no_connection/1, is_ack_required/1]). +-export([dispatch/3]). +-export([maybe_ack/1, maybe_nack_dropped/1, nack_no_connection/1, is_ack_required/1]). %% for testing -export([subscribers/2]). @@ -38,6 +40,7 @@ -define(SERVER, ?MODULE). -define(TAB, emqx_shared_subscription). +-define(SHARED_SUBS, emqx_shared_subscriber). -define(ALIVE_SUBS, emqx_alive_shared_subscribers). -define(SHARED_SUB_QOS1_DISPATCH_TIMEOUT_SECONDS, 5). -define(ack, shared_sub_ack). @@ -48,8 +51,6 @@ -record(state, {pmon}). -record(emqx_shared_subscription, {group, topic, subpid}). --include("emqx_mqtt.hrl"). - %%------------------------------------------------------------------------------ %% Mnesia bootstrap %%------------------------------------------------------------------------------ @@ -72,16 +73,11 @@ mnesia(copy) -> start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). -subscribe(undefined, _Topic, _SubPid) -> - ok; subscribe(Group, Topic, SubPid) when is_pid(SubPid) -> - mnesia:dirty_write(?TAB, record(Group, Topic, SubPid)), - gen_server:cast(?SERVER, {monitor, SubPid}). + gen_server:call(?SERVER, {subscribe, Group, Topic, SubPid}). -unsubscribe(undefined, _Topic, _SubPid) -> - ok; unsubscribe(Group, Topic, SubPid) when is_pid(SubPid) -> - mnesia:dirty_delete_object(?TAB, record(Group, Topic, SubPid)). + gen_server:call(?SERVER, {unsubscribe, Group, Topic, SubPid}). record(Group, Topic, SubPid) -> #emqx_shared_subscription{group = Group, topic = Topic, subpid = SubPid}. @@ -251,14 +247,16 @@ do_pick_subscriber(Group, Topic, round_robin, _ClientId, Count) -> subscribers(Group, Topic) -> ets:select(?TAB, [{{emqx_shared_subscription, Group, Topic, '$1'}, [], ['$1']}]). -%%----------------------------------------------------------------------------- +%%------------------------------------------------------------------------------ %% gen_server callbacks -%%----------------------------------------------------------------------------- +%%------------------------------------------------------------------------------ init([]) -> - {atomic, PMon} = mnesia:transaction(fun init_monitors/0), + _ = emqx_router:set_mode(protected), mnesia:subscribe({table, ?TAB, simple}), - ets:new(?ALIVE_SUBS, [named_table, {read_concurrency, true}, protected]), + {atomic, PMon} = mnesia:transaction(fun init_monitors/0), + ok = emqx_tables:new(?SHARED_SUBS, [protected, bag]), + ok = emqx_tables:new(?ALIVE_SUBS, [protected, set, {read_concurrency, true}]), {ok, update_stats(#state{pmon = PMon})}. init_monitors() -> @@ -267,14 +265,29 @@ init_monitors() -> emqx_pmon:monitor(SubPid, Mon) end, emqx_pmon:new(), ?TAB). +handle_call({subscribe, Group, Topic, SubPid}, _From, State = #state{pmon = PMon}) -> + mnesia:dirty_write(?TAB, record(Group, Topic, SubPid)), + case ets:member(?SHARED_SUBS, {Group, Topic}) of + true -> ok; + false -> ok = emqx_router:add_route(Topic, {Group, node()}) + end, + ok = maybe_insert_alive_tab(SubPid), + true = ets:insert(?SHARED_SUBS, {{Group, Topic}, SubPid}), + {reply, ok, update_stats(State#state{pmon = emqx_pmon:monitor(SubPid, PMon)})}; + +handle_call({unsubscribe, Group, Topic, SubPid}, _From, State) -> + mnesia:dirty_delete_object(?TAB, record(Group, Topic, SubPid)), + true = ets:delete_object(?SHARED_SUBS, {{Group, Topic}, SubPid}), + case ets:member(?SHARED_SUBS, {Group, Topic}) of + true -> ok; + false -> ok = emqx_router:delete_route(Topic, {Group, node()}) + end, + {reply, ok, State}; + handle_call(Req, _From, State) -> emqx_logger:error("[SharedSub] unexpected call: ~p", [Req]), {reply, ignored, State}. -handle_cast({monitor, SubPid}, State= #state{pmon = PMon}) -> - NewPmon = emqx_pmon:monitor(SubPid, PMon), - ok = maybe_insert_alive_tab(SubPid), - {noreply, update_stats(State#state{pmon = NewPmon})}; handle_cast(Msg, State) -> emqx_logger:error("[SharedSub] unexpected cast: ~p", [Msg]), {noreply, State}. @@ -316,12 +329,18 @@ maybe_insert_alive_tab(Pid) when is_pid(Pid) -> ets:insert(?ALIVE_SUBS, {Pid}), cleanup_down(SubPid) -> ?IS_LOCAL_PID(SubPid) orelse ets:delete(?ALIVE_SUBS, SubPid), lists:foreach( - fun(Record) -> - mnesia:dirty_delete_object(?TAB, Record) - end,mnesia:dirty_match_object(#emqx_shared_subscription{_ = '_', subpid = SubPid})). + fun(Record = #emqx_shared_subscription{topic = Topic, group = Group}) -> + ok = mnesia:dirty_delete_object(?TAB, Record), + true = ets:delete_object(?SHARED_SUBS, {{Group, Topic}, SubPid}), + case ets:member(?SHARED_SUBS, {Group, Topic}) of + true -> ok; + false -> ok = emqx_router:delete_route(Topic, {Group, node()}) + end + end, mnesia:dirty_match_object(#emqx_shared_subscription{_ = '_', subpid = SubPid})). update_stats(State) -> - emqx_stats:setstat('subscriptions/shared/count', 'subscriptions/shared/max', ets:info(?TAB, size)), State. + emqx_stats:setstat('subscriptions/shared/count', 'subscriptions/shared/max', ets:info(?TAB, size)), + State. %% Return 'true' if the subscriber process is alive AND not in the failed list is_active_sub(Pid, FailedSubs) -> diff --git a/src/emqx_sm.erl b/src/emqx_sm.erl index 8f9a3e3cb..d178a8ae7 100644 --- a/src/emqx_sm.erl +++ b/src/emqx_sm.erl @@ -202,10 +202,10 @@ notify(Event) -> init([]) -> TabOpts = [public, set, {write_concurrency, true}], - _ = emqx_tables:new(?SESSION_TAB, [{read_concurrency, true} | TabOpts]), - _ = emqx_tables:new(?SESSION_P_TAB, TabOpts), - _ = emqx_tables:new(?SESSION_ATTRS_TAB, TabOpts), - _ = emqx_tables:new(?SESSION_STATS_TAB, TabOpts), + ok = emqx_tables:new(?SESSION_TAB, [{read_concurrency, true} | TabOpts]), + ok = emqx_tables:new(?SESSION_P_TAB, TabOpts), + ok = emqx_tables:new(?SESSION_ATTRS_TAB, TabOpts), + ok = emqx_tables:new(?SESSION_STATS_TAB, TabOpts), emqx_stats:update_interval(sm_stats, fun ?MODULE:stats_fun/0), {ok, #{session_pmon => emqx_pmon:new()}}. diff --git a/src/emqx_stats.erl b/src/emqx_stats.erl index 61ff6cbc3..790c397b9 100644 --- a/src/emqx_stats.erl +++ b/src/emqx_stats.erl @@ -152,7 +152,7 @@ cast(Msg) -> %%------------------------------------------------------------------------------ init(#{tick_ms := TickMs}) -> - _ = emqx_tables:new(?TAB, [set, public, {write_concurrency, true}]), + ok = emqx_tables:new(?TAB, [public, set, {write_concurrency, true}]), Stats = lists:append([?CONNECTION_STATS, ?SESSION_STATS, ?PUBSUB_STATS, ?ROUTE_STATS, ?RETAINED_STATS]), true = ets:insert(?TAB, [{Name, 0} || Name <- Stats]), diff --git a/src/emqx_tables.erl b/src/emqx_tables.erl index 330c87d9c..9b3ebfeae 100644 --- a/src/emqx_tables.erl +++ b/src/emqx_tables.erl @@ -17,10 +17,12 @@ -export([new/2]). %% Create a named_table ets. +-spec(new(atom(), list()) -> ok). new(Tab, Opts) -> case ets:info(Tab, name) of undefined -> - ets:new(Tab, lists:usort([named_table | Opts])); - Tab -> Tab + _ = ets:new(Tab, lists:usort([named_table | Opts])), + ok; + Tab -> ok end. diff --git a/src/emqx_zone.erl b/src/emqx_zone.erl index dd183dbdf..d119abe52 100644 --- a/src/emqx_zone.erl +++ b/src/emqx_zone.erl @@ -68,7 +68,7 @@ stop() -> %%------------------------------------------------------------------------------ init([]) -> - _ = emqx_tables:new(?TAB, [set, {read_concurrency, true}]), + ok = emqx_tables:new(?TAB, [set, {read_concurrency, true}]), {ok, element(2, handle_info(reload, #{timer => undefined}))}. handle_call(force_reload, _From, State) ->