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 2023/12/14 21:04:58 UTC
(couchdb) branch make-sure-to-reply-to-index-waiters created (now 103bb9c49)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch make-sure-to-reply-to-index-waiters
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at 103bb9c49 Make sure to reply to couch_index_server clients
This branch includes the following new commits:
new 103bb9c49 Make sure to reply to couch_index_server clients
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(couchdb) 01/01: Make sure to reply to couch_index_server clients
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch make-sure-to-reply-to-index-waiters
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 103bb9c49d0b98e531c05776e5483febb8e1770c
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Thu Dec 14 16:03:04 2023 -0500
Make sure to reply to couch_index_server clients
Previously we didin't always reply to waiting `get_index` calls as sometimes we
clobbered or deleted waiters in by_sig ets table.
Ref: https://github.com/apache/couchdb/pull/4491
---
src/couch_index/src/couch_index_server.erl | 66 ++++++++++++++++++++----------
1 file changed, 45 insertions(+), 21 deletions(-)
diff --git a/src/couch_index/src/couch_index_server.erl b/src/couch_index/src/couch_index_server.erl
index 549f8a226..4fc8bcb71 100644
--- a/src/couch_index/src/couch_index_server.erl
+++ b/src/couch_index/src/couch_index_server.erl
@@ -171,9 +171,9 @@ handle_call({async_open, {DbName, DDocId, Sig}, {ok, Pid}}, {OpenerPid, _} = Fro
unlink(OpenerPid),
ets:delete(State#st.openers, OpenerPid),
gen_server:reply(FromOpener, ok),
- [{_, Waiters}] = ets:lookup(State#st.by_sig, {DbName, Sig}),
+ Waiters = get_waiters(DbName, Sig, State),
add_to_ets(DbName, Sig, DDocId, Pid, State),
- [gen_server:reply(From, {ok, Pid}) || From <- Waiters],
+ ok = reply_to_waiters(Waiters, {ok, Pid}),
% Flush opener exit messages in case it died before we unlinked it
ok = flush_exit_messages_from(OpenerPid),
{noreply, State};
@@ -182,9 +182,9 @@ handle_call({async_error, {DbName, _DDocId, Sig}, Error}, {OpenerPid, _} = FromO
unlink(OpenerPid),
ets:delete(State#st.openers, OpenerPid),
gen_server:reply(FromOpener, ok),
- [{_, Waiters}] = ets:lookup(State#st.by_sig, {DbName, Sig}),
+ Waiters = get_waiters(DbName, Sig, State),
ets:delete(State#st.by_sig, {DbName, Sig}),
- [gen_server:reply(From, Error) || From <- Waiters],
+ ok = reply_to_waiters(Waiters, Error),
% Flush opener exit messages in case it died before we unlinked it
ok = flush_exit_messages_from(OpenerPid),
{noreply, State};
@@ -207,8 +207,8 @@ handle_cast({add_to_ets, [Pid, DbName, DDocId, Sig]}, State) ->
handle_cast({rem_from_ets, [DbName, DDocId, Sig]}, State) ->
ets:delete_object(State#st.by_db, {DbName, {DDocId, Sig}}),
{noreply, State};
-handle_cast({rem_from_ets, [DbName]}, State) ->
- rem_from_ets(DbName, State),
+handle_cast({rem_from_ets, [DbName, Reason]}, State) ->
+ rem_from_ets_with_reply(DbName, Reason, State),
{noreply, State}.
handle_info({'EXIT', Pid, Reason}, Server) ->
@@ -218,10 +218,11 @@ handle_info({'EXIT', Pid, Reason}, Server) ->
|| {_, {DDocId, _}} <-
ets:match_object(Server#st.by_db, {DbName, {'$1', Sig}})
],
- rem_from_ets(DbName, Sig, DDocIds, Pid, Server)
+ rem_from_ets_with_reply(DbName, Sig, DDocIds, Reason, Server)
end,
case ets:lookup(Server#st.by_pid, Pid) of
[{Pid, {DbName, Sig}}] ->
+ ets:delete(Server#st.by_pid, Pid),
Cleanup(DbName, Sig);
[] when Reason /= normal ->
case ets:lookup(Server#st.openers, Pid) of
@@ -285,11 +286,16 @@ reset_indexes(DbName, #st{} = State) ->
ets:lookup(State#st.by_db, DbName)
),
Fun = fun({Sig, DDocIds}) ->
- [{_, Pid}] = ets:lookup(State#st.by_sig, {DbName, Sig}),
- unlink(Pid),
- gen_server:cast(Pid, delete),
- ok = flush_exit_messages_from(Pid),
- rem_from_ets(DbName, Sig, DDocIds, Pid, State)
+ case ets:lookup(State#st.by_sig, {DbName, Sig}) of
+ [{_, Pid}] when is_pid(Pid) ->
+ unlink(Pid),
+ gen_server:cast(Pid, delete),
+ ok = flush_exit_messages_from(Pid),
+ ets:delete(State#st.by_pid, Pid);
+ _ ->
+ ok
+ end,
+ rem_from_ets_with_reply(DbName, Sig, DDocIds, {error, index_reset}, State)
end,
lists:foreach(Fun, dict:to_list(SigDDocIds)),
% We only need one of the index servers to do this.
@@ -301,22 +307,35 @@ reset_indexes(DbName, #st{} = State) ->
ok
end.
+get_waiters(DbName, Sig, St = #st{}) ->
+ case ets:lookup(St#st.by_sig, {DbName, Sig}) of
+ [{_, Waiters}] when is_list(Waiters) -> Waiters;
+ [{_, Pid}] when is_pid(Pid) -> [];
+ [] -> []
+ end.
+
+
+reply_to_waiters(Waiters, Reply) when is_list(Waiters) ->
+ [gen_server:reply(From, Reply) || From <- Waiters],
+ ok.
+
add_to_ets(DbName, Sig, DDocId, Pid, #st{} = St) ->
ets:insert(St#st.by_sig, {{DbName, Sig}, Pid}),
ets:insert(St#st.by_pid, {Pid, {DbName, Sig}}),
ets:insert(St#st.by_db, {DbName, {DDocId, Sig}}).
-rem_from_ets(DbName, Sig, DDocIds, Pid, #st{} = St) ->
+rem_from_ets_with_reply(DbName, Sig, DDocIds, Reply, #st{} = St) ->
+ Waiters = get_waiters(DbName, Sig, St),
ets:delete(St#st.by_sig, {DbName, Sig}),
- ets:delete(St#st.by_pid, Pid),
lists:foreach(
fun(DDocId) ->
ets:delete_object(St#st.by_db, {DbName, {DDocId, Sig}})
end,
DDocIds
- ).
+ ),
+ ok = reply_to_waiters(Waiters, Reply).
-rem_from_ets(DbName, #st{} = State) ->
+rem_from_ets_with_reply(DbName, Reply, #st{} = State) ->
SigDDocIds = lists:foldl(
fun({_, {DDocId, Sig}}, DDict) ->
dict:append(Sig, DDocId, DDict)
@@ -325,10 +344,15 @@ rem_from_ets(DbName, #st{} = State) ->
ets:lookup(State#st.by_db, DbName)
),
Fun = fun({Sig, DDocIds}) ->
- [{_, Pid}] = ets:lookup(State#st.by_sig, {DbName, Sig}),
- unlink(Pid),
- ok = flush_exit_messages_from(Pid),
- rem_from_ets(DbName, Sig, DDocIds, Pid, State)
+ case ets:lookup(State#st.by_sig, {DbName, Sig}) of
+ [{_, Pid}] when is_pid(Pid) ->
+ unlink(Pid),
+ ok = flush_exit_messages_from(Pid),
+ ets:delete(State#st.by_pid, Pid);
+ _ ->
+ ok
+ end,
+ rem_from_ets_with_reply(DbName, Sig, DDocIds, Reply, State)
end,
lists:foreach(Fun, dict:to_list(SigDDocIds)).
@@ -380,7 +404,7 @@ handle_db_event(<<"shards/", _/binary>> = DbName, {ddoc_updated, DDocId}, St) ->
couch_log:warning("~p: handle_db_event ~p for db ~p, reason ~p, stack ~p", [
?MODULE, Class, DbName, Reason, Stack
]),
- gen_server:cast(St#st.server_name, {rem_from_ets, [DbName]}),
+ gen_server:cast(St#st.server_name, {rem_from_ets, [DbName, Reason]}),
{ok, St}
end;
handle_db_event(DbName, {ddoc_updated, DDocId}, St) ->