summaryrefslogtreecommitdiff
path: root/src/couchdb/couch_query_servers.erl
diff options
context:
space:
mode:
Diffstat (limited to 'src/couchdb/couch_query_servers.erl')
-rw-r--r--src/couchdb/couch_query_servers.erl75
1 files changed, 37 insertions, 38 deletions
diff --git a/src/couchdb/couch_query_servers.erl b/src/couchdb/couch_query_servers.erl
index ff89a67a..cb760a10 100644
--- a/src/couchdb/couch_query_servers.erl
+++ b/src/couchdb/couch_query_servers.erl
@@ -15,7 +15,7 @@
-export([start_link/0]).
--export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2,code_change/3,stop/0]).
+-export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2,code_change/3]).
-export([start_doc_map/2, map_docs/2, stop_doc_map/1]).
-export([reduce/3, rereduce/3,validate_doc_update/5]).
-export([filter_docs/5]).
@@ -38,9 +38,6 @@
start_link() ->
gen_server:start_link({local, couch_query_servers}, couch_query_servers, [], []).
-stop() ->
- exit(whereis(couch_query_servers), normal).
-
start_doc_map(Lang, Functions) ->
Proc = get_os_process(Lang),
lists:foreach(fun(FunctionSource) ->
@@ -240,17 +237,18 @@ init([]) ->
ok = couch_config:register(
fun("query_servers" ++ _, _) ->
- ?MODULE:stop()
+ supervisor:terminate_child(couch_secondary_services, query_servers),
+ supervisor:restart_child(couch_secondary_services, query_servers)
end),
ok = couch_config:register(
fun("native_query_servers" ++ _, _) ->
- ?MODULE:stop()
+ supervisor:terminate_child(couch_secondary_services, query_servers),
+ [supervisor:restart_child(couch_secondary_services, query_servers)]
end),
Langs = ets:new(couch_query_server_langs, [set, private]),
PidProcs = ets:new(couch_query_server_pid_langs, [set, private]),
LangProcs = ets:new(couch_query_server_procs, [set, private]),
- InUse = ets:new(couch_query_server_used, [set, private]),
% 'query_servers' specifies an OS command-line to execute.
lists:foreach(fun({Lang, Command}) ->
true = ets:insert(Langs, {?l2b(Lang),
@@ -265,75 +263,71 @@ init([]) ->
process_flag(trap_exit, true),
{ok, {Langs, % Keyed by language name, value is {Mod,Func,Arg}
PidProcs, % Keyed by PID, valus is a #proc record.
- LangProcs, % Keyed by language name, value is a #proc record
- InUse % Keyed by PID, value is #proc record.
+ LangProcs % Keyed by language name, value is a #proc record
}}.
-terminate(_Reason, _Server) ->
+terminate(_Reason, {_Langs, PidProcs, _LangProcs}) ->
+ [couch_util:shutdown_sync(P) || {P,_} <- ets:tab2list(PidProcs)],
ok.
-handle_call({get_proc, #doc{body={Props}}=DDoc, DDocKey}, _From, {Langs, PidProcs, LangProcs, InUse}=Server) ->
+handle_call({get_proc, #doc{body={Props}}=DDoc, DDocKey}, _From, {Langs, PidProcs, LangProcs}=Server) ->
% Note to future self. Add max process limit.
Lang = proplists:get_value(<<"language">>, Props, <<"javascript">>),
case ets:lookup(LangProcs, Lang) of
[{Lang, [P|Rest]}] ->
% find a proc in the set that has the DDoc
case proc_with_ddoc(DDoc, DDocKey, [P|Rest]) of
- {ok, Proc} ->
- % looks like the proc isn't getting dropped from the list.
- % we need to change this to take a fun for equality checking
- % so we can do a comparison on portnum
- rem_from_list(LangProcs, Lang, Proc),
- add_to_list(InUse, Lang, Proc),
- {reply, {ok, Proc, get_query_server_config()}, Server};
- Error ->
- {reply, Error, Server}
- end;
+ {ok, Proc} ->
+ rem_from_list(LangProcs, Lang, Proc),
+ {reply, {ok, Proc, get_query_server_config()}, Server};
+ Error ->
+ {reply, Error, Server}
+ end;
_ ->
case (catch new_process(Langs, Lang)) of
{ok, Proc} ->
add_value(PidProcs, Proc#proc.pid, Proc),
case proc_with_ddoc(DDoc, DDocKey, [Proc]) of
- {ok, Proc2} ->
- rem_from_list(LangProcs, Lang, Proc),
- add_to_list(InUse, Lang, Proc2),
- {reply, {ok, Proc2, get_query_server_config()}, Server};
- Error ->
- {reply, Error, Server}
- end;
+ {ok, Proc2} ->
+ {reply, {ok, Proc2, get_query_server_config()}, Server};
+ Error ->
+ {reply, Error, Server}
+ end;
Error ->
{reply, Error, Server}
end
end;
-handle_call({get_proc, Lang}, _From, {Langs, PidProcs, LangProcs, InUse}=Server) ->
+handle_call({get_proc, Lang}, _From, {Langs, PidProcs, LangProcs}=Server) ->
% Note to future self. Add max process limit.
case ets:lookup(LangProcs, Lang) of
[{Lang, [Proc|_]}] ->
- add_value(PidProcs, Proc#proc.pid, Proc),
rem_from_list(LangProcs, Lang, Proc),
- add_to_list(InUse, Lang, Proc),
{reply, {ok, Proc, get_query_server_config()}, Server};
_ ->
case (catch new_process(Langs, Lang)) of
{ok, Proc} ->
add_value(PidProcs, Proc#proc.pid, Proc),
- add_to_list(InUse, Lang, Proc),
{reply, {ok, Proc, get_query_server_config()}, Server};
Error ->
{reply, Error, Server}
end
end;
-handle_call({ret_proc, Proc}, _From, {_, _, LangProcs, InUse}=Server) ->
+handle_call({unlink_proc, Pid}, _From, {_, PidProcs, _}=Server) ->
+ rem_value(PidProcs, Pid),
+ unlink(Pid),
+ {reply, ok, Server};
+handle_call({ret_proc, Proc}, _From, {_, PidProcs, LangProcs}=Server) ->
% Along with max process limit, here we should check
% if we're over the limit and discard when we are.
+ add_value(PidProcs, Proc#proc.pid, Proc),
add_to_list(LangProcs, Proc#proc.lang, Proc),
- rem_from_list(InUse, Proc#proc.lang, Proc),
+ link(Proc#proc.pid),
{reply, true, Server}.
handle_cast(_Whatever, Server) ->
{noreply, Server}.
-handle_info({'EXIT', Pid, Status}, {_, PidProcs, LangProcs, InUse}=Server) ->
+handle_info({'EXIT', Pid, Status}, {_, PidProcs, LangProcs}=Server) ->
case ets:lookup(PidProcs, Pid) of
[{Pid, Proc}] ->
case Status of
@@ -342,11 +336,14 @@ handle_info({'EXIT', Pid, Status}, {_, PidProcs, LangProcs, InUse}=Server) ->
end,
rem_value(PidProcs, Pid),
catch rem_from_list(LangProcs, Proc#proc.lang, Proc),
- catch rem_from_list(InUse, Proc#proc.lang, Proc),
{noreply, Server};
[] ->
- ?LOG_DEBUG("Unknown linked process died: ~p (reason: ~p)", [Pid, Status]),
- {stop, Status, Server}
+ case Status of
+ normal ->
+ {noreply, Server};
+ _ ->
+ {stop, Status, Server}
+ end
end.
code_change(_OldVsn, State, _Extra) ->
@@ -423,6 +420,7 @@ get_ddoc_process(#doc{} = DDoc, DDocKey) ->
proc_set_timeout(Proc, list_to_integer(couch_config:get(
"couchdb", "os_process_timeout", "5000"))),
link(Proc#proc.pid),
+ gen_server:call(couch_query_servers, {unlink_proc, Proc#proc.pid}),
Proc;
_ ->
catch proc_stop(Proc),
@@ -440,6 +438,7 @@ get_os_process(Lang) ->
proc_set_timeout(Proc, list_to_integer(couch_config:get(
"couchdb", "os_process_timeout", "5000"))),
link(Proc#proc.pid),
+ gen_server:call(couch_query_servers, {unlink_proc, Proc#proc.pid}),
Proc;
_ ->
catch proc_stop(Proc),
generated by cgit v1.2.3 (git 2.25.1) at 2024-07-21 20:49:08 +0000