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

2 files changed, 34 insertions, 50 deletions

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([]) ->