cloudi_service_map_reduce

%-*-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-2020 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-2020 Michael Truog
%%% @version 2.0.1 {@date} {@time}
%%%------------------------------------------------------------------------

-module(cloudi_service_map_reduce).
-author('mjtruog at protonmail dot com').

-behaviour(cloudi_service).

%% external interface
-export([resume/2,
         resume/3,
         suspend/2,
         suspend/3]).

%% 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_MAP_REDUCE_MODULE,     undefined).
-define(DEFAULT_MAP_REDUCE_ARGUMENTS,         []).
-define(DEFAULT_NAME,               "controller").
-define(DEFAULT_CONCURRENCY,                 1.0). % schedulers multiplier
-define(DEFAULT_LOG_EXECUTION_TIME,         true).

-type map_send_args() :: nonempty_list().
-export_type([map_send_args/0]).

-record(state,
    {
        map_reduce_module :: module(),
        map_reduce_state :: any(),
        map_reduce_name :: string(),
        map_count :: pos_integer(),
        log_execution_time :: boolean(),
        map_requests :: #{cloudi_service:trans_id() := map_send_args()},
        time_running :: cloudi_timestamp:seconds_monotonic(),
        suspended = false :: boolean(),
        elapsed_seconds = 0 :: non_neg_integer()
    }).

-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(),
        time_start :: cloudi_timestamp:seconds_monotonic()
    }).

-record(init_end,
    {
        state :: undefined | #state{},
        error = undefined :: any()
    }).

-define(INFO_RETRY_INTERVAL, 500). % milliseconds

%%%------------------------------------------------------------------------
%%% 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
%%%------------------------------------------------------------------------

-type agent() :: cloudi:agent().
-type service_name() :: cloudi:service_name().
-type timeout_milliseconds() :: cloudi:timeout_milliseconds().
-type module_response(Result) ::
    {{ok, Result}, AgentNew :: agent()} |
    {{error, cloudi:error_reason()}, AgentNew :: agent()}.

-spec resume(Agent :: agent(),
             Prefix :: service_name()) ->
    module_response(ok | {error, any()}).

resume(Agent, Prefix) ->
    cloudi:send_sync(Agent, Prefix, resume).

-spec resume(Agent :: agent(),
             Prefix :: service_name(),
             Timeout :: timeout_milliseconds()) ->
    module_response(ok | {error, any()}).

resume(Agent, Prefix, Timeout) ->
    cloudi:send_sync(Agent, Prefix, resume, Timeout).

-spec suspend(Agent :: agent(),
              Prefix :: service_name()) ->
    module_response(ok | {error, any()}).

suspend(Agent, Prefix) ->
    cloudi:send_sync(Agent, Prefix, suspend).

-spec suspend(Agent :: agent(),
              Prefix :: service_name(),
              Timeout :: timeout_milliseconds()) ->
    module_response(ok | {error, any()}).

suspend(Agent, Prefix, Timeout) ->
    cloudi:send_sync(Agent, Prefix, suspend, Timeout).

%%%------------------------------------------------------------------------
%%% 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}],
    [MapReduceModule, MapReduceArgs, Name, Concurrency, LogExecutionTime] =
        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),
    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,
                           time_start = TimeStart},
    cloudi_service:subscribe(Dispatcher, Name),
    {ok, undefined}.

cloudi_service_handle_request(_RequestType, _Name, _Pattern,
                              _RequestInfo, Request,
                              _Timeout, _Priority, _TransId, _Pid,
                              State, Dispatcher) ->
    request(Request, State, Dispatcher).

cloudi_service_handle_info(#init_begin{service = Service} = InitBegin,
                           undefined, Dispatcher) ->
    % cloudi_service_map_reduce_new/5 execution occurs outside of
    % cloudi_service_init/4 to allow send_sync and recv_async function calls
    % because no Erlang process linking/spawning/etc. should be occurring,
    % only algorithmic initialization.  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_link(fun() ->
        case init(InitBegin, Dispatcher) of
            {noreply, State} ->
                Service ! #init_end{state = State},
                ok;
            {stop, Reason, State} when Reason /= undefined ->
                Service ! #init_end{state = State,
                                    error = Reason},
                ok
        end,
        true = erlang:unlink(Service)
    end),
    {noreply, undefined};
cloudi_service_handle_info(#init_end{state = State,
                                     error = Error},
                           undefined, _Dispatcher) ->
    if
        Error =:= undefined ->
            {noreply, State};
        true ->
            {stop, Error, State}
    end;
cloudi_service_handle_info(Request, undefined, Dispatcher) ->
    % waiting for #init_end{} still
    erlang:send_after(?INFO_RETRY_INTERVAL,
                      cloudi_service:self(Dispatcher), Request),
    {noreply, undefined};
cloudi_service_handle_info(#timeout_async_active{trans_id = TransId} = Request,
                           #state{map_reduce_module = MapReduceModule,
                                  map_reduce_state = MapReduceState,
                                  map_requests = MapRequests} = State,
                           Dispatcher) ->
    case maps:find(TransId, MapRequests) of
        {ok, [_ | SendArgs]} ->
            case MapReduceModule:
                 cloudi_service_map_reduce_resend([Dispatcher | SendArgs],
                                                  MapReduceState) of
                {ok, SendArgsNew, MapReduceStateNew} ->
                    case map_send_request(SendArgsNew,
                                          maps:remove(TransId, 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;
        error ->
            cloudi_service_map_reduce_info(Request, State, Dispatcher)
    end;
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} = State,
                           Dispatcher) ->
    case maps:find(TransId, MapRequests) of
        {ok, [_ | SendArgs]} ->
            case MapReduceModule:
                 cloudi_service_map_reduce_recv([Dispatcher | SendArgs],
                                                ResponseInfo, Response,
                                                Timeout, TransId,
                                                MapReduceState, Dispatcher) of
                {ok, MapReduceStateNew} ->
                    MapRequestsNew = maps:remove(TransId, MapRequests),
                    StateNew = State#state{map_reduce_state = MapReduceStateNew,
                                           map_requests = MapRequestsNew},
                    map_check_continue(StateNew, Dispatcher);
                {done, MapReduceStateNew} ->
                    MapRequestsNew = maps:remove(TransId, MapRequests),
                    StateNew = State#state{map_reduce_state = MapReduceStateNew,
                                           map_requests = MapRequestsNew},
                    map_check_done(StateNew);
                {error, _} = Error ->
                    {stop, Error, State}
            end;
        error ->
            cloudi_service_map_reduce_info(Request, State, Dispatcher)
    end;
cloudi_service_handle_info(Request, State, Dispatcher) ->
    cloudi_service_map_reduce_info(Request, State, Dispatcher).

cloudi_service_terminate(_Reason, _Timeout, undefined) ->
    ok;
cloudi_service_terminate(shutdown, _Timeout,
                         #state{log_execution_time = true,
                                map_requests = #{},
                                time_running = TimeRunningStart,
                                elapsed_seconds = ElapsedSeconds}) ->
    TimeRunningEnd = cloudi_timestamp:seconds_monotonic(),
    ElapsedSecondsNew = ElapsedSeconds + (TimeRunningEnd - TimeRunningStart),
    ?LOG_INFO("total time taken was ~p hours",
              [hours_elapsed(ElapsedSecondsNew)]),
    ok;
cloudi_service_terminate(_Reason, _Timeout, #state{}) ->
    ok.

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

init(#init_begin{prefix = Prefix,
                 timeout = Timeout,
                 map_reduce_module = MapReduceModule,
                 map_reduce_args = MapReduceArgs,
                 map_reduce_name = MapReduceName,
                 concurrency = Concurrency,
                 log_execution_time = LogExecutionTime,
                 time_start = TimeStart},
     Dispatcher) ->
    MapCount = cloudi_concurrency:count(Concurrency),
    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{map_reduce_module = MapReduceModule,
                            map_reduce_state = MapReduceStateNew,
                            map_reduce_name = MapReduceName,
                            map_count = MapCount,
                            log_execution_time = LogExecutionTime,
                            map_requests = MapRequests,
                            time_running = TimeStart}};
                {error, _} = Error ->
                    {stop, Error, undefined}
            end;
        {error, _} = Error ->
            {stop, Error, undefined}
    end.

request(_, undefined, _Dispatcher) ->
    {reply, {error, init_pending}, undefined};
request(suspend,
        #state{map_reduce_module = MapReduceModule,
               map_reduce_name = MapReduceName,
               time_running = TimeRunningStart,
               suspended = false,
               elapsed_seconds = ElapsedSeconds} = State, _Dispatcher) ->
    TimeRunningEnd = cloudi_timestamp:seconds_monotonic(),
    ElapsedSecondsNew = ElapsedSeconds + (TimeRunningEnd - TimeRunningStart),
    ?LOG_INFO("~s ~ts suspended", [MapReduceModule, MapReduceName]),
    {reply, ok,
     State#state{suspended = true,
                 elapsed_seconds = ElapsedSecondsNew}};
request(suspend,
        #state{suspended = true} = State, _Dispatcher) ->
    {reply, {error, already_suspended}, State};
request(resume,
        #state{map_reduce_module = MapReduceModule,
               map_reduce_state = MapReduceState,
               map_reduce_name = MapReduceName,
               map_count = MapCount,
               map_requests = MapRequests,
               suspended = true} = State, Dispatcher) ->
    TimeRunningStart = cloudi_timestamp:seconds_monotonic(),
    case map_send(MapCount - maps:size(MapRequests), MapRequests,
                  MapReduceModule, MapReduceState, Dispatcher) of
        {ok, MapRequestsNew, MapReduceStateNew} ->
            ?LOG_INFO("~s ~ts resumed", [MapReduceModule, MapReduceName]),
            {reply, ok,
             State#state{map_reduce_state = MapReduceStateNew,
                         map_requests = MapRequestsNew,
                         time_running = TimeRunningStart,
                         suspended = false}};
        {error, _} = Error ->
            {stop, Error, State}
    end;
request(resume,
        #state{suspended = false} = State, _Dispatcher) ->
    {reply, {error, already_resumed}, State}.

cloudi_service_map_reduce_info(Request, State, Dispatcher) ->
    #state{map_reduce_module = MapReduceModule,
           map_reduce_state = MapReduceState} = State,
    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) ->
    case erlang:apply(cloudi_service, send_async_active, SendArgs) of
        {ok, TransId} ->
            {ok, maps:put(TransId, SendArgs, MapRequests)};
        {error, _} = Error ->
            Error
    end.

map_check_continue(#state{suspended = true} = State, _Dispatcher) ->
    {noreply, State};
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.

hours_elapsed(ElapsedSeconds) ->
    erlang:round((ElapsedSeconds / (60 * 60)) * 10) / 10.