From 60a6b3e7ffa209da31d563beae33726412318957 Mon Sep 17 00:00:00 2001 From: John Christopher Anderson Date: Wed, 2 Dec 2009 23:05:28 +0000 Subject: removed add_fun optimization for filters, this provides better concurrency at the cost of some function transfer/eval overhead, which will be optimized in a future patch. git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@886334 13f79535-47bb-0310-9956-ffa450edef68 --- src/couchdb/couch_httpd_db.erl | 61 ++++++++++++++++--------------------- src/couchdb/couch_query_servers.erl | 23 +++++--------- 2 files changed, 34 insertions(+), 50 deletions(-) (limited to 'src/couchdb') diff --git a/src/couchdb/couch_httpd_db.erl b/src/couchdb/couch_httpd_db.erl index 22d18d2f..e1c778a9 100644 --- a/src/couchdb/couch_httpd_db.erl +++ b/src/couchdb/couch_httpd_db.erl @@ -82,7 +82,7 @@ start_sending_changes(Resp, _Else) -> send_chunk(Resp, "{\"results\":[\n"). handle_changes_req(#httpd{method='GET',path_parts=[DbName|_]}=Req, Db) -> - {FilterFun, EndFilterFun} = make_filter_funs(Req, Db), + FilterFun = make_filter_fun(Req, Db), {ok, Info} = couch_db:get_db_info(Db), Seq = proplists:get_value(update_seq, Info), {Dir, StartSeq} = case couch_httpd:qs_value(Req, "descending", "false") of @@ -110,7 +110,7 @@ handle_changes_req(#httpd{method='GET',path_parts=[DbName|_]}=Req, Db) -> {httpd, clients_requesting_changes}), try keep_sending_changes(Req, Resp, Db, StartSeq, <<"">>, Timeout, - TimeoutFun, ResponseType, Limit, FilterFun, EndFilterFun) + TimeoutFun, ResponseType, Limit, FilterFun) after couch_db_update_notifier:stop(Notify), get_rest_db_updated() % clean out any remaining update messages @@ -123,7 +123,7 @@ handle_changes_req(#httpd{method='GET',path_parts=[DbName|_]}=Req, Db) -> start_sending_changes(Resp, ResponseType), {ok, {_, LastSeq, _Prepend, _, _, _, _, _}} = send_changes(Req, Resp, Db, Dir, StartSeq, <<"">>, "normal", - Limit, FilterFun, EndFilterFun), + Limit, FilterFun), end_sending_changes(Resp, LastSeq, ResponseType) end) end; @@ -154,9 +154,9 @@ end_sending_changes(Resp, EndSeq, _Else) -> end_json_response(Resp). keep_sending_changes(#httpd{user_ctx=UserCtx,path_parts=[DbName|_]}=Req, Resp, - Db, StartSeq, Prepend, Timeout, TimeoutFun, ResponseType, Limit, Filter, End) -> + Db, StartSeq, Prepend, Timeout, TimeoutFun, ResponseType, Limit, Filter) -> {ok, {_, EndSeq, Prepend2, _, _, _, NewLimit, _}} = send_changes(Req, Resp, Db, fwd, StartSeq, - Prepend, ResponseType, Limit, Filter, End), + Prepend, ResponseType, Limit, Filter), couch_db:close(Db), if Limit > NewLimit, ResponseType == "longpoll" -> @@ -167,7 +167,7 @@ keep_sending_changes(#httpd{user_ctx=UserCtx,path_parts=[DbName|_]}=Req, Resp, case couch_db:open(DbName, [{user_ctx, UserCtx}]) of {ok, Db2} -> keep_sending_changes(Req, Resp, Db2, EndSeq, Prepend2, Timeout, - TimeoutFun, ResponseType, NewLimit, Filter, End); + TimeoutFun, ResponseType, NewLimit, Filter); _Else -> end_sending_changes(Resp, EndSeq, ResponseType) end; @@ -178,7 +178,7 @@ keep_sending_changes(#httpd{user_ctx=UserCtx,path_parts=[DbName|_]}=Req, Resp, changes_enumerator(DocInfos, {Db, _, _, FilterFun, Resp, "continuous", Limit, IncludeDocs}) -> [#doc_info{id=Id, high_seq=Seq, revs=[#rev_info{deleted=Del,rev=Rev}|_]}|_] = DocInfos, - Results0 = [FilterFun(DocInfo) || DocInfo <- DocInfos], + Results0 = FilterFun(DocInfos), Results = [Result || Result <- Results0, Result /= null], Go = if Limit =< 1 -> stop; true -> ok end, case Results of @@ -191,7 +191,7 @@ changes_enumerator(DocInfos, {Db, _, _, FilterFun, Resp, "continuous", Limit, In end; changes_enumerator(DocInfos, {Db, _, Prepend, FilterFun, Resp, _, Limit, IncludeDocs}) -> [#doc_info{id=Id, high_seq=Seq, revs=[#rev_info{deleted=Del,rev=Rev}|_]}|_] = DocInfos, - Results0 = [FilterFun(DocInfo) || DocInfo <- DocInfos], + Results0 = FilterFun(DocInfos), Results = [Result || Result <- Results0, Result /= null], Go = if Limit =< 1 -> stop; true -> ok end, case Results of @@ -212,27 +212,24 @@ changes_row(_, Seq, Id, Del, Results, _, false) -> deleted_item(true) -> [{deleted,true}]; deleted_item(_) -> []. -send_changes(Req, Resp, Db, Dir, StartSeq, Prepend, ResponseType, Limit, FilterFun, End) -> +send_changes(Req, Resp, Db, Dir, StartSeq, Prepend, ResponseType, Limit, FilterFun) -> Style = list_to_existing_atom( couch_httpd:qs_value(Req, "style", "main_only")), IncludeDocs = list_to_existing_atom( couch_httpd:qs_value(Req, "include_docs", "false")), - try - couch_db:changes_since(Db, Style, StartSeq, fun changes_enumerator/2, - [{dir, Dir}], {Db, StartSeq, Prepend, FilterFun, Resp, ResponseType, Limit, IncludeDocs}) - after - End() - end. + couch_db:changes_since(Db, Style, StartSeq, fun changes_enumerator/2, + [{dir, Dir}], {Db, StartSeq, Prepend, FilterFun, Resp, ResponseType, Limit, IncludeDocs}). -make_filter_funs(Req, Db) -> +make_filter_fun(Req, Db) -> Filter = couch_httpd:qs_value(Req, "filter", ""), case [list_to_binary(couch_httpd:unquote(Part)) || Part <- string:tokens(Filter, "/")] of [] -> - {fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) -> - {[{rev, couch_doc:rev_to_str(Rev)}]} - end, - fun() -> ok end}; + fun(DocInfos) -> + % doing this as a batch is more efficient for external filters + [{[{rev, couch_doc:rev_to_str(Rev)}]} || + #doc_info{revs=[#rev_info{rev=Rev}|_]} <- DocInfos] + end; [DName, FName] -> DesignId = <<"_design/", DName/binary>>, case couch_db:open_doc(Db, DesignId) of @@ -244,21 +241,15 @@ make_filter_funs(Req, Db) -> throw({bad_request, "invalid filter function"}) end, Lang = proplists:get_value(<<"language">>, Props, <<"javascript">>), - {ok, Pid} = couch_query_servers:start_filter(Lang, FilterSrc), - FilterFun = fun(DInfo = #doc_info{revs=[#rev_info{rev=Rev}|_]}) -> - {ok, Doc} = couch_db:open_doc(Db, DInfo, [deleted]), - {ok, Pass} = couch_query_servers:filter_doc(Pid, Doc, Req, Db), - case Pass of - true -> - {[{rev, couch_doc:rev_to_str(Rev)}]}; - false -> - null - end - end, - EndFilterFun = fun() -> - couch_query_servers:end_filter(Pid) - end, - {FilterFun, EndFilterFun}; + fun(DocInfos) -> + Docs = [Doc || {ok, Doc} <- [ + {ok, Doc} = couch_db:open_doc(Db, DInfo, [deleted]) + || DInfo <- DocInfos]], + {ok, Passes} = couch_query_servers:filter_docs(Lang, FilterSrc, Docs, Req, Db), + [{[{rev, couch_doc:rev_to_str(Rev)}]} + || #doc_info{revs=[#rev_info{rev=Rev}|_]} <- DocInfos, + Pass <- Passes, Pass == true] + end; _Error -> throw({bad_request, "invalid design doc"}) end; diff --git a/src/couchdb/couch_query_servers.erl b/src/couchdb/couch_query_servers.erl index 9ab4ff5a..4ac56727 100644 --- a/src/couchdb/couch_query_servers.erl +++ b/src/couchdb/couch_query_servers.erl @@ -20,7 +20,7 @@ -export([reduce/3, rereduce/3,validate_doc_update/5]). -export([render_doc_show/6, render_doc_update/6, start_view_list/2, render_list_head/4, render_list_row/4, render_list_tail/1]). --export([start_filter/2, filter_doc/4, end_filter/1]). +-export([filter_docs/5]). % -export([test/0]). -include("couch_db.hrl"). @@ -231,22 +231,15 @@ render_list_tail(Proc) -> ok = ret_os_process(Proc), JsonResp. -start_filter(Lang, FilterSrc) -> - Proc = get_os_process(Lang), - true = proc_prompt(Proc, [<<"add_fun">>, FilterSrc]), - {ok, Proc}. - -filter_doc(Proc, Doc, Req, Db) -> +filter_docs(Lang, Src, Docs, Req, Db) -> JsonReq = couch_httpd_external:json_req_obj(Req, Db), - JsonDoc = couch_doc:to_json_obj(Doc, [revs]), + JsonDocs = [couch_doc:to_json_obj(Doc, [revs]) || Doc <- Docs], JsonCtx = couch_util:json_user_ctx(Db), - [true, [Pass]] = proc_prompt(Proc, - [<<"filter">>, [JsonDoc], JsonReq, JsonCtx]), - {ok, Pass}. - -end_filter(Proc) -> - ok = ret_os_process(Proc). - + Proc = get_os_process(Lang), + [true, Passes] = proc_prompt(Proc, + [<<"filter">>, Src, JsonDocs, JsonReq, JsonCtx]), + ret_os_process(Proc), + {ok, Passes}. init([]) -> -- cgit v1.2.3