<commit_before>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2007-2010 Basho Technologies, Inc.  All Rights Reserved.
%%
%% -------------------------------------------------------------------

-module(riak_search_index_fsm).
-behaviour(gen_fsm).

%% API
-export([start_link/0]).
-export([index_terms/2, index_terms/3, done/1]).

%% gen_fsm callbacks
-export([init/1, handle_event/3,
         handle_sync_event/4, handle_info/3, terminate/3, code_change/4]).
-export([idle/2, idle/3,
         receiving/2, receiving/3]).

-define(RECV_TIMEOUT_USECS, 5000).
-define(INDEX_TIMEOUT_USECS, 5000).

-record(state, {terms=[],          % Terms left to index,
                num_terms=0,       % Number of terms
                ref,               % Reference for current batch of terms sent out
                replies_left,      % Count of replies pending
                timer,             % Timer reference for batch
                overload_waiters=[], % List to reply to once out of overload
                done_waiters=[],   % List to reply to once terms is empty
                batch_size,        % Size of index batches to send
                overload_thresh}). % Threshold for num_terms to trigger overload


-type index() :: term().
-type field() :: term().
-type idxterm() :: term().
-type idxvalue() :: term().
-type props() :: list().
-type index_term() :: {index(),field(),idxterm(),idxvalue(),props()}.
-type index_terms() :: [index_term()].
-type overload_state() :: ok | overload.
-type statename() :: idle | receiving.
-type event() :: {reference(), {indexed, node()}} |
                 {reference(), recv_timeout} |
                 empty.
-type sync_event() :: {index, index_terms()} |
                      done.


%%%===================================================================
%%% API
%%%===================================================================

-spec start_link() -> {ok, pid()} | {error,term()}.
start_link() ->
    gen_fsm:start_link(?MODULE, [], []).

%% Index the list of index/field/term/value/props
-spec index_terms(pid(), index_terms()) -> ok.
index_terms(Pid, Terms) ->
    index_terms(Pid, Terms, ?INDEX_TIMEOUT_USECS).

-spec index_terms(pid(), index_terms() ,timeout()) -> ok.
index_terms(Pid, Terms, Timeout) ->
    gen_fsm:sync_send_event(Pid, {index, Terms}, Timeout).

%% Signal we are done and wait for the FSM to complete and exit
-spec done(pid()) -> ok.
done(Pid) ->
    gen_fsm:sync_send_event(Pid, done, infinity).

%%%===================================================================
%%% gen_fsm callbacks
%%%===================================================================

%% @private
init([]) ->
    BatchSize = app_helper:get_env(riak_search, index_batch_size, 20),
    OverloadThresh = app_helper:get_env(riak_search, index_overload_thresh, 100),
    {ok, idle, #state{batch_size=BatchSize,overload_thresh=OverloadThresh}}.

%% @private
%% Handle gen_fsm:send_event
-spec idle(event(),#state{}) -> {next_state, statename(), #state{}} | {stop, term(), #state{}}.
idle({_Ref, {indexed, _Partition}}, State) ->
    %% Ignore stale reply from previous batch
    {next_state, idle, State};

%% send_batch/1 ran out of terms to send - decide if caller is waiting
%% for us to exit.
idle(empty, State) ->
    case State#state.done_waiters of
        [] ->
            {next_state, idle, State};
        Waiters ->
            lists:foreach(fun(From) -> gen_fsm:reply(From, ok) end, Waiters),
            {stop, normal, State}
    end.

%% @private
%% Handle sync_send_event
-spec idle(sync_event(),term(),#state{}) -> {reply, term(), statename(), #state{}} |
                                            {next_state, statename(), #state{}} |
                                            {stop, term(), term(), #state{}}.
idle({index, Terms}, From, #state{terms = []} = State) ->
    {Status, State1} = add_terms(Terms, From, State),
    {StateName, NewState} = send_batch(State1),
    case Status of
        ok ->
            {reply, ok, StateName, NewState};
        overload ->
            {next_state, StateName, NewState}
    end;
idle(done, _From, #state{terms = []} = State) ->
    {stop, normal, ok, State}.

%% @private
%% handle gen_fsm:send_event
-spec receiving(event(),#state{}) -> {next_state, statename(), #state{}}.
receiving({Ref, {indexed, _Node}}, #state{ref = Ref}=State) ->
    %% Reply for the current batch
    case State#state.replies_left - 1 of
        0 ->
            cancel_recv_timeout(State#state.timer),
            {StateName, NewState} = send_batch(State),
            {next_state, StateName, NewState};
        RepliesLeft ->
            {next_state, receiving, State#state{replies_left = RepliesLeft}}
    end;
receiving({Ref, recv_timeout}, #state{ref=Ref} = State) ->    
    %% Receive timeout for current batch - for now just continue
    %% on to the next batch
    {StateName, NewState} = send_batch(State#state{timer=undefined}),
    {next_state, StateName, NewState};

receiving({_Ref, {indexed, _Partition}}, State) ->
    %% Ignore stale reply from previous batch
    {next_state, waiting, State};
receiving(empty, State) ->
    %% Got given some more work after the terms list was empty - ignore
    {next_state, receiving, State}.

%% @private
%% Handle sync_send_event
-spec receiving(sync_event(),term(),#state{}) -> {reply, term(), statename(), #state{}} |
                                                 {next_state, statename(), #state{}}.
receiving({index, Terms}, From, State) ->
    case add_terms(Terms, From, State) of
        {ok, NewState} ->
            {reply, ok, receiving, NewState};
        {overload, NewState} -><commit_msg>There is no 'waiting' state in this FSM; per Jon, making it 'receiving'<commit_after>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2007-2010 Basho Technologies, Inc.  All Rights Reserved.
%%
%% -------------------------------------------------------------------

-module(riak_search_index_fsm).
-behaviour(gen_fsm).

%% API
-export([start_link/0]).
-export([index_terms/2, index_terms/3, done/1]).

%% gen_fsm callbacks
-export([init/1, handle_event/3,
         handle_sync_event/4, handle_info/3, terminate/3, code_change/4]).
-export([idle/2, idle/3,
         receiving/2, receiving/3]).

-define(RECV_TIMEOUT_USECS, 5000).
-define(INDEX_TIMEOUT_USECS, 5000).

-record(state, {terms=[],          % Terms left to index,
                num_terms=0,       % Number of terms
                ref,               % Reference for current batch of terms sent out
                replies_left,      % Count of replies pending
                timer,             % Timer reference for batch
                overload_waiters=[], % List to reply to once out of overload
                done_waiters=[],   % List to reply to once terms is empty
                batch_size,        % Size of index batches to send
                overload_thresh}). % Threshold for num_terms to trigger overload


-type index() :: term().
-type field() :: term().
-type idxterm() :: term().
-type idxvalue() :: term().
-type props() :: list().
-type index_term() :: {index(),field(),idxterm(),idxvalue(),props()}.
-type index_terms() :: [index_term()].
-type overload_state() :: ok | overload.
-type statename() :: idle | receiving.
-type event() :: {reference(), {indexed, node()}} |
                 {reference(), recv_timeout} |
                 empty.
-type sync_event() :: {index, index_terms()} |
                      done.


%%%===================================================================
%%% API
%%%===================================================================

-spec start_link() -> {ok, pid()} | {error,term()}.
start_link() ->
    gen_fsm:start_link(?MODULE, [], []).

%% Index the list of index/field/term/value/props
-spec index_terms(pid(), index_terms()) -> ok.
index_terms(Pid, Terms) ->
    index_terms(Pid, Terms, ?INDEX_TIMEOUT_USECS).

-spec index_terms(pid(), index_terms() ,timeout()) -> ok.
index_terms(Pid, Terms, Timeout) ->
    gen_fsm:sync_send_event(Pid, {index, Terms}, Timeout).

%% Signal we are done and wait for the FSM to complete and exit
-spec done(pid()) -> ok.
done(Pid) ->
    gen_fsm:sync_send_event(Pid, done, infinity).

%%%===================================================================
%%% gen_fsm callbacks
%%%===================================================================

%% @private
init([]) ->
    BatchSize = app_helper:get_env(riak_search, index_batch_size, 20),
    OverloadThresh = app_helper:get_env(riak_search, index_overload_thresh, 100),
    {ok, idle, #state{batch_size=BatchSize,overload_thresh=OverloadThresh}}.

%% @private
%% Handle gen_fsm:send_event
-spec idle(event(),#state{}) -> {next_state, statename(), #state{}} | {stop, term(), #state{}}.
idle({_Ref, {indexed, _Partition}}, State) ->
    %% Ignore stale reply from previous batch
    {next_state, idle, State};

%% send_batch/1 ran out of terms to send - decide if caller is waiting
%% for us to exit.
idle(empty, State) ->
    case State#state.done_waiters of
        [] ->
            {next_state, idle, State};
        Waiters ->
            lists:foreach(fun(From) -> gen_fsm:reply(From, ok) end, Waiters),
            {stop, normal, State}
    end.

%% @private
%% Handle sync_send_event
-spec idle(sync_event(),term(),#state{}) -> {reply, term(), statename(), #state{}} |
                                            {next_state, statename(), #state{}} |
                                            {stop, term(), term(), #state{}}.
idle({index, Terms}, From, #state{terms = []} = State) ->
    {Status, State1} = add_terms(Terms, From, State),
    {StateName, NewState} = send_batch(State1),
    case Status of
        ok ->
            {reply, ok, StateName, NewState};
        overload ->
            {next_state, StateName, NewState}
    end;
idle(done, _From, #state{terms = []} = State) ->
    {stop, normal, ok, State}.

%% @private
%% handle gen_fsm:send_event
-spec receiving(event(),#state{}) -> {next_state, statename(), #state{}}.
receiving({Ref, {indexed, _Node}}, #state{ref = Ref}=State) ->
    %% Reply for the current batch
    case State#state.replies_left - 1 of
        0 ->
            cancel_recv_timeout(State#state.timer),
            {StateName, NewState} = send_batch(State),
            {next_state, StateName, NewState};
        RepliesLeft ->
            {next_state, receiving, State#state{replies_left = RepliesLeft}}
    end;
receiving({Ref, recv_timeout}, #state{ref=Ref} = State) ->    
    %% Receive timeout for current batch - for now just continue
    %% on to the next batch
    {StateName, NewState} = send_batch(State#state{timer=undefined}),
    {next_state, StateName, NewState};

receiving({_Ref, {indexed, _Partition}}, State) ->
    %% Ignore stale reply from previous batch
    {next_state, receiving, State};
receiving(empty, State) ->
    %% Got given some more work after the terms list was empty - ignore
    {next_state, receiving, State}.

%% @private
%% Handle sync_send_event
-spec receiving(sync_event(),term(),#state{}) -> {reply, term(), statename(), #state{}} |
                                                 {next_state, statename(), #state{}}.
receiving({index, Terms}, From, State) ->
    case add_terms(Terms, From, State) of
        {ok, NewState} ->
            {reply, ok, receiving, NewState};
        {overload, NewState} ->
{'fraction_non_alphanumeric': 0.21714086227962995, 'fraction_numerical': 0.007156571827544074, 'mean_word_length': 2.7484461890742558, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5003417', 'n_tokens_mistral': 3521, 'n_tokens_neox': 3359, 'n_words': 876}