Current section
Files
Jump to
Current section
Files
src/cloudi_service_map_reduce.erl
%-*-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 (Abstract) Map-Reduce Service==
%%% This module provides an Erlang behaviour for fault-tolerant,
%%% database agnostic map-reduce. See the hexpi test for example usage.
%%% @end
%%%
%%% MIT License
%%%
%%% Copyright (c) 2012-2023 Michael Truog <mjtruog at protonmail dot com>
%%%
%%% 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 <mjtruog at protonmail dot com>
%%% @copyright 2012-2023 Michael Truog
%%% @version 2.0.6 {@date} {@time}
%%%------------------------------------------------------------------------
-module(cloudi_service_map_reduce).
-author('mjtruog at protonmail dot com').
-behaviour(cloudi_service).
%% external interface
-export([aspect_suspend/1,
aspect_resume/1,
elapsed_seconds/0,
map_size/0]).
%% cloudi_service callbacks
-export([cloudi_service_init/4,
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").
-define(DEFAULT_MAP_REDUCE_MODULE, undefined).
-define(DEFAULT_MAP_REDUCE_ARGUMENTS, []).
-define(DEFAULT_NAME, "controller").
-define(DEFAULT_CONCURRENCY, 1.0).
% logical cpu count multiplier
% (calculated the same way as count_process and count_thread in
% service configuration).
-define(DEFAULT_LOG_EXECUTION_TIME, true).
-define(DEFAULT_RETRY, 3).
% Max retries with a Timeout value of ?TIMEOUT_MAX_ERLANG
% before exiting with retry_max.
% The Timeout value needs to be provided in the map_send_args()
% and is typically increasing due to service request failures.
-define(DEFAULT_RETRY_DELAY, 0). % milliseconds
-type map_send_args() :: nonempty_list().
-export_type([map_send_args/0]).
-record(map_send,
{
send_args
:: map_send_args(),
retry_count
:: non_neg_integer()
}).
-record(state,
{
service
:: pid(),
map_reduce_module
:: module(),
map_reduce_state
:: any(),
map_reduce_name
:: string(),
map_count
:: pos_integer(),
log_execution_time
:: boolean(),
retry
:: non_neg_integer(),
retry_delay
:: non_neg_integer(),
map_requests
:: #{cloudi_service:trans_id() := #map_send{}},
time_running
:: cloudi_timestamp:seconds_monotonic(),
suspended = false
:: boolean(),
elapsed_seconds = 0
:: non_neg_integer()
}).
-record(init_state,
{
service
:: pid(),
info_queued = []
:: list()
}).
-record(init_begin,
{
service
:: pid(),
prefix
:: string(),
timeout
:: cloudi_service_api:timeout_initialize_value_milliseconds(),
map_reduce_module
:: module(),
map_reduce_args
:: list(),
map_reduce_name
:: string(),
concurrency
:: number(),
log_execution_time
:: boolean(),
retry
:: non_neg_integer(),
retry_delay
:: non_neg_integer(),
time_start
:: cloudi_timestamp:seconds_monotonic()
}).
-record(init_end,
{
state
:: undefined | #state{},
error = undefined
:: any()
}).
-record(cloudi_service_map_reduce_info,
{
map_requests_size
:: non_neg_integer(),
time_running
:: cloudi_timestamp:seconds_monotonic(),
elapsed_seconds
:: non_neg_integer()
}).
-define(CALLBACK_INFO_PDICT_KEY, cloudi_service_map_reduce_info).
%%%------------------------------------------------------------------------
%%% Callback functions from behavior
%%%------------------------------------------------------------------------
-callback cloudi_service_map_reduce_new(ModuleReduceArgs :: list(),
Count :: pos_integer(),
Prefix :: string(),
Timeout ::
cloudi_service_api:
timeout_initialize_value_milliseconds(),
Dispatcher :: pid()) ->
{'ok', ModuleReduceState :: any()} |
{'error', Reason :: any()}.
-callback cloudi_service_map_reduce_send(ModuleReduceState :: any(),
Dispatcher :: pid()) ->
{'ok', SendArgs :: map_send_args(), ModuleReduceStateNew :: any()} |
{'done', ModuleReduceStateNew :: any()} |
{'error', Reason :: any()}.
-callback cloudi_service_map_reduce_resend(SendArgs :: map_send_args(),
ModuleReduceState :: any()) ->
{'ok', SendArgsNew :: map_send_args(), ModuleReduceStateNew :: any()} |
{'error', Reason :: any()}.
-callback cloudi_service_map_reduce_recv(SendArgs :: map_send_args(),
ResponseInfo :: any(),
Response :: any(),
Timeout :: non_neg_integer(),
TransId :: binary(),
ModuleReduceState :: any(),
Dispatcher :: pid()) ->
{'ok', ModuleReduceStateNew :: any()} |
{'done', ModuleReduceStateNew :: any()} |
{'error', Reason :: any()}.
-callback cloudi_service_map_reduce_info(Request :: any(),
ModuleReduceState :: any(),
Dispatcher :: pid()) ->
{'ok', ModuleReduceStateNew :: any()} |
{'done', ModuleReduceStateNew :: any()} |
{'error', Reason :: any()}.
%%%------------------------------------------------------------------------
%%% External interface functions
%%%------------------------------------------------------------------------
%%-------------------------------------------------------------------------
%% @doc
%% ===Function for aspects_suspend service configuration option.===
%% Add as {cloudi_service_map_reduce, aspect_suspend}.
%% @end
%%-------------------------------------------------------------------------
-spec aspect_suspend(State :: #state{}) ->
{ok, #state{}}.
aspect_suspend(#state{map_reduce_module = MapReduceModule,
map_reduce_name = MapReduceName,
time_running = TimeRunningStart,
suspended = false,
elapsed_seconds = ElapsedSeconds} = State) ->
TimeRunningEnd = cloudi_timestamp:seconds_monotonic(),
ElapsedSecondsNew = ElapsedSeconds + (TimeRunningEnd - TimeRunningStart),
?LOG_INFO("~s ~ts suspended",
[MapReduceModule, cloudi_service_name:utf8(MapReduceName)]),
{ok,
State#state{suspended = true,
elapsed_seconds = ElapsedSecondsNew}}.
%%-------------------------------------------------------------------------
%% @doc
%% ===Function for aspects_resume service configuration option.===
%% Add as {cloudi_service_map_reduce, aspect_resume}.
%% @end
%%-------------------------------------------------------------------------
-spec aspect_resume(State :: #state{}) ->
{ok, #state{}}.
aspect_resume(#state{map_reduce_module = MapReduceModule,
map_reduce_name = MapReduceName,
suspended = true} = State) ->
TimeRunningStart = cloudi_timestamp:seconds_monotonic(),
?LOG_INFO("~s ~ts resumed",
[MapReduceModule, cloudi_service_name:utf8(MapReduceName)]),
{ok,
State#state{time_running = TimeRunningStart,
suspended = false}}.
%%-------------------------------------------------------------------------
%% @doc
%% ===Get the current elapsed seconds in a Map-Reduce callback function.===
%% @end
%%-------------------------------------------------------------------------
-spec elapsed_seconds() ->
cloudi_service_api:seconds().
elapsed_seconds() ->
#cloudi_service_map_reduce_info{
time_running = TimeRunningStart,
elapsed_seconds = ElapsedSeconds
} = erlang:get(?CALLBACK_INFO_PDICT_KEY),
ElapsedSeconds + (cloudi_timestamp:seconds_monotonic() - TimeRunningStart).
%%-------------------------------------------------------------------------
%% @doc
%% ===Get the current number of map service requests pending in a Map-Reduce callback function.===
%% The maximum integer value returned is the integer equivalent of the
%% concurrency argument.
%% @end
%%-------------------------------------------------------------------------
-spec map_size() ->
non_neg_integer().
map_size() ->
#cloudi_service_map_reduce_info{
map_requests_size = MapRequestsSize
} = erlang:get(?CALLBACK_INFO_PDICT_KEY),
MapRequestsSize.
%%%------------------------------------------------------------------------
%%% Callback functions from cloudi_service
%%%------------------------------------------------------------------------
cloudi_service_init(Args, Prefix, Timeout, Dispatcher) ->
Defaults = [
{map_reduce, ?DEFAULT_MAP_REDUCE_MODULE},
{map_reduce_args, ?DEFAULT_MAP_REDUCE_ARGUMENTS},
{name, ?DEFAULT_NAME},
{concurrency, ?DEFAULT_CONCURRENCY},
{log_execution_time, ?DEFAULT_LOG_EXECUTION_TIME},
{retry, ?DEFAULT_RETRY},
{retry_delay, ?DEFAULT_RETRY_DELAY}],
[MapReduceModule, MapReduceArgs, Name, Concurrency, LogExecutionTime,
Retry, RetryDelay0] =
cloudi_proplists:take_values(Defaults, Args),
TimeStart = cloudi_timestamp:seconds_monotonic(),
true = is_atom(MapReduceModule) andalso (MapReduceModule /= undefined),
true = is_list(MapReduceArgs),
true = is_number(Concurrency) andalso (Concurrency > 0),
true = is_boolean(LogExecutionTime),
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)),
case application:load(MapReduceModule) of
ok ->
ok = reltool_util:application_start(MapReduceModule,
[], Timeout);
{error, {already_loaded, MapReduceModule}} ->
ok = reltool_util:application_start(MapReduceModule,
[], Timeout);
{error, _} ->
ok = reltool_util:module_loaded(MapReduceModule)
end,
% cloudi_service_init/4 is always executed by the service process
Service = self(),
Service ! #init_begin{service = Service,
prefix = Prefix,
timeout = Timeout,
map_reduce_module = MapReduceModule,
map_reduce_args = MapReduceArgs,
map_reduce_name = Name,
concurrency = Concurrency,
log_execution_time = LogExecutionTime,
retry = Retry,
retry_delay = RetryDelayN,
time_start = TimeStart},
ok = cloudi_service:subscribe(Dispatcher, Name),
{ok, #init_state{service = Service}}.
cloudi_service_handle_info(#init_begin{service = Service} = InitBegin,
#init_state{} = InitState,
Dispatcher) ->
% cloudi_service_map_reduce_new/5 execution occurs outside of
% cloudi_service_init/4 to allow send_sync and recv_async function calls
% (only algorithmic initialization should be occurring).
% Initialization is done in a temporary process so there is
% no blocking problem (with send_sync or recv_async)
% if duo_mode == true. No timeout is enforced on
% cloudi_service_map_reduce_new/5 execution.
_ = erlang:spawn(fun() ->
true = erlang:link(Service),
case init(InitBegin, Dispatcher) of
{noreply, State} ->
Service ! #init_end{state = State},
ok;
{stop, Reason, State} ->
true = Reason /= undefined,
Service ! #init_end{state = State,
error = Reason},
ok
end,
true = erlang:unlink(Service)
end),
{noreply, InitState};
cloudi_service_handle_info(#init_end{} = InitEnd,
#init_state{} = InitState,
_Dispatcher) ->
init_end(InitEnd, InitState);
cloudi_service_handle_info(Request,
#init_state{info_queued = InfoQueued} = InitState,
_Dispatcher) ->
{noreply, InitState#init_state{info_queued = [Request | InfoQueued]}};
cloudi_service_handle_info(#timeout_async_active{trans_id = TransId} = Request,
#state{map_requests = MapRequests} = State,
Dispatcher) ->
case maps:take(TransId, MapRequests) of
{MapRequest, MapRequestsNew} ->
map_retry(MapRequest,
State#state{map_requests = MapRequestsNew},
Dispatcher);
error ->
map_info(Request, State, Dispatcher)
end;
cloudi_service_handle_info({cloudi_service_map_reduce_retry, MapRequest},
State, Dispatcher) ->
map_resend(MapRequest, State, Dispatcher);
cloudi_service_handle_info(#return_async_active{response_info = ResponseInfo,
response = Response,
timeout = Timeout,
trans_id = TransId} = Request,
#state{map_reduce_module = MapReduceModule,
map_reduce_state = MapReduceState,
map_requests = MapRequests,
time_running = TimeRunningStart,
elapsed_seconds = ElapsedSeconds} = State,
Dispatcher) ->
case maps:take(TransId, MapRequests) of
{#map_send{send_args = [_ | SendArgs]},
MapRequestsNew} ->
ok = callback_info_set(maps:size(MapRequestsNew),
TimeRunningStart, ElapsedSeconds),
case MapReduceModule:
cloudi_service_map_reduce_recv([Dispatcher | SendArgs],
ResponseInfo, Response,
Timeout, TransId,
MapReduceState, Dispatcher) of
{ok, MapReduceStateNew} ->
StateNew = State#state{map_reduce_state = MapReduceStateNew,
map_requests = MapRequestsNew},
map_check_continue(StateNew, Dispatcher);
{done, MapReduceStateNew} ->
StateNew = State#state{map_reduce_state = MapReduceStateNew,
map_requests = MapRequestsNew},
map_check_done(StateNew);
{error, _} = Error ->
{stop, Error, State}
end;
error ->
map_info(Request, State, Dispatcher)
end;
cloudi_service_handle_info(Request, State, Dispatcher) ->
map_info(Request, State, Dispatcher).
cloudi_service_terminate(shutdown, _Timeout,
#state{log_execution_time = true,
map_requests = #{},
time_running = TimeRunningStart,
suspended = Suspended,
elapsed_seconds = ElapsedSeconds}) ->
ElapsedSecondsNew = if
Suspended =:= true ->
ElapsedSeconds;
Suspended =:= false ->
TimeRunningEnd = cloudi_timestamp:seconds_monotonic(),
ElapsedSeconds + (TimeRunningEnd - TimeRunningStart)
end,
?LOG_INFO("total time taken was ~p hours",
[hours_elapsed(ElapsedSecondsNew)]),
ok;
cloudi_service_terminate(_Reason, _Timeout, _State) ->
ok.
%%%------------------------------------------------------------------------
%%% Private functions
%%%------------------------------------------------------------------------
init(#init_begin{service = Service,
prefix = Prefix,
timeout = Timeout,
map_reduce_module = MapReduceModule,
map_reduce_args = MapReduceArgs,
map_reduce_name = MapReduceName,
concurrency = Concurrency,
log_execution_time = LogExecutionTime,
retry = Retry,
retry_delay = RetryDelay,
time_start = TimeStart},
Dispatcher) ->
MapCount = cloudi_concurrency:count(Concurrency),
ok = callback_info_set(TimeStart),
case MapReduceModule:
cloudi_service_map_reduce_new(MapReduceArgs, MapCount,
Prefix, Timeout, Dispatcher) of
{ok, MapReduceState} ->
case map_send(MapCount, #{},
MapReduceModule, MapReduceState, Dispatcher) of
{ok, MapRequests, MapReduceStateNew} ->
{noreply,
#state{service = Service,
map_reduce_module = MapReduceModule,
map_reduce_state = MapReduceStateNew,
map_reduce_name = MapReduceName,
map_count = MapCount,
log_execution_time = LogExecutionTime,
retry = Retry,
retry_delay = RetryDelay,
map_requests = MapRequests,
time_running = TimeStart}};
{error, _} = Error ->
{stop, Error, undefined}
end;
{error, _} = Error ->
{stop, Error, undefined}
end.
init_end(#init_end{state = State,
error = undefined},
#init_state{service = Service,
info_queued = InfoQueued}) ->
ok = init_end_send(lists:reverse(InfoQueued), Service),
{noreply, State};
init_end(#init_end{state = State,
error = Error},
#init_state{}) ->
{stop, Error, State}.
init_end_send([], _) ->
ok;
init_end_send([Request | InfoQueued], Service) ->
Service ! Request,
init_end_send(InfoQueued, Service).
map_retry(MapRequest, State, Dispatcher) ->
case retry(MapRequest, State) of
true ->
{noreply, State};
false ->
map_resend(MapRequest, State, Dispatcher)
end.
map_resend(#map_send{send_args = [_ | SendArgsTail],
retry_count = RetryCount},
#state{map_reduce_module = MapReduceModule,
map_reduce_state = MapReduceState,
map_requests = MapRequests,
time_running = TimeRunningStart,
elapsed_seconds = ElapsedSeconds} = State, Dispatcher) ->
SendArgs = [Dispatcher | SendArgsTail],
ok = callback_info_set(maps:size(MapRequests),
TimeRunningStart, ElapsedSeconds),
case MapReduceModule:
cloudi_service_map_reduce_resend(SendArgs, MapReduceState) of
{ok, SendArgsNew, MapReduceStateNew} ->
case map_send_request(SendArgsNew, RetryCount, MapRequests) of
{ok, MapRequestsNew} ->
{noreply,
State#state{map_reduce_state = MapReduceStateNew,
map_requests = MapRequestsNew}};
{error, _} = Error ->
{stop, Error, State}
end;
{error, _} = Error ->
{stop, Error, State}
end.
map_info(Request, State, Dispatcher) ->
#state{map_reduce_module = MapReduceModule,
map_reduce_state = MapReduceState,
map_requests = MapRequests,
time_running = TimeRunningStart,
elapsed_seconds = ElapsedSeconds} = State,
ok = callback_info_set(maps:size(MapRequests),
TimeRunningStart, ElapsedSeconds),
case MapReduceModule:
cloudi_service_map_reduce_info(Request, MapReduceState, Dispatcher) of
{ok, MapReduceStateNew} ->
{noreply, State#state{map_reduce_state = MapReduceStateNew}};
{done, MapReduceStateNew} ->
map_check_done(State#state{map_reduce_state = MapReduceStateNew});
{error, _} = Error ->
{stop, Error, State}
end.
map_send(MapRequests, MapReduceModule, MapReduceState, Dispatcher) ->
map_send(1, MapRequests, MapReduceModule, MapReduceState, Dispatcher).
map_send(0, MapRequests, _MapReduceModule, MapReduceState, _Dispatcher) ->
{ok, MapRequests, MapReduceState};
map_send(Count, MapRequests, MapReduceModule, MapReduceState, Dispatcher) ->
case MapReduceModule:
cloudi_service_map_reduce_send(MapReduceState, Dispatcher) of
{ok, SendArgs, MapReduceStateNew} ->
case map_send_request(SendArgs, MapRequests) of
{ok, MapRequestsNew} ->
map_send(Count - 1, MapRequestsNew,
MapReduceModule, MapReduceStateNew, Dispatcher);
{error, _} = Error ->
Error
end;
{done, MapReduceStateNew} ->
{ok, MapRequests, MapReduceStateNew};
{error, _} = Error ->
Error
end.
map_send_request(SendArgs, MapRequests) ->
map_send_request(SendArgs, 0, MapRequests).
map_send_request(SendArgs, RetryCount, MapRequests) ->
case erlang:apply(cloudi_service, send_async_active, SendArgs) of
{ok, TransId} ->
MapRequestsNew = maps:put(TransId,
#map_send{send_args = SendArgs,
retry_count = RetryCount},
MapRequests),
{ok, MapRequestsNew};
{error, _} = Error ->
Error
end.
map_check_continue(#state{map_reduce_module = MapReduceModule,
map_reduce_state = MapReduceState,
map_requests = MapRequests} = State, Dispatcher) ->
case map_send(MapRequests, MapReduceModule, MapReduceState, Dispatcher) of
{ok, MapRequestsNew, MapReduceStateNew} ->
{noreply,
State#state{map_reduce_state = MapReduceStateNew,
map_requests = MapRequestsNew}};
{error, _} = Error ->
{stop, Error, State}
end.
map_check_done(#state{map_requests = MapRequests} = State) ->
case maps:size(MapRequests) of
0 ->
{stop, shutdown, State};
_ ->
{noreply, State}
end.
retry(#map_send{send_args = [_Dispatcher, _Name, _Request,
?TIMEOUT_MAX_ERLANG],
retry_count = RetryCount} = MapRequest,
#state{service = Service,
retry = Retry,
retry_delay = RetryDelay}) ->
if
RetryCount < Retry ->
retry_delay(RetryDelay, Service, MapRequest);
true ->
erlang:exit(retry_max)
end;
retry(#map_send{send_args = [_Dispatcher, _Name, _Request,
?TIMEOUT_MAX_ERLANG, _PatternPid],
retry_count = RetryCount} = MapRequest,
#state{service = Service,
retry = Retry,
retry_delay = RetryDelay}) ->
if
RetryCount < Retry ->
retry_delay(RetryDelay, Service, MapRequest);
true ->
erlang:exit(retry_max)
end;
retry(#map_send{send_args = [_Dispatcher, _Name, _RequestInfo, _Request,
?TIMEOUT_MAX_ERLANG, _Priority],
retry_count = RetryCount} = MapRequest,
#state{service = Service,
retry = Retry,
retry_delay = RetryDelay}) ->
if
RetryCount < Retry ->
retry_delay(RetryDelay, Service, MapRequest);
true ->
erlang:exit(retry_max)
end;
retry(#map_send{send_args = [_Dispatcher, _Name, _RequestInfo, _Request,
?TIMEOUT_MAX_ERLANG, _Priority, _PatternPid],
retry_count = RetryCount} = MapRequest,
#state{service = Service,
retry = Retry,
retry_delay = RetryDelay}) ->
if
RetryCount < Retry ->
retry_delay(RetryDelay, Service, MapRequest);
true ->
erlang:exit(retry_max)
end;
retry(_, _) ->
false.
retry_delay(RetryDelay, Service,
#map_send{retry_count = RetryCount} = MapRequest) ->
MapRequestNew = MapRequest#map_send{retry_count = RetryCount + 1},
_ = erlang:send_after(RetryDelay, Service,
{cloudi_service_map_reduce_retry, MapRequestNew}),
true.
callback_info_set(TimeRunningStart) ->
callback_info_set(0, TimeRunningStart, 0).
callback_info_set(MapRequestsSize, TimeRunningStart, ElapsedSeconds) ->
_ = erlang:put(?CALLBACK_INFO_PDICT_KEY,
#cloudi_service_map_reduce_info{
map_requests_size = MapRequestsSize,
time_running = TimeRunningStart,
elapsed_seconds = ElapsedSeconds}),
ok.
hours_elapsed(ElapsedSeconds) ->
erlang:round((ElapsedSeconds / (60 * 60)) * 10) / 10.