% 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_proc_manager).
-behaviour(gen_server).
-behaviour(config_listener).
-vsn(3).

-export([
    start_link/0,
    get_proc/3,
    get_proc/1,
    ret_proc/1,
    os_proc_idle/1,
    get_proc_count/0,
    get_stale_proc_count/0,
    new_proc/1,
    reload/0,
    terminate_stale_procs/0,
    get_servers_from_env/1
]).

-export([
    init/1,
    terminate/2,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    code_change/3
]).

-export([
    handle_config_change/5,
    handle_config_terminate/3
]).

-include_lib("couch/include/couch_db.hrl").

-define(PROCS, couch_proc_manager_procs).
-define(WAITERS, couch_proc_manager_waiters).
-define(OPENING, couch_proc_manager_opening).
-define(SERVERS, couch_proc_manager_servers).
-define(COUNTERS, couch_proc_manager_counters).
-define(IDLE_BY_DB, couch_proc_manager_idle_by_db).
-define(IDLE_ACCESS, couch_proc_manager_idle_access).
-define(RELISTEN_DELAY, 5000).

-record(state, {
    config,
    threshold_ts,
    hard_limit,
    soft_limit
}).

-type docid() :: iodata().
-type revision() :: {integer(), binary()}.

-record(client, {
    wait_key :: {binary(), integer(), gen_server:reply_tag()} | '_',
    from :: undefined | {pid(), gen_server:reply_tag()} | '_',
    lang :: binary() | '_',
    ddoc :: undefined | #doc{} | '_',
    db_key :: undefined | binary(),
    ddoc_key :: undefined | {DDocId :: docid(), Rev :: revision()} | '_'
}).

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

get_proc(#doc{body = {Props}} = DDoc, DbKey, {_DDocId, _Rev} = DDocKey) ->
    LangStr = couch_util:get_value(<<"language">>, Props, <<"javascript">>),
    Lang = couch_util:to_binary(LangStr),
    Client = #client{lang = Lang, ddoc = DDoc, db_key = DbKey, ddoc_key = DDocKey},
    Timeout = get_os_process_timeout(),
    gen_server:call(?MODULE, {get_proc, Client}, Timeout).

get_proc(LangStr) ->
    Lang = couch_util:to_binary(LangStr),
    Client = #client{lang = Lang},
    Timeout = get_os_process_timeout(),
    gen_server:call(?MODULE, {get_proc, Client}, Timeout).

ret_proc(#proc{} = Proc) ->
    gen_server:call(?MODULE, {ret_proc, Proc}, infinity).

os_proc_idle(Proc) when is_pid(Proc) ->
    gen_server:cast(?MODULE, {os_proc_idle, Proc}).

get_proc_count() ->
    try
        ets:info(?PROCS, size) + ets:info(?OPENING, size)
    catch
        error:badarg ->
            0
    end.

get_stale_proc_count() ->
    gen_server:call(?MODULE, get_stale_proc_count).

reload() ->
    gen_server:call(?MODULE, set_threshold_ts).

terminate_stale_procs() ->
    gen_server:call(?MODULE, terminate_stale_procs).

init([]) ->
    process_flag(trap_exit, true),
    ok = config:listen_for_changes(?MODULE, undefined),

    % Main process table. Pid -> #proc{}
    ets:new(?PROCS, [named_table, {read_concurrency, true}, {keypos, #proc.pid}]),

    % #client{} waiters ordered by {Lang, timestamp(), Ref}
    ets:new(?WAITERS, [named_table, ordered_set, {keypos, #client.wait_key}]),

    % Async process openers. Pid -> #client{}
    ets:new(?OPENING, [named_table]),

    % Configured language servers Lang -> Start MFA | Command
    ets:new(?SERVERS, [named_table]),

    % Idle Pids. Ordered to allow partial key lookups {Lang, DbKey, Pid} -> DDocs
    ets:new(?IDLE_BY_DB, [named_table, ordered_set]),

    % Idle Db tagged pids ordered by last use. {Lang, timestamp(), Pid} -> true
    ets:new(?IDLE_ACCESS, [named_table, ordered_set]),

    % Lang -> number of procs spawn for that lang
    ets:new(?COUNTERS, [named_table]),

    ets:insert(?SERVERS, get_servers_from_env("COUCHDB_QUERY_SERVER_")),
    ets:insert(?SERVERS, get_servers_from_env("COUCHDB_NATIVE_QUERY_SERVER_")),
    ets:insert(?SERVERS, [{"QUERY", {mango_native_proc, start_link, []}}]),
    maybe_configure_erlang_native_servers(),

    {ok, #state{
        config = get_proc_config(),
        threshold_ts = timestamp(),
        hard_limit = get_hard_limit(),
        soft_limit = get_soft_limit()
    }}.

terminate(_Reason, _State) ->
    foreach_proc(fun(#proc{pid = P}) -> couch_util:shutdown_sync(P) end).

handle_call(get_stale_proc_count, _From, State) ->
    #state{threshold_ts = T0} = State,
    MatchSpec = [{#proc{threshold_ts = '$1', _ = '_'}, [{'<', '$1', T0}], [true]}],
    {reply, ets:select_count(?PROCS, MatchSpec), State};
handle_call({get_proc, #client{} = Client}, From, State) ->
    add_waiting_client(Client#client{from = From}),
    ok = flush_waiters(State, Client#client.lang),
    {noreply, State};
handle_call({ret_proc, #proc{} = Proc}, From, State) ->
    #proc{client = Ref, pid = Pid} = Proc,
    erlang:demonitor(Ref, [flush]),
    gen_server:reply(From, true),
    case ets:lookup(?PROCS, Pid) of
        [#proc{} = ProcInt] ->
            ok = return_proc(State, ProcInt);
        [] ->
            % Proc must've died and we already
            % cleared it out of the table in
            % the handle_info clause.
            ok
    end,
    {noreply, State};
handle_call(set_threshold_ts, _From, State) ->
    Fun = fun
        (#proc{client = undefined} = Proc) -> ok = remove_proc(Proc);
        (_) -> ok
    end,
    ok = foreach_proc(Fun),
    {reply, ok, State#state{threshold_ts = timestamp()}};
handle_call(terminate_stale_procs, _From, #state{threshold_ts = Ts1} = State) ->
    Fun = fun
        (#proc{client = undefined, threshold_ts = Ts2} = Proc) ->
            case Ts1 > Ts2 of
                true -> ok = remove_proc(Proc);
                false -> ok
            end;
        (_) ->
            ok
    end,
    foreach_proc(Fun),
    {reply, ok, State};
handle_call(_Call, _From, State) ->
    {reply, ignored, State}.

handle_cast({os_proc_idle, Pid}, #state{soft_limit = SoftLimit} = State) ->
    case ets:lookup(?PROCS, Pid) of
        [#proc{client = undefined, db_key = DbKey, lang = Lang} = Proc] ->
            IsOverSoftLimit = get_count(Lang) >= SoftLimit,
            IsTagged = DbKey =/= undefined,
            case IsOverSoftLimit orelse IsTagged of
                true ->
                    couch_log:debug("Closing tagged or idle OS Process: ~p", [Pid]),
                    ok = remove_proc(Proc);
                false ->
                    ok
            end;
        _ ->
            State
    end,
    {noreply, State};
handle_cast(reload_config, State) ->
    NewState = State#state{
        config = get_proc_config(),
        hard_limit = get_hard_limit(),
        soft_limit = get_soft_limit()
    },
    maybe_configure_erlang_native_servers(),
    lists:foreach(
        fun({Lang, _}) ->
            ok = flush_waiters(NewState, Lang)
        end,
        ets:tab2list(?COUNTERS)
    ),
    {noreply, NewState};
handle_cast(_Msg, State) ->
    {noreply, State}.

handle_info(shutdown, State) ->
    {stop, shutdown, State};
handle_info({'EXIT', Pid, {spawn_ok, Proc0, undefined = _From}}, State) ->
    % Use ets:take/2 to assert that opener existed before removing. Also assert that
    % the pid matches and the the client was a bogus client
    [{Pid, #client{from = undefined}}] = ets:take(?OPENING, Pid),
    Proc = Proc0#proc{client = undefined},
    link(Proc#proc.pid),
    ets:insert(?PROCS, Proc),
    insert_idle_by_db(Proc),
    {noreply, State};
handle_info({'EXIT', Pid, {spawn_ok, Proc0, {ClientPid, _} = From}}, State) ->
    % Use ets:take/2 to assert that opener existed before removing
    [{Pid, #client{}}] = ets:take(?OPENING, Pid),
    link(Proc0#proc.pid),
    Proc = assign_proc(ClientPid, Proc0),
    gen_server:reply(From, {ok, Proc, State#state.config}),
    {noreply, State};
handle_info({'EXIT', Pid, spawn_error}, State) ->
    % Assert when removing that we always expect the opener to have been there
    [{Pid, #client{lang = Lang}}] = ets:take(?OPENING, Pid),
    dec_count(Lang),
    ok = flush_waiters(State, Lang),
    {noreply, State};
handle_info({'EXIT', Pid, Reason}, State) ->
    couch_log:info("~p ~p died ~p", [?MODULE, Pid, Reason]),
    case ets:lookup(?PROCS, Pid) of
        [#proc{} = Proc] ->
            ok = remove_proc(Proc),
            ok = flush_waiters(State, Proc#proc.lang);
        [] ->
            ok
    end,
    {noreply, State};
handle_info({'DOWN', Ref, _, _, _Reason}, #state{} = State) ->
    case ets:match_object(?PROCS, #proc{client = Ref, _ = '_'}) of
        [#proc{} = Proc] -> ok = return_proc(State, Proc);
        [] -> ok
    end,
    {noreply, State};
handle_info(restart_config_listener, State) ->
    ok = config:listen_for_changes(?MODULE, nil),
    {noreply, State};
handle_info(_Msg, State) ->
    {noreply, State}.

code_change(_OldVsn, #state{} = State, _Extra) ->
    {ok, State}.

handle_config_terminate(_, stop, _) ->
    ok;
handle_config_terminate(_Server, _Reason, _State) ->
    gen_server:cast(?MODULE, reload_config),
    erlang:send_after(?RELISTEN_DELAY, whereis(?MODULE), restart_config_listener).

handle_config_change("native_query_servers", _, _, _, _) ->
    gen_server:cast(?MODULE, reload_config),
    {ok, undefined};
handle_config_change("query_server_config", _, _, _, _) ->
    gen_server:cast(?MODULE, reload_config),
    {ok, undefined};
handle_config_change(_, _, _, _, _) ->
    {ok, undefined}.

find_proc(#client{ddoc_key = undefined} = Client, _CanSpawn) ->
    #client{lang = Lang} = Client,
    % Find an unowned process first, if that fails find an owned one
    case find_proc(Lang, undefined, '_') of
        {ok, Proc} ->
            {ok, Proc};
        not_found ->
            case find_proc(Lang, '_', '_') of
                {ok, Proc} ->
                    {ok, Proc};
                Else ->
                    Else
            end;
        Else ->
            Else
    end;
find_proc(#client{} = Client, CanSpawn) ->
    #client{
        lang = Lang,
        ddoc = DDoc,
        db_key = DbKey,
        ddoc_key = DDocKey
    } = Client,
    case find_proc(Lang, DbKey, DDocKey) of
        not_found ->
            % Find a ddoc process used by the same db at least
            case find_proc(Lang, DbKey, '_') of
                {ok, Proc} ->
                    teach_ddoc(DDoc, DbKey, DDocKey, Proc);
                not_found ->
                    % Pick a process not used by any ddoc
                    case find_proc(Lang, undefined, '_') of
                        {ok, Proc} ->
                            replenish_untagged_pool(Lang, CanSpawn),
                            teach_ddoc(DDoc, DbKey, DDocKey, Proc);
                        Else ->
                            Else
                    end;
                Else ->
                    Else
            end;
        {ok, Proc} ->
            {ok, Proc};
        Else ->
            Else
    end.

find_proc(Lang, DbPat, DDocKey) when
    DbPat =:= '_' orelse DbPat =:= undefined orelse is_binary(DbPat),
    DDocKey =:= '_' orelse is_tuple(DDocKey)
->
    Pattern = {{Lang, DbPat, '$1'}, '$2'},
    Guards =
        case DDocKey of
            '_' -> [];
            {_, _} -> [{map_get, {const, DDocKey}, '$2'}]
        end,
    MSpec = [{Pattern, Guards, ['$1']}],
    case ets:select_reverse(?IDLE_BY_DB, MSpec, 1) of
        '$end_of_table' ->
            not_found;
        {[Pid], _Continuation} when is_pid(Pid) ->
            [#proc{client = undefined} = Proc] = ets:lookup(?PROCS, Pid),
            % Once it's found it's not idle any longer and it might be
            % "tought" a new ddoc, so its db_key might change
            remove_idle_by_db(Proc),
            remove_idle_access(Proc),
            {ok, Proc}
    end.

spawn_proc(#client{} = Client) ->
    Pid = spawn_link(?MODULE, new_proc, [Client]),
    ets:insert(?OPENING, {Pid, Client}),
    inc_count(Client#client.lang).

% This instance was spawned without a client to replenish
% the untagged pool asynchronously
new_proc(#client{from = undefined} = Client) ->
    #client{lang = Lang} = Client,
    Resp =
        try
            case new_proc_int(undefined, Lang) of
                {ok, Proc} ->
                    {spawn_ok, Proc, undefined};
                _Error ->
                    spawn_error
            end
        catch
            _:_ ->
                spawn_error
        end,
    exit(Resp);
new_proc(#client{ddoc = undefined, ddoc_key = undefined} = Client) ->
    #client{from = From, lang = Lang} = Client,
    Resp =
        try
            case new_proc_int(From, Lang) of
                {ok, Proc} ->
                    {spawn_ok, Proc, From};
                Error ->
                    gen_server:reply(From, {error, Error}),
                    spawn_error
            end
        catch
            _:_ ->
                spawn_error
        end,
    exit(Resp);
new_proc(#client{} = Client) ->
    #client{
        from = From,
        lang = Lang,
        ddoc = DDoc,
        db_key = DbKey,
        ddoc_key = DDocKey
    } = Client,
    Resp =
        try
            case new_proc_int(From, Lang) of
                {ok, NewProc} ->
                    {ok, Proc} = teach_ddoc(DDoc, DbKey, DDocKey, NewProc),
                    {spawn_ok, Proc, From};
                Error ->
                    gen_server:reply(From, {error, Error}),
                    spawn_error
            end
        catch
            _:_ ->
                spawn_error
        end,
    exit(Resp).

replenish_untagged_pool(Lang, _CanSpawn = true) ->
    % After an untagged instance is tagged, we try to replenish
    % the untagged pool asynchronously. Here we are using a "bogus"
    % #client{} with an undefined from field.
    ok = spawn_proc(#client{lang = Lang, from = undefined});
replenish_untagged_pool(_Lang, _CanSpawn = false) ->
    ok.

reap_idle(Num, <<_/binary>> = Lang) when is_integer(Num), Num >= 1 ->
    case ets:match_object(?IDLE_ACCESS, {{Lang, '_', '_'}, '_'}, Num) of
        '$end_of_table' ->
            0;
        {Objects = [_ | _], _} ->
            ok = reap_idle(Objects),
            length(Objects)
    end.

reap_idle([]) ->
    ok;
reap_idle([{{_Lang, _Ts, Pid}, true} | Rest]) ->
    case ets:lookup(?PROCS, Pid) of
        % Do an extra assert that client is undefined
        [#proc{client = undefined} = Proc] ->
            ok = remove_proc(Proc);
        [] ->
            ok
    end,
    reap_idle(Rest).

insert_idle_access(#proc{db_key = undefined}, _Ts) ->
    % Only tagged proc are index in ?IDLE_ACCESS
    ok;
insert_idle_access(#proc{db_key = <<_/binary>>} = Proc, Ts) ->
    #proc{lang = Lang, pid = Pid} = Proc,
    % Lang is used for partially bound key access
    % Pid is for uniqueness as time is not strictly monotonic
    true = ets:insert_new(?IDLE_ACCESS, {{Lang, Ts, Pid}, true}),
    ok.

remove_idle_access(#proc{db_key = undefined}) ->
    % Only tagged procs are indexed in ?IDLE_ACCESS
    ok;
remove_idle_access(#proc{db_key = <<_/binary>>} = Proc) ->
    #proc{last_use_ts = Ts, lang = Lang, pid = Pid} = Proc,
    true = ets:delete(?IDLE_ACCESS, {Lang, Ts, Pid}),
    ok.

insert_idle_by_db(#proc{} = Proc) ->
    #proc{lang = Lang, pid = Pid, db_key = Db, ddoc_keys = #{} = DDocs} = Proc,
    % An extra assert that only expect to insert a new object
    true = ets:insert_new(?IDLE_BY_DB, {{Lang, Db, Pid}, DDocs}),
    ok.

remove_idle_by_db(#proc{} = Proc) ->
    #proc{lang = Lang, pid = Pid, db_key = Db} = Proc,
    true = ets:delete(?IDLE_BY_DB, {Lang, Db, Pid}),
    ok.

split_string_if_longer(String, Pos) ->
    case length(String) > Pos of
        true -> lists:split(Pos, String);
        false -> false
    end.

split_by_char(String, Char) ->
    %% 17.5 doesn't have string:split
    %% the function doesn't handle errors
    %% it is designed to be used only in specific context
    Pos = string:chr(String, Char),
    {Key, [_Eq | Value]} = lists:split(Pos - 1, String),
    {Key, Value}.

get_servers_from_env(Spec) ->
    SpecLen = length(Spec),
    % loop over os:getenv(), match SPEC_
    lists:filtermap(
        fun(EnvStr) ->
            case split_string_if_longer(EnvStr, SpecLen) of
                {Spec, Rest} ->
                    {true, split_by_char(Rest, $=)};
                _ ->
                    false
            end
        end,
        os:getenv()
    ).

get_query_server(LangStr) ->
    case ets:lookup(?SERVERS, string:to_upper(LangStr)) of
        [{_, Command}] -> Command;
        _ -> undefined
    end.

native_query_server_enabled() ->
    % 1. [native_query_server] enable_erlang_query_server = true | false
    % 2. if [native_query_server] erlang == {couch_native_process, start_link, []} -> pretend true as well
    NativeEnabled = config:get_boolean("native_query_servers", "enable_erlang_query_server", false),
    NativeLegacyConfig = config:get("native_query_servers", "erlang", ""),
    NativeLegacyEnabled = NativeLegacyConfig =:= "{couch_native_process, start_link, []}",
    NativeEnabled orelse NativeLegacyEnabled.

maybe_configure_erlang_native_servers() ->
    case native_query_server_enabled() of
        true ->
            ets:insert(?SERVERS, [
                {"ERLANG", {couch_native_process, start_link, []}}
            ]);
        _Else ->
            ok
    end.

new_proc_int(From, Lang) when is_binary(Lang) ->
    LangStr = binary_to_list(Lang),
    case get_query_server(LangStr) of
        undefined ->
            case From of
                undefined -> ok;
                {_, _} -> gen_server:reply(From, {unknown_query_language, Lang})
            end;
        {M, F, A} ->
            {ok, Pid} = apply(M, F, A),
            make_proc(Pid, Lang, M);
        Command ->
            {ok, Pid} = couch_os_process:start_link(Command),
            make_proc(Pid, Lang, couch_os_process)
    end.

teach_ddoc(DDoc, DbKey, {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.
    JsonDoc = couch_doc:to_json_obj(DDoc, []),
    Prompt = [<<"ddoc">>, <<"new">>, DDocId, JsonDoc],
    true = couch_query_servers:proc_prompt(Proc, Prompt),
    % we should remove any other ddocs keys for this docid
    % because the query server overwrites without the rev
    Keys2 = maps:filter(fun({Id, _}, true) -> Id =/= DDocId end, Keys),

    % add ddoc to the proc
    {ok, Proc#proc{db_key = DbKey, ddoc_keys = Keys2#{DDocKey => true}}}.

make_proc(Pid, Lang, Mod) when is_binary(Lang) ->
    Proc = #proc{
        lang = Lang,
        pid = Pid,
        prompt_fun = {Mod, prompt},
        set_timeout_fun = {Mod, set_timeout},
        stop_fun = {Mod, stop},
        threshold_ts = timestamp(),
        last_use_ts = timestamp()
    },
    unlink(Pid),
    {ok, Proc}.

assign_proc(Pid, #proc{client = undefined} = Proc0) when is_pid(Pid) ->
    Proc = Proc0#proc{client = erlang:monitor(process, Pid)},
    % It's important to insert the proc here instead of doing an update_element
    % as we might have updated the db_key or ddoc_keys in teach_ddoc/4
    ets:insert(?PROCS, Proc),
    Proc;
assign_proc(#client{} = Client, #proc{client = undefined} = Proc) ->
    {Pid, _} = Client#client.from,
    assign_proc(Pid, Proc).

return_proc(#state{} = State, #proc{} = Proc) ->
    #proc{pid = Pid, lang = Lang} = Proc,
    case is_process_alive(Pid) of
        true ->
            case Proc#proc.threshold_ts < State#state.threshold_ts of
                true ->
                    ok = remove_proc(Proc);
                false ->
                    gen_server:cast(Pid, garbage_collect),
                    Ts = timestamp(),
                    true = ets:update_element(?PROCS, Pid, [
                        {#proc.client, undefined},
                        {#proc.last_use_ts, Ts}
                    ]),
                    Proc1 = Proc#proc{client = undefined, last_use_ts = Ts},
                    insert_idle_access(Proc1, Ts),
                    insert_idle_by_db(Proc1)
            end;
        false ->
            ok = remove_proc(Proc)
    end,
    ok = flush_waiters(State, Lang).

remove_proc(#proc{pid = Pid} = Proc) ->
    remove_idle_access(Proc),
    remove_idle_by_db(Proc),
    ets:delete(?PROCS, Pid),
    case is_process_alive(Pid) of
        true ->
            unlink(Pid),
            gen_server:cast(Pid, stop);
        false ->
            ok
    end,
    dec_count(Proc#proc.lang).

flush_waiters(#state{} = State, Lang) ->
    #state{hard_limit = HardLimit, config = {[_ | _] = Cfg}} = State,
    TimeoutMSec = couch_util:get_value(<<"timeout">>, Cfg),
    Timeout = erlang:convert_time_unit(TimeoutMSec, millisecond, native),
    StaleLimit = timestamp() - Timeout,
    case get_waiting_client(Lang) of
        #client{wait_key = {_, T, _}} = Client when is_integer(T), T < StaleLimit ->
            % Client waited too long and the gen_server call timeout
            % likey fired already, don't bother allocating a process for it
            remove_waiting_client(Client),
            flush_waiters(State, Lang);
        #client{from = From} = Client ->
            CanSpawn = get_count(Lang) < HardLimit,
            case find_proc(Client, CanSpawn) of
                {ok, Proc0} ->
                    Proc = assign_proc(Client, Proc0),
                    gen_server:reply(From, {ok, Proc, State#state.config}),
                    remove_waiting_client(Client),
                    flush_waiters(State, Lang);
                {error, Error} ->
                    gen_server:reply(From, {error, Error}),
                    remove_waiting_client(Client),
                    flush_waiters(State, Lang);
                not_found when CanSpawn ->
                    ok = spawn_proc(Client),
                    remove_waiting_client(Client),
                    flush_waiters(State, Lang);
                not_found ->
                    % 10% of limit
                    ReapBatch = round(HardLimit * 0.1 + 1),
                    case reap_idle(ReapBatch, Lang) of
                        N when is_integer(N), N > 0 ->
                            % We may have room available to spawn
                            case get_count(Lang) < HardLimit of
                                true ->
                                    ok = spawn_proc(Client),
                                    remove_waiting_client(Client),
                                    flush_waiters(State, Lang);
                                false ->
                                    ok
                            end;
                        0 ->
                            ok
                    end
            end;
        undefined ->
            ok
    end.

add_waiting_client(#client{from = {_Pid, Tag}, lang = Lang} = Client) ->
    % Use Lang in the key first since we can look it up using a partially bound
    % in get_waiting_client/2. Use the reply tag to provide uniqueness.
    Key = {Lang, timestamp(), Tag},
    true = ets:insert_new(?WAITERS, Client#client{wait_key = Key}).

-spec get_waiting_client(Lang :: binary()) -> undefined | #client{}.
get_waiting_client(Lang) ->
    % Use a partially bound key (Lang) to avoid scanning unrelated procs
    Key = {Lang, '_', '_'},
    case ets:match_object(?WAITERS, #client{wait_key = Key, _ = '_'}, 1) of
        '$end_of_table' ->
            undefined;
        {[#client{} = Client], _} ->
            Client
    end.

remove_waiting_client(#client{wait_key = Key}) ->
    ets:delete(?WAITERS, Key).

get_proc_config() ->
    Limit = config:get_boolean("query_server_config", "reduce_limit", true),
    Timeout = get_os_process_timeout(),
    {[
        {<<"reduce_limit">>, Limit},
        {<<"timeout">>, Timeout}
    ]}.

get_hard_limit() ->
    config:get_integer("query_server_config", "os_process_limit", 100).

get_soft_limit() ->
    config:get_integer("query_server_config", "os_process_soft_limit", 100).

get_os_process_timeout() ->
    config:get_integer("couchdb", "os_process_timeout", 5000).

timestamp() ->
    erlang:monotonic_time().

foreach_proc(Fun) when is_function(Fun, 1) ->
    FoldFun = fun(#proc{} = Proc, ok) ->
        Fun(Proc),
        ok
    end,
    ok = ets:foldl(FoldFun, ok, ?PROCS).

inc_count(Lang) ->
    ets:update_counter(?COUNTERS, Lang, 1, {Lang, 0}),
    ok.

dec_count(Lang) ->
    ets:update_counter(?COUNTERS, Lang, -1, {Lang, 0}),
    ok.

get_count(Lang) ->
    case ets:lookup(?COUNTERS, Lang) of
        [{_, Count}] when is_integer(Count) ->
            Count;
        [] ->
            0
    end.