%-*-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 %%% 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 %%% @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.