move ssl negotiation into reader process

thus freeing up the acceptor to handle new connections

2 files changed, 38 insertions, 42 deletions

diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 1bc17a32..b45d9942 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -160,36 +160,31 @@ node_listeners(Node) ->
 on_node_down(Node) ->
     ok = mnesia:dirty_delete(rabbit_listener, Node).
 
-start_client(Sock) ->
+start_client(Sock, SockTransform) ->
     {ok, Child} = supervisor:start_child(rabbit_tcp_client_sup, []),
     ok = rabbit_net:controlling_process(Sock, Child),
-    Child ! {go, Sock},
+    Child ! {go, Sock, SockTransform},
     Child.
 
+start_client(Sock) ->
+    start_client(Sock, fun (S) -> {ok, S} end).
+
 start_ssl_client(SslOpts, Sock) ->
-    case rabbit_net:peername(Sock) of
-        {ok, {PeerAddress, PeerPort}} ->
-            PeerIp = inet_parse:ntoa(PeerAddress),
-            case ssl:ssl_accept(Sock, SslOpts) of
-                {ok, SslSock} ->
-                    rabbit_log:info("upgraded TCP connection "
-                                    "from ~s:~p to SSL~n",
-                                    [PeerIp, PeerPort]),
-                    RabbitSslSock = #ssl_socket{tcp = Sock, ssl = SslSock},
-                    start_client(RabbitSslSock);
-                {error, Reason} ->
-                    gen_tcp:close(Sock),
-                    rabbit_log:error("failed to upgrade TCP connection "
-                                     "from ~s:~p to SSL: ~n~p~n",
-                                     [PeerIp, PeerPort, Reason]),
-                    {error, Reason}
-            end;
-        {error, Reason} ->
-            gen_tcp:close(Sock),
-            rabbit_log:error("failed to upgrade TCP connection to SSL: ~p~n",
-                             [Reason]),
-            {error, Reason}
-    end.
+    start_client(
+      Sock,
+      fun (Sock1) ->
+              case catch ssl:ssl_accept(Sock1, SslOpts) of
+                  {ok, SslSock} ->
+                      rabbit_log:info("upgraded TCP connection ~p to SSL~n",
+                                      [self()]),
+                      {ok, #ssl_socket{tcp = Sock1, ssl = SslSock}};
+                  {error, Reason} ->
+                      {error, {ssl_upgrade_error, Reason}};
+                  {'EXIT', Reason} ->
+                      {error, {ssl_upgrade_failure, Reason}}
+              
+              end
+      end).
 
 connections() ->
     [Pid || {_, Pid, _, _} <- supervisor:which_children(
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index e21485b5..406ee1ed 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -142,7 +142,8 @@ start_link() ->
 init(Parent) ->
     Deb = sys:debug_options([]),
     receive
-        {go, Sock} -> start_connection(Parent, Deb, Sock)
+        {go, Sock, SockTransform} ->
+            start_connection(Parent, Deb, Sock, SockTransform)
     end.
 
 system_continue(Parent, Deb, State) ->
@@ -192,27 +193,27 @@ teardown_profiling(Value) ->
 
 inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F).
 
-peername(Sock) ->   
-    try
-        {Address, Port} = inet_op(fun () -> rabbit_net:peername(Sock) end),
-        AddressS = inet_parse:ntoa(Address),
-        {AddressS, Port}
-    catch
-        Ex -> rabbit_log:error("error on TCP connection ~p:~p~n",
-                               [self(), Ex]),
-              rabbit_log:info("closing TCP connection ~p", [self()]),
-              exit(normal)
+socket_op(Sock, Fun) ->   
+    case Fun(Sock) of
+        {ok, Res}       -> Res;
+        {error, Reason} -> rabbit_log:error("error on TCP connection ~p:~p~n",
+                                            [self(), Reason]),
+                           rabbit_log:info("closing TCP connection ~p~n",
+                                           [self()]),
+                           exit(normal)
     end.
 
-start_connection(Parent, Deb, ClientSock) ->
+start_connection(Parent, Deb, Sock, SockTransform) ->
     process_flag(trap_exit, true),
-    {PeerAddressS, PeerPort} = peername(ClientSock),
+    {PeerAddress, PeerPort} = socket_op(Sock, fun rabbit_net:peername/1),
+    PeerAddressS = inet_parse:ntoa(PeerAddress),
+    rabbit_log:info("starting TCP connection ~p from ~s:~p~n",
+                    [self(), PeerAddressS, PeerPort]),
+    ClientSock = socket_op(Sock, SockTransform),
+    erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(),
+                      handshake_timeout),
     ProfilingValue = setup_profiling(),
     try 
-        rabbit_log:info("starting TCP connection ~p from ~s:~p~n",
-                        [self(), PeerAddressS, PeerPort]),
-        erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(),
-                          handshake_timeout),
         mainloop(Parent, Deb, switch_callback(
                                 #v1{sock = ClientSock,
                                     connection = #connection{