diff options
| -rw-r--r-- | src/couchdb/Makefile.am | 2 | ||||
| -rw-r--r-- | src/couchdb/couch_db_update_notifier.erl | 21 | ||||
| -rw-r--r-- | src/couchdb/couch_os_process.erl | 161 | ||||
| -rw-r--r-- | src/couchdb/couch_query_servers.erl | 248 |
4 files changed, 277 insertions, 155 deletions
diff --git a/src/couchdb/Makefile.am b/src/couchdb/Makefile.am index 9db1539b..76d3f13e 100644 --- a/src/couchdb/Makefile.am +++ b/src/couchdb/Makefile.am @@ -55,6 +55,7 @@ source_files = \ couch_httpd_misc_handlers.erl \ couch_key_tree.erl \ couch_log.erl \ + couch_os_process.erl \ couch_query_servers.erl \ couch_rep.erl \ couch_server.erl \ @@ -85,6 +86,7 @@ compiled_files = \ couch_httpd_misc_handlers.beam \ couch_key_tree.beam \ couch_log.beam \ + couch_os_process.beam \ couch_query_servers.beam \ couch_rep.beam \ couch_server.beam \ diff --git a/src/couchdb/couch_db_update_notifier.erl b/src/couchdb/couch_db_update_notifier.erl index f2f0342c..68948afb 100644 --- a/src/couchdb/couch_db_update_notifier.erl +++ b/src/couchdb/couch_db_update_notifier.erl @@ -37,12 +37,14 @@ stop(Pid) -> couch_event_sup:stop(Pid). init(Exec) when is_list(Exec) -> % an exe - Port = open_port({spawn, Exec}, [stream, exit_status, hide]), - {ok, Port}; + {ok, couch_os_process:start_link(Exec, [], [stream, exit_status, hide])}; init(Else) -> {ok, Else}. -terminate(_Reason, _Port) -> +terminate(_Reason, Pid) when is_pid(Pid) -> + couch_os_process:stop(Pid), + ok; +terminate(_Reason, _State) -> ok. handle_event(Event, Fun) when is_function(Fun, 1) -> @@ -51,16 +53,17 @@ handle_event(Event, Fun) when is_function(Fun, 1) -> handle_event(Event, {Fun, FunAcc}) -> FunAcc2 = Fun(Event, FunAcc), {ok, {Fun, FunAcc2}}; -handle_event({EventAtom, DbName}, Port) -> +handle_event({EventAtom, DbName}, Pid) -> Obj = {[{type, list_to_binary(atom_to_list(EventAtom))}, {db, DbName}]}, - true = port_command(Port, ?JSON_ENCODE(Obj) ++ "\n"), - {ok, Port}. + true = couch_os_process:write(Pid, Obj), + {ok, Pid}. handle_call(_Request, State) -> - {ok, ok, State}. + {reply, ok, State}. -handle_info({'EXIT', _, _Reason}, _Port) -> - remove_handler. +handle_info({'EXIT', Pid, Reason}, Pid) -> + ?LOG_ERROR("Update notification process ~p died: ~p", [Pid, Reason]), + {stop, nil}. code_change(_OldVsn, State, _Extra) -> {ok, State}. diff --git a/src/couchdb/couch_os_process.erl b/src/couchdb/couch_os_process.erl new file mode 100644 index 00000000..c0b98946 --- /dev/null +++ b/src/couchdb/couch_os_process.erl @@ -0,0 +1,161 @@ +% 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_os_process). +-behaviour(gen_server). + +-export([start_link/1, start_link/2, start_link/3, stop/1]). +-export([set_timeout/2, write/2, read/1, prompt/2, async/3]). +-export([writeline/2, readline/1, writejson/2, readjson/1]). +-export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2, code_change/3]). + +-include("couch_db.hrl"). + +-define(PORT_OPTIONS, [stream, {line, 1024}, binary, exit_status, hide]). + +-record(os_proc, + {command, + port, + writer, + reader, + timeout + }). + +start_link(Command) -> + start_link(Command, []). +start_link(Command, Options) -> + start_link(Command, Options, ?PORT_OPTIONS). +start_link(Command, Options, PortOptions) -> + gen_server:start_link(couch_os_process, [Command, Options, PortOptions], []). + +stop(Pid) -> + gen_server:cast(Pid, stop). + +% Read/Write API +set_timeout(Pid, TimeOut) when is_integer(TimeOut) -> + ok = gen_server:call(Pid, {set_timeout, TimeOut}). + +write(Pid, Data) -> + gen_server:call(Pid, {write, Data}). + +read(Pid) -> + gen_server:call(Pid, read). + +prompt(Pid, Data) -> + gen_server:call(Pid, {prompt, Data}). + +async(Pid, Data, CallBack) -> + gen_server:cast(Pid, {async, Data, CallBack}). + +% Utility functions for reading and writing +% in custom functions +writeline(OsProc, Data) when is_record(OsProc, os_proc) -> + port_command(OsProc#os_proc.port, Data ++ "\n"). + +readline(OsProc) when is_record(OsProc, os_proc) -> + readline(OsProc, []). +readline(OsProc, Acc) when is_record(OsProc, os_proc) -> + #os_proc{port=Port} = OsProc, + receive + {Port, {data, {noeol, Data}}} -> + readline(OsProc, [Data|Acc]); + {Port, {data, {eol, Data}}} -> + lists:reverse(Acc, Data); + {Port, Err} -> + catch port_close(Port), + throw({os_process_error, Err}) + after OsProc#os_proc.timeout -> + catch port_close(Port), + throw({os_process_error, "OS process timed out."}) + end. + +% Standard JSON functions +writejson(OsProc, Data) when is_record(OsProc, os_proc) -> + true = writeline(OsProc, ?JSON_ENCODE(Data)). + +readjson(OsProc) when is_record(OsProc, os_proc) -> + Line = readline(OsProc), + case ?JSON_DECODE(Line) of + {[{<<"log">>,Msg}]} when is_binary(Msg) -> + % we got a message to log. Log it and continue + ?LOG_INFO("OS Process Log Message: ~s", [Msg]), + readjson(OsProc); + {[{<<"error">>, Id}, {<<"reason">>, Reason}]} -> + throw({list_to_atom(binary_to_list(Id)),Reason}); + {[{<<"reason">>, Reason}, {<<"error">>, Id}]} -> + throw({list_to_atom(binary_to_list(Id)),Reason}); + Result -> + Result + end. + +% gen_server API +init([Command, Options, PortOptions]) -> + BaseProc = #os_proc{ + command=Command, + port=open_port({spawn, Command}, PortOptions), + writer=fun writejson/2, + reader=fun readjson/1, + timeout=5000 + }, + OsProc = + lists:foldl(fun(Opt, Proc) -> + case Opt of + {writer, Writer} when is_function(Writer) -> + Proc#os_proc{writer=Writer}; + {reader, Reader} when is_function(Reader) -> + Proc#os_proc{reader=Reader}; + {timeout, TimeOut} when is_integer(TimeOut) -> + Proc#os_proc{timeout=TimeOut} + end + end, BaseProc, Options), + {ok, OsProc}. + +terminate(_Reason, #os_proc{port=nil}) -> + ok; +terminate(_Reason, #os_proc{port=Port}) -> + catch port_close(Port), + ok. + +handle_call({set_timeout, TimeOut}, _From, OsProc) -> + {reply, ok, OsProc#os_proc{timeout=TimeOut}}; +handle_call({write, Data}, _From, OsProc) -> + Writer = OsProc#os_proc.writer, + {reply, Writer(OsProc, Data), OsProc}; +handle_call(read, _From, OsProc) -> + Reader = OsProc#os_proc.reader, + {reply, Reader(OsProc), OsProc}; +handle_call({prompt, Data}, _From, OsProc) -> + #os_proc{writer=Writer, reader=Reader} = OsProc, + Writer(OsProc, Data), + {reply, Reader(OsProc), OsProc}. + +handle_cast({async, Data, CallBack}, OsProc) -> + #os_proc{writer=Writer, reader=Reader} = OsProc, + Writer(OsProc, Data), + CallBack(Reader(OsProc)), + {noreply, OsProc}; +handle_cast(stop, OsProc) -> + {stop, normal, OsProc}; +handle_cast(Msg, OsProc) -> + ?LOG_DEBUG("OS Proc: Unknown cast: ~p", [Msg]), + {noreply, OsProc}. + +handle_info({Port, {exit_status, Status}}, #os_proc{port=Port}=OsProc) -> + ?LOG_ERROR("OS Process died with status: ~p", [Status]), + {stop, error, OsProc}; +handle_info(Msg, OsProc) -> + ?LOG_DEBUG("OS Proc: Unknown info: ~p", [Msg]), + {noreply, OsProc}. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + diff --git a/src/couchdb/couch_query_servers.erl b/src/couchdb/couch_query_servers.erl index 8465a632..b694d5e3 100644 --- a/src/couchdb/couch_query_servers.erl +++ b/src/couchdb/couch_query_servers.erl @@ -28,71 +28,20 @@ start_link() -> stop() -> exit(whereis(couch_query_servers), close). -readline(Port) -> - readline(Port, []). - -readline(Port, Acc) -> - case get(query_server_timeout) of - undefined -> - Timeout = list_to_integer(couch_config:get( - "couchdb", "view_timeout", "5000")), - put(query_server_timeout, Timeout); - Timeout -> ok - end, - receive - {Port, {data, {noeol, Data}}} -> - readline(Port, [Data|Acc]); - {Port, {data, {eol, Data}}} -> - lists:reverse(Acc, Data); - {Port, Err} -> - catch port_close(Port), - throw({external_process_error, Err}) - after Timeout -> - catch port_close(Port), - throw({external_process_error, "External process timed out"}) - end. - -read_json(Port) -> - Line = readline(Port), - case ?JSON_DECODE(Line) of - {[{<<"log">>,Msg}]} when is_binary(Msg) -> - % we got a message to log. Log it and continue - ?LOG_INFO("Query Server Log Message: ~s", [Msg]), - read_json(Port); - Else -> - Else - end. - -% send command and get a response. -prompt(Port, Json) -> - Bin = iolist_to_binary([?JSON_ENCODE(Json) , "\n"]), - true = port_command(Port, Bin), - case read_json(Port) of - {[{<<"error">>, Id}, {<<"reason">>, Reason}]} -> - throw({Id,Reason}); - {[{<<"reason">>, Reason}, {<<"error">>, Id}]} -> - throw({Id,Reason}); - Result -> - Result - end. - - start_doc_map(Lang, Functions) -> - Port = get_linked_port(Lang), - % send the functions as json strings + Pid = get_os_process(Lang), lists:foreach(fun(FunctionSource) -> - true = prompt(Port, [<<"add_fun">>, FunctionSource]) - end, - Functions), - {ok, {Lang, Port}}. + true = couch_os_process:prompt(Pid, [<<"add_fun">>, FunctionSource]) + end, Functions), + {ok, {Lang, Pid}}. -map_docs({_Lang, Port}, Docs) -> +map_docs({_Lang, Pid}, Docs) -> % send the documents Results = lists:map( fun(Doc) -> Json = couch_doc:to_json_obj(Doc, []), - FunsResults = prompt(Port, [<<"map_doc">>, Json]), + FunsResults = couch_os_process:prompt(Pid, [<<"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: @@ -111,30 +60,8 @@ map_docs({_Lang, Port}, Docs) -> stop_doc_map(nil) -> ok; -stop_doc_map({Lang, Port}) -> - return_linked_port(Lang, Port). - -get_linked_port(Lang) -> - case gen_server:call(couch_query_servers, {get_port, Lang}) of - {ok, Port0} -> - link(Port0), - true = prompt(Port0, [<<"reset">>]), - Port0; - {empty, Cmd} -> - ?LOG_INFO("Spawning new ~s instance.", [Lang]), - open_port({spawn, Cmd}, [stream, - {line, 1000}, - binary, - exit_status, - hide]); - Error -> - throw(Error) - end. - -return_linked_port(Lang, Port) -> - ok = gen_server:call(couch_query_servers, {return_port, {Lang, Port}}), - true = unlink(Port), - ok. +stop_doc_map({Lang, Pid}) -> + ok = ret_os_process(Lang, Pid). group_reductions_results([]) -> []; @@ -153,29 +80,29 @@ group_reductions_results(List) -> rereduce(_Lang, [], _ReducedValues) -> {ok, []}; rereduce(Lang, RedSrcs, ReducedValues) -> - Port = get_linked_port(Lang), + Pid = get_os_process(Lang), Grouped = group_reductions_results(ReducedValues), Results = lists:zipwith( fun(FunSrc, Values) -> [true, [Result]] = - prompt(Port, [<<"rereduce">>, [FunSrc], Values]), + couch_os_process:prompt(Pid, [<<"rereduce">>, [FunSrc], Values]), Result end, RedSrcs, Grouped), - return_linked_port(Lang, Port), + ok = ret_os_process(Lang, Pid), {ok, Results}. reduce(_Lang, [], _KVs) -> {ok, []}; reduce(Lang, RedSrcs, KVs) -> - Port = get_linked_port(Lang), - [true, Results] = prompt(Port, + Pid = get_os_process(Lang), + [true, Results] = couch_os_process:prompt(Pid, [<<"reduce">>, RedSrcs, KVs]), - return_linked_port(Lang, Port), + ok = ret_os_process(Lang, Pid), {ok, Results}. validate_doc_update(Lang, FunSrc, EditDoc, DiskDoc, Ctx) -> - Port = get_linked_port(Lang), + Pid = get_os_process(Lang), JsonEditDoc = couch_doc:to_json_obj(EditDoc, [revs]), JsonDiskDoc = if DiskDoc == nil -> @@ -183,7 +110,7 @@ validate_doc_update(Lang, FunSrc, EditDoc, DiskDoc, Ctx) -> true -> couch_doc:to_json_obj(DiskDoc, [revs]) end, - try prompt(Port, + try couch_os_process:prompt(Pid, [<<"validate">>, FunSrc, JsonEditDoc, JsonDiskDoc, Ctx]) of 1 -> ok; @@ -192,7 +119,7 @@ validate_doc_update(Lang, FunSrc, EditDoc, DiskDoc, Ctx) -> {[{<<"unauthorized">>, Message}]} -> throw({unauthorized, Message}) after - return_linked_port(Lang, Port) + ok = ret_os_process(Lang, Pid) end. init([]) -> @@ -206,72 +133,101 @@ init([]) -> fun("query_servers" ++ _, _) -> ?MODULE:stop() end), - - QueryServers = couch_config:get("query_servers"), - QueryServers2 = - [{list_to_binary(Lang), Path} || {Lang, Path} <- QueryServers], - - {ok, {QueryServers2, []}}. + + Langs = ets:new(couch_query_server_langs, [set, private]), + PidLangs = ets:new(couch_query_server_pid_langs, [set, private]), + Pids = ets:new(couch_query_server_procs, [set, private]), + InUse = ets:new(couch_query_server_used, [set, private]), + lists:foreach(fun({Lang, Command}) -> + true = ets:insert(Langs, {?l2b(Lang), Command}) + end, couch_config:get("query_servers")), + {ok, {Langs, PidLangs, Pids, InUse}}. terminate(_Reason, _Server) -> ok. -handle_call({get_port, Lang}, {FromPid, _}, {QueryServerList, LangPorts}) -> - case proplists:get_value(Lang, LangPorts) of - undefined -> - case proplists:get_value(Lang, QueryServerList) of - undefined -> % not a supported language - {reply, {query_language_unknown, Lang}, {QueryServerList, LangPorts}}; - ServerCmd -> - {reply, {empty, ServerCmd}, {QueryServerList, LangPorts}} - end; - Port -> - Result = - case catch port_connect(Port, FromPid) of - true -> - true = unlink(Port), - {ok, Port}; - Error -> - catch port_close(Port), - Error - end, - {reply, Result, {QueryServerList, LangPorts -- [{Lang,Port}]}} - end; -handle_call({return_port, {Lang, Port}}, _From, {QueryServerList, LangPorts}) -> - case catch port_connect(Port, self()) of - true -> - {reply, ok, {QueryServerList, [{Lang, Port} | LangPorts]}}; +handle_call({get_proc, Lang}, _From, {Langs, PidLangs, Pids, InUse}=Server) -> + % Note to future self. Add max process limit. + case ets:lookup(Pids, Lang) of + [{Lang, [Pid|_]}] -> + add_value(PidLangs, Pid, Lang), + rem_from_list(Pids, Lang, Pid), + add_to_list(InUse, Lang, Pid), + true = couch_os_process:prompt(Pid, [<<"reset">>]), + {reply, Pid, Server}; _ -> - catch port_close(Port), - {reply, ok, {QueryServerList, LangPorts}} - end. - -handle_cast(_Whatever, {Cmd, Ports}) -> - {noreply, {Cmd, Ports}}. - -handle_info({Port, {exit_status, Status}}, {QueryServerList, LangPorts}) -> - case lists:keysearch(Port, 2, LangPorts) of - {value, {Lang, _}} -> + {ok, Pid} = new_process(Langs, Lang), + add_to_list(InUse, Lang, Pid), + {reply, Pid, Server} + end; +handle_call({ret_proc, Lang, Pid}, _From, {_, _, Pids, InUse}=Server) -> + % Along with max process limit, here we should check + % if we're over the limit and discard when we are. + add_to_list(Pids, Lang, Pid), + rem_from_list(InUse, Lang, Pid), + {reply, true, Server}. + +handle_cast(_Whatever, Server) -> + {noreply, Server}. + +handle_info({'EXIT', Pid, Status}, {Langs, PidLangs, Pids, InUse}) -> + case ets:lookup(PidLangs, Pid) of + [{Pid, Lang}] -> case Status of - 0 -> ok; - _ -> ?LOG_ERROR("Abnormal shutdown of ~s query server process (exit_status: ~w).", [Lang, Status]) + normal -> ok; + _ -> ?LOG_DEBUG("Linked process died abnromally: ~p (reason: ~p)", [Pid, Status]) end, - {noreply, {QueryServerList, lists:keydelete(Port, 2, LangPorts)}}; - _ -> - ?LOG_ERROR("Unknown linked port/process crash: ~p", [Port]) + {ok, { + Langs, + rem_value(PidLangs, Pid), + rem_from_list(Pids, Lang, Pid), + rem_from_list(InUse, Lang, Pid) + }}; + [] -> + ?LOG_DEBUG("Unknown linked process died: ~p (reason: ~p)", [Pid, Status]), + {ok, {Langs, PidLangs, Pids, InUse}} end. code_change(_OldVsn, State, _Extra) -> {ok, State}. -% test() -> -% test("../js/js -f main.js"). -% -% test(Cmd) -> -% start_link(Cmd), -% {ok, DocMap} = start_doc_map(<<"javascript">>, [<<"function(doc) {if (doc[0] == 'a') return doc[1];}">>]), -% {ok, Results} = map_docs(DocMap, [#doc{body={"a", "b"}}, #doc{body={"c", "d"}},#doc{body={"a", "c"}}]), -% io:format("Results: ~w~n", [Results]), -% stop_doc_map(DocMap), -% ok. +% Private API + +new_process(Langs, Lang) -> + Proc = + case ets:lookup(Langs, Lang) of + [{Lang, Command}] -> + couch_os_process:start_link(Command); + _ -> + throw({unknown_query_language, Lang}) + end, + Proc. + +get_os_process(Lang) -> + gen_server:call(couch_query_servers, {get_proc, Lang}). + +ret_os_process(Lang, Pid) -> + true = gen_server:call(couch_query_servers, {ret_proc, Lang, Pid}), + ok. + +add_value(Tid, Key, Value) -> + true = ets:insert(Tid, {Key, Value}). + +rem_value(Tid, Key) -> + true = ets:insert(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) -> + case ets:lookup(Tid, Key) of + [{Key, Vals}] -> + ets:insert(Tid, {Key, [Val || Val <- Vals, Val /= Value]}); + [] -> ok + end. |