%-*-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 Funnel Service== %%% The funnel service is a way of using duplicate service request sends %%% (possibly from separate service processes processing the same data) %%% with the goal of making the sending service fault-tolerant. %%% %%% The funnel service is the opposite of cloudi_service_quorum due to %%% receiving many service requests that may be duplicates and only sending %%% unique service requests with distributed fault-tolerance. %%% A duplicate service request has the same service name, request info and %%% request. Each duplicate still contains a unique trans_id and is only %%% a duplicate while the timeout time periods elapse. %%% %%% The funnel service creates its own service request for the first %%% unique service request it receives and any response it receives will be %%% used for replies to duplicate service requests. The last recent %%% timeout time period before the response is received is used for %%% retaining the response data for future duplicate service requests. %%% %%% With the funnel service's sensitivity to a service request's %%% timeout value, other more robust approaches to fault-tolerance %%% should be preferred %%% (e.g., cloudi_service_queue and/or cloudi_service_quorum). %%% The funnel service provides a way to make a CloudI service %%% fault-tolerant without modifying the service %%% (assuming the send destinations are easy to point at the funnel service) %%% but it does add latency to the service request sends and may require %%% larger timeout values. If timeout values are too small, %%% duplicate service requests may not be recognized by the funnel service %%% and it could send more than a single service request for a group %%% of duplicate service requests it receives. %%% %%% The funnel service could receive separate service request sends from %%% separate instances of cloudi_service_cron that are using the %%% same initialization arguments %%% (the cron expression arguments could also include {send_mcast, true} %%% to ensure all cloudi_service_funnel processes get the cron expression %%% service request data, {send_args_info, true} is necessary to make each %%% cron event's service request unique). That would provide %%% cloudi_service_cron execution with distributed fault-tolerance that %%% relies on cloudi_service_funnel execution. %%% @end %%% %%% MIT License %%% %%% Copyright (c) 2020-2022 Michael Truog %%% %%% Permission is hereby granted, free of charge, to any person obtaining a %%% copy of this software and associated documentation files (the "Software"), %%% to deal in the Software without restriction, including without limitation %%% the rights to use, copy, modify, merge, publish, distribute, sublicense, %%% and/or sell copies of the Software, and to permit persons to whom the %%% Software is furnished to do so, subject to the following conditions: %%% %%% The above copyright notice and this permission notice shall be included in %%% all copies or substantial portions of the Software. %%% %%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR %%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, %%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE %%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER %%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING %%% FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER %%% DEALINGS IN THE SOFTWARE. %%% %%% @author Michael Truog %%% @copyright 2020-2022 Michael Truog %%% @version 2.0.5 {@date} {@time} %%%------------------------------------------------------------------------ -module(cloudi_service_funnel). -author('mjtruog at protonmail dot com'). -behaviour(cloudi_service). %% cloudi_crdt update functions -export([senders_crdt_merge/2, request_crdt_merge/2, request_send_retry_crdt/1, response_store_crdt/2, response_timeout_crdt/1]). %% 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_constants.hrl"). -include_lib("cloudi_core/include/cloudi_logger.hrl"). -include_lib("cloudi_core/include/cloudi_service.hrl"). -include_lib("cloudi_core/include/cloudi_crdt.hrl"). -define(DEFAULT_NAME, "funnel"). % Funnel name used for incoming service requests % (creates subscription Prefix ++ FunnelName ++ "*" with the matching % suffix used as the request's service name) -define(DEFAULT_NODE_COUNT, 1). % Count of connected nodes using the funnel service % with the same name argument. -define(DEFAULT_RECEIVE_ONLY, false). % Do not send from this service instance % (the node may be expected to be unreliable). -define(DEFAULT_REQUESTS_EXTEND_LIVENESS, false). % Should later requests extend the liveness of the % response data, delaying its expiration past the timeout % of the first request? -define(DEFAULT_RETRY, 0). -define(DEFAULT_RETRY_DELAY, 0). % milliseconds -type request_key() :: {cloudi_service:service_name(), cloudi_service:request_info(), cloudi_service:request()}. -type process_index() :: non_neg_integer(). -type sender_id() :: {node(), process_index()}. -record(request, { name :: cloudi_service:service_name(), pattern :: cloudi_service:service_name_pattern(), timeout_last :: cloudi_service:timeout_value_milliseconds(), priority_min :: cloudi_service:priority_value(), trans_id_first :: cloudi_service:trans_id(), trans_id_last :: cloudi_service:trans_id(), pending :: list({cloudi_service:request_type(), cloudi_service:timeout_value_milliseconds(), cloudi_service:trans_id(), pid()}), sender_id :: sender_id(), response = undefined :: undefined | {cloudi_service:response_info(), cloudi_service:response()}, retry_count = 0 :: non_neg_integer() }). -record(senders, { sender_ids :: list(sender_id()), length :: non_neg_integer(), merges :: pos_integer() }). -record(state, { sender_id :: sender_id(), service :: cloudi_service:source(), name :: nonempty_string(), init_done :: boolean(), receive_only :: boolean(), requests_extend_liveness :: boolean(), retry :: non_neg_integer(), retry_delay :: non_neg_integer(), crdt :: cloudi_crdt:state(), sent = #{} :: #{cloudi_service:trans_id() := request_key()} }). %%%------------------------------------------------------------------------ %%% External interface functions %%%------------------------------------------------------------------------ %%%------------------------------------------------------------------------ %%% Callback functions from cloudi_service %%%------------------------------------------------------------------------ cloudi_service_init(Args, Prefix, _Timeout, Dispatcher) -> Defaults = [ {name, ?DEFAULT_NAME}, {node_count, ?DEFAULT_NODE_COUNT}, {receive_only, ?DEFAULT_RECEIVE_ONLY}, {requests_extend_liveness, ?DEFAULT_REQUESTS_EXTEND_LIVENESS}, {retry, ?DEFAULT_RETRY}, {retry_delay, ?DEFAULT_RETRY_DELAY}], [Name, NodeCount, ReceiveOnly, RequestsExtendLiveness, Retry, RetryDelay0] = cloudi_proplists:take_values(Defaults, Args), false = cloudi_service_name:pattern(Prefix), true = cloudi_service:destination_refresh_immediate(Dispatcher), true = is_list(Name) andalso is_integer(hd(Name)), false = lists:member($/, Name), false = cloudi_service_name:pattern(Name), true = is_integer(NodeCount) andalso (NodeCount >= 1), true = is_boolean(ReceiveOnly), true = is_boolean(RequestsExtendLiveness), true = is_integer(Retry) andalso (Retry >= 0), RetryDelayN = cloudi_args_type: period_to_milliseconds(RetryDelay0, 0, ?TIMEOUT_MAX_ERLANG), true = ((Retry == 0) andalso (RetryDelayN == 0)) orelse ((Retry > 0) andalso (RetryDelayN >= 0)), ProcessIndex = cloudi_service:process_index(Dispatcher), SenderId = {node(), ProcessIndex}, Service = cloudi_service:self(Dispatcher), InitialDataF = crdt_data_restart_f(Service), CRDT0 = cloudi_crdt:new(Dispatcher, [{service_name, "crdt_" ++ Name}, {node_count, NodeCount}, {initial_data_function, InitialDataF}, {retry, 20}, % 5 minutes total {retry_delay, {15, seconds}}, {priority_default_offset, -1}]), CRDTN = cloudi_crdt:events_subscriptions(Dispatcher, [assign, update], CRDT0), Service ! crdt_data_init, {ok, #state{sender_id = SenderId, service = Service, name = Name, init_done = false, receive_only = ReceiveOnly, requests_extend_liveness = RequestsExtendLiveness, retry = Retry, retry_delay = RetryDelayN, crdt = CRDTN}}. cloudi_service_handle_request(RequestType, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source, #state{crdt = CRDT0} = State, Dispatcher) -> case cloudi_service_name:parse(Name, Pattern) of [FunnelName] -> request(FunnelName, RequestType, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source, State, Dispatcher); [] -> {ok, CRDTN} = cloudi_crdt: handle_request(RequestType, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source, CRDT0, Dispatcher), {noreply, State#state{crdt = CRDTN}} end. cloudi_service_handle_info({response_timeout, RequestKey}, #state{crdt = CRDT0} = State, Dispatcher) -> UpdateF = response_timeout_crdt, CRDTN = cloudi_crdt:update_clear_id(Dispatcher, RequestKey, ?MODULE, UpdateF, UpdateF, CRDT0), {noreply, State#state{crdt = CRDTN}}; cloudi_service_handle_info({retry, RequestKey}, #state{crdt = CRDT} = State, Dispatcher) -> RequestValue = cloudi_crdt:get(Dispatcher, RequestKey, CRDT), StateNew = request_send(RequestKey, RequestValue, State, Dispatcher), {noreply, StateNew}; cloudi_service_handle_info(crdt_data_init, #state{sender_id = SenderId, receive_only = ReceiveOnly, crdt = CRDT0} = State, Dispatcher) -> % create a consistent store of sender_ids among all the service processes % (required to ensure concurrent service requests choose the same sender_id % when sending a service request for the first unique request). SendersValue = if ReceiveOnly =:= true -> #senders{sender_ids = [], length = 0, merges = 1}; ReceiveOnly =:= false -> #senders{sender_ids = [SenderId], length = 1, merges = 1} end, UpdateF = senders_crdt_merge, CRDTN = cloudi_crdt:update_assign_id(Dispatcher, senders, SendersValue, ?MODULE, UpdateF, SendersValue, {UpdateF, SenderId}, CRDT0), {noreply, State#state{crdt = CRDTN}}; cloudi_service_handle_info({crdt_data_restart, Data}, State, Dispatcher) -> StateNew = crdt_data_restart(Data, State, Dispatcher), {noreply, StateNew}; cloudi_service_handle_info(#crdt_event{type = assign, id = request_crdt_merge, key = RequestKey, new = {value, RequestValue}}, State, Dispatcher) -> StateNew = request_send(RequestKey, RequestValue, State, Dispatcher), {noreply, StateNew}; cloudi_service_handle_info(#crdt_event{type = update, id = {request_send_retry_crdt, SenderIdRetry}, key = RequestKey}, #state{sender_id = SenderId, service = Service, retry_delay = RetryDelay} = State, _Dispatcher) -> ok = sender_only(fun() -> erlang:send_after(RetryDelay, Service, {retry, RequestKey}) end, SenderId, SenderIdRetry), {noreply, State}; cloudi_service_handle_info(#crdt_event{type = update, id = response_store_crdt, key = RequestKey, old = {value, RequestValueOld}, new = {value, RequestValueNew}}, #state{sender_id = SenderId, service = Service} = State, Dispatcher) -> #request{name = Name, pattern = Pattern, pending = Pending, sender_id = SenderIdRequest} = RequestValueOld, #request{timeout_last = TimeoutLast, trans_id_last = TransIdLast, pending = [], response = {ResponseInfo, Response}} = RequestValueNew, ok = sender_only(fun() -> ok = response_return(Pending, Name, Pattern, ResponseInfo, Response, cloudi_trans_id:microseconds(), Dispatcher), ok = response_timeout(TimeoutLast, TransIdLast, RequestKey, Service) end, SenderId, SenderIdRequest), {noreply, State}; cloudi_service_handle_info(#crdt_event{type = update, id = response_timeout_crdt, key = RequestKey, new = {value, RequestValue}}, #state{sender_id = SenderId, service = Service} = State, _Dispatcher) -> #request{timeout_last = TimeoutLast, trans_id_last = TransIdLast, sender_id = SenderIdRequest, response = {_, _}} = RequestValue, ok = sender_only(fun() -> ok = response_timeout(TimeoutLast, TransIdLast, RequestKey, Service) end, SenderId, SenderIdRequest), {noreply, State}; cloudi_service_handle_info(#crdt_event{id = {senders_crdt_merge, SenderIdMerge}, key = senders, new = {value, SendersValue}}, #state{sender_id = SenderId, name = Name, init_done = false, crdt = CRDT} = State, Dispatcher) -> InitDone = senders_ready(SendersValue, cloudi_crdt:crdt_size(CRDT), SenderIdMerge, SenderId), if InitDone =:= true -> % ensure that all cloudi_service_funnel service instances % do not have receive_only set to true #senders{sender_ids = [_ | _]} = SendersValue, ok = cloudi_service:subscribe(Dispatcher, Name ++ "*"); InitDone =:= false -> ok end, {noreply, State#state{init_done = InitDone}}; cloudi_service_handle_info(#crdt_event{}, State, _Dispatcher) -> {noreply, State}; cloudi_service_handle_info(Request, #state{crdt = CRDT0, sent = Sent} = State, Dispatcher) -> case cloudi_crdt:handle_info(Request, CRDT0, Dispatcher) of {ok, CRDTN} -> {noreply, State#state{crdt = CRDTN}}; {ignored, CRDTN} -> StateNew = case Request of #return_async_active{response_info = ResponseInfo, response = Response, trans_id = TransId} -> {RequestKey, SentNew} = maps:take(TransId, Sent), response_store(RequestKey, ResponseInfo, Response, State#state{crdt = CRDTN, sent = SentNew}, Dispatcher); #timeout_async_active{trans_id = TransId} -> {RequestKey, SentNew} = maps:take(TransId, Sent), request_send_retry(RequestKey, State#state{sent = SentNew}, Dispatcher) end, {noreply, StateNew} end. cloudi_service_terminate(_Reason, _Timeout, _State) -> ok. %%%------------------------------------------------------------------------ %%% Private functions %%%------------------------------------------------------------------------ senders_crdt_merge(#senders{sender_ids = SenderIds, merges = 1}, #senders{sender_ids = SenderIdsOld, merges = MergesOld} = SendersOld) -> SenderIdsNew = lists:umerge(SenderIdsOld, SenderIds), SendersOld#senders{sender_ids = SenderIdsNew, length = length(SenderIdsNew), merges = MergesOld + 1}. senders_ready(#senders{merges = MergesExpected}, MergesExpected, _, _) -> true; senders_ready(#senders{merges = Merges}, MergesExpected, SenderId, SenderId) when Merges > MergesExpected -> true; % restart occurred senders_ready(_, _, _, _) -> false. crdt_data_restart_f(Service) -> fun(Data) -> Service ! {crdt_data_restart, Data} end. crdt_data_restart(Data, State, Dispatcher) -> % process data after a restart that will not be present in events % is used to create new service requests for this sender, % if no response is already present maps:fold(fun(RequestKey, RequestValue, StateNext) -> request_send(RequestKey, RequestValue, StateNext, Dispatcher) end, State, maps:remove(senders, Data)). request(FunnelName, RequestType, Name, Pattern, RequestInfo, Request, Timeout, Priority, TransId, Source, #state{requests_extend_liveness = RequestsExtendLiveness, crdt = CRDT0} = State, Dispatcher) -> RequestKey = {FunnelName, RequestInfo, Request}, case cloudi_crdt:find(Dispatcher, RequestKey, CRDT0) of {ok, #request{response = {ResponseInfo, Response}}} -> {reply, ResponseInfo, Response, State}; _ -> Pending = [{RequestType, Timeout, TransId, Source}], SenderIdRequest = request_sender_id(RequestKey, CRDT0, Dispatcher), RequestValue = #request{name = Name, pattern = Pattern, timeout_last = Timeout, priority_min = Priority, trans_id_first = TransId, trans_id_last = TransId, pending = Pending, sender_id = SenderIdRequest}, UpdateF = request_crdt_merge, CRDTN = cloudi_crdt:update_assign_id(Dispatcher, RequestKey, RequestValue, ?MODULE, UpdateF, [RequestValue, RequestsExtendLiveness], UpdateF, CRDT0), {noreply, State#state{crdt = CRDTN}} end. request_sender_id(RequestKey, CRDT, Dispatcher) -> % load balances with a hash function that is used in the same way % in all service processes % (to ensure the sender_id is consistently assigned when duplicate % service requests are concurrently received in separate processes) SendersValue = cloudi_crdt:get(Dispatcher, senders, CRDT), #senders{sender_ids = SenderIds, length = Length} = SendersValue, lists:nth(erlang:phash2(RequestKey, Length) + 1, SenderIds). request_crdt_merge([#request{name = Name, pattern = Pattern, timeout_last = Timeout, priority_min = Priority, trans_id_first = TransId, trans_id_last = TransId, pending = [PendingValue]}, RequestsExtendLiveness], #request{name = Name, pattern = Pattern, timeout_last = TimeoutLastOld, priority_min = PriorityMinOld, trans_id_first = TransIdFirstOld, trans_id_last = TransIdLastOld, pending = PendingOld} = RequestValueOld) -> MicroSeconds = cloudi_trans_id:microseconds(TransId), MicroSecondsFirstOld = cloudi_trans_id:microseconds(TransIdFirstOld), MicroSecondsLastOld = cloudi_trans_id:microseconds(TransIdLastOld), TransIdFirst = if MicroSecondsFirstOld < MicroSeconds -> TransIdFirstOld; true -> TransId end, {TimeoutLast, TransIdLast} = if RequestsExtendLiveness =:= true, MicroSecondsLastOld < MicroSeconds -> {Timeout, TransId}; true -> {TimeoutLastOld, TransIdLastOld} end, PriorityMin = erlang:min(PriorityMinOld, Priority), RequestValueOld#request{timeout_last = TimeoutLast, priority_min = PriorityMin, trans_id_first = TransIdFirst, trans_id_last = TransIdLast, pending = [PendingValue | PendingOld]}. request_send(RequestKey, #request{timeout_last = TimeoutLast, priority_min = PriorityMin, trans_id_last = TransIdLast, sender_id = SenderId, response = undefined} = RequestValue, #state{sender_id = SenderId, sent = Sent} = State, Dispatcher) -> {FunnelName, RequestInfo, Request} = RequestKey, Timeout = timeout_current(TimeoutLast, TransIdLast), case cloudi_service:send_async_active(Dispatcher, FunnelName, RequestInfo, Request, Timeout, PriorityMin) of {ok, TransId} -> State#state{sent = maps:put(TransId, RequestKey, Sent)}; {error, timeout} -> request_send_retry(RequestKey, RequestValue, State, Dispatcher) end; request_send(_, _, State, _) -> State. request_send_retry(RequestKey, #state{crdt = CRDT} = State, Dispatcher) -> RequestValue = cloudi_crdt:get(Dispatcher, RequestKey, CRDT), request_send_retry(RequestKey, RequestValue, State, Dispatcher). request_send_retry(RequestKey, #request{retry_count = RetryCount}, #state{sender_id = SenderId, retry = RetryCountMax, crdt = CRDT0} = State, Dispatcher) -> if RetryCount < RetryCountMax -> UpdateF = request_send_retry_crdt, CRDTN = cloudi_crdt:update_id(Dispatcher, RequestKey, ?MODULE, UpdateF, {UpdateF, SenderId}, CRDT0), State#state{crdt = CRDTN}; RetryCount == RetryCountMax -> ResponseInfo = <<>>, Response = <<>>, response_store(RequestKey, ResponseInfo, Response, State, Dispatcher) end. request_send_retry_crdt(#request{retry_count = RetryCount} = RequestValue) -> RequestValue#request{retry_count = RetryCount + 1}. response_store(RequestKey, ResponseInfo, Response, #state{crdt = CRDT0} = State, Dispatcher) -> ResponseData = {ResponseInfo, Response}, CRDT1 = cloudi_crdt:flush_next_write(Dispatcher, CRDT0), UpdateF = response_store_crdt, CRDTN = cloudi_crdt:update_id(Dispatcher, RequestKey, ?MODULE, UpdateF, ResponseData, UpdateF, CRDT1), State#state{crdt = CRDTN}. response_store_crdt(ResponseData, RequestValue) -> RequestValue#request{pending = [], response = ResponseData}. response_return([], _, _, _, _, _, _) -> ok; response_return([{RequestType, Timeout, TransId, Source} | Pending], Name, Pattern, ResponseInfo, Response, MicroSecondsNow, Dispatcher) -> MilliSecondsElapsed = (MicroSecondsNow - cloudi_trans_id:microseconds(TransId)) div 1000, if MilliSecondsElapsed =< Timeout -> cloudi_service:return_nothrow(Dispatcher, RequestType, Name, Pattern, ResponseInfo, Response, Timeout, TransId, Source); true -> ok end, response_return(Pending, Name, Pattern, ResponseInfo, Response, MicroSecondsNow, Dispatcher). response_timeout(TimeoutLast, TransIdLast, RequestKey, Service) -> ResponseTimeout = timeout_current(TimeoutLast, TransIdLast), erlang:send_after(ResponseTimeout, Service, {response_timeout, RequestKey}), ok. response_timeout_crdt(#request{timeout_last = TimeoutLast, trans_id_last = TransIdLast} = RequestValue) -> case timeout_current(TimeoutLast, TransIdLast) of 0 -> undefined; _ -> RequestValue end. timeout_current(TimeoutLast, TransIdLast) -> MilliSecondsElapsed = (cloudi_trans_id:microseconds() - cloudi_trans_id:microseconds(TransIdLast)) div 1000, if TimeoutLast < MilliSecondsElapsed -> 0; true -> TimeoutLast - MilliSecondsElapsed end. sender_only(F, SenderId, SenderId) -> _ = F(), ok; sender_only(_, _, _) -> ok.