cloudi_service_queue

%-*-Mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*-
% ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et nomod:
%%%
%%%------------------------------------------------------------------------
%%% @doc
%%% ==CloudI Persistent Queue Service==
%%% Use Write Ahead Logging (WAL) to persist service requests.  This
%%% service provides a leaner alternative to persistent messaging queues
%%% (e.g., when compared to RabbitMQ, both topics/queues).
%%% The filesystem is used for keeping the queue service requests persistent
%%% and the queue file path is provided to this service's configuration
%%% arguments.  You must make sure the queue file path is unique.  If you
%%% use a process count higher than 1, make sure to have "${I}" within the
%%% file path, so the process index is used within the file path.
%%%
%%% The fault_isolation service argument determines the fault-tolerance
%%% guarantee for the request/response exchange.  When fault_isolation is
%%% set to 'destination' (the default), the sender is isolated from
%%% destination instability only.  So, this means persistence to the
%%% filesystem begins and ends within this service and the source could
%%% fail to receive the response due to its own instability.
%%%
%%% When fault_isolation is set to 'both', both the sender and the
%%% destination are isolated from instability.  Persistence of the request
%%% begins when this service receives the incoming service request, but
%%% persistence ends after the source receives a service request that contains
%%% the response.  So, 'both' does not assume the source Erlang pid
%%% remains alive during the request/response exchange.  If you need
%%% requests to survive an Erlang VM restart, this is the mode you should use.
%%% (This assumes the source is also meant to receive the response, which
%%%  doesn't need to be the case with 'both'.  Only a valid service name
%%%  needs to be specified for the destination of the response
%%%  (with a <<"service_name">> key/value entry in the
%%%   RequestInfo of the initial service request) and it doesn't need to be
%%%  the source of the request).
%%%
%%% The retry service argument controls the number of retries during the
%%% lifetime of this service's instance.  So, this means that after an
%%% Erlang VM restart, the retry counter will start from 0
%%% after the initial retry that occurs when the WAL is read upon startup.
%%%
%%% The amount of time the service request is persisted is always limited by
%%% the timeout of the service request.  Tracking the time taken by a
%%% service request depends on the Erlang VM time-keeping being dependable
%%% which depends on the OS time-keeping not varying wildly
%%% (otherwise service requests may timeout quicker due to the system time
%%%  moving into the future).  If the fault_isolation service argument is
%%% set to 'both', the original service request timeout value will be used
%%% for the service request send to the destination and the service request
%%% send containing the response.
%%%
%%% If the retry service argument is set higher than 0, any retry attempts
%%% will occur during the time period defined by the timeout of the
%%% service request (i.e., a service request is only retried if its
%%% timeout has not expired).  Any usage of the retry_delay service argument
%%% will contribute to time elapsed during the time period defined by the
%%% timeout of the service request.
%%%
%%% To make sure cloudi_service_queue gets a service request timeout quickly
%%% (i.e., without depending on the timeout elapsing locally, despite the
%%%  timeout being a small value), it is common to set the service
%%% configuration options request_timeout_immediate_max and
%%% response_timeout_immediate_max to 0.  If the retry_delay service argument
%%% is used, setting the request_name_lookup service configuration option to
%%% async is best if the service request destinations are expected to rarely
%%% be present (i.e., only appear when anticipating the receive of data).
%%% @end
%%%
%%% BSD LICENSE
%%% 
%%% Copyright (c) 2014-2016, Michael Truog <mjtruog at gmail dot com>
%%% All rights reserved.
%%% 
%%% Redistribution and use in source and binary forms, with or without
%%% modification, are permitted provided that the following conditions are met:
%%% 
%%%     * Redistributions of source code must retain the above copyright
%%%       notice, this list of conditions and the following disclaimer.
%%%     * Redistributions in binary form must reproduce the above copyright
%%%       notice, this list of conditions and the following disclaimer in
%%%       the documentation and/or other materials provided with the
%%%       distribution.
%%%     * All advertising materials mentioning features or use of this
%%%       software must display the following acknowledgment:
%%%         This product includes software developed by Michael Truog
%%%     * The name of the author may not be used to endorse or promote
%%%       products derived from this software without specific prior
%%%       written permission
%%% 
%%% THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
%%% CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
%%% INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
%%% OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
%%% DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
%%% CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
%%% SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
%%% BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
%%% SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
%%% INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
%%% WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
%%% NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
%%% OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
%%% DAMAGE.
%%%
%%% @author Michael Truog <mjtruog [at] gmail (dot) com>
%%% @copyright 2014-2016 Michael Truog
%%% @version 1.5.4 {@date} {@time}
%%%------------------------------------------------------------------------

-module(cloudi_service_queue).
-author('mjtruog [at] gmail (dot) com').

-behaviour(cloudi_service).

%% cloudi_service callbacks
-export([cloudi_service_init/4,
         cloudi_service_handle_request/11,
         cloudi_service_handle_info/3,
         cloudi_service_terminate/3]).

-include_lib("cloudi_core/include/cloudi_logger.hrl").
-include_lib("cloudi_core/include/cloudi_service.hrl").

-define(DEFAULT_FILE,                   undefined). % see below:
        % required argument, string
        % use "$I" or "${I}" for the process index within the string
        % so unique files are created when the configuration count_process > 1
-define(DEFAULT_COMPRESSION,                    0). % zlib compression 0..9
-define(DEFAULT_RETRY,                          0).
-define(DEFAULT_RETRY_DELAY,                    0). % milliseconds
-define(DEFAULT_FAULT_ISOLATION,      destination). % | both

% fault_isolation: destination
-record(destination_request,
    {
        type :: cloudi_service:request_type(),
        name :: cloudi_service:service_name(),
        pattern :: cloudi_service:service_name_pattern(),
        request_info :: cloudi_service:request_info(),
        request :: cloudi_service:request(),
        timeout :: cloudi_service:timeout_value_milliseconds(),
        priority :: cloudi_service:priority(),
        trans_id :: cloudi_service:trans_id(),
        pid :: cloudi_service:source()
    }).
-type request_destination_mode() :: #destination_request{}.
% fault_isolation: both
-record(both_request,
    {
        name :: cloudi_service:service_name(),
        request_info :: cloudi_service:request_info(),
        request :: cloudi_service:request(),
        timeout :: cloudi_service:timeout_value_milliseconds(),
        priority :: cloudi_service:priority(),
        trans_id :: cloudi_service:trans_id(),
        next_name :: cloudi_service:service_name(),
        next_trans_id :: cloudi_service:trans_id()
    }).
-record(both_response,
    {
        name :: cloudi_service:service_name(),
        response_info :: cloudi_service:response_info(),
        response :: cloudi_service:response(),
        timeout :: cloudi_service:timeout_value_milliseconds(),
        priority :: cloudi_service:priority(),
        trans_id :: cloudi_service:trans_id()
    }).
-type request_both_mode() :: #both_request{} | #both_response{}.
% cloudi_write_ahead_logging 
-type request() :: request_destination_mode() |
                   request_both_mode().
-export_type([request/0]).

-record(state,
    {
        service :: pid(),
        logging :: cloudi_write_ahead_logging:state(),
        mode :: destination | both,
        retry :: non_neg_integer(),
        retry_delay :: non_neg_integer(),
        retry_f :: cloudi_write_ahead_logging:retry_function()
    }).

%%%------------------------------------------------------------------------
%%% External interface functions
%%%------------------------------------------------------------------------

%%%------------------------------------------------------------------------
%%% Callback functions from cloudi_service
%%%------------------------------------------------------------------------

cloudi_service_init(Args, Prefix, _Timeout, Dispatcher) ->
    Defaults = [
        {file,                         ?DEFAULT_FILE},
        {compression,           ?DEFAULT_COMPRESSION},
        {retry,                       ?DEFAULT_RETRY},
        {retry_delay,           ?DEFAULT_RETRY_DELAY},
        {fault_isolation,   ?DEFAULT_FAULT_ISOLATION}],
    [FilePath, Compression, Retry, RetryDelay,
     Mode] = cloudi_proplists:take_values(Defaults, Args),
    false = trie:is_pattern(Prefix),
    true = is_list(FilePath) andalso is_integer(hd(FilePath)),
    true = is_integer(Compression) andalso
           (Compression >= 0) andalso (Compression =< 9),
    true = is_integer(Retry) andalso (Retry >= 0),
    true = is_integer(RetryDelay) andalso
           (RetryDelay >= 0) andalso (RetryDelay =< 4294967295),
    true = ((Retry == 0) andalso (RetryDelay == 0)) orelse
           ((Retry > 0) andalso (RetryDelay >= 0)),
    true = ((Mode =:= destination) orelse (Mode =:= both)),
    I = erlang:integer_to_list(cloudi_service:process_index(Dispatcher)),
    Environment = trie:store("I", I,
                                      cloudi_environment:lookup()),
    QueueFilePath = cloudi_environment:transform(FilePath, Environment),
    Service = cloudi_service:self(Dispatcher),
    DispatcherPid = cloudi_service:dispatcher(Dispatcher),
    RetryF = fun(T, RetryT) ->
        retry(T, RetryT, Mode, DispatcherPid, Service)
    end,
    Logging = cloudi_write_ahead_logging:new(QueueFilePath,
                                             Compression,
                                             RetryF),
    cloudi_service:subscribe(Dispatcher, "*"),
    {ok, #state{service = Service,
                logging = Logging,
                mode = Mode,
                retry = Retry,
                retry_delay = RetryDelay,
                retry_f = RetryF}}.

cloudi_service_handle_request(Type, Name, Pattern, RequestInfo, Request,
                              Timeout, Priority, TransId, Pid,
                              #state{service = Service,
                                     logging = Logging,
                                     mode = destination} = State,
                              Dispatcher) ->
    [QueueName] = cloudi_service_name:parse(Name, Pattern),
    ChunkRequest = #destination_request{type = Type,
                                        name = Name,
                                        pattern = Pattern,
                                        request_info = RequestInfo,
                                        request = Request,
                                        timeout = Timeout,
                                        priority = Priority,
                                        trans_id = TransId,
                                        pid = Pid},
    {Chunk, NextLogging} = cloudi_write_ahead_logging:
                           store_start(ChunkRequest, Logging),
    % PERSISTENCE START:
    % at this point the service request has been persisted and
    % can be restarted
    {ok, QueueTransId} = send_async_active(QueueName,
                                           RequestInfo, Request,
                                           Timeout, Priority,
                                           Dispatcher, Service),
    NewLogging = cloudi_write_ahead_logging:
                 store_end(QueueTransId, Chunk, NextLogging),
    {noreply, State#state{logging = NewLogging}};
cloudi_service_handle_request(_Type, Name, Pattern, RequestInfo, Request,
                              Timeout, Priority, TransId, _Pid,
                              #state{service = Service,
                                     logging = Logging,
                                     mode = both} = State,
                              Dispatcher) ->
    RequestMetaData = cloudi_request_info:key_value_parse(RequestInfo),
    case cloudi_key_value:find(<<"service_name">>, RequestMetaData) of
        {ok, NextNameAnyType} ->
            [QueueName] = cloudi_service_name:parse(Name, Pattern),
            NextName = if
                is_binary(NextNameAnyType) ->
                    erlang:binary_to_list(NextNameAnyType);
                is_list(NextNameAnyType), is_integer(hd(NextNameAnyType)) ->
                    NextNameAnyType
            end,
            NextTransId = cloudi_service:trans_id(Dispatcher),
            ChunkRequest = #both_request{name = QueueName,
                                         request_info = RequestInfo,
                                         request = Request,
                                         timeout = Timeout,
                                         priority = Priority,
                                         trans_id = TransId,
                                         next_name = NextName,
                                         next_trans_id = NextTransId},
            {Chunk, NextLogging} = cloudi_write_ahead_logging:
                                   store_start(ChunkRequest, Logging),
            % PERSISTENCE START:
            % at this point the service request has been persisted and
            % can be restarted
            {ok, QueueTransId} = send_async_active(QueueName,
                                                   RequestInfo, Request,
                                                   Timeout, Priority,
                                                   Dispatcher, Service),
            NewLogging = cloudi_write_ahead_logging:
                         store_end(QueueTransId, Chunk, NextLogging),
            {reply, NextTransId, State#state{logging = NewLogging}};
        error ->
            ?LOG_ERROR("service_name not found in RequestInfo for ~s",
                       [cloudi_trans_id:to_string(TransId, nodash)]),
            {reply, <<>>, State}
    end.

cloudi_service_handle_info(#return_async_active{response_info = ResponseInfo,
                                                response = Response,
                                                timeout = Timeout,
                                                trans_id = QueueTransId},
                           #state{logging = Logging,
                                  mode = destination} = State,
                           Dispatcher) ->
    {#destination_request{type = Type,
                          name = Name,
                          pattern = Pattern,
                          trans_id = TransId,
                          pid = Pid},
     NewLogging} = cloudi_write_ahead_logging:
                   erase(QueueTransId, Logging),
    % PERSISTENCE END:
    % at this point the service request is no longer persisted
    cloudi_service:return_nothrow(Dispatcher, Type, Name, Pattern,
                                  ResponseInfo, Response,
                                  Timeout, TransId, Pid),
    {noreply, State#state{logging = NewLogging}};
cloudi_service_handle_info(#return_async_active{response_info = ResponseInfo,
                                                response = Response,
                                                trans_id = QueueTransId},
                           #state{service = Service,
                                  logging = Logging,
                                  mode = both} = State,
                           Dispatcher) ->
    UpdateF = fun
        (#both_request{timeout = Timeout,
                       priority = Priority,
                       next_name = NextName,
                       next_trans_id = NextTransId}) ->
            {NextTransId,
             #both_response{name = NextName,
                            response_info = ResponseInfo,
                            response = Response,
                            timeout = Timeout,
                            priority = Priority,
                            trans_id = NextTransId}};
        (#both_response{}) ->
            undefined
    end,
    {NewChunkRequest,
     NewLogging} = cloudi_write_ahead_logging:
                   update(QueueTransId, UpdateF, Logging),
    case NewChunkRequest of
        undefined ->
            % PERSISTENCE END:
            % at this point the service request is no longer persisted
            % because both the request and the response were delivered
            {noreply, State#state{logging = NewLogging}};
        #both_response{name = NextName,
                       response_info = ResponseInfo,
                       response = Response,
                       timeout = NextTimeout,
                       priority = NextPriority,
                       trans_id = NextTransId} ->
            % deliver the response as a service request
            {ok, _} = send_async_active(NextName, ResponseInfo, Response,
                                        NextTimeout, NextPriority, NextTransId,
                                        Dispatcher, Service),
            {noreply, State#state{logging = NewLogging}}
    end;
cloudi_service_handle_info(#timeout_async_active{trans_id = QueueTransId},
                           #state{logging = Logging,
                                  retry = Retry,
                                  retry_delay = 0,
                                  retry_f = RetryF} = State,
                           _Dispatcher) ->
    NewLogging = cloudi_write_ahead_logging:
                 erase_retry(QueueTransId, Retry, RetryF, Logging),
    % PERSISTENCE END (if a retry didn't occur):
    % at this point the service request/response is no longer persisted
    % due to its timeout value
    {noreply, State#state{logging = NewLogging}};
cloudi_service_handle_info(#timeout_async_active{trans_id = QueueTransId},
                           #state{service = Service,
                                  retry_delay = RetryDelay} = State,
                           _Dispatcher)
    when RetryDelay > 0 ->
    erlang:send_after(RetryDelay, Service, {retry_delay, QueueTransId}),
    {noreply, State};
cloudi_service_handle_info({retry_delay, QueueTransId},
                           #state{logging = Logging,
                                  retry = Retry,
                                  retry_f = RetryF} = State,
                           _Dispatcher) ->
    NewLogging = cloudi_write_ahead_logging:
                 erase_retry(QueueTransId, Retry, RetryF, Logging),
    % PERSISTENCE END (if a retry didn't occur):
    % at this point the service request/response is no longer persisted
    % due to its timeout value
    {noreply, State#state{logging = NewLogging}};
cloudi_service_handle_info(Request, State, _Dispatcher) ->
    ?LOG_WARN("Unknown info \"~p\"", [Request]),
    {noreply, State}.

cloudi_service_terminate(_Reason, _Timeout, #state{}) ->
    ok.

%%%------------------------------------------------------------------------
%%% Private functions
%%%------------------------------------------------------------------------

-compile({inline, [{retry, 5}]}).
-spec retry(request(),
            Retry :: boolean(),
            Mode :: destination | both,
            Dispatcher :: cloudi_service:dispatcher(),
            Service :: cloudi_service:source()) ->
    {ok, cloudi_service:trans_id()} |
    {error, timeout}.

retry(#destination_request{name = Name,
                           pattern = Pattern,
                           request_info = RequestInfo,
                           request = Request,
                           timeout = Timeout,
                           priority = Priority,
                           trans_id = TransId,
                           pid = Pid}, true, destination,
      Dispatcher, Service) ->
    Age = (cloudi_trans_id:microseconds() -
           cloudi_trans_id:microseconds(TransId)) div 1000 + 100, % milliseconds
    case erlang:is_process_alive(Pid) of
        false ->
            {error, timeout};
        true when Age >= Timeout ->
            {error, timeout};
        true ->
            [QueueName] = cloudi_service_name:parse(Name, Pattern),
            NewTimeout = Timeout - Age,
            send_async_active(QueueName, RequestInfo, Request,
                              NewTimeout, Priority,
                              Dispatcher, Service)
    end;
retry(#both_request{name = QueueName,
                    request_info = RequestInfo,
                    request = Request,
                    timeout = Timeout,
                    priority = Priority,
                    trans_id = TransId}, Retry, both,
      Dispatcher, Service) ->
    Age = (cloudi_trans_id:microseconds() -
           cloudi_trans_id:microseconds(TransId)) div 1000 + 100, % milliseconds
    if
        Age >= Timeout ->
            % an empty response will be handled due to the request timeout
            % (always occurs, even without using retries)
            Service ! #return_async_active{name = QueueName,
                                           pattern = QueueName,
                                           response_info = <<>>,
                                           response = <<>>,
                                           timeout = Timeout,
                                           trans_id = TransId},
            {ok, TransId};
        Retry =:= true ->
            NewTimeout = Timeout - Age,
            send_async_active(QueueName, RequestInfo, Request,
                              NewTimeout, Priority,
                              Dispatcher, Service);
        Retry =:= false ->
            {error, timeout}
    end;
retry(#both_response{name = NextName,
                     response_info = ResponseInfo,
                     response = Response,
                     timeout = NextTimeout,
                     priority = NextPriority,
                     trans_id = NextTransId}, true, both,
      Dispatcher, Service) ->
    send_async_active(NextName, ResponseInfo, Response,
                      NextTimeout, NextPriority, NextTransId,
                      Dispatcher, Service);
retry(_, false, _, _, _) ->
    {error, timeout}.

send_async_active(Name, ResponseInfo, Response, Timeout, Priority, TransId,
                  Dispatcher, Service) ->
    Result = case cloudi_service:get_pid(Dispatcher, Name, Timeout) of
        {ok, PatternPid} ->
            cloudi_service:send_async_active(Dispatcher, Name,
                                             ResponseInfo, Response,
                                             Timeout, Priority,
                                             TransId, PatternPid);
        {error, _} = Error ->
            Error
    end,
    case Result of
        {ok, _} = Success ->
            Success;
        {error, timeout} ->
            Service ! #timeout_async_active{trans_id = TransId},
            {ok, TransId}
    end.

send_async_active(Name, RequestInfo, Request, Timeout, Priority,
                  Dispatcher, Service) ->
    case cloudi_service:send_async_active(Dispatcher, Name,
                                          RequestInfo, Request,
                                          Timeout, Priority) of
        {ok, _} = Success ->
            Success;
        {error, timeout} ->
            TransId = cloudi_service:trans_id(Dispatcher),
            Service ! #timeout_async_active{trans_id = TransId},
            {ok, TransId}
    end.