You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by fd...@apache.org on 2011/11/15 14:00:52 UTC
[2/3] git commit: Shutdown view group on ddoc update
Shutdown view group on ddoc update
If a design document is updated (or deleted), its associated
view groups that are open will be shutdown as soon as they
finish serving all their requests. Not doing this prevented
a proper cleanup of outdated view files (file descriptor leaks)
and unncessary processes in the system (old view groups and index
file ref counters, for e.g.).
COUCHDB-1309
Project: http://git-wip-us.apache.org/repos/asf/couchdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb/commit/ee942109
Tree: http://git-wip-us.apache.org/repos/asf/couchdb/tree/ee942109
Diff: http://git-wip-us.apache.org/repos/asf/couchdb/diff/ee942109
Branch: refs/heads/master
Commit: ee942109e39d88947e82672c526d8277f9824b65
Parents: 34516a8
Author: Filipe David Borba Manana <fd...@apache.org>
Authored: Sun Nov 13 18:22:42 2011 +0000
Committer: Filipe David Borba Manana <fd...@apache.org>
Committed: Tue Nov 15 13:00:11 2011 +0000
----------------------------------------------------------------------
src/couch_index/src/couch_index.erl | 36 +++++++++-
src/couch_index/src/couch_index_server.erl | 45 ++++++++----
test/etap/200-view-group-no-db-leaks.t | 84 +++++++++++++++++++----
3 files changed, 134 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb/blob/ee942109/src/couch_index/src/couch_index.erl
----------------------------------------------------------------------
diff --git a/src/couch_index/src/couch_index.erl b/src/couch_index/src/couch_index.erl
index 3b27b0f..5086048 100644
--- a/src/couch_index/src/couch_index.erl
+++ b/src/couch_index/src/couch_index.erl
@@ -34,7 +34,8 @@
compactor,
waiters=[],
commit_delay,
- committed=true
+ committed=true,
+ shutdown=false
}).
@@ -201,8 +202,13 @@ handle_cast({config_change, NewDelay}, State) ->
{noreply, State#st{commit_delay=MsDelay}};
handle_cast({updated, NewIdxState}, State) ->
{noreply, NewState} = handle_cast({new_state, NewIdxState}, State),
- maybe_restart_updater(NewState),
- {noreply, NewState};
+ 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, commit_delay=Delay} = State,
CurrSeq = Mod:get(update_seq, NewIdxState),
@@ -231,6 +237,30 @@ 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]) 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}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/ee942109/src/couch_index/src/couch_index_server.erl
----------------------------------------------------------------------
diff --git a/src/couch_index/src/couch_index_server.erl b/src/couch_index/src/couch_index_server.erl
index e5f4571..975f44d 100644
--- a/src/couch_index/src/couch_index_server.erl
+++ b/src/couch_index/src/couch_index_server.erl
@@ -70,10 +70,10 @@ get_index(Module, IdxState) ->
init([]) ->
process_flag(trap_exit, true),
couch_config:register(fun ?MODULE:config_change/2),
- couch_db_update_notifier:start_link(fun ?MODULE:update_notify/1),
ets:new(?BY_SIG, [protected, set, named_table]),
ets:new(?BY_PID, [private, set, named_table]),
- ets:new(?BY_DB, [private, bag, named_table]),
+ ets:new(?BY_DB, [protected, bag, named_table]),
+ couch_db_update_notifier:start_link(fun ?MODULE:update_notify/1),
RootDir = couch_index_util:root_dir(),
% Deprecation warning if it wasn't index_dir
case couch_config:get("couchdb", "index_dir") of
@@ -104,13 +104,13 @@ handle_call({get_index, {_Mod, _IdxState, DbName, Sig}=Args}, From, State) ->
[{_, Pid}] when is_pid(Pid) ->
{reply, {ok, Pid}, State}
end;
-handle_call({async_open, {DbName, Sig}, {ok, Pid}}, _From, State) ->
+handle_call({async_open, {DbName, DDocId, Sig}, {ok, Pid}}, _From, State) ->
[{_, Waiters}] = ets:lookup(?BY_SIG, {DbName, Sig}),
[gen_server:reply(From, {ok, Pid}) || From <- Waiters],
link(Pid),
- add_to_ets(DbName, Sig, Pid),
+ add_to_ets(DbName, Sig, DDocId, Pid),
{reply, ok, State};
-handle_call({async_error, {DbName, Sig}, Error}, _From, State) ->
+handle_call({async_error, {DbName, _DDocId, Sig}, Error}, _From, State) ->
[{_, Waiters}] = ets:lookup(?BY_SIG, {DbName, Sig}),
[gen_server:reply(From, Error) || From <- Waiters],
ets:delete(?BY_SIG, {DbName, Sig}),
@@ -128,7 +128,9 @@ handle_cast({reset_indexes, DbName}, State) ->
handle_info({'EXIT', Pid, Reason}, Server) ->
case ets:lookup(?BY_PID, Pid) of
[{Pid, DbName, Sig}] ->
- rem_from_ets(DbName, Sig, Pid);
+ [{DbName, {DDocId, Sig}}] =
+ ets:match_object(?BY_DB, {DbName, {'$1', Sig}}),
+ rem_from_ets(DbName, Sig, DDocId, Pid);
[] when Reason /= normal ->
exit(Reason);
_Else ->
@@ -142,37 +144,40 @@ code_change(_OldVsn, State, _Extra) ->
new_index({Mod, IdxState, DbName, Sig}) ->
+ DDocId = Mod:get(idx_name, IdxState),
case couch_index:start_link({Mod, IdxState}) of
{ok, Pid} ->
- gen_server:call(?MODULE, {async_open, {DbName, Sig}, {ok, Pid}}),
+ ok = gen_server:call(
+ ?MODULE, {async_open, {DbName, DDocId, Sig}, {ok, Pid}}),
unlink(Pid);
Error ->
- gen_server:call(?MODULE, {async_error, {DbName, Sig}, Error})
+ ok = gen_server:call(
+ ?MODULE, {async_error, {DbName, DDocId, Sig}, Error})
end.
reset_indexes(DbName, Root) ->
% shutdown all the updaters and clear the files, the db got changed
- Fun = fun({_, Sig}) ->
+ Fun = fun({_, {DDocId, Sig}}) ->
[{_, Pid}] = ets:lookup(?BY_SIG, {DbName, Sig}),
couch_util:shutdown_sync(Pid),
- rem_from_ets(DbName, Sig, Pid)
+ rem_from_ets(DbName, Sig, DDocId, Pid)
end,
lists:foreach(Fun, ets:lookup(?BY_DB, DbName)),
Path = Root ++ "/." ++ binary_to_list(DbName) ++ "_design",
couch_file:nuke_dir(Root, Path).
-add_to_ets(DbName, Sig, Pid) ->
+add_to_ets(DbName, Sig, DDocId, Pid) ->
ets:insert(?BY_SIG, {{DbName, Sig}, Pid}),
ets:insert(?BY_PID, {Pid, {DbName, Sig}}),
- ets:insert(?BY_DB, {DbName, Sig}).
+ ets:insert(?BY_DB, {DbName, {DDocId, Sig}}).
-rem_from_ets(DbName, Sig, Pid) ->
+rem_from_ets(DbName, Sig, DDocId, Pid) ->
ets:delete(?BY_SIG, {DbName, Sig}),
ets:delete(?BY_PID, Pid),
- ets:delete_object(?BY_DB, {DbName, Sig}).
+ ets:delete_object(?BY_DB, {DbName, {DDocId, Sig}}).
config_change("couchdb", "view_index_dir") ->
@@ -185,6 +190,18 @@ update_notify({deleted, DbName}) ->
gen_server:cast(?MODULE, {reset_indexes, DbName});
update_notify({created, DbName}) ->
gen_server:cast(?MODULE, {reset_indexes, DbName});
+update_notify({ddoc_updated, {DbName, DDocId}}) ->
+ case ets:match_object(?BY_DB, {DbName, {DDocId, '$1'}}) of
+ [] ->
+ ok;
+ [{DbName, {DDocId, Sig}}] ->
+ case ets:lookup(?BY_SIG, {DbName, Sig}) of
+ [{_, IndexPid}] ->
+ (catch gen_server:cast(IndexPid, ddoc_updated));
+ [] ->
+ ok
+ end
+ end;
update_notify(_) ->
ok.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/ee942109/test/etap/200-view-group-no-db-leaks.t
----------------------------------------------------------------------
diff --git a/test/etap/200-view-group-no-db-leaks.t b/test/etap/200-view-group-no-db-leaks.t
index 7ad7439..4586ff4 100755
--- a/test/etap/200-view-group-no-db-leaks.t
+++ b/test/etap/200-view-group-no-db-leaks.t
@@ -52,7 +52,7 @@ ddoc_name() -> <<"foo">>.
main(_) ->
test_util:init_code_path(),
- etap:plan(18),
+ etap:plan(28),
case (catch test()) of
ok ->
etap:end_tests();
@@ -72,7 +72,7 @@ test() ->
create_db(),
create_docs(),
- create_design_doc(),
+ {ok, DDocRev} = create_design_doc(),
{ok, IndexerPid} = couch_index_server:get_index(
couch_mrview_index, test_db_name(), <<"_design/", (ddoc_name())/binary>>
@@ -80,12 +80,12 @@ test() ->
etap:is(is_pid(IndexerPid), true, "got view group pid"),
etap:is(is_process_alive(IndexerPid), true, "view group pid is alive"),
- query_view(),
+ query_view(3, null, false),
check_db_ref_count(),
etap:is(is_process_alive(IndexerPid), true, "view group pid is alive"),
create_new_doc(<<"doc1000">>),
- query_view(),
+ query_view(4, null, false),
check_db_ref_count(),
etap:is(is_process_alive(IndexerPid), true, "view group pid is alive"),
@@ -104,17 +104,39 @@ test() ->
etap:is(is_process_alive(IndexerPid), true, "view group pid is alive"),
create_new_doc(<<"doc1001">>),
- query_view(),
+ query_view(5, null, false),
check_db_ref_count(),
etap:is(is_process_alive(IndexerPid), true, "view group pid is alive"),
+ etap:diag("updating the design document with a new view definition"),
+ {ok, _NewDDocRev} = update_ddoc_view(DDocRev),
+
+ {ok, NewIndexerPid} = couch_index_server:get_index(
+ couch_mrview_index, test_db_name(), <<"_design/", (ddoc_name())/binary>>
+ ),
+ etap:is(is_pid(NewIndexerPid), true, "got new view group pid"),
+ etap:is(is_process_alive(NewIndexerPid), true, "new view group pid is alive"),
+ etap:isnt(NewIndexerPid, IndexerPid, "new view group has a different pid"),
+ etap:diag("querying view with ?stale=ok, must return empty row set"),
+ query_view(0, foo, ok),
+ etap:diag("querying view (without stale), must return 5 rows with value 1"),
+ query_view(5, 1, false),
MonRef = erlang:monitor(process, IndexerPid),
- ok = couch_server:delete(test_db_name(), []),
receive
{'DOWN', MonRef, _, _, _} ->
- etap:diag("view group is dead after DB deletion")
+ etap:diag("old view group is dead after ddoc update")
+ after 5000 ->
+ etap:bail("old view group is not dead after ddoc update")
+ end,
+
+ etap:diag("deleting database"),
+ MonRef2 = erlang:monitor(process, NewIndexerPid),
+ ok = couch_server:delete(test_db_name(), []),
+ receive
+ {'DOWN', MonRef2, _, _, _} ->
+ etap:diag("new view group is dead after DB deletion")
after 5000 ->
- etap:bail("view group did not die after DB deletion")
+ etap:bail("new view group did not die after DB deletion")
end,
ok = timer:sleep(1000),
@@ -222,9 +244,27 @@ create_design_doc() ->
]}}
]}}
]}),
- {ok, _} = couch_db:update_docs(Db, [DDoc]),
+ {ok, Rev} = couch_db:update_doc(Db, DDoc, []),
couch_db:ensure_full_commit(Db),
- couch_db:close(Db).
+ couch_db:close(Db),
+ {ok, Rev}.
+
+update_ddoc_view(DDocRev) ->
+ {ok, Db} = couch_db:open(test_db_name(), [admin_user_ctx()]),
+ DDoc = couch_doc:from_json_obj({[
+ {<<"_id">>, <<"_design/", (ddoc_name())/binary>>},
+ {<<"_rev">>, couch_doc:rev_to_str(DDocRev)},
+ {<<"language">>, <<"javascript">>},
+ {<<"views">>, {[
+ {<<"bar">>, {[
+ {<<"map">>, <<"function(doc) { emit(doc._id, 1); }">>}
+ ]}}
+ ]}}
+ ]}),
+ {ok, NewRev} = couch_db:update_doc(Db, DDoc, []),
+ couch_db:ensure_full_commit(Db),
+ couch_db:close(Db),
+ {ok, NewRev}.
create_new_doc(Id) ->
{ok, Db} = couch_db:open(test_db_name(), [admin_user_ctx()]),
@@ -240,10 +280,26 @@ db_url() ->
"http://" ++ get(addr) ++ ":" ++ get(port) ++ "/" ++
binary_to_list(test_db_name()).
-query_view() ->
- {ok, Code, _Headers, _Body} = test_util:request(
- db_url() ++ "/_design/" ++ binary_to_list(ddoc_name()) ++ "/_view/bar",
+query_view(ExpectedRowCount, ExpectedRowValue, Stale) ->
+ {ok, Code, _Headers, Body} = test_util:request(
+ db_url() ++ "/_design/" ++ binary_to_list(ddoc_name()) ++ "/_view/bar"
+ ++ case Stale of
+ false -> [];
+ _ -> "?stale=" ++ atom_to_list(Stale)
+ end,
[],
get),
etap:is(Code, 200, "got view response"),
- ok.
+ {Props} = ejson:decode(Body),
+ Rows = couch_util:get_value(<<"rows">>, Props, []),
+ etap:is(length(Rows), ExpectedRowCount, "result set has correct # of rows"),
+ lists:foreach(
+ fun({Row}) ->
+ case couch_util:get_value(<<"value">>, Row) of
+ ExpectedRowValue ->
+ ok;
+ _ ->
+ etap:bail("row has incorrect value")
+ end
+ end,
+ Rows).