Move the global GC server from 'emqx-recon' to 'emqx' project (#3190)

etc/emqx.conf

@@ -249,6 +249,18 @@ node.dist_buffer_size = 8MB
 ## vm.args: +e Number
 node.max_ets_tables = 256000
 
+## Global GC Interval.
+##
+## Value: Duration
+##
+## Examples:
+##  - 2h:  2 hours
+##  - 30m: 30 minutes
+##  - 20s: 20 seconds
+##
+## Defaut: 15 minutes
+node.global_gc_interval = 15m
+
 ## Tweak GC to run more often.
 ##
 ## Value: Number [0-65535]

priv/emqx.schema

@@ -205,8 +205,6 @@ end}.
   {default, "emqx@127.0.0.1"}
 ]}.
 
-
-
 %% @doc Specify SSL Options in the file if using SSL for erlang distribution
 {mapping, "node.ssl_dist_optfile", "vm_args.-ssl_dist_optfile", [
   {datatype, string},
@@ -287,6 +285,11 @@ end}.
  end
 }.
 
+%% @doc Global GC Interval
+{mapping, "node.global_gc_interval", "emqx.global_gc_interval", [
+  {datatype, {duration, s}}
+]}.
+
 %% @doc http://www.erlang.org/doc/man/erlang.html#system_flag-2
 {mapping, "node.fullsweep_after", "vm_args.-env ERL_FULLSWEEP_AFTER", [
   {default, 1000},

src/emqx_global_gc.erl

@@ -0,0 +1,97 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_global_gc).
+
+-behaviour(gen_server).
+
+-include("types.hrl").
+
+-export([start_link/0, stop/0]).
+
+-export([run/0]).
+
+%% gen_server callbacks
+-export([ init/1
+        , handle_call/3
+        , handle_cast/2
+        , handle_info/2
+        , terminate/2
+        , code_change/3
+        ]).
+
+%% 5 minutes
+%% -define(DEFAULT_INTERVAL, 300000).
+
+%%--------------------------------------------------------------------
+%% APIs
+%%--------------------------------------------------------------------
+
+-spec(start_link() -> startlink_ret()).
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+-spec(run() -> {ok, timer:time()}).
+run() -> gen_server:call(?MODULE, run, infinity).
+
+-spec(stop() -> ok).
+stop() -> gen_server:stop(?MODULE).
+
+%%--------------------------------------------------------------------
+%% gen_server callbacks
+%%--------------------------------------------------------------------
+
+init([]) ->
+    {ok, ensure_timer(#{timer => undefined})}.
+
+handle_call(run, _From, State) ->
+    {Time, _} = timer:tc(fun run_gc/0),
+    {reply, {ok, Time div 1000}, State, hibernate};
+
+handle_call(_Req, _From, State) ->
+    {reply, ignored, State}.
+
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+handle_info({timeout, TRef, run}, State = #{timer := TRef}) ->
+    run_gc(),
+    {noreply, ensure_timer(State), hibernate};
+
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+terminate(_Reason, _State) ->
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%--------------------------------------------------------------------
+%% Internel function
+%%--------------------------------------------------------------------
+
+ensure_timer(State) ->
+    case emqx:get_env(global_gc_interval) of
+        undefined -> State;
+        Interval  -> TRef = emqx_misc:start_timer(timer:seconds(Interval), run),
+                     State#{timer := TRef}
+    end.
+
+run_gc() ->
+    [garbage_collect(P) || P <- processes(),
+                           {status, waiting} == process_info(P, status)].
+

src/emqx_kernel_sup.erl

@@ -27,7 +27,8 @@ start_link() ->
 
 init([]) ->
     {ok, {{one_for_one, 10, 100},
-          [child_spec(emqx_pool_sup, supervisor),
+          [child_spec(emqx_global_gc, worker),
+           child_spec(emqx_pool_sup, supervisor),
            child_spec(emqx_hooks, worker),
            child_spec(emqx_stats, worker),
            child_spec(emqx_metrics, worker),
@@ -40,7 +41,8 @@ child_spec(M, worker) ->
       restart  => permanent,
       shutdown => 5000,
       type     => worker,
-      modules  => [M]};
+      modules  => [M]
+     };
 
 child_spec(M, supervisor) ->
     #{id       => M,
@@ -48,6 +50,6 @@ child_spec(M, supervisor) ->
       restart  => permanent,
       shutdown => infinity,
       type     => supervisor,
-      modules  => [M]}.
-
+      modules  => [M]
+     }.
 

test/emqx_global_gc_SUITE.erl

@@ -0,0 +1,33 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_global_gc_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+
+all() -> emqx_ct:all(?MODULE).
+
+t_run_gc(_) ->
+    ok = application:set_env(emqx, global_gc_interval, 1),
+    {ok, _} = emqx_global_gc:start_link(),
+    ok = timer:sleep(1500),
+    {ok, MilliSecs} = emqx_global_gc:run(),
+    ct:print("Global GC: ~w(ms)~n", [MilliSecs]),
+    emqx_global_gc:stop().
+