PR Comments

epgsql · Oct 6, 2014 · 8ab8c3c · seth · Oct 6, 2014 · seth
1 parent a3e21df
commit 8ab8c3c
Show file tree

Hide file tree

Showing 3 changed files with 38 additions and 31 deletions.
diff --git a/src/pooler.erl b/src/pooler.erl
@@ -180,17 +180,16 @@ accept_member(PoolName, MemberPid) ->
 take_member(PoolName) when is_atom(PoolName) orelse is_pid(PoolName) ->
     gen_server:call(PoolName, {take_member, 0}, infinity).
 
-%% @doc Obtain exclusive access to a member from 'PoolName'.
+%% @doc Obtain exclusive access to a member of 'PoolName'.
 %%
-%% If no free members are available, reply is deferred for up to
-%% timeout. During this window, if accept member is called
-%% the next queued requestor will be given the accepted member.
-%% If timeout is set to zero, will return immediately with
-%% either a Pid or error_no_members.
+%% If no members are available, wait for up to Timeout milliseconds for a member
+%% to become available. Waiting requests are served in FIFO order. If no member
+%% is available within the specified timeout, error_no_members is returned.
+%% Timeout can be either milliseconds as integer or {duration, time_unit}
 %%
--spec take_member(atom() | pid(), non_neg_integer()) -> pid() | error_no_members.
+-spec take_member(atom() | pid(), non_neg_integer() | time_spec()) -> pid() | error_no_members.
 take_member(PoolName, Timeout) when is_atom(PoolName) orelse is_pid(PoolName) ->
-    gen_server:call(PoolName, {take_member, Timeout}, infinity).
+    gen_server:call(PoolName, {take_member, time_as_millis(Timeout)}, infinity).
 
 
 %% @doc Take a member from a randomly selected member of the group
@@ -313,7 +312,7 @@ init(#pool{}=Pool) ->
 set_member_sup(#pool{} = Pool, MemberSup) ->
     Pool#pool{member_sup = MemberSup}.
 
-handle_call({take_member, Timeout}, From = {APid,_}, #pool{} = Pool) when is_pid(APid) ->
+handle_call({take_member, Timeout}, From = {APid, _}, #pool{} = Pool) when is_pid(APid) ->
     maybe_reply(take_member_from_pool_queued(Pool, From, Timeout));
 
 handle_call({return_member, Pid, Status}, {_CPid, _Tag}, Pool) ->
@@ -582,7 +581,7 @@ take_member_from_pool_queued(Pool0 = #pool{queue_max = QMax,
             send_metric(Pool1, queue_max_reached, {inc, 1}, counter),
             {error_no_members, Pool1};
         {{error_no_members, Pool1 = #pool{queued_requestors = QueuedRequestors}}, QueueCount} ->
-            timer:send_after(time_as_millis(Timeout), {requestor_timeout, From}),
+            timer:send_after(Timeout, {requestor_timeout, From}),
             send_metric(Pool1, queue_count, QueueCount, histogram),
             {queued, Pool1#pool{queued_requestors = queue:in(From, QueuedRequestors)}};
         {{Member, NewPool}, _} when is_pid(Member) ->

diff --git a/src/pooler.hrl b/src/pooler.hrl
@@ -81,7 +81,7 @@
           %% and Exometer format.
           metrics_api = folsom :: 'folsom' | 'exometer',
 
-          %% A queue of requestors for deferred take member requests
+          %% A queue of requestors for blocking take member requests
           queued_requestors = queue:new() :: queue:queue({pid(), _}),
           %% The max depth of the queue
           queue_max = 50

diff --git a/test/pooler_tests.erl b/test/pooler_tests.erl
@@ -743,13 +743,19 @@ pooler_integration_queueing_test_() ->
      % setup
      fun() ->
              Pool = [{name, test_pool_1},
-                       {max_count, 10},
-                       {queue_max, 10},
-                       {init_count, 0},
-                       {metrics, fake_metrics},
-                       {member_start_timeout, {5, sec}},
-                       {start_mfa,
-                        {pooled_gs, start_link, [{"type-0", fun pooler_tests:sleep_for_configured_timeout/0 }]}}],
+                     {max_count, 10},
+                     {queue_max, 10},
+                     {init_count, 0},
+                     {metrics, fake_metrics},
+                     {member_start_timeout, {5, sec}},
+                     {start_mfa,
+                      {pooled_gs, start_link, [
+                                               {"type-0",
+                                                fun pooler_tests:sleep_for_configured_timeout/0 }
+                                              ]
+                      }
+                     }
+                    ],
 
              application:set_env(pooler, pools, [Pool]),
              fake_metrics:start_link(),
@@ -760,7 +766,7 @@ pooler_integration_queueing_test_() ->
              fake_metrics:stop(),
              application:stop(pooler)
      end,
-     %
+                                                %
      [
       fun(_) ->
               fun() ->
@@ -787,7 +793,9 @@ pooler_integration_queueing_test_() ->
               fun() ->
                       application:set_env(pooler, sleep_time, 10),
                       ?assertEqual(0, (dump_pool(test_pool_1))#pool.free_count),
-                      [?assertEqual(pooler:take_member(test_pool_1, 0), error_no_members) || _ <- lists:seq(1, (dump_pool(test_pool_1))#pool.max_count)],
+                      [
+                       ?assertEqual(pooler:take_member(test_pool_1, 0), error_no_members) ||
+                          _ <- lists:seq(1, (dump_pool(test_pool_1))#pool.max_count)],
                       timer:sleep(50),
                       %Next request should be available
                       Pid = pooler:take_member(test_pool_1, 0),
@@ -800,19 +808,19 @@ pooler_integration_queueing_test_() ->
                       % fill to queue_max, next request should return immediately with no_members
                       % Will return a if queue max is not enforced.
                       application:set_env(pooler, sleep_time, 100),
-                     [ proc_lib:spawn(fun() -> Val = pooler:take_member(test_pool_1, 200),
-                                      ?assert(is_pid(Val)),
-                                      pooler:return_member(Val) end)
-                                      || _ <- lists:seq(1, (dump_pool(test_pool_1))#pool.max_count)
-                                         ],
+                      [ proc_lib:spawn(fun() -> Val = pooler:take_member(test_pool_1, 200),
+                                                ?assert(is_pid(Val)),
+                                                pooler:return_member(Val) end)
+                        || _ <- lists:seq(1, (dump_pool(test_pool_1))#pool.max_count)
+                      ],
                       timer:sleep(50),
                       ?assertEqual(10, queue:len((dump_pool(test_pool_1))#pool.queued_requestors)),
                       ?assertEqual(pooler:take_member(test_pool_1, 500), error_no_members),
                       ExpectKeys = lists:sort([<<"pooler.test_pool_1.error_no_members_count">>,
-                                        <<"pooler.test_pool_1.events">>,
-                                        <<"pooler.test_pool_1.take_rate">>,
-                                        <<"pooler.test_pool_1.queue_count">>,
-                                        <<"pooler.test_pool_1.queue_max_reached">>]),
+                                               <<"pooler.test_pool_1.events">>,
+                                               <<"pooler.test_pool_1.take_rate">>,
+                                               <<"pooler.test_pool_1.queue_count">>,
+                                               <<"pooler.test_pool_1.queue_max_reached">>]),
                       Metrics = fake_metrics:get_metrics(),
                       GotKeys = lists:usort([ Name || {Name, _, _} <- Metrics ]),
                       ?assertEqual(ExpectKeys, GotKeys),
@@ -824,8 +832,8 @@ pooler_integration_queueing_test_() ->
               end
       end
      ]
-     }.
-    
+    }.
+
 
 pooler_integration_test_() ->
     {foreach,