Backport/william/v4.2.12

.github/workflows/elvis_lint.yaml

@@ -1,6 +1,7 @@
 name: Elvis Linter
 
-on: [pull_request]
+on: [push]
+
 
 jobs:
   build:

etc/emqx.conf

@@ -724,6 +724,7 @@ zone.external.acl_deny_action = ignore
 ## Numbers delimited by `|'. Zero or negative is to disable.
 zone.external.force_gc_policy = 16000|16MB
 
+
 ## Max message queue length and total heap size to force shutdown
 ## connection/session process.
 ## Message queue here is the Erlang process mailbox, but not the number
@@ -736,6 +737,16 @@ zone.external.force_gc_policy = 16000|16MB
 ##   - 1000|32MB  on ARCH_32 sytem
 #zone.external.force_shutdown_policy = 10000|64MB
 
+## Performance toggle for subscribe/unsubscribe wildcard topic.
+## Change this toggle only when there are many wildcard topics.
+## Value: Enum
+##  - key:   mnesia translational updates with per-key locks. recommended for single node setup.
+##  - tab:   mnesia translational updates with table lock. recommended for multi-nodes setup.
+##  - global: global lock protected updates. recommended for larger cluster.
+## NOTE: when changing from/to 'global' lock, it requires all nodes in the cluster
+## to be stopped before the change.
+broker.perf.route_lock_type = key
+
 ## Maximum MQTT packet size allowed.
 ##
 ## Value: Bytes

priv/emqx.schema

@@ -2105,6 +2105,18 @@ end}.
   {datatype, flag}
 ]}.
 
+%% @doc Performance toggle for subscribe/unsubscribe wildcard topic.
+%%      Change this toggle only when there are many wildcard topics.
+%% key:   mnesia translational updates with per-key locks. recommended for single node setup.
+%% tab:   mnesia translational updates with table lock. recommended for multi-nodes setup.
+%% global: global lock protected updates. recommended for larger cluster.
+%% NOTE: when changing from/to 'global' lock, it requires a full cluster restart.
+%%
+{mapping, "broker.perf.route_lock_type", "emqx.route_lock_type", [
+  {default, key},
+  {datatype, {enum, [key, tab, global]}}
+]}.
+
 %%--------------------------------------------------------------------
 %% System Monitor
 %%--------------------------------------------------------------------

src/emqx.appup.src

@@ -8,23 +8,49 @@
     end,
 {VSN,
   [
+   {"4.2.11", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, []},
+      {load_module, emqx_router, soft_purge, soft_purge, [emqx_trie]}
+    ]},
     {"4.2.10", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, []},
+      {load_module, emqx_router, soft_purge, soft_purge, [emqx_trie]},
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {"4.2.9", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, []},
+      {load_module, emqx_router, soft_purge, soft_purge, [emqx_trie]},
       {load_module, emqx_connection, brutal_purge, soft_purge, []},
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {<<"4.2.[34567]">>, [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, []},
+      {load_module, emqx_router, soft_purge, soft_purge, [emqx_trie]},
       {load_module, emqx_frame, brutal_purge, soft_purge, []},
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {"4.2.2", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, []},
+      {load_module, emqx_router, soft_purge, soft_purge, [emqx_trie]},
       {load_module, emqx_frame, brutal_purge, soft_purge, []},
       {load_module, emqx_metrics, brutal_purge, soft_purge, []},
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {"4.2.1", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, []},
+      {load_module, emqx_router, soft_purge, soft_purge, [emqx_trie]},
       {load_module, emqx_frame, brutal_purge, soft_purge, []},
       {load_module, emqx_metrics, brutal_purge, soft_purge, []},
       {load_module, emqx_channel, brutal_purge, soft_purge, []},
@@ -33,6 +59,10 @@
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {"4.2.0", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, []},
+      {load_module, emqx_router, soft_purge, soft_purge, [emqx_trie]},
       {load_module, emqx_frame, brutal_purge, soft_purge, []},
       {load_module, emqx_metrics, brutal_purge, soft_purge, []},
       {load_module, emqx_channel, brutal_purge, soft_purge, []},
@@ -47,23 +77,49 @@
     {<<".*">>, []}
   ],
   [
+   {"4.2.11", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, [emqx_router]},
+      {load_module, emqx_router, soft_purge, soft_purge, []}
+    ]},
     {"4.2.10", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, [emqx_router]},
+      {load_module, emqx_router, soft_purge, soft_purge, []},
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {"4.2.9", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, [emqx_router]},
+      {load_module, emqx_router, soft_purge, soft_purge, []},
       {load_module, emqx_connection, brutal_purge, soft_purge, []},
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {<<"4.2.[34567]">>, [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, [emqx_router]},
+      {load_module, emqx_router, soft_purge, soft_purge, []},
       {load_module, emqx_frame, brutal_purge, soft_purge, []},
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {"4.2.2", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, [emqx_router]},
+      {load_module, emqx_router, soft_purge, soft_purge, []},
       {load_module, emqx_frame, brutal_purge, soft_purge, []},
       {load_module, emqx_metrics, brutal_purge, soft_purge, []},
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {"4.2.1", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, [emqx_router]},
+      {load_module, emqx_router, soft_purge, soft_purge, []},
       {load_module, emqx_frame, brutal_purge, soft_purge, []},
       {load_module, emqx_metrics, brutal_purge, soft_purge, []},
       {load_module, emqx_channel, brutal_purge, soft_purge, []},
@@ -72,6 +128,10 @@
       {load_module, emqx_ws_connection, brutal_purge, soft_purge, []}
     ]},
     {"4.2.0", [
+      {load_module, emqx_router_sup, soft_purge, soft_purge, []},
+      {load_module, emqx_broker, soft_purge, soft_purge, []},
+      {load_module, emqx_trie, soft_purge, soft_purge, [emqx_router]},
+      {load_module, emqx_router, soft_purge, soft_purge, []},
       {load_module, emqx_frame, brutal_purge, soft_purge, []},
       {load_module, emqx_metrics, brutal_purge, soft_purge, []},
       {load_module, emqx_channel, brutal_purge, soft_purge, []},

src/emqx_broker.erl

@@ -419,7 +419,7 @@ safe_update_stats(Tab, Stat, MaxStat) ->
 -compile({inline, [call/2, cast/2, pick/1]}).
 
 call(Broker, Req) ->
-    gen_server:call(Broker, Req).
+    gen_server:call(Broker, Req, infinity).
 
 cast(Broker, Msg) ->
     gen_server:cast(Broker, Msg).

src/emqx_router.erl

@@ -118,7 +118,8 @@ do_add_route(Topic, Dest) when is_binary(Topic) ->
         false ->
             ok = emqx_router_helper:monitor(Dest),
             case emqx_topic:wildcard(Topic) of
-                true  -> trans(fun insert_trie_route/1, [Route]);
+                true  ->
+                    maybe_trans(fun insert_trie_route/1, [Route]);
                 false -> insert_direct_route(Route)
             end
     end.
@@ -164,7 +165,8 @@ do_delete_route(Topic) when is_binary(Topic) ->
 do_delete_route(Topic, Dest) ->
     Route = #route{topic = Topic, dest = Dest},
     case emqx_topic:wildcard(Topic) of
-        true  -> trans(fun delete_trie_route/1, [Route]);
+        true  ->
+            maybe_trans(fun delete_trie_route/1, [Route]);
         false -> delete_direct_route(Route)
     end.
 
@@ -247,10 +249,59 @@ delete_trie_route(Route = #route{topic = Topic}) ->
     end.
 
 %% @private
+-spec(maybe_trans(function(), list(any())) -> ok | {error, term()}).
+maybe_trans(Fun, Args) ->
+    case persistent_term:get(emqx_route_lock_type, key) of
+        key ->
+            trans(Fun, Args);
+        global ->
+            lock_router(),
+            try mnesia:sync_dirty(Fun, Args)
+            after
+                unlock_router()
+            end;
+        tab ->
+            trans(fun() ->
+                          emqx_trie:lock_tables(),
+                          apply(Fun, Args)
+                  end, [])
+    end.
+
 -spec(trans(function(), list(any())) -> ok | {error, term()}).
 trans(Fun, Args) ->
-    case mnesia:transaction(Fun, Args) of
-        {atomic, Ok} -> Ok;
-        {aborted, Reason} -> {error, Reason}
+    %% trigger selective receive optimization of compiler,
+    %% ideal for handling bursty traffic.
+    Ref = erlang:make_ref(),
+    Owner = self(),
+    {WPid, RefMon} = spawn_monitor(
+                       fun() ->
+                               Res = case mnesia:transaction(Fun, Args) of
+                                         {atomic, Ok} -> Ok;
+                                         {aborted, Reason} -> {error, Reason}
+                                     end,
+                               Owner ! {Ref, Res}
+                       end),
+    receive
+        {Ref, TransRes} ->
+            receive
+                {'DOWN', RefMon, process, WPid, normal} -> ok
+            end,
+            TransRes;
+        {'DOWN', RefMon, process, WPid, Info} ->
+            {error, {trans_crash, Info}}
+    end.
+
+lock_router() ->
+    %% if Retry is not 0, global:set_lock could sleep a random time up to 8s.
+    %% Considering we have a limited number of brokers, it is safe to use sleep 1 ms.
+    case global:set_lock({?MODULE, self()}, [node() | nodes()], 0) of
+        false ->
+            %% Force to sleep 1ms instead.
+            timer:sleep(1),
+            lock_router();
+        true ->
+            ok
     end.
 
+unlock_router() ->
+    global:del_lock({?MODULE, self()}).

src/emqx_router_sup.erl

@@ -34,6 +34,10 @@ init([]) ->
                type     => worker,
                modules  => [emqx_router_helper]},
 
+    ok = persistent_term:put(emqx_route_lock_type,
+                             application:get_env(emqx, route_lock_type, key)
+                            ),
+
     %% Router pool
     RouterPool = emqx_pool_sup:spec([router_pool, hash,
                                      {emqx_router, start_link, []}]),

src/emqx_trie.erl

@@ -31,7 +31,9 @@
         , delete/1
         ]).
 
--export([empty/0]).
+-export([ empty/0
+        , lock_tables/0
+        ]).
 
 -ifdef(TEST).
 -compile(export_all).
@@ -120,6 +122,11 @@ delete(Topic) when is_binary(Topic) ->
 empty() ->
     ets:info(?TRIE_TAB, size) == 0.
 
+-spec lock_tables() -> ok.
+lock_tables() ->
+    mnesia:write_lock_table(?TRIE_TAB),
+    mnesia:write_lock_table(?TRIE_NODE_TAB).
+
 %%--------------------------------------------------------------------
 %% Internal functions
 %%--------------------------------------------------------------------