Implement a hash-based subscription sharding

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

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], []).
+
+%%------------------------------------------------------------------------------
+%% 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
+%% 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).
-
--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);
+    subscribe(Topic, SubId, #{});
+subscribe(Topic, SubOpts) when is_binary(Topic), is_map(SubOpts) ->
+    subscribe(Topic, 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()).
-
--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).
+    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(), 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())
-      -> [{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)).
-
-subscription(Topic, Subscriber) ->
-    {Topic, ets:lookup_element(?SUBOPTION, {Topic, Subscriber}, 2)}.
-
--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(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.
+%%------------------------------------------------------------------------------
+%% 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(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(subscriptions(pid() | emqx_types:subid())
+      -> [{emqx_topic:topic(), emqx_types:subopts()}]).
+subscriptions(SubPid) ->
+    [{Topic, safe_lookup_element(?SUBOPTION, {SubPid, Topic}, #{})}
+      || Topic <- safe_lookup_element(?SUBSCRIPTION, SubPid, [])].
+
+-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(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(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).
-
-wait_for_reply(Tag, Timeout) ->
-    receive
-        {Tag, Reply} -> Reply
-    after Timeout ->
-        exit(timeout)
+safe_lookup_element(Tab, Key, Def) ->
+    try ets:lookup_element(Tab, Key, 2) catch error:badarg -> Def end.
+
+%%------------------------------------------------------------------------------
+%% 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)).

src/emqx_broker_helper.erl

@@ -16,63 +16,82 @@
 
 -behaviour(gen_server).
 
+-compile({no_auto_import, [monitor/2]}).
+
 -export([start_link/0]).
--export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+-export([monitor/2]).
+-export([create_seq/1, reclaim_seq/1]).
 
-%% internal export
--export([stats_fun/0]).
+-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.
-

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]),
-
-    %% Shared subscription
-    SharedSub = {shared_sub, {emqx_shared_sub, start_link, []},
-                 permanent, 5000, worker, [emqx_shared_sub]},
-
-    %% Broker helper
-    Helper = {broker_helper, {emqx_broker_helper, start_link, []},
-              permanent, 5000, worker, [emqx_broker_helper]},
-
     %% Broker pool
+    PoolSize = emqx_vm:schedulers() * 2,
     BrokerPool = emqx_pool_sup:spec(emqx_broker_pool,
-                                    [broker, hash, emqx_vm:schedulers() * 2,
+                                    [broker, hash, PoolSize,
                                      {emqx_broker, start_link, []}]),
+    %% Shared subscription
+    SharedSub = #{id       => shared_sub,
+                  start    => {emqx_shared_sub, start_link, []},
+                  restart  => permanent,
+                  shutdown => 2000,
+                  type     => worker,
+                  modules  => [emqx_shared_sub]},
 
-    {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]).
+    %% Broker helper
+    Helper = #{id       => helper,
+               start    => {emqx_broker_helper, start_link, []},
+               restart  => permanent,
+               shutdown => 2000,
+               type     => worker,
+               modules  => [emqx_broker_helper]},
+
+    {ok, {{one_for_all, 0, 1}, [BrokerPool, SharedSub, Helper]}}.
 

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}),

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.
+

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);

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)).