You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2017/06/02 15:34:38 UTC
[couchdb] branch master updated: Notify couch_index_processes on
all shards when ddoc updated
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/couchdb.git
The following commit(s) were added to refs/heads/master by this push:
new 994fe33 Notify couch_index_processes on all shards when ddoc updated
994fe33 is described below
commit 994fe33bb02ca6a1d1b2d807a486c3b386426999
Author: Mayya Sharipova <ma...@ca.ibm.com>
AuthorDate: Mon Apr 24 17:13:18 2017 -0400
Notify couch_index_processes on all shards when ddoc updated
Before when a design doc is updated/deleted, only one couch_index
process was notified - the one which shard contained a design doc.
couch_index processes from other shards still continued to exist,
and indexing activities for these processes were still be going on.
The patch notifies couch_index_processes on all shards
COUCHDB-3400
---
src/couch_index/src/couch_index.erl | 23 ++++
src/couch_index/src/couch_index_server.erl | 16 +++
.../test/couch_index_ddoc_updated_tests.erl | 144 +++++++++++++++++++++
3 files changed, 183 insertions(+)
diff --git a/src/couch_index/src/couch_index.erl b/src/couch_index/src/couch_index.erl
index c86f5e1..9da928d 100644
--- a/src/couch_index/src/couch_index.erl
+++ b/src/couch_index/src/couch_index.erl
@@ -257,6 +257,29 @@ handle_cast(delete, State) ->
#st{mod=Mod, idx_state=IdxState} = State,
ok = Mod:delete(IdxState),
{stop, normal, State};
+handle_cast({ddoc_updated, DDocResult}, State) ->
+ #st{mod = Mod, idx_state = IdxState, waiters = Waiters} = State,
+ Shutdown = case DDocResult of
+ {not_found, deleted} ->
+ true;
+ {ok, DDoc} ->
+ DbName = Mod:get(db_name, IdxState),
+ couch_util:with_db(DbName, fun(Db) ->
+ {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(ddoc_updated, State) ->
#st{mod = Mod, idx_state = IdxState, waiters = Waiters} = State,
DbName = Mod:get(db_name, IdxState),
diff --git a/src/couch_index/src/couch_index_server.erl b/src/couch_index/src/couch_index_server.erl
index 4e86f5e..92b8c8e 100644
--- a/src/couch_index/src/couch_index_server.erl
+++ b/src/couch_index/src/couch_index_server.erl
@@ -251,6 +251,22 @@ handle_db_event(DbName, created, St) ->
handle_db_event(DbName, deleted, St) ->
gen_server:cast(?MODULE, {reset_indexes, DbName}),
{ok, St};
+handle_db_event(<<"shards/", _/binary>> = DbName, {ddoc_updated,
+ DDocId}, St) ->
+ DDocResult = couch_util:with_db(DbName, fun(Db) ->
+ couch_db:open_doc(Db, DDocId, [ejson_body, ?ADMIN_CTX])
+ end),
+ DbShards = [mem3:name(Sh) || Sh <- mem3:local_shards(mem3:dbname(DbName))],
+ lists:foreach(fun(DbShard) ->
+ lists:foreach(fun({_DbShard, {_DDocId, Sig}}) ->
+ case ets:lookup(?BY_SIG, {DbShard, Sig}) of
+ [{_, IndexPid}] -> (catch
+ gen_server:cast(IndexPid, {ddoc_updated, DDocResult}));
+ [] -> []
+ end
+ end, ets:match_object(?BY_DB, {DbShard, {DDocId, '$1'}}))
+ end, DbShards),
+ {ok, St};
handle_db_event(DbName, {ddoc_updated, DDocId}, St) ->
lists:foreach(fun({_DbName, {_DDocId, Sig}}) ->
case ets:lookup(?BY_SIG, {DbName, Sig}) of
diff --git a/src/couch_index/test/couch_index_ddoc_updated_tests.erl b/src/couch_index/test/couch_index_ddoc_updated_tests.erl
new file mode 100644
index 0000000..007f569
--- /dev/null
+++ b/src/couch_index/test/couch_index_ddoc_updated_tests.erl
@@ -0,0 +1,144 @@
+% 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_ddoc_updated_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+
+start() ->
+ fake_index(),
+ Ctx = test_util:start_couch([mem3, fabric]),
+ DbName = ?tempdb(),
+ ok = fabric:create_db(DbName, [?ADMIN_CTX]),
+ {Ctx, DbName}.
+
+
+stop({Ctx, DbName}) ->
+ (catch meck:unload(test_index)),
+ ok = fabric:delete_db(DbName, [?ADMIN_CTX]),
+ DbDir = config:get("couchdb", "database_dir", "."),
+ WaitFun = fun() ->
+ filelib:fold_files(DbDir, <<".*", DbName/binary, "\.[0-9]+.*">>,
+ true, fun(_F, _A) -> wait end, ok)
+ end,
+ ok = test_util:wait(WaitFun),
+ test_util:stop_couch(Ctx),
+ ok.
+
+
+ddoc_update_test_() ->
+ {
+ "Check ddoc update actions",
+ {
+ setup,
+ fun start/0, fun stop/1,
+ fun check_all_indexers_exit_on_ddoc_change/1
+ }
+ }.
+
+
+check_all_indexers_exit_on_ddoc_change({_Ctx, DbName}) ->
+ ?_test(begin
+ [DbShard1 | RestDbShards] = lists:map(fun(Sh) ->
+ {ok, ShardDb} = couch_db:open(mem3:name(Sh), []),
+ ShardDb
+ end, mem3:local_shards(mem3:dbname(DbName))),
+
+ % create a DDoc on Db1
+ DDocID = <<"idx_name">>,
+ DDocJson = couch_doc:from_json_obj({[
+ {<<"_id">>, DDocID},
+ {<<"value">>, 1}
+ ]}),
+ {ok, _Rev} = couch_db:update_doc(DbShard1, DDocJson, []),
+ {ok, DbShard} = couch_db:reopen(DbShard1),
+ {ok, DDoc} = couch_db:open_doc(
+ DbShard, DDocID, [ejson_body, ?ADMIN_CTX]),
+ DbShards = [DbShard | RestDbShards],
+ N = length(DbShards),
+
+ % run couch_index process for each shard database
+ ok = meck:reset(test_index),
+ lists:foreach(fun(ShardDb) ->
+ couch_index_server:get_index(test_index, ShardDb, DDoc)
+ end, DbShards),
+
+ IndexesBefore = get_indexes_by_ddoc(DDocID, N),
+ ?assertEqual(N, length(IndexesBefore)),
+
+ AliveBefore = lists:filter(fun erlang:is_process_alive/1, IndexesBefore),
+ ?assertEqual(N, length(AliveBefore)),
+
+ % update ddoc
+ DDocJson2 = couch_doc:from_json_obj({[
+ {<<"_id">>, DDocID},
+ {<<"value">>, 2},
+ {<<"_rev">>, couch_doc:rev_to_str(DDoc#doc.revs)}
+ ]}),
+ {ok, _} = couch_db:update_doc(DbShard, DDocJson2, []),
+
+ % assert that all index processes exit after ddoc updated
+ ok = meck:reset(test_index),
+ couch_index_server:handle_db_event(
+ DbShard#db.name, {ddoc_updated, DDocID}, {st, ""}),
+
+ ok = meck:wait(N, test_index, init, ['_', '_'], 5000),
+ IndexesAfter = get_indexes_by_ddoc(DDocID, 0),
+ ?assertEqual(0, length(IndexesAfter)),
+
+ %% assert that previously running indexes are gone
+ AliveAfter = lists:filter(fun erlang:is_process_alive/1, IndexesBefore),
+ ?assertEqual(0, length(AliveAfter)),
+ ok
+ end).
+
+
+fake_index() ->
+ ok = meck:new([test_index], [non_strict]),
+ ok = meck:expect(test_index, init, fun(Db, DDoc) ->
+ {ok, {couch_db:name(Db), DDoc}}
+ end),
+ ok = meck:expect(test_index, open, fun(_Db, State) ->
+ {ok, State}
+ end),
+ ok = meck:expect(test_index, get, fun
+ (db_name, {DbName, _DDoc}) ->
+ DbName;
+ (idx_name, {_DbName, DDoc}) ->
+ DDoc#doc.id;
+ (signature, {_DbName, DDoc}) ->
+ couch_crypto:hash(md5, term_to_binary(DDoc));
+ (update_seq, Seq) ->
+ Seq
+ end).
+
+
+get_indexes_by_ddoc(DDocID, N) ->
+ Indexes = test_util:wait(fun() ->
+ Indxs = ets:match_object(
+ couchdb_indexes_by_db, {'$1', {DDocID, '$2'}}),
+ case length(Indxs) == N of
+ true ->
+ Indxs;
+ false ->
+ wait
+ end
+ end),
+ lists:foldl(fun({DbName, {_DDocID, Sig}}, Acc) ->
+ case ets:lookup(couchdb_indexes_by_sig, {DbName, Sig}) of
+ [{_, Pid}] -> [Pid|Acc];
+ _ -> Acc
+ end
+ end, [], Indexes).
+
--
To stop receiving notification emails like this one, please contact
['"commits@couchdb.apache.org" <co...@couchdb.apache.org>'].