Use 1 writer process per AMQP 1.0 connection

AMQP 0.9.1 uses a separate rabbit_writer Erlang process per
AMQP 0.9.1 channel.

Prior to this commit, AMQP 1.0 used a separate rabbit_amqp1_0_writer
Erlang process per AMQP 1.0 session.

Advantage of single writer proc per session (prior to this commit):
* High parallelism for serialising packets if multiple sessions within
  a connection write heavily at the same time.

This commit uses a single writer process per AMQP 1.0 connection that is
shared across all AMQP 1.0 sessions.

Advantages of single writer proc per connection (this commit):
* Lower memory usage with hundreds of thousands of AMQP 1.0 sessions
* Less TCP and IP header overhead given that the single writer process
  can accumulate across all sessions bytes worth a MSS before flushing
  the socket.

In other words, this commit decides that a reader / writer process pair
per AMQP 1.0 connection is good enough for bi-directional TRANSFER flows.
Having a writer per session is too heavy. The final goal by previous
commits and follow-up commits is to reduce the total number of Erlang
processes to allow hundreds of thousands of AMQP clients to connect
while keeping resource usage in RabbitMQ at a low level.

We still ensure high thoughput by having separate reader, writer, and
session processes.

Author: ansd

deps/rabbit_common/src/rabbit_net.erl

@@ -32,11 +32,11 @@
                  {raw, non_neg_integer(), non_neg_integer(), binary()}].
 -type hostname() :: inet:hostname().
 -type ip_port() :: inet:port_number().
--type rabbit_proxy_socket() :: {'rabbit_proxy_socket', ranch_transport:socket(), ranch_proxy_header:proxy_info()}.
+-type proxy_socket() :: {'rabbit_proxy_socket', ranch_transport:socket(), ranch_proxy_header:proxy_info()}.
 % -type host_or_ip() :: binary() | inet:ip_address().
 -spec is_ssl(socket()) -> boolean().
 -spec ssl_info(socket()) -> 'nossl' | ok_val_or_error([{atom(), any()}]).
--spec proxy_ssl_info(socket(), rabbit_proxy_socket() | 'undefined') -> 'nossl' | ok_val_or_error([{atom(), any()}]).
+-spec proxy_ssl_info(socket(), proxy_socket() | 'undefined') -> 'nossl' | ok_val_or_error([{atom(), any()}]).
 -spec controlling_process(socket(), pid()) -> ok_or_any_error().
 -spec getstat(socket(), [stat_option()]) ->
           ok_val_or_error([{stat_option(), integer()}]).

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_reader.erl

@@ -28,10 +28,16 @@
 
 %%--------------------------------------------------------------------------
 
--record(v1, {parent, sock, connection, callback, recv_len, pending_recv,
-             connection_state, heartbeater, helper_sup,
-             channel_sup_sup_pid, buf, buf_len, throttle, proxy_socket,
-             tracked_channels}).
+-record(v1,
+        {
+         parent,
+         sock :: rabbit_net:socket(),
+         connection, callback, recv_len, pending_recv,
+         connection_state, heartbeater, helper_sup,
+         channel_sup_sup_pid, buf, buf_len, throttle, proxy_socket,
+         tracked_channels,
+         writer :: rabbit_types:option(pid())
+        }).
 
 -record(v1_connection, {name :: binary(),
                         host,
@@ -385,64 +391,62 @@ parse_1_0_frame(Payload, _Channel) ->
         _    -> {Perf, Rest}
     end.
 
-handle_1_0_connection_frame(#'v1_0.open'{ max_frame_size = ClientFrameMax,
-                                          channel_max = ClientChannelMax,
-                                          idle_time_out = IdleTimeout,
-                                          hostname = Hostname },
-                            State = #v1{
-                              connection_state = starting,
-                              connection = Connection,
-                              throttle   = Throttle,
-                              helper_sup = HelperSupPid,
-                              sock = Sock}) ->
+handle_1_0_connection_frame(
+  #'v1_0.open'{max_frame_size = ClientFrameMax,
+               channel_max = ClientChannelMax,
+               idle_time_out = IdleTimeout,
+               hostname = Hostname },
+  #v1{connection_state = starting,
+      connection = Connection,
+      throttle   = Throttle,
+      helper_sup = HelperSupPid,
+      sock = Sock} = State0) ->
     ClientHeartbeatSec = case IdleTimeout of
-                             undefined        -> 0;
+                             undefined -> 0;
                              {uint, Interval} -> Interval div 1000
                          end,
-    FrameMax           = case ClientFrameMax of
-                             undefined -> unlimited;
-                             {_, FM}   -> FM
-                         end,
+    FrameMax = case ClientFrameMax of
+                   undefined -> unlimited;
+                   {_, FM}   -> FM
+               end,
     {ok, HeartbeatSec} = application:get_env(rabbit, heartbeat),
-    State1 =
-        if (FrameMax =/= unlimited) and (FrameMax < ?FRAME_1_0_MIN_SIZE) ->
-                protocol_error(?V_1_0_AMQP_ERROR_FRAME_SIZE_TOO_SMALL,
-                               "frame_max=~w < ~w min size",
-                               [FrameMax, ?FRAME_1_0_MIN_SIZE]);
-           true ->
-                SendFun =
-                    fun() ->
-                            Frame =
-                                amqp10_binary_generator:build_heartbeat_frame(),
-                            catch rabbit_net:send(Sock, Frame)
-                    end,
-
-                Parent = self(),
-                ReceiveFun =
-                    fun() ->
-                            Parent ! heartbeat_timeout
-                    end,
-                %% [2.4.5] the value in idle-time-out SHOULD be half the peer's
-                %%         actual timeout threshold
-                ReceiverHeartbeatSec = lists:min([HeartbeatSec * 2, 4294967]),
-                %% TODO: only start heartbeat receive timer at next next frame
-                Heartbeater =
-                    rabbit_heartbeat:start(HelperSupPid, Sock,
-                                           ClientHeartbeatSec, SendFun,
-                                           ReceiverHeartbeatSec, ReceiveFun),
-                State#v1{connection_state = running,
-                         connection = Connection#v1_connection{
-                                                   frame_max = FrameMax,
-                                                   hostname  = Hostname},
-                         heartbeater = Heartbeater}
-        end,
+    if (FrameMax =/= unlimited) and (FrameMax < ?FRAME_1_0_MIN_SIZE) ->
+           protocol_error(?V_1_0_AMQP_ERROR_FRAME_SIZE_TOO_SMALL,
+                          "frame_max=~w < ~w min size",
+                          [FrameMax, ?FRAME_1_0_MIN_SIZE]);
+       true ->
+           ok
+    end,
+
+    SendFun = fun() ->
+                      Frame = amqp10_binary_generator:build_heartbeat_frame(),
+                      catch rabbit_net:send(Sock, Frame)
+              end,
+    Parent = self(),
+    ReceiveFun = fun() ->
+                         Parent ! heartbeat_timeout
+                 end,
+    %% [2.4.5] the value in idle-time-out SHOULD be half the peer's
+    %%         actual timeout threshold
+    ReceiverHeartbeatSec = lists:min([HeartbeatSec * 2, 4294967]),
+    %% TODO: only start heartbeat receive timer at next next frame
+    Heartbeater = rabbit_heartbeat:start(
+                    HelperSupPid, Sock,
+                    ClientHeartbeatSec, SendFun,
+                    ReceiverHeartbeatSec, ReceiveFun),
+    State1 = State0#v1{connection_state = running,
+                       connection = Connection#v1_connection{
+                                      frame_max = FrameMax,
+                                      hostname  = Hostname},
+                       heartbeater = Heartbeater},
+    State2 = start_writer(State1),
     HostnameVal = case Hostname of
-                    undefined -> undefined;
-                    null -> undefined;
-                    {utf8, Val} -> Val
+                      undefined -> undefined;
+                      null -> undefined;
+                      {utf8, Val} -> Val
                   end,
     rabbit_log:debug("AMQP 1.0 connection.open frame: hostname = ~ts, extracted vhost = ~ts, idle_timeout = ~tp" ,
-                    [HostnameVal, vhost(Hostname), HeartbeatSec * 1000]),
+                     [HostnameVal, vhost(Hostname), HeartbeatSec * 1000]),
     %% TODO enforce channel_max
     ok = send_on_channel0(
            Sock,
@@ -452,20 +456,35 @@ handle_1_0_connection_frame(#'v1_0.open'{ max_frame_size = ClientFrameMax,
                         container_id   = {utf8, rabbit_nodes:cluster_name()},
                         properties     = server_properties()}),
     Conserve = rabbit_alarm:register(self(), {?MODULE, conserve_resources, []}),
+    State = State2#v1{throttle = Throttle#throttle{alarmed_by = Conserve}},
 
-    Infos = infos(?EVENT_KEYS, State1),
+    Infos = infos(?EVENT_KEYS, State),
     ok = rabbit_core_metrics:connection_created(
            proplists:get_value(pid, Infos),
            Infos),
     ok = rabbit_event:notify(connection_created, Infos),
     ok = rabbit_networking:register_connection(self()),
 
-    control_throttle(
-      State1#v1{throttle = Throttle#throttle{alarmed_by = Conserve}});
+    control_throttle(State);
 
 handle_1_0_connection_frame(_Frame, State) ->
     maybe_close(State#v1{connection_state = closing}).
 
+start_writer(#v1{helper_sup = SupPid,
+                 sock = Sock,
+                 connection = #v1_connection{frame_max = FrameMax}} = State) ->
+    ChildSpec = #{id => writer,
+                  start => {rabbit_amqp1_0_writer, start_link,
+                            [Sock, FrameMax, self()]},
+                  restart => transient,
+                  significant => true,
+                  shutdown => ?WORKER_WAIT,
+                  type => worker,
+                  modules => [rabbit_amqp1_0_writer]
+                 },
+    {ok, Pid} = supervisor:start_child(SupPid, ChildSpec),
+    State#v1{writer = Pid}.
+
 handle_1_0_session_frame(Channel, Frame, State) ->
     case maps:get(Channel, State#v1.tracked_channels, undefined) of
         undefined ->
@@ -640,8 +659,7 @@ send_on_channel0(Sock, Method) ->
     send_on_channel0(Sock, Method, amqp10_framing).
 
 send_on_channel0(Sock, Method, Framing) ->
-    ok = rabbit_amqp1_0_writer:internal_send_command(
-           Sock, 0, Method, Framing).
+    ok = rabbit_amqp1_0_writer:internal_send_command(Sock, Method, Framing).
 
 %% End 1-0
 
@@ -728,23 +746,27 @@ untrack_channel(Channel, State) ->
         error -> State
     end.
 
-send_to_new_1_0_session(Channel, Frame, State) ->
-    #v1{sock = Sock,
-        channel_sup_sup_pid = ChanSupSup,
-        connection = #v1_connection{frame_max = FrameMax,
-                                    hostname  = Hostname,
-                                    user      = User}
-       } = State,
-    %% Note: the equivalent, start_channel is in channel_sup_sup
-
-    case rabbit_amqp1_0_session_sup_sup:start_session(
-           %% NB subtract fixed frame header size
-           ChanSupSup, {amqp10_framing, Sock, Channel,
-                        case FrameMax of
-                            unlimited -> unlimited;
-                            _         -> FrameMax - 8
-                        end,
-                        self(), User, vhost(Hostname)}) of
+send_to_new_1_0_session(
+  Channel, Frame,
+  #v1{channel_sup_sup_pid = ChanSupSup,
+      connection = #v1_connection{frame_max = FrameMax0,
+                                  hostname = Hostname,
+                                  user = User},
+      writer = WriterPid} = State) ->
+    %% Subtract fixed frame header size.
+    FrameMax = case FrameMax0 of
+                   unlimited -> unlimited;
+                   _ -> FrameMax0 - 8
+               end,
+    ChildArg = {amqp10_framing,
+                Channel,
+                FrameMax,
+                self(),
+                WriterPid,
+                User,
+                vhost(Hostname)},
+    %% The equivalent, start_channel is in channel_sup_sup.
+    case rabbit_amqp1_0_session_sup_sup:start_session(ChanSupSup, ChildArg) of
         {ok, _ChSupPid, ChFrPid} ->
             erlang:monitor(process, ChFrPid),
             ModifiedState = track_channel(Channel, ChFrPid, State),

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -162,16 +162,17 @@ handle_info({'EXIT', _Pid, Reason}, State) ->
 
 handle_cast({frame, Frame, FlowPid},
             #state{reader_pid = ReaderPid,
-                   writer_pid = Sock} = State) ->
+                   writer_pid = WriterPid,
+                   channel_num = Ch} = State) ->
     credit_flow:ack(FlowPid),
     try handle_control(Frame, State) of
         {reply, Replies, NewState} when is_list(Replies) ->
             lists:foreach(fun (Reply) ->
-                                  rabbit_amqp1_0_writer:send_command(Sock, Reply)
+                                  rabbit_amqp1_0_writer:send_command(WriterPid, Ch, Reply)
                           end, Replies),
             {noreply, NewState};
         {reply, Reply, NewState} ->
-            rabbit_amqp1_0_writer:send_command(Sock, Reply),
+            rabbit_amqp1_0_writer:send_command(WriterPid, Ch, Reply),
             {noreply, NewState};
         {noreply, _} = NoreplyNewstate ->
             NoreplyNewstate;
@@ -182,17 +183,18 @@ handle_cast({frame, Frame, FlowPid},
             End = #'v1_0.end'{ error = Reason },
             rabbit_log:warning("Closing session for connection ~tp:~n~tp",
                                [ReaderPid, Reason]),
-            ok = rabbit_amqp1_0_writer:send_command_sync(Sock, End),
+            ok = rabbit_amqp1_0_writer:send_command_sync(WriterPid, Ch, End),
             {stop, normal, State};
           exit:normal ->
               {stop, normal, State};
           _:Reason:Stacktrace ->
               {stop, {Reason, Stacktrace}, State}
     end;
 handle_cast({queue_event, _, _} = QEvent,
-            #state{writer_pid = WriterPid} = State0) ->
+            #state{writer_pid = WriterPid,
+                   channel_num = Ch} = State0) ->
     {Reply, State} = handle_queue_event(QEvent, State0),
-    [rabbit_amqp1_0_writer:send_command(WriterPid, F) ||
+    [rabbit_amqp1_0_writer:send_command(WriterPid, Ch, F) ||
      F <- flow_fields(Reply, State)],
     %% TODO noreply_coalesce(State) as done in rabbit_channel to process further msgs
     %% from the mailbox before sending the disposition with confirms / rejects
@@ -443,6 +445,7 @@ handle_control(Flow = #'v1_0.flow'{handle = Handle}, State0) ->
 handle_control(#'v1_0.detach'{handle = Handle,
                               closed = Closed},
                #state{writer_pid = WriterPid,
+                      channel_num = Ch,
                       queue_states = QStates0,
                       vhost = Vhost,
                       user = #user{username = Username}} = State0) ->
@@ -469,15 +472,16 @@ handle_control(#'v1_0.detach'{handle = Handle,
                             State0
                     end
             end,
-    ok = rabbit_amqp1_0_writer:send_command_sync(WriterPid,
-                                                 #'v1_0.detach'{handle = Handle,
-                                                                closed = Closed}),
+    ok = rabbit_amqp1_0_writer:send_command_sync(
+           WriterPid, Ch, #'v1_0.detach'{handle = Handle,
+                                         closed = Closed}),
     {noreply, State};
 
-handle_control(#'v1_0.end'{}, #state{writer_pid = Sock,
+handle_control(#'v1_0.end'{}, #state{writer_pid = WriterPid,
+                                     channel_num = Ch,
                                      queue_states = QStates}) ->
     ok = rabbit_queue_type:close(QStates),
-    ok = rabbit_amqp1_0_writer:send_command_sync(Sock, #'v1_0.end'{}),
+    ok = rabbit_amqp1_0_writer:send_command_sync(WriterPid, Ch, #'v1_0.end'{}),
     stop;
 
 handle_control(Frame, _State) ->
@@ -488,6 +492,7 @@ handle_control(Frame, _State) ->
 send_pending_transfers(#state{outgoing_window = LocalSpace,
                               remote_incoming_window = RemoteSpace,
                               writer_pid = WriterPid,
+                              channel_num = Ch,
                               pending_transfers = Buf0,
                               queue_states = QStates} = State0)
   when RemoteSpace > 0 andalso LocalSpace > 0 ->
@@ -506,12 +511,12 @@ send_pending_transfers(#state{outgoing_window = LocalSpace,
                           {ok, rabbit_classic_queue} ->
                               fun(T, C) ->
                                       rabbit_amqp1_0_writer:send_command_and_notify(
-                                        WriterPid, QPid, self(), T, C)
+                                        WriterPid, Ch, QPid, self(), T, C)
                               end;
                           {ok, _QType} ->
                               fun(T, C) ->
                                       rabbit_amqp1_0_writer:send_command(
-                                        WriterPid, T, C)
+                                        WriterPid, Ch, T, C)
                               end
                       end,
             %% rabbit_basic:maybe_gc_large_msg(Content, GCThreshold)
@@ -532,10 +537,11 @@ send_pending_transfers(#state{outgoing_window = LocalSpace,
             end
     end;
 send_pending_transfers(#state{remote_incoming_window = RemoteSpace,
-                              writer_pid = WriterPid} = State0)
+                              writer_pid = WriterPid,
+                              channel_num = Ch} = State0)
   when RemoteSpace > 0 ->
     {Flow = #'v1_0.flow'{}, State} = bump_outgoing_window(State0),
-    rabbit_amqp1_0_writer:send_command(WriterPid, flow_fields(Flow, State)),
+    rabbit_amqp1_0_writer:send_command(WriterPid, Ch, flow_fields(Flow, State)),
     send_pending_transfers(State);
 send_pending_transfers(State) ->
     State.