Fix/timeout leak (#497)

fix socket leaks

make sure that pending connections are correctly removed from the queue, which fix a race condition with the manager.
fix #462

Author: benoitc

examples/test_deadlock.erl

@@ -0,0 +1,38 @@
+#!/usr/bin/env escript
+%% -*- erlang -*-
+%%! -pa ../_build/default/lib/hackney/ebin  -pa ../_build/default/lib/*/ebin  -pa ../_build/default/lib/certifi/ebin  -pa ../_build/default/lib/idna/ebin  -pa ../_build/default/lib/metrics/ebin -pa ../_build/default/lib/mimerl/ebin   -pa ../_build/default/lib/ssl_verify_fun/ebin -pa ../_build/default/lib/unicode_util_compat/ebin
+
+-module(test_async).
+
+-export([main/1]).
+
+deadlock(Url, N) ->
+  Fun = fun() ->
+          {ok, _, _Headers, Ref} = hackney:get(Url, [], <<>>, [{connect_timeout, 1000}, {recv_timeout, 1000}]),
+          {ok, _Body} = hackney:body(Ref)
+        end,
+  Pids = [spawn(
+    fun() ->
+      spawn_link(Fun),
+      timer:sleep(500),
+      exit(timeout)
+    end) || _I <- lists:seq(1, N)],
+  MRefs = [erlang:monitor(process, Pid) || Pid <- Pids],
+  wait_pids(MRefs),
+  ok.
+
+wait_pids([]) -> ok;
+wait_pids(MRefs) ->
+  receive
+    {'DOWN', MRef, process, _, _} ->
+      wait_pids(MRefs -- [MRef])
+  end.
+
+
+main(_) ->
+  {ok, _}= application:ensure_all_started(hackney),
+  io:format("start test~n", []),
+  deadlock("https://httparrot.herokuapp.com/delay/5", 550),
+  io:format("pools are ~p~n", [ets:tab2list(hackney_pool)]),
+  io:format("state of the pool:~n ~p~n", [hackney_pool:get_stats(default)]),
+  ok.

src/hackney_pool.erl

@@ -16,10 +16,13 @@
   checkout/4,
   checkin/2]).
 
--export([start_pool/2,
+-export([
+  get_stats/1,
+  start_pool/2,
   stop_pool/1,
   find_pool/1,
-  notify/2]).
+  notify/2
+]).
 
 
 -export([count/1, count/2,
@@ -45,7 +48,8 @@
   max_connections,
   timeout,
   clients = dict:new(),
-  queues = dict:new(),  % Dest => queue of Froms
+  queues = dict:new(),  % Dest => queue of Froms,
+  pending = dict:new(),
   connections = dict:new(),
   sockets = dict:new(),
   nb_waiters=0}).
@@ -99,6 +103,9 @@ checkin({_Name, Ref, Dest, Owner, Transport}, Socket) ->
       ok
   end.
 
+get_stats(Pool) ->
+  gen_server:call(find_pool(Pool), stats).
+
 
 %% @doc start a pool
 start_pool(Name, Options) ->
@@ -235,6 +242,8 @@ init([Name, Options]) ->
   {ok, #state{name=Name, metrics=Engine, max_connections=MaxConn,
     timeout=Timeout}}.
 
+handle_call(stats, _From, State) ->
+  {reply, handle_stats(State), State};
 handle_call(count, _From, #state{sockets=Sockets}=State) ->
   {reply, dict:size(Sockets), State};
 handle_call(timeout, _From, #state{timeout=Timeout}=State) ->
@@ -247,7 +256,7 @@ handle_call({checkout, Dest, Pid, RequestRef}, From, State) ->
     max_connections=MaxConn,
     clients=Clients,
     queues = Queues,
-    nb_waiters = NbWaiters} = State,
+    pending = Pending} = State,
 
   {Reply, State2} = find_connection(Dest, Pid, State),
   case Reply of
@@ -259,12 +268,11 @@ handle_call({checkout, Dest, Pid, RequestRef}, From, State) ->
       case dict:size(Clients) >= MaxConn of
         true ->
           Queues2 = add_to_queue(Dest, From, RequestRef, Queues),
-          NbWaiters2 = NbWaiters + 1,
-          _ = metrics:update_histogram(Engine,
-            [hackney_pool, PoolName, queue_count],
-            NbWaiters2),
-          {noreply, State2#state{queues = Queues2,
-            nb_waiters=NbWaiters2}};
+          Pending2 =add_pending(RequestRef, From, Dest, Pending),
+          _ = metrics:update_histogram(
+            Engine, [hackney_pool, PoolName, queue_count], dict:size(Pending2)
+          ),
+          {noreply, State2#state{queues = Queues2, pending = Pending2}};
         false ->
           State3 = monitor_client(Dest, RequestRef, State2),
           ok = update_usage(State3),
@@ -307,11 +315,12 @@ handle_cast({set_timeout, NewTimeout}, State) ->
   {noreply, State#state{timeout=NewTimeout}};
 
 handle_cast({checkout_cancel, Dest, Ref}, State) ->
-  {Queues, Removed} = del_from_queue(Dest, Ref, State#state.queues),
+  #state{queues=Queues, pending=Pending} = State,
+  {Queues2, Removed} = del_from_queue(Dest, Ref, Queues),
   case Removed of
     true ->
-      NbWaiters = State#state.nb_waiters - 1,
-      {noreply, State#state{queues=Queues, nb_waiters=NbWaiters}};
+      Pending2 = del_pending(Ref, Pending),
+      {noreply, State#state{queues=Queues2, nb_waiters=Pending2}};
     false ->
       % we leak the socket here but 'DOWN' will mop up for us when it times out
       {noreply, dequeue(Dest, Ref, State)}
@@ -338,7 +347,8 @@ handle_info({'DOWN', Ref, request, _Pid, _Reason}, State) ->
     {ok, Dest} ->
       {noreply, dequeue(Dest, Ref, State)};
     error ->
-      {noreply, State}
+      NewState = remove_pending(Ref, State),
+      {noreply, NewState}
   end;
 handle_info(_, State) ->
   {noreply, State}.
@@ -360,17 +370,18 @@ terminate(_Reason, #state{name=PoolName, metrics=Engine, sockets=Sockets}) ->
 %% internals
 
 dequeue(Dest, Ref, State) ->
-  Clients2 = dict:erase(Ref, State#state.clients),
-  case queue_out(Dest, State#state.queues) of
+  #state{clients=Clients, queues=Queues, pending=Pending} = State,
+  Clients2 = dict:erase(Ref, Clients),
+  case queue_out(Dest, Queues) of
     empty ->
       State#state{clients = Clients2};
     {ok, {From, Ref2}, Queues2} ->
-      NbWaiters = State#state.nb_waiters - 1,
-      _ = metrics:update_histogram(State#state.metrics,
-        [hackney_pool, State#state.name, queue_count], NbWaiters),
+      Pending2 = del_pending(Ref, Pending),
+      _ = metrics:update_histogram(
+        State#state.metrics, [hackney_pool, State#state.name, queue_count], dict:size(Pending2)
+      ),
       gen_server:reply(From, {error, no_socket, self()}),
-      State2 = State#state{queues = Queues2, clients = Clients2,
-        nb_waiters=NbWaiters},
+      State2 = State#state{queues = Queues2, clients = Clients2, pending=Pending2},
       monitor_client(Dest, Ref2, State2)
   end.
 
@@ -387,8 +398,7 @@ find_connection({_Host, _Port, Transport}=Dest, Pid,
               cancel_timer(S, Timer),
               NewConns = update_connections(Rest, Dest, Conns),
               NewSockets = dict:erase(S, Sockets),
-              NewState = State#state{connections=NewConns,
-                sockets=NewSockets},
+              NewState = State#state{connections=NewConns, sockets=NewSockets},
               {{ok, S, self()}, NewState};
             {error, badarg} ->
               %% something happened here normally the PID died,
@@ -509,26 +519,25 @@ queue_out({_Host, _Port, _Transport} = Dest, Queues) ->
 %% @private
 %%------------------------------------------------------------------------------
 deliver_socket(Socket, {_, _, Transport} = Dest, State) ->
-  case queue_out(Dest, State#state.queues) of
+  #state{queues = Queues, pending=Pending} = State,
+  case queue_out(Dest, Queues) of
     empty ->
       store_socket(Dest, Socket, State);
     {ok, {{PidWaiter, _} = FromWaiter, Ref}, Queues2} ->
-      NbWaiters = State#state.nb_waiters - 1,
-      _ = metrics:update_histogram(State#state.metrics,
-        [hackney_pool, State#state.name, queue_count],
-        NbWaiters),
+      Pending2 = del_pending(Ref, Pending),
+      _ = metrics:update_histogram(
+        State#state.metrics, [hackney_pool, State#state.name, queue_count], dict:size(Pending2)
+      ),
       case Transport:controlling_process(Socket, PidWaiter) of
         ok ->
           gen_server:reply(FromWaiter, {ok, Socket, self()}),
-          monitor_client(Dest, Ref,
-            State#state{queues = Queues2,
-              nb_waiters=NbWaiters});
+          monitor_client(Dest, Ref, State#state{queues = Queues2, pending=Pending2});
         _Error ->
           % Something wrong, close the socket
-            catch Transport:close(Socket),
+          _ = (catch Transport:close(Socket)),
           %% and let the waiter connect to a new one
           gen_server:reply(FromWaiter, {error, no_socket, self()}),
-          State#state{queues = Queues2, nb_waiters = NbWaiters}
+          State#state{queues = Queues2, pending = Pending2}
       end
   end.
 
@@ -544,6 +553,37 @@ sync_socket(Transport, Socket) ->
     true
   end.
 
+
+add_pending(Ref, From, Dest, Pending) ->
+  dict:store(Ref, {From, Dest}, Pending).
+
+del_pending(Ref, Pending) ->
+  dict:erase(Ref, Pending).
+
+remove_pending(Ref, #state{queues=Queues0, pending=Pending0} = State) ->
+  case dict:find(Ref, Pending0) of
+    {ok, {From, Dest}} ->
+      Pending1 = dict:erase(Ref, Pending0),
+      Queues1 = case dict:find(Dest, Queues0) of
+                  {ok, Q0} ->
+                    Q1 = queue:filter(
+                      fun
+                        (PendingReq) when PendingReq =:= {From, Ref} -> false;
+                        (_) -> true
+                      end,
+                      Q0
+                    ),
+                    dict:store(Dest, Q1, Queues0);
+                  error ->
+                    Queues0
+                end,
+      State#state{queues=Queues1, pending=Pending1};
+    error ->
+      State
+  end.
+
+
+
 %------------------------------------------------------------------------------
 %% @private
 %%------------------------------------------------------------------------------
@@ -581,3 +621,12 @@ update_usage(
   _ = metrics:update_histogram(Engine, [hackney_pool, PoolName, free_count],
     dict:size(Sockets) - 1),
   ok.
+
+
+handle_stats(State) ->
+  #state{name=PoolName, max_connections=Max, sockets=Sockets, clients=Clients, pending=Pending} = State,
+  [{name, PoolName},
+   {max, Max},
+   {in_use_count,  dict:size(Clients)},
+   {free_count, dict:size(Sockets)},
+   {queue_count, dict:size(Pending)}].