src/smoosh/src/smoosh_channel.erl

% Licensed under the Apache License, Version 2.0 (the "License"); you may not
% use this file except in compliance with the License. You may obtain a copy of
% the License at
%
%   http://www.apache.org/licenses/LICENSE-2.0
%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
% License for the specific language governing permissions and limitations under
% the License.

-module(smoosh_channel).
-behaviour(gen_server).
-vsn(1).
-include_lib("couch/include/couch_db.hrl").

% public api.
-export([start_link/1, close/1, suspend/1, resume/1, get_status/1]).
-export([enqueue/3, last_updated/2, flush/1]).

% gen_server api.
-export([
    init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    code_change/3,
    terminate/2
]).

% records.

-record(state, {
    active = [],
    name,
    waiting = smoosh_priority_queue:new(),
    paused = true,
    starting = []
}).

% public functions.

start_link(Name) ->
    gen_server:start_link(?MODULE, Name, []).

suspend(ServerRef) ->
    gen_server:call(ServerRef, suspend).

resume(ServerRef) ->
    gen_server:call(ServerRef, resume).

enqueue(ServerRef, Object, Priority) ->
    gen_server:cast(ServerRef, {enqueue, Object, Priority}).

last_updated(ServerRef, Object) ->
    gen_server:call(ServerRef, {last_updated, Object}).

get_status(ServerRef) ->
    gen_server:call(ServerRef, status).

close(ServerRef) ->
    gen_server:call(ServerRef, close).

flush(ServerRef) ->
    gen_server:call(ServerRef, flush).

% gen_server functions.

init(Name) ->
    schedule_unpause(),
    erlang:send_after(60 * 1000, self(), check_window),
    {ok, #state{name = Name}}.

handle_call({last_updated, Object}, _From, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    LastUpdated = smoosh_priority_queue:last_updated(Object, State#state.waiting),
    {reply, LastUpdated, State};
handle_call(suspend, _From, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    #state{active = Active} = State,
    [
        catch erlang:suspend_process(Pid, [unless_suspending])
     || {_, Pid} <- Active
    ],
    {reply, ok, State#state{paused = true}};
handle_call(resume, _From, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    #state{active = Active} = State,
    [catch erlang:resume_process(Pid) || {_, Pid} <- Active],
    {reply, ok, State#state{paused = false}};
handle_call(status, _From, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    {reply,
        {ok, [
            {active, length(State#state.active)},
            {starting, length(State#state.starting)},
            {waiting, smoosh_priority_queue:info(State#state.waiting)}
        ]},
        State};
handle_call(close, _From, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    {stop, normal, ok, State};
handle_call(flush, _From, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    {reply, ok, State#state{waiting = smoosh_priority_queue:new()}}.

handle_cast({enqueue, _Object, 0}, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    {noreply, State};
handle_cast({enqueue, Object, Priority}, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    {noreply, maybe_start_compaction(add_to_queue(Object, Priority, State))}.

% We accept noproc here due to possibly having monitored a restarted compaction
% pid after it finished.
handle_info({'DOWN', Ref, _, Job, Reason}, State0) when
    Reason == normal;
    Reason == noproc
->
    {ok, State} = code_change(nil, State0, nil),
    #state{active = Active, starting = Starting} = State,
    {noreply,
        maybe_start_compaction(
            State#state{
                active = lists:keydelete(Job, 2, Active),
                starting = lists:keydelete(Ref, 1, Starting)
            }
        )};
handle_info({'DOWN', Ref, _, Job, Reason}, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    #state{active = Active0, starting = Starting0} = State,
    case lists:keytake(Job, 2, Active0) of
        {value, {Key, _Pid}, Active1} ->
            State1 = maybe_remonitor_cpid(
                State#state{active = Active1},
                Key,
                Reason
            ),
            {noreply, maybe_start_compaction(State1)};
        false ->
            case lists:keytake(Ref, 1, Starting0) of
                {value, {_, Key}, Starting1} ->
                    couch_log:warning("failed to start compaction of ~p: ~p", [
                        smoosh_utils:stringify(Key), Reason
                    ]),
                    {ok, _} = timer:apply_after(5000, smoosh_server, enqueue, [Key]),
                    {noreply, maybe_start_compaction(State#state{starting = Starting1})};
                false ->
                    {noreply, State}
            end
    end;
handle_info({Ref, {ok, Pid}}, State0) when is_reference(Ref) ->
    {ok, State} = code_change(nil, State0, nil),
    case lists:keytake(Ref, 1, State#state.starting) of
        {value, {_, Key}, Starting1} ->
            couch_log:notice(
                "~s: Started compaction for ~s",
                [State#state.name, smoosh_utils:stringify(Key)]
            ),
            erlang:monitor(process, Pid),
            erlang:demonitor(Ref, [flush]),
            {noreply, State#state{
                active = [{Key, Pid} | State#state.active],
                starting = Starting1
            }};
        false ->
            {noreply, State}
    end;
handle_info(check_window, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    #state{paused = Paused, name = Name} = State,
    StrictWindow = smoosh_utils:get(Name, "strict_window", "false"),
    FinalState =
        case {not Paused, smoosh_utils:in_allowed_window(Name)} of
            {false, false} ->
                % already in desired state
                State;
            {true, true} ->
                % already in desired state
                State;
            {false, true} ->
                % resume is always safe even if we did not previously suspend
                {reply, ok, NewState} = handle_call(resume, nil, State),
                NewState;
            {true, false} ->
                if
                    StrictWindow =:= "true" ->
                        {reply, ok, NewState} = handle_call(suspend, nil, State),
                        NewState;
                    true ->
                        State#state{paused = true}
                end
        end,
    erlang:send_after(60 * 1000, self(), check_window),
    {noreply, FinalState};
handle_info(pause, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    {noreply, State#state{paused = true}};
handle_info(unpause, State0) ->
    {ok, State} = code_change(nil, State0, nil),
    {noreply, maybe_start_compaction(State#state{paused = false})}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, #state{} = State, _Extra) ->
    {ok, State}.

% private functions.

add_to_queue(Key, Priority, State) ->
    #state{active = Active, waiting = Q} = State,
    case lists:keymember(Key, 1, Active) of
        true ->
            State;
        false ->
            Capacity = list_to_integer(smoosh_utils:get(State#state.name, "capacity", "9999")),
            couch_log:notice(
                "~s: adding ~p to internal compactor queue with priority ~p",
                [State#state.name, Key, Priority]
            ),
            State#state{
                waiting = smoosh_priority_queue:in(Key, Priority, Priority, Capacity, Q)
            }
    end.

maybe_start_compaction(#state{paused = true} = State) ->
    State;
maybe_start_compaction(State) ->
    Concurrency = list_to_integer(
        smoosh_utils:get(
            State#state.name,
            "concurrency",
            "1"
        )
    ),
    if
        length(State#state.active) + length(State#state.starting) < Concurrency ->
            case smoosh_priority_queue:out(State#state.waiting) of
                false ->
                    State;
                {Key, Priority, Q} ->
                    try
                        State2 =
                            case start_compact(State, Key) of
                                false ->
                                    State;
                                State1 ->
                                    couch_log:notice(
                                        "~s: Starting compaction for ~s (priority ~p)",
                                        [State#state.name, smoosh_utils:stringify(Key), Priority]
                                    ),
                                    State1
                            end,
                        maybe_start_compaction(State2#state{waiting = Q})
                    catch
                        Class:Exception ->
                            couch_log:notice(
                                "~s: ~p ~p for ~s",
                                [
                                    State#state.name,
                                    Class,
                                    Exception,
                                    smoosh_utils:stringify(Key)
                                ]
                            ),
                            maybe_start_compaction(State#state{waiting = Q})
                    end
            end;
        true ->
            State
    end.

start_compact(State, {schema, DbName, GroupId}) ->
    case smoosh_utils:ignore_db({DbName, GroupId}) of
        false ->
            {ok, Pid} = couch_md_index_manager:get_group_pid(
                DbName,
                GroupId
            ),
            Ref = erlang:monitor(process, Pid),
            Pid ! {'$gen_call', {self(), Ref}, compact},
            State#state{starting = [{Ref, {schema, DbName, GroupId}} | State#state.starting]};
        _ ->
            false
    end;
start_compact(State, DbName) when is_list(DbName) ->
    start_compact(State, ?l2b(DbName));
start_compact(State, DbName) when is_binary(DbName) ->
    {ok, Db} = couch_db:open_int(DbName, []),
    try
        start_compact(State, Db)
    after
        couch_db:close(Db)
    end;
start_compact(State, {Shard, GroupId}) ->
    case smoosh_utils:ignore_db({Shard, GroupId}) of
        false ->
            DbName = mem3:dbname(Shard),
            {ok, Pid} = couch_index_server:get_index(
                couch_mrview_index, Shard, GroupId
            ),
            spawn(fun() -> cleanup_index_files(DbName, Shard) end),
            Ref = erlang:monitor(process, Pid),
            Pid ! {'$gen_call', {self(), Ref}, compact},
            State#state{starting = [{Ref, {Shard, GroupId}} | State#state.starting]};
        _ ->
            false
    end;
start_compact(State, Db) ->
    case smoosh_utils:ignore_db(Db) of
        false ->
            DbPid = couch_db:get_pid(Db),
            Key = couch_db:name(Db),
            case couch_db:get_compactor_pid(Db) of
                nil ->
                    Ref = erlang:monitor(process, DbPid),
                    DbPid ! {'$gen_call', {self(), Ref}, start_compact},
                    State#state{starting = [{Ref, Key} | State#state.starting]};
                % Compaction is already running, so monitor existing compaction pid.
                CPid ->
                    couch_log:notice(
                        "Db ~s continuing compaction",
                        [smoosh_utils:stringify(Key)]
                    ),
                    erlang:monitor(process, CPid),
                    State#state{active = [{Key, CPid} | State#state.active]}
            end;
        _ ->
            false
    end.

maybe_remonitor_cpid(State, DbName, Reason) when is_binary(DbName) ->
    {ok, Db} = couch_db:open_int(DbName, []),
    case couch_db:get_compactor_pid_sync(Db) of
        nil ->
            couch_log:warning(
                "exit for compaction of ~p: ~p",
                [smoosh_utils:stringify(DbName), Reason]
            ),
            {ok, _} = timer:apply_after(5000, smoosh_server, enqueue, [DbName]),
            State;
        CPid ->
            couch_log:notice(
                "~s compaction already running. Re-monitor Pid ~p",
                [smoosh_utils:stringify(DbName), CPid]
            ),
            erlang:monitor(process, CPid),
            State#state{active = [{DbName, CPid} | State#state.active]}
    end;
% not a database compaction, so ignore the pid check
maybe_remonitor_cpid(State, Key, Reason) ->
    couch_log:warning(
        "exit for compaction of ~p: ~p",
        [smoosh_utils:stringify(Key), Reason]
    ),
    {ok, _} = timer:apply_after(5000, smoosh_server, enqueue, [Key]),
    State.

schedule_unpause() ->
    WaitSecs = list_to_integer(config:get("smoosh", "wait_secs", "30")),
    erlang:send_after(WaitSecs * 1000, self(), unpause).

cleanup_index_files(DbName, _Shard) ->
    case config:get("smoosh", "cleanup_index_files", "false") of
        "true" ->
            fabric:cleanup_index_files(DbName);
        _ ->
            ok
    end.