You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by be...@apache.org on 2014/01/12 13:40:08 UTC
[43/52] [partial] move src/* to the root
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_query_servers.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_query_servers.erl b/apps/couch/src/couch_query_servers.erl
new file mode 100644
index 0000000..3b58cbe
--- /dev/null
+++ b/apps/couch/src/couch_query_servers.erl
@@ -0,0 +1,616 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_query_servers).
+-behaviour(gen_server).
+
+-export([start_link/0, config_change/1]).
+
+-export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2,code_change/3]).
+-export([start_doc_map/3, map_docs/2, map_doc_raw/2, stop_doc_map/1, raw_to_ejson/1]).
+-export([reduce/3, rereduce/3,validate_doc_update/5]).
+-export([filter_docs/5]).
+-export([filter_view/3]).
+
+-export([with_ddoc_proc/2, proc_prompt/2, ddoc_prompt/3, ddoc_proc_prompt/3, json_doc/1]).
+
+% For 210-os-proc-pool.t
+-export([get_os_process/1, ret_os_process/1]).
+
+-include("couch_db.hrl").
+
+-record(proc, {
+ pid,
+ lang,
+ ddoc_keys = [],
+ prompt_fun,
+ set_timeout_fun,
+ stop_fun
+}).
+
+-record(qserver, {
+ langs, % Keyed by language name, value is {Mod,Func,Arg}
+ pid_procs, % Keyed by PID, valus is a #proc record.
+ lang_procs, % Keyed by language name, value is a #proc record
+ lang_limits, % Keyed by language name, value is {Lang, Limit, Current}
+ waitlist = [],
+ config
+}).
+
+start_link() ->
+ gen_server:start_link({local, couch_query_servers}, couch_query_servers, [], []).
+
+start_doc_map(Lang, Functions, Lib) ->
+ Proc = get_os_process(Lang),
+ case Lib of
+ {[]} -> ok;
+ Lib ->
+ true = proc_prompt(Proc, [<<"add_lib">>, Lib])
+ end,
+ lists:foreach(fun(FunctionSource) ->
+ true = proc_prompt(Proc, [<<"add_fun">>, FunctionSource])
+ end, Functions),
+ {ok, Proc}.
+
+map_docs(Proc, Docs) ->
+ % send the documents
+ Results = lists:map(
+ fun(Doc) ->
+ Json = couch_doc:to_json_obj(Doc, []),
+
+ FunsResults = proc_prompt(Proc, [<<"map_doc">>, Json]),
+ % the results are a json array of function map yields like this:
+ % [FunResults1, FunResults2 ...]
+ % where funresults is are json arrays of key value pairs:
+ % [[Key1, Value1], [Key2, Value2]]
+ % Convert the key, value pairs to tuples like
+ % [{Key1, Value1}, {Key2, Value2}]
+ lists:map(
+ fun(FunRs) ->
+ [list_to_tuple(FunResult) || FunResult <- FunRs]
+ end,
+ FunsResults)
+ end,
+ Docs),
+ {ok, Results}.
+
+map_doc_raw(Proc, Doc) ->
+ Json = couch_doc:to_json_obj(Doc, []),
+ {ok, proc_prompt_raw(Proc, [<<"map_doc">>, Json])}.
+
+
+stop_doc_map(nil) ->
+ ok;
+stop_doc_map(Proc) ->
+ ok = ret_os_process(Proc).
+
+group_reductions_results([]) ->
+ [];
+group_reductions_results(List) ->
+ {Heads, Tails} = lists:foldl(
+ fun([H|T], {HAcc,TAcc}) ->
+ {[H|HAcc], [T|TAcc]}
+ end, {[], []}, List),
+ case Tails of
+ [[]|_] -> % no tails left
+ [Heads];
+ _ ->
+ [Heads | group_reductions_results(Tails)]
+ end.
+
+rereduce(_Lang, [], _ReducedValues) ->
+ {ok, []};
+rereduce(Lang, RedSrcs, ReducedValues) ->
+ Grouped = group_reductions_results(ReducedValues),
+ Results = lists:zipwith(
+ fun
+ (<<"_", _/binary>> = FunSrc, Values) ->
+ {ok, [Result]} = builtin_reduce(rereduce, [FunSrc], [[[], V] || V <- Values], []),
+ Result;
+ (FunSrc, Values) ->
+ os_rereduce(Lang, [FunSrc], Values)
+ end, RedSrcs, Grouped),
+ {ok, Results}.
+
+reduce(_Lang, [], _KVs) ->
+ {ok, []};
+reduce(Lang, RedSrcs, KVs) ->
+ {OsRedSrcs, BuiltinReds} = lists:partition(fun
+ (<<"_", _/binary>>) -> false;
+ (_OsFun) -> true
+ end, RedSrcs),
+ {ok, OsResults} = os_reduce(Lang, OsRedSrcs, KVs),
+ {ok, BuiltinResults} = builtin_reduce(reduce, BuiltinReds, KVs, []),
+ recombine_reduce_results(RedSrcs, OsResults, BuiltinResults, []).
+
+recombine_reduce_results([], [], [], Acc) ->
+ {ok, lists:reverse(Acc)};
+recombine_reduce_results([<<"_", _/binary>>|RedSrcs], OsResults, [BRes|BuiltinResults], Acc) ->
+ recombine_reduce_results(RedSrcs, OsResults, BuiltinResults, [BRes|Acc]);
+recombine_reduce_results([_OsFun|RedSrcs], [OsR|OsResults], BuiltinResults, Acc) ->
+ recombine_reduce_results(RedSrcs, OsResults, BuiltinResults, [OsR|Acc]).
+
+os_reduce(_Lang, [], _KVs) ->
+ {ok, []};
+os_reduce(Lang, OsRedSrcs, KVs) ->
+ Proc = get_os_process(Lang),
+ OsResults = try proc_prompt(Proc, [<<"reduce">>, OsRedSrcs, KVs]) of
+ [true, Reductions] -> Reductions
+ after
+ ok = ret_os_process(Proc)
+ end,
+ {ok, OsResults}.
+
+os_rereduce(Lang, OsRedSrcs, KVs) ->
+ Proc = get_os_process(Lang),
+ try proc_prompt(Proc, [<<"rereduce">>, OsRedSrcs, KVs]) of
+ [true, [Reduction]] -> Reduction
+ after
+ ok = ret_os_process(Proc)
+ end.
+
+
+builtin_reduce(_Re, [], _KVs, Acc) ->
+ {ok, lists:reverse(Acc)};
+builtin_reduce(Re, [<<"_sum",_/binary>>|BuiltinReds], KVs, Acc) ->
+ Sum = builtin_sum_rows(KVs),
+ builtin_reduce(Re, BuiltinReds, KVs, [Sum|Acc]);
+builtin_reduce(reduce, [<<"_count",_/binary>>|BuiltinReds], KVs, Acc) ->
+ Count = length(KVs),
+ builtin_reduce(reduce, BuiltinReds, KVs, [Count|Acc]);
+builtin_reduce(rereduce, [<<"_count",_/binary>>|BuiltinReds], KVs, Acc) ->
+ Count = builtin_sum_rows(KVs),
+ builtin_reduce(rereduce, BuiltinReds, KVs, [Count|Acc]);
+builtin_reduce(Re, [<<"_stats",_/binary>>|BuiltinReds], KVs, Acc) ->
+ Stats = builtin_stats(Re, KVs),
+ builtin_reduce(Re, BuiltinReds, KVs, [Stats|Acc]).
+
+builtin_sum_rows(KVs) ->
+ lists:foldl(fun
+ ([_Key, Value], Acc) when is_number(Value), is_number(Acc) ->
+ Acc + Value;
+ ([_Key, Value], Acc) when is_list(Value), is_list(Acc) ->
+ sum_terms(Acc, Value);
+ ([_Key, Value], Acc) when is_number(Value), is_list(Acc) ->
+ sum_terms(Acc, [Value]);
+ ([_Key, Value], Acc) when is_list(Value), is_number(Acc) ->
+ sum_terms([Acc], Value);
+ (_Else, _Acc) ->
+ throw({invalid_value, <<"builtin _sum function requires map values to be numbers or lists of numbers">>})
+ end, 0, KVs).
+
+sum_terms([], []) ->
+ [];
+sum_terms([_|_]=Xs, []) ->
+ Xs;
+sum_terms([], [_|_]=Ys) ->
+ Ys;
+sum_terms([X|Xs], [Y|Ys]) when is_number(X), is_number(Y) ->
+ [X+Y | sum_terms(Xs,Ys)];
+sum_terms(_, _) ->
+ throw({invalid_value, <<"builtin _sum function requires map values to be numbers or lists of numbers">>}).
+
+builtin_stats(reduce, []) ->
+ {[]};
+builtin_stats(reduce, [[_,First]|Rest]) when is_number(First) ->
+ Stats = lists:foldl(fun([_K,V], {S,C,Mi,Ma,Sq}) when is_number(V) ->
+ {S+V, C+1, lists:min([Mi, V]), lists:max([Ma, V]), Sq+(V*V)};
+ (_, _) ->
+ throw({invalid_value,
+ <<"builtin _stats function requires map values to be numbers">>})
+ end, {First,1,First,First,First*First}, Rest),
+ {Sum, Cnt, Min, Max, Sqr} = Stats,
+ {[{sum,Sum}, {count,Cnt}, {min,Min}, {max,Max}, {sumsqr,Sqr}]};
+
+builtin_stats(rereduce, [[_,First]|Rest]) ->
+ {[{sum,Sum0}, {count,Cnt0}, {min,Min0}, {max,Max0}, {sumsqr,Sqr0}]} = First,
+ Stats = lists:foldl(fun([_K,Red], {S,C,Mi,Ma,Sq}) ->
+ {[{sum,Sum}, {count,Cnt}, {min,Min}, {max,Max}, {sumsqr,Sqr}]} = Red,
+ {Sum+S, Cnt+C, lists:min([Min, Mi]), lists:max([Max, Ma]), Sqr+Sq}
+ end, {Sum0,Cnt0,Min0,Max0,Sqr0}, Rest),
+ {Sum, Cnt, Min, Max, Sqr} = Stats,
+ {[{sum,Sum}, {count,Cnt}, {min,Min}, {max,Max}, {sumsqr,Sqr}]}.
+
+% use the function stored in ddoc.validate_doc_update to test an update.
+validate_doc_update(DDoc, EditDoc, DiskDoc, Ctx, SecObj) ->
+ JsonEditDoc = couch_doc:to_json_obj(EditDoc, [revs]),
+ JsonDiskDoc = json_doc(DiskDoc),
+ case ddoc_prompt(DDoc, [<<"validate_doc_update">>], [JsonEditDoc, JsonDiskDoc, Ctx, SecObj]) of
+ 1 ->
+ ok;
+ {[{<<"forbidden">>, Message}]} ->
+ throw({forbidden, Message});
+ {[{<<"unauthorized">>, Message}]} ->
+ throw({unauthorized, Message})
+ end.
+
+json_doc(nil) -> null;
+json_doc(Doc) ->
+ couch_doc:to_json_obj(Doc, [revs]).
+
+filter_view(DDoc, VName, Docs) ->
+ JsonDocs = [couch_doc:to_json_obj(Doc, [revs]) || Doc <- Docs],
+ [true, Passes] = ddoc_prompt(DDoc, [<<"views">>, VName, <<"map">>], [JsonDocs]),
+ {ok, Passes}.
+
+filter_docs(Req, Db, DDoc, FName, Docs) ->
+ JsonReq = case Req of
+ {json_req, JsonObj} ->
+ JsonObj;
+ #httpd{} = HttpReq ->
+ couch_httpd_external:json_req_obj(HttpReq, Db)
+ end,
+ JsonDocs = [couch_doc:to_json_obj(Doc, [revs]) || Doc <- Docs],
+ [true, Passes] = ddoc_prompt(DDoc, [<<"filters">>, FName],
+ [JsonDocs, JsonReq]),
+ {ok, Passes}.
+
+ddoc_proc_prompt({Proc, DDocId}, FunPath, Args) ->
+ proc_prompt(Proc, [<<"ddoc">>, DDocId, FunPath, Args]).
+
+ddoc_prompt(DDoc, FunPath, Args) ->
+ with_ddoc_proc(DDoc, fun({Proc, DDocId}) ->
+ proc_prompt(Proc, [<<"ddoc">>, DDocId, FunPath, Args])
+ end).
+
+with_ddoc_proc(#doc{id=DDocId,revs={Start, [DiskRev|_]}}=DDoc, Fun) ->
+ Rev = couch_doc:rev_to_str({Start, DiskRev}),
+ DDocKey = {DDocId, Rev},
+ Proc = get_ddoc_process(DDoc, DDocKey),
+ try Fun({Proc, DDocId})
+ after
+ ok = ret_os_process(Proc)
+ end.
+
+init([]) ->
+ % register async to avoid deadlock on restart_child
+ Self = self(),
+ spawn(couch_config, register, [fun ?MODULE:config_change/1, Self]),
+
+ Langs = ets:new(couch_query_server_langs, [set, private]),
+ LangLimits = ets:new(couch_query_server_lang_limits, [set, private]),
+ PidProcs = ets:new(couch_query_server_pid_langs, [set, private]),
+ LangProcs = ets:new(couch_query_server_procs, [set, private]),
+
+ ProcTimeout = list_to_integer(couch_config:get(
+ "couchdb", "os_process_timeout", "5000")),
+ ReduceLimit = list_to_atom(
+ couch_config:get("query_server_config","reduce_limit","true")),
+ OsProcLimit = list_to_integer(
+ couch_config:get("query_server_config","os_process_limit","10")),
+
+ % 'query_servers' specifies an OS command-line to execute.
+ lists:foreach(fun({Lang, Command}) ->
+ true = ets:insert(LangLimits, {?l2b(Lang), OsProcLimit, 0}),
+ true = ets:insert(Langs, {?l2b(Lang),
+ couch_os_process, start_link, [Command]})
+ end, couch_config:get("query_servers")),
+ % 'native_query_servers' specifies a {Module, Func, Arg} tuple.
+ lists:foreach(fun({Lang, SpecStr}) ->
+ {ok, {Mod, Fun, SpecArg}} = couch_util:parse_term(SpecStr),
+ true = ets:insert(LangLimits, {?l2b(Lang), 0, 0}), % 0 means no limit
+ true = ets:insert(Langs, {?l2b(Lang),
+ Mod, Fun, SpecArg})
+ end, couch_config:get("native_query_servers")),
+
+
+ process_flag(trap_exit, true),
+ {ok, #qserver{
+ langs = Langs, % Keyed by language name, value is {Mod,Func,Arg}
+ pid_procs = PidProcs, % Keyed by PID, valus is a #proc record.
+ lang_procs = LangProcs, % Keyed by language name, value is a #proc record
+ lang_limits = LangLimits, % Keyed by language name, value is {Lang, Limit, Current}
+ config = {[{<<"reduce_limit">>, ReduceLimit},{<<"timeout">>, ProcTimeout}]}
+ }}.
+
+terminate(_Reason, #qserver{pid_procs=PidProcs}) ->
+ [couch_util:shutdown_sync(P) || {P,_} <- ets:tab2list(PidProcs)],
+ ok.
+
+handle_call({get_proc, DDoc1, DDocKey}, From, Server) ->
+ #doc{body = {Props}} = DDoc = couch_doc:with_ejson_body(DDoc1),
+ Lang = couch_util:get_value(<<"language">>, Props, <<"javascript">>),
+ case lang_proc(Lang, Server, fun(Procs) ->
+ % find a proc in the set that has the DDoc
+ proc_with_ddoc(DDoc, DDocKey, Procs)
+ end) of
+ {ok, Proc} ->
+ {reply, {ok, Proc, Server#qserver.config}, Server};
+ wait ->
+ {noreply, add_to_waitlist({DDoc, DDocKey}, From, Server)};
+ Error ->
+ {reply, Error, Server}
+ end;
+handle_call({get_proc, Lang}, From, Server) ->
+ case lang_proc(Lang, Server, fun([P|_Procs]) ->
+ {ok, P}
+ end) of
+ {ok, Proc} ->
+ {reply, {ok, Proc, Server#qserver.config}, Server};
+ wait ->
+ {noreply, add_to_waitlist({Lang}, From, Server)};
+ Error ->
+ {reply, Error, Server}
+ end;
+handle_call({unlink_proc, Pid}, _From, Server) ->
+ unlink(Pid),
+ {reply, ok, Server};
+handle_call({ret_proc, Proc}, _From, #qserver{
+ pid_procs=PidProcs,
+ lang_procs=LangProcs}=Server) ->
+ % Along with max process limit, here we should check
+ % if we're over the limit and discard when we are.
+ case is_process_alive(Proc#proc.pid) of
+ true ->
+ add_value(PidProcs, Proc#proc.pid, Proc),
+ add_to_list(LangProcs, Proc#proc.lang, Proc),
+ link(Proc#proc.pid);
+ false ->
+ ok
+ end,
+ {reply, true, service_waitlist(Server)}.
+
+handle_cast(_Whatever, Server) ->
+ {noreply, Server}.
+
+handle_info({'EXIT', _, _}, Server) ->
+ {noreply, Server};
+handle_info({'DOWN', _, process, Pid, Status}, #qserver{
+ pid_procs=PidProcs,
+ lang_procs=LangProcs,
+ lang_limits=LangLimits}=Server) ->
+ case ets:lookup(PidProcs, Pid) of
+ [{Pid, Proc}] ->
+ case Status of
+ normal -> ok;
+ _ -> ?LOG_DEBUG("Linked process died abnormally: ~p (reason: ~p)", [Pid, Status])
+ end,
+ rem_value(PidProcs, Pid),
+ catch rem_from_list(LangProcs, Proc#proc.lang, Proc),
+ [{Lang, Lim, Current}] = ets:lookup(LangLimits, Proc#proc.lang),
+ true = ets:insert(LangLimits, {Lang, Lim, Current-1}),
+ {noreply, service_waitlist(Server)};
+ [] ->
+ case Status of
+ normal ->
+ {noreply, Server};
+ _ ->
+ {stop, Status, Server}
+ end
+ end.
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+config_change("query_servers") ->
+ supervisor:terminate_child(couch_secondary_services, query_servers),
+ supervisor:restart_child(couch_secondary_services, query_servers);
+config_change("native_query_servers") ->
+ supervisor:terminate_child(couch_secondary_services, query_servers),
+ supervisor:restart_child(couch_secondary_services, query_servers);
+config_change("query_server_config") ->
+ supervisor:terminate_child(couch_secondary_services, query_servers),
+ supervisor:restart_child(couch_secondary_services, query_servers).
+
+% Private API
+
+add_to_waitlist(Info, From, #qserver{waitlist=Waitlist}=Server) ->
+ Server#qserver{waitlist=[{Info, From}|Waitlist]}.
+
+service_waitlist(#qserver{waitlist=[]}=Server) ->
+ Server;
+service_waitlist(#qserver{waitlist=Waitlist}=Server) ->
+ [Oldest|RevWList] = lists:reverse(Waitlist),
+ case service_waiting(Oldest, Server) of
+ ok ->
+ Server#qserver{waitlist=lists:reverse(RevWList)};
+ wait ->
+ Server#qserver{waitlist=Waitlist}
+ end.
+
+% todo get rid of duplication
+service_waiting({{#doc{body={Props}}=DDoc, DDocKey}, From}, Server) ->
+ Lang = couch_util:get_value(<<"language">>, Props, <<"javascript">>),
+ case lang_proc(Lang, Server, fun(Procs) ->
+ % find a proc in the set that has the DDoc
+ proc_with_ddoc(DDoc, DDocKey, Procs)
+ end) of
+ {ok, Proc} ->
+ gen_server:reply(From, {ok, Proc, Server#qserver.config}),
+ ok;
+ wait -> % this should never happen
+ wait;
+ Error ->
+ gen_server:reply(From, Error),
+ ok
+ end;
+service_waiting({{Lang}, From}, Server) ->
+ case lang_proc(Lang, Server, fun([P|_Procs]) ->
+ {ok, P}
+ end) of
+ {ok, Proc} ->
+ gen_server:reply(From, {ok, Proc, Server#qserver.config}),
+ ok;
+ wait -> % this should never happen
+ wait;
+ Error ->
+ gen_server:reply(From, Error),
+ ok
+ end.
+
+lang_proc(Lang, #qserver{
+ langs=Langs,
+ pid_procs=PidProcs,
+ lang_procs=LangProcs,
+ lang_limits=LangLimits}, PickFun) ->
+ % Note to future self. Add max process limit.
+ case ets:lookup(LangProcs, Lang) of
+ [{Lang, [P|Procs]}] ->
+ {ok, Proc} = PickFun([P|Procs]),
+ rem_from_list(LangProcs, Lang, Proc),
+ {ok, Proc};
+ _ ->
+ case (catch new_process(Langs, LangLimits, Lang)) of
+ {ok, Proc} ->
+ add_value(PidProcs, Proc#proc.pid, Proc),
+ PickFun([Proc]);
+ ErrorOrWait ->
+ ErrorOrWait
+ end
+ end.
+
+new_process(Langs, LangLimits, Lang) ->
+ [{Lang, Lim, Current}] = ets:lookup(LangLimits, Lang),
+ if (Lim == 0) or (Current < Lim) -> % Lim == 0 means no limit
+ % we are below the limit for our language, make a new one
+ case ets:lookup(Langs, Lang) of
+ [{Lang, Mod, Func, Arg}] ->
+ {ok, Pid} = apply(Mod, Func, Arg),
+ erlang:monitor(process, Pid),
+ true = ets:insert(LangLimits, {Lang, Lim, Current+1}),
+ {ok, #proc{lang=Lang,
+ pid=Pid,
+ % Called via proc_prompt, proc_set_timeout, and proc_stop
+ prompt_fun={Mod, prompt},
+ set_timeout_fun={Mod, set_timeout},
+ stop_fun={Mod, stop}}};
+ _ ->
+ {unknown_query_language, Lang}
+ end;
+ true ->
+ wait
+ end.
+
+proc_with_ddoc(DDoc, DDocKey, LangProcs) ->
+ DDocProcs = lists:filter(fun(#proc{ddoc_keys=Keys}) ->
+ lists:any(fun(Key) ->
+ Key == DDocKey
+ end, Keys)
+ end, LangProcs),
+ case DDocProcs of
+ [DDocProc|_] ->
+ ?LOG_DEBUG("DDocProc found for DDocKey: ~p",[DDocKey]),
+ {ok, DDocProc};
+ [] ->
+ [TeachProc|_] = LangProcs,
+ ?LOG_DEBUG("Teach ddoc to new proc ~p with DDocKey: ~p",[TeachProc, DDocKey]),
+ {ok, SmartProc} = teach_ddoc(DDoc, DDocKey, TeachProc),
+ {ok, SmartProc}
+ end.
+
+proc_prompt(Proc, Args) ->
+ case proc_prompt_raw(Proc, Args) of
+ {json, Json} ->
+ ?JSON_DECODE(Json);
+ EJson ->
+ EJson
+ end.
+
+proc_prompt_raw(#proc{prompt_fun = {Mod, Func}} = Proc, Args) ->
+ apply(Mod, Func, [Proc#proc.pid, Args]).
+
+raw_to_ejson({json, Json}) ->
+ ?JSON_DECODE(Json);
+raw_to_ejson(EJson) ->
+ EJson.
+
+proc_stop(Proc) ->
+ {Mod, Func} = Proc#proc.stop_fun,
+ apply(Mod, Func, [Proc#proc.pid]).
+
+proc_set_timeout(Proc, Timeout) ->
+ {Mod, Func} = Proc#proc.set_timeout_fun,
+ apply(Mod, Func, [Proc#proc.pid, Timeout]).
+
+teach_ddoc(DDoc, {DDocId, _Rev}=DDocKey, #proc{ddoc_keys=Keys}=Proc) ->
+ % send ddoc over the wire
+ % we only share the rev with the client we know to update code
+ % but it only keeps the latest copy, per each ddoc, around.
+ true = proc_prompt(Proc, [<<"ddoc">>, <<"new">>, DDocId, couch_doc:to_json_obj(DDoc, [])]),
+ % we should remove any other ddocs keys for this docid
+ % because the query server overwrites without the rev
+ Keys2 = [{D,R} || {D,R} <- Keys, D /= DDocId],
+ % add ddoc to the proc
+ {ok, Proc#proc{ddoc_keys=[DDocKey|Keys2]}}.
+
+get_ddoc_process(#doc{} = DDoc, DDocKey) ->
+ % remove this case statement
+ case gen_server:call(couch_query_servers, {get_proc, DDoc, DDocKey}, infinity) of
+ {ok, Proc, {QueryConfig}} ->
+ % process knows the ddoc
+ case (catch proc_prompt(Proc, [<<"reset">>, {QueryConfig}])) of
+ true ->
+ proc_set_timeout(Proc, couch_util:get_value(<<"timeout">>, QueryConfig)),
+ link(Proc#proc.pid),
+ gen_server:call(couch_query_servers, {unlink_proc, Proc#proc.pid}, infinity),
+ Proc;
+ _ ->
+ catch proc_stop(Proc),
+ get_ddoc_process(DDoc, DDocKey)
+ end;
+ Error ->
+ throw(Error)
+ end.
+
+get_os_process(Lang) ->
+ case gen_server:call(couch_query_servers, {get_proc, Lang}, infinity) of
+ {ok, Proc, {QueryConfig}} ->
+ case (catch proc_prompt(Proc, [<<"reset">>, {QueryConfig}])) of
+ true ->
+ proc_set_timeout(Proc, couch_util:get_value(<<"timeout">>, QueryConfig)),
+ link(Proc#proc.pid),
+ gen_server:call(couch_query_servers, {unlink_proc, Proc#proc.pid}, infinity),
+ Proc;
+ _ ->
+ catch proc_stop(Proc),
+ get_os_process(Lang)
+ end;
+ Error ->
+ throw(Error)
+ end.
+
+ret_os_process(Proc) ->
+ true = gen_server:call(couch_query_servers, {ret_proc, Proc}, infinity),
+ catch unlink(Proc#proc.pid),
+ ok.
+
+add_value(Tid, Key, Value) ->
+ true = ets:insert(Tid, {Key, Value}).
+
+rem_value(Tid, Key) ->
+ true = ets:delete(Tid, Key).
+
+add_to_list(Tid, Key, Value) ->
+ case ets:lookup(Tid, Key) of
+ [{Key, Vals}] ->
+ true = ets:insert(Tid, {Key, [Value|Vals]});
+ [] ->
+ true = ets:insert(Tid, {Key, [Value]})
+ end.
+
+rem_from_list(Tid, Key, Value) when is_record(Value, proc)->
+ Pid = Value#proc.pid,
+ case ets:lookup(Tid, Key) of
+ [{Key, Vals}] ->
+ % make a new values list that doesn't include the Value arg
+ NewValues = [Val || #proc{pid=P}=Val <- Vals, P /= Pid],
+ ets:insert(Tid, {Key, NewValues});
+ [] -> ok
+ end;
+rem_from_list(Tid, Key, Value) ->
+ case ets:lookup(Tid, Key) of
+ [{Key, Vals}] ->
+ % make a new values list that doesn't include the Value arg
+ NewValues = [Val || Val <- Vals, Val /= Value],
+ ets:insert(Tid, {Key, NewValues});
+ [] -> ok
+ end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_ref_counter.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_ref_counter.erl b/apps/couch/src/couch_ref_counter.erl
new file mode 100644
index 0000000..a774f46
--- /dev/null
+++ b/apps/couch/src/couch_ref_counter.erl
@@ -0,0 +1,111 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_ref_counter).
+-behaviour(gen_server).
+
+-export([start/1, init/1, terminate/2, handle_call/3, handle_cast/2, code_change/3, handle_info/2]).
+-export([drop/1,drop/2,add/1,add/2,count/1]).
+
+start(ChildProcs) ->
+ gen_server:start(couch_ref_counter, {self(), ChildProcs}, []).
+
+
+drop(RefCounterPid) ->
+ drop(RefCounterPid, self()).
+
+drop(RefCounterPid, Pid) ->
+ gen_server:call(RefCounterPid, {drop, Pid}, infinity).
+
+
+add(RefCounterPid) ->
+ add(RefCounterPid, self()).
+
+add(RefCounterPid, Pid) ->
+ gen_server:call(RefCounterPid, {add, Pid}, infinity).
+
+count(RefCounterPid) ->
+ gen_server:call(RefCounterPid, count).
+
+% server functions
+
+-record(srv,
+ {
+ referrers=dict:new(), % a dict of each ref counting proc.
+ child_procs=[]
+ }).
+
+init({Pid, ChildProcs}) ->
+ [link(ChildProc) || ChildProc <- ChildProcs],
+ Referrers = dict:from_list([{Pid, {erlang:monitor(process, Pid), 1}}]),
+ {ok, #srv{referrers=Referrers, child_procs=ChildProcs}}.
+
+
+terminate(_Reason, #srv{child_procs=ChildProcs}) ->
+ [couch_util:shutdown_sync(Pid) || Pid <- ChildProcs],
+ ok.
+
+
+handle_call({add, Pid},_From, #srv{referrers=Referrers}=Srv) ->
+ Referrers2 =
+ case dict:find(Pid, Referrers) of
+ error ->
+ dict:store(Pid, {erlang:monitor(process, Pid), 1}, Referrers);
+ {ok, {MonRef, RefCnt}} ->
+ dict:store(Pid, {MonRef, RefCnt + 1}, Referrers)
+ end,
+ {reply, ok, Srv#srv{referrers=Referrers2}};
+handle_call(count, _From, Srv) ->
+ {monitors, Monitors} = process_info(self(), monitors),
+ {reply, length(Monitors), Srv};
+handle_call({drop, Pid}, _From, #srv{referrers=Referrers}=Srv) ->
+ Referrers2 =
+ case dict:find(Pid, Referrers) of
+ {ok, {MonRef, 1}} ->
+ erlang:demonitor(MonRef, [flush]),
+ dict:erase(Pid, Referrers);
+ {ok, {MonRef, Num}} ->
+ dict:store(Pid, {MonRef, Num-1}, Referrers);
+ error ->
+ Referrers
+ end,
+ Srv2 = Srv#srv{referrers=Referrers2},
+ case should_close() of
+ true ->
+ {stop,normal,ok,Srv2};
+ false ->
+ {reply, ok, Srv2}
+ end.
+
+handle_cast(Msg, _Srv)->
+ exit({unknown_msg,Msg}).
+
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+handle_info({'DOWN', MonRef, _, Pid, _}, #srv{referrers=Referrers}=Srv) ->
+ {ok, {MonRef, _RefCount}} = dict:find(Pid, Referrers),
+ Srv2 = Srv#srv{referrers=dict:erase(Pid, Referrers)},
+ case should_close() of
+ true ->
+ {stop,normal,Srv2};
+ false ->
+ {noreply,Srv2}
+ end.
+
+
+should_close() ->
+ case process_info(self(), monitors) of
+ {monitors, []} -> true;
+ _ -> false
+ end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_secondary_sup.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_secondary_sup.erl b/apps/couch/src/couch_secondary_sup.erl
new file mode 100644
index 0000000..6dd5604
--- /dev/null
+++ b/apps/couch/src/couch_secondary_sup.erl
@@ -0,0 +1,49 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_secondary_sup).
+-behaviour(supervisor).
+-export([init/1, start_link/0]).
+
+start_link() ->
+ supervisor:start_link({local,couch_secondary_services}, ?MODULE, []).
+
+init([]) ->
+ SecondarySupervisors = [
+ {couch_db_update_notifier_sup,
+ {couch_db_update_notifier_sup, start_link, []},
+ permanent,
+ infinity,
+ supervisor,
+ [couch_db_update_notifier_sup]},
+
+ {couch_plugin_event,
+ {gen_event, start_link, [{local, couch_plugin}]},
+ permanent,
+ brutal_kill,
+ worker,
+ dynamic}
+ ],
+ Children = SecondarySupervisors ++ [
+ begin
+ {ok, {Module, Fun, Args}} = couch_util:parse_term(SpecStr),
+
+ {list_to_atom(Name),
+ {Module, Fun, Args},
+ permanent,
+ brutal_kill,
+ worker,
+ [Module]}
+ end
+ || {Name, SpecStr}
+ <- couch_config:get("daemons"), SpecStr /= ""],
+ {ok, {{one_for_one, 10, 3600}, Children}}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_server.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_server.erl b/apps/couch/src/couch_server.erl
new file mode 100644
index 0000000..4b80dfb
--- /dev/null
+++ b/apps/couch/src/couch_server.erl
@@ -0,0 +1,493 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_server).
+-behaviour(gen_server).
+
+-export([open/2,create/2,delete/2,get_version/0,get_version/1,get_uuid/0]).
+-export([all_databases/0, all_databases/2]).
+-export([init/1, handle_call/3,sup_start_link/0]).
+-export([handle_cast/2,code_change/3,handle_info/2,terminate/2]).
+-export([dev_start/0,is_admin/2,has_admins/0,get_stats/0]).
+
+-include("couch_db.hrl").
+
+-record(server,{
+ root_dir = [],
+ dbname_regexp,
+ max_dbs_open=100,
+ dbs_open=0,
+ start_time=""
+ }).
+
+dev_start() ->
+ couch:stop(),
+ up_to_date = make:all([load, debug_info]),
+ couch:start().
+
+get_version() ->
+ couch:release_version().
+get_version(short) ->
+ %% strip git hash from version string
+ [Version|_Rest] = string:tokens(get_version(), "+"),
+ Version.
+
+
+get_uuid() ->
+ case couch_config:get("couchdb", "uuid", nil) of
+ nil ->
+ UUID = couch_uuids:random(),
+ couch_config:set("couchdb", "uuid", ?b2l(UUID)),
+ UUID;
+ UUID -> ?l2b(UUID)
+ end.
+
+get_stats() ->
+ {ok, #server{start_time=Time,dbs_open=Open}} =
+ gen_server:call(couch_server, get_server),
+ [{start_time, ?l2b(Time)}, {dbs_open, Open}].
+
+sup_start_link() ->
+ gen_server:start_link({local, couch_server}, couch_server, [], []).
+
+open(DbName, Options0) ->
+ Options = maybe_add_sys_db_callbacks(DbName, Options0),
+ case gen_server:call(couch_server, {open, DbName, Options}, infinity) of
+ {ok, Db} ->
+ Ctx = couch_util:get_value(user_ctx, Options, #user_ctx{}),
+ {ok, Db#db{user_ctx=Ctx}};
+ Error ->
+ Error
+ end.
+
+create(DbName, Options0) ->
+ Options = maybe_add_sys_db_callbacks(DbName, Options0),
+ case gen_server:call(couch_server, {create, DbName, Options}, infinity) of
+ {ok, Db} ->
+ Ctx = couch_util:get_value(user_ctx, Options, #user_ctx{}),
+ {ok, Db#db{user_ctx=Ctx}};
+ Error ->
+ Error
+ end.
+
+delete(DbName, Options) ->
+ gen_server:call(couch_server, {delete, DbName, Options}, infinity).
+
+maybe_add_sys_db_callbacks(DbName, Options) when is_binary(DbName) ->
+ maybe_add_sys_db_callbacks(?b2l(DbName), Options);
+maybe_add_sys_db_callbacks(DbName, Options) ->
+ case couch_config:get("replicator", "db", "_replicator") of
+ DbName ->
+ [
+ {before_doc_update, fun couch_replicator_manager:before_doc_update/2},
+ {after_doc_read, fun couch_replicator_manager:after_doc_read/2},
+ sys_db | Options
+ ];
+ _ ->
+ case couch_config:get("couch_httpd_auth", "authentication_db", "_users") of
+ DbName ->
+ [
+ {before_doc_update, fun couch_users_db:before_doc_update/2},
+ {after_doc_read, fun couch_users_db:after_doc_read/2},
+ sys_db | Options
+ ];
+ _ ->
+ Options
+ end
+ end.
+
+check_dbname(#server{dbname_regexp=RegExp}, DbName) ->
+ case re:run(DbName, RegExp, [{capture, none}]) of
+ nomatch ->
+ case DbName of
+ "_users" -> ok;
+ "_replicator" -> ok;
+ _Else ->
+ {error, illegal_database_name, DbName}
+ end;
+ match ->
+ ok
+ end.
+
+is_admin(User, ClearPwd) ->
+ case couch_config:get("admins", User) of
+ "-hashed-" ++ HashedPwdAndSalt ->
+ [HashedPwd, Salt] = string:tokens(HashedPwdAndSalt, ","),
+ couch_util:to_hex(crypto:sha(ClearPwd ++ Salt)) == HashedPwd;
+ _Else ->
+ false
+ end.
+
+has_admins() ->
+ couch_config:get("admins") /= [].
+
+get_full_filename(Server, DbName) ->
+ filename:join([Server#server.root_dir, "./" ++ DbName ++ ".couch"]).
+
+hash_admin_passwords() ->
+ hash_admin_passwords(true).
+
+hash_admin_passwords(Persist) ->
+ lists:foreach(
+ fun({User, ClearPassword}) ->
+ HashedPassword = couch_passwords:hash_admin_password(ClearPassword),
+ couch_config:set("admins", User, ?b2l(HashedPassword), Persist)
+ end, couch_passwords:get_unhashed_admins()).
+
+init([]) ->
+ % read config and register for configuration changes
+
+ % just stop if one of the config settings change. couch_server_sup
+ % will restart us and then we will pick up the new settings.
+
+ RootDir = couch_config:get("couchdb", "database_dir", "."),
+ MaxDbsOpen = list_to_integer(
+ couch_config:get("couchdb", "max_dbs_open")),
+ Self = self(),
+ ok = couch_config:register(
+ fun("couchdb", "database_dir") ->
+ exit(Self, config_change)
+ end),
+ ok = couch_config:register(
+ fun("couchdb", "max_dbs_open", Max) ->
+ gen_server:call(couch_server,
+ {set_max_dbs_open, list_to_integer(Max)})
+ end),
+ ok = couch_file:init_delete_dir(RootDir),
+ hash_admin_passwords(),
+ ok = couch_config:register(
+ fun("admins", _Key, _Value, Persist) ->
+ % spawn here so couch_config doesn't try to call itself
+ spawn(fun() -> hash_admin_passwords(Persist) end)
+ end, false),
+ {ok, RegExp} = re:compile("^[a-z][a-z0-9\\_\\$()\\+\\-\\/]*$"),
+ ets:new(couch_dbs_by_name, [set, private, named_table]),
+ ets:new(couch_dbs_by_pid, [set, private, named_table]),
+ ets:new(couch_dbs_by_lru, [ordered_set, private, named_table]),
+ ets:new(couch_sys_dbs, [set, private, named_table]),
+ process_flag(trap_exit, true),
+ {ok, #server{root_dir=RootDir,
+ dbname_regexp=RegExp,
+ max_dbs_open=MaxDbsOpen,
+ start_time=couch_util:rfc1123_date()}}.
+
+terminate(_Reason, _Srv) ->
+ lists:foreach(
+ fun({_, {_, Pid, _}}) ->
+ couch_util:shutdown_sync(Pid)
+ end,
+ ets:tab2list(couch_dbs_by_name)).
+
+all_databases() ->
+ {ok, DbList} = all_databases(
+ fun(DbName, Acc) -> {ok, [DbName | Acc]} end, []),
+ {ok, lists:usort(DbList)}.
+
+all_databases(Fun, Acc0) ->
+ {ok, #server{root_dir=Root}} = gen_server:call(couch_server, get_server),
+ NormRoot = couch_util:normpath(Root),
+ FinalAcc = try
+ filelib:fold_files(Root, "^[a-z0-9\\_\\$()\\+\\-]*[\\.]couch$", true,
+ fun(Filename, AccIn) ->
+ NormFilename = couch_util:normpath(Filename),
+ case NormFilename -- NormRoot of
+ [$/ | RelativeFilename] -> ok;
+ RelativeFilename -> ok
+ end,
+ case Fun(?l2b(filename:rootname(RelativeFilename, ".couch")), AccIn) of
+ {ok, NewAcc} -> NewAcc;
+ {stop, NewAcc} -> throw({stop, Fun, NewAcc})
+ end
+ end, Acc0)
+ catch throw:{stop, Fun, Acc1} ->
+ Acc1
+ end,
+ {ok, FinalAcc}.
+
+
+maybe_close_lru_db(#server{dbs_open=NumOpen, max_dbs_open=MaxOpen}=Server)
+ when NumOpen < MaxOpen ->
+ {ok, Server};
+maybe_close_lru_db(#server{dbs_open=NumOpen}=Server) ->
+ % must free up the lru db.
+ case try_close_lru(now()) of
+ ok ->
+ {ok, Server#server{dbs_open=NumOpen - 1}};
+ Error -> Error
+ end.
+
+try_close_lru(StartTime) ->
+ LruTime = get_lru(),
+ if LruTime > StartTime ->
+ % this means we've looped through all our opened dbs and found them
+ % all in use.
+ {error, all_dbs_active};
+ true ->
+ [{_, DbName}] = ets:lookup(couch_dbs_by_lru, LruTime),
+ [{_, {opened, MainPid, LruTime}}] = ets:lookup(couch_dbs_by_name, DbName),
+ case couch_db:is_idle(MainPid) of
+ true ->
+ ok = shutdown_idle_db(DbName, MainPid, LruTime);
+ false ->
+ % this still has referrers. Go ahead and give it a current lru time
+ % and try the next one in the table.
+ NewLruTime = now(),
+ true = ets:insert(couch_dbs_by_name, {DbName, {opened, MainPid, NewLruTime}}),
+ true = ets:insert(couch_dbs_by_pid, {MainPid, DbName}),
+ true = ets:delete(couch_dbs_by_lru, LruTime),
+ true = ets:insert(couch_dbs_by_lru, {NewLruTime, DbName}),
+ try_close_lru(StartTime)
+ end
+ end.
+
+get_lru() ->
+ get_lru(ets:first(couch_dbs_by_lru)).
+
+get_lru(LruTime) ->
+ [{LruTime, DbName}] = ets:lookup(couch_dbs_by_lru, LruTime),
+ case ets:member(couch_sys_dbs, DbName) of
+ false ->
+ LruTime;
+ true ->
+ [{_, {opened, MainPid, _}}] = ets:lookup(couch_dbs_by_name, DbName),
+ case couch_db:is_idle(MainPid) of
+ true ->
+ NextLru = ets:next(couch_dbs_by_lru, LruTime),
+ ok = shutdown_idle_db(DbName, MainPid, LruTime),
+ get_lru(NextLru);
+ false ->
+ get_lru(ets:next(couch_dbs_by_lru, LruTime))
+ end
+ end.
+
+shutdown_idle_db(DbName, MainPid, LruTime) ->
+ couch_util:shutdown_sync(MainPid),
+ true = ets:delete(couch_dbs_by_lru, LruTime),
+ true = ets:delete(couch_dbs_by_name, DbName),
+ true = ets:delete(couch_dbs_by_pid, MainPid),
+ true = ets:delete(couch_sys_dbs, DbName),
+ ok.
+
+open_async(Server, From, DbName, Filepath, Options) ->
+ Parent = self(),
+ Opener = spawn_link(fun() ->
+ Res = couch_db:start_link(DbName, Filepath, Options),
+ gen_server:call(
+ Parent, {open_result, DbName, Res, Options}, infinity
+ ),
+ unlink(Parent),
+ case Res of
+ {ok, DbReader} ->
+ unlink(DbReader);
+ _ ->
+ ok
+ end
+ end),
+ true = ets:insert(couch_dbs_by_name, {DbName, {opening, Opener, [From]}}),
+ true = ets:insert(couch_dbs_by_pid, {Opener, DbName}),
+ DbsOpen = case lists:member(sys_db, Options) of
+ true ->
+ true = ets:insert(couch_sys_dbs, {DbName, true}),
+ Server#server.dbs_open;
+ false ->
+ Server#server.dbs_open + 1
+ end,
+ Server#server{dbs_open = DbsOpen}.
+
+handle_call({set_max_dbs_open, Max}, _From, Server) ->
+ {reply, ok, Server#server{max_dbs_open=Max}};
+handle_call(get_server, _From, Server) ->
+ {reply, {ok, Server}, Server};
+handle_call({open_result, DbName, {ok, OpenedDbPid}, Options}, _From, Server) ->
+ link(OpenedDbPid),
+ [{DbName, {opening,Opener,Froms}}] = ets:lookup(couch_dbs_by_name, DbName),
+ lists:foreach(fun({FromPid,_}=From) ->
+ gen_server:reply(From,
+ catch couch_db:open_ref_counted(OpenedDbPid, FromPid))
+ end, Froms),
+ LruTime = now(),
+ true = ets:insert(couch_dbs_by_name,
+ {DbName, {opened, OpenedDbPid, LruTime}}),
+ true = ets:delete(couch_dbs_by_pid, Opener),
+ true = ets:insert(couch_dbs_by_pid, {OpenedDbPid, DbName}),
+ true = ets:insert(couch_dbs_by_lru, {LruTime, DbName}),
+ case lists:member(create, Options) of
+ true ->
+ couch_db_update_notifier:notify({created, DbName});
+ false ->
+ ok
+ end,
+ {reply, ok, Server};
+handle_call({open_result, DbName, {error, eexist}, Options}, From, Server) ->
+ handle_call({open_result, DbName, file_exists, Options}, From, Server);
+handle_call({open_result, DbName, Error, Options}, _From, Server) ->
+ [{DbName, {opening,Opener,Froms}}] = ets:lookup(couch_dbs_by_name, DbName),
+ lists:foreach(fun(From) ->
+ gen_server:reply(From, Error)
+ end, Froms),
+ true = ets:delete(couch_dbs_by_name, DbName),
+ true = ets:delete(couch_dbs_by_pid, Opener),
+ DbsOpen = case lists:member(sys_db, Options) of
+ true ->
+ true = ets:delete(couch_sys_dbs, DbName),
+ Server#server.dbs_open;
+ false ->
+ Server#server.dbs_open - 1
+ end,
+ {reply, ok, Server#server{dbs_open = DbsOpen}};
+handle_call({open, DbName, Options}, {FromPid,_}=From, Server) ->
+ LruTime = now(),
+ case ets:lookup(couch_dbs_by_name, DbName) of
+ [] ->
+ open_db(DbName, Server, Options, From);
+ [{_, {opening, Opener, Froms}}] ->
+ true = ets:insert(couch_dbs_by_name, {DbName, {opening, Opener, [From|Froms]}}),
+ {noreply, Server};
+ [{_, {opened, MainPid, PrevLruTime}}] ->
+ true = ets:insert(couch_dbs_by_name, {DbName, {opened, MainPid, LruTime}}),
+ true = ets:delete(couch_dbs_by_lru, PrevLruTime),
+ true = ets:insert(couch_dbs_by_lru, {LruTime, DbName}),
+ {reply, couch_db:open_ref_counted(MainPid, FromPid), Server}
+ end;
+handle_call({create, DbName, Options}, From, Server) ->
+ case ets:lookup(couch_dbs_by_name, DbName) of
+ [] ->
+ open_db(DbName, Server, [create | Options], From);
+ [_AlreadyRunningDb] ->
+ {reply, file_exists, Server}
+ end;
+handle_call({delete, DbName, _Options}, _From, Server) ->
+ DbNameList = binary_to_list(DbName),
+ case check_dbname(Server, DbNameList) of
+ ok ->
+ FullFilepath = get_full_filename(Server, DbNameList),
+ UpdateState =
+ case ets:lookup(couch_dbs_by_name, DbName) of
+ [] -> false;
+ [{_, {opening, Pid, Froms}}] ->
+ couch_util:shutdown_sync(Pid),
+ true = ets:delete(couch_dbs_by_name, DbName),
+ true = ets:delete(couch_dbs_by_pid, Pid),
+ [gen_server:reply(F, not_found) || F <- Froms],
+ true;
+ [{_, {opened, Pid, LruTime}}] ->
+ couch_util:shutdown_sync(Pid),
+ true = ets:delete(couch_dbs_by_name, DbName),
+ true = ets:delete(couch_dbs_by_pid, Pid),
+ true = ets:delete(couch_dbs_by_lru, LruTime),
+ true
+ end,
+ Server2 = case UpdateState of
+ true ->
+ DbsOpen = case ets:member(couch_sys_dbs, DbName) of
+ true ->
+ true = ets:delete(couch_sys_dbs, DbName),
+ Server#server.dbs_open;
+ false ->
+ Server#server.dbs_open - 1
+ end,
+ Server#server{dbs_open = DbsOpen};
+ false ->
+ Server
+ end,
+
+ %% Delete any leftover .compact files. If we don't do this a subsequent
+ %% request for this DB will try to open the .compact file and use it.
+ couch_file:delete(Server#server.root_dir, FullFilepath ++ ".compact"),
+
+ case couch_file:delete(Server#server.root_dir, FullFilepath) of
+ ok ->
+ couch_db_update_notifier:notify({deleted, DbName}),
+ {reply, ok, Server2};
+ {error, enoent} ->
+ {reply, not_found, Server2};
+ Else ->
+ {reply, Else, Server2}
+ end;
+ Error ->
+ {reply, Error, Server}
+ end.
+
+handle_cast(Msg, _Server) ->
+ exit({unknown_cast_message, Msg}).
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+handle_info({'EXIT', _Pid, config_change}, Server) ->
+ {noreply, shutdown, Server};
+handle_info({'EXIT', Pid, Reason}, Server) ->
+ Server2 = case ets:lookup(couch_dbs_by_pid, Pid) of
+ [{Pid, DbName}] ->
+
+ % If the Pid is known, the name should be as well.
+ % If not, that's an error, which is why there is no [] clause.
+ case ets:lookup(couch_dbs_by_name, DbName) of
+ [{_, {opening, Pid, Froms}}] ->
+ Msg = case Reason of
+ snappy_nif_not_loaded ->
+ io_lib:format(
+ "To open the database `~s`, Apache CouchDB "
+ "must be built with Erlang OTP R13B04 or higher.",
+ [DbName]
+ );
+ true ->
+ io_lib:format("Error opening database ~p: ~p", [DbName, Reason])
+ end,
+ ?LOG_ERROR(Msg, []),
+ lists:foreach(
+ fun(F) -> gen_server:reply(F, {bad_otp_release, Msg}) end,
+ Froms
+ );
+ [{_, {opened, Pid, LruTime}}] ->
+ ?LOG_ERROR(
+ "Unexpected exit of database process ~p [~p]: ~p",
+ [Pid, DbName, Reason]
+ ),
+ true = ets:delete(couch_dbs_by_lru, LruTime)
+ end,
+
+ true = ets:delete(couch_dbs_by_pid, DbName),
+ true = ets:delete(couch_dbs_by_name, DbName),
+
+ case ets:lookup(couch_sys_dbs, DbName) of
+ [{DbName, _}] ->
+ true = ets:delete(couch_sys_dbs, DbName),
+ Server;
+ [] ->
+ Server#server{dbs_open = Server#server.dbs_open - 1}
+ end
+ end,
+ {noreply, Server2};
+handle_info(Error, _Server) ->
+ ?LOG_ERROR("Unexpected message, restarting couch_server: ~p", [Error]),
+ exit(kill).
+
+open_db(DbName, Server, Options, From) ->
+ DbNameList = binary_to_list(DbName),
+ case check_dbname(Server, DbNameList) of
+ ok ->
+ Filepath = get_full_filename(Server, DbNameList),
+ case lists:member(sys_db, Options) of
+ true ->
+ {noreply, open_async(Server, From, DbName, Filepath, Options)};
+ false ->
+ case maybe_close_lru_db(Server) of
+ {ok, Server2} ->
+ {noreply, open_async(Server2, From, DbName, Filepath, Options)};
+ CloseError ->
+ {reply, CloseError, Server}
+ end
+ end;
+ Error ->
+ {reply, Error, Server}
+ end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_server_sup.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_server_sup.erl b/apps/couch/src/couch_server_sup.erl
new file mode 100644
index 0000000..7f37677
--- /dev/null
+++ b/apps/couch/src/couch_server_sup.erl
@@ -0,0 +1,117 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_server_sup).
+-behaviour(supervisor).
+
+
+-export([start_link/1,stop/0, couch_config_start_link_wrapper/2,
+ restart_core_server/0, config_change/2]).
+
+-include("couch_db.hrl").
+
+%% supervisor callbacks
+-export([init/1]).
+
+start_link(IniFiles) ->
+ case whereis(couch_server_sup) of
+ undefined ->
+ start_server(IniFiles);
+ _Else ->
+ {error, already_started}
+ end.
+
+restart_core_server() ->
+ init:restart().
+
+couch_config_start_link_wrapper(IniFiles, FirstConfigPid) ->
+ case is_process_alive(FirstConfigPid) of
+ true ->
+ link(FirstConfigPid),
+ {ok, FirstConfigPid};
+ false -> couch_config:start_link(IniFiles)
+ end.
+
+start_server(IniFiles) ->
+ case init:get_argument(pidfile) of
+ {ok, [PidFile]} ->
+ case file:write_file(PidFile, os:getpid()) of
+ ok -> ok;
+ {error, Reason} ->
+ io:format("Failed to write PID file ~s: ~s",
+ [PidFile, file:format_error(Reason)])
+ end;
+ _ -> ok
+ end,
+
+ {ok, ConfigPid} = couch_config:start_link(IniFiles),
+
+ LogLevel = couch_config:get("log", "level", "info"),
+ % announce startup
+ io:format("Apache CouchDB ~s (LogLevel=~s) is starting.~n", [
+ couch_server:get_version(),
+ LogLevel
+ ]),
+ case LogLevel of
+ "debug" ->
+ io:format("Configuration Settings ~p:~n", [IniFiles]),
+ [io:format(" [~s] ~s=~p~n", [Module, Variable, Value])
+ || {{Module, Variable}, Value} <- couch_config:all()];
+ _ -> ok
+ end,
+
+ BaseChildSpecs =
+ {{one_for_all, 10, 3600},
+ [{couch_config,
+ {couch_server_sup, couch_config_start_link_wrapper, [IniFiles, ConfigPid]},
+ permanent,
+ brutal_kill,
+ worker,
+ [couch_config]},
+ {couch_primary_services,
+ {couch_primary_sup, start_link, []},
+ permanent,
+ infinity,
+ supervisor,
+ [couch_primary_sup]},
+ {couch_secondary_services,
+ {couch_secondary_sup, start_link, []},
+ permanent,
+ infinity,
+ supervisor,
+ [couch_secondary_sup]}
+ ]},
+
+ % ensure these applications are running
+ application:start(ibrowse),
+ application:start(crypto),
+
+ {ok, Pid} = supervisor:start_link(
+ {local, couch_server_sup}, couch_server_sup, BaseChildSpecs),
+
+ % launch the icu bridge
+ % just restart if one of the config settings change.
+ couch_config:register(fun ?MODULE:config_change/2, Pid),
+
+ unlink(ConfigPid),
+
+ {ok, Pid}.
+
+stop() ->
+ catch exit(whereis(couch_server_sup), normal).
+
+config_change("daemons", _) ->
+ supervisor:terminate_child(couch_server_sup, couch_secondary_services),
+ supervisor:restart_child(couch_server_sup, couch_secondary_services).
+
+init(ChildSpecs) ->
+ {ok, ChildSpecs}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_stats_aggregator.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_stats_aggregator.erl b/apps/couch/src/couch_stats_aggregator.erl
new file mode 100644
index 0000000..6090355
--- /dev/null
+++ b/apps/couch/src/couch_stats_aggregator.erl
@@ -0,0 +1,297 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_stats_aggregator).
+-behaviour(gen_server).
+
+-export([start/0, start/1, stop/0]).
+-export([all/0, all/1, get/1, get/2, get_json/1, get_json/2, collect_sample/0]).
+
+-export([init/1, terminate/2, code_change/3]).
+-export([handle_call/3, handle_cast/2, handle_info/2]).
+
+-record(aggregate, {
+ description = <<"">>,
+ seconds = 0,
+ count = 0,
+ current = null,
+ sum = null,
+ mean = null,
+ variance = null,
+ stddev = null,
+ min = null,
+ max = null,
+ samples = []
+}).
+
+
+start() ->
+ PrivDir = couch_util:priv_dir(),
+ start(filename:join(PrivDir, "stat_descriptions.cfg")).
+
+start(FileName) ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [FileName], []).
+
+stop() ->
+ gen_server:cast(?MODULE, stop).
+
+all() ->
+ ?MODULE:all(0).
+all(Time) when is_binary(Time) ->
+ ?MODULE:all(list_to_integer(binary_to_list(Time)));
+all(Time) when is_atom(Time) ->
+ ?MODULE:all(list_to_integer(atom_to_list(Time)));
+all(Time) when is_integer(Time) ->
+ Aggs = ets:match(?MODULE, {{'$1', Time}, '$2'}),
+ Stats = lists:map(fun([Key, Agg]) -> {Key, Agg} end, Aggs),
+ case Stats of
+ [] ->
+ {[]};
+ _ ->
+ Ret = lists:foldl(fun({{Mod, Key}, Agg}, Acc) ->
+ CurrKeys = case proplists:lookup(Mod, Acc) of
+ none -> [];
+ {Mod, {Keys}} -> Keys
+ end,
+ NewMod = {[{Key, to_json_term(Agg)} | CurrKeys]},
+ [{Mod, NewMod} | proplists:delete(Mod, Acc)]
+ end, [], Stats),
+ {Ret}
+ end.
+
+get(Key) ->
+ ?MODULE:get(Key, 0).
+get(Key, Time) when is_binary(Time) ->
+ ?MODULE:get(Key, list_to_integer(binary_to_list(Time)));
+get(Key, Time) when is_atom(Time) ->
+ ?MODULE:get(Key, list_to_integer(atom_to_list(Time)));
+get(Key, Time) when is_integer(Time) ->
+ case ets:lookup(?MODULE, {make_key(Key), Time}) of
+ [] -> #aggregate{seconds=Time};
+ [{_, Agg}] -> Agg
+ end.
+
+get_json(Key) ->
+ get_json(Key, 0).
+get_json(Key, Time) ->
+ to_json_term(?MODULE:get(Key, Time)).
+
+collect_sample() ->
+ gen_server:call(?MODULE, collect_sample, infinity).
+
+
+init(StatDescsFileName) ->
+ % Create an aggregate entry for each {description, rate} pair.
+ ets:new(?MODULE, [named_table, set, protected]),
+ SampleStr = couch_config:get("stats", "samples", "[0]"),
+ {ok, Samples} = couch_util:parse_term(SampleStr),
+ {ok, Descs} = file:consult(StatDescsFileName),
+ lists:foreach(fun({Sect, Key, Value}) ->
+ lists:foreach(fun(Secs) ->
+ Agg = #aggregate{
+ description=list_to_binary(Value),
+ seconds=Secs
+ },
+ ets:insert(?MODULE, {{{Sect, Key}, Secs}, Agg})
+ end, Samples)
+ end, Descs),
+
+ Self = self(),
+ ok = couch_config:register(
+ fun("stats", _) -> exit(Self, config_change) end
+ ),
+
+ Rate = list_to_integer(couch_config:get("stats", "rate", "1000")),
+ % TODO: Add timer_start to kernel start options.
+ {ok, TRef} = timer:apply_after(Rate, ?MODULE, collect_sample, []),
+ {ok, {TRef, Rate}}.
+
+terminate(_Reason, {TRef, _Rate}) ->
+ timer:cancel(TRef),
+ ok.
+
+handle_call(collect_sample, _, {OldTRef, SampleInterval}) ->
+ timer:cancel(OldTRef),
+ {ok, TRef} = timer:apply_after(SampleInterval, ?MODULE, collect_sample, []),
+ % Gather new stats values to add.
+ Incs = lists:map(fun({Key, Value}) ->
+ {Key, {incremental, Value}}
+ end, couch_stats_collector:all(incremental)),
+ Abs = lists:map(fun({Key, Values}) ->
+ couch_stats_collector:clear(Key),
+ Values2 = case Values of
+ X when is_list(X) -> X;
+ Else -> [Else]
+ end,
+ {_, Mean} = lists:foldl(fun(Val, {Count, Curr}) ->
+ {Count+1, Curr + (Val - Curr) / (Count+1)}
+ end, {0, 0}, Values2),
+ {Key, {absolute, Mean}}
+ end, couch_stats_collector:all(absolute)),
+
+ Values = Incs ++ Abs,
+ Now = erlang:now(),
+ lists:foreach(fun({{Key, Rate}, Agg}) ->
+ NewAgg = case proplists:lookup(Key, Values) of
+ none ->
+ rem_values(Now, Agg);
+ {Key, {Type, Value}} ->
+ NewValue = new_value(Type, Value, Agg#aggregate.current),
+ Agg2 = add_value(Now, NewValue, Agg),
+ rem_values(Now, Agg2)
+ end,
+ ets:insert(?MODULE, {{Key, Rate}, NewAgg})
+ end, ets:tab2list(?MODULE)),
+ {reply, ok, {TRef, SampleInterval}}.
+
+handle_cast(stop, State) ->
+ {stop, normal, State}.
+
+handle_info(_Info, State) ->
+ {noreply, State}.
+
+code_change(_OldVersion, State, _Extra) ->
+ {ok, State}.
+
+
+new_value(incremental, Value, null) ->
+ Value;
+new_value(incremental, Value, Current) ->
+ Value - Current;
+new_value(absolute, Value, _Current) ->
+ Value.
+
+add_value(Time, Value, #aggregate{count=Count, seconds=Secs}=Agg) when Count < 1 ->
+ Samples = case Secs of
+ 0 -> [];
+ _ -> [{Time, Value}]
+ end,
+ Agg#aggregate{
+ count=1,
+ current=Value,
+ sum=Value,
+ mean=Value,
+ variance=0.0,
+ stddev=null,
+ min=Value,
+ max=Value,
+ samples=Samples
+ };
+add_value(Time, Value, Agg) ->
+ #aggregate{
+ count=Count,
+ current=Current,
+ sum=Sum,
+ mean=Mean,
+ variance=Variance,
+ samples=Samples
+ } = Agg,
+
+ NewCount = Count + 1,
+ NewMean = Mean + (Value - Mean) / NewCount,
+ NewVariance = Variance + (Value - Mean) * (Value - NewMean),
+ StdDev = case NewCount > 1 of
+ false -> null;
+ _ -> math:sqrt(NewVariance / (NewCount - 1))
+ end,
+ Agg2 = Agg#aggregate{
+ count=NewCount,
+ current=Current + Value,
+ sum=Sum + Value,
+ mean=NewMean,
+ variance=NewVariance,
+ stddev=StdDev,
+ min=lists:min([Agg#aggregate.min, Value]),
+ max=lists:max([Agg#aggregate.max, Value])
+ },
+ case Agg2#aggregate.seconds of
+ 0 -> Agg2;
+ _ -> Agg2#aggregate{samples=[{Time, Value} | Samples]}
+ end.
+
+rem_values(Time, Agg) ->
+ Seconds = Agg#aggregate.seconds,
+ Samples = Agg#aggregate.samples,
+ Pred = fun({When, _Value}) ->
+ timer:now_diff(Time, When) =< (Seconds * 1000000)
+ end,
+ {Keep, Remove} = lists:splitwith(Pred, Samples),
+ Agg2 = lists:foldl(fun({_, Value}, Acc) ->
+ rem_value(Value, Acc)
+ end, Agg, Remove),
+ Agg2#aggregate{samples=Keep}.
+
+rem_value(_Value, #aggregate{count=Count, seconds=Secs}) when Count =< 1 ->
+ #aggregate{seconds=Secs};
+rem_value(Value, Agg) ->
+ #aggregate{
+ count=Count,
+ sum=Sum,
+ mean=Mean,
+ variance=Variance
+ } = Agg,
+
+ OldMean = (Mean * Count - Value) / (Count - 1),
+ OldVariance = Variance - (Value - OldMean) * (Value - Mean),
+ OldCount = Count - 1,
+ StdDev = case OldCount > 1 of
+ false -> null;
+ _ -> math:sqrt(clamp_value(OldVariance / (OldCount - 1)))
+ end,
+ Agg#aggregate{
+ count=OldCount,
+ sum=Sum-Value,
+ mean=clamp_value(OldMean),
+ variance=clamp_value(OldVariance),
+ stddev=StdDev
+ }.
+
+to_json_term(Agg) ->
+ {Min, Max} = case Agg#aggregate.seconds > 0 of
+ false ->
+ {Agg#aggregate.min, Agg#aggregate.max};
+ _ ->
+ case length(Agg#aggregate.samples) > 0 of
+ true ->
+ Extract = fun({_Time, Value}) -> Value end,
+ Samples = lists:map(Extract, Agg#aggregate.samples),
+ {lists:min(Samples), lists:max(Samples)};
+ _ ->
+ {null, null}
+ end
+ end,
+ {[
+ {description, Agg#aggregate.description},
+ {current, round_value(Agg#aggregate.sum)},
+ {sum, round_value(Agg#aggregate.sum)},
+ {mean, round_value(Agg#aggregate.mean)},
+ {stddev, round_value(Agg#aggregate.stddev)},
+ {min, Min},
+ {max, Max}
+ ]}.
+
+make_key({Mod, Val}) when is_integer(Val) ->
+ {Mod, list_to_atom(integer_to_list(Val))};
+make_key(Key) ->
+ Key.
+
+round_value(Val) when not is_number(Val) ->
+ Val;
+round_value(Val) when Val == 0 ->
+ Val;
+round_value(Val) ->
+ erlang:round(Val * 1000.0) / 1000.0.
+
+clamp_value(Val) when Val > 0.00000000000001 ->
+ Val;
+clamp_value(_) ->
+ 0.0.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_stats_collector.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_stats_collector.erl b/apps/couch/src/couch_stats_collector.erl
new file mode 100644
index 0000000..f7b9bb4
--- /dev/null
+++ b/apps/couch/src/couch_stats_collector.erl
@@ -0,0 +1,136 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+% todo
+% - remove existance check on increment(), decrement() and record(). have
+% modules initialize counters on startup.
+
+-module(couch_stats_collector).
+
+-behaviour(gen_server).
+
+-export([start/0, stop/0]).
+-export([all/0, all/1, get/1, increment/1, decrement/1, record/2, clear/1]).
+-export([track_process_count/1, track_process_count/2]).
+
+-export([init/1, terminate/2, code_change/3]).
+-export([handle_call/3, handle_cast/2, handle_info/2]).
+
+-define(HIT_TABLE, stats_hit_table).
+-define(ABS_TABLE, stats_abs_table).
+
+start() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+stop() ->
+ gen_server:call(?MODULE, stop).
+
+all() ->
+ ets:tab2list(?HIT_TABLE) ++ abs_to_list().
+
+all(Type) ->
+ case Type of
+ incremental -> ets:tab2list(?HIT_TABLE);
+ absolute -> abs_to_list()
+ end.
+
+get(Key) ->
+ case ets:lookup(?HIT_TABLE, Key) of
+ [] ->
+ case ets:lookup(?ABS_TABLE, Key) of
+ [] ->
+ nil;
+ AbsVals ->
+ lists:map(fun({_, Value}) -> Value end, AbsVals)
+ end;
+ [{_, Counter}] ->
+ Counter
+ end.
+
+increment(Key) ->
+ Key2 = make_key(Key),
+ case catch ets:update_counter(?HIT_TABLE, Key2, 1) of
+ {'EXIT', {badarg, _}} ->
+ catch ets:insert(?HIT_TABLE, {Key2, 1}),
+ ok;
+ _ ->
+ ok
+ end.
+
+decrement(Key) ->
+ Key2 = make_key(Key),
+ case catch ets:update_counter(?HIT_TABLE, Key2, -1) of
+ {'EXIT', {badarg, _}} ->
+ catch ets:insert(?HIT_TABLE, {Key2, -1}),
+ ok;
+ _ -> ok
+ end.
+
+record(Key, Value) ->
+ catch ets:insert(?ABS_TABLE, {make_key(Key), Value}).
+
+clear(Key) ->
+ catch ets:delete(?ABS_TABLE, make_key(Key)).
+
+track_process_count(Stat) ->
+ track_process_count(self(), Stat).
+
+track_process_count(Pid, Stat) ->
+ MonitorFun = fun() ->
+ Ref = erlang:monitor(process, Pid),
+ receive {'DOWN', Ref, _, _, _} -> ok end,
+ couch_stats_collector:decrement(Stat)
+ end,
+ case (catch couch_stats_collector:increment(Stat)) of
+ ok -> spawn(MonitorFun);
+ _ -> ok
+ end.
+
+
+init(_) ->
+ ets:new(?HIT_TABLE, [named_table, set, public]),
+ ets:new(?ABS_TABLE, [named_table, duplicate_bag, public]),
+ {ok, nil}.
+
+terminate(_Reason, _State) ->
+ ok.
+
+handle_call(stop, _, State) ->
+ {stop, normal, stopped, State}.
+
+handle_cast(foo, State) ->
+ {noreply, State}.
+
+handle_info(_Info, State) ->
+ {noreply, State}.
+
+code_change(_OldVersion, State, _Extra) ->
+ {ok, State}.
+
+
+make_key({Module, Key}) when is_integer(Key) ->
+ {Module, list_to_atom(integer_to_list(Key))};
+make_key(Key) ->
+ Key.
+
+abs_to_list() ->
+ SortedKVs = lists:sort(ets:tab2list(?ABS_TABLE)),
+ lists:foldl(fun({Key, Val}, Acc) ->
+ case Acc of
+ [] ->
+ [{Key, [Val]}];
+ [{Key, Prev} | Rest] ->
+ [{Key, [Val | Prev]} | Rest];
+ Others ->
+ [{Key, [Val]} | Others]
+ end
+ end, [], SortedKVs).
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_stream.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_stream.erl b/apps/couch/src/couch_stream.erl
new file mode 100644
index 0000000..959feef
--- /dev/null
+++ b/apps/couch/src/couch_stream.erl
@@ -0,0 +1,299 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_stream).
+-behaviour(gen_server).
+
+% public API
+-export([open/1, open/2, close/1]).
+-export([foldl/4, foldl/5, foldl_decode/6, range_foldl/6]).
+-export([copy_to_new_stream/3, write/2]).
+
+% gen_server callbacks
+-export([init/1, terminate/2, code_change/3]).
+-export([handle_cast/2, handle_call/3, handle_info/2]).
+
+-include("couch_db.hrl").
+
+-define(DEFAULT_BUFFER_SIZE, 4096).
+
+-record(stream,
+ {fd = 0,
+ written_pointers=[],
+ buffer_list = [],
+ buffer_len = 0,
+ max_buffer,
+ written_len = 0,
+ md5,
+ % md5 of the content without any transformation applied (e.g. compression)
+ % needed for the attachment upload integrity check (ticket 558)
+ identity_md5,
+ identity_len = 0,
+ encoding_fun,
+ end_encoding_fun
+ }).
+
+
+%%% Interface functions %%%
+
+open(Fd) ->
+ open(Fd, []).
+
+open(Fd, Options) ->
+ gen_server:start_link(couch_stream, {Fd, Options}, []).
+
+close(Pid) ->
+ gen_server:call(Pid, close, infinity).
+
+copy_to_new_stream(Fd, PosList, DestFd) ->
+ {ok, Dest} = open(DestFd),
+ foldl(Fd, PosList,
+ fun(Bin, _) ->
+ ok = write(Dest, Bin)
+ end, ok),
+ close(Dest).
+
+foldl(_Fd, [], _Fun, Acc) ->
+ Acc;
+foldl(Fd, [Pos|Rest], Fun, Acc) ->
+ {ok, Bin} = couch_file:pread_iolist(Fd, Pos),
+ foldl(Fd, Rest, Fun, Fun(Bin, Acc)).
+
+foldl(Fd, PosList, <<>>, Fun, Acc) ->
+ foldl(Fd, PosList, Fun, Acc);
+foldl(Fd, PosList, Md5, Fun, Acc) ->
+ foldl(Fd, PosList, Md5, couch_util:md5_init(), Fun, Acc).
+
+foldl_decode(Fd, PosList, Md5, Enc, Fun, Acc) ->
+ {DecDataFun, DecEndFun} = case Enc of
+ gzip ->
+ ungzip_init();
+ identity ->
+ identity_enc_dec_funs()
+ end,
+ Result = foldl_decode(
+ DecDataFun, Fd, PosList, Md5, couch_util:md5_init(), Fun, Acc
+ ),
+ DecEndFun(),
+ Result.
+
+foldl(_Fd, [], Md5, Md5Acc, _Fun, Acc) ->
+ Md5 = couch_util:md5_final(Md5Acc),
+ Acc;
+foldl(Fd, [{Pos, _Size}], Md5, Md5Acc, Fun, Acc) -> % 0110 UPGRADE CODE
+ foldl(Fd, [Pos], Md5, Md5Acc, Fun, Acc);
+foldl(Fd, [Pos], Md5, Md5Acc, Fun, Acc) ->
+ {ok, Bin} = couch_file:pread_iolist(Fd, Pos),
+ Md5 = couch_util:md5_final(couch_util:md5_update(Md5Acc, Bin)),
+ Fun(Bin, Acc);
+foldl(Fd, [{Pos, _Size}|Rest], Md5, Md5Acc, Fun, Acc) ->
+ foldl(Fd, [Pos|Rest], Md5, Md5Acc, Fun, Acc);
+foldl(Fd, [Pos|Rest], Md5, Md5Acc, Fun, Acc) ->
+ {ok, Bin} = couch_file:pread_iolist(Fd, Pos),
+ foldl(Fd, Rest, Md5, couch_util:md5_update(Md5Acc, Bin), Fun, Fun(Bin, Acc)).
+
+range_foldl(Fd, PosList, From, To, Fun, Acc) ->
+ range_foldl(Fd, PosList, From, To, 0, Fun, Acc).
+
+range_foldl(_Fd, _PosList, _From, To, Off, _Fun, Acc) when Off >= To ->
+ Acc;
+range_foldl(Fd, [Pos|Rest], From, To, Off, Fun, Acc) when is_integer(Pos) -> % old-style attachment
+ {ok, Bin} = couch_file:pread_iolist(Fd, Pos),
+ range_foldl(Fd, [{Pos, iolist_size(Bin)}] ++ Rest, From, To, Off, Fun, Acc);
+range_foldl(Fd, [{_Pos, Size}|Rest], From, To, Off, Fun, Acc) when From > Off + Size ->
+ range_foldl(Fd, Rest, From, To, Off + Size, Fun, Acc);
+range_foldl(Fd, [{Pos, Size}|Rest], From, To, Off, Fun, Acc) ->
+ {ok, Bin} = couch_file:pread_iolist(Fd, Pos),
+ Bin1 = if
+ From =< Off andalso To >= Off + Size -> Bin; %% the whole block is covered
+ true ->
+ PrefixLen = clip(From - Off, 0, Size),
+ PostfixLen = clip(Off + Size - To, 0, Size),
+ MatchLen = Size - PrefixLen - PostfixLen,
+ <<_Prefix:PrefixLen/binary,Match:MatchLen/binary,_Postfix:PostfixLen/binary>> = iolist_to_binary(Bin),
+ Match
+ end,
+ range_foldl(Fd, Rest, From, To, Off + Size, Fun, Fun(Bin1, Acc)).
+
+clip(Value, Lo, Hi) ->
+ if
+ Value < Lo -> Lo;
+ Value > Hi -> Hi;
+ true -> Value
+ end.
+
+foldl_decode(_DecFun, _Fd, [], Md5, Md5Acc, _Fun, Acc) ->
+ Md5 = couch_util:md5_final(Md5Acc),
+ Acc;
+foldl_decode(DecFun, Fd, [{Pos, _Size}], Md5, Md5Acc, Fun, Acc) ->
+ foldl_decode(DecFun, Fd, [Pos], Md5, Md5Acc, Fun, Acc);
+foldl_decode(DecFun, Fd, [Pos], Md5, Md5Acc, Fun, Acc) ->
+ {ok, EncBin} = couch_file:pread_iolist(Fd, Pos),
+ Md5 = couch_util:md5_final(couch_util:md5_update(Md5Acc, EncBin)),
+ Bin = DecFun(EncBin),
+ Fun(Bin, Acc);
+foldl_decode(DecFun, Fd, [{Pos, _Size}|Rest], Md5, Md5Acc, Fun, Acc) ->
+ foldl_decode(DecFun, Fd, [Pos|Rest], Md5, Md5Acc, Fun, Acc);
+foldl_decode(DecFun, Fd, [Pos|Rest], Md5, Md5Acc, Fun, Acc) ->
+ {ok, EncBin} = couch_file:pread_iolist(Fd, Pos),
+ Bin = DecFun(EncBin),
+ Md5Acc2 = couch_util:md5_update(Md5Acc, EncBin),
+ foldl_decode(DecFun, Fd, Rest, Md5, Md5Acc2, Fun, Fun(Bin, Acc)).
+
+gzip_init(Options) ->
+ case couch_util:get_value(compression_level, Options, 0) of
+ Lvl when Lvl >= 1 andalso Lvl =< 9 ->
+ Z = zlib:open(),
+ % 15 = ?MAX_WBITS (defined in the zlib module)
+ % the 16 + ?MAX_WBITS formula was obtained by inspecting zlib:gzip/1
+ ok = zlib:deflateInit(Z, Lvl, deflated, 16 + 15, 8, default),
+ {
+ fun(Data) ->
+ zlib:deflate(Z, Data)
+ end,
+ fun() ->
+ Last = zlib:deflate(Z, [], finish),
+ ok = zlib:deflateEnd(Z),
+ ok = zlib:close(Z),
+ Last
+ end
+ };
+ _ ->
+ identity_enc_dec_funs()
+ end.
+
+ungzip_init() ->
+ Z = zlib:open(),
+ zlib:inflateInit(Z, 16 + 15),
+ {
+ fun(Data) ->
+ zlib:inflate(Z, Data)
+ end,
+ fun() ->
+ ok = zlib:inflateEnd(Z),
+ ok = zlib:close(Z)
+ end
+ }.
+
+identity_enc_dec_funs() ->
+ {
+ fun(Data) -> Data end,
+ fun() -> [] end
+ }.
+
+write(_Pid, <<>>) ->
+ ok;
+write(Pid, Bin) ->
+ gen_server:call(Pid, {write, Bin}, infinity).
+
+
+init({Fd, Options}) ->
+ {EncodingFun, EndEncodingFun} =
+ case couch_util:get_value(encoding, Options, identity) of
+ identity ->
+ identity_enc_dec_funs();
+ gzip ->
+ gzip_init(Options)
+ end,
+ {ok, #stream{
+ fd=Fd,
+ md5=couch_util:md5_init(),
+ identity_md5=couch_util:md5_init(),
+ encoding_fun=EncodingFun,
+ end_encoding_fun=EndEncodingFun,
+ max_buffer=couch_util:get_value(
+ buffer_size, Options, ?DEFAULT_BUFFER_SIZE)
+ }
+ }.
+
+terminate(_Reason, _Stream) ->
+ ok.
+
+handle_call({write, Bin}, _From, Stream) ->
+ BinSize = iolist_size(Bin),
+ #stream{
+ fd = Fd,
+ written_len = WrittenLen,
+ written_pointers = Written,
+ buffer_len = BufferLen,
+ buffer_list = Buffer,
+ max_buffer = Max,
+ md5 = Md5,
+ identity_md5 = IdenMd5,
+ identity_len = IdenLen,
+ encoding_fun = EncodingFun} = Stream,
+ if BinSize + BufferLen > Max ->
+ WriteBin = lists:reverse(Buffer, [Bin]),
+ IdenMd5_2 = couch_util:md5_update(IdenMd5, WriteBin),
+ case EncodingFun(WriteBin) of
+ [] ->
+ % case where the encoder did some internal buffering
+ % (zlib does it for example)
+ WrittenLen2 = WrittenLen,
+ Md5_2 = Md5,
+ Written2 = Written;
+ WriteBin2 ->
+ {ok, Pos, _} = couch_file:append_binary(Fd, WriteBin2),
+ WrittenLen2 = WrittenLen + iolist_size(WriteBin2),
+ Md5_2 = couch_util:md5_update(Md5, WriteBin2),
+ Written2 = [{Pos, iolist_size(WriteBin2)}|Written]
+ end,
+
+ {reply, ok, Stream#stream{
+ written_len=WrittenLen2,
+ written_pointers=Written2,
+ buffer_list=[],
+ buffer_len=0,
+ md5=Md5_2,
+ identity_md5=IdenMd5_2,
+ identity_len=IdenLen + BinSize}};
+ true ->
+ {reply, ok, Stream#stream{
+ buffer_list=[Bin|Buffer],
+ buffer_len=BufferLen + BinSize,
+ identity_len=IdenLen + BinSize}}
+ end;
+handle_call(close, _From, Stream) ->
+ #stream{
+ fd = Fd,
+ written_len = WrittenLen,
+ written_pointers = Written,
+ buffer_list = Buffer,
+ md5 = Md5,
+ identity_md5 = IdenMd5,
+ identity_len = IdenLen,
+ encoding_fun = EncodingFun,
+ end_encoding_fun = EndEncodingFun} = Stream,
+
+ WriteBin = lists:reverse(Buffer),
+ IdenMd5Final = couch_util:md5_final(couch_util:md5_update(IdenMd5, WriteBin)),
+ WriteBin2 = EncodingFun(WriteBin) ++ EndEncodingFun(),
+ Md5Final = couch_util:md5_final(couch_util:md5_update(Md5, WriteBin2)),
+ Result = case WriteBin2 of
+ [] ->
+ {lists:reverse(Written), WrittenLen, IdenLen, Md5Final, IdenMd5Final};
+ _ ->
+ {ok, Pos, _} = couch_file:append_binary(Fd, WriteBin2),
+ StreamInfo = lists:reverse(Written, [{Pos, iolist_size(WriteBin2)}]),
+ StreamLen = WrittenLen + iolist_size(WriteBin2),
+ {StreamInfo, StreamLen, IdenLen, Md5Final, IdenMd5Final}
+ end,
+ {stop, normal, Result, Stream}.
+
+handle_cast(_Msg, State) ->
+ {noreply,State}.
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+handle_info(_Info, State) ->
+ {noreply, State}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_task_status.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_task_status.erl b/apps/couch/src/couch_task_status.erl
new file mode 100644
index 0000000..e23b560
--- /dev/null
+++ b/apps/couch/src/couch_task_status.erl
@@ -0,0 +1,151 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_task_status).
+-behaviour(gen_server).
+
+% This module is used to track the status of long running tasks.
+% Long running tasks register themselves, via a call to add_task/1, and then
+% update their status properties via update/1. The status of a task is a
+% list of properties. Each property is a tuple, with the first element being
+% either an atom or a binary and the second element must be an EJSON value. When
+% a task updates its status, it can override some or all of its properties.
+% The properties {started_on, UnitTimestamp}, {updated_on, UnixTimestamp} and
+% {pid, ErlangPid} are automatically added by this module.
+% When a tracked task dies, its status will be automatically removed from
+% memory. To get the tasks list, call the all/0 function.
+
+-export([start_link/0, stop/0]).
+-export([all/0, add_task/1, update/1, get/1, set_update_frequency/1]).
+-export([is_task_added/0]).
+
+-export([init/1, terminate/2, code_change/3]).
+-export([handle_call/3, handle_cast/2, handle_info/2]).
+
+-include("couch_db.hrl").
+
+-define(set(L, K, V), lists:keystore(K, 1, L, {K, V})).
+
+
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+stop() ->
+ gen_server:cast(?MODULE, stop).
+
+
+all() ->
+ gen_server:call(?MODULE, all).
+
+
+add_task(Props) ->
+ put(task_status_update, {{0, 0, 0}, 0}),
+ Ts = timestamp(),
+ TaskProps = lists:ukeysort(
+ 1, [{started_on, Ts}, {updated_on, Ts} | Props]),
+ put(task_status_props, TaskProps),
+ gen_server:call(?MODULE, {add_task, TaskProps}).
+
+
+is_task_added() ->
+ is_list(erlang:get(task_status_props)).
+
+
+set_update_frequency(Msecs) ->
+ put(task_status_update, {{0, 0, 0}, Msecs * 1000}).
+
+
+update(Props) ->
+ MergeProps = lists:ukeysort(1, Props),
+ TaskProps = lists:ukeymerge(1, MergeProps, erlang:get(task_status_props)),
+ put(task_status_props, TaskProps),
+ maybe_persist(TaskProps).
+
+
+get(Props) when is_list(Props) ->
+ TaskProps = erlang:get(task_status_props),
+ [couch_util:get_value(P, TaskProps) || P <- Props];
+get(Prop) ->
+ TaskProps = erlang:get(task_status_props),
+ couch_util:get_value(Prop, TaskProps).
+
+
+maybe_persist(TaskProps0) ->
+ {LastUpdateTime, Frequency} = erlang:get(task_status_update),
+ case timer:now_diff(Now = now(), LastUpdateTime) >= Frequency of
+ true ->
+ put(task_status_update, {Now, Frequency}),
+ TaskProps = ?set(TaskProps0, updated_on, timestamp(Now)),
+ gen_server:cast(?MODULE, {update_status, self(), TaskProps});
+ false ->
+ ok
+ end.
+
+
+init([]) ->
+ % read configuration settings and register for configuration changes
+ ets:new(?MODULE, [ordered_set, protected, named_table]),
+ {ok, nil}.
+
+
+terminate(_Reason,_State) ->
+ ok.
+
+
+handle_call({add_task, TaskProps}, {From, _}, Server) ->
+ case ets:lookup(?MODULE, From) of
+ [] ->
+ true = ets:insert(?MODULE, {From, TaskProps}),
+ erlang:monitor(process, From),
+ {reply, ok, Server};
+ [_] ->
+ {reply, {add_task_error, already_registered}, Server}
+ end;
+handle_call(all, _, Server) ->
+ All = [
+ [{pid, ?l2b(pid_to_list(Pid))} | TaskProps]
+ ||
+ {Pid, TaskProps} <- ets:tab2list(?MODULE)
+ ],
+ {reply, All, Server}.
+
+
+handle_cast({update_status, Pid, NewProps}, Server) ->
+ case ets:lookup(?MODULE, Pid) of
+ [{Pid, _CurProps}] ->
+ ?LOG_DEBUG("New task status for ~p: ~p", [Pid, NewProps]),
+ true = ets:insert(?MODULE, {Pid, NewProps});
+ _ ->
+ % Task finished/died in the meanwhile and we must have received
+ % a monitor message before this call - ignore.
+ ok
+ end,
+ {noreply, Server};
+handle_cast(stop, State) ->
+ {stop, normal, State}.
+
+handle_info({'DOWN', _MonitorRef, _Type, Pid, _Info}, Server) ->
+ %% should we also erlang:demonitor(_MonitorRef), ?
+ ets:delete(?MODULE, Pid),
+ {noreply, Server}.
+
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+
+timestamp() ->
+ timestamp(now()).
+
+timestamp({Mega, Secs, _}) ->
+ Mega * 1000000 + Secs.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/e62a4fc1/apps/couch/src/couch_users_db.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_users_db.erl b/apps/couch/src/couch_users_db.erl
new file mode 100644
index 0000000..9b875ba
--- /dev/null
+++ b/apps/couch/src/couch_users_db.erl
@@ -0,0 +1,121 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_users_db).
+
+-export([before_doc_update/2, after_doc_read/2, strip_non_public_fields/1]).
+
+-include("couch_db.hrl").
+
+-define(NAME, <<"name">>).
+-define(PASSWORD, <<"password">>).
+-define(DERIVED_KEY, <<"derived_key">>).
+-define(PASSWORD_SCHEME, <<"password_scheme">>).
+-define(PBKDF2, <<"pbkdf2">>).
+-define(ITERATIONS, <<"iterations">>).
+-define(SALT, <<"salt">>).
+-define(replace(L, K, V), lists:keystore(K, 1, L, {K, V})).
+
+% If the request's userCtx identifies an admin
+% -> save_doc (see below)
+%
+% If the request's userCtx.name is null:
+% -> save_doc
+% // this is an anonymous user registering a new document
+% // in case a user doc with the same id already exists, the anonymous
+% // user will get a regular doc update conflict.
+% If the request's userCtx.name doesn't match the doc's name
+% -> 404 // Not Found
+% Else
+% -> save_doc
+before_doc_update(Doc, #db{user_ctx = UserCtx} = Db) ->
+ #user_ctx{name=Name} = UserCtx,
+ DocName = get_doc_name(Doc),
+ case (catch couch_db:check_is_admin(Db)) of
+ ok ->
+ save_doc(Doc);
+ _ when Name =:= DocName orelse Name =:= null ->
+ save_doc(Doc);
+ _ ->
+ throw(not_found)
+ end.
+
+% If newDoc.password == null || newDoc.password == undefined:
+% ->
+% noop
+% Else -> // calculate password hash server side
+% newDoc.password_sha = hash_pw(newDoc.password + salt)
+% newDoc.salt = salt
+% newDoc.password = null
+save_doc(#doc{body={Body}} = Doc) ->
+ case couch_util:get_value(?PASSWORD, Body) of
+ null -> % server admins don't have a user-db password entry
+ Doc;
+ undefined ->
+ Doc;
+ ClearPassword ->
+ Iterations = list_to_integer(couch_config:get("couch_httpd_auth", "iterations", "1000")),
+ Salt = couch_uuids:random(),
+ DerivedKey = couch_passwords:pbkdf2(ClearPassword, Salt, Iterations),
+ Body0 = [{?PASSWORD_SCHEME, ?PBKDF2}, {?ITERATIONS, Iterations}|Body],
+ Body1 = ?replace(Body0, ?DERIVED_KEY, DerivedKey),
+ Body2 = ?replace(Body1, ?SALT, Salt),
+ Body3 = proplists:delete(?PASSWORD, Body2),
+ Doc#doc{body={Body3}}
+ end.
+
+% If the doc is a design doc
+% If the request's userCtx identifies an admin
+% -> return doc
+% Else
+% -> 403 // Forbidden
+% If the request's userCtx identifies an admin
+% -> return doc
+% If the request's userCtx.name doesn't match the doc's name
+% -> 404 // Not Found
+% Else
+% -> return doc
+after_doc_read(#doc{id = <<?DESIGN_DOC_PREFIX, _/binary>>} = Doc, Db) ->
+ case (catch couch_db:check_is_admin(Db)) of
+ ok ->
+ Doc;
+ _ ->
+ throw({forbidden,
+ <<"Only administrators can view design docs in the users database.">>})
+ end;
+after_doc_read(Doc, #db{user_ctx = UserCtx} = Db) ->
+ #user_ctx{name=Name} = UserCtx,
+ DocName = get_doc_name(Doc),
+ case (catch couch_db:check_is_admin(Db)) of
+ ok ->
+ Doc;
+ _ when Name =:= DocName ->
+ Doc;
+ _ ->
+ Doc1 = strip_non_public_fields(Doc),
+ case Doc1 of
+ #doc{body={[]}} ->
+ throw(not_found);
+ _ ->
+ Doc1
+ end
+ end.
+
+get_doc_name(#doc{id= <<"org.couchdb.user:", Name/binary>>}) ->
+ Name;
+get_doc_name(_) ->
+ undefined.
+
+strip_non_public_fields(#doc{body={Props}}=Doc) ->
+ Public = re:split(couch_config:get("couch_httpd_auth", "public_fields", ""),
+ "\\s*,\\s*", [{return, binary}]),
+ Doc#doc{body={[{K, V} || {K, V} <- Props, lists:member(K, Public)]}}.