blob: 2b19ccaaea0a1cd450e6b835b311bf6c5f60fe50 [file] [log] [blame]
% 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(couch_index).
-behaviour(gen_server).
-behaviour(config_listener).
-vsn(1).
%% API
-export([start_link/1, stop/1, get_state/2, get_info/1]).
-export([trigger_update/2]).
-export([compact/1, compact/2, get_compactor_pid/1]).
-export([config_change/3]).
%% gen_server callbacks
-export([init/1, terminate/2, code_change/3]).
-export([handle_call/3, handle_cast/2, handle_info/2]).
% config_listener api
-export([handle_config_change/5, handle_config_terminate/3]).
-include_lib("couch/include/couch_db.hrl").
-define(CHECK_INTERVAL, 600000). % 10 minutes
-record(st, {
mod,
idx_state,
updater,
compactor,
waiters=[],
commit_delay,
committed=true,
shutdown=false
}).
start_link({Module0, IdxState0}) ->
[Module, IdxState] = couch_index_plugin:before_open(Module0, IdxState0),
proc_lib:start_link(?MODULE, init, [{Module, IdxState}]).
stop(Pid) ->
gen_server:cast(Pid, stop).
get_state(Pid, RequestSeq) ->
gen_server:call(Pid, {get_state, RequestSeq}, infinity).
get_info(Pid) ->
gen_server:call(Pid, get_info).
trigger_update(Pid, UpdateSeq) ->
gen_server:cast(Pid, {trigger_update, UpdateSeq}).
compact(Pid) ->
compact(Pid, []).
compact(Pid, Options) ->
{ok, CPid} = gen_server:call(Pid, compact),
case lists:member(monitor, Options) of
true -> {ok, erlang:monitor(process, CPid)};
false -> ok
end.
get_compactor_pid(Pid) ->
gen_server:call(Pid, get_compactor_pid).
config_change("query_server_config", "commit_freq", NewValue) ->
ok = gen_server:cast(?MODULE, {config_update, NewValue}).
init({Mod, IdxState}) ->
ok = config:listen_for_changes(?MODULE, nil),
DbName = Mod:get(db_name, IdxState),
erlang:send_after(?CHECK_INTERVAL, self(), maybe_close),
Resp = couch_util:with_db(DbName, fun(Db) ->
case Mod:open(Db, IdxState) of
{ok, IdxSt} ->
couch_db:monitor(Db),
{ok, IdxSt};
Error ->
Error
end
end),
case Resp of
{ok, NewIdxState} ->
{ok, UPid} = couch_index_updater:start_link(self(), Mod),
{ok, CPid} = couch_index_compactor:start_link(self(), Mod),
Delay = config:get("query_server_config", "commit_freq", "5"),
MsDelay = 1000 * list_to_integer(Delay),
State = #st{
mod=Mod,
idx_state=NewIdxState,
updater=UPid,
compactor=CPid,
commit_delay=MsDelay
},
Args = [
Mod:get(db_name, IdxState),
Mod:get(idx_name, IdxState),
couch_index_util:hexsig(Mod:get(signature, IdxState))
],
couch_log:info("Opening index for db: ~s idx: ~s sig: ~p", Args),
proc_lib:init_ack({ok, self()}),
gen_server:enter_loop(?MODULE, [], State);
Other ->
proc_lib:init_ack(Other)
end.
terminate(Reason, State) ->
#st{mod=Mod, idx_state=IdxState}=State,
Mod:close(IdxState),
send_all(State#st.waiters, Reason),
couch_util:shutdown_sync(State#st.updater),
couch_util:shutdown_sync(State#st.compactor),
Args = [
Mod:get(db_name, IdxState),
Mod:get(idx_name, IdxState),
couch_index_util:hexsig(Mod:get(signature, IdxState)),
Reason
],
couch_log:info("Closing index for db: ~s idx: ~s sig: ~p~nreason: ~p", Args),
ok.
handle_call({get_state, ReqSeq}, From, State) ->
#st{
mod=Mod,
idx_state=IdxState,
waiters=Waiters
} = State,
IdxSeq = Mod:get(update_seq, IdxState),
case ReqSeq =< IdxSeq of
true ->
{reply, {ok, IdxState}, State};
_ -> % View update required
couch_index_updater:run(State#st.updater, IdxState),
Waiters2 = [{From, ReqSeq} | Waiters],
{noreply, State#st{waiters=Waiters2}, infinity}
end;
handle_call(get_info, _From, State) ->
#st{mod=Mod} = State,
IdxState = State#st.idx_state,
{ok, Info0} = Mod:get(info, IdxState),
IsUpdating = couch_index_updater:is_running(State#st.updater),
IsCompacting = couch_index_compactor:is_running(State#st.compactor),
IdxSeq = Mod:get(update_seq, IdxState),
GetCommSeq = fun(Db) -> couch_db:get_committed_update_seq(Db) end,
DbName = Mod:get(db_name, IdxState),
CommittedSeq = couch_util:with_db(DbName, GetCommSeq),
Info = Info0 ++ [
{updater_running, IsUpdating},
{compact_running, IsCompacting},
{waiting_commit, State#st.committed == false},
{waiting_clients, length(State#st.waiters)},
{pending_updates, max(CommittedSeq - IdxSeq, 0)}
],
{reply, {ok, Info}, State};
handle_call(reset, _From, State) ->
#st{
mod=Mod,
idx_state=IdxState
} = State,
{ok, NewIdxState} = Mod:reset(IdxState),
{reply, {ok, NewIdxState}, State#st{idx_state=NewIdxState}};
handle_call(compact, _From, State) ->
Resp = couch_index_compactor:run(State#st.compactor, State#st.idx_state),
{reply, Resp, State};
handle_call(get_compactor_pid, _From, State) ->
{reply, {ok, State#st.compactor}, State};
handle_call({compacted, NewIdxState}, _From, State) ->
#st{
mod=Mod,
idx_state=OldIdxState,
updater=Updater,
commit_delay=Delay
} = State,
assert_signature_match(Mod, OldIdxState, NewIdxState),
NewSeq = Mod:get(update_seq, NewIdxState),
OldSeq = Mod:get(update_seq, OldIdxState),
% For indices that require swapping files, we have to make sure we're
% up to date with the current index. Otherwise indexes could roll back
% (perhaps considerably) to previous points in history.
case NewSeq >= OldSeq of
true ->
{ok, NewIdxState1} = Mod:swap_compacted(OldIdxState, NewIdxState),
% Restart the indexer if it's running.
case couch_index_updater:is_running(Updater) of
true -> ok = couch_index_updater:restart(Updater, NewIdxState1);
false -> ok
end,
case State#st.committed of
true -> erlang:send_after(Delay, self(), commit);
false -> ok
end,
{reply, ok, State#st{
idx_state=NewIdxState1,
committed=false
}};
_ ->
{reply, recompact, State}
end.
handle_cast({config_change, NewDelay}, State) ->
MsDelay = 1000 * list_to_integer(NewDelay),
{noreply, State#st{commit_delay=MsDelay}};
handle_cast({trigger_update, UpdateSeq}, State) ->
#st{
mod=Mod,
idx_state=IdxState
} = State,
case UpdateSeq =< Mod:get(update_seq, IdxState) of
true ->
{noreply, State};
false ->
couch_index_updater:run(State#st.updater, IdxState),
{noreply, State}
end;
handle_cast({updated, NewIdxState}, State) ->
{noreply, NewState} = handle_cast({new_state, NewIdxState}, State),
case NewState#st.shutdown andalso (NewState#st.waiters =:= []) of
true ->
{stop, normal, NewState};
false ->
maybe_restart_updater(NewState),
{noreply, NewState}
end;
handle_cast({new_state, NewIdxState}, State) ->
#st{
mod=Mod,
idx_state=OldIdxState,
commit_delay=Delay
} = State,
assert_signature_match(Mod, OldIdxState, NewIdxState),
CurrSeq = Mod:get(update_seq, NewIdxState),
Args = [
Mod:get(db_name, NewIdxState),
Mod:get(idx_name, NewIdxState),
CurrSeq
],
couch_log:debug("Updated index for db: ~s idx: ~s seq: ~B", Args),
Rest = send_replies(State#st.waiters, CurrSeq, NewIdxState),
case State#st.committed of
true -> erlang:send_after(Delay, self(), commit);
false -> ok
end,
{noreply, State#st{
idx_state=NewIdxState,
waiters=Rest,
committed=false
}};
handle_cast({update_error, Error}, State) ->
send_all(State#st.waiters, Error),
{noreply, State#st{waiters=[]}};
handle_cast(stop, State) ->
{stop, normal, State};
handle_cast(delete, State) ->
#st{mod=Mod, idx_state=IdxState} = State,
ok = Mod:delete(IdxState),
{stop, normal, State};
handle_cast(ddoc_updated, State) ->
#st{mod = Mod, idx_state = IdxState, waiters = Waiters} = State,
DbName = Mod:get(db_name, IdxState),
DDocId = Mod:get(idx_name, IdxState),
Shutdown = couch_util:with_db(DbName, fun(Db) ->
case couch_db:open_doc(Db, DDocId, [ejson_body, ?ADMIN_CTX]) of
{not_found, deleted} ->
true;
{ok, DDoc} ->
{ok, NewIdxState} = Mod:init(Db, DDoc),
Mod:get(signature, NewIdxState) =/= Mod:get(signature, IdxState)
end
end),
case Shutdown of
true ->
case Waiters of
[] ->
{stop, normal, State};
_ ->
{noreply, State#st{shutdown = true}}
end;
false ->
{noreply, State#st{shutdown = false}}
end;
handle_cast(_Mesg, State) ->
{stop, unhandled_cast, State}.
handle_info(commit, #st{committed=true}=State) ->
{noreply, State};
handle_info(commit, State) ->
#st{mod=Mod, idx_state=IdxState, commit_delay=Delay} = State,
DbName = Mod:get(db_name, IdxState),
IdxName = Mod:get(idx_name, IdxState),
GetCommSeq = fun(Db) -> couch_db:get_committed_update_seq(Db) end,
CommittedSeq = couch_util:with_db(DbName, GetCommSeq),
case CommittedSeq >= Mod:get(update_seq, IdxState) of
true ->
% Commit the updates
ok = Mod:commit(IdxState),
couch_event:notify(DbName, {index_commit, IdxName}),
{noreply, State#st{committed=true}};
_ ->
% We can't commit the header because the database seq that's
% fully committed to disk is still behind us. If we committed
% now and the database lost those changes our view could be
% forever out of sync with the database. But a crash before we
% commit these changes, no big deal, we only lose incremental
% changes since last committal.
erlang:send_after(Delay, self(), commit),
{noreply, State}
end;
handle_info(maybe_close, State) ->
% We need to periodically check if our index file still
% exists on disk because index cleanups don't notify
% the couch_index process when a file has been deleted. If
% we don't check for this condition then the index can
% remain open indefinitely wasting disk space.
%
% We make sure that we're idle before closing by looking
% to see if we have any clients waiting for an update.
Mod = State#st.mod,
case State#st.waiters of
[] ->
case Mod:index_file_exists(State#st.idx_state) of
true ->
erlang:send_after(?CHECK_INTERVAL, self(), maybe_close),
{noreply, State};
false ->
{stop, normal, State}
end;
_ ->
erlang:send_after(?CHECK_INTERVAL, self, maybe_close),
{noreply, State}
end;
handle_info({'DOWN', _, _, _Pid, _}, #st{mod=Mod, idx_state=IdxState}=State) ->
Args = [Mod:get(db_name, IdxState), Mod:get(idx_name, IdxState)],
couch_log:info("Index shutdown by monitor notice for db: ~s idx: ~s", Args),
catch send_all(State#st.waiters, shutdown),
{stop, normal, State#st{waiters=[]}}.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
handle_config_change("query_server_config", "commit_freq", Val, _, _) ->
{ok, gen_server:cast(?MODULE, {config_update, Val})};
handle_config_change(_, _, _, _, _) ->
{ok, nil}.
handle_config_terminate(_Server, stop, _State) -> ok;
handle_config_terminate(_Server, _Reason, _State) ->
spawn(fun() ->
timer:sleep(5000),
config:listen_for_changes(?MODULE, nil)
end).
maybe_restart_updater(#st{waiters=[]}) ->
ok;
maybe_restart_updater(#st{mod=Mod, idx_state=IdxState}=State) ->
couch_util:with_db(Mod:get(db_name, IdxState), fun(Db) ->
UpdateSeq = couch_db:get_update_seq(Db),
CommittedSeq = couch_db:get_committed_update_seq(Db),
CanUpdate = UpdateSeq > CommittedSeq,
UOpts = Mod:get(update_options, IdxState),
case CanUpdate and lists:member(committed_only, UOpts) of
true -> couch_db:ensure_full_commit(Db);
false -> ok
end
end),
couch_index_updater:run(State#st.updater, IdxState).
send_all(Waiters, Reply) ->
[gen_server:reply(From, Reply) || {From, _} <- Waiters].
send_replies(Waiters, UpdateSeq, IdxState) ->
Pred = fun({_, S}) -> S =< UpdateSeq end,
{ToSend, Remaining} = lists:partition(Pred, Waiters),
[gen_server:reply(From, {ok, IdxState}) || {From, _} <- ToSend],
Remaining.
assert_signature_match(Mod, OldIdxState, NewIdxState) ->
case {Mod:get(signature, OldIdxState), Mod:get(signature, NewIdxState)} of
{Sig, Sig} -> ok;
_ -> erlang:error(signature_mismatch)
end.