Merge pull request #4599 from esl/max-worker-queue-len

Implement worker queue limit for outgoing pools

Author: telezynski

doc/configuration/outgoing-connections.md

@@ -57,6 +57,17 @@ Number of workers to be started by the pool.
 
 Number of milliseconds after which a call to the pool will time out.
 
+### `outgoing_pools.*.*.max_worker_queue_len`
+* **Syntax:** non-negative integer
+* **Default:** not set
+* **Example:** `max_worker_queue_len = 1000`
+
+Maximum number of requests waiting in the incoming message queue of a worker. By default there is no such limit.
+When this queue length is reached for all workers, further incoming requests will be dropped.
+
+!!! Note
+    This option is applicable only to the `best_worker` strategy. Using it for other strategies is not allowed.
+
 ## Connection options
 
 Options specific to a pool connection are defined in a subsection starting with `[outgoing_pools.*.*.connection]`.
@@ -338,13 +349,6 @@ Sets the RabbitMQ Virtual Host. The host needs to exist, as it is **not** create
 
 Enables/disables one-to-one publishers confirms.
 
-### `outgoing_pools.rabbit.*.connection.max_worker_queue_len`
-* **Syntax:** non-negative integer or `"infinity"`
-* **Default:** `1000`
-* **Example:** `max_worker_queue_len = "infinity"`
-
-Sets a limit of messages in a worker's mailbox above which the worker starts dropping the messages. If a worker message queue length reaches the limit, messages from the head of the queue are dropped until the queue length is again below the limit. Use `infinity` to disable.
-
 ---
 To enable TLS, you need to include the [TLS section](#tls-options) in the connection options.
 

doc/migrations/6.5.0_6.x.x.md

@@ -0,0 +1,7 @@
+## Configuration options
+
+### RabbitMQ connection pool
+
+The RabbitMQ-specific connection option `max_worker_queue_len` is now replaced with a more generic [worker pool option](../configuration/outgoing-connections.md#outgoing_poolsmax_worker_queue_len) with the same name.
+Contrary to the removed option, it now ensures that the queue limit is never exceeded.
+Make sure you use the new option instead of the removed one.

mkdocs.yml

@@ -219,6 +219,7 @@ nav:
       - '6.3.1 to 6.3.2': 'migrations/6.3.1_6.3.2.md'
       - '6.3.3 to 6.4.0': 'migrations/6.3.3_6.4.0.md'
       - '6.4.0 to 6.5.0': 'migrations/6.4.0_6.5.0.md'
+      - '6.5.0 to 6.x.x': 'migrations/6.5.0_6.x.x.md'
       - 'MAM MUC migration helper': 'migrations/jid-from-mam-muc-script.md'
   - 'Contributions to the Ecosystem': 'Contributions.md'
   - 'MongooseIM History': 'History.md'

src/config/mongoose_config_spec.erl

@@ -452,7 +452,9 @@ wpool(ExtraDefaults) ->
                        <<"strategy">> => #option{type = atom,
                                                  validate = {enum, wpool_strategy_values()}},
                        <<"call_timeout">> => #option{type = integer,
-                                                     validate = positive}
+                                                     validate = positive},
+                       <<"max_worker_queue_len">> => #option{type = integer,
+                                                             validate = non_negative}
                       },
              defaults = maps:merge(#{<<"workers">> => 10,
                                      <<"strategy">> => best_worker,
@@ -544,8 +546,6 @@ outgoing_pool_connection(<<"rabbit">>) ->
                  <<"virtual_host">> => #option{type = binary,
                                                validate = non_empty},
                  <<"confirms_enabled">> => #option{type = boolean},
-                 <<"max_worker_queue_len">> => #option{type = int_or_infinity,
-                                                       validate = non_negative},
                  <<"tls">> => tls([client])
                 },
        include = always,
@@ -554,8 +554,7 @@ outgoing_pool_connection(<<"rabbit">>) ->
                     <<"username">> => <<"guest">>,
                     <<"password">> => <<"guest">>,
                     <<"virtual_host">> => <<"/">>,
-                    <<"confirms_enabled">> => false,
-                    <<"max_worker_queue_len">> => 1000}
+                    <<"confirms_enabled">> => false}
       };
 outgoing_pool_connection(<<"rdbms">>) ->
     #section{
@@ -1048,7 +1047,7 @@ check_auth_method(Method, Opts) ->
 
 process_pool([Tag, Type | _], AllOpts = #{scope := ScopeIn, connection := Connection}) ->
     Scope = pool_scope(ScopeIn),
-    Opts = maps:without([scope, host, connection], AllOpts),
+    Opts = verify_pool_strategy(maps:without([scope, host, connection], AllOpts)),
     #{type => b2a(Type),
       scope => Scope,
       tag => b2a(Tag),
@@ -1059,9 +1058,17 @@ process_host_config_pool([Tag, Type, _Pools, {host, HT} | _], AllOpts = #{connec
     #{type => b2a(Type),
       scope => HT,
       tag => b2a(Tag),
-      opts => maps:remove(connection, AllOpts),
+      opts => verify_pool_strategy(maps:remove(connection, AllOpts)),
       conn_opts => Connection}.
 
+verify_pool_strategy(Opts = #{strategy := Strategy, max_worker_queue_len := _})
+  when Strategy =/= best_worker ->
+    error(#{what => invalid_worker_pool_strategy_option,
+            text => <<"max_worker_queue_len can only be set for the best_worker strategy">>,
+            pool_options => Opts});
+verify_pool_strategy(Opts) ->
+    Opts.
+
 pool_scope(host) -> host_type;
 pool_scope(host_type) -> host_type;
 pool_scope(global) -> global.

src/mongoose_rabbit_worker.erl

@@ -44,8 +44,7 @@
                   username := binary(),
                   password := binary(),
                   virtual_host := binary(),
-                  confirms_enabled := boolean(),
-                  max_worker_queue_len := non_neg_integer() | infinity}.
+                  confirms_enabled := boolean()}.
 
 -type publish_result() :: boolean() | timeout | {channel_exception, any(), any()}.
 
@@ -71,20 +70,25 @@ init(State) ->
     {ok, establish_rabbit_connection(State)}.
 
 -spec handle_call({amqp_call, mongoose_amqp:method()}, gen_server:from(), state()) ->
-          {reply, request_dropped | {ok | error | exit | throw, mongoose_amqp:method() | atom()},
-           state()}.
-handle_call(Req, From, State) ->
-    maybe_handle_request(fun do_handle_call/3, [Req, From, State],
-                         {reply, request_dropped, State}).
+          {reply, {ok | error | exit | throw, mongoose_amqp:method() | atom()}, state()}.
+handle_call({amqp_call, Method}, _From, State = #{channel := Channel}) ->
+    try amqp_channel:call(Channel, Method) of
+        Res ->
+            {reply, {ok, Res}, State}
+    catch
+        Error:Reason ->
+            {reply, {Error, Reason}, maybe_restart_rabbit_connection(State)}
+    end.
 
 -spec handle_cast({amqp_publish, mongoose_amqp:method(), mongoose_amqp:message()}, state()) ->
           {noreply, state()}.
-handle_cast(Req, State) ->
-    maybe_handle_request(fun do_handle_cast/2, [Req, State], {noreply, State}).
+handle_cast({amqp_publish, Method, Payload}, State) ->
+    handle_amqp_publish(Method, Payload, State).
 
 -spec handle_info(term(), state()) -> {noreply, state()}.
 handle_info(Req, State) ->
-    maybe_handle_request(fun do_handle_info/2, [Req, State], {noreply, State}).
+    ?UNEXPECTED_INFO(Req),
+    {noreply, State}.
 
 -spec terminate(term(), state()) -> ok.
 terminate(_Reason, #{connection := Connection, channel := Channel,
@@ -96,27 +100,6 @@ terminate(_Reason, #{connection := Connection, channel := Channel,
 %%% Internal functions
 %%%===================================================================
 
--spec do_handle_call({amqp_call, mongoose_amqp:method()}, gen_server:from(), state()) ->
-          {reply, {ok | error | exit | throw, mongoose_amqp:method() | atom()}, state()}.
-do_handle_call({amqp_call, Method}, _From, State = #{channel := Channel}) ->
-    try amqp_channel:call(Channel, Method) of
-        Res ->
-            {reply, {ok, Res}, State}
-    catch
-        Error:Reason ->
-            {reply, {Error, Reason}, maybe_restart_rabbit_connection(State)}
-    end.
-
--spec do_handle_cast({amqp_publish, mongoose_amqp:method(), mongoose_amqp:message()}, state()) ->
-          {noreply, state()}.
-do_handle_cast({amqp_publish, Method, Payload}, State) ->
-    handle_amqp_publish(Method, Payload, State).
-
--spec do_handle_info(term(), state()) -> {noreply, state()}.
-do_handle_info(Req, State) ->
-    ?UNEXPECTED_INFO(Req),
-    {noreply, State}.
-
 -spec handle_amqp_publish(mongoose_amqp:method(), mongoose_amqp:message(), state()) ->
           {noreply, state()}.
 handle_amqp_publish(Method, Payload, State = #{host_type := HostType, pool_tag := PoolTag}) ->
@@ -229,26 +212,3 @@ maybe_enable_confirms(Channel, #{confirms_enabled := true}) ->
     ok;
 maybe_enable_confirms(_Channel, #{}) ->
     ok.
-
--spec maybe_handle_request(Callback :: function(), Args :: [term()], Reply :: term()) -> term().
-maybe_handle_request(Callback, Args, Reply) ->
-    #{opts := #{max_worker_queue_len := Limit}} = lists:last(Args),
-    case is_msq_queue_max_limit_reached(Limit) of
-        false ->
-            apply(Callback, Args);
-        true ->
-            ?LOG_WARNING(#{what => rabbit_worker_request_dropped,
-                           reason => queue_message_length_limit_reached,
-                           limit => Limit}),
-            Reply
-    end.
-
--spec is_msq_queue_max_limit_reached(Limit :: infinity | non_neg_integer()) -> boolean().
-is_msq_queue_max_limit_reached(infinity) -> false;
-is_msq_queue_max_limit_reached(Limit) ->
-    case process_info(self(), message_queue_len) of
-        {_, QueueLen} when QueueLen > Limit ->
-            true;
-        _Else ->
-            false
-    end.

src/wpool/mongoose_wpool.erl

@@ -234,7 +234,7 @@ get_worker(PoolType, HostType) ->
 get_worker(PoolType, HostType, Tag) ->
     case get_pool(PoolType, HostType, Tag) of
         {ok, #mongoose_wpool{strategy = Strategy} = Pool} ->
-            Worker = wpool_pool:Strategy(make_pool_name(Pool)),
+            Worker = get_wpool_worker(make_pool_name(Pool), Strategy),
             {ok, whereis(Worker)};
         Err ->
             Err
@@ -381,7 +381,14 @@ expand_pools(Pools, PerHostType, HostTypes) ->
 prepare_pool_map(Pool = #{scope := HT, opts := Opts}) ->
     %% Rename "scope" field to "host_type" and change wpool opts to a KV list
     Pool1 = maps:remove(scope, Pool),
-    Pool1#{host_type => HT, opts => maps:to_list(Opts)}.
+    Pool1#{host_type => HT, opts => maps:to_list(prepare_pool_opts(Opts))}.
+
+-spec prepare_pool_opts(pool_map_in()) -> pool_map_in().
+prepare_pool_opts(Opts = #{strategy := best_worker, max_worker_queue_len := MaxQueueLen}) ->
+    Opts1 = maps:remove(max_worker_queue_len, Opts),
+    Opts1#{strategy := fun(Name) -> best_worker_with_max_queue_len(Name, MaxQueueLen) end};
+prepare_pool_opts(Opts) ->
+    Opts.
 
 -spec get_unique_types([pool_map_in()], [pool_map_in()]) -> [pool_type()].
 get_unique_types(Pools, HostTypeSpecific) ->
@@ -395,6 +402,12 @@ get_pool(PoolType, HostType, Tag) ->
         [Pool] -> {ok, Pool}
     end.
 
+-spec get_wpool_worker(wpool:name(), wpool:strategy()) -> proc_name().
+get_wpool_worker(PoolName, Strategy) when is_atom(Strategy) ->
+    wpool_pool:Strategy(PoolName);
+get_wpool_worker(PoolName, StrategyFun) when is_function(StrategyFun, 1) ->
+    StrategyFun(PoolName).
+
 -spec instrumentation(pool_type(), host_type_or_global(), tag()) -> [mongoose_instrument:spec()].
 instrumentation(PoolType, HostType, Tag) ->
     CallbackModule = make_callback_module_name(PoolType),
@@ -404,3 +417,13 @@ instrumentation(PoolType, HostType, Tag) ->
         false ->
             []
     end.
+
+-spec best_worker_with_max_queue_len(wpool:name(), pos_integer()) -> atom().
+best_worker_with_max_queue_len(Name, MaxQueueLen) ->
+    Worker = wpool_pool:best_worker(Name),
+    case process_info(whereis(Worker), message_queue_len) of
+        {_, QueueLen} when QueueLen >= MaxQueueLen ->
+            exit(no_available_workers);
+        _ ->
+            Worker
+    end.

test/common/config_parser_helper.erl

@@ -362,9 +362,8 @@ options("outgoing_pools") ->
                           root_dn => <<"cn=admin,dc=example,dc=com">>,
                           servers => ["ldap-server.example.com"]}},
          #{type => rabbit, scope => host_type, tag => event_pusher,
-           opts => #{workers => 20},
-           conn_opts => #{confirms_enabled => true,
-                          max_worker_queue_len => 100}},
+           opts => #{workers => 20, max_worker_queue_len => 100},
+           conn_opts => #{confirms_enabled => true}},
          #{type => rdbms,
            opts => #{workers => 5},
            conn_opts => #{query_timeout => 5000, keepalive_interval => 30,
@@ -829,8 +828,7 @@ default_pool_conn_opts(rabbit) ->
       username => <<"guest">>,
       password => <<"guest">>,
       virtual_host => <<"/">>,
-      confirms_enabled => false,
-      max_worker_queue_len => 1000};
+      confirms_enabled => false};
 default_pool_conn_opts(redis) ->
     #{host => "127.0.0.1",
       port => 6379,

test/config_parser_SUITE.erl

@@ -1105,14 +1105,12 @@ pool_rabbit_connection(_Config) ->
     ?cfg(P ++ [password], <<"pass">>, T(#{<<"password">> => <<"pass">>})),
     ?cfg(P ++ [virtual_host], <<"vh">>, T(#{<<"virtual_host">> => <<"vh">>})),
     ?cfg(P ++ [confirms_enabled], true, T(#{<<"confirms_enabled">> => true})),
-    ?cfg(P ++ [max_worker_queue_len], 100, T(#{<<"max_worker_queue_len">> => 100})),
     ?err(T(#{<<"host">> => <<>>})),
     ?err(T(#{<<"port">> => 123456})),
     ?err(T(#{<<"username">> => <<>>})),
     ?err(T(#{<<"password">> => <<>>})),
     ?err(T(#{<<"virtual_host">> => <<>>})),
-    ?err(T(#{<<"confirms_enabled">> => <<"yes">>})),
-    ?err(T(#{<<"max_worker_queue_len">> => -1})).
+    ?err(T(#{<<"confirms_enabled">> => <<"yes">>})).
 
 pool_rabbit_connection_tls(_Config) ->
     P = [outgoing_pools, 1, conn_opts, tls],
@@ -1152,9 +1150,13 @@ test_pool_opts(Type, Required) ->
     ?cfg(P, default_config([outgoing_pools, Type, default, opts]), T(Required)),
     ?cfg(P ++ [workers], 11, T(Required#{<<"workers">> => 11})),
     ?cfg(P ++ [strategy], random_worker, T(Required#{<<"strategy">> => <<"random_worker">>})),
+    ?cfg(P ++ [max_worker_queue_len], 1000, T(Required#{<<"max_worker_queue_len">> => 1000})),
     ?cfg(P ++ [call_timeout], 999, T(Required#{<<"call_timeout">> => 999})),
     ?err(T(Required#{<<"workers">> => 0})),
     ?err(T(Required#{<<"strategy">> => <<"worst_worker">>})),
+    ?err(T(Required#{<<"max_worker_queue_len">> => -1})),
+    ?err(T(Required#{<<"strategy">> => <<"random_worker">>,
+                     <<"max_worker_queue_len">> => 1000})), % this opt is only for best_worker
     ?err(T(Required#{<<"call_timeout">> => 0})).
 
 test_just_tls_client(P, T) ->

test/config_parser_SUITE_data/outgoing_pools.toml

@@ -47,14 +47,14 @@
 [outgoing_pools.rabbit.event_pusher]
   scope = "host_type"
   workers = 20
+  max_worker_queue_len = 100
 
   [outgoing_pools.rabbit.event_pusher.connection]
     host = "localhost"
     port = 5672
     username = "guest"
     password = "guest"
     confirms_enabled = true
-    max_worker_queue_len = 100
 
 [outgoing_pools.ldap.default]
   scope = "host_type"

test/mongoose_rabbit_worker_SUITE.erl

@@ -33,9 +33,7 @@
 all() ->
     [
      no_request_in_worker_queue_is_lost_when_amqp_call_fails,
-     worker_creates_fresh_amqp_conection_and_channel_when_amqp_call_fails,
-     worker_processes_msgs_when_queue_msg_len_limit_is_not_reached,
-     worker_drops_msgs_when_queue_msg_len_limit_is_reached
+     worker_creates_fresh_amqp_conection_and_channel_when_amqp_call_fails
     ].
 
 %%--------------------------------------------------------------------
@@ -125,46 +123,6 @@ worker_creates_fresh_amqp_conection_and_channel_when_amqp_call_fails(Config) ->
     ?assert(is_pid(Channel3)),
     ?assertNotMatch(ConnectionAndChannel2, ConnectionAndChannel3).
 
-
-worker_processes_msgs_when_queue_msg_len_limit_is_not_reached(Config) ->
-    %% given
-    Worker = proplists:get_value(worker_pid, Config),
-    Ref = make_ref(),
-    Lock = lock_fun(),
-    SendBack = send_back_fun(),
-
-    %% when
-    gen_server:cast(Worker, {amqp_publish, {Lock, [Ref]}, ok}),
-    gen_server:cast(Worker, {amqp_publish, {SendBack, [self(), Ref]}, ok}),
-    [gen_server:cast(Worker, {amqp_publish, ok, ok})
-     || _ <- lists:seq(1, ?MAX_QUEUE_LEN-1)],
-
-    %% unlock the worker
-    Worker ! Ref,
-
-    %% then
-    ?assertReceivedMatch(Ref, 100).
-
-worker_drops_msgs_when_queue_msg_len_limit_is_reached(Config) ->
-    %% given
-    Worker = proplists:get_value(worker_pid, Config),
-    Ref = make_ref(),
-    Lock = lock_fun(),
-    SendBack = send_back_fun(),
-
-    %% when
-    gen_server:cast(Worker, {amqp_publish, {Lock, [Ref]}, ok}),
-    gen_server:cast(Worker, {amqp_publish, {SendBack, [self(), Ref]}, ok}),
-    [gen_server:cast(Worker, {amqp_publish, ok, ok})
-     || _ <- lists:seq(1, ?MAX_QUEUE_LEN+1)],
-
-    %% unlock the worker
-    Worker ! Ref,
-
-    %% then
-    ?assertError({assertReceivedMatch_failed, _},
-                 ?assertReceivedMatch(Ref, 100)).
-
 %%--------------------------------------------------------------------
 %% Helpers
 %%--------------------------------------------------------------------