| % 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_query_servers). |
| -behaviour(gen_server). |
| |
| -export([start_link/0, config_change/1]). |
| |
| -export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2,code_change/3]). |
| -export([start_doc_map/3, map_docs/2, map_doc_raw/2, stop_doc_map/1, raw_to_ejson/1]). |
| -export([reduce/3, rereduce/3,validate_doc_update/5]). |
| -export([filter_docs/5]). |
| -export([filter_view/3]). |
| |
| -export([with_ddoc_proc/2, proc_prompt/2, ddoc_prompt/3, ddoc_proc_prompt/3, json_doc/1]). |
| |
| % For 210-os-proc-pool.t |
| -export([get_os_process/1, ret_os_process/1]). |
| |
| -include("couch_db.hrl"). |
| |
| -record(proc, { |
| pid, |
| lang, |
| ddoc_keys = [], |
| prompt_fun, |
| set_timeout_fun, |
| stop_fun |
| }). |
| |
| -record(qserver, { |
| langs, % Keyed by language name, value is {Mod,Func,Arg} |
| pid_procs, % Keyed by PID, valus is a #proc record. |
| lang_procs, % Keyed by language name, value is a #proc record |
| lang_limits, % Keyed by language name, value is {Lang, Limit, Current} |
| waitlist = [], |
| config |
| }). |
| |
| start_link() -> |
| gen_server:start_link({local, couch_query_servers}, couch_query_servers, [], []). |
| |
| start_doc_map(Lang, Functions, Lib) -> |
| Proc = get_os_process(Lang), |
| case Lib of |
| {[]} -> ok; |
| Lib -> |
| true = proc_prompt(Proc, [<<"add_lib">>, Lib]) |
| end, |
| lists:foreach(fun(FunctionSource) -> |
| true = proc_prompt(Proc, [<<"add_fun">>, FunctionSource]) |
| end, Functions), |
| {ok, Proc}. |
| |
| map_docs(Proc, Docs) -> |
| % send the documents |
| Results = lists:map( |
| fun(Doc) -> |
| Json = couch_doc:to_json_obj(Doc, []), |
| |
| FunsResults = proc_prompt(Proc, [<<"map_doc">>, Json]), |
| % the results are a json array of function map yields like this: |
| % [FunResults1, FunResults2 ...] |
| % where funresults is are json arrays of key value pairs: |
| % [[Key1, Value1], [Key2, Value2]] |
| % Convert the key, value pairs to tuples like |
| % [{Key1, Value1}, {Key2, Value2}] |
| lists:map( |
| fun(FunRs) -> |
| [list_to_tuple(FunResult) || FunResult <- FunRs] |
| end, |
| FunsResults) |
| end, |
| Docs), |
| {ok, Results}. |
| |
| map_doc_raw(Proc, Doc) -> |
| Json = couch_doc:to_json_obj(Doc, []), |
| {ok, proc_prompt_raw(Proc, [<<"map_doc">>, Json])}. |
| |
| |
| stop_doc_map(nil) -> |
| ok; |
| stop_doc_map(Proc) -> |
| ok = ret_os_process(Proc). |
| |
| group_reductions_results([]) -> |
| []; |
| group_reductions_results(List) -> |
| {Heads, Tails} = lists:foldl( |
| fun([H|T], {HAcc,TAcc}) -> |
| {[H|HAcc], [T|TAcc]} |
| end, {[], []}, List), |
| case Tails of |
| [[]|_] -> % no tails left |
| [Heads]; |
| _ -> |
| [Heads | group_reductions_results(Tails)] |
| end. |
| |
| rereduce(_Lang, [], _ReducedValues) -> |
| {ok, []}; |
| rereduce(Lang, RedSrcs, ReducedValues) -> |
| Grouped = group_reductions_results(ReducedValues), |
| Results = lists:zipwith( |
| fun |
| (<<"_", _/binary>> = FunSrc, Values) -> |
| {ok, [Result]} = builtin_reduce(rereduce, [FunSrc], [[[], V] || V <- Values], []), |
| Result; |
| (FunSrc, Values) -> |
| os_rereduce(Lang, [FunSrc], Values) |
| end, RedSrcs, Grouped), |
| {ok, Results}. |
| |
| reduce(_Lang, [], _KVs) -> |
| {ok, []}; |
| reduce(Lang, RedSrcs, KVs) -> |
| {OsRedSrcs, BuiltinReds} = lists:partition(fun |
| (<<"_", _/binary>>) -> false; |
| (_OsFun) -> true |
| end, RedSrcs), |
| {ok, OsResults} = os_reduce(Lang, OsRedSrcs, KVs), |
| {ok, BuiltinResults} = builtin_reduce(reduce, BuiltinReds, KVs, []), |
| recombine_reduce_results(RedSrcs, OsResults, BuiltinResults, []). |
| |
| recombine_reduce_results([], [], [], Acc) -> |
| {ok, lists:reverse(Acc)}; |
| recombine_reduce_results([<<"_", _/binary>>|RedSrcs], OsResults, [BRes|BuiltinResults], Acc) -> |
| recombine_reduce_results(RedSrcs, OsResults, BuiltinResults, [BRes|Acc]); |
| recombine_reduce_results([_OsFun|RedSrcs], [OsR|OsResults], BuiltinResults, Acc) -> |
| recombine_reduce_results(RedSrcs, OsResults, BuiltinResults, [OsR|Acc]). |
| |
| os_reduce(_Lang, [], _KVs) -> |
| {ok, []}; |
| os_reduce(Lang, OsRedSrcs, KVs) -> |
| Proc = get_os_process(Lang), |
| OsResults = try proc_prompt(Proc, [<<"reduce">>, OsRedSrcs, KVs]) of |
| [true, Reductions] -> Reductions |
| after |
| ok = ret_os_process(Proc) |
| end, |
| {ok, OsResults}. |
| |
| os_rereduce(Lang, OsRedSrcs, KVs) -> |
| Proc = get_os_process(Lang), |
| try proc_prompt(Proc, [<<"rereduce">>, OsRedSrcs, KVs]) of |
| [true, [Reduction]] -> Reduction |
| after |
| ok = ret_os_process(Proc) |
| end. |
| |
| |
| builtin_reduce(_Re, [], _KVs, Acc) -> |
| {ok, lists:reverse(Acc)}; |
| builtin_reduce(Re, [<<"_sum",_/binary>>|BuiltinReds], KVs, Acc) -> |
| Sum = builtin_sum_rows(KVs), |
| builtin_reduce(Re, BuiltinReds, KVs, [Sum|Acc]); |
| builtin_reduce(reduce, [<<"_count",_/binary>>|BuiltinReds], KVs, Acc) -> |
| Count = length(KVs), |
| builtin_reduce(reduce, BuiltinReds, KVs, [Count|Acc]); |
| builtin_reduce(rereduce, [<<"_count",_/binary>>|BuiltinReds], KVs, Acc) -> |
| Count = builtin_sum_rows(KVs), |
| builtin_reduce(rereduce, BuiltinReds, KVs, [Count|Acc]); |
| builtin_reduce(Re, [<<"_stats",_/binary>>|BuiltinReds], KVs, Acc) -> |
| Stats = builtin_stats(Re, KVs), |
| builtin_reduce(Re, BuiltinReds, KVs, [Stats|Acc]). |
| |
| builtin_sum_rows(KVs) -> |
| lists:foldl(fun |
| ([_Key, Value], Acc) when is_number(Value), is_number(Acc) -> |
| Acc + Value; |
| ([_Key, Value], Acc) when is_list(Value), is_list(Acc) -> |
| sum_terms(Acc, Value); |
| ([_Key, Value], Acc) when is_number(Value), is_list(Acc) -> |
| sum_terms(Acc, [Value]); |
| ([_Key, Value], Acc) when is_list(Value), is_number(Acc) -> |
| sum_terms([Acc], Value); |
| (_Else, _Acc) -> |
| throw({invalid_value, <<"builtin _sum function requires map values to be numbers or lists of numbers">>}) |
| end, 0, KVs). |
| |
| sum_terms([], []) -> |
| []; |
| sum_terms([_|_]=Xs, []) -> |
| Xs; |
| sum_terms([], [_|_]=Ys) -> |
| Ys; |
| sum_terms([X|Xs], [Y|Ys]) when is_number(X), is_number(Y) -> |
| [X+Y | sum_terms(Xs,Ys)]; |
| sum_terms(_, _) -> |
| throw({invalid_value, <<"builtin _sum function requires map values to be numbers or lists of numbers">>}). |
| |
| builtin_stats(reduce, []) -> |
| {[]}; |
| builtin_stats(reduce, [[_,First]|Rest]) when is_number(First) -> |
| Stats = lists:foldl(fun([_K,V], {S,C,Mi,Ma,Sq}) when is_number(V) -> |
| {S+V, C+1, lists:min([Mi, V]), lists:max([Ma, V]), Sq+(V*V)}; |
| (_, _) -> |
| throw({invalid_value, |
| <<"builtin _stats function requires map values to be numbers">>}) |
| end, {First,1,First,First,First*First}, Rest), |
| {Sum, Cnt, Min, Max, Sqr} = Stats, |
| {[{sum,Sum}, {count,Cnt}, {min,Min}, {max,Max}, {sumsqr,Sqr}]}; |
| |
| builtin_stats(rereduce, [[_,First]|Rest]) -> |
| {[{sum,Sum0}, {count,Cnt0}, {min,Min0}, {max,Max0}, {sumsqr,Sqr0}]} = First, |
| Stats = lists:foldl(fun([_K,Red], {S,C,Mi,Ma,Sq}) -> |
| {[{sum,Sum}, {count,Cnt}, {min,Min}, {max,Max}, {sumsqr,Sqr}]} = Red, |
| {Sum+S, Cnt+C, lists:min([Min, Mi]), lists:max([Max, Ma]), Sqr+Sq} |
| end, {Sum0,Cnt0,Min0,Max0,Sqr0}, Rest), |
| {Sum, Cnt, Min, Max, Sqr} = Stats, |
| {[{sum,Sum}, {count,Cnt}, {min,Min}, {max,Max}, {sumsqr,Sqr}]}. |
| |
| % use the function stored in ddoc.validate_doc_update to test an update. |
| validate_doc_update(DDoc, EditDoc, DiskDoc, Ctx, SecObj) -> |
| JsonEditDoc = couch_doc:to_json_obj(EditDoc, [revs]), |
| JsonDiskDoc = json_doc(DiskDoc), |
| case ddoc_prompt(DDoc, [<<"validate_doc_update">>], [JsonEditDoc, JsonDiskDoc, Ctx, SecObj]) of |
| 1 -> |
| ok; |
| {[{<<"forbidden">>, Message}]} -> |
| throw({forbidden, Message}); |
| {[{<<"unauthorized">>, Message}]} -> |
| throw({unauthorized, Message}) |
| end. |
| |
| json_doc(nil) -> null; |
| json_doc(Doc) -> |
| couch_doc:to_json_obj(Doc, [revs]). |
| |
| filter_view(DDoc, VName, Docs) -> |
| JsonDocs = [couch_doc:to_json_obj(Doc, [revs]) || Doc <- Docs], |
| [true, Passes] = ddoc_prompt(DDoc, [<<"views">>, VName, <<"map">>], [JsonDocs]), |
| {ok, Passes}. |
| |
| filter_docs(Req, Db, DDoc, FName, Docs) -> |
| JsonReq = case Req of |
| {json_req, JsonObj} -> |
| JsonObj; |
| #httpd{} = HttpReq -> |
| couch_httpd_external:json_req_obj(HttpReq, Db) |
| end, |
| JsonDocs = [couch_doc:to_json_obj(Doc, [revs]) || Doc <- Docs], |
| [true, Passes] = ddoc_prompt(DDoc, [<<"filters">>, FName], |
| [JsonDocs, JsonReq]), |
| {ok, Passes}. |
| |
| ddoc_proc_prompt({Proc, DDocId}, FunPath, Args) -> |
| proc_prompt(Proc, [<<"ddoc">>, DDocId, FunPath, Args]). |
| |
| ddoc_prompt(DDoc, FunPath, Args) -> |
| with_ddoc_proc(DDoc, fun({Proc, DDocId}) -> |
| proc_prompt(Proc, [<<"ddoc">>, DDocId, FunPath, Args]) |
| end). |
| |
| with_ddoc_proc(#doc{id=DDocId,revs={Start, [DiskRev|_]}}=DDoc, Fun) -> |
| Rev = couch_doc:rev_to_str({Start, DiskRev}), |
| DDocKey = {DDocId, Rev}, |
| Proc = get_ddoc_process(DDoc, DDocKey), |
| try Fun({Proc, DDocId}) |
| after |
| ok = ret_os_process(Proc) |
| end. |
| |
| init([]) -> |
| % register async to avoid deadlock on restart_child |
| Self = self(), |
| spawn(couch_config, register, [fun ?MODULE:config_change/1, Self]), |
| |
| Langs = ets:new(couch_query_server_langs, [set, private]), |
| LangLimits = ets:new(couch_query_server_lang_limits, [set, private]), |
| PidProcs = ets:new(couch_query_server_pid_langs, [set, private]), |
| LangProcs = ets:new(couch_query_server_procs, [set, private]), |
| |
| ProcTimeout = list_to_integer(couch_config:get( |
| "couchdb", "os_process_timeout", "5000")), |
| ReduceLimit = list_to_atom( |
| couch_config:get("query_server_config","reduce_limit","true")), |
| OsProcLimit = list_to_integer( |
| couch_config:get("query_server_config","os_process_limit","10")), |
| |
| % 'query_servers' specifies an OS command-line to execute. |
| lists:foreach(fun({Lang, Command}) -> |
| true = ets:insert(LangLimits, {?l2b(Lang), OsProcLimit, 0}), |
| true = ets:insert(Langs, {?l2b(Lang), |
| couch_os_process, start_link, [Command]}) |
| end, couch_config:get("query_servers")), |
| % 'native_query_servers' specifies a {Module, Func, Arg} tuple. |
| lists:foreach(fun({Lang, SpecStr}) -> |
| {ok, {Mod, Fun, SpecArg}} = couch_util:parse_term(SpecStr), |
| true = ets:insert(LangLimits, {?l2b(Lang), 0, 0}), % 0 means no limit |
| true = ets:insert(Langs, {?l2b(Lang), |
| Mod, Fun, SpecArg}) |
| end, couch_config:get("native_query_servers")), |
| |
| |
| process_flag(trap_exit, true), |
| {ok, #qserver{ |
| langs = Langs, % Keyed by language name, value is {Mod,Func,Arg} |
| pid_procs = PidProcs, % Keyed by PID, valus is a #proc record. |
| lang_procs = LangProcs, % Keyed by language name, value is a #proc record |
| lang_limits = LangLimits, % Keyed by language name, value is {Lang, Limit, Current} |
| config = {[{<<"reduce_limit">>, ReduceLimit},{<<"timeout">>, ProcTimeout}]} |
| }}. |
| |
| terminate(_Reason, #qserver{pid_procs=PidProcs}) -> |
| [couch_util:shutdown_sync(P) || {P,_} <- ets:tab2list(PidProcs)], |
| ok. |
| |
| handle_call({get_proc, DDoc1, DDocKey}, From, Server) -> |
| #doc{body = {Props}} = DDoc = couch_doc:with_ejson_body(DDoc1), |
| Lang = couch_util:get_value(<<"language">>, Props, <<"javascript">>), |
| case lang_proc(Lang, Server, fun(Procs) -> |
| % find a proc in the set that has the DDoc |
| proc_with_ddoc(DDoc, DDocKey, Procs) |
| end) of |
| {ok, Proc} -> |
| {reply, {ok, Proc, Server#qserver.config}, Server}; |
| wait -> |
| {noreply, add_to_waitlist({DDoc, DDocKey}, From, Server)}; |
| Error -> |
| {reply, Error, Server} |
| end; |
| handle_call({get_proc, Lang}, From, Server) -> |
| case lang_proc(Lang, Server, fun([P|_Procs]) -> |
| {ok, P} |
| end) of |
| {ok, Proc} -> |
| {reply, {ok, Proc, Server#qserver.config}, Server}; |
| wait -> |
| {noreply, add_to_waitlist({Lang}, From, Server)}; |
| Error -> |
| {reply, Error, Server} |
| end; |
| handle_call({unlink_proc, Pid}, _From, Server) -> |
| unlink(Pid), |
| {reply, ok, Server}; |
| handle_call({ret_proc, Proc}, _From, #qserver{ |
| pid_procs=PidProcs, |
| lang_procs=LangProcs}=Server) -> |
| % Along with max process limit, here we should check |
| % if we're over the limit and discard when we are. |
| case is_process_alive(Proc#proc.pid) of |
| true -> |
| add_value(PidProcs, Proc#proc.pid, Proc), |
| add_to_list(LangProcs, Proc#proc.lang, Proc), |
| link(Proc#proc.pid); |
| false -> |
| ok |
| end, |
| {reply, true, service_waitlist(Server)}. |
| |
| handle_cast(_Whatever, Server) -> |
| {noreply, Server}. |
| |
| handle_info({'EXIT', _, _}, Server) -> |
| {noreply, Server}; |
| handle_info({'DOWN', _, process, Pid, Status}, #qserver{ |
| pid_procs=PidProcs, |
| lang_procs=LangProcs, |
| lang_limits=LangLimits}=Server) -> |
| case ets:lookup(PidProcs, Pid) of |
| [{Pid, Proc}] -> |
| case Status of |
| normal -> ok; |
| _ -> ?LOG_DEBUG("Linked process died abnormally: ~p (reason: ~p)", [Pid, Status]) |
| end, |
| rem_value(PidProcs, Pid), |
| catch rem_from_list(LangProcs, Proc#proc.lang, Proc), |
| [{Lang, Lim, Current}] = ets:lookup(LangLimits, Proc#proc.lang), |
| true = ets:insert(LangLimits, {Lang, Lim, Current-1}), |
| {noreply, service_waitlist(Server)}; |
| [] -> |
| case Status of |
| normal -> |
| {noreply, Server}; |
| _ -> |
| {stop, Status, Server} |
| end |
| end. |
| |
| code_change(_OldVsn, State, _Extra) -> |
| {ok, State}. |
| |
| config_change("query_servers") -> |
| supervisor:terminate_child(couch_secondary_services, query_servers), |
| supervisor:restart_child(couch_secondary_services, query_servers); |
| config_change("native_query_servers") -> |
| supervisor:terminate_child(couch_secondary_services, query_servers), |
| supervisor:restart_child(couch_secondary_services, query_servers); |
| config_change("query_server_config") -> |
| supervisor:terminate_child(couch_secondary_services, query_servers), |
| supervisor:restart_child(couch_secondary_services, query_servers). |
| |
| % Private API |
| |
| add_to_waitlist(Info, From, #qserver{waitlist=Waitlist}=Server) -> |
| Server#qserver{waitlist=[{Info, From}|Waitlist]}. |
| |
| service_waitlist(#qserver{waitlist=[]}=Server) -> |
| Server; |
| service_waitlist(#qserver{waitlist=Waitlist}=Server) -> |
| [Oldest|RevWList] = lists:reverse(Waitlist), |
| case service_waiting(Oldest, Server) of |
| ok -> |
| Server#qserver{waitlist=lists:reverse(RevWList)}; |
| wait -> |
| Server#qserver{waitlist=Waitlist} |
| end. |
| |
| % todo get rid of duplication |
| service_waiting({{#doc{body={Props}}=DDoc, DDocKey}, From}, Server) -> |
| Lang = couch_util:get_value(<<"language">>, Props, <<"javascript">>), |
| case lang_proc(Lang, Server, fun(Procs) -> |
| % find a proc in the set that has the DDoc |
| proc_with_ddoc(DDoc, DDocKey, Procs) |
| end) of |
| {ok, Proc} -> |
| gen_server:reply(From, {ok, Proc, Server#qserver.config}), |
| ok; |
| wait -> % this should never happen |
| wait; |
| Error -> |
| gen_server:reply(From, Error), |
| ok |
| end; |
| service_waiting({{Lang}, From}, Server) -> |
| case lang_proc(Lang, Server, fun([P|_Procs]) -> |
| {ok, P} |
| end) of |
| {ok, Proc} -> |
| gen_server:reply(From, {ok, Proc, Server#qserver.config}), |
| ok; |
| wait -> % this should never happen |
| wait; |
| Error -> |
| gen_server:reply(From, Error), |
| ok |
| end. |
| |
| lang_proc(Lang, #qserver{ |
| langs=Langs, |
| pid_procs=PidProcs, |
| lang_procs=LangProcs, |
| lang_limits=LangLimits}, PickFun) -> |
| % Note to future self. Add max process limit. |
| case ets:lookup(LangProcs, Lang) of |
| [{Lang, [P|Procs]}] -> |
| {ok, Proc} = PickFun([P|Procs]), |
| rem_from_list(LangProcs, Lang, Proc), |
| {ok, Proc}; |
| _ -> |
| case (catch new_process(Langs, LangLimits, Lang)) of |
| {ok, Proc} -> |
| add_value(PidProcs, Proc#proc.pid, Proc), |
| PickFun([Proc]); |
| ErrorOrWait -> |
| ErrorOrWait |
| end |
| end. |
| |
| new_process(Langs, LangLimits, Lang) -> |
| [{Lang, Lim, Current}] = ets:lookup(LangLimits, Lang), |
| if (Lim == 0) or (Current < Lim) -> % Lim == 0 means no limit |
| % we are below the limit for our language, make a new one |
| case ets:lookup(Langs, Lang) of |
| [{Lang, Mod, Func, Arg}] -> |
| {ok, Pid} = apply(Mod, Func, Arg), |
| erlang:monitor(process, Pid), |
| true = ets:insert(LangLimits, {Lang, Lim, Current+1}), |
| {ok, #proc{lang=Lang, |
| pid=Pid, |
| % Called via proc_prompt, proc_set_timeout, and proc_stop |
| prompt_fun={Mod, prompt}, |
| set_timeout_fun={Mod, set_timeout}, |
| stop_fun={Mod, stop}}}; |
| _ -> |
| {unknown_query_language, Lang} |
| end; |
| true -> |
| wait |
| end. |
| |
| proc_with_ddoc(DDoc, DDocKey, LangProcs) -> |
| DDocProcs = lists:filter(fun(#proc{ddoc_keys=Keys}) -> |
| lists:any(fun(Key) -> |
| Key == DDocKey |
| end, Keys) |
| end, LangProcs), |
| case DDocProcs of |
| [DDocProc|_] -> |
| ?LOG_DEBUG("DDocProc found for DDocKey: ~p",[DDocKey]), |
| {ok, DDocProc}; |
| [] -> |
| [TeachProc|_] = LangProcs, |
| ?LOG_DEBUG("Teach ddoc to new proc ~p with DDocKey: ~p",[TeachProc, DDocKey]), |
| {ok, SmartProc} = teach_ddoc(DDoc, DDocKey, TeachProc), |
| {ok, SmartProc} |
| end. |
| |
| proc_prompt(Proc, Args) -> |
| case proc_prompt_raw(Proc, Args) of |
| {json, Json} -> |
| ?JSON_DECODE(Json); |
| EJson -> |
| EJson |
| end. |
| |
| proc_prompt_raw(#proc{prompt_fun = {Mod, Func}} = Proc, Args) -> |
| apply(Mod, Func, [Proc#proc.pid, Args]). |
| |
| raw_to_ejson({json, Json}) -> |
| ?JSON_DECODE(Json); |
| raw_to_ejson(EJson) -> |
| EJson. |
| |
| proc_stop(Proc) -> |
| {Mod, Func} = Proc#proc.stop_fun, |
| apply(Mod, Func, [Proc#proc.pid]). |
| |
| proc_set_timeout(Proc, Timeout) -> |
| {Mod, Func} = Proc#proc.set_timeout_fun, |
| apply(Mod, Func, [Proc#proc.pid, Timeout]). |
| |
| teach_ddoc(DDoc, {DDocId, _Rev}=DDocKey, #proc{ddoc_keys=Keys}=Proc) -> |
| % send ddoc over the wire |
| % we only share the rev with the client we know to update code |
| % but it only keeps the latest copy, per each ddoc, around. |
| true = proc_prompt(Proc, [<<"ddoc">>, <<"new">>, DDocId, couch_doc:to_json_obj(DDoc, [])]), |
| % we should remove any other ddocs keys for this docid |
| % because the query server overwrites without the rev |
| Keys2 = [{D,R} || {D,R} <- Keys, D /= DDocId], |
| % add ddoc to the proc |
| {ok, Proc#proc{ddoc_keys=[DDocKey|Keys2]}}. |
| |
| get_ddoc_process(#doc{} = DDoc, DDocKey) -> |
| % remove this case statement |
| case gen_server:call(couch_query_servers, {get_proc, DDoc, DDocKey}, infinity) of |
| {ok, Proc, {QueryConfig}} -> |
| % process knows the ddoc |
| case (catch proc_prompt(Proc, [<<"reset">>, {QueryConfig}])) of |
| true -> |
| proc_set_timeout(Proc, couch_util:get_value(<<"timeout">>, QueryConfig)), |
| link(Proc#proc.pid), |
| gen_server:call(couch_query_servers, {unlink_proc, Proc#proc.pid}, infinity), |
| Proc; |
| _ -> |
| catch proc_stop(Proc), |
| get_ddoc_process(DDoc, DDocKey) |
| end; |
| Error -> |
| throw(Error) |
| end. |
| |
| get_os_process(Lang) -> |
| case gen_server:call(couch_query_servers, {get_proc, Lang}, infinity) of |
| {ok, Proc, {QueryConfig}} -> |
| case (catch proc_prompt(Proc, [<<"reset">>, {QueryConfig}])) of |
| true -> |
| proc_set_timeout(Proc, couch_util:get_value(<<"timeout">>, QueryConfig)), |
| link(Proc#proc.pid), |
| gen_server:call(couch_query_servers, {unlink_proc, Proc#proc.pid}, infinity), |
| Proc; |
| _ -> |
| catch proc_stop(Proc), |
| get_os_process(Lang) |
| end; |
| Error -> |
| throw(Error) |
| end. |
| |
| ret_os_process(Proc) -> |
| true = gen_server:call(couch_query_servers, {ret_proc, Proc}, infinity), |
| catch unlink(Proc#proc.pid), |
| ok. |
| |
| add_value(Tid, Key, Value) -> |
| true = ets:insert(Tid, {Key, Value}). |
| |
| rem_value(Tid, Key) -> |
| true = ets:delete(Tid, Key). |
| |
| add_to_list(Tid, Key, Value) -> |
| case ets:lookup(Tid, Key) of |
| [{Key, Vals}] -> |
| true = ets:insert(Tid, {Key, [Value|Vals]}); |
| [] -> |
| true = ets:insert(Tid, {Key, [Value]}) |
| end. |
| |
| rem_from_list(Tid, Key, Value) when is_record(Value, proc)-> |
| Pid = Value#proc.pid, |
| case ets:lookup(Tid, Key) of |
| [{Key, Vals}] -> |
| % make a new values list that doesn't include the Value arg |
| NewValues = [Val || #proc{pid=P}=Val <- Vals, P /= Pid], |
| ets:insert(Tid, {Key, NewValues}); |
| [] -> ok |
| end; |
| rem_from_list(Tid, Key, Value) -> |
| case ets:lookup(Tid, Key) of |
| [{Key, Vals}] -> |
| % make a new values list that doesn't include the Value arg |
| NewValues = [Val || Val <- Vals, Val /= Value], |
| ets:insert(Tid, {Key, NewValues}); |
| [] -> ok |
| end. |
