src/couch/test/eunit/couchdb_os_proc_pool.erl - couchdb - Git at Google

 % 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(couchdb_os_proc_pool).

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

 -define(TDEF_FE(Name), fun(Arg) -> {atom_to_list(Name), ?_test(Name(Arg))} end).

 -define(TIMEOUT, 1000).


 setup() ->
     Ctx = test_util:start_couch(),
     meck:new(couch_os_process, [passthrough]),
     meck:new(couch_proc_manager, [passthrough]),
     ok = setup_config(),
     Ctx.

 teardown(Ctx) ->
     ok = teardown_config(),
     meck:unload(),
     test_util:stop_couch(Ctx),
     ok.

 os_proc_pool_test_() ->
     {
         "OS processes pool tests",
         {
             foreach,
             fun setup/0,
             fun teardown/1,
             [
                 ?TDEF_FE(should_block_new_proc_on_full_pool),
                 ?TDEF_FE(should_free_slot_on_proc_unexpected_exit),
                 ?TDEF_FE(should_reuse_known_proc),
                 ?TDEF_FE(should_process_waiting_queue_as_fifo),
                 ?TDEF_FE(should_reduce_pool_on_idle_os_procs),
                 ?TDEF_FE(should_reduce_pool_of_tagged_processes_on_idle),
                 ?TDEF_FE(should_not_return_broken_process_to_the_pool),
                 ?TDEF_FE(oldest_tagged_process_is_reaped),
                 ?TDEF_FE(untagged_process_is_replenished),
                 ?TDEF_FE(exact_ddoc_tagged_process_is_picked_first),
                 ?TDEF_FE(db_tagged_process_is_second_choice),
                 ?TDEF_FE(if_no_tagged_process_found_new_must_be_spawned),
                 ?TDEF_FE(db_tag_none_works),
                 ?TDEF_FE(stale_procs_are_cleaned),
                 ?TDEF_FE(bad_query_language)
             ]
         }
     }.

 should_block_new_proc_on_full_pool(_) ->
     Client1 = spawn_client(),
     Client2 = spawn_client(),
     Client3 = spawn_client(),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),
     ?assertEqual(ok, ping_client(Client3)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),
     Proc3 = get_client_proc(Client3, "3"),

     ?assertNotEqual(Proc1, Proc2),
     ?assertNotEqual(Proc2, Proc3),
     ?assertNotEqual(Proc3, Proc1),

     Client4 = spawn_client(),
     ?assertEqual(timeout, ping_client(Client4)),

     ?assertEqual(ok, stop_client(Client1)),
     ?assertEqual(ok, ping_client(Client4)),

     Proc4 = get_client_proc(Client4, "4"),

     ?assertEqual(Proc1#proc.pid, Proc4#proc.pid),
     ?assertNotEqual(Proc1#proc.client, Proc4#proc.client),

     stop_clients([Client2, Client3, Client4]).

 should_free_slot_on_proc_unexpected_exit(_) ->
     Client1 = spawn_client(),
     Client2 = spawn_client(),
     Client3 = spawn_client(),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),
     ?assertEqual(ok, ping_client(Client3)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),
     Proc3 = get_client_proc(Client3, "3"),

     ?assertNotEqual(Proc1#proc.pid, Proc2#proc.pid),
     ?assertNotEqual(Proc1#proc.client, Proc2#proc.client),
     ?assertNotEqual(Proc2#proc.pid, Proc3#proc.pid),
     ?assertNotEqual(Proc2#proc.client, Proc3#proc.client),
     ?assertNotEqual(Proc3#proc.pid, Proc1#proc.pid),
     ?assertNotEqual(Proc3#proc.client, Proc1#proc.client),

     ?assertEqual(ok, kill_client(Client1)),

     Client4 = spawn_client(),
     ?assertEqual(ok, ping_client(Client4)),

     Proc4 = get_client_proc(Client4, "4"),

     ?assertEqual(Proc4#proc.pid, Proc1#proc.pid),
     ?assertNotEqual(Proc4#proc.client, Proc1#proc.client),
     ?assertNotEqual(Proc2#proc.pid, Proc4#proc.pid),
     ?assertNotEqual(Proc2#proc.client, Proc4#proc.client),
     ?assertNotEqual(Proc3#proc.pid, Proc4#proc.pid),
     ?assertNotEqual(Proc3#proc.client, Proc4#proc.client),

     stop_clients([Client2, Client3, Client4]).

 should_reuse_known_proc(_) ->
     Db = <<"db">>,
     Client1 = spawn_client(Db, <<"ddoc1">>),
     Client2 = spawn_client(Db, <<"ddoc2">>),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),
     ?assertNotEqual(Proc1#proc.pid, Proc2#proc.pid),

     ?assertEqual(ok, stop_client(Client1)),
     ?assertEqual(ok, stop_client(Client2)),
     ?assert(is_process_alive(Proc1#proc.pid)),
     ?assert(is_process_alive(Proc2#proc.pid)),

     Client1Again = spawn_client(Db, <<"ddoc1">>),
     ?assertEqual(ok, ping_client(Client1Again)),
     Proc1Again = get_client_proc(Client1Again, "1-again"),
     ?assertEqual(Proc1#proc.pid, Proc1Again#proc.pid),
     ?assertNotEqual(Proc1#proc.client, Proc1Again#proc.client),
     ?assertEqual(ok, stop_client(Client1Again)).

 should_process_waiting_queue_as_fifo(_) ->
     Db = <<"db">>,
     meck:reset(couch_proc_manager),
     Client1 = spawn_client(Db, <<"ddoc1">>),
     meck:wait(1, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),
     Client2 = spawn_client(Db, <<"ddoc2">>),
     meck:wait(2, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),
     Client3 = spawn_client(Db, <<"ddoc3">>),
     meck:wait(3, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),
     Client4 = spawn_client(Db, <<"ddoc4">>),
     meck:wait(4, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),
     Client5 = spawn_client(Db, <<"ddoc5">>),
     meck:wait(5, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),
     ?assertEqual(ok, ping_client(Client3)),
     ?assertEqual(timeout, ping_client(Client4)),
     ?assertEqual(timeout, ping_client(Client5)),

     Proc1 = get_client_proc(Client1, "1"),
     ?assertEqual(ok, stop_client(Client1)),
     ?assertEqual(ok, ping_client(Client4)),
     Proc4 = get_client_proc(Client4, "4"),

     ?assertNotEqual(Proc4#proc.client, Proc1#proc.client),
     ?assertEqual(Proc1#proc.pid, Proc4#proc.pid),
     ?assertEqual(timeout, ping_client(Client5)),

     ?assertEqual(ok, stop_client(Client2)),
     ?assertEqual(ok, stop_client(Client3)),
     ?assertEqual(ok, stop_client(Client4)),
     ?assertEqual(ok, stop_client(Client5)).

 should_reduce_pool_on_idle_os_procs(_) ->
     %% os_process_idle_limit is in sec
     cfg_set("os_process_idle_limit", "1"),

     Db = undefined,
     Client1 = spawn_client(Db, <<"ddoc1">>),
     Client2 = spawn_client(Db, <<"ddoc2">>),
     Client3 = spawn_client(Db, <<"ddoc3">>),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),
     ?assertEqual(ok, ping_client(Client3)),

     ?assertEqual(3, couch_proc_manager:get_proc_count()),

     ?assertEqual(ok, stop_client(Client1)),
     ?assertEqual(ok, stop_client(Client2)),
     ?assertEqual(ok, stop_client(Client3)),

     % granularity of idle limit is in seconds
     timer:sleep(1000),
     wait_process_count(1).

 should_reduce_pool_of_tagged_processes_on_idle(_) ->
     %% os_process_idle_limit is in sec
     cfg_set("os_process_idle_limit", "1"),

     Db = <<"reduce_pool_on_idle_db">>,
     Client1 = spawn_client(Db, <<"ddoc1">>),
     Client2 = spawn_client(Db, <<"ddoc2">>),
     Client3 = spawn_client(Db, <<"ddoc3">>),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),
     ?assertEqual(ok, ping_client(Client3)),

     ?assertEqual(3, couch_proc_manager:get_proc_count()),

     stop_clients([Client1, Client2, Client3]),

     timer:sleep(1000),
     wait_process_count(0).

 oldest_tagged_process_is_reaped(_) ->
     Client1 = spawn_client(<<"db1">>, <<"ddoc1">>),
     Client2 = spawn_client(<<"db2">>, <<"ddoc1">>),
     Client3 = spawn_client(<<"db3">>, <<"ddoc1">>),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),
     ?assertEqual(ok, ping_client(Client3)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),
     Proc3 = get_client_proc(Client3, "3"),

     ?assert(all_alive_all_different([Proc1, Proc2, Proc3])),

     stop_clients([Client1, Client2, Client3]),

     % All procs should be released back into the pool
     wait_tagged_idle_count(3),

     % Processes should be alive
     ?assert(all_alive([Proc1, Proc2, Proc3])),

     % Spawning a new tagged proc with a different tag should kill
     % the oldest unused proc and spawn a new one
     Client4 = spawn_client(<<"db4">>, <<"ddoc1">>),
     ?assertEqual(ok, ping_client(Client4)),
     Proc4 = get_client_proc(Client4, "4"),

     ?assert(all_alive_all_different([Proc2, Proc3, Proc4])),
     ?assertNot(is_process_alive(Proc1#proc.pid)),

     ?assertEqual(ok, stop_client(Client4)).

 untagged_process_is_replenished(_) ->
     Client1 = spawn_client(),
     Client2 = spawn_client(),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),

     ?assert(all_alive_all_different([Proc1, Proc2])),

     stop_clients([Client1, Client2]),

     % All procs should be released back into the pool
     % and they are now all untagged idle
     wait_idle_count(2),
     ?assertEqual(0, tagged_idle_count()),

     % Processes should still be alive
     ?assert(all_alive([Proc1, Proc2])),

     % Spawning a new tagged proc should tag one of the procs
     % and also asynchronously replenish the untagged pool
     Client3 = spawn_client(<<"db">>, <<"ddoc1">>),
     ?assertEqual(ok, ping_client(Client3)),

     % The process is one of the previously untagged ones
     Proc3 = get_client_proc(Client3, "3"),
     Pid3 = Proc3#proc.pid,
     ?assert(lists:member(Pid3, proc_pids([Proc1, Proc2]))),

     % wait for replinishment
     wait_idle_count(2),

     ?assertEqual(ok, stop_client(Client3)).

 exact_ddoc_tagged_process_is_picked_first(_) ->
     Client1 = spawn_client(<<"db">>, <<"ddoc1">>),
     Client2 = spawn_client(<<"db">>, <<"ddoc2">>),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),

     ?assert(all_alive_all_different([Proc1, Proc2])),

     stop_clients([Client1, Client2]),

     % All procs should be released back into the pool
     % and they now tagged and idle
     wait_tagged_idle_count(2),
     wait_idle_count(2),

     % Processes should still be alive
     ?assert(all_alive([Proc1, Proc2])),

     % Spawning a new tagged proc should pick the one with
     % matching ddoc
     Client3 = spawn_client(<<"db">>, <<"ddoc1">>),
     ?assertEqual(ok, ping_client(Client3)),
     Proc3 = get_client_proc(Client3, "3"),
     ?assertEqual(Proc1#proc.pid, Proc3#proc.pid),

     ?assertEqual(ok, stop_client(Client3)).

 db_tagged_process_is_second_choice(_) ->
     Client1 = spawn_client(<<"db1">>, <<"ddoc1">>),
     Client2 = spawn_client(<<"db2">>, <<"ddoc2">>),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),

     ?assert(all_alive_all_different([Proc1, Proc2])),

     stop_clients([Client1, Client2]),

     % All procs should be released back into the pool
     % and they now tagged and idle
     wait_tagged_idle_count(2),
     wait_idle_count(2),

     % Processes should still be alive
     ?assert(all_alive([Proc1, Proc2])),

     % Spawning a new tagged proc should pick the one with
     % the matching ddoc
     Client3 = spawn_client(<<"db1">>, <<"ddoc3">>),
     ?assertEqual(ok, ping_client(Client3)),
     Proc3 = get_client_proc(Client3, "3"),
     ?assertEqual(Proc1#proc.pid, Proc3#proc.pid),

     ?assertEqual(ok, stop_client(Client3)).

 if_no_tagged_process_found_new_must_be_spawned(_) ->
     Client1 = spawn_client(<<"db1">>, <<"ddoc">>),
     Client2 = spawn_client(<<"db2">>, <<"ddoc">>),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),

     ?assert(all_alive_all_different([Proc1, Proc2])),

     stop_clients([Client1, Client2]),

     % All procs should be released back into the pool
     % and they now tagged and idle
     wait_tagged_idle_count(2),
     wait_idle_count(2),

     % Processes should still be alive
     ?assert(all_alive([Proc1, Proc2])),

     % If new tagged process with new db should spawn
     % new process never pick up an existing one
     Client3 = spawn_client(<<"db3">>, <<"ddoc">>),
     ?assertEqual(ok, ping_client(Client3)),
     Proc3 = get_client_proc(Client3, "3"),
     ?assertNotEqual(Proc1#proc.pid, Proc3#proc.pid),
     ?assertNotEqual(Proc2#proc.pid, Proc3#proc.pid),

     % db1 and db2 procs should still be sitting idle
     ?assertEqual(2, tagged_idle_count()),

     % After 3rd proc returns to the pool there should
     % be 3 tagged idle processes
     ?assertEqual(ok, stop_client(Client3)),
     wait_tagged_idle_count(3),
     ?assertEqual(3, idle_count()).

 db_tag_none_works(_) ->
     cfg_set("db_tag", "none"),
     Client1 = spawn_client(undefined, <<"ddoc1">>),
     Client2 = spawn_client(undefined, <<"ddoc2">>),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),

     ?assert(all_alive_all_different([Proc1, Proc2])),

     stop_clients([Client1, Client2]),

     % All procs should be released back into the pool
     % they should be untagged effectively
     wait_idle_count(2),
     ?assertEqual(0, tagged_idle_count()),

     % Processes should still be alive
     ?assert(all_alive([Proc1, Proc2])),

     % If new tagged process with new db should spawn
     % new process and pick based on ddoc id matching
     Client3 = spawn_client(undefined, <<"ddoc1">>),
     ?assertEqual(ok, ping_client(Client3)),
     Proc3 = get_client_proc(Client3, "3"),
     ?assertEqual(Proc1#proc.pid, Proc3#proc.pid),
     ?assertNotEqual(Proc2#proc.pid, Proc3#proc.pid),

     wait_idle_count(1),

     % After 3rd client stop there should be 2 idle
     % untagged procs
     ?assertEqual(ok, stop_client(Client3)),
     wait_idle_count(2),
     ?assertEqual(0, tagged_idle_count()).

 stale_procs_are_cleaned(_) ->
     Client1 = spawn_client(),
     Client2 = spawn_client(),

     ?assertEqual(ok, ping_client(Client1)),
     ?assertEqual(ok, ping_client(Client2)),

     Proc1 = get_client_proc(Client1, "1"),
     Proc2 = get_client_proc(Client2, "2"),

     ?assert(all_alive_all_different([Proc1, Proc2])),

     ?assertEqual(0, couch_proc_manager:get_stale_proc_count()),
     ?assertEqual(ok, couch_proc_manager:reload()),
     ?assertEqual(2, couch_proc_manager:get_stale_proc_count()),

     stop_clients([Client1, Client2]),
     ?assertEqual(ok, couch_proc_manager:terminate_stale_procs()),
     wait_idle_count(0),
     ?assertEqual(0, couch_proc_manager:get_proc_count()).

 bad_query_language(_) ->
     Expect = {unknown_query_language, <<"bad">>},
     ?assertThrow(Expect, couch_query_servers:get_os_process(<<"bad">>)).

 should_not_return_broken_process_to_the_pool(_) ->
     cfg_set("os_process_soft_limit", "1"),
     cfg_set("os_process_limit", "1"),

     Db = <<"thedb">>,
     DDoc1 = ddoc(<<"_design/ddoc1">>),

     meck:reset(couch_os_process),

     ?assertEqual(0, couch_proc_manager:get_proc_count()),
     ok = couch_query_servers:with_ddoc_proc(Db, DDoc1, fun(_) -> ok end),
     ?assertEqual(0, meck:num_calls(couch_os_process, stop, 1)),
     ?assertEqual(1, couch_proc_manager:get_proc_count()),

     ?assertError(
         bad,
         couch_query_servers:with_ddoc_proc(Db, DDoc1, fun(_) ->
             error(bad)
         end)
     ),
     ?assertEqual(1, meck:num_calls(couch_os_process, stop, 1)),

     WaitFun = fun() ->
         case couch_proc_manager:get_proc_count() of
             0 -> ok;
             N when is_integer(N), N > 0 -> wait
         end
     end,
     case test_util:wait(WaitFun, 5000) of
         timeout -> error(timeout);
         _ -> ok
     end,
     ?assertEqual(0, couch_proc_manager:get_proc_count()),

     DDoc2 = ddoc(<<"_design/ddoc2">>),
     ok = couch_query_servers:with_ddoc_proc(Db, DDoc2, fun(_) -> ok end),
     ?assertEqual(1, meck:num_calls(couch_os_process, stop, 1)),
     ?assertEqual(1, couch_proc_manager:get_proc_count()).


 ddoc(DDocId) ->
     #doc{
         id = DDocId,
         revs = {1, [<<"abc">>]},
         body = {[
             {<<"language">>, <<"javascript">>},
             {<<"views">>, {[
                 {<<"v1">>, {[
                     {<<"map">>, <<"function(doc) {emit(doc.value,1);}">>}
                 ]}}
             ]}}
         ]}
     }.

 setup_config() ->
     config:set("native_query_servers", "enable_erlang_query_server", "true", false),
     config:set("query_server_config", "os_process_limit", "3", false),
     config:set("query_server_config", "os_process_soft_limit", "2", false),
     ok.

 teardown_config() ->
     config:delete("native_query_servers", "enable_erlang_query_server", false),
     config:delete("query_server_config", "os_process_limit", false),
     config:delete("query_server_config", "os_process_soft_limit", false),
     config:delete("query_server_config", "db_tag", false),
     ok.

 spawn_client() ->
     Parent = self(),
     Ref = make_ref(),
     Pid = spawn(fun() ->
         Proc = couch_query_servers:get_os_process(<<"erlang">>),
         loop(Parent, Ref, Proc)
     end),
     {Pid, Ref}.

 spawn_client(Db, DDocId) ->
     Parent = self(),
     Ref = make_ref(),
     Pid = spawn(fun() ->
         DDocKey = {DDocId, <<"1-abcdefgh">>},
         DDoc = #doc{body={[{<<"language">>, <<"erlang">>}]}},
         Proc = couch_query_servers:get_ddoc_process(DDoc, Db, DDocKey),
         loop(Parent, Ref, Proc)
     end),
     {Pid, Ref}.

 ping_client({Pid, Ref}) ->
     Pid ! ping,
     receive
         {pong, Ref} ->
             ok
     after ?TIMEOUT ->
         timeout
     end.

 get_client_proc({Pid, Ref}, ClientName) ->
     Pid ! get_proc,
     receive
         {proc, Ref, Proc} -> Proc
     after ?TIMEOUT ->
         erlang:error({assertion_failed,
                      [{module, ?MODULE},
                       {line, ?LINE},
                       {reason, "Timeout getting client "
                                ++ ClientName ++ " proc"}]})
     end.

 stop_client({Pid, Ref}) ->
     MRef = erlang:monitor(process, Pid),
     Pid ! stop,
     receive
         {stop, Ref} ->
             receive
                 {'DOWN', MRef, process, Pid, _} -> ok
             end,
             ok
     after ?TIMEOUT ->
         erlang:demonitor(MRef, [flush]),
         timeout
     end.

 kill_client({Pid, Ref}) ->
     MRef = erlang:monitor(process, Pid),
     Pid ! die,
     receive
         {die, Ref} ->
             receive
                 {'DOWN', MRef, process, Pid, _} -> ok
             end,
             ok
     after ?TIMEOUT ->
         erlang:demonitor(MRef, [flush]),
         timeout
     end.

 loop(Parent, Ref, Proc) ->
     receive
         ping ->
             Parent ! {pong, Ref},
             loop(Parent, Ref, Proc);
         get_proc  ->
             Parent ! {proc, Ref, Proc},
             loop(Parent, Ref, Proc);
         stop ->
             couch_query_servers:ret_os_process(Proc),
             Parent ! {stop, Ref};
         die ->
             Parent ! {die, Ref},
             exit(some_error)
     end.

 proc_pids(Procs) ->
     [P#proc.pid || P <- Procs].

 all_alive(Procs) ->
     lists:all(fun is_process_alive/1, proc_pids(Procs)).

 all_different(Procs) ->
     lists:usort(proc_pids(Procs)) =:= lists:sort(proc_pids(Procs)).

 all_alive_all_different(Procs) ->
     all_alive(Procs) andalso all_different(Procs).

 idle_count() ->
     ets:info(couch_proc_manager_idle_by_db, size).

 tagged_idle_count() ->
     ets:info(couch_proc_manager_idle_access, size).

 stop_clients(Clients) ->
     Fun = fun(C) -> ?assertEqual(ok, stop_client(C)) end,
     lists:map(Fun, Clients).

 wait_tagged_idle_count(N) ->
     WaitFun = fun() ->
         case tagged_idle_count() == N of
             true -> ok;
             false -> wait
         end
     end,
     case test_util:wait(WaitFun, 5000) of
         timeout -> error(timeout);
         _ -> ok
     end,
     ?assertEqual(N, tagged_idle_count()).

 wait_idle_count(N) ->
     WaitFun = fun() ->
         case idle_count() == N of
             true -> ok;
             false -> wait
         end
     end,
     case test_util:wait(WaitFun, 5000) of
         timeout -> error(timeout);
         _ -> ok
     end,
     ?assertEqual(N, idle_count()).

 wait_process_count(N) ->
     WaitFun = fun() ->
         case couch_proc_manager:get_proc_count() == N of
             true -> ok;
             false -> wait
         end
     end,
     case test_util:wait(WaitFun, 5000) of
         timeout -> error(timeout);
         _ -> ok
     end,
     ?assertEqual(N, couch_proc_manager:get_proc_count()).

 cfg_set(K, V) ->
     config:set("query_server_config", K, V, false),
     ok.
	% 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(couchdb_os_proc_pool).

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

	-define(TDEF_FE(Name), fun(Arg) -> {atom_to_list(Name), ?_test(Name(Arg))} end).

	-define(TIMEOUT, 1000).


	setup() ->
	Ctx = test_util:start_couch(),
	meck:new(couch_os_process, [passthrough]),
	meck:new(couch_proc_manager, [passthrough]),
	ok = setup_config(),
	Ctx.

	teardown(Ctx) ->
	ok = teardown_config(),
	meck:unload(),
	test_util:stop_couch(Ctx),
	ok.

	os_proc_pool_test_() ->
	{
	"OS processes pool tests",
	{
	foreach,
	fun setup/0,
	fun teardown/1,
	[
	?TDEF_FE(should_block_new_proc_on_full_pool),
	?TDEF_FE(should_free_slot_on_proc_unexpected_exit),
	?TDEF_FE(should_reuse_known_proc),
	?TDEF_FE(should_process_waiting_queue_as_fifo),
	?TDEF_FE(should_reduce_pool_on_idle_os_procs),
	?TDEF_FE(should_reduce_pool_of_tagged_processes_on_idle),
	?TDEF_FE(should_not_return_broken_process_to_the_pool),
	?TDEF_FE(oldest_tagged_process_is_reaped),
	?TDEF_FE(untagged_process_is_replenished),
	?TDEF_FE(exact_ddoc_tagged_process_is_picked_first),
	?TDEF_FE(db_tagged_process_is_second_choice),
	?TDEF_FE(if_no_tagged_process_found_new_must_be_spawned),
	?TDEF_FE(db_tag_none_works),
	?TDEF_FE(stale_procs_are_cleaned),
	?TDEF_FE(bad_query_language)
	]
	}
	}.

	should_block_new_proc_on_full_pool(_) ->
	Client1 = spawn_client(),
	Client2 = spawn_client(),
	Client3 = spawn_client(),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),
	?assertEqual(ok, ping_client(Client3)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),
	Proc3 = get_client_proc(Client3, "3"),

	?assertNotEqual(Proc1, Proc2),
	?assertNotEqual(Proc2, Proc3),
	?assertNotEqual(Proc3, Proc1),

	Client4 = spawn_client(),
	?assertEqual(timeout, ping_client(Client4)),

	?assertEqual(ok, stop_client(Client1)),
	?assertEqual(ok, ping_client(Client4)),

	Proc4 = get_client_proc(Client4, "4"),

	?assertEqual(Proc1#proc.pid, Proc4#proc.pid),
	?assertNotEqual(Proc1#proc.client, Proc4#proc.client),

	stop_clients([Client2, Client3, Client4]).

	should_free_slot_on_proc_unexpected_exit(_) ->
	Client1 = spawn_client(),
	Client2 = spawn_client(),
	Client3 = spawn_client(),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),
	?assertEqual(ok, ping_client(Client3)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),
	Proc3 = get_client_proc(Client3, "3"),

	?assertNotEqual(Proc1#proc.pid, Proc2#proc.pid),
	?assertNotEqual(Proc1#proc.client, Proc2#proc.client),
	?assertNotEqual(Proc2#proc.pid, Proc3#proc.pid),
	?assertNotEqual(Proc2#proc.client, Proc3#proc.client),
	?assertNotEqual(Proc3#proc.pid, Proc1#proc.pid),
	?assertNotEqual(Proc3#proc.client, Proc1#proc.client),

	?assertEqual(ok, kill_client(Client1)),

	Client4 = spawn_client(),
	?assertEqual(ok, ping_client(Client4)),

	Proc4 = get_client_proc(Client4, "4"),

	?assertEqual(Proc4#proc.pid, Proc1#proc.pid),
	?assertNotEqual(Proc4#proc.client, Proc1#proc.client),
	?assertNotEqual(Proc2#proc.pid, Proc4#proc.pid),
	?assertNotEqual(Proc2#proc.client, Proc4#proc.client),
	?assertNotEqual(Proc3#proc.pid, Proc4#proc.pid),
	?assertNotEqual(Proc3#proc.client, Proc4#proc.client),

	stop_clients([Client2, Client3, Client4]).

	should_reuse_known_proc(_) ->
	Db = <<"db">>,
	Client1 = spawn_client(Db, <<"ddoc1">>),
	Client2 = spawn_client(Db, <<"ddoc2">>),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),
	?assertNotEqual(Proc1#proc.pid, Proc2#proc.pid),

	?assertEqual(ok, stop_client(Client1)),
	?assertEqual(ok, stop_client(Client2)),
	?assert(is_process_alive(Proc1#proc.pid)),
	?assert(is_process_alive(Proc2#proc.pid)),

	Client1Again = spawn_client(Db, <<"ddoc1">>),
	?assertEqual(ok, ping_client(Client1Again)),
	Proc1Again = get_client_proc(Client1Again, "1-again"),
	?assertEqual(Proc1#proc.pid, Proc1Again#proc.pid),
	?assertNotEqual(Proc1#proc.client, Proc1Again#proc.client),
	?assertEqual(ok, stop_client(Client1Again)).

	should_process_waiting_queue_as_fifo(_) ->
	Db = <<"db">>,
	meck:reset(couch_proc_manager),
	Client1 = spawn_client(Db, <<"ddoc1">>),
	meck:wait(1, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),
	Client2 = spawn_client(Db, <<"ddoc2">>),
	meck:wait(2, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),
	Client3 = spawn_client(Db, <<"ddoc3">>),
	meck:wait(3, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),
	Client4 = spawn_client(Db, <<"ddoc4">>),
	meck:wait(4, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),
	Client5 = spawn_client(Db, <<"ddoc5">>),
	meck:wait(5, couch_proc_manager, handle_call, [{get_proc, '_'}, '_', '_'], 1000),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),
	?assertEqual(ok, ping_client(Client3)),
	?assertEqual(timeout, ping_client(Client4)),
	?assertEqual(timeout, ping_client(Client5)),

	Proc1 = get_client_proc(Client1, "1"),
	?assertEqual(ok, stop_client(Client1)),
	?assertEqual(ok, ping_client(Client4)),
	Proc4 = get_client_proc(Client4, "4"),

	?assertNotEqual(Proc4#proc.client, Proc1#proc.client),
	?assertEqual(Proc1#proc.pid, Proc4#proc.pid),
	?assertEqual(timeout, ping_client(Client5)),

	?assertEqual(ok, stop_client(Client2)),
	?assertEqual(ok, stop_client(Client3)),
	?assertEqual(ok, stop_client(Client4)),
	?assertEqual(ok, stop_client(Client5)).

	should_reduce_pool_on_idle_os_procs(_) ->
	%% os_process_idle_limit is in sec
	cfg_set("os_process_idle_limit", "1"),

	Db = undefined,
	Client1 = spawn_client(Db, <<"ddoc1">>),
	Client2 = spawn_client(Db, <<"ddoc2">>),
	Client3 = spawn_client(Db, <<"ddoc3">>),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),
	?assertEqual(ok, ping_client(Client3)),

	?assertEqual(3, couch_proc_manager:get_proc_count()),

	?assertEqual(ok, stop_client(Client1)),
	?assertEqual(ok, stop_client(Client2)),
	?assertEqual(ok, stop_client(Client3)),

	% granularity of idle limit is in seconds
	timer:sleep(1000),
	wait_process_count(1).

	should_reduce_pool_of_tagged_processes_on_idle(_) ->
	%% os_process_idle_limit is in sec
	cfg_set("os_process_idle_limit", "1"),

	Db = <<"reduce_pool_on_idle_db">>,
	Client1 = spawn_client(Db, <<"ddoc1">>),
	Client2 = spawn_client(Db, <<"ddoc2">>),
	Client3 = spawn_client(Db, <<"ddoc3">>),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),
	?assertEqual(ok, ping_client(Client3)),

	?assertEqual(3, couch_proc_manager:get_proc_count()),

	stop_clients([Client1, Client2, Client3]),

	timer:sleep(1000),
	wait_process_count(0).

	oldest_tagged_process_is_reaped(_) ->
	Client1 = spawn_client(<<"db1">>, <<"ddoc1">>),
	Client2 = spawn_client(<<"db2">>, <<"ddoc1">>),
	Client3 = spawn_client(<<"db3">>, <<"ddoc1">>),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),
	?assertEqual(ok, ping_client(Client3)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),
	Proc3 = get_client_proc(Client3, "3"),

	?assert(all_alive_all_different([Proc1, Proc2, Proc3])),

	stop_clients([Client1, Client2, Client3]),

	% All procs should be released back into the pool
	wait_tagged_idle_count(3),

	% Processes should be alive
	?assert(all_alive([Proc1, Proc2, Proc3])),

	% Spawning a new tagged proc with a different tag should kill
	% the oldest unused proc and spawn a new one
	Client4 = spawn_client(<<"db4">>, <<"ddoc1">>),
	?assertEqual(ok, ping_client(Client4)),
	Proc4 = get_client_proc(Client4, "4"),

	?assert(all_alive_all_different([Proc2, Proc3, Proc4])),
	?assertNot(is_process_alive(Proc1#proc.pid)),

	?assertEqual(ok, stop_client(Client4)).

	untagged_process_is_replenished(_) ->
	Client1 = spawn_client(),
	Client2 = spawn_client(),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),

	?assert(all_alive_all_different([Proc1, Proc2])),

	stop_clients([Client1, Client2]),

	% All procs should be released back into the pool
	% and they are now all untagged idle
	wait_idle_count(2),
	?assertEqual(0, tagged_idle_count()),

	% Processes should still be alive
	?assert(all_alive([Proc1, Proc2])),

	% Spawning a new tagged proc should tag one of the procs
	% and also asynchronously replenish the untagged pool
	Client3 = spawn_client(<<"db">>, <<"ddoc1">>),
	?assertEqual(ok, ping_client(Client3)),

	% The process is one of the previously untagged ones
	Proc3 = get_client_proc(Client3, "3"),
	Pid3 = Proc3#proc.pid,
	?assert(lists:member(Pid3, proc_pids([Proc1, Proc2]))),

	% wait for replinishment
	wait_idle_count(2),

	?assertEqual(ok, stop_client(Client3)).

	exact_ddoc_tagged_process_is_picked_first(_) ->
	Client1 = spawn_client(<<"db">>, <<"ddoc1">>),
	Client2 = spawn_client(<<"db">>, <<"ddoc2">>),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),

	?assert(all_alive_all_different([Proc1, Proc2])),

	stop_clients([Client1, Client2]),

	% All procs should be released back into the pool
	% and they now tagged and idle
	wait_tagged_idle_count(2),
	wait_idle_count(2),

	% Processes should still be alive
	?assert(all_alive([Proc1, Proc2])),

	% Spawning a new tagged proc should pick the one with
	% matching ddoc
	Client3 = spawn_client(<<"db">>, <<"ddoc1">>),
	?assertEqual(ok, ping_client(Client3)),
	Proc3 = get_client_proc(Client3, "3"),
	?assertEqual(Proc1#proc.pid, Proc3#proc.pid),

	?assertEqual(ok, stop_client(Client3)).

	db_tagged_process_is_second_choice(_) ->
	Client1 = spawn_client(<<"db1">>, <<"ddoc1">>),
	Client2 = spawn_client(<<"db2">>, <<"ddoc2">>),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),

	?assert(all_alive_all_different([Proc1, Proc2])),

	stop_clients([Client1, Client2]),

	% All procs should be released back into the pool
	% and they now tagged and idle
	wait_tagged_idle_count(2),
	wait_idle_count(2),

	% Processes should still be alive
	?assert(all_alive([Proc1, Proc2])),

	% Spawning a new tagged proc should pick the one with
	% the matching ddoc
	Client3 = spawn_client(<<"db1">>, <<"ddoc3">>),
	?assertEqual(ok, ping_client(Client3)),
	Proc3 = get_client_proc(Client3, "3"),
	?assertEqual(Proc1#proc.pid, Proc3#proc.pid),

	?assertEqual(ok, stop_client(Client3)).

	if_no_tagged_process_found_new_must_be_spawned(_) ->
	Client1 = spawn_client(<<"db1">>, <<"ddoc">>),
	Client2 = spawn_client(<<"db2">>, <<"ddoc">>),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),

	?assert(all_alive_all_different([Proc1, Proc2])),

	stop_clients([Client1, Client2]),

	% All procs should be released back into the pool
	% and they now tagged and idle
	wait_tagged_idle_count(2),
	wait_idle_count(2),

	% Processes should still be alive
	?assert(all_alive([Proc1, Proc2])),

	% If new tagged process with new db should spawn
	% new process never pick up an existing one
	Client3 = spawn_client(<<"db3">>, <<"ddoc">>),
	?assertEqual(ok, ping_client(Client3)),
	Proc3 = get_client_proc(Client3, "3"),
	?assertNotEqual(Proc1#proc.pid, Proc3#proc.pid),
	?assertNotEqual(Proc2#proc.pid, Proc3#proc.pid),

	% db1 and db2 procs should still be sitting idle
	?assertEqual(2, tagged_idle_count()),

	% After 3rd proc returns to the pool there should
	% be 3 tagged idle processes
	?assertEqual(ok, stop_client(Client3)),
	wait_tagged_idle_count(3),
	?assertEqual(3, idle_count()).

	db_tag_none_works(_) ->
	cfg_set("db_tag", "none"),
	Client1 = spawn_client(undefined, <<"ddoc1">>),
	Client2 = spawn_client(undefined, <<"ddoc2">>),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),

	?assert(all_alive_all_different([Proc1, Proc2])),

	stop_clients([Client1, Client2]),

	% All procs should be released back into the pool
	% they should be untagged effectively
	wait_idle_count(2),
	?assertEqual(0, tagged_idle_count()),

	% Processes should still be alive
	?assert(all_alive([Proc1, Proc2])),

	% If new tagged process with new db should spawn
	% new process and pick based on ddoc id matching
	Client3 = spawn_client(undefined, <<"ddoc1">>),
	?assertEqual(ok, ping_client(Client3)),
	Proc3 = get_client_proc(Client3, "3"),
	?assertEqual(Proc1#proc.pid, Proc3#proc.pid),
	?assertNotEqual(Proc2#proc.pid, Proc3#proc.pid),

	wait_idle_count(1),

	% After 3rd client stop there should be 2 idle
	% untagged procs
	?assertEqual(ok, stop_client(Client3)),
	wait_idle_count(2),
	?assertEqual(0, tagged_idle_count()).

	stale_procs_are_cleaned(_) ->
	Client1 = spawn_client(),
	Client2 = spawn_client(),

	?assertEqual(ok, ping_client(Client1)),
	?assertEqual(ok, ping_client(Client2)),

	Proc1 = get_client_proc(Client1, "1"),
	Proc2 = get_client_proc(Client2, "2"),

	?assert(all_alive_all_different([Proc1, Proc2])),

	?assertEqual(0, couch_proc_manager:get_stale_proc_count()),
	?assertEqual(ok, couch_proc_manager:reload()),
	?assertEqual(2, couch_proc_manager:get_stale_proc_count()),

	stop_clients([Client1, Client2]),
	?assertEqual(ok, couch_proc_manager:terminate_stale_procs()),
	wait_idle_count(0),
	?assertEqual(0, couch_proc_manager:get_proc_count()).

	bad_query_language(_) ->
	Expect = {unknown_query_language, <<"bad">>},
	?assertThrow(Expect, couch_query_servers:get_os_process(<<"bad">>)).

	should_not_return_broken_process_to_the_pool(_) ->
	cfg_set("os_process_soft_limit", "1"),
	cfg_set("os_process_limit", "1"),

	Db = <<"thedb">>,
	DDoc1 = ddoc(<<"_design/ddoc1">>),

	meck:reset(couch_os_process),

	?assertEqual(0, couch_proc_manager:get_proc_count()),
	ok = couch_query_servers:with_ddoc_proc(Db, DDoc1, fun(_) -> ok end),
	?assertEqual(0, meck:num_calls(couch_os_process, stop, 1)),
	?assertEqual(1, couch_proc_manager:get_proc_count()),

	?assertError(
	bad,
	couch_query_servers:with_ddoc_proc(Db, DDoc1, fun(_) ->
	error(bad)
	end)
	),
	?assertEqual(1, meck:num_calls(couch_os_process, stop, 1)),

	WaitFun = fun() ->
	case couch_proc_manager:get_proc_count() of
	0 -> ok;
	N when is_integer(N), N > 0 -> wait
	end
	end,
	case test_util:wait(WaitFun, 5000) of
	timeout -> error(timeout);
	_ -> ok
	end,
	?assertEqual(0, couch_proc_manager:get_proc_count()),

	DDoc2 = ddoc(<<"_design/ddoc2">>),
	ok = couch_query_servers:with_ddoc_proc(Db, DDoc2, fun(_) -> ok end),
	?assertEqual(1, meck:num_calls(couch_os_process, stop, 1)),
	?assertEqual(1, couch_proc_manager:get_proc_count()).


	ddoc(DDocId) ->
	#doc{
	id = DDocId,
	revs = {1, [<<"abc">>]},
	body = {[
	{<<"language">>, <<"javascript">>},
	{<<"views">>, {[
	{<<"v1">>, {[
	{<<"map">>, <<"function(doc) {emit(doc.value,1);}">>}
	]}}
	]}}
	]}
	}.

	setup_config() ->
	config:set("native_query_servers", "enable_erlang_query_server", "true", false),
	config:set("query_server_config", "os_process_limit", "3", false),
	config:set("query_server_config", "os_process_soft_limit", "2", false),
	ok.

	teardown_config() ->
	config:delete("native_query_servers", "enable_erlang_query_server", false),
	config:delete("query_server_config", "os_process_limit", false),
	config:delete("query_server_config", "os_process_soft_limit", false),
	config:delete("query_server_config", "db_tag", false),
	ok.

	spawn_client() ->
	Parent = self(),
	Ref = make_ref(),
	Pid = spawn(fun() ->
	Proc = couch_query_servers:get_os_process(<<"erlang">>),
	loop(Parent, Ref, Proc)
	end),
	{Pid, Ref}.

	spawn_client(Db, DDocId) ->
	Parent = self(),
	Ref = make_ref(),
	Pid = spawn(fun() ->
	DDocKey = {DDocId, <<"1-abcdefgh">>},
	DDoc = #doc{body={[{<<"language">>, <<"erlang">>}]}},
	Proc = couch_query_servers:get_ddoc_process(DDoc, Db, DDocKey),
	loop(Parent, Ref, Proc)
	end),
	{Pid, Ref}.

	ping_client({Pid, Ref}) ->
	Pid ! ping,
	receive
	{pong, Ref} ->
	ok
	after ?TIMEOUT ->
	timeout
	end.

	get_client_proc({Pid, Ref}, ClientName) ->
	Pid ! get_proc,
	receive
	{proc, Ref, Proc} -> Proc
	after ?TIMEOUT ->
	erlang:error({assertion_failed,
	[{module, ?MODULE},
	{line, ?LINE},
	{reason, "Timeout getting client "
	++ ClientName ++ " proc"}]})
	end.

	stop_client({Pid, Ref}) ->
	MRef = erlang:monitor(process, Pid),
	Pid ! stop,
	receive
	{stop, Ref} ->
	receive
	{'DOWN', MRef, process, Pid, _} -> ok
	end,
	ok
	after ?TIMEOUT ->
	erlang:demonitor(MRef, [flush]),
	timeout
	end.

	kill_client({Pid, Ref}) ->
	MRef = erlang:monitor(process, Pid),
	Pid ! die,
	receive
	{die, Ref} ->
	receive
	{'DOWN', MRef, process, Pid, _} -> ok
	end,
	ok
	after ?TIMEOUT ->
	erlang:demonitor(MRef, [flush]),
	timeout
	end.

	loop(Parent, Ref, Proc) ->
	receive
	ping ->
	Parent ! {pong, Ref},
	loop(Parent, Ref, Proc);
	get_proc ->
	Parent ! {proc, Ref, Proc},
	loop(Parent, Ref, Proc);
	stop ->
	couch_query_servers:ret_os_process(Proc),
	Parent ! {stop, Ref};
	die ->
	Parent ! {die, Ref},
	exit(some_error)
	end.

	proc_pids(Procs) ->
	[P#proc.pid \|\| P <- Procs].

	all_alive(Procs) ->
	lists:all(fun is_process_alive/1, proc_pids(Procs)).

	all_different(Procs) ->
	lists:usort(proc_pids(Procs)) =:= lists:sort(proc_pids(Procs)).

	all_alive_all_different(Procs) ->
	all_alive(Procs) andalso all_different(Procs).

	idle_count() ->
	ets:info(couch_proc_manager_idle_by_db, size).

	tagged_idle_count() ->
	ets:info(couch_proc_manager_idle_access, size).

	stop_clients(Clients) ->
	Fun = fun(C) -> ?assertEqual(ok, stop_client(C)) end,
	lists:map(Fun, Clients).

	wait_tagged_idle_count(N) ->
	WaitFun = fun() ->
	case tagged_idle_count() == N of
	true -> ok;
	false -> wait
	end
	end,
	case test_util:wait(WaitFun, 5000) of
	timeout -> error(timeout);
	_ -> ok
	end,
	?assertEqual(N, tagged_idle_count()).

	wait_idle_count(N) ->
	WaitFun = fun() ->
	case idle_count() == N of
	true -> ok;
	false -> wait
	end
	end,
	case test_util:wait(WaitFun, 5000) of
	timeout -> error(timeout);
	_ -> ok
	end,
	?assertEqual(N, idle_count()).

	wait_process_count(N) ->
	WaitFun = fun() ->
	case couch_proc_manager:get_proc_count() == N of
	true -> ok;
	false -> wait
	end
	end,
	case test_util:wait(WaitFun, 5000) of
	timeout -> error(timeout);
	_ -> ok
	end,
	?assertEqual(N, couch_proc_manager:get_proc_count()).

	cfg_set(K, V) ->
	config:set("query_server_config", K, V, false),
	ok.