pool: fix race conditions

there was a possible race codition when creating/destroying a connection
ID. This changes simply cache a hash of the connection to make it more
consistent. We keep the number of items in the cache to 1000.

* changes

- refactor hackney_connections to be a simple LRU cache
- fix hackney_connection and hackney_pool modules

Author: benoitc

rebar.lock

@@ -1,6 +1,7 @@
 {"1.1.0",
 [{<<"certifi">>,{pkg,<<"certifi">>,<<"2.5.2">>},0},
  {<<"idna">>,{pkg,<<"idna">>,<<"6.0.1">>},0},
+ {<<"lru">>,{pkg,<<"lru">>,<<"2.4.0">>},0},
  {<<"metrics">>,{pkg,<<"metrics">>,<<"1.0.1">>},0},
  {<<"mimerl">>,{pkg,<<"mimerl">>,<<"1.2.0">>},0},
  {<<"parse_trans">>,{pkg,<<"parse_trans">>,<<"3.3.0">>},0},
@@ -10,6 +11,7 @@
 {pkg_hash,[
  {<<"certifi">>, <<"B7CFEAE9D2ED395695DD8201C57A2D019C0C43ECAF8B8BCB9320B40D6662F340">>},
  {<<"idna">>, <<"1D038FB2E7668CE41FBF681D2C45902E52B3CB9E9C77B55334353B222C2EE50C">>},
+ {<<"lru">>, <<"A8F9967CA9B6F260BAA19E2EFB2AEB3853A3F5BD5F8416F537A672294B38C1BC">>},
  {<<"metrics">>, <<"25F094DEA2CDA98213CECC3AEFF09E940299D950904393B2A29D191C346A8486">>},
  {<<"mimerl">>, <<"67E2D3F571088D5CFD3E550C383094B47159F3EEE8FFA08E64106CDF5E981BE3">>},
  {<<"parse_trans">>, <<"09765507A3C7590A784615CFD421D101AEC25098D50B89D7AA1D66646BC571C1">>},

src/hackney_connection.erl

@@ -2,7 +2,6 @@
 
 
 -export([new/1,
-         release/1,
          get_property/2,
          is_ssl/1]).
 
@@ -27,13 +26,10 @@ new(#client{transport=Transport,
   Host1 = string_compat:to_lower(Host0),
   ConnectOptions = connect_options(Transport, Host1, ClientOptions),
   Tunnel = maybe_tunnel(Transport),
-  Id = hackney_connections:new_connection_id(Transport, Host1, Port, ConnectOptions),
+  Id = new_connection_id(Transport, Host1, Port, ConnectOptions),
   Connection = new_connection_r(Transport, Host1, Port, Id, Tunnel),
   {Connection, ConnectOptions}.
 
-release(#connection{id=Id}) ->
-  hackney_connections:release_connection_id(Id).
-
 get_property(transport, #connection{transport=Transport}) -> Transport;
 get_property(host, #connection{host=Host}) -> Host;
 get_property(port, #connection{port=Port}) -> Port;
@@ -45,7 +41,6 @@ is_ssl(#connection{transport=hackney_ssl}) -> true;
 is_ssl(#connection{}) -> false;
 is_ssl(_) -> erlang:error(badarg).
 
-
 controlling_process(#connection{transport=Transport}, Socket, Owner) ->
   Transport:controlling_process(Socket, Owner).
 
@@ -77,6 +72,18 @@ close(#connection{transport=Transport}, Socket) ->
   Transport:close(Socket).
 
 
+new_connection_id(Transport, Host, Port, ConnectionOptions) ->
+  Key = {Transport, Host, Port, ConnectionOptions},
+  case hackney_connections:lookup(Key) of
+    {ok, Id} ->
+      Id;
+    error ->
+      Id = erlang:phash2(Key),
+      hackney_connections:insert(Key, Id),
+      Id
+  end.
+
+
 new_connection_r(Transport, Host, Port, Id, Tunnel) ->
   #connection{transport=Transport,
               host=Host,

src/hackney_connections.erl

@@ -1,9 +1,14 @@
 -module(hackney_connections).
 -behaviour(gen_server).
 
+%% API
+-export([
+  insert/2,
+  lookup/1,
+  delete/1,
+  get_num_entries/0
+]).
 
--export([new_connection_id/4,
-         release_connection_id/1]).
 
 -export([start_link/0]).
 
@@ -19,60 +24,69 @@
 
 -define(SERVER, ?MODULE).
 
+-record(state, {tab,lru}).
+-define(MAX_ITEMS, 1000).
 
-new_connection_id(Transport, Host, Port, ConnectOptions) ->
-  ConnectionKey = {Transport, Host, Port, ConnectOptions},
-  gen_server:call(?SERVER, {new_session_id, ConnectionKey}).
 
+insert(Key, Id) ->
+  gen_server:call(?SERVER, {insert, Key, Id}).
 
-release_connection_id(Id) ->
-  gen_server:cast(?SERVER, {release_connection_id, Id}).
+lookup(Key) ->
+  gen_server:call(?SERVER, {lookup, Key}).
 
+delete(Key) ->
+  gen_server:call(?SERVER, {delete, Key}).
 
+get_num_entries() ->
+  gen_server:call(?SERVER, num_entries).
 
 start_link() ->
   gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
 
 init([]) ->
-    _ = ets:new(?CONNECTIONS, [set, named_table, public,
-                               {read_concurrency, true},
-                               {write_concurrency, true}]),
-  {ok, {}}.
+  Tab = ets:new(?SERVER, [set, {read_concurrency, true}]),
+  {ok, #state{tab=Tab, lru=new_lrulist()}}.
 
 
-handle_call({new_session_id, ConnectionKey}, _From, State) ->
-  case ets:lookup(?CONNECTIONS, ConnectionKey) of
-    [{_, Id}] ->
-      ets:update_counter(?CONNECTIONS, Id, {3, 1}),
-      {reply, Id, State};
+
+handle_call({insert, Key, Id}, From, #state{tab=Tab, lru=Lru} = State) ->
+  true = ets:insert(Tab, {Key, Id}),
+  gen_server:reply(From, true),
+  Lru2 = push(Key, Lru),
+  NewState = case has_exceeded(State) of
+               true ->
+                 evict(State#state{lru=Lru2});
+               false ->
+                 State#state{lru=Lru2 }
+             end,
+  {noreply, NewState};
+
+
+handle_call({lookup, Key}, From, State = #state{tab=Tab, lru=Lru}) ->
+  case ets:lookup(Tab, Key) of
+    [{Key, Id}] ->
+      gen_server:reply(From, {ok, Id}),
+      Lru2 = touch(Key, Lru),
+      {noreply, State#state{lru=Lru2}};
     [] ->
-      Id = make_ref(),
-      ets:insert(?CONNECTIONS, [{ConnectionKey, Id},
-                                {Id, ConnectionKey, 1}]),
-      {reply, Id, State}
+      {reply, error, State}
   end;
-handle_call(_Msg, _From, State) ->
-  {reply, badarg, State}.
 
+handle_call({delete, Key}, From, State =  #state{tab=Tab, lru=Lru}) ->
+  _ = ets:delete(Tab, Key),
+  gen_server:reply(From, ok),
+  Lru2 = drop(Key, Lru),
+  {reply, true, State#state{lru=Lru2}};
+
+
+handle_call(num_entries, _From, State = #state{tab=Tab}) ->
+  {reply, ets:info(Tab, size) , State};
+
+
+handle_call(_Msg, _From, State) ->
+  {reply, {error, bad_call}, State}.
 
-handle_cast({release_connection_id, Id},  State) ->
-  RefCount = try ets:update_counter(?CONNECTIONS, Id, {3, -1})
-             catch
-               error:badarg -> -1
-             end,
-  if
-    RefCount =< 0 ->
-      case ets:take(?CONNECTIONS, Id) of
-        [{_, ConnKey, _}] ->
-          ets:delete(?CONNECTIONS, ConnKey);
-        [] ->
-          ok
-      end;
-    true ->
-      ok
-  end,
-  {noreply, State};
 handle_cast(_Msg, State) ->
   {noreply, State}.
 
@@ -88,3 +102,43 @@ code_change(_OldVsn, State, _Extra) ->
 terminate(_Reason, _State) ->
   ok.
 
+
+
+has_exceeded(#state{ tab = Tab}) ->
+  (ets:info(Tab, size) > ?MAX_ITEMS).
+
+evict(State = #state{tab=Tab, lru=Lru}) ->
+  case pop(Lru) of
+    {ok, Key, Lru2} ->
+      _ = ets:delete(Tab, Key),
+      State#state{lru=Lru2};
+    empty ->
+      State
+  end.
+
+
+%% ========
+%% LRU LIST
+
+
+new_lrulist() -> [].
+
+push(LruElement, LruList) ->
+  case lists:member(LruElement, LruList) of
+    true -> LruList;
+    false -> LruList ++ [LruElement]
+  end.
+
+pop([]) -> empty;
+pop([ El | Rest] ) -> {ok, El, Rest}.
+
+touch(LruElement, [LruElement|Rest]) ->
+  Rest ++ [LruElement];
+touch(LruElement, [H|Rest]) ->
+  [H|touch(LruElement, Rest)];
+touch(LruElement, []) ->
+  [LruElement].
+
+drop(LruElement, LruList) ->
+  lists:delete(LruElement, LruList).
+

src/hackney_pool.erl

@@ -91,28 +91,23 @@ checkout(Host, _Port, Transport, #client{options=Opts,
 
           {ok, {PoolName, RequestRef, Connection, Owner, Transport}, Socket};
         {error, timeout} ->
-          hackney_connections:release_connection_id(Connection),
           _ = metrics:increment_counter(Metrics, [hackney, Host, connect_timeout]),
           {error, timeout};
         Error ->
           ?report_trace("connect error", []),
-          hackney_connections:release_connection_id(Connection),
           _ = metrics:increment_counter(Metrics, [hackney, Host, connect_error]),
           Error
       end;
     {error, Reason} ->
-      hackney_connections:release_connection_id(Connection),
       {error, Reason};
     {'EXIT', {timeout, _}} ->
-      hackney_connections:release_connection_id(Connection),
       % socket will still checkout so to avoid deadlock we send in a cancellation
       gen_server:cast(Pool, {checkout_cancel, Connection, RequestRef}),
       {error, checkout_timeout}
   end.
 
 %% @doc release a socket in the pool
 checkin({_Name, Ref, Connection, Owner, Transport}, Socket) ->
-  hackney_connection:release(Connection),
   hackney_connection:setopts(Connection, Socket, [{active, false}]),
   case hackney_connection:sync_socket(Connection, Socket) of
     true ->
@@ -401,7 +396,6 @@ dequeue(Dest, Ref, State) ->
     empty ->
       State#state{clients = Clients2};
     {ok, {From, Ref2}, Queues2} ->
-      hackney_connections:release_connection_id(Dest),
       Pending2 = del_pending(Ref, Pending),
       _ = metrics:update_histogram(
             State#state.metrics, [hackney_pool, State#state.name, queue_count], dict:size(Pending2)
@@ -451,7 +445,6 @@ remove_socket(Socket, #state{connections=Conns, sockets=Sockets}=State) ->
     {ok, {Connection, Timer}} ->
       cancel_timer(Socket, Timer),
       catch hackney_connection:close(Connection, Socket),
-      hackney_connections:release_connection_id(Connection),
       ConnSockets = lists:delete(Socket, dict:fetch(Connection, Conns)),
       NewConns = update_connections(ConnSockets, Connection, Conns),
       NewSockets = dict:erase(Socket, Sockets),
@@ -562,7 +555,6 @@ deliver_socket(Socket, Connection, State) ->
           gen_server:reply(FromWaiter, {ok, Socket, self()}),
           monitor_client(Connection, Ref, State#state{queues = Queues2, pending=Pending2});
         _Error ->
-          hackney_connection:release(Connection),
           % Something wrong, close the socket
           _ = (catch hackney_connection:close(Connection, Socket)),
           %% and let the waiter connect to a new one