12580002997

Committed 02 Jan 2025 09:05AM UTC coverage: 82.034%. First build

Build # 12580002997

Build Type

Pull #14475

github

Committed by

web-flow

Commit Message

Merge 2f431ed61 into d5a56c20b

Pull Request Pull Request #14475: refactor(limiter): refactor and simplify the limiter

Run Details

165 of 233 new or added lines in 20 files covered. (70.82%)

56317 of 68651 relevant lines covered (82.03%)

15172.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.0

/apps/emqx/src/emqx_limiter/src/emqx_limiter.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2021-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(emqx_limiter).

%% API
-export([
    check/2,
    restore/2
]).

-export([get_cfg/2, get_names_cfg/2]).

-export([
    internal_allocator/0,
    default_alloc_interval/0,
    calc_capacity/1, calc_capacity/2
]).

-export_type([type/0, limiter/0, zone/0, limiter_name/0]).

-type type() :: private | shared | infinity.
-type limiter() :: #{
    type := type(),
    module := module(),
    atom() => term()
}.

-type zone() :: atom().
-type limiter_name() :: max_conn | messages | bytes.

-define(DEFAULT_ALLOC_INTERVAL, 100).

%%--------------------------------------------------------------------
%%  call backs
%%--------------------------------------------------------------------

-callback check(non_neg_integer(), limiter()) -> {boolean(), limiter()}.
-callback restore(non_neg_integer(), limiter()) -> limiter().

%%--------------------------------------------------------------------
%%  API
%%--------------------------------------------------------------------
check(_Need, undefined) ->
    {ok, undefined};
check(Need, #{module := Mod} = Limiter) ->
    Mod:check(Need, Limiter).

restore(Consumed, #{module := Mod} = Limiter) ->
    Mod:restore(Consumed, Limiter).

get_cfg(Name, Cfg) ->
    NameStr = erlang:atom_to_list(Name),
    {ok, RateKey} = emqx_utils:safe_to_existing_atom(NameStr ++ "_rate"),
    case maps:get(RateKey, Cfg, infinity) of
        infinity ->
            undefined;
        Rate ->
            {ok, BurstKey} = emqx_utils:safe_to_existing_atom(NameStr ++ "_burst"),
            Burst = maps:get(BurstKey, Cfg, 0),
            #{rate => Rate, burst => Burst}
    end.

get_names_cfg(Names, Cfg) ->
    Keys = lists:foldl(
        fun(Name, Acc) ->
            NameStr = erlang:atom_to_list(Name),
            {ok, RateKey} = emqx_utils:safe_to_existing_atom(NameStr ++ "_rate"),
            {ok, BurstKey} = emqx_utils:safe_to_existing_atom(NameStr ++ "_burst"),
            [RateKey, BurstKey | Acc]
        end,
        [],
        Names
    ),
    maps:with(Keys, Cfg).

internal_allocator() ->
    <<"internal_allocator">>.

default_alloc_interval() ->
    ?DEFAULT_ALLOC_INTERVAL.

%% Capacity = rate * interval
%% but if interval is less than 1 second, use 1 second instead of interval,
%% so we can ensure that capacity is at least greater than 1
calc_capacity(Rate) ->
    calc_capacity(Rate, default_alloc_interval()).

calc_capacity(Rate, Interval) ->
    erlang:ceil(Rate * erlang:max(Interval, 1000)).

%%--------------------------------------------------------------------
%%  Internal functions
%%--------------------------------------------------------------------

1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2021-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
3	%%
4	%% Licensed under the Apache License, Version 2.0 (the "License");
5	%% you may not use this file except in compliance with the License.
6	%% You may obtain a copy of the License at
7	%%
8	%% http://www.apache.org/licenses/LICENSE-2.0
9	%%
10	%% Unless required by applicable law or agreed to in writing, software
11	%% distributed under the License is distributed on an "AS IS" BASIS,
12	%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	%% See the License for the specific language governing permissions and
14	%% limitations under the License.
15	%%--------------------------------------------------------------------
16
17	-module(emqx_limiter).
18
19	%% API
20	-export([
21	check/2,
22	restore/2
23	]).
24
25	-export([get_cfg/2, get_names_cfg/2]).
26
27	-export([
28	internal_allocator/0,
29	default_alloc_interval/0,
30	calc_capacity/1, calc_capacity/2
31	]).
32
33	-export_type([type/0, limiter/0, zone/0, limiter_name/0]).
34
35	-type type() :: private \| shared \| infinity.
36	-type limiter() :: #{
37	type := type(),
38	module := module(),
39	atom() => term()
40	}.
41
42	-type zone() :: atom().
43	-type limiter_name() :: max_conn \| messages \| bytes.
44
45	-define(DEFAULT_ALLOC_INTERVAL, 100).
46
47	%%--------------------------------------------------------------------
48	%% call backs
49	%%--------------------------------------------------------------------
50
51	-callback check(non_neg_integer(), limiter()) -> {boolean(), limiter()}.
52	-callback restore(non_neg_integer(), limiter()) -> limiter().
53
54	%%--------------------------------------------------------------------
55	%% API
56	%%--------------------------------------------------------------------
57	check(_Need, undefined) ->
NEW 58	{ok, undefined};	×
59	check(Need, #{module := Mod} = Limiter) ->
60	Mod:check(Need, Limiter).	155✔
61
62	restore(Consumed, #{module := Mod} = Limiter) ->
NEW 63	Mod:restore(Consumed, Limiter).	×
64
65	get_cfg(Name, Cfg) ->
66	NameStr = erlang:atom_to_list(Name),	32,212✔
67	{ok, RateKey} = emqx_utils:safe_to_existing_atom(NameStr ++ "_rate"),	32,212✔
68	case maps:get(RateKey, Cfg, infinity) of	32,212✔
69	infinity ->
70	undefined;	32,204✔
71	Rate ->
72	{ok, BurstKey} = emqx_utils:safe_to_existing_atom(NameStr ++ "_burst"),	8✔
73	Burst = maps:get(BurstKey, Cfg, 0),	8✔
74	#{rate => Rate, burst => Burst}	8✔
75	end.
76
77	get_names_cfg(Names, Cfg) ->
78	Keys = lists:foldl(	4,699✔
79	fun(Name, Acc) ->
80	NameStr = erlang:atom_to_list(Name),	9,398✔
81	{ok, RateKey} = emqx_utils:safe_to_existing_atom(NameStr ++ "_rate"),	9,398✔
82	{ok, BurstKey} = emqx_utils:safe_to_existing_atom(NameStr ++ "_burst"),	9,398✔
83	[RateKey, BurstKey \| Acc]	9,398✔
84	end,
85	[],
86	Names
87	),
88	maps:with(Keys, Cfg).	4,699✔
89
90	internal_allocator() ->
91	<<"internal_allocator">>.	1,550✔
92
93	default_alloc_interval() ->
94	?DEFAULT_ALLOC_INTERVAL.	7,780✔
95
96	%% Capacity = rate * interval
97	%% but if interval is less than 1 second, use 1 second instead of interval,
98	%% so we can ensure that capacity is at least greater than 1
99	calc_capacity(Rate) ->
100	calc_capacity(Rate, default_alloc_interval()).	1,991✔
101
102	calc_capacity(Rate, Interval) ->
103	erlang:ceil(Rate * erlang:max(Interval, 1000)).	2,087✔
104
105	%%--------------------------------------------------------------------
106	%% Internal functions
107	%%--------------------------------------------------------------------

emqx / emqx / 12580002997

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous