Шаблон блокирующего сервера-таймера.

gen_bserver.erl

%%% @file gen_bserver.erl
%%%
%%%     Шаблон блокирующего сервера-таймера.
%%%     Используется для долгих функций, которые
%%%     не могут быть выполнены параллельно,
%%%     но должны запускаться регулярно.
%%%     События происходят по истечению
%%%         GEN_BSERVER_TIMEOUT.
%%%     timeout сервера и функцию обработки occupate
%%%     можно менять в процессе выполнения.
%%%
%%%     Кроме того, указанные модуль можно использовать в качестве behaviour.
%%%     Для этого нужно определить функцию occupate. И в этом случае,
%%%     функцию можно переопределять во время выполнения.
%%%     Функция occupate по-умолчанию (для этого модуля) определена
%%%     в его конце.
%%%
%%%     ВАЖНО:
%%%         Для функции occupate важен только побочный эффект.
%%%         Эта функция не может возвращать свой результат в явном виде.
%%%

-module(gen_bserver).
-behaviour(gen_server).

%% --------------------------------------------------------------------
%% Макросы
%% --------------------------------------------------------------------
-define(GEN_BSERVER_TIMEOUT, 1000).      %% раз в секунду

-ifndef(D).
    -define(D(S),       io:format(S)).
    -define(D(F, P),    io:format(F, P)).
-endif.

%% --------------------------------------------------------------------
%% Описание состояния сервера
%% --------------------------------------------------------------------
-record(state,{
    % своборен \ занят
        occupancy   =   free                    :: free | busy,
    % текущий рабочий процесс
        worker      =   null                    :: null | pid(),
    % время отклика
        timeout     =   ?GEN_BSERVER_TIMEOUT    :: integer(),
    % функция обработки
        occupate    =   fun()-> ok end          :: function()
}).

%% --------------------------------------------------------------------
%% Внешний экспорт
%% --------------------------------------------------------------------
-export([
    start_link/0,   %% стартует сервер по-умолчанию
    start_link/1,   %% стартует сервер с заданным timeout 
    start_link/2,   %% стартует сервер с заданным timeout и occupate
    get_state/0,    %% возвращает состояние сервера
    get_timeout/0,  %% возвращает timeout сервера
    get_occupate/0, %% возвращает функцию обработки сервера
    set_timeout/1,  %% задает timeout сервера
    set_occupate/1, %% задает функцию обработки сервера
    free/0,         %% меняет работающий процесс
    stop/0          %% останавливает сервер
]).

%% стартует сервер по-умолчанию
-spec start_link()                                          -> {ok, pid()}.

%% стартует сервер по-умолчанию с  timeout = Timeout
-spec start_link(Timeout::integer())                        -> {ok, pid()}.

%% стартует сервер с  timeout = Timeout, occupate = Occupate
-spec start_link
    (Timeout::integer(), Occupate::function())  -> {ok, pid()};
    (Timeout::integer(), Omodule::atom())       -> {ok, pid()}.

%% возвращает состояние сервера
-spec get_state()           -> record(state).

%% возвращает timeout сервера
-spec get_timeout()             -> integer().

%% возвращает функцию обработки сервера
-spec get_occupate()            -> function().

%% задает timeout сервера
-spec set_timeout(integer())    -> none().

%% задает функцию обработки сервера
-spec set_occupate(function())  -> none().

%% меняет работающий процесс
-spec free()            -> none().

%% останавливает сервер
-spec stop()            -> none().

%% --------------------------------------------------------------------
%% Обратные вызовы spawn_link/3
%% --------------------------------------------------------------------
-export([
    occupate/0  %% долгая функция обработки
]).

%% --------------------------------------------------------------------
%% Обратные вызовы gen_server
%% --------------------------------------------------------------------
-export([
    init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2,
    code_change/3
]).

%% --------------------------------------------------------------------
%% Обратные вызовы для behaviour
%% --------------------------------------------------------------------
-export([behaviour_info/1]).

%% ====================================================================
%% Внешние функции
%% ====================================================================

%%
%% @doc Определяет набор функций, которые должны быть реализованы,
%%      если модуль будет использован в качестве behaviour.
%% @spec behaviour() -> undefined|list(tuple(atom(), integer())).
%%
behaviour_info(callbacks) ->
    [{occupate,0}];
behaviour_info(_Other) ->
    undefined.

%%
%% @doc Стартует сервер с заданным timeout и определенной функцией Occupate
%% @spec start_link(Timeout::integer(), Occupate::function()) -> {ok, pid()}.
%%
start_link(Timeout, Occupate) when erlang:is_function(Occupate, 0)->
    gen_server:start_link(
        {local, ?MODULE},
        ?MODULE,
        [Timeout, Occupate],
        []
    );
%%
%% @doc Стартует сервер с заданным timeout и определенной функцией
%%      Omodule:occupate/0. Функция должна быть экспортирована
%%      из модуля Omodule.
%% @spec start_link(Timeout::integer(), Omodule::atom()) -> {ok, pid()}.
%%
start_link(Timeout, Omodule) when erlang:is_atom(Omodule)->
    start_link(Timeout, fun Omodule:occupate/0).

%%
%% @doc Стартует сервер с заданным timeout и ?MODULE:occupate.
%%      Функцию occupate можно переопределить статически (переписать ?MODULE),
%%      или динамически в процессе выполнения.
%% @spec start_link(Timeout::integer()) -> {ok, pid()}.
%%
start_link(Timeout)->
    start_link(Timeout, fun occupate/0).

%%
%% @doc Стартует сервер c ?GEN_BSERVER_TIMEOUT и ?MODULE:occupate.
%%      Оба этих парамента можно переопределить в процессе выполнения.
%% @spec start_link() -> {ok, pid()}.
%%
start_link()->
    start_link(?GEN_BSERVER_TIMEOUT).

%%
%% @doc Возвращает состояние сервера (использует синхронный запрос)
%% @spec get_state() -> record(state).
%%
get_state()->
    gen_server:call(?MODULE, {'get', state}).

%%
%% @doc Возвращает timeout сервера (использует синхронный запрос)
%% @spec get_timeout() -> integer().
%%
get_timeout()->
    gen_server:call(?MODULE, {'get', timeout}).

%%
%% @doc Возвращает функцию сервера (использует синхронный запрос)
%% @spec get_occupate() -> function().
%%
get_occupate()->
    gen_server:call(?MODULE, {'get', occupate}).

%%
%% @doc Устанавливает timeout сервера (использует асинхронный запрос)
%% @spec set_timeout(integer()) -> none()..
%%
set_timeout(Timeout)->
    gen_server:cast(?MODULE, {set, {timeout, Timeout}}).

%%
%% @doc Устанавливает функцию сервера (иcпользует асинхронный запрос)
%% @spec set_occupate(integer()) -> none().
%%
set_occupate(Occupate)->
    gen_server:cast(?MODULE, {set, {occupate, Occupate}}).

%%
%% @doc Перезапускает рабочий процесс сервера
%% @spec free() -> none()
%%
free()->
    gen_server:cast(?MODULE, free).

stop()->
    gen_server:cast(?MODULE, stop).
    
%% ====================================================================
%% Функции gen_server
%% ====================================================================

%% --------------------------------------------------------------------
%% Функция: init/1
%% Описание: инициализирует сервер
%% Возвращает:
%%          {ok, State}               |
%%          {ok, State, Timeout}
%% --------------------------------------------------------------------
%%
%% @doc Инициализирует сервер начальным состянием и задает таймаут.
%%      Кроме того включает перехват сигналов выхода
%%          process_flag(trap_exit, true)
%% @spec init([]) -> {ok, #state{}, integer()}..
%%

init([Timeout, Occupate]) ->
    process_flag(trap_exit, true),
    {ok, #state{timeout=Timeout,occupate=Occupate}, Timeout}.

%% --------------------------------------------------------------------
%% Функция:     handle_call/3
%% Описание:    Обрабатывает call сообшения
%% Возвращает:  {reply, Reply, State}          |
%%              {reply, Reply, State, Timeout} |
%%              {noreply, State}               |
%%              {noreply, State, Timeout}      |
%%              {stop, Reason, Reply, State}   | (terminate/2 is called)
%%              {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------

%%
%% @doc Возвращает timeout сервера по синхронному запросу
%% @spec handle_call({'get', timeout}, atom()|pid(), record(state)) ->
%%          {reply, record(state), record(state)}.
%%
handle_call({'get', timeout}, _from,  State) ->
    {reply, State#state.timeout, State};

%%
%% @doc Возвращает функцию сервера по синхронному запросу
%% @spec handle_call({'get', timeout}, atom()|pid(), record(state)) ->
%%          {reply, record(state), record(state)}.
%%
handle_call({'get', occupate}, _from, State) ->
    {reply, State#state.occupate, State};

%%
%% @doc Возвращает состояние сервера по синхронному запросу
%% @spec handle_call({'get', state}, atom()|pid(), record(state)) ->
%%          {reply, record(state), record(state)}.
%%
handle_call({'get', state}, _from, State) ->
    {reply, State, State}.

%% --------------------------------------------------------------------
%% Функция:     handle_coccupancy вать ast/2
%% Описание:    Обрабатывает cast сообшения
%% Возвращает:  {noreply, State}          |
%%              {noreply, State, Timeout} |
%%              {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------


%%
%% @doc Выставляет параметр состояние сервера timeout = Timeout
%%      по aсинхронному запросу.
%% @spec handle_cast({set, {timeout, integer()}}, record(state)) ->
%%          {noreply, record(state)}.
%%
handle_cast({set, {timeout, Timeout}}, State) ->
    {noreply, State#state{timeout=Timeout}};

%%
%% @doc Выставляет параметр состояние сервера occupate = Occupate
%%      по aсинхронному запросу.
%% @spec handle_cast({set, {occupate, function()}}, record(state)) ->
%%          {noreply, record(state)}.
%%
handle_cast({set, {occupate, Occupate}}, State) ->
    {noreply, State#state{occupate=Occupate}};

%%
%% @doc Выставляет параметр состояния сервера occupancy=free, worker=null
%%      Это по сути означает перезапуск рабочего процесса сервера,
%%      после истечения GEN_BSERVER_TIMEOUT
%% @spec handle_cast(free, record(state)) -> {noreply, record(state)}.
%%
handle_cast(free, #state{occupancy=busy, worker=Worker}=State) ->
    ?D("handle_cast(free, #state{occupancy=busy})~n"),
    exit(Worker, normal),
    {noreply,
        State#state{
            occupancy=free,
            worker=null
        }
    };

handle_cast(stop, State) ->
    {stop, normal, State}.

%% --------------------------------------------------------------------
%% Функция:     handle_info/2
%% Описание:    Обрабатывает  все не call cast сообшения
%% Возвращает:  {noreply, State}          |
%%              {noreply, State, Timeout} |
%%              {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------

%%
%% @doc Начинает выполнение целевой операции --- cбора статистики.
%%      Сборка может оказаться долше, чем GEN_BSERVER_TIMEOUT,
%%      Сборку выполняем отдельным пототком, чтобы не блокировать этот сервер.
%% @spec handle_info(timeout, record(state)) ->
%%          {noreply, record(state), integer()}.
%%
handle_info(timeout, #state{occupancy=free, worker=null}=State) ->
    ?D("handle_info(timeout, #state{occupancy=free,worker=null}~n"),
    {noreply,
        State#state{
            occupancy = busy,
            %% Выполняем долгую опрерацию отдельным потоком,
            %%  чтобы не блокировать сам сервер.
            worker = spawn_link(State#state.occupate)
        },
        State#state.timeout
    };

%%
%% @doc Обрабатывает получение сообщения, по истечению GEN_BSERVER_TIMEOUT
%% @spec handle_info(timeout, record(state)) ->
%%          {noreply, record(state), integer()}.
%%
handle_info(timeout, #state{occupancy=busy}=State) ->
    ?D("handle_info(timeout,#state{occupancy=busy})~n"),
    {noreply,
        State,
        State#state.timeout
    };

%%
%% @doc Обрабатывает окончание обработки статистики
%% @spec handle_info({'EXIT',atom()|pid(),normal}, record(state)) ->
%%          {noreply, record(state), integer()}.
%%
handle_info(
        {'EXIT',Worker,normal},
        #state{occupancy=busy,worker=Worker}=State
    )->
    ?D("worker ~p closed ~n", [Worker]),
    {noreply,
        State#state{
            occupancy=free,
            worker=null
        },
        State#state.timeout
    }.

%% --------------------------------------------------------------------
%% Функция:     terminate/2
%% Описание:    Зааершает работу сервера
%% Возвращает:  none (ignored by gen_server)
%% --------------------------------------------------------------------

%%
%% @doc Обрабатывает остановку сервера. В общем случае ничего не делает.
%% @spec terminate(any(), record(state)) -> none()
%%
terminate(_reason, _state) ->
    ok.

%% --------------------------------------------------------------------
%% Функция:     code_change/3
%% Описание:    Производит смену состояния сервера при перезагрузке кода
%% Возвращает:  {ok, NewState}
%% --------------------------------------------------------------------

%%
%% @doc Производит смену состояния сервера при перезагрузке кода.
%%      Производит смему рабочего процесса, и выясняет состояние сервера
%%      после этого.
%% @spec code_change(any(), record(state), any()) -> {ok, record(state)}
%%
code_change(_old_vsn, _state, _extra) ->
    ?MODULE:free(),
    New_state = ?MODULE:state(),
    {ok, New_state}.

%% --------------------------------------------------------------------
%% Внутренние функции
%% --------------------------------------------------------------------

%%
%% @doc Некоторая долгая операция,
%%      которая должна выполняться регулярно,
%%      но не может выполняться параллельно.
%% @spec occupate()-> none()
%%
occupate()->
    ?D("~p:occupate with pid = ~p ~n", [?MODULE, self()]),
    test_occupate(1000000000).

%%
%% @doc Иммитирует работу некоторой долгой операции.
%% @spec test_occupate(integer())-> none()
%%
test_occupate(0) ->
    ok;
test_occupate(X) ->
    test_occupate(X - 1).

stat_srv.erl

%%% @file stat_srv.erl
%%%
%%%     Таймер-сервер для сбора статистики.
%%%     События происходят по истечению,
%%%     ?STAT_SRV_TIMEOUT
%%%     Является наследником gen_bserver
%%% 
%%%
%%%

-module(stat_srv).
-extends(gen_bserver).
-behaviour(gen_bserver).

%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------

%% --------------------------------------------------------------------
%% Макросы
%% --------------------------------------------------------------------
-define(STAT_SRV_TIMEOUT, 1000).      %%                раз в секунду
%-define(STAT_SRV_TIMEOUT, 86400000). %% 24*60*60*1000; раз в сутки

-ifndef(D).
    -define(D(S),       io:format(S)).
    -define(D(F, P),    io:format(F, P)).
-endif.


%% --------------------------------------------------------------------
%% Внешний экспорт
%% --------------------------------------------------------------------

-export([
    start_link/0,   %% стартует сервер по-умолчанию
    occupate/0
                    %% остальные функции унаследованы от gen_bserver
]).

%% стартует сервер сбора статистики
-spec start_link()  -> {ok, pid()}.
-spec occupate()    -> {ok, pid()}.

%% ====================================================================
%% Внешние функции
%% ====================================================================

%%
%% @doc Стартует сервер сбора статистики
%% @spec start_link() -> {ok, pid()}.
%%
start_link()->
    % gen_bserver:start_link(?STAT_SRV_TIMEOUT, fun dao_stat:fetch_smi/0).
    gen_bserver:start_link(?STAT_SRV_TIMEOUT, ?MODULE).

%%
%% @doc Функция сбора статистики
%% @spec fetch_stat()-> none()
%%
occupate() ->
    ?D("~p:occupate with pid = ~p ~n", [?MODULE, self()]),
    % dao_stat:fetch_smi().
    test_occupate(1000000000).

%%
%% @doc Имитирует работу некоторой долгой операции.
%% @spec test_occupate(integer())-> none()
%%
test_occupate(0) ->
    ok;
test_occupate(X) ->
    test_occupate(X - 1).