Remove one supervisory level

Given that we now have 1 writer per AMQP 1.0 connection, this commit
changes the supervisor hierarchy such that only 1 additional process
(rabbit_amqp1_0_session) is created per AMQP 1.0 session.

Author: ansd

deps/rabbitmq_amqp1_0/BUILD.bazel

@@ -50,7 +50,6 @@ rabbitmq_app(
     priv = [":priv"],
     deps = [
         "//deps/amqp10_common:erlang_app",
-        "//deps/amqp_client:erlang_app",
         "//deps/rabbit:erlang_app",
         "//deps/rabbit_common:erlang_app",
     ],

deps/rabbitmq_amqp1_0/app.bzl

@@ -14,7 +14,6 @@ def all_beam_files(name = "all_beam_files"):
             "src/rabbit_amqp1_0_reader.erl",
             "src/rabbit_amqp1_0_session.erl",
             "src/rabbit_amqp1_0_session_sup.erl",
-            "src/rabbit_amqp1_0_session_sup_sup.erl",
             "src/rabbit_amqp1_0_util.erl",
             "src/rabbit_amqp1_0_writer.erl",
         ],
@@ -24,7 +23,6 @@ def all_beam_files(name = "all_beam_files"):
         erlc_opts = "//:erlc_opts",
         deps = [
             "//deps/amqp10_common:erlang_app",
-            "//deps/amqp_client:erlang_app",
             "//deps/rabbit_common:erlang_app",
             "//deps/rabbitmq_cli:erlang_app",
         ],
@@ -45,7 +43,6 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/rabbit_amqp1_0_reader.erl",
             "src/rabbit_amqp1_0_session.erl",
             "src/rabbit_amqp1_0_session_sup.erl",
-            "src/rabbit_amqp1_0_session_sup_sup.erl",
             "src/rabbit_amqp1_0_util.erl",
             "src/rabbit_amqp1_0_writer.erl",
         ],
@@ -55,7 +52,6 @@ def all_test_beam_files(name = "all_test_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = [
             "//deps/amqp10_common:erlang_app",
-            "//deps/amqp_client:erlang_app",
             "//deps/rabbit_common:erlang_app",
             "//deps/rabbitmq_cli:erlang_app",
         ],
@@ -86,7 +82,6 @@ def all_srcs(name = "all_srcs"):
             "src/rabbit_amqp1_0_reader.erl",
             "src/rabbit_amqp1_0_session.erl",
             "src/rabbit_amqp1_0_session_sup.erl",
-            "src/rabbit_amqp1_0_session_sup_sup.erl",
             "src/rabbit_amqp1_0_util.erl",
             "src/rabbit_amqp1_0_writer.erl",
         ],

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_reader.erl

@@ -33,8 +33,10 @@
          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,
+         connection_state, heartbeater,
+         helper_sup :: pid(),
+         session_sup :: rabbit_types:option(pid()),
+         buf, buf_len, throttle, proxy_socket,
          tracked_channels,
          writer :: rabbit_types:option(pid())
         }).
@@ -630,25 +632,23 @@ start_1_0_connection(amqp,
 
 start_1_0_connection0(Mode, State = #v1{connection = Connection,
                                         helper_sup = HelperSup}) ->
-    ChannelSupSupPid =
-        case Mode of
-            sasl ->
-                undefined;
-            amqp ->
-                StartMFA = {rabbit_amqp1_0_session_sup_sup, start_link, []},
-                ChildSpec = #{id => channel_sup_sup,
-                              start => StartMFA,
-                              restart => transient,
-                              significant => true,
-                              shutdown => infinity,
-                              type => supervisor,
-                              modules => [rabbit_amqp1_0_session_sup_sup]},
-                {ok, Pid} = supervisor:start_child(HelperSup, ChildSpec),
-                Pid
-        end,
+    SessionSup = case Mode of
+                     sasl ->
+                         undefined;
+                     amqp ->
+                         ChildSpec = #{id => session_sup,
+                                       start => {rabbit_amqp1_0_session_sup, start_link, [self()]},
+                                       restart => transient,
+                                       significant => true,
+                                       shutdown => infinity,
+                                       type => supervisor,
+                                       modules => [rabbit_amqp1_0_session_sup]},
+                         {ok, Pid} = supervisor:start_child(HelperSup, ChildSpec),
+                         Pid
+                 end,
     switch_callback(State#v1{connection = Connection#v1_connection{
                                             timeout_sec = ?NORMAL_TIMEOUT},
-                             channel_sup_sup_pid = ChannelSupSupPid,
+                             session_sup = SessionSup,
                              connection_state = starting},
                     {frame_header_1_0, Mode}, 8).
 
@@ -734,9 +734,10 @@ auth_phase_1_0(Response,
               handshake, 8)
     end.
 
-track_channel(Channel, ChFrPid, State) ->
-    rabbit_log:debug("AMQP 1.0 opened channel = ~tp " , [{Channel, ChFrPid}]),
-    State#v1{tracked_channels = maps:put(Channel, ChFrPid, State#v1.tracked_channels)}.
+track_channel(ChannelNum, SessionPid, State) ->
+    rabbit_log:debug("AMQP 1.0 created session process ~p for channel number ~b",
+                     [SessionPid, ChannelNum]),
+    State#v1{tracked_channels = maps:put(ChannelNum, SessionPid, State#v1.tracked_channels)}.
 
 untrack_channel(Channel, State) ->
     case maps:take(Channel, State#v1.tracked_channels) of
@@ -747,8 +748,8 @@ untrack_channel(Channel, State) ->
     end.
 
 send_to_new_1_0_session(
-  Channel, Frame,
-  #v1{channel_sup_sup_pid = ChanSupSup,
+  ChannelNum, Frame,
+  #v1{session_sup = SessionSup,
       connection = #v1_connection{frame_max = FrameMax0,
                                   hostname = Hostname,
                                   user = User},
@@ -758,23 +759,20 @@ send_to_new_1_0_session(
                    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),
+    ChildArgs = [WriterPid,
+                 ChannelNum,
+                 FrameMax,
+                 User,
+                 vhost(Hostname)],
+    case rabbit_amqp1_0_session_sup:start_session(SessionSup, ChildArgs) of
+        {ok, SessionPid} ->
+            erlang:monitor(process, SessionPid),
+            ModifiedState = track_channel(ChannelNum, SessionPid, State),
             rabbit_log_connection:info(
                         "AMQP 1.0 connection ~tp: "
                         "user '~ts' authenticated and granted access to vhost '~ts'",
                         [self(), User#user.username, vhost(Hostname)]),
-            ok = rabbit_amqp1_0_session:process_frame(ChFrPid, Frame),
+            ok = rabbit_amqp1_0_session:process_frame(SessionPid, Frame),
             ModifiedState;
         {error, {not_allowed, _}} ->
             rabbit_log:error("AMQP 1.0: user '~ts' is not allowed to access virtual host '~ts'",

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session.erl

@@ -35,7 +35,7 @@
 % [2.8.10]
 -type sequence_no() :: 0..16#ffffffff.
 
--export([start_link/1]).
+-export([start_link/6]).
 -export([process_frame/2,
          get_info/1]).
 -export([init/1,
@@ -115,7 +115,8 @@
                 % confirmed = [] :: [[non_neg_integer(),...]]
                }).
 
-start_link(Args) ->
+start_link(ReaderPid, WriterPid, ChannelNum, FrameMax, User, Vhost) ->
+    Args = {ReaderPid, WriterPid, ChannelNum, FrameMax, User, Vhost},
     gen_server:start_link(?MODULE, Args, []).
 
 get_info(Pid) ->
@@ -125,7 +126,7 @@ process_frame(Pid, Frame) ->
     credit_flow:send(Pid),
     gen_server:cast(Pid, {frame, Frame, self()}).
 
-init({Channel, ReaderPid, WriterPid, User, Vhost, FrameMax}) ->
+init({ReaderPid, WriterPid, ChannelNum, FrameMax, User, Vhost}) ->
     process_flag(trap_exit, true),
     %% TODO tick_timer with consumer_timeout and permission expiry as done in channel?
     % put(permission_cache_can_expire, rabbit_access_control:permission_cache_can_expire(User)),
@@ -134,9 +135,8 @@ init({Channel, ReaderPid, WriterPid, User, Vhost, FrameMax}) ->
                 frame_max = FrameMax,
                 user = User,
                 vhost = Vhost,
-                channel_num            = Channel,
-                next_publish_id        = 0
-               }}.
+                channel_num = ChannelNum,
+                next_publish_id = 0}}.
 
 terminate(_Reason, _State) ->
     ok.

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session_sup.erl

@@ -10,41 +10,31 @@
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 
--export([start_link/1]).
--export([init/1]).
+%% client API
+-export([start_link/1,
+         start_session/2]).
 
--export_type([start_link_arg/0]).
--type start_link_arg() :: {amqp10_framing,
-                           rabbit_channel:channel_number(),
-                           non_neg_integer() | 'unlimited',
-                           pid(),
-                           pid(),
-                           rabbit_types:user(),
-                           rabbit_types:vhost()}.
+%% supervisor callback
+-export([init/1]).
 
--spec start_link(start_link_arg()) ->
-    {'ok', pid(), pid()} | {'error', term()}.
-start_link({amqp10_framing, Channel, FrameMax, ReaderPid, WriterPid, User, VHost}) ->
-    {ok, SupPid} = supervisor:start_link(?MODULE, []),
-    SessionSpec = #{id => channel,
-                    start =>
-                    {rabbit_amqp1_0_session, start_link,
-                     [{Channel, ReaderPid, WriterPid, User, VHost, FrameMax}]},
-                    restart => transient,
-                    significant => true,
-                    shutdown => ?WORKER_WAIT,
-                    type => worker,
-                    modules => [rabbit_amqp1_0_session]},
-    case supervisor:start_child(SupPid, SessionSpec) of
-        {ok, ChannelPid} ->
-            {ok, SupPid, ChannelPid};
-        {error, Reason} ->
-            {error, Reason}
-    end.
+-spec start_link(Reader :: pid()) ->
+    supervisor:startlink_ret().
+start_link(ReaderPid) ->
+    supervisor:start_link(?MODULE, ReaderPid).
 
-init([]) ->
-    SupFlags = #{strategy => one_for_all,
+init(ReaderPid) ->
+    SupFlags = #{strategy => simple_one_for_one,
                  intensity => 0,
-                 period => 1,
-                 auto_shutdown => any_significant},
-    {ok, {SupFlags, []}}.
+                 period => 1},
+    ChildSpec = #{id => amqp1_0_session,
+                  start => {rabbit_amqp1_0_session, start_link, [ReaderPid]},
+                  restart => temporary,
+                  shutdown => ?WORKER_WAIT,
+                  type => worker,
+                  modules => [rabbit_amqp1_0_session]},
+    {ok, {SupFlags, [ChildSpec]}}.
+
+-spec start_session(pid(), list()) ->
+    supervisor:startchild_ret().
+start_session(SessionSupPid, Args) ->
+    supervisor:start_child(SessionSupPid, Args).

deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_session_sup_sup.erl

@@ -435,7 +435,7 @@ module Test =
                 receiver.Close()
             with
             | :? Amqp.AmqpException as ae ->
-                assertEqual (ae.Error.Condition) (Symbol cond)
+                assertEqual (Symbol cond) (ae.Error.Condition)
             | _ -> failwith "invalid expection thrown"
 
     let authFailure uri =

deps/rabbitmq_amqp1_0/test/system_SUITE_data/fsharp-tests/Program.fs

@@ -360,5 +360,5 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/protocol_interop_SUITE.beam"],
         app_name = "rabbitmq_mqtt",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_stomp:erlang_app"],
+        deps = ["//deps/amqp10_common:erlang_app", "//deps/amqp_client:erlang_app", "//deps/rabbitmq_stomp:erlang_app"],
     )

deps/rabbitmq_mqtt/app.bzl

@@ -853,7 +853,6 @@ rabbitmq_amqp1_0:
 - rabbit_amqp1_0_reader
 - rabbit_amqp1_0_session
 - rabbit_amqp1_0_session_sup
-- rabbit_amqp1_0_session_sup_sup
 - rabbit_amqp1_0_util
 - rabbit_amqp1_0_writer
 rabbitmq_auth_backend_cache: