Compare commits
合併到: SisMaker:master
SisMaker:amiramix-master
SisMaker:develop
SisMaker:develop-0.1.8
SisMaker:develop-0.1.9
SisMaker:develop-optional_stats
SisMaker:develop-pid
SisMaker:develop-sidejob
SisMaker:develop-sj-0.2.0
SisMaker:master
pull from: SisMaker:develop-sidejob
SisMaker:amiramix-master
SisMaker:develop
SisMaker:develop-0.1.8
SisMaker:develop-0.1.9
SisMaker:develop-optional_stats
SisMaker:develop-pid
SisMaker:develop-sidejob
SisMaker:develop-sj-0.2.0
SisMaker:master
1 次程式碼提交
1 次程式碼提交
| 作者 | SHA1 | 備註 | 提交日期 |
|---|---|---|---|
 Pedram Nimreezi
|
71eb9ed5ff | Add job batching, statistics, timing and linearization options | 10 年之前 |
共有 14 個檔案被更改，包括 2727 行新增 和 324 行删除
分割檢視
Diff Options
-
+42 -44README.org
-
+861 -175src/glc.erl
-
+373 -82src/glc_code.erl
-
+2 -0src/glc_lib.erl
-
+313 -7src/gr_counter.erl
-
+17 -8src/gr_manager.erl
-
+9 -7src/gr_param.erl
-
+162 -0src/gr_sidejob_supervisor.erl
-
+2 -1src/gr_sup.erl
-
+611 -0src/gr_worker.erl
-
+217 -0src/gr_worker_job.erl
-
+66 -0src/gr_worker_job_sup.erl
-
+43 -0src/gr_worker_sup.erl
-
+9 -0src/gre.erl
+ 42
- 44
README.org
查看文件
+ 861
- 175
src/glc.erl
文件差異過大導致無法顯示
查看文件
+ 373
- 82
src/glc_code.erl
查看文件
+ 2
- 0
src/glc_lib.erl
查看文件
+ 313
- 7
src/gr_counter.erl
查看文件
+ 17
- 8
src/gr_manager.erl
查看文件
+ 9
- 7
src/gr_param.erl
查看文件
+ 162
- 0
src/gr_sidejob_supervisor.erl
查看文件
| @ -0,0 +1,162 @@ | |||
| %% ------------------------------------------------------------------- | |||
| %% | |||
| %% Copyright (c) 2013 Basho Technologies, Inc. All Rights Reserved. | |||
| %% | |||
| %% This file is provided to you 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. | |||
| %% | |||
| %% ------------------------------------------------------------------- | |||
| %% @doc | |||
| %% This module implements a sidejob_worker behavior that operates as a | |||
| %% parallel, capacity-limited supervisor of dynamic, transient children. | |||
| -module(gr_sidejob_supervisor). | |||
| -behaviour(gen_server). | |||
| %% API | |||
| -export([start_child/4, spawn/2, spawn/4, which_children/1]). | |||
| %% sidejob_worker callbacks | |||
| -export([current_usage/1, rate/1]). | |||
| %% gen_server callbacks | |||
| -export([init/1, handle_call/3, handle_cast/2, handle_info/2, | |||
| terminate/2, code_change/3]). | |||
| -record(state, {name, | |||
| children=sets:new(), | |||
| spawned=0, | |||
| died=0}). | |||
| -type resource() :: atom(). | |||
| %%%=================================================================== | |||
| %%% API | |||
| %%%=================================================================== | |||
| -spec start_child(resource(), module(), atom(), term()) -> {ok, pid()} | | |||
| {error, overload} | | |||
| {error, term()}. | |||
| start_child(Name, Mod, Fun, Args) -> | |||
| case glc:call(Name, {start_child, Mod, Fun, Args}, infinity) of | |||
| overload -> | |||
| {error, overload}; | |||
| Other -> | |||
| Other | |||
| end. | |||
| -spec spawn(resource(), function() | {module(), atom(), [term()]}) -> {ok, pid()} | {error, overload}. | |||
| spawn(Name, Fun) -> | |||
| case glc:call(Name, {spawn, Fun}, infinity) of | |||
| overload -> | |||
| {error, overload}; | |||
| Other -> | |||
| Other | |||
| end. | |||
| -spec spawn(resource(), module(), atom(), [term()]) -> {ok, pid()} | | |||
| {error, overload}. | |||
| spawn(Name, Mod, Fun, Args) -> | |||
| ?MODULE:spawn(Name, {Mod, Fun, Args}). | |||
| -spec which_children(resource()) -> [pid()]. | |||
| which_children(Module) -> | |||
| {ok, Workers} = Module:get(workers), | |||
| Children = [gen_server:call(global:whereis_name(Worker), get_children) | |||
| || Worker <- tuple_to_list(Workers)], | |||
| lists:flatten(Children). | |||
| %%%=================================================================== | |||
| %%% gen_server callbacks | |||
| %%%=================================================================== | |||
| init([Name]) -> | |||
| process_flag(trap_exit, true), | |||
| {ok, #state{name=Name}}. | |||
| handle_call(get_children, _From, State=#state{children=Children}) -> | |||
| {reply, sets:to_list(Children), State}; | |||
| handle_call({start_child, Mod, Fun, Args}, _From, State) -> | |||
| Result = (catch apply(Mod, Fun, Args)), | |||
| {Reply, State2} = case Result of | |||
| {ok, Pid} when is_pid(Pid) -> | |||
| {Result, add_child(Pid, State)}; | |||
| {ok, Pid, _Info} when is_pid(Pid) -> | |||
| {Result, add_child(Pid, State)}; | |||
| ignore -> | |||
| {{ok, undefined}, State}; | |||
| {error, _} -> | |||
| {Result, State}; | |||
| Error -> | |||
| {{error, Error}, State} | |||
| end, | |||
| {reply, Reply, State2}; | |||
| handle_call({spawn, Fun}, _From, State) -> | |||
| Pid = case Fun of | |||
| _ when is_function(Fun) -> | |||
| spawn_link(Fun); | |||
| {M, F, A} -> | |||
| spawn_link(M, F, A) | |||
| end, | |||
| State2 = add_child(Pid, State), | |||
| {reply, Pid, State2}; | |||
| handle_call(_Request, _From, State) -> | |||
| {reply, ok, State}. | |||
| handle_cast(_Msg, State) -> | |||
| {noreply, State}. | |||
| handle_info({'EXIT', Pid, Reason}, State=#state{children=Children, | |||
| died=Died}) -> | |||
| case sets:is_element(Pid, Children) of | |||
| true -> | |||
| Children2 = sets:del_element(Pid, Children), | |||
| Died2 = Died + 1, | |||
| State2 = State#state{children=Children2, died=Died2}, | |||
| {noreply, State2}; | |||
| false -> | |||
| {stop, Reason, State} | |||
| end; | |||
| handle_info(_Info, State) -> | |||
| {noreply, State}. | |||
| terminate(_Reason, _State) -> | |||
| ok. | |||
| code_change(_OldVsn, State, _Extra) -> | |||
| {ok, State}. | |||
| current_usage(#state{children=Children}) -> | |||
| {message_queue_len, Pending} = process_info(self(), message_queue_len), | |||
| Current = sets:size(Children), | |||
| Pending + Current. | |||
| rate(State=#state{spawned=Spawned, died=Died}) -> | |||
| State2 = State#state{spawned=0, | |||
| died=0}, | |||
| {Spawned, Died, State2}. | |||
| %%%=================================================================== | |||
| %%% Internal functions | |||
| %%%=================================================================== | |||
| add_child(Pid, State=#state{children=Children, spawned=Spawned}) -> | |||
| Children2 = sets:add_element(Pid, Children), | |||
| Spawned2 = Spawned + 1, | |||
| State#state{children=Children2, spawned=Spawned2}. | |||
+ 2
- 1
src/gr_sup.erl
查看文件
+ 611
- 0
src/gr_worker.erl
查看文件
| @ -0,0 +1,611 @@ | |||
| %% Copyright (c) 2013, Pedram Nimreezi <deadzen@deadzen.com> | |||
| %% | |||
| %% Permission to use, copy, modify, and/or distribute this software for any | |||
| %% purpose with or without fee is hereby granted, provided that the above | |||
| %% copyright notice and this permission notice appear in all copies. | |||
| %% | |||
| %% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES | |||
| %% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF | |||
| %% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR | |||
| %% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES | |||
| %% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN | |||
| %% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF | |||
| %% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. | |||
| -module(gr_worker). | |||
| -behaviour(gen_server). | |||
| %% API | |||
| -export([start_link/8, | |||
| list/1, insert_queue/4, batch_queue/3, | |||
| delete/2, lookup/2, lookup_element/2, | |||
| info/1, info_size/1]). | |||
| %% gen_server callbacks | |||
| -export([init/1, | |||
| handle_call/3, | |||
| handle_cast/2, | |||
| handle_info/2, | |||
| terminate/2, | |||
| code_change/3]). | |||
| -include_lib("stdlib/include/qlc.hrl"). | |||
| -record(state, {module, table_id, timer_ref, worker_ets, stats_enabled, | |||
| batch_cursor, batch_limit, batch_delay, batch_results=[], | |||
| jobs_linearized, last_ts, reporter, queue_limit, full, | |||
| sequence=0, enqueue=0, dequeue=0, waiting=[]}). | |||
| -define(ATTEMPTS, 1). | |||
| %%%=================================================================== | |||
| %%% API | |||
| %%%=================================================================== | |||
| list(Server) -> | |||
| case (catch gen_server:call(Server, list, infinity)) of | |||
| {'EXIT', _Reason} -> | |||
| list(gr_manager:wait_for_pid(Server)); | |||
| Else -> Else | |||
| end. | |||
| info_size(Server) -> | |||
| case (catch gen_server:call(Server, info_size, infinity)) of | |||
| {'EXIT', _Reason} -> | |||
| info_size(gr_manager:wait_for_pid(Server)); | |||
| Else -> Else | |||
| end. | |||
| delete(Server, Term) -> | |||
| case (catch gen_server:call(Server, {delete, Term}, infinity)) of | |||
| {'EXIT', _Reason} -> | |||
| delete(gr_manager:wait_for_pid(Server), Term); | |||
| Else -> Else | |||
| end. | |||
| batch_queue(Server, Module, Size) when is_atom(Server) -> | |||
| case whereis(Server) of | |||
| undefined -> | |||
| batch_queue(gr_manager:wait_for_pid(Server), Module, Size); | |||
| Pid -> | |||
| case erlang:is_process_alive(Pid) of | |||
| true -> | |||
| batch_queue(Pid, Module, Size); | |||
| false -> | |||
| ServerPid = gr_manager:wait_for_pid(Server), | |||
| batch_queue(ServerPid, Module, Size) | |||
| end | |||
| end; | |||
| batch_queue(Server, Module, Size) when is_pid(Server) -> | |||
| gen_server:cast(Server, {batch_queue, {Server, Module, Size}}). | |||
| insert_queue(Server, Module, {K, {V, A}} = Term, Evt) when is_integer(A) -> | |||
| case (catch gen_server:call(Server, {insert, {Module, {K, {V, A, Evt}}}}, infinity)) of | |||
| {'EXIT', _Reason} -> | |||
| insert_queue(gr_manager:wait_for_pid(Server), Module, Term, Evt); | |||
| Else -> Else | |||
| end; | |||
| insert_queue(Server, Module, {K, V} = Term, Evt) -> | |||
| case (catch gen_server:call(Server, {insert, {Module, {K, {V, ?ATTEMPTS, Evt}}}}, infinity)) of | |||
| {'EXIT', _Reason} -> | |||
| insert_queue(gr_manager:wait_for_pid(Server), Module, Term, Evt); | |||
| Else -> Else | |||
| end. | |||
| lookup(Server, Term) -> | |||
| case (catch gen_server:call(Server, {lookup, Term}, infinity)) of | |||
| {'EXIT', _Reason} -> | |||
| lookup(gr_manager:wait_for_pid(Server), Term); | |||
| Else -> Else | |||
| end. | |||
| lookup_element(Server, Term) -> | |||
| case (catch gen_server:call(Server, {lookup_element, Term}, infinity)) of | |||
| {'EXIT', _Reason} -> | |||
| lookup_element(gr_manager:wait_for_pid(Server), Term); | |||
| Else -> Else | |||
| end. | |||
| info(Server) -> | |||
| case (catch gen_server:call(Server, info, infinity)) of | |||
| {'EXIT', _Reason} -> | |||
| info(gr_manager:wait_for_pid(Server)); | |||
| Else -> Else | |||
| end. | |||
| %%-------------------------------------------------------------------- | |||
| %% @doc | |||
| %% Starts the server | |||
| %% | |||
| %% @spec start_link(Name) -> {ok, Pid} | ignore | {error, Error} | |||
| %% @end | |||
| %%-------------------------------------------------------------------- | |||
| start_link(Name, Module, Counts, StatsEnabled, JobsLinearized, | |||
| QueueLimit, BatchLimit, BatchDelay) -> | |||
| gen_server:start_link({local, Name}, ?MODULE, | |||
| [Name, Module, Counts, StatsEnabled, JobsLinearized, | |||
| QueueLimit, BatchLimit, BatchDelay], []). | |||
| %%%=================================================================== | |||
| %%% gen_server callbacks | |||
| %%%=================================================================== | |||
| %%-------------------------------------------------------------------- | |||
| %% @private | |||
| %% @doc | |||
| %% Initializes the server | |||
| %% | |||
| %% @spec init(Args) -> {ok, State} | | |||
| %% {ok, State, Timeout} | | |||
| %% ignore | | |||
| %% {stop, Reason} | |||
| %% @end | |||
| %%-------------------------------------------------------------------- | |||
| init([Name, Module, Reporter, StatsEnabled, JobsLinearized, | |||
| QueueLimit, BatchLimit, BatchDelay]) -> | |||
| process_flag(trap_exit, true), | |||
| {ok, #state{module=Module, | |||
| queue_limit=QueueLimit, | |||
| batch_limit=BatchLimit, | |||
| batch_delay=BatchDelay, | |||
| worker_ets=Name, | |||
| full=false, | |||
| timer_ref=schedule_tick(), | |||
| stats_enabled=StatsEnabled, | |||
| jobs_linearized=JobsLinearized, | |||
| last_ts=erlang:now(), | |||
| reporter=Reporter}}. | |||
| %%-------------------------------------------------------------------- | |||
| %% @private | |||
| %% @doc | |||
| %% Handling call messages | |||
| %% | |||
| %% @spec handle_call(Request, From, State) -> | |||
| %% {reply, Reply, State} | | |||
| %% {reply, Reply, State, Timeout} | | |||
| %% {noreply, State} | | |||
| %% {noreply, State, Timeout} | | |||
| %% {stop, Reason, Reply, State} | | |||
| %% {stop, Reason, State} | |||
| %% @end | |||
| %%-------------------------------------------------------------------- | |||
| handle_call(Call, From, State) when is_atom(Call), | |||
| Call =:= info; Call =:= info_size; | |||
| Call =:= list -> | |||
| TableId = State#state.table_id, | |||
| Waiting = State#state.waiting, | |||
| case TableId of | |||
| undefined -> {noreply, State#state{waiting=[{Call, From}|Waiting]}}; | |||
| _ when Call =:= list -> | |||
| {reply, handle_list(TableId), State}; | |||
| _ when Call =:= info -> | |||
| {reply, handle_info(TableId), State}; | |||
| _ when Call =:= info_size -> | |||
| {reply, handle_info_size(TableId), State} | |||
| end; | |||
| handle_call({Call, Term}, From, | |||
| #state{module=Module, table_id=TableId, | |||
| waiting=Waiting, stats_enabled=StatsEnabled, | |||
| reporter=Reporter, full=Full}=State) when is_atom(Call), | |||
| Call =:= insert; Call =:= lookup; | |||
| Call =:= delete; Call =:= lookup_element -> | |||
| case TableId of | |||
| undefined -> | |||
| {noreply, State#state{waiting=[{{Call, Term}, From}|Waiting]}}; | |||
| _ when Call =:= lookup -> | |||
| {reply, handle_lookup(TableId, Term), State}; | |||
| _ when Call =:= delete -> | |||
| {reply, handle_delete(TableId, Term), State}; | |||
| _ when Call =:= insert -> | |||
| ok = maybe_batch(State), | |||
| {Term1, State1} = maybe_rekey(Module, Term, State), | |||
| {Reply, State2} = case (not Full) of | |||
| true -> {handle_insert(TableId, Term1), | |||
| update_enqueue_rate(State1, 1)}; | |||
| false when StatsEnabled -> | |||
| gr_counter:update_counter(Reporter, queue_reject, 1), | |||
| {false, State1}; | |||
| false -> {false, State1} | |||
| end, | |||
| {reply, Reply, State2}; | |||
| _ when Call =:= lookup_element -> | |||
| {reply, handle_lookup_element(TableId, Term), State} | |||
| end; | |||
| handle_call(_Request, _From, State) -> | |||
| Reply = {error, unhandled_message}, | |||
| {reply, Reply, State}. | |||
| %%-------------------------------------------------------------------- | |||
| %% @private | |||
| %% @doc | |||
| %% Handling cast messages | |||
| %% | |||
| %% @spec handle_cast(Msg, State) -> {noreply, State} | | |||
| %% {noreply, State, Timeout} | | |||
| %% {stop, Reason, State} | |||
| %% @end | |||
| %%-------------------------------------------------------------------- | |||
| handle_cast({batch_queue, {_Module, _Server, Size} = Term}=Call, | |||
| #state{table_id=TableId, waiting=Waiting, | |||
| batch_cursor=Cursor} = State) -> | |||
| State2 = case TableId of | |||
| undefined -> State#state{waiting=[Call|Waiting]}; | |||
| _ when Size =:= 0, Cursor =/= undefined -> | |||
| qlc:delete_cursor(Cursor), | |||
| State#state{batch_cursor=undefined}; | |||
| _ -> handle_batch_queue(TableId, Term, State) | |||
| end, | |||
| {noreply, State2}; | |||
| handle_cast(_Msg, State) -> | |||
| {noreply, State}. | |||
| %%-------------------------------------------------------------------- | |||
| %% @private | |||
| %% @doc | |||
| %% Handling all non call/cast messages | |||
| %% | |||
| %% @spec handle_info(Info, State) -> {noreply, State} | | |||
| %% {noreply, State, Timeout} | | |||
| %% {stop, Reason, State} | |||
| %% @end | |||
| %%-------------------------------------------------------------------- | |||
| handle_info({'ETS-TRANSFER', TableId, _Pid, _Data}, #state{waiting=Waiting}=State0) -> | |||
| _ = [ gen_server:reply(From, perform_call(TableId, Call)) | |||
| || {Call, From} <- Waiting, Call =/= batch_queue ], | |||
| State = case [ Term || {batch_queue, Term} <- Waiting ] of | |||
| [] -> State0; | |||
| [Term|_] -> % we care about, at most, 1 of these. | |||
| handle_batch_queue(TableId, Term, State0) | |||
| end, | |||
| {noreply, State#state{table_id=TableId, waiting=[]}}; | |||
| handle_info('gr_worker_tick', #state{module=Module, table_id=TableId, | |||
| worker_ets=Server, batch_limit=Size, | |||
| queue_limit=QueueLimit}=State0) -> | |||
| State = tick(State0), | |||
| schedule_tick(), | |||
| State1 = case handle_info_size(TableId) of | |||
| TableSize when TableSize > 0, TableSize < QueueLimit -> | |||
| batch_queue(Server, Module, Size), | |||
| State#state{full=false}; | |||
| 0 -> State#state{full=false}; | |||
| _ -> %% leave nothing behind | |||
| batch_queue(Server, Module, Size), | |||
| State#state{full=true} | |||
| end, | |||
| {noreply, State1}; | |||
| handle_info({success, {Ref, {_Pid, {Key, _Val, _Attempts, _Evt}}}}, | |||
| #state{table_id=TableId, batch_results=Results} = State) -> | |||
| handle_delete(TableId, Key), | |||
| {noreply, State#state{batch_results = lists:keydelete(Ref, 1, Results)}}; | |||
| handle_info({failure, {Ref, {_Pid, {Key, _Val, Attempts, _Evt}}}}, | |||
| #state{table_id=TableId, batch_results=Results} = State0) -> | |||
| State = case Attempts of | |||
| 0 -> handle_delete(TableId, Key), | |||
| State0#state{batch_results = lists:keydelete(Ref, 1, Results)}; | |||
| _ -> State0 | |||
| end, | |||
| {noreply, State}; | |||
| handle_info({'EXIT', Pid, normal}, #state{batch_cursor=Cursor} = State) when | |||
| Pid =:= element(1, element(2, Cursor)) -> | |||
| {noreply, State#state{batch_cursor = undefined}}; | |||
| handle_info(_Info, 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, State) -> void() | |||
| %% @end | |||
| %%-------------------------------------------------------------------- | |||
| terminate(_Reason, #state{timer_ref=undefined}=_State) -> | |||
| ok; | |||
| terminate(_Reason, #state{timer_ref=Ref}=_State) -> | |||
| erlang:cancel_timer(Ref), | |||
| ok. | |||
| %%-------------------------------------------------------------------- | |||
| %% @private | |||
| %% @doc | |||
| %% Convert process state when code is changed | |||
| %% | |||
| %% @spec code_change(OldVsn, State, Extra) -> {ok, NewState} | |||
| %% @end | |||
| %%-------------------------------------------------------------------- | |||
| code_change(_OldVsn, State, _Extra) -> | |||
| {ok, State}. | |||
| %%%=================================================================== | |||
| %%% Internal functions | |||
| %%%=================================================================== | |||
| perform_call(TableId, Call) -> | |||
| case Call of | |||
| list -> | |||
| handle_list(TableId); | |||
| info -> | |||
| handle_info(TableId); | |||
| info_size -> | |||
| handle_info_size(TableId); | |||
| {insert, Term} -> | |||
| handle_insert(TableId, Term); | |||
| {delete, Term} -> | |||
| handle_delete(TableId, Term); | |||
| {lookup, Term} -> | |||
| handle_lookup(TableId, Term); | |||
| {lookup_element, Term} -> | |||
| handle_lookup_element(TableId, Term) | |||
| end. | |||
| handle_list(TableId) -> | |||
| ets:tab2list(TableId). | |||
| handle_info(TableId) -> | |||
| ets:info(TableId). | |||
| handle_info_size(TableId) -> | |||
| ets:info(TableId, size). | |||
| handle_delete(TableId, Term) -> | |||
| ets:delete(TableId, Term). | |||
| handle_insert(TableId, {_Module, {Key, {Val, Attempts, Evt}}}) -> | |||
| TS = os:timestamp(), | |||
| Term = {Key, {Val, Attempts, gre:append({enqueued, TS}, Evt)}}, | |||
| ets:insert(TableId, Term). | |||
| handle_lookup(TableId, Term) -> | |||
| ets:lookup(TableId, Term). | |||
| handle_lookup_element(TableId, Term) -> | |||
| ets:lookup_element(TableId, Term, 2). | |||
| query_handle(TableId, JobsLinearized) -> | |||
| QH0 = qlc:q([{K,E} || {K, {_, A, _}=E} | |||
| <- ets:table(TableId), A > 0 ]), | |||
| case JobsLinearized of | |||
| true -> qlc:sort(QH0, [{order, ascending}]); | |||
| false -> QH0 | |||
| end. | |||
| handle_batch_queue(TableId, Term, #state{batch_cursor=undefined, | |||
| jobs_linearized=JobsLinearized}=State) -> | |||
| QH = query_handle(TableId, JobsLinearized), | |||
| BatchState = State#state{ batch_cursor=qlc:cursor(QH) }, | |||
| handle_batch_queue(TableId, Term, BatchState); | |||
| handle_batch_queue(TableId, {Server, Module, Size} = _Term, | |||
| #state{reporter=Reporter, | |||
| batch_cursor=Cursor, | |||
| batch_delay=BatchDelay, | |||
| batch_results=BatchResults, | |||
| stats_enabled=StatsEnabled, | |||
| jobs_linearized=JobsLinearized}=State) -> | |||
| Len = handle_info_size(TableId), | |||
| case get_batch_queue(Cursor, Size) of | |||
| {ok, []} when BatchResults =:= [], Len > 0 -> | |||
| QH = query_handle(TableId, JobsLinearized), | |||
| BatchState = case Cursor of | |||
| undefined -> State#state{ batch_cursor=qlc:cursor(QH) }; | |||
| _ -> (catch qlc:delete_cursor(Cursor)), | |||
| State#state{ batch_cursor=qlc:cursor(QH) } | |||
| end, | |||
| ok = batch_queue(Server, Module, Size), | |||
| BatchState; | |||
| {ok, []} -> | |||
| %ok = batch_queue(Server, Module, 0), | |||
| State; | |||
| {ok, Batch} when JobsLinearized -> | |||
| Self = self(), | |||
| Ref0 = make_ref(), | |||
| spawn(fun() -> | |||
| Results = lists:map(fun({Key, {Val, Attempts, Evt}}) -> | |||
| execute(Self, Module, Reporter, StatsEnabled, | |||
| {Key, {Val, Attempts, Evt}}) | |||
| end, Batch), | |||
| Self ! {Ref0, Results}, | |||
| Sleep = 10 * BatchDelay, | |||
| timer:sleep(Sleep), | |||
| ok = batch_queue(Server, Module, Size) | |||
| end), | |||
| handle_batch_results(Ref0, State); | |||
| {ok, Batch} -> | |||
| Self = self(), | |||
| Ref0 = make_ref(), | |||
| spawn(fun() -> | |||
| Pids = lists:map(fun({Key, {Val, Attempts, Evt}}) -> | |||
| spawn_monitor(fun() -> | |||
| exit(execute(Self, Module, Reporter, StatsEnabled, | |||
| {Key, {Val, Attempts, Evt}})) | |||
| end) | |||
| end, Batch), | |||
| Results = [receive {'DOWN', Ref, _, _, Reason} -> Reason | |||
| end || {_, Ref} <- Pids], | |||
| Self ! {Ref0, Results}, | |||
| Sleep = 10 * BatchDelay, | |||
| timer:sleep(Sleep), | |||
| ok = batch_queue(Server, Module, Size) | |||
| end), | |||
| handle_batch_results(Ref0, State) | |||
| end. | |||
| tick(State=#state{stats_enabled=false}) -> | |||
| State; | |||
| tick(#state{module=Module, table_id=_TableId, reporter=Reporter}=State) -> | |||
| {In, Out, State2} = flush_current_rate(State), | |||
| Usage = message_queue_len(State), | |||
| %Usage = current_usage(State), | |||
| gr_counter:queue_report(Reporter, Module, Usage, In, Out), | |||
| %ets:select_delete(TableId,[{{'$1',{'$2', '$3', '$4'}},[{'<', '$3', 1}],['true']}]), | |||
| State2. | |||
| maybe_time_queue(_Reporter, _Event, false) -> ok; | |||
| maybe_time_queue(Reporter, Event, true=_StatsEnabled) -> | |||
| QueueTime = timer:now_diff(gre:fetch(dequeued, Event), | |||
| gre:fetch(enqueued, Event)), | |||
| gr_counter:update_counter(Reporter, queue_time, QueueTime), | |||
| ok. | |||
| get_batch_queue(Cursor, Size) -> | |||
| case qlc:next_answers(Cursor, Size) of | |||
| [] -> | |||
| {ok, []}; | |||
| List -> | |||
| {ok, List} | |||
| end. | |||
| execute(Self, Module, Reporter, StatsEnabled, {Key, {Val, Attempts, Evt}}) -> | |||
| Obj = {Key, {Val, Attempts, Evt}}, | |||
| Now = os:timestamp(), | |||
| QueueTime = timer:now_diff(Now, gre:fetch(enqueued, Evt)), | |||
| Event = gre:merge(gre:make([{id, Key}, %{event_handled, false}, | |||
| {queuetime, QueueTime}, {event_linear, true}, | |||
| {attempt, Attempts}, {dequeued, Now}], [list]), Evt), | |||
| ok = maybe_time_queue(Reporter, Event, StatsEnabled) , | |||
| SubRef = make_ref(), | |||
| MySelf = {Self, SubRef}, | |||
| Success = fun(_) -> success(Obj, MySelf) end, | |||
| Failure = fun(_) -> failure(Obj, MySelf) end, | |||
| %%{SubRef, glc:run(Module, Val, Event, Success, Failure)} | |||
| case glc:call(Module, {spawn, {glc, run, [Module, Val, Event, Success, Failure]}}, infinity) of | |||
| {error, overload} = _Err -> | |||
| %Failure(Err), | |||
| undefined; | |||
| Pid -> | |||
| {SubRef, {Pid, Obj}} | |||
| end. | |||
| -spec handle_batch_results(reference(), #state{}) -> #state{}. | |||
| handle_batch_results(Ref, State) -> | |||
| Results = receive {Ref, Res0} -> Res0 end, | |||
| BatchResults = [ R || R <- Results, R =/= undefined], | |||
| update_dequeue_rate(State#state{batch_results=BatchResults}, | |||
| length(BatchResults)). | |||
| -spec success({binary(), fun(), non_neg_integer()}, {pid(), reference()}) -> ok. | |||
| success({Key, {Val, _Attempts, Evt}}, {Pid, Ref}) -> | |||
| Attempts = 0, | |||
| Obj = {Ref, {Pid, {Key, Val, Attempts, Evt}}}, | |||
| Pid ! {success, Obj}, | |||
| ok. | |||
| -spec failure({binary(), fun(), non_neg_integer()}, {pid(), reference()}) -> ok. | |||
| failure({Key, {Val, Attempts0, Evt}} = _Term, {Pid, Ref}) when Attempts0 > 0 -> | |||
| Attempts = Attempts0 - 1, | |||
| Obj = {Ref, {Pid, {Key, Val, Attempts, Evt}}}, | |||
| Pid ! {failure, Obj}, | |||
| ok; | |||
| failure({Key, {Val, _Attempts, Evt}}, {Pid, Ref}) -> | |||
| Attempts = -1, | |||
| Obj = {Ref, {Pid, {Key, Val, Attempts, Evt}}}, | |||
| Pid ! {failure, Obj}, | |||
| ok. | |||
| schedule_tick() -> | |||
| erlang:send_after(1000, self(), 'gr_worker_tick'). | |||
| message_queue_len(#state{}) -> | |||
| {message_queue_len, Len} = process_info(self(), message_queue_len), | |||
| Len. | |||
| update_enqueue_rate(State=#state{enqueue=Enqueue}, Len) -> | |||
| State#state{enqueue=Enqueue + Len}. | |||
| update_dequeue_rate(State=#state{dequeue=Dequeue}, Len) -> | |||
| State#state{dequeue=Dequeue + Len}. | |||
| flush_current_rate(State=#state{enqueue=Enqueue, dequeue=Dequeue}) -> | |||
| State2 = State#state{enqueue=0, dequeue=0}, | |||
| {Enqueue, Dequeue, State2}. | |||
| -spec maybe_batch(#state{}) -> ok. | |||
| maybe_batch(#state{batch_cursor=Cursor, worker_ets=Server, | |||
| module=Module, batch_limit=Size} = _State) -> | |||
| case is_tuple(Cursor) of | |||
| true -> ok; | |||
| false -> ok = batch_queue(Server, Module, Size) | |||
| end. | |||
| -spec maybe_rekey(atom(), term(), #state{}) -> {term(), #state{}}. | |||
| maybe_rekey(Module, Term, State) -> | |||
| JobsLinearized = State#state.jobs_linearized, | |||
| KeyUndefined = element(1, element(2, Term)) =:= undefined, | |||
| case (JobsLinearized or KeyUndefined) of | |||
| true -> | |||
| {ok, {TS, Seq, Key}} = | |||
| get_next_id(State#state.last_ts, | |||
| State#state.sequence), | |||
| Val = element(2, element(2, Term)), | |||
| {{Module, {list_to_binary(integer_to_list(Key)), Val}}, | |||
| State#state{last_ts=TS, sequence=Seq}}; | |||
| false -> {Term, State} %% doesn't matter anymore | |||
| end. | |||
| get_next_id(TS, Seq) -> | |||
| case get_next_seq(TS, Seq) of | |||
| backwards_clock -> | |||
| erlang:error(backwards_clock, [{TS, Seq}]); | |||
| exhausted -> | |||
| % Retry after a millisecond | |||
| timer:sleep(1), | |||
| get_next_id(TS, Seq); | |||
| {ok, Time, NewSeq} -> | |||
| {ok, {Time, NewSeq, construct_id(Time, NewSeq)}} | |||
| end. | |||
| get_next_seq({Megas, Secs, Micros} = Time, Seq) -> | |||
| Now = erlang:now(), | |||
| {NowMegas, NowSecs, NowMicros} = Now, | |||
| if | |||
| % Time is essentially equal at the millisecond | |||
| Megas =:= NowMegas, | |||
| Secs =:= NowSecs, | |||
| NowMicros div 1000 =:= Micros div 1000 -> | |||
| case (Seq + 1) rem 4096 of | |||
| 0 -> exhausted; | |||
| NewSeq -> {ok, Now, NewSeq} | |||
| end; | |||
| % Woops, clock was moved backwards by NTP | |||
| Now < Time -> | |||
| backwards_clock; | |||
| % New millisecond | |||
| true -> | |||
| {ok, Now, 0} | |||
| end. | |||
| construct_id({Megas, Secs, Micros}, Seq) -> | |||
| Millis = Micros div 1000, | |||
| Combined = (Megas * 1000000 + Secs) * 1000 + Millis, | |||
| <<Integer:54/integer>> = <<0:1, Combined:41/integer-unsigned, | |||
| Seq:12/integer-unsigned>>, | |||
| Integer. | |||
+ 217
- 0
src/gr_worker_job.erl
查看文件
| @ -0,0 +1,217 @@ | |||
| %% Slight variation of sidejob_worker | |||
| -module(gr_worker_job). | |||
| -behaviour(gen_server). | |||
| %% API | |||
| -export([start/2, stop/2, start_link/6, spawn_fun/5]). | |||
| %% gen_server callbacks | |||
| -export([init/1, handle_call/3, handle_cast/2, handle_info/2, | |||
| terminate/2, code_change/3]). | |||
| %%%%%%%%%% | |||
| -record(state, {id, module, mod, modstate, timer_ref, worker_ets, reporter, | |||
| stats_enabled, usage, limit, width, last_mq=0, enqueue=0, dequeue=0}). | |||
| %% Interface | |||
| start_link(Module, Reporter, StatsEnabled, Limit, Width, Id) -> | |||
| Name = glc:workers_name(Module, Id), | |||
| case gen_server:start_link(?MODULE, [Name, Module, Reporter, StatsEnabled, Limit, Width, Id], []) of | |||
| {ok, Pid} -> yes = global:re_register_name(Name, Pid), | |||
| {ok, Pid}; | |||
| Else -> Else | |||
| end. | |||
| spawn_fun(Module, Fun, Event, OnSuccess, OnFailure) -> | |||
| {ok, Pid} = start(glc:workers_sup(Module), undefined), | |||
| ok = gen_server:cast(Pid, {run, Module, Fun, Event, OnSuccess, OnFailure}), | |||
| {ok, Pid}. | |||
| start(Sup, Id) -> | |||
| supervisor:start_child(Sup, [Id]). | |||
| stop(Sup, Pid) -> | |||
| supervisor:terminate_child(Sup, Pid). | |||
| init([Name, Module, Reporter, StatsEnabled, Limit, Width, Id]) -> | |||
| Mod = gr_sidejob_supervisor, | |||
| case Mod:init([Module]) of | |||
| {ok, ModState} -> | |||
| process_flag(trap_exit, true), | |||
| Exports = proplists:get_value(exports, Mod:module_info()), | |||
| Usage = case lists:member({current_usage, 1}, Exports) of | |||
| true -> | |||
| custom; | |||
| false -> | |||
| default | |||
| end, | |||
| ets:insert(Name, [{usage, 0}, {full, 0}]), | |||
| {ok, #state{module=Module, id=Id, | |||
| mod=gr_sidejob_supervisor, | |||
| modstate=ModState, | |||
| usage=Usage, | |||
| limit=Limit, | |||
| width=Width, | |||
| worker_ets=Name, | |||
| reporter=Reporter, | |||
| stats_enabled=StatsEnabled, | |||
| timer_ref=schedule_tick()}}; | |||
| Else -> Else | |||
| end. | |||
| handle_cast({run, Module, Fun, Event, Success, Failure}, State) -> | |||
| case glc:run(Module, Fun, Event) of | |||
| ok -> Success(undefined); | |||
| {ok, Result} -> Success(Result); | |||
| Else -> Failure(Else) | |||
| end, | |||
| {stop, normal, State}; | |||
| handle_cast(Request, State=#state{mod=Mod, | |||
| modstate=ModState}) -> | |||
| Result = Mod:handle_cast(Request, ModState), | |||
| {Pos, ModState2} = case Result of | |||
| {noreply,NewState} -> | |||
| {2, NewState}; | |||
| {noreply,NewState,hibernate} -> | |||
| {2, NewState}; | |||
| {noreply,NewState,_Timeout} -> | |||
| {2, NewState}; | |||
| {stop,_Reason,NewState} -> | |||
| {3, NewState} | |||
| end, | |||
| State2 = State#state{modstate=ModState2}, | |||
| State3 = update_rate(update_usage(State2)), | |||
| Return = setelement(Pos, Result, State3), | |||
| Return; | |||
| handle_cast(_Message, State) -> | |||
| {noreply, State}. | |||
| handle_info('gr_worker_job_tick', State) -> | |||
| State2 = tick(State), | |||
| schedule_tick(), | |||
| {noreply, State2}; | |||
| handle_info(Info, State=#state{mod=Mod, | |||
| modstate=ModState}) -> | |||
| Result = Mod:handle_info(Info, ModState), | |||
| {Pos, ModState2} = case Result of | |||
| {noreply,NewState} -> | |||
| {2, NewState}; | |||
| {noreply,NewState,hibernate} -> | |||
| {2, NewState}; | |||
| {noreply,NewState,_Timeout} -> | |||
| {2, NewState}; | |||
| {stop,_Reason,NewState} -> | |||
| {3, NewState} | |||
| end, | |||
| State2 = State#state{modstate=ModState2}, | |||
| State3 = update_rate(update_usage(State2)), | |||
| Return = setelement(Pos, Result, State3), | |||
| Return; | |||
| handle_info(_Message, State) -> | |||
| {noreply, State}. | |||
| handle_call(Request, From, State=#state{mod=Mod, | |||
| modstate=ModState}) -> | |||
| Result = Mod:handle_call(Request, From, ModState), | |||
| {Pos, ModState2} = case Result of | |||
| {reply,_Reply,NewState} -> | |||
| {3, NewState}; | |||
| {reply,_Reply,NewState,hibernate} -> | |||
| {3, NewState}; | |||
| {reply,_Reply,NewState,_Timeout} -> | |||
| {3, NewState}; | |||
| {noreply,NewState} -> | |||
| {2, NewState}; | |||
| {noreply,NewState,hibernate} -> | |||
| {2, NewState}; | |||
| {noreply,NewState,_Timeout} -> | |||
| {2, NewState}; | |||
| {stop,_Reason,_Reply,NewState} -> | |||
| {4, NewState}; | |||
| {stop,_Reason,NewState} -> | |||
| {3, NewState} | |||
| end, | |||
| State2 = State#state{modstate=ModState2}, | |||
| State3 = update_rate(update_usage(State2)), | |||
| Return = setelement(Pos, Result, State3), | |||
| Return; | |||
| handle_call(_Request, _From, State) -> | |||
| {reply, nop, State}. | |||
| code_change(_OldVsn, State, _Extra) -> | |||
| {ok, State}. | |||
| terminate(_Reason, #state{timer_ref=undefined}=_State) -> | |||
| ok; | |||
| terminate(_Reason, #state{timer_ref=Ref}=_State) -> | |||
| erlang:cancel_timer(Ref), | |||
| ok; | |||
| terminate(_Reason, _State) -> | |||
| ok. | |||
| tick(State=#state{stats_enabled=false}) -> | |||
| {_In, _Out, State2} = flush_current_rate(State), | |||
| State2; | |||
| tick(State=#state{id=Id, module=_Module, reporter=Reporter}) -> | |||
| Usage = current_usage(State), | |||
| {In, Out, State2} = flush_current_rate(State), | |||
| gr_counter:job_report(Reporter, Id, Usage, In, Out), | |||
| State2. | |||
| current_usage(#state{usage=default}) -> | |||
| {message_queue_len, Len} = process_info(self(), message_queue_len), | |||
| Len; | |||
| current_usage(#state{usage=custom, mod=Mod, modstate=ModState}) -> | |||
| Mod:current_usage(ModState). | |||
| update_usage(State=#state{worker_ets=ETS, width=Width, limit=Limit}) -> | |||
| Usage = current_usage(State), | |||
| Full = case Usage >= (Limit div Width) of | |||
| true -> | |||
| 1; | |||
| false -> | |||
| 0 | |||
| end, | |||
| ets:insert(ETS, [{usage, Usage}, | |||
| {full, Full}]), | |||
| State. | |||
| update_rate(State=#state{usage=custom}) -> | |||
| %% Assume this is updated internally in the custom module | |||
| State; | |||
| update_rate(State=#state{usage=default, last_mq=_LastLen}) -> | |||
| {message_queue_len, Len} = process_info(self(), message_queue_len), | |||
| Enqueue = Len + 1, | |||
| %Enqueue = Len - LastLen + 1, | |||
| Dequeue = State#state.dequeue + 1, | |||
| State#state{enqueue=Enqueue, dequeue=Dequeue, last_mq=Len}. | |||
| flush_current_rate(State=#state{ | |||
| usage=default, | |||
| enqueue=Enqueue, | |||
| dequeue=Dequeue}) -> | |||
| State2 = State#state{enqueue=0, dequeue=0}, | |||
| {Enqueue, Dequeue, State2}; | |||
| flush_current_rate(State=#state{usage=custom, mod=Mod, modstate=ModState}) -> | |||
| {Enqueue, Dequeue, ModState2} = Mod:rate(ModState), | |||
| State2 = State#state{modstate=ModState2}, | |||
| {Enqueue, Dequeue, State2}. | |||
| schedule_tick() -> | |||
| erlang:send_after(1000, self(), 'gr_worker_job_tick'). | |||
+ 66
- 0
src/gr_worker_job_sup.erl
查看文件
| @ -0,0 +1,66 @@ | |||
| %% Copyright (c) 2015, Pedram Nimreezi <deadzen@deadzen.com> | |||
| %% | |||
| %% Permission to use, copy, modify, and/or distribute this software for any | |||
| %% purpose with or without fee is hereby granted, provided that the above | |||
| %% copyright notice and this permission notice appear in all copies. | |||
| %% | |||
| %% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES | |||
| %% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF | |||
| %% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR | |||
| %% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES | |||
| %% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN | |||
| %% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF | |||
| %% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. | |||
| %% @doc Worker supervisor for all goldrush job runs. | |||
| %% | |||
| -module(gr_worker_job_sup). | |||
| -behaviour(supervisor). | |||
| -type startlink_err() :: {'already_started', pid()} | 'shutdown' | term(). | |||
| -type startlink_ret() :: {'ok', pid()} | 'ignore' | {'error', startlink_err()}. | |||
| %% API | |||
| -export([start_link/6]). | |||
| %% Supervisor callbacks | |||
| -export([init/1]). | |||
| %% =================================================================== | |||
| %% API functions | |||
| %% =================================================================== | |||
| %% @hidden | |||
| -spec start_link(atom(), atom(), atom(), boolean(), pos_integer(), | |||
| pos_integer()) -> startlink_ret(). | |||
| start_link(Name, Module, Reporter, StatsEnabled, Limit, Width) -> | |||
| supervisor:start_link({local, Name}, ?MODULE, [Module, Reporter, | |||
| StatsEnabled, Limit, Width]). | |||
| %% =================================================================== | |||
| %% Supervisor callbacks | |||
| %% =================================================================== | |||
| %% @hidden | |||
| -spec init([]) -> {ok, { {simple_one_for_one, 50, 10}, [supervisor:child_spec()]} }. | |||
| init([Module, Reporter, StatsEnabled, Limit, Width]) -> | |||
| ok = init_workers(Module, Width), | |||
| %% init stats | |||
| ChildSpec = {gr_worker_job, | |||
| {gr_worker_job, start_link, [Module, Reporter, StatsEnabled, | |||
| Limit, Width]}, | |||
| temporary, brutal_kill, worker, [gr_worker_job]}, | |||
| {ok, { {simple_one_for_one, 50, 10}, [ChildSpec]}}. | |||
| -spec init_workers(atom(), pos_integer() | [atom()]) -> ok. | |||
| init_workers(Module, Width) when is_integer(Width) -> | |||
| Workers = glc:job_workers(Module, Width), | |||
| init_workers(Module, Workers); | |||
| init_workers(Module, [Worker|Workers]) -> | |||
| TableId = ets:new(Worker, [named_table, public]), %% @todo make not public | |||
| true = ets:insert(TableId, [{usage, 0}, {full, 0}]), | |||
| init_workers(Module, Workers); | |||
| init_workers(_Module, []) -> | |||
| ok. | |||
+ 43
- 0
src/gr_worker_sup.erl
查看文件
| @ -0,0 +1,43 @@ | |||
| %% Copyright (c) 2015, Pedram Nimreezi <deadzen@deadzen.com> | |||
| %% | |||
| %% Permission to use, copy, modify, and/or distribute this software for any | |||
| %% purpose with or without fee is hereby granted, provided that the above | |||
| %% copyright notice and this permission notice appear in all copies. | |||
| %% | |||
| %% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES | |||
| %% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF | |||
| %% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR | |||
| %% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES | |||
| %% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN | |||
| %% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF | |||
| %% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. | |||
| %% @doc Worker supervisor for all goldrush job runs. | |||
| %% | |||
| -module(gr_worker_sup). | |||
| -behaviour(supervisor). | |||
| -type startlink_err() :: {'already_started', pid()} | 'shutdown' | term(). | |||
| -type startlink_ret() :: {'ok', pid()} | 'ignore' | {'error', startlink_err()}. | |||
| %% API | |||
| -export([start_link/0]). | |||
| %% Supervisor callbacks | |||
| -export([init/1]). | |||
| %% =================================================================== | |||
| %% API functions | |||
| %% =================================================================== | |||
| %% @hidden | |||
| -spec start_link() -> startlink_ret(). | |||
| start_link() -> | |||
| supervisor:start_link({local, ?MODULE}, ?MODULE, []). | |||
| %% =================================================================== | |||
| %% Supervisor callbacks | |||
| %% =================================================================== | |||
| %% @hidden | |||
| -spec init([]) -> {ok, { {one_for_one, 50, 10}, [supervisor:child_spec()]} }. | |||
| init(_Args) -> | |||
| {ok, { {one_for_one, 50, 10}, []} }. | |||