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

4 files changed, 39 insertions, 54 deletions

diff --git a/share/server/filter.js b/share/server/filter.js
index 23536db0..a683146a 100644
--- a/share/server/filter.js
+++ b/share/server/filter.js
@@ -11,11 +11,13 @@
 // the License.
 
 var Filter = {
-  filter : function(docs, req, userCtx) {
+  filter : function(funSrc, docs, req, userCtx) {
+    var filterFun = compileFunction(funSrc);
+    
     var results = [];
     try {
       for (var i=0; i < docs.length; i++) {
-        results.push((funs[0](docs[i], req, userCtx) && true) || false);
+        results.push((filterFun(docs[i], req, userCtx) && true) || false);
       };
       respond([true, results]);
     } catch (error) {
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([]) ->
 
diff --git a/test/view_server/query_server_spec.rb b/test/view_server/query_server_spec.rb
index 4697efca..0d443567 100644
--- a/test/view_server/query_server_spec.rb
+++ b/test/view_server/query_server_spec.rb
@@ -606,10 +606,9 @@ describe "query server normal case" do
     before(:all) do
       @fun = functions["filter-basic"][LANGUAGE]
       @qs.reset!
-      @qs.add_fun(@fun).should == true
     end
     it "should only return true for good docs" do
-      @qs.run(["filter", [{"key"=>"bam", "good" => true}, {"foo" => "bar"}, {"good" => true}], {"req" => "foo"}]).
+      @qs.run(["filter", @fun, [{"key"=>"bam", "good" => true}, {"foo" => "bar"}, {"good" => true}], {"req" => "foo"}]).
         should ==  [true, [true, false, true]]
     end
   end