You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2009/07/20 20:46:30 UTC
svn commit: r795953 - /couchdb/trunk/src/couchdb/couch_httpd_db.erl
Author: damien
Date: Mon Jul 20 18:46:30 2009
New Revision: 795953
URL: http://svn.apache.org/viewvc?rev=795953&view=rev
Log:
Optimize filtered _changes to use an OS process only when actively filtering changes with the continuous=true option.
Modified:
couchdb/trunk/src/couchdb/couch_httpd_db.erl
Modified: couchdb/trunk/src/couchdb/couch_httpd_db.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_httpd_db.erl?rev=795953&r1=795952&r2=795953&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_httpd_db.erl (original)
+++ couchdb/trunk/src/couchdb/couch_httpd_db.erl Mon Jul 20 18:46:30 2009
@@ -66,38 +66,32 @@
handle_changes_req(#httpd{method='GET',path_parts=[DbName|_]}=Req, Db) ->
StartSeq = list_to_integer(couch_httpd:qs_value(Req, "since", "0")),
- Filter = couch_httpd:qs_value(Req, "filter", nil),
- {ok, FilterFun, EndFilterFun} = make_filter_funs(Req, Db, Filter),
- try
- {ok, Resp} = start_json_response(Req, 200),
- send_chunk(Resp, "{\"results\":[\n"),
- case couch_httpd:qs_value(Req, "continuous", "false") of
- "true" ->
- Self = self(),
- {ok, Notify} = couch_db_update_notifier:start_link(
- fun({_, DbName0}) when DbName0 == DbName ->
- Self ! db_updated;
- (_) ->
- ok
- end),
- {Timeout, TimeoutFun} = get_changes_timeout(Req, Resp),
- couch_stats_collector:track_process_count(Self,
- {httpd, clients_requesting_changes}),
- try
- keep_sending_changes(Req, Resp, Db, StartSeq, <<"">>, Timeout, TimeoutFun, FilterFun)
- after
- couch_db_update_notifier:stop(Notify),
- get_rest_db_updated() % clean out any remaining update messages
- end;
+ {ok, Resp} = start_json_response(Req, 200),
+ send_chunk(Resp, "{\"results\":[\n"),
+ case couch_httpd:qs_value(Req, "continuous", "false") of
+ "true" ->
+ Self = self(),
+ {ok, Notify} = couch_db_update_notifier:start_link(
+ fun({_, DbName0}) when DbName0 == DbName ->
+ Self ! db_updated;
+ (_) ->
+ ok
+ end),
+ {Timeout, TimeoutFun} = get_changes_timeout(Req, Resp),
+ couch_stats_collector:track_process_count(Self,
+ {httpd, clients_requesting_changes}),
+ try
+ keep_sending_changes(Req, Resp, Db, StartSeq, <<"">>, Timeout, TimeoutFun)
+ after
+ couch_db_update_notifier:stop(Notify),
+ get_rest_db_updated() % clean out any remaining update messages
+ end;
- "false" ->
- {ok, {LastSeq, _Prepend}} =
- send_changes(Req, Resp, Db, StartSeq, <<"">>, FilterFun),
- send_chunk(Resp, io_lib:format("\n],\n\"last_seq\":~w}\n", [LastSeq])),
- send_chunk(Resp, "")
- end
- after
- EndFilterFun()
+ "false" ->
+ {ok, {LastSeq, _Prepend}} =
+ send_changes(Req, Resp, Db, StartSeq, <<"">>),
+ send_chunk(Resp, io_lib:format("\n],\n\"last_seq\":~w}\n", [LastSeq])),
+ send_chunk(Resp, "")
end;
handle_changes_req(#httpd{path_parts=[_,<<"_changes">>]}=Req, _Db) ->
@@ -118,22 +112,24 @@
after 0 -> updated
end.
-keep_sending_changes(#httpd{user_ctx=UserCtx,path_parts=[DbName|_]}=Req, Resp, Db, StartSeq, Prepend, Timeout, TimeoutFun, FilterFun) ->
- {ok, {EndSeq, Prepend2}} = send_changes(Req, Resp, Db, StartSeq, Prepend, FilterFun),
+keep_sending_changes(#httpd{user_ctx=UserCtx,path_parts=[DbName|_]}=Req, Resp, Db, StartSeq, Prepend, Timeout, TimeoutFun) ->
+ {ok, {EndSeq, Prepend2}} = send_changes(Req, Resp, Db, StartSeq, Prepend),
couch_db:close(Db),
case wait_db_updated(Timeout, TimeoutFun) of
updated ->
{ok, Db2} = couch_db:open(DbName, [{user_ctx, UserCtx}]),
- keep_sending_changes(Req, Resp, Db2, EndSeq, Prepend2, Timeout, TimeoutFun, FilterFun);
+ keep_sending_changes(Req, Resp, Db2, EndSeq, Prepend2, Timeout, TimeoutFun);
stop ->
send_chunk(Resp, io_lib:format("\n],\n\"last_seq\":~w}\n", [EndSeq])),
send_chunk(Resp, "")
end.
-send_changes(Req, Resp, Db, StartSeq, Prepend0, FilterFun) ->
+send_changes(Req, Resp, Db, StartSeq, Prepend0) ->
Style = list_to_existing_atom(
couch_httpd:qs_value(Req, "style", "main_only")),
- couch_db:changes_since(Db, Style, StartSeq,
+ {FilterFun, EndFilterFun} = make_filter_funs(Req, Db),
+ try
+ couch_db:changes_since(Db, Style, StartSeq,
fun([#doc_info{id=Id, high_seq=Seq}|_]=DocInfos, {_, Prepend}) ->
Results0 = [FilterFun(DocInfo) || DocInfo <- DocInfos],
Results = [Result || Result <- Results0, Result /= null],
@@ -146,16 +142,20 @@
{changes,Results}]})]),
{ok, {Seq, <<",\n">>}}
end
- end, {StartSeq, Prepend0}).
+ end, {StartSeq, Prepend0})
+ after
+ EndFilterFun()
+ end.
-make_filter_funs(_Req, _Db, nil) ->
- {ok, fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) ->
- {[{rev, couch_doc:rev_to_str(Rev)}]}
- end,
- fun() -> ok end};
-make_filter_funs(Req, Db, Filter) ->
+make_filter_funs(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};
[DName, FName] ->
DesignId = <<"_design/", DName/binary>>,
#doc{body={Props}} = couch_httpd_db:couch_doc_open(Db, DesignId, nil, []),
@@ -175,7 +175,7 @@
EndFilterFun = fun() ->
couch_query_servers:end_filter(Pid)
end,
- {ok, FilterFun, EndFilterFun};
+ {FilterFun, EndFilterFun};
_Else ->
throw({bad_request,
"filter parameter must be of the form `designname/filtername`"})