Use 1 instead of 4 processes per WebMQTT connection

Reducing number of Erlang processes allows better scaling with millions
of clients connecting via MQTT over WebSockets.

In this commit, we partially revert
9c153b2
where support for heartbeats were introduced.

Prior to this commit there were 4 processes per WebMQTT connection
supervised by ranch_conns_sup in the ranch application:
1. rabbit_web_mqtt_connection_sup
2. rabbit_web_mqtt_connection_sup (id=rabbit_web_mqtt_keepalive_sup)
3. rabbit_hearbeat (receiver)
4. rabbit_web_mqtt_handler connection process

After this commit, there is only the 4th process supervised directly by
ranch_conns_sup.

Author: ansd

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl

@@ -80,14 +80,14 @@ cowboy_ws_connection_pid(RanchConnPid) ->
 mqtt_init() ->
   CowboyOpts0  = maps:from_list(get_env(cowboy_opts, [])),
   CowboyWsOpts = maps:from_list(get_env(cowboy_ws_opts, [])),
-
   Routes = cowboy_router:compile([{'_', [
       {get_env(ws_path, "/ws"), rabbit_web_mqtt_handler, [{ws_opts, CowboyWsOpts}]}
   ]}]),
-  CowboyOpts = CowboyOpts0#{env          => #{dispatch => Routes},
-                            middlewares  => [cowboy_router, rabbit_web_mqtt_middleware, cowboy_handler],
-                            proxy_header => get_env(proxy_protocol, false),
-                            stream_handlers => [rabbit_web_mqtt_stream_handler, cowboy_stream_h]},
+  CowboyOpts = CowboyOpts0#{
+                 env => #{dispatch => Routes},
+                 proxy_header => get_env(proxy_protocol, false),
+                 stream_handlers => [rabbit_web_mqtt_stream_handler, cowboy_stream_h]
+                },
   case get_env(tcp_config, []) of
       []       -> ok;
       TCPConf0 -> start_tcp_listener(TCPConf0, CowboyOpts)
@@ -99,59 +99,55 @@ mqtt_init() ->
   ok.
 
 start_tcp_listener(TCPConf0, CowboyOpts) ->
-  {TCPConf, IpStr, Port} = get_tcp_conf(TCPConf0),
-  RanchTransportOpts = #{
-    socket_opts => TCPConf,
-    connection_type => supervisor,
-    max_connections => get_max_connections(),
-    num_acceptors => get_env(num_tcp_acceptors, 10),
-    num_conns_sups => get_env(num_conns_sup, 1)
-  },
-  case ranch:start_listener(rabbit_networking:ranch_ref(TCPConf),
-                            ranch_tcp,
-                            RanchTransportOpts,
-                            rabbit_web_mqtt_connection_sup,
-                            CowboyOpts) of
-      {ok, _}                       -> ok;
-      {error, {already_started, _}} -> ok;
-      {error, ErrTCP}               ->
-          rabbit_log_connection:error(
-              "Failed to start a WebSocket (HTTP) listener. Error: ~p,"
-              " listener settings: ~p",
+    {TCPConf, IpStr, Port} = get_tcp_conf(TCPConf0),
+    RanchRef = rabbit_networking:ranch_ref(TCPConf),
+    RanchTransportOpts =
+    #{
+      socket_opts => TCPConf,
+      max_connections => get_max_connections(),
+      num_acceptors => get_env(num_tcp_acceptors, 10),
+      num_conns_sups => get_env(num_conns_sup, 1)
+     },
+    case cowboy:start_clear(RanchRef, RanchTransportOpts, CowboyOpts) of
+        {ok, _} ->
+            ok;
+        {error, {already_started, _}} ->
+            ok;
+        {error, ErrTCP} ->
+            rabbit_log_connection:error(
+              "Failed to start a WebSocket (HTTP) listener. Error: ~p, listener settings: ~p",
               [ErrTCP, TCPConf]),
-          throw(ErrTCP)
-  end,
-  listener_started(?TCP_PROTOCOL, TCPConf),
-  rabbit_log:info("rabbit_web_mqtt: listening for HTTP connections on ~s:~w",
-                  [IpStr, Port]).
+            throw(ErrTCP)
+    end,
+    listener_started(?TCP_PROTOCOL, TCPConf),
+    rabbit_log:info("rabbit_web_mqtt: listening for HTTP connections on ~s:~w",
+                    [IpStr, Port]).
 
 start_tls_listener(TLSConf0, CowboyOpts) ->
-  rabbit_networking:ensure_ssl(),
-  {TLSConf, TLSIpStr, TLSPort} = get_tls_conf(TLSConf0),
-  RanchTransportOpts = #{
-    socket_opts => TLSConf,
-    connection_type => supervisor,
-    max_connections => get_max_connections(),
-    num_acceptors => get_env(num_ssl_acceptors, 10),
-    num_conns_sups => get_env(num_conns_sup, 1)
-  },
-  case ranch:start_listener(rabbit_networking:ranch_ref(TLSConf),
-                            ranch_ssl,
-                            RanchTransportOpts,
-                            rabbit_web_mqtt_connection_sup,
-                            CowboyOpts) of
-      {ok, _}                       -> ok;
-      {error, {already_started, _}} -> ok;
-      {error, ErrTLS}               ->
-          rabbit_log_connection:error(
-              "Failed to start a TLS WebSocket (HTTPS) listener. Error: ~p,"
-              " listener settings: ~p",
+    rabbit_networking:ensure_ssl(),
+    {TLSConf, TLSIpStr, TLSPort} = get_tls_conf(TLSConf0),
+    RanchRef = rabbit_networking:ranch_ref(TLSConf),
+    RanchTransportOpts =
+    #{
+      socket_opts => TLSConf,
+      max_connections => get_max_connections(),
+      num_acceptors => get_env(num_ssl_acceptors, 10),
+      num_conns_sups => get_env(num_conns_sup, 1)
+     },
+    case cowboy:start_tls(RanchRef, RanchTransportOpts, CowboyOpts) of
+        {ok, _} ->
+            ok;
+        {error, {already_started, _}} ->
+            ok;
+        {error, ErrTLS} ->
+            rabbit_log_connection:error(
+              "Failed to start a TLS WebSocket (HTTPS) listener. Error: ~p, listener settings: ~p",
               [ErrTLS, TLSConf]),
-          throw(ErrTLS)
-  end,
-  listener_started(?TLS_PROTOCOL, TLSConf),
-  rabbit_log:info("rabbit_web_mqtt: listening for HTTPS connections on ~s:~w",
-                  [TLSIpStr, TLSPort]).
+            throw(ErrTLS)
+    end,
+    listener_started(?TLS_PROTOCOL, TLSConf),
+    rabbit_log:info("rabbit_web_mqtt: listening for HTTPS connections on ~s:~w",
+                    [TLSIpStr, TLSPort]).
 
 listener_started(Protocol, Listener) ->
     Port = rabbit_misc:pget(port, Listener),

deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_connection_sup.erl

@@ -25,8 +25,6 @@
 
 -record(state, {
     conn_name,
-    keepalive,
-    keepalive_sup,
     parse_state,
     proc_state,
     state,
@@ -44,6 +42,13 @@
 %%TODO move from deprecated callback results to new callback results
 %% see cowboy_websocket.erl
 
+%%TODO call rabbit_networking:register_non_amqp_connection/1 so that we are notified
+%% when need to force load the 'connection_created' event for the management plugin, see
+%% https://github.com/rabbitmq/rabbitmq-management-agent/issues/58
+%% https://github.com/rabbitmq/rabbitmq-server/blob/90cc0e2abf944141feedaf3190d7b6d8b4741b11/deps/rabbitmq_stream/src/rabbit_stream_reader.erl#L536
+%% https://github.com/rabbitmq/rabbitmq-server/blob/90cc0e2abf944141feedaf3190d7b6d8b4741b11/deps/rabbitmq_stream/src/rabbit_stream_reader.erl#L189
+%% https://github.com/rabbitmq/rabbitmq-server/blob/7eb4084cf879fe6b1d26dd3c837bd180cb3a8546/deps/rabbitmq_management_agent/src/rabbit_mgmt_db_handler.erl#L72
+
 %% cowboy_sub_protcol
 upgrade(Req, Env, Handler, HandlerState) ->
     upgrade(Req, Env, Handler, HandlerState, #{}).
@@ -64,7 +69,6 @@ takeover(Parent, Ref, Socket, Transport, Opts, Buffer, {Handler, HandlerState})
 %% cowboy_websocket
 init(Req, Opts) ->
     {PeerAddr, _PeerPort} = maps:get(peer, Req),
-    {_, KeepaliveSup} = lists:keyfind(keepalive_sup, 1, Opts),
     SockInfo = maps:get(proxy_header, Req, undefined),
     WsOpts0 = proplists:get_value(ws_opts, Opts, #{}),
     %%TODO return idle_timeout?
@@ -83,8 +87,6 @@ init(Req, Opts) ->
                    cowboy_req:set_resp_header(<<"sec-websocket-protocol">>, SecWsProtocol, Req)
            end,
     {?MODULE, Req2, #state{
-                       keepalive          = {none, none},
-                       keepalive_sup      = KeepaliveSup,
                        parse_state        = rabbit_mqtt_frame:initial_state(),
                        state              = running,
                        conserve_resources = false,
@@ -101,7 +103,7 @@ websocket_init(State0 = #state{socket = Sock, peername = PeerAddr}) ->
                       conn_name          = ConnStr,
                       socket             = Sock
                      },
-            rabbit_log_connection:info("accepting Web MQTT connection ~p (~s)", [self(), ConnStr]),
+            rabbit_log_connection:info("Accepting Web MQTT connection ~p (~s)", [self(), ConnStr]),
             RealSocket = rabbit_net:unwrap_socket(Sock),
             ProcessorState = rabbit_mqtt_processor:initial_state(RealSocket,
                                                                  ConnStr,
@@ -173,18 +175,12 @@ websocket_info({'$gen_cast', {close_connection, Reason}}, State = #state{ proc_s
     rabbit_log_connection:warning("Web MQTT disconnecting client with ID '~s' (~p), reason: ~s",
                  [rabbit_mqtt_processor:info(client_id, ProcState), ConnName, Reason]),
     stop(State);
-websocket_info({start_keepalive, Keepalive},
-               State = #state{ socket = Sock, keepalive_sup = KeepaliveSup }) ->
-    %% Only the client has the responsibility for sending keepalives
-    SendFun = fun() -> ok end,
-    Parent = self(),
-    ReceiveFun = fun() -> Parent ! keepalive_timeout end,
-    Heartbeater = rabbit_heartbeat:start(
-                    KeepaliveSup, Sock, 0, SendFun, Keepalive, ReceiveFun),
-    {ok, State #state { keepalive = Heartbeater }, hibernate};
-websocket_info(keepalive_timeout, State = #state{conn_name = ConnStr}) ->
-    rabbit_log_connection:error("closing Web MQTT connection ~p (keepalive timeout)", [ConnStr]),
-    stop(State);
+websocket_info({start_keepalive, _Keepalive}, State) ->
+    %%TODO use timer as done in rabbit_mqtt_reader
+    {ok, State, hibernate};
+% websocket_info(keepalive_timeout, State = #state{conn_name = ConnStr}) ->
+%     rabbit_log_connection:error("closing Web MQTT connection ~p (keepalive timeout)", [ConnStr]),
+%     stop(State);
 websocket_info(emit_stats, State) ->
     {ok, emit_stats(State), hibernate};
 websocket_info({ra_event, _From, Evt},
@@ -284,13 +280,13 @@ handle_credits(State0) ->
 control_throttle(State = #state{ state              = CS,
                                  conserve_resources = Mem }) ->
     case {CS, Mem orelse credit_flow:blocked()} of
-        {running,   true} -> ok = rabbit_heartbeat:pause_monitor(
-                                    State#state.keepalive),
-                             State #state{ state = blocked };
-        {blocked,  false} -> ok = rabbit_heartbeat:resume_monitor(
-                                    State#state.keepalive),
-                             State #state{ state = running };
-        {_,            _} -> State
+        %%TODO cancel / resume keepalive timer as done in rabbit_mqtt_reader
+        {running, true} ->
+            State #state{state = blocked};
+        {blocked,false} ->
+            State #state{state = running};
+        {_, _} ->
+            State
     end.
 
 send_reply(Frame, PState) ->