12741272085

Committed 13 Jan 2025 05:33AM UTC coverage: 82.355%. First build

Build # 12741272085

Build Type

Pull #14475

github

Committed by

web-flow

Commit Message

Merge b56877530 into 0fc8025be

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

Run Details

199 of 249 new or added lines in 21 files covered. (79.92%)

57201 of 69457 relevant lines covered (82.35%)

15186.26 hits per line

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

51.85

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

%%--------------------------------------------------------------------
%% Copyright (c) 2019-2025 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_manager).

-behaviour(gen_server).

-include_lib("emqx/include/logger.hrl").
-include_lib("stdlib/include/ms_transform.hrl").

%% API
-export([start_link/0]).

-export([
    find_bucket/2,
    insert_bucket/3,
    delete_bucket/2
]).

-export([
    register_name/2,
    unregister_name/1,
    whereis_name/1,
    send/2
]).

%% gen_server callbacks
-export([
    init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2
]).

-export_type([limiter_name/0, bucket_ref/0]).

-type bucket_ref() :: emqx_limiter_bucket_ref:bucket_ref().
-type limiter_name_key() :: {emqx_limiter:zone(), emqx_limiter:limiter_name()}.
-type tab_key() :: emqx_limiter:zone() | limiter_name_key().
-type tab_value() :: pid() | bucket_ref().

-type limiter_name() ::
    emqx_limiter:limiter_name()
    | emqx_retainer_dispatcher.

-define(TAB, ?MODULE).

%% counter record in ets table
-record(?TAB, {
    key :: tab_key(),
    value :: tab_value()
}).

-define(LIMITER_NAME(Zone, Name), {Zone, Name}).

%%--------------------------------------------------------------------
%%  name callback
%%--------------------------------------------------------------------
register_name(Name, Pid) ->
    case whereis_name(Name) of
        undefined ->
            ets:insert(?TAB, #?TAB{key = Name, value = Pid}),
            yes;
        _ ->
            no
    end.

unregister_name(Name) ->
    ets:delete(?TAB, Name).

whereis_name(Name) ->
    case ets:lookup(?TAB, Name) of
        [#?TAB{value = Pid}] ->
            Pid;
        _ ->
            undefined
    end.

send(Name, Msg) ->
    case whereis_name(Name) of
        Pid when is_pid(Pid) ->
            Pid ! Msg,
            Pid;
        undefined ->
            exit({badarg, {Name, Msg}})
    end.

%%--------------------------------------------------------------------
%%  API
%%--------------------------------------------------------------------

-spec find_bucket(emqx_limiter_allocator:allocator_name(), limiter_name()) ->
    {ok, bucket_ref()} | undefined.
find_bucket(Zone, Name) ->
    case ets:lookup(?TAB, ?LIMITER_NAME(Zone, Name)) of
        [#?TAB{value = Bucket}] ->
            {ok, Bucket};
        _ ->
            undefined
    end.

-spec insert_bucket(
    emqx_limiter_allocator:allocator_name(),
    limiter_name(),
    bucket_ref()
) -> boolean().
insert_bucket(Zone, Name, Bucket) ->
    ets:insert(
        ?TAB,
        #?TAB{key = ?LIMITER_NAME(Zone, Name), value = Bucket}
    ).

-spec delete_bucket(emqx_limiter_allocator:allocator_name(), limiter_name()) -> true.
delete_bucket(Zone, Name) ->
    ets:delete(?TAB, ?LIMITER_NAME(Zone, Name)).

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

%%--------------------------------------------------------------------
%%  gen_server callbacks
%%--------------------------------------------------------------------
init([]) ->
    _ = ets:new(?TAB, [
        set,
        public,
        named_table,
        {keypos, #?TAB.key},
        {write_concurrency, true},
        {read_concurrency, true},
        {heir, erlang:whereis(emqx_limiter_sup), none}
    ]),
    {ok, #{}}.

handle_call(Req, _From, State) ->
    ?SLOG(error, #{msg => "unexpected_call", call => Req}),
    {reply, ignore, State}.

handle_cast(Req, State) ->
    ?SLOG(error, #{msg => "unexpected_cast", cast => Req}),
    {noreply, State}.

handle_info(Info, State) ->
    ?SLOG(error, #{msg => "unexpected_info", info => Info}),
    {noreply, State}.

terminate(_Reason, _State) ->
    ok.

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

1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2019-2025 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_manager).
18
19	-behaviour(gen_server).
20
21	-include_lib("emqx/include/logger.hrl").
22	-include_lib("stdlib/include/ms_transform.hrl").
23
24	%% API
25	-export([start_link/0]).
26
27	-export([
28	find_bucket/2,
29	insert_bucket/3,
30	delete_bucket/2
31	]).
32
33	-export([
34	register_name/2,
35	unregister_name/1,
36	whereis_name/1,
37	send/2
38	]).
39
40	%% gen_server callbacks
41	-export([
42	init/1,
43	handle_call/3,
44	handle_cast/2,
45	handle_info/2,
46	terminate/2
47	]).
48
49	-export_type([limiter_name/0, bucket_ref/0]).
50
51	-type bucket_ref() :: emqx_limiter_bucket_ref:bucket_ref().
52	-type limiter_name_key() :: {emqx_limiter:zone(), emqx_limiter:limiter_name()}.
53	-type tab_key() :: emqx_limiter:zone() \| limiter_name_key().
54	-type tab_value() :: pid() \| bucket_ref().
55
56	-type limiter_name() ::
57	emqx_limiter:limiter_name()
58	\| emqx_retainer_dispatcher.
59
60	-define(TAB, ?MODULE).
61
62	%% counter record in ets table
63	-record(?TAB, {
64	key :: tab_key(),
65	value :: tab_value()
66	}).
67
68	-define(LIMITER_NAME(Zone, Name), {Zone, Name}).
69
70	%%--------------------------------------------------------------------
71	%% name callback
72	%%--------------------------------------------------------------------
73	register_name(Name, Pid) ->
74	case whereis_name(Name) of	1,179✔
75	undefined ->
76	ets:insert(?TAB, #?TAB{key = Name, value = Pid}),	1,179✔
77	yes;	1,179✔
78	_ ->
NEW 79	no	×
80	end.
81
82	unregister_name(Name) ->
NEW 83	ets:delete(?TAB, Name).	×
84
85	whereis_name(Name) ->
86	case ets:lookup(?TAB, Name) of	2,692✔
87	[#?TAB{value = Pid}] ->
88	Pid;	334✔
89	_ ->
90	undefined	2,358✔
91	end.
92
93	send(Name, Msg) ->
NEW 94	case whereis_name(Name) of	×
95	Pid when is_pid(Pid) ->
NEW 96	Pid ! Msg,	×
NEW 97	Pid;	×
98	undefined ->
NEW 99	exit({badarg, {Name, Msg}})	×
100	end.
101
102	%%--------------------------------------------------------------------
103	%% API
104	%%--------------------------------------------------------------------
105
106	-spec find_bucket(emqx_limiter_allocator:allocator_name(), limiter_name()) ->
107	{ok, bucket_ref()} \| undefined.
108	find_bucket(Zone, Name) ->
109	case ets:lookup(?TAB, ?LIMITER_NAME(Zone, Name)) of	34,339✔
110	[#?TAB{value = Bucket}] ->
111	{ok, Bucket};	441✔
112	_ ->
113	undefined	33,898✔
114	end.
115
116	-spec insert_bucket(
117	emqx_limiter_allocator:allocator_name(),
118	limiter_name(),
119	bucket_ref()
120	) -> boolean().
121	insert_bucket(Zone, Name, Bucket) ->
122	ets:insert(	191✔
123	?TAB,
124	#?TAB{key = ?LIMITER_NAME(Zone, Name), value = Bucket}
125	).
126
127	-spec delete_bucket(emqx_limiter_allocator:allocator_name(), limiter_name()) -> true.
128	delete_bucket(Zone, Name) ->
129	ets:delete(?TAB, ?LIMITER_NAME(Zone, Name)).	142✔
130
131	start_link() ->
132	gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).	1,179✔
133
134	%%--------------------------------------------------------------------
135	%% gen_server callbacks
136	%%--------------------------------------------------------------------
137	init([]) ->
138	_ = ets:new(?TAB, [	1,179✔
139	set,
140	public,
141	named_table,
142	{keypos, #?TAB.key},
143	{write_concurrency, true},
144	{read_concurrency, true},
145	{heir, erlang:whereis(emqx_limiter_sup), none}
146	]),
147	{ok, #{}}.	1,179✔
148
149	handle_call(Req, _From, State) ->
150	?SLOG(error, #{msg => "unexpected_call", call => Req}),	×
151	{reply, ignore, State}.	×
152
153	handle_cast(Req, State) ->
154	?SLOG(error, #{msg => "unexpected_cast", cast => Req}),	×
155	{noreply, State}.	×
156
157	handle_info(Info, State) ->
158	?SLOG(error, #{msg => "unexpected_info", info => Info}),	×
159	{noreply, State}.	×
160
161	terminate(_Reason, _State) ->
162	ok.	×
163
164	%%--------------------------------------------------------------------
165	%% Internal functions
166	%%--------------------------------------------------------------------

emqx / emqx / 12741272085

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