task worker pool

apps/iot/src/iot_api.erl

@@ -63,7 +63,6 @@ get_event_period_settings() ->
             hackney:close(ClientRef),
             {ok, RespBody};
         {ok, HttpCode, _, ClientRef} ->
-            {ok, RespBody} = hackney:body(ClientRef),
             hackney:close(ClientRef),
             {error, {http_error, HttpCode}};
         {error, Reason} ->

apps/iot/src/iot_event_period_settings.erl

@@ -8,12 +8,13 @@
 %%%-------------------------------------------------------------------
 -module(iot_event_period_settings).
 -author("anlicheng").
+-include_lib("stdlib/include/qlc.hrl").
 
 -behaviour(gen_server).
 
 %% API
 -export([start_link/0]).
--export([get_throttle/1]).
+-export([get_throttle/1, debug_info/0]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
@@ -41,6 +42,10 @@
 %%% API
 %%%===================================================================
 
+debug_info() ->
+    Q = qlc:q([E || E <- ets:table(?TAB_NAME)]),
+    qlc:e(Q).
+
 %% 获取设置的时间周期
 -spec get_throttle(GroupKey :: any()) -> integer().
 get_throttle(GroupKey) ->

apps/iot/src/iot_host.erl

@@ -497,7 +497,8 @@ handle_event(cast, {handle, {ai_event, Event0}}, ?STATE_ACTIVATED, State = #stat
                     end,
                     iot_device:change_status(DevicePid, ?DEVICE_ONLINE),
 
-                    iot_ai_router:route_uuid(DeviceUUID, EventType, Params)
+                    %iot_ai_router:route_uuid(DeviceUUID, EventType, Params)
+                    iot_device:ai_event(DevicePid, EventType, Params)
             end;
         Event when is_map(Event) ->
             lager:warning("[iot_host] host: ~p, event: ~p, not supported", [UUID, Event]);

apps/iot/src/iot_task.erl

@@ -21,7 +21,8 @@
 -define(SERVER, ?MODULE).
 
 -record(state, {
-    counter = 0
+    counter = 0,
+    pool_pid :: pid()
 }).
 
 %%%===================================================================
@@ -52,7 +53,9 @@ start_link() ->
     {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term()} | ignore).
 init([]) ->
-    {ok, #state{}}.
+    %% 启动工作的线程池
+    {ok, PoolPid} = poolboy:start_link([{size, 10}, {max_overflow, 50}, {worker_module, iot_task_worker}], []),
+    {ok, #state{pool_pid = PoolPid}}.
 
 %% @private
 %% @doc Handling call messages
@@ -73,8 +76,9 @@ handle_call(_Request, _From, State = #state{}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_cast({submit, Task}, State = #state{}) ->
-    spawn_monitor(fun() -> Task() end),
+handle_cast({submit, Task}, State = #state{pool_pid = PoolPid}) ->
+    %% 限制异步任务的最大并发数, 避免进程被耗尽
+    poolboy:transaction(PoolPid, fun(WorkerPid) -> iot_task_worker:execute(WorkerPid, Task) end),
     {noreply, State};
 handle_cast(debug_info, State = #state{counter = Counter}) ->
     lager:debug("[iot_task] execute task_num: ~p", [Counter]),
@@ -86,13 +90,6 @@ handle_cast(debug_info, State = #state{counter = Counter}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-%% Task进程挂掉
-handle_info({'DOWN', _MRef, process, _Pid, normal}, State=#state{counter = Counter}) ->
-    {noreply, State#state{counter = Counter + 1}};
-handle_info({'DOWN', _MRef, process, _Pid, Reason}, State) ->
-    lager:notice("[iot_task] task process down with reason: ~p", [Reason]),
-    {noreply, State};
-
 handle_info(_Info, State = #state{}) ->
     {noreply, State}.
 

apps/iot/src/iot_task_worker.erl

@@ -0,0 +1,110 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2024, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 12. 7月 2024 10:16
+%%%-------------------------------------------------------------------
+-module(iot_task_worker).
+-author("anlicheng").
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0]).
+-export([execute/2]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+-record(state, {
+
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec execute(Pid :: pid(), Task :: fun()) -> ok.
+execute(Pid, Task) when is_pid(Pid), is_function(Task, 0) ->
+    gen_server:call(Pid, {execute, Task}).
+
+%% @doc Spawns the server and registers the local name (unique)
+-spec(start_link() ->
+    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+%% @private
+%% @doc Initializes the server
+-spec(init(Args :: term()) ->
+    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term()} | ignore).
+init([]) ->
+    {ok, #state{}}.
+
+%% @private
+%% @doc Handling call messages
+-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
+    State :: #state{}) ->
+    {reply, Reply :: term(), NewState :: #state{}} |
+    {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_call({execute, Task}, _From, State = #state{}) ->
+    case catch Task() of
+        {error, Reason} ->
+            lager:warning("[iot_task_worker] execute task get error: ~p", Reason);
+        _ ->
+            ok
+    end,
+    {reply, ok, State}.
+
+%% @private
+%% @doc Handling cast messages
+-spec(handle_cast(Request :: term(), State :: #state{}) ->
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_cast(_Request, State = #state{}) ->
+    {noreply, State}.
+
+%% @private
+%% @doc Handling all non call/cast messages
+-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_info(_Info, State = #state{}) ->
+    {noreply, State}.
+
+%% @private
+%% @doc This function is called by a gen_server when it is about to
+%% terminate. It should be the opposite of Module:init/1 and do any
+%% necessary cleaning up. When it returns, the gen_server terminates
+%% with Reason. The return value is ignored.
+-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
+    State :: #state{}) -> term()).
+terminate(_Reason, _State = #state{}) ->
+    ok.
+
+%% @private
+%% @doc Convert process state when code is changed
+-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
+    Extra :: term()) ->
+    {ok, NewState :: #state{}} | {error, Reason :: term()}).
+code_change(_OldVsn, State = #state{}, _Extra) ->
+    {ok, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================