From 46bf4b727f0fae37b017f194983122c50d3f34e5 Mon Sep 17 00:00:00 2001 From: John Christopher Anderson Date: Mon, 20 Jul 2009 04:11:36 +0000 Subject: Initial checkin of _changes filters. The prime weak-spot for this approach is that it maintains an OS-process per connected filtered _changes consumer. I'm pretty sure we'll be able to work around this without changing the API, but it'll involve a lot of OS-process bookkeeping. Those enhancements should generally improve show & list performance as well. Punting on them for now, first wanted to get _changes filters implemented so people could give feedback. git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@795687 13f79535-47bb-0310-9956-ffa450edef68 --- src/couchdb/couch_db.erl | 6 +-- src/couchdb/couch_httpd_db.erl | 104 ++++++++++++++++++++++++------------ src/couchdb/couch_httpd_show.erl | 14 ++--- src/couchdb/couch_query_servers.erl | 15 ++++++ src/couchdb/couch_util.erl | 18 ++++++- 5 files changed, 106 insertions(+), 51 deletions(-) (limited to 'src/couchdb') diff --git a/src/couchdb/couch_db.erl b/src/couchdb/couch_db.erl index 21bb1e5e..4e64846c 100644 --- a/src/couchdb/couch_db.erl +++ b/src/couchdb/couch_db.erl @@ -279,11 +279,9 @@ validate_doc_update(#db{validate_doc_funs=[]}, _Doc, _GetDiskDocFun) -> ok; validate_doc_update(_Db, #doc{id= <<"_local/",_/binary>>}, _GetDiskDocFun) -> ok; -validate_doc_update(#db{name=DbName,user_ctx=Ctx}=Db, Doc, GetDiskDocFun) -> +validate_doc_update(Db, Doc, GetDiskDocFun) -> DiskDoc = GetDiskDocFun(), - JsonCtx = {[{<<"db">>, DbName}, - {<<"name">>,Ctx#user_ctx.name}, - {<<"roles">>,Ctx#user_ctx.roles}]}, + JsonCtx = couch_util:json_user_ctx(Db), try [case Fun(Doc, DiskDoc, JsonCtx) of ok -> ok; Error -> throw(Error) diff --git a/src/couchdb/couch_httpd_db.erl b/src/couchdb/couch_httpd_db.erl index 0ec1cbda..9d7dda6b 100644 --- a/src/couchdb/couch_httpd_db.erl +++ b/src/couchdb/couch_httpd_db.erl @@ -66,33 +66,38 @@ get_changes_timeout(Req, Resp) -> 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, <<"">>), - send_chunk(Resp, io_lib:format("\n],\n\"last_seq\":~w}\n", [LastSeq])), - send_chunk(Resp, "") + "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() end; handle_changes_req(#httpd{path_parts=[_,<<"_changes">>]}=Req, _Db) -> @@ -113,27 +118,23 @@ get_rest_db_updated() -> after 0 -> updated end. -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), +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), 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); + keep_sending_changes(Req, Resp, Db2, EndSeq, Prepend2, Timeout, TimeoutFun, FilterFun); 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) -> +send_changes(Req, Resp, Db, StartSeq, Prepend0, FilterFun) -> Style = list_to_existing_atom( couch_httpd:qs_value(Req, "style", "main_only")), couch_db:changes_since(Db, Style, StartSeq, fun([#doc_info{id=Id, high_seq=Seq}|_]=DocInfos, {_, Prepend}) -> - FilterFun = - fun(#doc_info{revs=[#rev_info{rev=Rev}|_]}) -> - {[{rev, couch_doc:rev_to_str(Rev)}]} - end, Results0 = [FilterFun(DocInfo) || DocInfo <- DocInfos], Results = [Result || Result <- Results0, Result /= null], case Results of @@ -147,6 +148,39 @@ send_changes(Req, Resp, Db, StartSeq, Prepend0) -> end end, {StartSeq, Prepend0}). +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) -> + case [list_to_binary(couch_httpd:unquote(Part)) + || Part <- string:tokens(Filter, "/")] of + [DName, FName] -> + DesignId = <<"_design/", DName/binary>>, + #doc{body={Props}} = couch_httpd_db:couch_doc_open(Db, DesignId, nil, []), + Lang = proplists:get_value(<<"language">>, Props, <<"javascript">>), + FilterSrc = couch_util:get_nested_json_value({Props}, [<<"filters">>, FName]), + {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), + {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, + {ok, FilterFun, EndFilterFun}; + _Else -> + throw({bad_request, + "filter parameter must be of the form `designname/filtername`"}) + end. + handle_compact_req(#httpd{method='POST',path_parts=[DbName,_,Id|_]}=Req, _Db) -> ok = couch_view_compactor:start_compact(DbName, Id), send_json(Req, 202, {[{ok, true}]}); @@ -747,7 +781,7 @@ couch_doc_from_req(Req, DocId, Json) -> % Useful for debugging % couch_doc_open(Db, DocId) -> -% couch_doc_open(Db, DocId, [], []). +% couch_doc_open(Db, DocId, nil, []). couch_doc_open(Db, DocId, Rev, Options) -> case Rev of diff --git a/src/couchdb/couch_httpd_show.erl b/src/couchdb/couch_httpd_show.erl index 692ea23d..a73105b4 100644 --- a/src/couchdb/couch_httpd_show.erl +++ b/src/couchdb/couch_httpd_show.erl @@ -45,7 +45,7 @@ handle_doc_show(Req, DesignName, ShowName, DocId, Db) -> DesignId = <<"_design/", DesignName/binary>>, #doc{body={Props}} = couch_httpd_db:couch_doc_open(Db, DesignId, nil, []), Lang = proplists:get_value(<<"language">>, Props, <<"javascript">>), - ShowSrc = get_nested_json_value({Props}, [<<"shows">>, ShowName]), + ShowSrc = couch_util:get_nested_json_value({Props}, [<<"shows">>, ShowName]), Doc = case DocId of nil -> nil; _ -> @@ -78,18 +78,10 @@ handle_view_list(Req, DesignName, ListName, ViewName, Db, Keys) -> DesignId = <<"_design/", DesignName/binary>>, #doc{body={Props}} = couch_httpd_db:couch_doc_open(Db, DesignId, nil, []), Lang = proplists:get_value(<<"language">>, Props, <<"javascript">>), - ListSrc = get_nested_json_value({Props}, [<<"lists">>, ListName]), + ListSrc = couch_util:get_nested_json_value({Props}, [<<"lists">>, ListName]), send_view_list_response(Lang, ListSrc, ViewName, DesignId, Req, Db, Keys). -get_nested_json_value({Props}, [Key|Keys]) -> - case proplists:get_value(Key, Props, nil) of - nil -> throw({not_found, <<"missing json key: ", Key/binary>>}); - Value -> get_nested_json_value(Value, Keys) - end; -get_nested_json_value(Value, []) -> - Value; -get_nested_json_value(_NotJSONObj, _) -> - throw({not_found, json_mismatch}). + send_view_list_response(Lang, ListSrc, ViewName, DesignId, Req, Db, Keys) -> Stale = couch_httpd_view:get_stale_type(Req), diff --git a/src/couchdb/couch_query_servers.erl b/src/couchdb/couch_query_servers.erl index 02018e33..1d452796 100644 --- a/src/couchdb/couch_query_servers.erl +++ b/src/couchdb/couch_query_servers.erl @@ -20,6 +20,7 @@ -export([reduce/3, rereduce/3,validate_doc_update/5]). -export([render_doc_show/6, start_view_list/2, render_list_head/4, render_list_row/3, render_list_tail/1]). +-export([start_filter/2, filter_doc/4, end_filter/1]). % -export([test/0]). -include("couch_db.hrl"). @@ -211,8 +212,22 @@ render_list_tail({Lang, Pid}) -> ok = ret_os_process(Lang, Pid), JsonResp. +start_filter(Lang, FilterSrc) -> + Pid = get_os_process(Lang), + true = couch_os_process:prompt(Pid, [<<"add_fun">>, FilterSrc]), + {ok, {Lang, Pid}}. +filter_doc({_Lang, Pid}, Doc, Req, Db) -> + JsonReq = couch_httpd_external:json_req_obj(Req, Db), + JsonDoc = couch_doc:to_json_obj(Doc, [revs]), + JsonCtx = couch_util:json_user_ctx(Db), + [true, [Pass]] = couch_os_process:prompt(Pid, + [<<"filter">>, [JsonDoc], JsonReq, JsonCtx]), + {ok, Pass}. +end_filter({Lang, Pid}) -> + ok = ret_os_process(Lang, Pid). + init([]) -> diff --git a/src/couchdb/couch_util.erl b/src/couchdb/couch_util.erl index 817572bb..db9f937c 100644 --- a/src/couchdb/couch_util.erl +++ b/src/couchdb/couch_util.erl @@ -17,7 +17,7 @@ -export([new_uuid/0, rand32/0, implode/2, collate/2, collate/3]). -export([abs_pathname/1,abs_pathname/2, trim/1, ascii_lower/1]). -export([encodeBase64/1, decodeBase64/1, to_hex/1,parse_term/1,dict_find/3]). --export([file_read_size/1]). +-export([file_read_size/1, get_nested_json_value/2, json_user_ctx/1]). -export([to_binary/1, to_list/1]). -include("couch_db.hrl"). @@ -75,6 +75,22 @@ parse_term(List) -> erl_parse:parse_term(Tokens). +get_nested_json_value({Props}, [Key|Keys]) -> + case proplists:get_value(Key, Props, nil) of + nil -> throw({not_found, <<"missing json key: ", Key/binary>>}); + Value -> get_nested_json_value(Value, Keys) + end; +get_nested_json_value(Value, []) -> + Value; +get_nested_json_value(_NotJSONObj, _) -> + throw({not_found, json_mismatch}). + +json_user_ctx(#db{name=DbName, user_ctx=Ctx}) -> + {[{<<"db">>, DbName}, + {<<"name">>,Ctx#user_ctx.name}, + {<<"roles">>,Ctx#user_ctx.roles}]}. + + % returns a random integer rand32() -> crypto:rand_uniform(0, 16#100000000). -- cgit v1.2.3