src/smoosh/src/smoosh_channel.erl - couchdb - Git at Google

 % 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, activate/1, get_status/1]).
 -export([enqueue/3, last_updated/2, flush/1, is_key/2, is_activated/1, persist/1]).

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

 -define(VSN, 1).
 -define(CHECKPOINT_INTERVAL_IN_MSEC, 180000).

 -ifndef(TEST).
 -define(START_DELAY_IN_MSEC, 60000).
 -define(ACTIVATE_DELAY_IN_MSEC, 30000).
 -else.
 -define(START_DELAY_IN_MSEC, 0).
 -define(ACTIVATE_DELAY_IN_MSEC, 0).
 -endif.

 % records.

 % When the state is set to activated = true, the channel has completed the state
 % recovery process that occurs on (re)start and is accepting new compaction jobs.
 % Note: if activated = false and a request for a new compaction job is received,
 % smoosh will enqueue this new job after the state recovery process has finished.
 % When the state is set to paused = false, the channel is actively compacting any
 % compaction jobs that are scheduled.
 % See operator_guide.md --> State diagram.

 -record(state, {
     active = [],
     name,
     waiting,
     paused = true,
     starting = [],
     activated = false,
     requests = []
 }).

 % 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_and_activate).

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

 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).

 is_key(ServerRef, Key) ->
     gen_server:call(ServerRef, {is_key, Key}).

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

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

 % gen_server functions.

 init(Name) ->
     erlang:send_after(60 * 1000, self(), check_window),
     process_flag(trap_exit, true),
     Waiting = smoosh_priority_queue:new(Name),
     State = #state{name = Name, waiting = Waiting, paused = true, activated = false},
     erlang:send_after(?START_DELAY_IN_MSEC, self(), start_recovery),
     {ok, State}.

 handle_call({last_updated, Object}, _From, State) ->
     LastUpdated = smoosh_priority_queue:last_updated(Object, State#state.waiting),
     {reply, LastUpdated, State};
 handle_call(suspend, _From, State) ->
     #state{active = Active} = State,
     [
         catch erlang:suspend_process(Pid, [unless_suspending])
      || {_, Pid} <- Active
     ],
     {reply, ok, State#state{paused = true}};
 handle_call(resume_and_activate, _From, State) ->
     #state{active = Active} = State,
     [catch erlang:resume_process(Pid) || {_, Pid} <- Active],
     {reply, ok, State#state{paused = false, activated = true}};
 handle_call(activate, _From, State) ->
     {reply, ok, State#state{activated = true}};
 handle_call(status, _From, State) ->
     {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, State) ->
     {stop, normal, ok, State};
 handle_call(flush, _From, #state{waiting = Q} = State) ->
     {reply, ok, State#state{waiting = smoosh_priority_queue:flush(Q)}};
 handle_call({is_key, Key}, _From, #state{waiting = Waiting} = State) ->
     {reply, smoosh_priority_queue:is_key(Key, Waiting), State};
 handle_call(is_activated, _From, #state{activated = Activated} = State0) ->
     {reply, Activated, State0};
 handle_call(persist, _From, State) ->
     persist_queue(State),
     {reply, ok, State}.

 handle_cast({enqueue, _Object, 0}, #state{} = State) ->
     {noreply, State};
 handle_cast({enqueue, Object, Priority}, #state{activated = true} = State) ->
     {noreply, maybe_start_compaction(add_to_queue(Object, Priority, State))};
 handle_cast({enqueue, Object, Priority}, #state{activated = false, requests = Requests} = State0) ->
     couch_log:notice(
         "~p Channel is not activated yet. Adding ~p to requests with priority ~p.", [
             ?MODULE,
             Object,
             Priority
         ]
     ),
     {noreply, State0#state{requests = [{Object, Priority} | Requests]}}.

 % We accept noproc here due to possibly having monitored a restarted compaction
 % pid after it finished.
 handle_info({'DOWN', Ref, _, Job, Reason}, State) when
     Reason == normal;
     Reason == noproc
 ->
     #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}, State) ->
     #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}}, State) when is_reference(Ref) ->
     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, State) ->
     #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_and_activate, 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(start_recovery, #state{name = Name, waiting = Waiting0} = State0) ->
     RecActive = recover(active_file_name(Name)),
     Waiting1 = lists:foldl(
         fun(DbName, Acc) ->
             case couch_server:exists(DbName) andalso couch_db:is_compacting(DbName) of
                 true ->
                     Priority = smoosh_server:get_priority(Name, DbName),
                     smoosh_priority_queue:in(DbName, Priority, Priority, Acc);
                 false ->
                     Acc
             end
         end,
         Waiting0,
         RecActive
     ),
     State1 = maybe_start_compaction(State0#state{paused = false, waiting = Waiting1}),
     couch_log:notice(
         "~p Previously active compaction jobs (if any) have been successfully recovered and restarted.",
         [?MODULE]
     ),
     erlang:send_after(?ACTIVATE_DELAY_IN_MSEC, self(), activate),
     {noreply, State1#state{paused = true}};
 handle_info(activate, State) ->
     {noreply, activate_channel(State)};
 handle_info(persist, State) ->
     persist_queue(State),
     erlang:send_after(?CHECKPOINT_INTERVAL_IN_MSEC, self(), persist),
     {noreply, State};
 handle_info(pause, State) ->
     {noreply, State#state{paused = true}};
 handle_info(unpause, State) ->
     {noreply, maybe_start_compaction(State#state{paused = false})}.

 terminate(_Reason, _State) ->
     ok.

 persist_queue(State) ->
     write_state_to_file(State).

 recover(FilePath) ->
     case do_recover(FilePath) of
         {ok, List} ->
             List;
         error ->
             []
     end.

 do_recover(FilePath) ->
     case file:read_file(FilePath) of
         {ok, Content} ->
             <<Vsn, Binary/binary>> = Content,
             try parse_state(Vsn, ?VSN, Binary) of
                 Term ->
                     couch_log:notice(
                         "~p Successfully restored state file ~s", [?MODULE, FilePath]
                     ),
                     {ok, Term}
             catch
                 error:Reason ->
                     couch_log:error(
                         "~p Invalid state file (~p). Deleting ~s", [?MODULE, Reason, FilePath]
                     ),
                     file:delete(FilePath),
                     error
             end;
         {error, enoent} ->
             couch_log:notice(
                 "~p (~p) State file ~s does not exist. Not restoring.", [?MODULE, enoent, FilePath]
             ),
             error;
         {error, Reason} ->
             couch_log:error(
                 "~p Cannot read the state file (~p). Deleting ~s", [?MODULE, Reason, FilePath]
             ),
             file:delete(FilePath),
             error
     end.

 parse_state(1, ?VSN, Binary) ->
     erlang:binary_to_term(Binary, [safe]);
 parse_state(Vsn, ?VSN, _) ->
     error({unsupported_version, Vsn}).

 write_state_to_file(#state{name = Name, active = Active, starting = Starting, waiting = Waiting}) ->
     Active1 = lists:foldl(
         fun({DbName, _}, Acc) ->
             [DbName | Acc]
         end,
         [],
         Active
     ),
     Starting1 = lists:foldl(
         fun({_, DbName}, Acc) ->
             [DbName | Acc]
         end,
         [],
         Starting
     ),
     smoosh_utils:write_to_file(Active1, active_file_name(Name), ?VSN),
     smoosh_utils:write_to_file(Starting1, starting_file_name(Name), ?VSN),
     smoosh_priority_queue:write_to_file(Waiting).

 active_file_name(Name) ->
     filename:join(config:get("smoosh", "state_dir", "."), Name ++ ".active").

 starting_file_name(Name) ->
     filename:join(config:get("smoosh", "state_dir", "."), Name ++ ".starting").

 % 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_activate(#state{activated = true} = State) ->
     State;
 maybe_activate(State) ->
     activate_channel(State).

 activate_channel(#state{name = Name, waiting = Waiting0, requests = Requests0} = State0) ->
     RecStarting = recover(starting_file_name(Name)),
     Starting = lists:foldl(
         fun(DbName, Acc) ->
             case couch_server:exists(DbName) of
                 true ->
                     Priority = smoosh_server:get_priority(Name, DbName),
                     smoosh_priority_queue:in(DbName, Priority, Priority, Acc);
                 false ->
                     Acc
             end
         end,
         Waiting0,
         RecStarting
     ),
     Waiting1 = smoosh_priority_queue:recover(Starting),
     Requests1 = lists:reverse(Requests0),
     Waiting2 = lists:foldl(
         fun({DbName, Priority}, Acc) ->
             case couch_server:exists(DbName) of
                 true ->
                     smoosh_priority_queue:in(DbName, Priority, Priority, Acc);
                 false ->
                     Acc
             end
         end,
         Waiting1,
         Requests1
     ),
     State1 = maybe_start_compaction(State0#state{
         waiting = Waiting2, paused = false, activated = true, requests = []
     }),
     handle_info(persist, State1),
     schedule_unpause(),
     State1#state{paused = true}.

 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 ->
                     maybe_activate(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.
	% 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, activate/1, get_status/1]).
	-export([enqueue/3, last_updated/2, flush/1, is_key/2, is_activated/1, persist/1]).

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

	-define(VSN, 1).
	-define(CHECKPOINT_INTERVAL_IN_MSEC, 180000).

	-ifndef(TEST).
	-define(START_DELAY_IN_MSEC, 60000).
	-define(ACTIVATE_DELAY_IN_MSEC, 30000).
	-else.
	-define(START_DELAY_IN_MSEC, 0).
	-define(ACTIVATE_DELAY_IN_MSEC, 0).
	-endif.

	% records.

	% When the state is set to activated = true, the channel has completed the state
	% recovery process that occurs on (re)start and is accepting new compaction jobs.
	% Note: if activated = false and a request for a new compaction job is received,
	% smoosh will enqueue this new job after the state recovery process has finished.
	% When the state is set to paused = false, the channel is actively compacting any
	% compaction jobs that are scheduled.
	% See operator_guide.md --> State diagram.

	-record(state, {
	active = [],
	name,
	waiting,
	paused = true,
	starting = [],
	activated = false,
	requests = []
	}).

	% 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_and_activate).

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

	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).

	is_key(ServerRef, Key) ->
	gen_server:call(ServerRef, {is_key, Key}).

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

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

	% gen_server functions.

	init(Name) ->
	erlang:send_after(60 * 1000, self(), check_window),
	process_flag(trap_exit, true),
	Waiting = smoosh_priority_queue:new(Name),
	State = #state{name = Name, waiting = Waiting, paused = true, activated = false},
	erlang:send_after(?START_DELAY_IN_MSEC, self(), start_recovery),
	{ok, State}.

	handle_call({last_updated, Object}, _From, State) ->
	LastUpdated = smoosh_priority_queue:last_updated(Object, State#state.waiting),
	{reply, LastUpdated, State};
	handle_call(suspend, _From, State) ->
	#state{active = Active} = State,
	[
	catch erlang:suspend_process(Pid, [unless_suspending])
	\|\| {_, Pid} <- Active
	],
	{reply, ok, State#state{paused = true}};
	handle_call(resume_and_activate, _From, State) ->
	#state{active = Active} = State,
	[catch erlang:resume_process(Pid) \|\| {_, Pid} <- Active],
	{reply, ok, State#state{paused = false, activated = true}};
	handle_call(activate, _From, State) ->
	{reply, ok, State#state{activated = true}};
	handle_call(status, _From, State) ->
	{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, State) ->
	{stop, normal, ok, State};
	handle_call(flush, _From, #state{waiting = Q} = State) ->
	{reply, ok, State#state{waiting = smoosh_priority_queue:flush(Q)}};
	handle_call({is_key, Key}, _From, #state{waiting = Waiting} = State) ->
	{reply, smoosh_priority_queue:is_key(Key, Waiting), State};
	handle_call(is_activated, _From, #state{activated = Activated} = State0) ->
	{reply, Activated, State0};
	handle_call(persist, _From, State) ->
	persist_queue(State),
	{reply, ok, State}.

	handle_cast({enqueue, _Object, 0}, #state{} = State) ->
	{noreply, State};
	handle_cast({enqueue, Object, Priority}, #state{activated = true} = State) ->
	{noreply, maybe_start_compaction(add_to_queue(Object, Priority, State))};
	handle_cast({enqueue, Object, Priority}, #state{activated = false, requests = Requests} = State0) ->
	couch_log:notice(
	"~p Channel is not activated yet. Adding ~p to requests with priority ~p.", [
	?MODULE,
	Object,
	Priority
	]
	),
	{noreply, State0#state{requests = [{Object, Priority} \| Requests]}}.

	% We accept noproc here due to possibly having monitored a restarted compaction
	% pid after it finished.
	handle_info({'DOWN', Ref, _, Job, Reason}, State) when
	Reason == normal;
	Reason == noproc
	->
	#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}, State) ->
	#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}}, State) when is_reference(Ref) ->
	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, State) ->
	#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_and_activate, 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(start_recovery, #state{name = Name, waiting = Waiting0} = State0) ->
	RecActive = recover(active_file_name(Name)),
	Waiting1 = lists:foldl(
	fun(DbName, Acc) ->
	case couch_server:exists(DbName) andalso couch_db:is_compacting(DbName) of
	true ->
	Priority = smoosh_server:get_priority(Name, DbName),
	smoosh_priority_queue:in(DbName, Priority, Priority, Acc);
	false ->
	Acc
	end
	end,
	Waiting0,
	RecActive
	),
	State1 = maybe_start_compaction(State0#state{paused = false, waiting = Waiting1}),
	couch_log:notice(
	"~p Previously active compaction jobs (if any) have been successfully recovered and restarted.",
	[?MODULE]
	),
	erlang:send_after(?ACTIVATE_DELAY_IN_MSEC, self(), activate),
	{noreply, State1#state{paused = true}};
	handle_info(activate, State) ->
	{noreply, activate_channel(State)};
	handle_info(persist, State) ->
	persist_queue(State),
	erlang:send_after(?CHECKPOINT_INTERVAL_IN_MSEC, self(), persist),
	{noreply, State};
	handle_info(pause, State) ->
	{noreply, State#state{paused = true}};
	handle_info(unpause, State) ->
	{noreply, maybe_start_compaction(State#state{paused = false})}.

	terminate(_Reason, _State) ->
	ok.

	persist_queue(State) ->
	write_state_to_file(State).

	recover(FilePath) ->
	case do_recover(FilePath) of
	{ok, List} ->
	List;
	error ->
	[]
	end.

	do_recover(FilePath) ->
	case file:read_file(FilePath) of
	{ok, Content} ->
	<<Vsn, Binary/binary>> = Content,
	try parse_state(Vsn, ?VSN, Binary) of
	Term ->
	couch_log:notice(
	"~p Successfully restored state file ~s", [?MODULE, FilePath]
	),
	{ok, Term}
	catch
	error:Reason ->
	couch_log:error(
	"~p Invalid state file (~p). Deleting ~s", [?MODULE, Reason, FilePath]
	),
	file:delete(FilePath),
	error
	end;
	{error, enoent} ->
	couch_log:notice(
	"~p (~p) State file ~s does not exist. Not restoring.", [?MODULE, enoent, FilePath]
	),
	error;
	{error, Reason} ->
	couch_log:error(
	"~p Cannot read the state file (~p). Deleting ~s", [?MODULE, Reason, FilePath]
	),
	file:delete(FilePath),
	error
	end.

	parse_state(1, ?VSN, Binary) ->
	erlang:binary_to_term(Binary, [safe]);
	parse_state(Vsn, ?VSN, _) ->
	error({unsupported_version, Vsn}).

	write_state_to_file(#state{name = Name, active = Active, starting = Starting, waiting = Waiting}) ->
	Active1 = lists:foldl(
	fun({DbName, _}, Acc) ->
	[DbName \| Acc]
	end,
	[],
	Active
	),
	Starting1 = lists:foldl(
	fun({_, DbName}, Acc) ->
	[DbName \| Acc]
	end,
	[],
	Starting
	),
	smoosh_utils:write_to_file(Active1, active_file_name(Name), ?VSN),
	smoosh_utils:write_to_file(Starting1, starting_file_name(Name), ?VSN),
	smoosh_priority_queue:write_to_file(Waiting).

	active_file_name(Name) ->
	filename:join(config:get("smoosh", "state_dir", "."), Name ++ ".active").

	starting_file_name(Name) ->
	filename:join(config:get("smoosh", "state_dir", "."), Name ++ ".starting").

	% 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_activate(#state{activated = true} = State) ->
	State;
	maybe_activate(State) ->
	activate_channel(State).

	activate_channel(#state{name = Name, waiting = Waiting0, requests = Requests0} = State0) ->
	RecStarting = recover(starting_file_name(Name)),
	Starting = lists:foldl(
	fun(DbName, Acc) ->
	case couch_server:exists(DbName) of
	true ->
	Priority = smoosh_server:get_priority(Name, DbName),
	smoosh_priority_queue:in(DbName, Priority, Priority, Acc);
	false ->
	Acc
	end
	end,
	Waiting0,
	RecStarting
	),
	Waiting1 = smoosh_priority_queue:recover(Starting),
	Requests1 = lists:reverse(Requests0),
	Waiting2 = lists:foldl(
	fun({DbName, Priority}, Acc) ->
	case couch_server:exists(DbName) of
	true ->
	smoosh_priority_queue:in(DbName, Priority, Priority, Acc);
	false ->
	Acc
	end
	end,
	Waiting1,
	Requests1
	),
	State1 = maybe_start_compaction(State0#state{
	waiting = Waiting2, paused = false, activated = true, requests = []
	}),
	handle_info(persist, State1),
	schedule_unpause(),
	State1#state{paused = true}.

	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 ->
	maybe_activate(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.