gproc pool

apps/emqttd/include/emqttd.hrl

@@ -57,7 +57,7 @@
 -record(mqtt_subscriber, {
     topic    :: binary(),
     qos = 0  :: 0 | 1 | 2,
-    subpid   :: pid()
+    pid      :: pid()
 }).
 
 -type mqtt_subscriber() :: #mqtt_subscriber{}.

apps/emqttd/src/emqttd_broker.erl

@@ -214,13 +214,13 @@ create(Topic) ->
     emqttd_pubsub:create(Topic).
 
 retain(Topic, Payload) when is_binary(Payload) ->
-    emqttd_router:route(broker, #mqtt_message{retain = true,
-                                              topic = Topic,
-                                              payload = Payload}).
+    emqttd_pubsub:publish(#mqtt_message{retain = true,
+                                        topic = Topic,
+                                        payload = Payload}).
 
 publish(Topic, Payload) when is_binary(Payload) ->
-    emqttd_router:route(broker, #mqtt_message{topic = Topic,
-                                              payload = Payload}).
+    emqttd_pubsub:publish(#mqtt_message{topic = Topic,
+                                        payload = Payload}).
 
 uptime(#state{started_at = Ts}) ->
     Secs = timer:now_diff(os:timestamp(), Ts) div 1000000,

apps/emqttd/src/emqttd_cm.erl

@@ -32,8 +32,6 @@
 
 -define(SERVER, ?MODULE).
 
--define(CLIENT_TABLE, mqtt_client).
-
 %% API Exports 
 -export([start_link/0]).
 
@@ -52,6 +50,8 @@
 
 -record(state, {tab}).
 
+-define(CLIENT_TABLE, mqtt_client).
+
 %%%=============================================================================
 %%% API
 %%%=============================================================================

apps/emqttd/src/emqttd_pooler.erl

@@ -0,0 +1,69 @@
+%%%-----------------------------------------------------------------------------
+%%% @Copyright (C) 2012-2015, Feng Lee <feng@emqtt.io>
+%%%
+%%% Permission is hereby granted, free of charge, to any person obtaining a copy
+%%% of this software and associated documentation files (the "Software"), to deal
+%%% in the Software without restriction, including without limitation the rights
+%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+%%% copies of the Software, and to permit persons to whom the Software is
+%%% furnished to do so, subject to the following conditions:
+%%%
+%%% The above copyright notice and this permission notice shall be included in all
+%%% copies or substantial portions of the Software.
+%%%
+%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+%%% SOFTWARE.
+%%%-----------------------------------------------------------------------------
+%%% @doc
+%%% emqttd pooler supervisor.
+%%%
+%%% @end
+%%%-----------------------------------------------------------------------------
+-module(emqttd_pooler).
+
+-author('feng@emqtt.io').
+
+-behaviour(gen_server).
+
+-define(SERVER, ?MODULE).
+
+%% API Exports 
+-export([start_link/1]).
+
+%% gen_server Function Exports
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-record(state, {id}).
+
+%%%=============================================================================
+%%% API
+%%%=============================================================================
+-spec start_link(I :: pos_integer()) -> {ok, pid()} | ignore | {error, any()}.
+start_link(I) ->
+    gen_server:start_link(?MODULE, [I], []).
+
+init([I]) ->
+    gproc_pool:connect_worker(pooler, {pooler, I}),
+    {ok, #state{id = I}}.
+
+handle_call(_Req, _From, State) ->
+    {reply, ok, State}.
+
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+terminate(_Reason, #state{id = I}) ->
+    gproc_pool:disconnect_worker(pooler, {pooler, I}), ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+

apps/emqttd/src/emqttd_pooler_sup.erl

@@ -0,0 +1,57 @@
+%%%-----------------------------------------------------------------------------
+%%% @Copyright (C) 2012-2015, Feng Lee <feng@emqtt.io>
+%%%
+%%% Permission is hereby granted, free of charge, to any person obtaining a copy
+%%% of this software and associated documentation files (the "Software"), to deal
+%%% in the Software without restriction, including without limitation the rights
+%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+%%% copies of the Software, and to permit persons to whom the Software is
+%%% furnished to do so, subject to the following conditions:
+%%%
+%%% The above copyright notice and this permission notice shall be included in all
+%%% copies or substantial portions of the Software.
+%%%
+%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+%%% SOFTWARE.
+%%%-----------------------------------------------------------------------------
+%%% @doc
+%%% emqttd pooler supervisor.
+%%%
+%%% @end
+%%%-----------------------------------------------------------------------------
+-module(emqttd_pooler_sup).
+
+-author('feng@emqtt.io').
+
+-include("emqttd.hrl").
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/0, start_link/1]).
+
+%% Supervisor callbacks
+-export([init/1]).
+
+start_link() ->
+    start_link(erlang:system_info(schedulers)).
+
+start_link(PoolSize) ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, [PoolSize]).
+
+init([PoolSize]) ->
+    gproc_pool:new(pooler, random, [{size, PoolSize}]),
+    Children = lists:map(
+                 fun(I) ->
+                         gproc_pool:add_worker(pooler, {pooler, I}, I),
+                         {{emqttd_pooler, I},
+                            {emqttd_pooler, start_link, [I]},
+                                permanent, 5000, worker, [emqttd_pooler]}
+                 end, lists:seq(1, PoolSize)),
+    {ok, {{one_for_all, 10, 100}, Children}}.
+

apps/emqttd/src/emqttd_pubsub.erl

@@ -32,20 +32,16 @@
 
 -include("emqttd.hrl").
 
--behaviour(gen_server).
-
--define(SERVER, ?MODULE).
-
--define(SUBACK_ERR, 128).
-
 %% Mnesia Callbacks
 -export([mnesia/1]).
 
 -boot_mnesia({mnesia, [boot]}).
 -copy_mnesia({mnesia, [copy]}).
 
+-behaviour(gen_server).
+
 %% API Exports 
--export([start_link/0]).
+-export([start_link/0, name/1]).
 
 -export([create/1,
          subscribe/1, subscribe/2,
@@ -58,6 +54,8 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
+-define(SUBACK_ERR, 128).
+
 -record(state, {submap :: map()}).
 
 %%%=============================================================================
@@ -76,7 +74,7 @@ mnesia(boot) ->
                 {ram_copies, [node()]},
                 {record_name, mqtt_subscriber},
                 {attributes, record_info(fields, mqtt_subscriber)},
-                {index, [subpid]},
+                {index, [pid]},
                 {local_content, true}]);
 
 mnesia(copy) ->
@@ -85,7 +83,9 @@ mnesia(copy) ->
 
 %%%=============================================================================
 %%% API
+%%%
 %%%=============================================================================
+%%%
 
 %%------------------------------------------------------------------------------
 %% @doc
@@ -93,9 +93,12 @@ mnesia(copy) ->
 %%
 %% @end
 %%------------------------------------------------------------------------------
--spec start_link() -> {ok, pid()} | ignore | {error, any()}.
-start_link() ->
-    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+-spec start_link(Opts) -> {ok, pid()} | ignore | {error, any()}.
+start_link(Opts) ->
+    gen_server:start_link(?MODULE, [], []).
+
+name(I) ->
+	list_to_atom("emqttd_pubsub_" ++ integer_to_list(I)).
 
 %%------------------------------------------------------------------------------
 %% @doc
@@ -103,47 +106,41 @@ start_link() ->
 %%
 %% @end
 %%------------------------------------------------------------------------------
--spec create(binary()) -> ok.
+-spec create(binary()) -> {atomic, ok} | {aborted, Reason :: any()}.
 create(Topic) when is_binary(Topic) ->
-    Record = #mqtt_topic{topic = Topic, node = node()},
-    {atomic, ok} = mnesia:transaction(fun insert_topic/1, [Record]), ok.
+    TopicRecord = #mqtt_topic{topic = Topic, node = node()},
+    Result = mnesia:transaction(fun create_topic/1, [TopicRecord]),
+    setstats(topics), Result.
 
 %%------------------------------------------------------------------------------
 %% @doc
-%% Subscribe topics
+%% Subscribe topic or topics.
 %%
 %% @end
 %%------------------------------------------------------------------------------
--spec subscribe({Topic, Qos} | list({Topic, Qos})) -> {ok, Qos | list(Qos)} when 
+-spec subscribe({Topic, Qos} | list({Topic, Qos})) -> {ok, Qos | list(Qos)} when
     Topic   :: binary(),
     Qos     :: mqtt_qos().
 subscribe(Topics = [{_Topic, _Qos} | _]) ->
     {ok, lists:map(fun({Topic, Qos}) ->
             case subscribe(Topic, Qos) of
-                {ok, GrantedQos} -> 
+                {ok, GrantedQos} ->
                     GrantedQos;
-                Error -> 
-                    lager:error("Failed to subscribe '~s': ~p", [Topic, Error]), 
+                {error, Error} -> 
+                    lager:error("subscribe '~s' error: ~p", [Topic, Error]), 
                     ?SUBACK_ERR
             end
         end, Topics)}.
 
--spec subscribe(Topic :: binary(), Qos :: mqtt_qos()) -> {ok, Qos :: mqtt_qos()}.
+-spec subscribe(Topic :: binary(), Qos :: mqtt_qos()) -> {ok, Qos :: mqtt_qos()} | {error, any()}.
 subscribe(Topic, Qos) when is_binary(Topic) andalso ?IS_QOS(Qos) ->
-    TopicRecord = #mqtt_topic{topic = Topic, node = node()},
-    Subscriber = #mqtt_subscriber{topic = Topic, qos = Qos, subpid = self()},
-    F = fun() ->
-            case insert_topic(TopicRecord) of
-               ok -> insert_subscriber(Subscriber);
-               Error -> Error
-            end
-        end,
-    case mnesia:transaction(F) of
+    case  create(Topic) of
         {atomic, ok} -> 
-            {ok, Qos};
-        {aborted, Reason} -> 
-            {error, Reason}
-    end.
+            Subscriber = #mqtt_subscriber{topic = Topic, qos = Qos, pid = self()},
+            ets:insert_new(?SUBSCRIBER_TAB, Subscriber),
+            {ok, Qos}; % Grant all qos
+        {aborted, Reason} ->
+            {error, Reason}.
 
 %%------------------------------------------------------------------------------
 %% @doc
@@ -153,15 +150,17 @@ subscribe(Topic, Qos) when is_binary(Topic) andalso ?IS_QOS(Qos) ->
 %%------------------------------------------------------------------------------
 -spec unsubscribe(binary() | list(binary())) -> ok.
 unsubscribe(Topic) when is_binary(Topic) ->
-    SubPid = self(),
+    Pattern = #mqtt_subscriber{topic = Topic, _ = '_', pid = self()},
+    ets:match_delete(?SUBSCRIBER_TAB, Pattern),
+
     TopicRecord = #mqtt_topic{topic = Topic, node = node()},
     F = fun() ->
         %%TODO record name...
-        Pattern = #mqtt_subscriber{topic = Topic, _ = '_', subpid = SubPid},
         [mnesia:delete_object(Sub) || Sub <- mnesia:match_object(Pattern)],
         try_remove_topic(TopicRecord)
     end,
-    {atomic, _} = mneisa:transaction(F), ok;
+    %{atomic, _} = mneisa:transaction(F), 
+    ok;
 
 unsubscribe(Topics = [Topic|_]) when is_binary(Topic) ->
     lists:foreach(fun(T) -> unsubscribe(T) end, Topics).
@@ -193,7 +192,7 @@ publish(Topic, Msg) when is_binary(Topic) ->
 %%------------------------------------------------------------------------------
 -spec dispatch(Topic :: binary(), Msg :: mqtt_message()) -> non_neg_integer().
 dispatch(Topic, Msg = #mqtt_message{qos = Qos}) when is_binary(Topic) ->
-    case mnesia:dirty_read(subscriber, Topic) of
+    case ets:lookup:(?SUBSCRIBER_TAB, Topic) of
         [] -> 
             %%TODO: not right when clusted...
             setstats(dropped);
@@ -307,15 +306,19 @@ code_change(_OldVsn, State, _Extra) ->
 %%%=============================================================================
 %%% Internal functions
 %%%=============================================================================
-insert_topic(Record = #mqtt_topic{topic = Topic}) ->
+
+-spec create_topic(#mqtt_topic{}) -> {atomic, ok} | {aborted, any()}.
+create_topic(TopicRecord = #mqtt_topic{topic = Topic}) ->
     case mnesia:wread({topic, Topic}) of
         [] ->
             ok = emqttd_trie:insert(Topic),
-            mnesia:write(topic, Record, write);
+            mnesia:write(topic, TopicRecord, write);
         Records ->
-            case lists:member(Record, Records) of
-                true -> ok;
-                false -> mnesia:write(topic, Record, write)
+            case lists:member(TopicRecord, Records) of
+                true -> 
+                    ok;
+                false -> 
+                    mnesia:write(topic, TopicRecord, write)
             end
     end.
 

apps/emqttd/src/emqttd_pubsub_sup.erl

@@ -0,0 +1,59 @@
+%%%-----------------------------------------------------------------------------
+%%% @Copyright (C) 2012-2015, Feng Lee <feng@emqtt.io>
+%%%
+%%% Permission is hereby granted, free of charge, to any person obtaining a copy
+%%% of this software and associated documentation files (the "Software"), to deal
+%%% in the Software without restriction, including without limitation the rights
+%%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+%%% copies of the Software, and to permit persons to whom the Software is
+%%% furnished to do so, subject to the following conditions:
+%%%
+%%% The above copyright notice and this permission notice shall be included in all
+%%% copies or substantial portions of the Software.
+%%%
+%%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+%%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+%%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+%%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+%%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+%%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+%%% SOFTWARE.
+%%%-----------------------------------------------------------------------------
+%%% @doc
+%%% emqttd pubsub supervisor.
+%%%
+%%% @end
+%%%-----------------------------------------------------------------------------
+-module(emqttd_pubsub_sup).
+
+-author('feng@emqtt.io').
+
+-include("emqttd.hrl").
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/1]).
+
+%% Supervisor callbacks
+-export([init/1]).
+
+start_link(Opts) ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, [Opts]).
+
+init([Opts]) ->
+    Schedulers = erlang:system_info(schedulers),
+    PoolSize = proplists:get_value(pool, Opts, Schedulers),
+    gproc_pool:new(pubsub, hash, [{size, PoolSize}]),
+    Children = lists:map(
+                 fun(I) ->
+                         gproc_pool:add_worker(pubsub, emqttd_pubsub:name(I), I),
+                         child(I, Opts)
+                 end, lists:seq(1, PoolSize)),
+    {ok, {{one_for_all, 10, 100}, Children}}.
+
+child(I, Opts) ->
+    {{emqttd_pubsub, I},
+        {emqttd_pubsub, start_link, [I, Opts]},
+            permanent, 5000, worker, [emqttd_pubsub]}.
+

apps/emqttd/src/emqttd_router.erl

@@ -1,129 +0,0 @@
-%%-----------------------------------------------------------------------------
-%% Copyright (c) 2012-2015, Feng Lee <feng@emqtt.io>
-%% 
-%% Permission is hereby granted, free of charge, to any person obtaining a copy
-%% of this software and associated documentation files (the "Software"), to deal
-%% in the Software without restriction, including without limitation the rights
-%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-%% copies of the Software, and to permit persons to whom the Software is
-%% furnished to do so, subject to the following conditions:
-%% 
-%% The above copyright notice and this permission notice shall be included in all
-%% copies or substantial portions of the Software.
-%% 
-%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-%% SOFTWARE.
-%%------------------------------------------------------------------------------
-
-%%TODO: route chain... statistics
--module(emqttd_router).
-
--include_lib("emqtt/include/emqtt.hrl").
-
--include("emqttd.hrl").
-
--behaviour(gen_server).
-
--define(SERVER, ?MODULE).
-
-%% API Function Exports
--export([start_link/0]).
-
-%%Router Chain--> --->In Out<---
--export([route/2]).
-
-%% Mnesia Callbacks
--export([mnesia/1]).
-
--boot_mnesia({mnesia, [boot]}).
--copy_mnesia({mnesia, [copy]}).
-
-%% gen_server Function Exports
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
-         terminate/2, code_change/3]).
-
--record(state, {}).
-
-%%%=============================================================================
-%%% Mnesia callbacks
-%%%=============================================================================
-mnesia(boot) ->
-    %% topic table
-    ok = emqttd_mnesia:create_table(topic, [
-                {type, bag},
-                {ram_copies, [node()]},
-                {record_name, mqtt_topic},
-                {attributes, record_info(fields, mqtt_topic)}]).
-
-mnesia(copy) ->
-    ok = emqttd_mnesia:copy_table(topic),
-
-%%%=============================================================================
-%%% API
-%%%=============================================================================
-
-%%------------------------------------------------------------------------------
-%% @doc
-%% Start emqttd router.
-%%
-%% @end
-%%------------------------------------------------------------------------------
--spec start_link() -> {ok, pid()} | ignore | {error, term()}.
-start_link() ->
-    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
-
-%%------------------------------------------------------------------------------
-%% @doc
-%% Route mqtt message. From is clienid or module.
-%%
-%% @end
-%%------------------------------------------------------------------------------
--spec route(From :: binary() | atom(), Msg :: mqtt_message()) -> ok.
-route(From, Msg) ->
-    lager:info("Route ~s from ~s", [emqtt_message:format(Msg), From]),
-    emqttd_msg_store:retain(Msg),
-	emqttd_pubsub:publish(emqtt_message:unset_flag(Msg)).
-
-%%%=============================================================================
-%%% gen_server callbacks
-%%%=============================================================================
-init([]) ->
-    TabId = ets:new(?CLIENT_TABLE, [bag,
-                                    named_table,
-                                    public,
-                                    {read_concurrency, true}]),
-    %% local subscriber table, not shared with other nodes 
-    ok = emqttd_mnesia:create_table(subscriber, [
-                {type, bag},
-                {ram_copies, [node()]},
-                {record_name, mqtt_subscriber},
-                {attributes, record_info(fields, mqtt_subscriber)},
-                {index, [subpid]},
-                {local_content, true}]);
-
-    {ok, #state{tab = TabId}}.
-
-handle_call(_Request, _From, State) ->
-    {reply, ok, State}.
-
-handle_cast(_Msg, State) ->
-    {noreply, State}.
-
-handle_info(_Info, State) ->
-    {noreply, State}.
-
-terminate(_Reason, _State) ->
-    ok.
-
-code_change(_OldVsn, State, _Extra) ->
-    {ok, State}.
-	
-%%%=============================================================================
-%%% Internal functions
-%%%=============================================================================
-

rebar.config

@@ -18,11 +18,12 @@
 {validate_app_modules, true}.
 
 {sub_dirs, [
-    "rel", 
+    "rel",
     "apps/emqtt",
     "apps/emqttd"]}.
 
 {deps, [
+	{gproc, ".*", {git, "git://github.com/uwiger/gproc.git", {branch, "master"}}},
 	{lager, ".*", {git, "git://github.com/basho/lager.git", {branch, "master"}}},
 	{esockd, "2.*", {git, "git://github.com/emqtt/esockd.git", {branch, "master"}}},
 	{mochiweb, ".*", {git, "git://github.com/slimpp/mochiweb.git", {branch, "master"}}}

rel/files/app.config

@@ -72,6 +72,8 @@
         {max_message_num, 100000},
         {max_playload_size, 16#ffff}
     ]},
+    %% PubSub
+    {pubsub, []},
     %% Broker
     {broker, [
         {sys_interval, 60}