diff --git a/src/emqx.erl b/src/emqx.erl index 72f1d6f81..3792cc4f8 100644 --- a/src/emqx.erl +++ b/src/emqx.erl @@ -26,7 +26,6 @@ %% PubSub management API -export([topics/0, subscriptions/1, subscribers/1, subscribed/2]). --export([get_subopts/2, set_subopts/3]). %% Hooks API -export([hook/2, hook/3, hook/4, unhook/2, run_hooks/2, run_hooks/3]). @@ -70,20 +69,18 @@ is_running(Node) -> subscribe(Topic) -> emqx_broker:subscribe(iolist_to_binary(Topic)). --spec(subscribe(emqx_topic:topic() | string(), emqx_types:subid() | pid()) -> ok). +-spec(subscribe(emqx_topic:topic() | string(), emqx_types:subid() | emqx_types:subopts()) -> ok). subscribe(Topic, SubId) when is_atom(SubId); is_binary(SubId)-> emqx_broker:subscribe(iolist_to_binary(Topic), SubId); -subscribe(Topic, SubPid) when is_pid(SubPid) -> - emqx_broker:subscribe(iolist_to_binary(Topic), SubPid). +subscribe(Topic, SubOpts) when is_map(SubOpts) -> + emqx_broker:subscribe(iolist_to_binary(Topic), SubOpts). --spec(subscribe(emqx_topic:topic() | string(), emqx_types:subid() | pid(), - emqx_types:subopts()) -> ok). -subscribe(Topic, SubId, Options) when is_atom(SubId); is_binary(SubId)-> - emqx_broker:subscribe(iolist_to_binary(Topic), SubId, Options); -subscribe(Topic, SubPid, Options) when is_pid(SubPid)-> - emqx_broker:subscribe(iolist_to_binary(Topic), SubPid, Options). +-spec(subscribe(emqx_topic:topic() | string(), + emqx_types:subid() | pid(), emqx_types:subopts()) -> ok). +subscribe(Topic, SubId, SubOpts) when (is_atom(SubId) orelse is_binary(SubId)), is_map(SubOpts) -> + emqx_broker:subscribe(iolist_to_binary(Topic), SubId, SubOpts). --spec(publish(emqx_types:message()) -> {ok, emqx_types:deliver_results()}). +-spec(publish(emqx_types:message()) -> emqx_types:deliver_results()). publish(Msg) -> emqx_broker:publish(Msg). @@ -91,26 +88,14 @@ publish(Msg) -> unsubscribe(Topic) -> emqx_broker:unsubscribe(iolist_to_binary(Topic)). --spec(unsubscribe(emqx_topic:topic() | string(), emqx_types:subid() | pid()) -> ok). -unsubscribe(Topic, SubId) when is_atom(SubId); is_binary(SubId) -> - emqx_broker:unsubscribe(iolist_to_binary(Topic), SubId); -unsubscribe(Topic, SubPid) when is_pid(SubPid) -> - emqx_broker:unsubscribe(iolist_to_binary(Topic), SubPid). +-spec(unsubscribe(emqx_topic:topic() | string(), emqx_types:subid()) -> ok). +unsubscribe(Topic, SubId) -> + emqx_broker:unsubscribe(iolist_to_binary(Topic), SubId). %%------------------------------------------------------------------------------ %% PubSub management API %%------------------------------------------------------------------------------ --spec(get_subopts(emqx_topic:topic() | string(), emqx_types:subscriber()) - -> emqx_types:subopts()). -get_subopts(Topic, Subscriber) -> - emqx_broker:get_subopts(iolist_to_binary(Topic), Subscriber). - --spec(set_subopts(emqx_topic:topic() | string(), emqx_types:subscriber(), - emqx_types:subopts()) -> boolean()). -set_subopts(Topic, Subscriber, Options) when is_map(Options) -> - emqx_broker:set_subopts(iolist_to_binary(Topic), Subscriber, Options). - -spec(topics() -> list(emqx_topic:topic())). topics() -> emqx_router:topics(). @@ -118,15 +103,15 @@ topics() -> emqx_router:topics(). subscribers(Topic) -> emqx_broker:subscribers(iolist_to_binary(Topic)). --spec(subscriptions(emqx_types:subscriber()) -> [{emqx_topic:topic(), emqx_types:subopts()}]). -subscriptions(Subscriber) -> - emqx_broker:subscriptions(Subscriber). +-spec(subscriptions(pid()) -> [{emqx_topic:topic(), emqx_types:subopts()}]). +subscriptions(SubPid) when is_pid(SubPid) -> + emqx_broker:subscriptions(SubPid). --spec(subscribed(emqx_topic:topic() | string(), pid() | emqx_types:subid()) -> boolean()). -subscribed(Topic, SubPid) when is_pid(SubPid) -> - emqx_broker:subscribed(iolist_to_binary(Topic), SubPid); -subscribed(Topic, SubId) when is_atom(SubId); is_binary(SubId) -> - emqx_broker:subscribed(iolist_to_binary(Topic), SubId). +-spec(subscribed(pid() | emqx_types:subid(), emqx_topic:topic() | string()) -> boolean()). +subscribed(SubPid, Topic) when is_pid(SubPid) -> + emqx_broker:subscribed(SubPid, iolist_to_binary(Topic)); +subscribed(SubId, Topic) when is_atom(SubId); is_binary(SubId) -> + emqx_broker:subscribed(SubId, iolist_to_binary(Topic)). %%------------------------------------------------------------------------------ %% Hooks API diff --git a/src/emqx_broker.erl b/src/emqx_broker.erl index ca4a86c87..3a3cde1fa 100644 --- a/src/emqx_broker.erl +++ b/src/emqx_broker.erl @@ -19,150 +19,130 @@ -include("emqx.hrl"). -export([start_link/2]). --export([subscribe/1, subscribe/2, subscribe/3, subscribe/4]). --export([multi_subscribe/1, multi_subscribe/2, multi_subscribe/3]). +-export([subscribe/1, subscribe/2, subscribe/3]). +-export([unsubscribe/1, unsubscribe/2]). +-export([subscriber_down/1]). -export([publish/1, safe_publish/1]). --export([unsubscribe/1, unsubscribe/2, unsubscribe/3]). --export([multi_unsubscribe/1, multi_unsubscribe/2, multi_unsubscribe/3]). -export([dispatch/2, dispatch/3]). -export([subscriptions/1, subscribers/1, subscribed/2]). --export([get_subopts/2, set_subopts/3]). +-export([get_subopts/2, set_subopts/2]). -export([topics/0]). +%% Stats fun +-export([stats_fun/0]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). --ifdef(TEST). --compile(export_all). --compile(nowarn_export_all). --endif. - --record(state, {pool, id, submap, submon}). --record(subscribe, {topic, subpid, subid, subopts = #{}}). --record(unsubscribe, {topic, subpid, subid}). - -%% The default request timeout +-define(SHARD, 1024). -define(TIMEOUT, 60000). -define(BROKER, ?MODULE). %% ETS tables --define(SUBOPTION, emqx_suboption). --define(SUBSCRIBER, emqx_subscriber). +-define(SUBID, emqx_subid). +-define(SUBOPTION, emqx_suboption). +-define(SUBSCRIBER, emqx_subscriber). -define(SUBSCRIPTION, emqx_subscription). +%% Gards -define(is_subid(Id), (is_binary(Id) orelse is_atom(Id))). --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}]). + _ = create_tabs(), + gen_server:start_link({local, emqx_misc:proc_name(?BROKER, Id)}, ?MODULE, [Pool, Id], []). %%------------------------------------------------------------------------------ -%% Subscribe +%% Create tabs +%%------------------------------------------------------------------------------ + +create_tabs() -> + TabOpts = [public, {read_concurrency, true}, {write_concurrency, true}], + + %% SubId: SubId -> SubPid1, SubPid2,... + _ = emqx_tables:new(?SUBID, [bag | TabOpts]), + %% SubOption: {SubPid, Topic} -> SubOption + _ = emqx_tables:new(?SUBOPTION, [set | TabOpts]), + + %% Subscription: SubPid -> Topic1, Topic2, Topic3, ... + %% duplicate_bag: o(1) insert + _ = emqx_tables:new(?SUBSCRIPTION, [duplicate_bag | TabOpts]), + + %% Subscriber: Topic -> SubPid1, SubPid2, SubPid3, ... + %% duplicate_bag: o(1) insert + emqx_tables:new(?SUBSCRIBER, [duplicate_bag | TabOpts]). + +%%------------------------------------------------------------------------------ +%% Subscribe API %%------------------------------------------------------------------------------ -spec(subscribe(emqx_topic:topic()) -> ok). subscribe(Topic) when is_binary(Topic) -> - subscribe(Topic, self()). + subscribe(Topic, undefined). --spec(subscribe(emqx_topic:topic(), pid() | emqx_types:subid()) -> ok). -subscribe(Topic, SubPid) when is_binary(Topic), is_pid(SubPid) -> - subscribe(Topic, SubPid, undefined); +-spec(subscribe(emqx_topic:topic(), emqx_types:subid() | emqx_types:subopts()) -> ok). subscribe(Topic, SubId) when is_binary(Topic), ?is_subid(SubId) -> - subscribe(Topic, self(), SubId). + subscribe(Topic, SubId, #{}); +subscribe(Topic, SubOpts) when is_binary(Topic), is_map(SubOpts) -> + subscribe(Topic, undefined, SubOpts). --spec(subscribe(emqx_topic:topic(), pid() | emqx_types:subid(), - emqx_types:subid() | emqx_types:subopts()) -> ok). -subscribe(Topic, SubPid, SubId) when is_binary(Topic), is_pid(SubPid), ?is_subid(SubId) -> - subscribe(Topic, SubPid, SubId, #{qos => 0}); -subscribe(Topic, SubPid, SubOpts) when is_binary(Topic), is_pid(SubPid), is_map(SubOpts) -> - subscribe(Topic, SubPid, undefined, SubOpts); +-spec(subscribe(emqx_topic:topic(), emqx_types:subid(), emqx_types:subopts()) -> ok). subscribe(Topic, SubId, SubOpts) when is_binary(Topic), ?is_subid(SubId), is_map(SubOpts) -> - subscribe(Topic, self(), SubId, SubOpts). - --spec(subscribe(emqx_topic:topic(), pid(), emqx_types:subid(), emqx_types:subopts()) -> ok). -subscribe(Topic, SubPid, SubId, SubOpts) when is_binary(Topic), is_pid(SubPid), - ?is_subid(SubId), is_map(SubOpts) -> - Broker = pick(SubPid), - SubReq = #subscribe{topic = Topic, subpid = SubPid, subid = SubId, subopts = SubOpts}, - wait_for_reply(async_call(Broker, SubReq), ?TIMEOUT). - --spec(multi_subscribe(emqx_types:topic_table()) -> ok). -multi_subscribe(TopicTable) when is_list(TopicTable) -> - multi_subscribe(TopicTable, self()). - --spec(multi_subscribe(emqx_types:topic_table(), pid() | emqx_types:subid()) -> ok). -multi_subscribe(TopicTable, SubPid) when is_pid(SubPid) -> - multi_subscribe(TopicTable, SubPid, undefined); -multi_subscribe(TopicTable, SubId) when ?is_subid(SubId) -> - multi_subscribe(TopicTable, self(), SubId). - --spec(multi_subscribe(emqx_types:topic_table(), pid(), emqx_types:subid()) -> ok). -multi_subscribe(TopicTable, SubPid, SubId) when is_pid(SubPid), ?is_subid(SubId) -> - Broker = pick(SubPid), - SubReq = fun(Topic, SubOpts) -> - #subscribe{topic = Topic, subpid = SubPid, subid = SubId, subopts = SubOpts} - end, - wait_for_replies([async_call(Broker, SubReq(Topic, SubOpts)) - || {Topic, SubOpts} <- TopicTable], ?TIMEOUT). + SubPid = self(), + 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}); + true -> ok + end. %%------------------------------------------------------------------------------ -%% Unsubscribe +%% Unsubscribe API %%------------------------------------------------------------------------------ -spec(unsubscribe(emqx_topic:topic()) -> ok). unsubscribe(Topic) when is_binary(Topic) -> - unsubscribe(Topic, self()). + SubPid = self(), + case ets:lookup(?SUBOPTION, {SubPid, Topic}) of + [{_, SubOpts}] -> + Group = maps:get(share, SubOpts, undefined), + 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}); + [] -> ok + end. --spec(unsubscribe(emqx_topic:topic(), pid() | emqx_types:subid()) -> ok). -unsubscribe(Topic, SubPid) when is_binary(Topic), is_pid(SubPid) -> - unsubscribe(Topic, SubPid, undefined); -unsubscribe(Topic, SubId) when is_binary(Topic), ?is_subid(SubId) -> - unsubscribe(Topic, self(), SubId). - --spec(unsubscribe(emqx_topic:topic(), pid(), emqx_types:subid()) -> ok). -unsubscribe(Topic, SubPid, SubId) when is_binary(Topic), is_pid(SubPid), ?is_subid(SubId) -> - Broker = pick(SubPid), - UnsubReq = #unsubscribe{topic = Topic, subpid = SubPid, subid = SubId}, - wait_for_reply(async_call(Broker, UnsubReq), ?TIMEOUT). - --spec(multi_unsubscribe([emqx_topic:topic()]) -> ok). -multi_unsubscribe(Topics) -> - multi_unsubscribe(Topics, self()). - --spec(multi_unsubscribe([emqx_topic:topic()], pid() | emqx_types:subid()) -> ok). -multi_unsubscribe(Topics, SubPid) when is_pid(SubPid) -> - multi_unsubscribe(Topics, SubPid, undefined); -multi_unsubscribe(Topics, SubId) when ?is_subid(SubId) -> - multi_unsubscribe(Topics, self(), SubId). - --spec(multi_unsubscribe([emqx_topic:topic()], pid(), emqx_types:subid()) -> ok). -multi_unsubscribe(Topics, SubPid, SubId) when is_pid(SubPid), ?is_subid(SubId) -> - Broker = pick(SubPid), - UnsubReq = fun(Topic) -> - #unsubscribe{topic = Topic, subpid = SubPid, subid = SubId} - end, - wait_for_replies([async_call(Broker, UnsubReq(Topic)) || Topic <- Topics], ?TIMEOUT). +-spec(unsubscribe(emqx_topic:topic(), emqx_types:subid()) -> ok). +unsubscribe(Topic, _SubId) when is_binary(Topic) -> + unsubscribe(Topic). %%------------------------------------------------------------------------------ %% Publish %%------------------------------------------------------------------------------ --spec(publish(emqx_types:message()) -> {ok, emqx_types:deliver_results()}). +-spec(publish(emqx_types:message()) -> emqx_types:deliver_results()). publish(Msg) when is_record(Msg, message) -> _ = emqx_tracer:trace(publish, Msg), - {ok, case emqx_hooks:run('message.publish', [], Msg) of - {ok, Msg1 = #message{topic = Topic}} -> - Delivery = route(aggre(emqx_router:match_routes(Topic)), delivery(Msg1)), - Delivery#delivery.results; - {stop, _} -> - emqx_logger:warning("Stop publishing: ~s", [emqx_message:format(Msg)]), - [] - end}. + case emqx_hooks:run('message.publish', [], Msg) of + {ok, Msg1 = #message{topic = Topic}} -> + Delivery = route(aggre(emqx_router:match_routes(Topic)), delivery(Msg1)), + Delivery#delivery.results; + {stop, _} -> + emqx_logger:warning("Stop publishing: ~s", [emqx_message:format(Msg)]), + [] + end. --spec(safe_publish(emqx_types:message()) -> ok). %% Called internally +-spec(safe_publish(emqx_types:message()) -> ok). safe_publish(Msg) when is_record(Msg, message) -> try publish(Msg) @@ -227,98 +207,113 @@ dispatch(Topic, Delivery = #delivery{message = Msg, results = Results}) -> emqx_hooks:run('message.dropped', [#{node => node()}, Msg]), inc_dropped_cnt(Topic), Delivery; - [Sub] -> %% optimize? - dispatch(Sub, Topic, Msg), + [SubPid] -> %% optimize? + dispatch(SubPid, Topic, Msg), Delivery#delivery{results = [{dispatch, Topic, 1}|Results]}; - Subscribers -> - Count = lists:foldl(fun(Sub, Acc) -> - dispatch(Sub, Topic, Msg), Acc + 1 - end, 0, Subscribers), + SubPids -> + Count = lists:foldl(fun(SubPid, Acc) -> + dispatch(SubPid, Topic, Msg), Acc + 1 + end, 0, SubPids), Delivery#delivery{results = [{dispatch, Topic, Count}|Results]} end. -dispatch({SubPid, _SubId}, Topic, Msg) when is_pid(SubPid) -> - SubPid ! {dispatch, Topic, Msg}; -dispatch({share, _Group, _Sub}, _Topic, _Msg) -> - ignored. +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. inc_dropped_cnt(<<"$SYS/", _/binary>>) -> ok; inc_dropped_cnt(_Topic) -> emqx_metrics:inc('messages/dropped'). --spec(subscribers(emqx_topic:topic()) -> [emqx_types:subscriber()]). +-spec(subscribers(emqx_topic:topic()) -> [pid()]). subscribers(Topic) -> - try ets:lookup_element(?SUBSCRIBER, Topic, 2) catch error:badarg -> [] end. + safe_lookup_element(?SUBSCRIBER, Topic, []). --spec(subscriptions(emqx_types:subscriber()) +%%------------------------------------------------------------------------------ +%% Subscriber is down +%%------------------------------------------------------------------------------ + +-spec(subscriber_down(pid()) -> true). +subscriber_down(SubPid) -> + lists:foreach( + fun(Sub = {_, 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}); + [] -> ok + end + end, ets:lookup(?SUBSCRIPTION, SubPid)), + ets:delete(?SUBSCRIPTION, SubPid). + +%%------------------------------------------------------------------------------ +%% Management APIs +%%------------------------------------------------------------------------------ + +-spec(subscriptions(pid() | emqx_types:subid()) -> [{emqx_topic:topic(), emqx_types:subopts()}]). -subscriptions(Subscriber) -> - lists:map(fun({_, {share, _Group, Topic}}) -> - subscription(Topic, Subscriber); - ({_, Topic}) -> - subscription(Topic, Subscriber) - end, ets:lookup(?SUBSCRIPTION, Subscriber)). +subscriptions(SubPid) -> + [{Topic, safe_lookup_element(?SUBOPTION, {SubPid, Topic}, #{})} + || Topic <- safe_lookup_element(?SUBSCRIPTION, SubPid, [])]. -subscription(Topic, Subscriber) -> - {Topic, ets:lookup_element(?SUBOPTION, {Topic, Subscriber}, 2)}. +-spec(subscribed(pid(), emqx_topic:topic()) -> boolean()). +subscribed(SubPid, Topic) when is_pid(SubPid) -> + ets:member(?SUBOPTION, {SubPid, Topic}); +subscribed(SubId, Topic) when ?is_subid(SubId) -> + %%FIXME:... SubId -> SubPid + ets:member(?SUBOPTION, {SubId, Topic}). --spec(subscribed(emqx_topic:topic(), pid() | emqx_types:subid() | emqx_types:subscriber()) -> boolean()). -subscribed(Topic, SubPid) when is_binary(Topic), is_pid(SubPid) -> - case ets:match_object(?SUBOPTION, {{Topic, {SubPid, '_'}}, '_'}, 1) of - {Match, _} -> - length(Match) >= 1; - '$end_of_table' -> - false - end; -subscribed(Topic, SubId) when is_binary(Topic), ?is_subid(SubId) -> - case ets:match_object(?SUBOPTION, {{Topic, {'_', SubId}}, '_'}, 1) of - {Match, _} -> - length(Match) >= 1; - '$end_of_table' -> - false - end; -subscribed(Topic, {SubPid, SubId}) when is_binary(Topic), is_pid(SubPid), ?is_subid(SubId) -> - ets:member(?SUBOPTION, {Topic, {SubPid, SubId}}). +-spec(get_subopts(pid(), emqx_topic:topic()) -> emqx_types:subopts()). +get_subopts(SubPid, Topic) when is_pid(SubPid), is_binary(Topic) -> + safe_lookup_element(?SUBOPTION, {SubPid, Topic}, #{}). --spec(get_subopts(emqx_topic:topic(), emqx_types:subscriber()) -> emqx_types:subopts()). -get_subopts(Topic, Subscriber) when is_binary(Topic) -> - try ets:lookup_element(?SUBOPTION, {Topic, Subscriber}, 2) - catch error:badarg -> [] - end. - --spec(set_subopts(emqx_topic:topic(), emqx_types:subscriber(), emqx_types:subopts()) -> boolean()). -set_subopts(Topic, Subscriber, Opts) when is_binary(Topic), is_map(Opts) -> - case ets:lookup(?SUBOPTION, {Topic, Subscriber}) of +-spec(set_subopts(emqx_topic:topic(), emqx_types:subopts()) -> boolean()). +set_subopts(Topic, NewOpts) when is_binary(Topic), is_map(NewOpts) -> + Sub = {self(), Topic}, + case ets:lookup(?SUBOPTION, Sub) of [{_, OldOpts}] -> - ets:insert(?SUBOPTION, {{Topic, Subscriber}, maps:merge(OldOpts, Opts)}); + ets:insert(?SUBOPTION, {Sub, maps:merge(OldOpts, NewOpts)}); [] -> false end. -async_call(Broker, Req) -> - From = {self(), Tag = make_ref()}, - ok = gen_server:cast(Broker, {From, Req}), - Tag. +-spec(topics() -> [emqx_topic:topic()]). +topics() -> + emqx_router:topics(). -wait_for_replies(Tags, Timeout) -> - lists:foreach( - fun(Tag) -> - wait_for_reply(Tag, Timeout) - end, Tags). +safe_lookup_element(Tab, Key, Def) -> + try ets:lookup_element(Tab, Key, 2) catch error:badarg -> Def end. -wait_for_reply(Tag, Timeout) -> - receive - {Tag, Reply} -> Reply - after Timeout -> - exit(timeout) +%%------------------------------------------------------------------------------ +%% Stats fun +%%------------------------------------------------------------------------------ + +stats_fun() -> + safe_update_stats(?SUBSCRIBER, 'subscribers/count', 'subscribers/max'), + safe_update_stats(?SUBSCRIPTION, 'subscriptions/count', 'subscriptions/max'), + safe_update_stats(?SUBOPTION, 'suboptions/count', 'suboptions/max'). + +safe_update_stats(Tab, Stat, MaxStat) -> + case ets:info(Tab, size) of + undefined -> ok; + Size -> emqx_stats:setstat(Stat, MaxStat, Size) end. -%% Pick a broker -pick(SubPid) when is_pid(SubPid) -> - gproc_pool:pick_worker(broker, SubPid). +%%------------------------------------------------------------------------------ +%% Pick and call +%%------------------------------------------------------------------------------ --spec(topics() -> [emqx_topic:topic()]). -topics() -> emqx_router:topics(). +call(Broker, Req) -> + gen_server:call(Broker, Req, ?TIMEOUT). + +%% Pick a broker +pick(Topic) -> + gproc_pool:pick_worker(broker, Topic). %%------------------------------------------------------------------------------ %% gen_server callbacks @@ -326,61 +321,32 @@ topics() -> emqx_router:topics(). init([Pool, Id]) -> true = gproc_pool:connect_worker(Pool, {Pool, Id}), - {ok, #state{pool = Pool, id = Id, submap = #{}, submon = emqx_pmon:new()}}. + {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(Req, _From, State) -> emqx_logger:error("[Broker] unexpected call: ~p", [Req]), {reply, ignored, State}. -handle_cast({From, #subscribe{topic = Topic, subpid = SubPid, subid = SubId, subopts = SubOpts}}, State) -> - Subscriber = {SubPid, SubId}, - case ets:member(?SUBOPTION, {Topic, Subscriber}) of - false -> - resubscribe(From, {Subscriber, SubOpts, Topic}, State); - true -> - case ets:lookup_element(?SUBOPTION, {Topic, Subscriber}, 2) =:= SubOpts of - true -> - gen_server:reply(From, ok), - {noreply, State}; - false -> - resubscribe(From, {Subscriber, SubOpts, Topic}, State) - end - end; - -handle_cast({From, #unsubscribe{topic = Topic, subpid = SubPid, subid = SubId}}, State) -> - Subscriber = {SubPid, SubId}, - case ets:lookup(?SUBOPTION, {Topic, Subscriber}) of - [{_, SubOpts}] -> - Group = maps:get(share, SubOpts, undefined), - true = do_unsubscribe(Group, Topic, Subscriber), - emqx_shared_sub:unsubscribe(Group, Topic, SubPid), - case ets:member(?SUBSCRIBER, Topic) of - false -> emqx_router:del_route(From, Topic, dest(Group)); - true -> gen_server:reply(From, ok) - end; - [] -> gen_server:reply(From, ok) - end, - {noreply, State}; - handle_cast(Msg, State) -> emqx_logger:error("[Broker] unexpected cast: ~p", [Msg]), {noreply, State}. -handle_info({'DOWN', _MRef, process, SubPid, Reason}, State = #state{submap = SubMap}) -> - case maps:find(SubPid, SubMap) of - {ok, SubIds} -> - lists:foreach(fun(SubId) -> subscriber_down({SubPid, SubId}) end, SubIds), - {noreply, demonitor_subscriber(SubPid, State)}; - error -> - emqx_logger:error("unexpected 'DOWN': ~p, reason: ~p", [SubPid, Reason]), - {noreply, State} - end; - handle_info(Info, State) -> emqx_logger:error("[Broker] unexpected info: ~p", [Info]), {noreply, State}. -terminate(_Reason, #state{pool = Pool, id = Id}) -> +terminate(_Reason, #{pool := Pool, id := Id}) -> gproc_pool:disconnect_worker(Pool, {Pool, Id}). code_change(_OldVsn, State, _Extra) -> @@ -390,69 +356,9 @@ code_change(_OldVsn, State, _Extra) -> %% Internal functions %%------------------------------------------------------------------------------ -resubscribe(From, {Subscriber, SubOpts, Topic}, State) -> - {SubPid, _} = Subscriber, - Group = maps:get(share, SubOpts, undefined), - true = do_subscribe(Group, Topic, Subscriber, SubOpts), - emqx_shared_sub:subscribe(Group, Topic, SubPid), - emqx_router:add_route(From, Topic, dest(Group)), - {noreply, monitor_subscriber(Subscriber, State)}. - -insert_subscriber(Group, Topic, Subscriber) -> - Subscribers = subscribers(Topic), - case lists:member(Subscriber, Subscribers) of - false -> - ets:insert(?SUBSCRIBER, {Topic, shared(Group, Subscriber)}); - _ -> - ok - end. - -do_subscribe(Group, Topic, Subscriber, SubOpts) -> - ets:insert(?SUBSCRIPTION, {Subscriber, shared(Group, Topic)}), - insert_subscriber(Group, Topic, Subscriber), - ets:insert(?SUBOPTION, {{Topic, Subscriber}, SubOpts}). - -do_unsubscribe(Group, Topic, Subscriber) -> - ets:delete_object(?SUBSCRIPTION, {Subscriber, shared(Group, Topic)}), - ets:delete_object(?SUBSCRIBER, {Topic, shared(Group, Subscriber)}), - ets:delete(?SUBOPTION, {Topic, Subscriber}). - -subscriber_down(Subscriber) -> - Topics = lists:map(fun({_, {share, Group, Topic}}) -> - {Topic, Group}; - ({_, Topic}) -> - {Topic, undefined} - end, ets:lookup(?SUBSCRIPTION, Subscriber)), - lists:foreach(fun({Topic, undefined}) -> - true = do_unsubscribe(undefined, Topic, Subscriber), - ets:member(?SUBSCRIBER, Topic) orelse emqx_router:del_route(Topic, dest(undefined)); - ({Topic, Group}) -> - true = do_unsubscribe(Group, Topic, Subscriber), - Groups = groups(Topic), - case lists:member(Group, lists:usort(Groups)) of - true -> ok; - false -> emqx_router:del_route(Topic, dest(Group)) - end - end, Topics). - -monitor_subscriber({SubPid, SubId}, State = #state{submap = SubMap, submon = SubMon}) -> - UpFun = fun(SubIds) -> lists:usort([SubId|SubIds]) end, - State#state{submap = maps:update_with(SubPid, UpFun, [SubId], SubMap), - submon = emqx_pmon:monitor(SubPid, SubMon)}. - -demonitor_subscriber(SubPid, State = #state{submap = SubMap, submon = SubMon}) -> - State#state{submap = maps:remove(SubPid, SubMap), - submon = emqx_pmon:demonitor(SubPid, SubMon)}. - dest(undefined) -> node(); dest(Group) -> {Group, node()}. shared(undefined, Name) -> Name; shared(Group, Name) -> {share, Group, Name}. -groups(Topic) -> - lists:foldl(fun({_, {share, Group, _}}, Acc) -> - [Group | Acc]; - ({_, _}, Acc) -> - Acc - end, [], ets:lookup(?SUBSCRIBER, Topic)). diff --git a/src/emqx_broker_helper.erl b/src/emqx_broker_helper.erl index e597a233e..35fe06f0d 100644 --- a/src/emqx_broker_helper.erl +++ b/src/emqx_broker_helper.erl @@ -16,63 +16,82 @@ -behaviour(gen_server). --export([start_link/0]). --export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). +-compile({no_auto_import, [monitor/2]}). -%% internal export --export([stats_fun/0]). +-export([start_link/0]). +-export([monitor/2]). +-export([create_seq/1, reclaim_seq/1]). + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, + code_change/3]). -define(HELPER, ?MODULE). +-define(SUBMON, emqx_submon). +-define(SUBSEQ, emqx_subseq). --record(state, {}). +-record(state, {pmon :: emqx_pmon:pmon()}). --spec(start_link() -> {ok, pid()} | ignore | {error, any()}). +-spec(start_link() -> emqx_types:startlink_ret()). start_link() -> gen_server:start_link({local, ?HELPER}, ?MODULE, [], []). +-spec(monitor(pid(), emqx_types:subid()) -> ok). +monitor(SubPid, SubId) when is_pid(SubPid) -> + case ets:lookup(?SUBMON, SubPid) of + [] -> + gen_server:cast(?HELPER, {monitor, SubPid, SubId}); + [{_, SubId}] -> + ok; + _Other -> + error(subid_conflict) + end. + +-spec(create_seq(emqx_topic:topic()) -> emqx_sequence:seqid()). +create_seq(Topic) -> + emqx_sequence:nextval(?SUBSEQ, Topic). + +-spec(reclaim_seq(emqx_topic:topic()) -> emqx_sequence:seqid()). +reclaim_seq(Topic) -> + emqx_sequence:reclaim(?SUBSEQ, Topic). + %%------------------------------------------------------------------------------ %% gen_server callbacks %%------------------------------------------------------------------------------ init([]) -> - %% Use M:F/A for callback, not anonymous function because - %% fun M:F/A is small, also no badfun risk during hot beam reload - emqx_stats:update_interval(broker_stats, fun ?MODULE:stats_fun/0), - {ok, #state{}, hibernate}. + %% SubSeq: Topic -> SeqId + _ = emqx_sequence:create(?SUBSEQ), + %% SubMon: SubPid -> SubId + _ = 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}. handle_call(Req, _From, State) -> emqx_logger:error("[BrokerHelper] unexpected call: ~p", [Req]), {reply, ignored, State}. +handle_cast({monitor, SubPid, SubId}, State = #state{pmon = PMon}) -> + true = ets:insert(?SUBMON, {SubPid, SubId}), + {noreply, State#state{pmon = emqx_pmon:monitor(SubPid, PMon)}}; + handle_cast(Msg, State) -> emqx_logger:error("[BrokerHelper] unexpected cast: ~p", [Msg]), {noreply, State}. +handle_info({'DOWN', _MRef, process, SubPid, _Reason}, State = #state{pmon = PMon}) -> + true = ets:delete(?SUBMON, SubPid), + ok = emqx_pool:async_submit(fun emqx_broker:subscriber_down/1, [SubPid]), + {noreply, State#state{pmon = emqx_pmon:erase(SubPid, PMon)}}; + handle_info(Info, State) -> emqx_logger:error("[BrokerHelper] unexpected info: ~p", [Info]), {noreply, State}. terminate(_Reason, #state{}) -> + _ = emqx_sequence:delete(?SUBSEQ), emqx_stats:cancel_update(broker_stats). code_change(_OldVsn, State, _Extra) -> {ok, State}. -%%------------------------------------------------------------------------------ -%% Internal functions -%%------------------------------------------------------------------------------ - -stats_fun() -> - safe_update_stats(emqx_subscriber, - 'subscribers/count', 'subscribers/max'), - safe_update_stats(emqx_subscription, - 'subscriptions/count', 'subscriptions/max'), - safe_update_stats(emqx_suboptions, - 'suboptions/count', 'suboptions/max'). - -safe_update_stats(Tab, Stat, MaxStat) -> - case ets:info(Tab, size) of - undefined -> ok; - Size -> emqx_stats:setstat(Stat, MaxStat, Size) - end. - diff --git a/src/emqx_broker_sup.erl b/src/emqx_broker_sup.erl index 51f6e72aa..a511e4154 100644 --- a/src/emqx_broker_sup.erl +++ b/src/emqx_broker_sup.erl @@ -20,8 +20,6 @@ -export([init/1]). --define(TAB_OPTS, [public, {read_concurrency, true}, {write_concurrency, true}]). - start_link() -> supervisor:start_link({local, ?MODULE}, ?MODULE, []). @@ -30,39 +28,26 @@ start_link() -> %%------------------------------------------------------------------------------ init([]) -> - %% Create the pubsub tables - ok = lists:foreach(fun create_tab/1, [subscription, subscriber, suboption]), - + %% Broker pool + PoolSize = emqx_vm:schedulers() * 2, + BrokerPool = emqx_pool_sup:spec(emqx_broker_pool, + [broker, hash, PoolSize, + {emqx_broker, start_link, []}]), %% Shared subscription - SharedSub = {shared_sub, {emqx_shared_sub, start_link, []}, - permanent, 5000, worker, [emqx_shared_sub]}, + SharedSub = #{id => shared_sub, + start => {emqx_shared_sub, start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => [emqx_shared_sub]}, %% Broker helper - Helper = {broker_helper, {emqx_broker_helper, start_link, []}, - permanent, 5000, worker, [emqx_broker_helper]}, + Helper = #{id => helper, + start => {emqx_broker_helper, start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => [emqx_broker_helper]}, - %% Broker pool - BrokerPool = emqx_pool_sup:spec(emqx_broker_pool, - [broker, hash, emqx_vm:schedulers() * 2, - {emqx_broker, start_link, []}]), - - {ok, {{one_for_all, 0, 1}, [SharedSub, Helper, BrokerPool]}}. - -%%------------------------------------------------------------------------------ -%% Create tables -%%------------------------------------------------------------------------------ - -create_tab(suboption) -> - %% Suboption: {Topic, Sub} -> [{qos, 1}] - emqx_tables:new(emqx_suboption, [set | ?TAB_OPTS]); - -create_tab(subscriber) -> - %% Subscriber: Topic -> Sub1, Sub2, Sub3, ..., SubN - %% duplicate_bag: o(1) insert - emqx_tables:new(emqx_subscriber, [duplicate_bag | ?TAB_OPTS]); - -create_tab(subscription) -> - %% Subscription: Sub -> Topic1, Topic2, Topic3, ..., TopicN - %% bag: o(n) insert - emqx_tables:new(emqx_subscription, [bag | ?TAB_OPTS]). + {ok, {{one_for_all, 0, 1}, [BrokerPool, SharedSub, Helper]}}. diff --git a/src/emqx_local_bridge.erl b/src/emqx_local_bridge.erl index 7c4e7cea1..df2dda686 100644 --- a/src/emqx_local_bridge.erl +++ b/src/emqx_local_bridge.erl @@ -61,7 +61,7 @@ init([Pool, Id, Node, Topic, Options]) -> true -> true = erlang:monitor_node(Node, true), Group = iolist_to_binary(["$bridge:", atom_to_list(Node), ":", Topic]), - emqx_broker:subscribe(Topic, self(), #{share => Group, qos => ?QOS_0}), + emqx_broker:subscribe(Topic, #{share => Group, qos => ?QOS_0}), State = parse_opts(Options, #state{node = Node, subtopic = Topic}), MQueue = emqx_mqueue:init(#{max_len => State#state.max_queue_len, store_qos0 => true}), diff --git a/src/emqx_sequence.erl b/src/emqx_sequence.erl index 62a882294..c4fc0fd08 100644 --- a/src/emqx_sequence.erl +++ b/src/emqx_sequence.erl @@ -14,45 +14,47 @@ -module(emqx_sequence). --export([create/0, create/1]). --export([generate/1, generate/2]). --export([reclaim/1, reclaim/2]). +-export([create/1, nextval/2, currval/2, reclaim/2, delete/1]). -type(key() :: term()). +-type(name() :: atom()). -type(seqid() :: non_neg_integer()). --define(DEFAULT_TAB, ?MODULE). +-export_type([seqid/0]). %% @doc Create a sequence. --spec(create() -> ok). -create() -> - create(?DEFAULT_TAB). - --spec(create(atom()) -> ok). -create(Tab) -> - _ = ets:new(Tab, [set, public, named_table, {write_concurrency, true}]), +-spec(create(name()) -> ok). +create(Name) -> + _ = ets:new(Name, [set, public, named_table, {write_concurrency, true}]), ok. -%% @doc Generate a sequence id. --spec(generate(key()) -> seqid()). -generate(Key) -> - generate(?DEFAULT_TAB, Key). +%% @doc Next value of the sequence. +-spec(nextval(name(), key()) -> seqid()). +nextval(Name, Key) -> + ets:update_counter(Name, Key, {2, 1}, {Key, 0}). --spec(generate(atom(), key()) -> seqid()). -generate(Tab, Key) -> - ets:update_counter(Tab, Key, {2, 1}, {Key, 0}). +%% @doc Current value of the sequence. +-spec(currval(name(), key()) -> seqid()). +currval(Name, Key) -> + try ets:lookup_element(Name, Key, 2) + catch + error:badarg -> 0 + end. %% @doc Reclaim a sequence id. --spec(reclaim(key()) -> seqid()). -reclaim(Key) -> - reclaim(?DEFAULT_TAB, Key). - --spec(reclaim(atom(), key()) -> seqid()). -reclaim(Tab, Key) -> - try ets:update_counter(Tab, Key, {2, -1, 0, 0}) of - 0 -> ets:delete_object(Tab, {Key, 0}), 0; +-spec(reclaim(name(), key()) -> seqid()). +reclaim(Name, Key) -> + try ets:update_counter(Name, Key, {2, -1, 0, 0}) of + 0 -> ets:delete_object(Name, {Key, 0}), 0; I -> I catch error:badarg -> 0 end. +%% @doc Delete the sequence. +delete(Name) -> + case ets:info(Name, name) of + Name -> ets:delete(Name); + undefined -> false + end. + diff --git a/src/emqx_session.erl b/src/emqx_session.erl index 12718b9fc..262a9a7a8 100644 --- a/src/emqx_session.erl +++ b/src/emqx_session.erl @@ -465,7 +465,7 @@ handle_cast({subscribe, FromPid, {PacketId, _Properties, TopicFilters}}, emqx_hooks:run('session.subscribed', [#{client_id => ClientId}, Topic, SubOpts#{first => false}]), SubMap; {ok, _SubOpts} -> - emqx_broker:set_subopts(Topic, {self(), ClientId}, SubOpts), + emqx_broker:set_subopts(Topic, SubOpts), %% Why??? emqx_hooks:run('session.subscribed', [#{client_id => ClientId}, Topic, SubOpts#{first => false}]), maps:put(Topic, SubOpts, SubMap); diff --git a/test/emqx_sequence_SUITE.erl b/test/emqx_sequence_SUITE.erl index 999a95723..f37b60d76 100644 --- a/test/emqx_sequence_SUITE.erl +++ b/test/emqx_sequence_SUITE.erl @@ -19,19 +19,19 @@ -include_lib("eunit/include/eunit.hrl"). --import(emqx_sequence, [generate/1, reclaim/1]). +-import(emqx_sequence, [nextval/2, reclaim/2]). all() -> [sequence_generate]. sequence_generate(_) -> - ok = emqx_sequence:create(), - ?assertEqual(1, generate(key)), - ?assertEqual(2, generate(key)), - ?assertEqual(3, generate(key)), - ?assertEqual(2, reclaim(key)), - ?assertEqual(1, reclaim(key)), - ?assertEqual(0, reclaim(key)), - ?assertEqual(false, ets:member(emqx_sequence, key)), - ?assertEqual(1, generate(key)). + ok = emqx_sequence:create(seqtab), + ?assertEqual(1, nextval(seqtab, key)), + ?assertEqual(2, nextval(seqtab, key)), + ?assertEqual(3, nextval(seqtab, key)), + ?assertEqual(2, reclaim(seqtab, key)), + ?assertEqual(1, reclaim(seqtab, key)), + ?assertEqual(0, reclaim(seqtab, key)), + ?assertEqual(false, ets:member(seqtab, key)), + ?assertEqual(1, nextval(seqtab, key)).