./ct_report/coverage/mongoose_wpool.COVER.html

1 %%%-------------------------------------------------------------------
2 %%% @doc
3 %%% This is here because there are pool options which have to be given when calling
4 %%% the pool (selection strategy, timeout), while we want to set it once for the pool and not
5 %%% worry about them later, hence additional storage.
6 %%% @end
7 %%%-------------------------------------------------------------------
8 -module(mongoose_wpool).
9 -author("bartlomiej.gorny@erlang-solutions.com").
10 -include("mongoose.hrl").
11
12 -record(mongoose_wpool, {
13 name :: pool_name(),
14 atom_name :: wpool:name(),
15 strategy :: wpool:strategy(),
16 call_timeout :: pos_integer()
17 }).
18
19 %% API
20 -export([ensure_started/0,
21 start/2, start/3, start/4, start/5,
22 stop/0, stop/1, stop/2, stop/3,
23 get_worker/2, get_worker/3,
24 call/2, call/3, call/4, call/5,
25 send_request/2, send_request/3, send_request/4, send_request/5,
26 cast/2, cast/3, cast/4, cast/5,
27 get_pool_settings/3, get_pools/0, stats/3]).
28
29 -export([start_sup_pool/3]).
30 -export([start_configured_pools/0, start_configured_pools/1, start_configured_pools/2]).
31 -export([is_configured/1]).
32 -export([make_pool_name/3]).
33 -export([call_start_callback/2]).
34
35 %% Mostly for tests
36 -export([expand_pools/3]).
37
38 -ignore_xref([call/2, cast/2, cast/3, expand_pools/3, get_worker/2,
39 send_request/2, send_request/3, send_request/4, send_request/5,
40 is_configured/2, is_configured/1, is_configured/1, start/2, start/3,
41 start/5, start_configured_pools/1, start_configured_pools/2, stats/3,
42 stop/1, stop/2]).
43
44 -type pool_type() :: redis | http | rdbms | cassandra | elastic | generic | rabbit | ldap.
45
46 %% Config scope
47 -type scope() :: global | host_type | mongooseim:host_type().
48 -type host_type_or_global() :: mongooseim:host_type_or_global().
49 -type tag() :: atom().
50
51 %% Name of a process
52 -type proc_name() :: atom().
53
54 %% ID of a pool. Used as a key for an ETS table
55 -type pool_name() :: {PoolType :: pool_type(),
56 HostType :: host_type_or_global(),
57 Tag :: tag()}.
58
59 -type pool_opts_in() :: map().
60 -type pool_opts() :: [wpool:option()].
61 -type conn_opts() :: map().
62
63 -type pool_map_in() :: #{type := pool_type(),
64 scope := scope(),
65 tag := tag(),
66 opts := pool_opts_in(),
67 conn_opts := conn_opts()}.
68 %% Pool map with expanded HostType argument instead of scope
69 -type pool_map() :: #{type := pool_type(),
70 host_type := host_type_or_global(),
71 tag := tag(),
72 opts := pool_opts(),
73 conn_opts := conn_opts()}.
74 -type pool_error() :: {pool_not_started, term()}.
75 -type worker_result() :: {ok, pid()} | {error, pool_error()}.
76 -type pool_record_result() :: {ok, #mongoose_wpool{}} | {error, pool_error()}.
77 -type start_result() :: {ok, pid()} | {error, term()}.
78 -type stop_result() :: ok | term().
79
80 -export_type([pool_type/0, tag/0, scope/0, proc_name/0, pool_name/0, pool_opts/0, conn_opts/0]).
81
82 -type callback_fun() :: init | start | is_supported_strategy | stop.
83
84 -callback init() -> ok | {error, term()}.
85 -callback start(host_type_or_global(), tag(), WPoolOpts :: pool_opts(), ConnOpts :: conn_opts()) ->
86 {ok, {pid(), proplists:proplist()}} | {ok, pid()} | {error, Reason :: term()}.
87 -callback is_supported_strategy(Strategy :: wpool:strategy()) -> boolean().
88 -callback stop(host_type_or_global(), tag()) -> ok.
89
90 -optional_callbacks([is_supported_strategy/1]).
91
92 ensure_started() ->
93 101 wpool:start(),
94 101 case whereis(mongoose_wpool_sup) of
95 undefined ->
96 
:-(
 mongoose_wpool_sup:start_link();
97 _ ->
98 101 ok
99 end,
100 101 ejabberd_sup:create_ets_table(
101 ?MODULE,
102 [named_table, public,
103 {read_concurrency, true},
104 {keypos, #mongoose_wpool.name},
105 {heir, whereis(mongoose_wpool_sup), undefined}]).
106
107 start_configured_pools() ->
108 101 Pools = mongoose_config:get_opt(outgoing_pools),
109 101 start_configured_pools(Pools).
110
111 start_configured_pools(PoolsIn) ->
112 121 start_configured_pools(PoolsIn, ?ALL_HOST_TYPES).
113
114 start_configured_pools(PoolsIn, HostTypes) ->
115 122 HostTypeSpecific = get_host_type_specific_pools(HostTypes),
116 122 Pools = expand_pools(PoolsIn, HostTypeSpecific, HostTypes),
117 122 [call_callback(init, PoolType, []) || PoolType <- get_unique_types(PoolsIn)],
118 122 [start(Pool) || Pool <- Pools].
119
120 -spec start(pool_map()) -> start_result().
121 start(#{type := PoolType, host_type := HostType, tag := Tag,
122 opts := PoolOpts, conn_opts := ConnOpts}) ->
123 324 start(PoolType, HostType, Tag, PoolOpts, ConnOpts).
124
125 -spec start(pool_type(), pool_opts()) -> start_result().
126 start(PoolType, PoolOpts) ->
127 
:-(
 start(PoolType, global, PoolOpts).
128
129 -spec start(pool_type(), host_type_or_global(), pool_opts()) -> start_result().
130 start(PoolType, HostType, PoolOpts) ->
131 
:-(
 start(PoolType, HostType, default, PoolOpts).
132
133 -spec start(pool_type(), host_type_or_global(), tag(),
134 pool_opts()) -> start_result().
135 start(PoolType, HostType, Tag, PoolOpts) ->
136 72 start(PoolType, HostType, Tag, PoolOpts, #{}).
137
138 -spec start(pool_type(), host_type_or_global(), tag(),
139 pool_opts(), conn_opts()) -> start_result().
140 start(PoolType, HostType, Tag, PoolOpts, ConnOpts) ->
141 396 {Opts0, WpoolOptsIn} = proplists:split(PoolOpts, [strategy, call_timeout]),
142 396 Opts = lists:append(Opts0),
143 396 Strategy = proplists:get_value(strategy, Opts, best_worker),
144 396 CallTimeout = proplists:get_value(call_timeout, Opts, 5000),
145 %% If a callback doesn't explicitly blacklist a strategy, let's proceed.
146 396 CallbackModule = make_callback_module_name(PoolType),
147 396 case catch CallbackModule:is_supported_strategy(Strategy) of
148 false ->
149 
:-(
 error({strategy_not_supported, PoolType, HostType, Tag, Strategy});
150 _ ->
151 396 start(PoolType, HostType, Tag, WpoolOptsIn, ConnOpts, Strategy, CallTimeout)
152 end.
153
154 -spec start(pool_type(), host_type_or_global(), tag(),
155 pool_opts(), conn_opts(), wpool:strategy(), pos_integer()) ->
156 start_result().
157 start(PoolType, HostType, Tag, WpoolOptsIn, ConnOpts, Strategy, CallTimeout) ->
158 396 case mongoose_wpool_mgr:start(PoolType, HostType, Tag, WpoolOptsIn, ConnOpts) of
159 {ok, Pid} ->
160 396 ets:insert(?MODULE, #mongoose_wpool{name = {PoolType, HostType, Tag},
161 atom_name = make_pool_name(PoolType, HostType, Tag),
162 strategy = Strategy,
163 call_timeout = CallTimeout}),
164 396 {ok, Pid};
165 Error ->
166 
:-(
 Error
167 end.
168
169 %% @doc this function starts the worker_pool's pool under a specific supervisor
170 %% in MongooseIM application.
171 %% It's needed for 2 reasons:
172 %% 1. We want to have a full control of all the pools and its restarts
173 %% 2. When a pool is started via wpool:start_pool it's supposed be called by a supervisor,
174 %% if not, there is no way to stop the pool.
175 -spec start_sup_pool(pool_type(), proc_name(), [wpool:option()]) ->
176 {ok, pid()} | {error, term()}.
177 start_sup_pool(PoolType, ProcName, WpoolOpts) ->
178 396 SupName = mongoose_wpool_type_sup:name(PoolType),
179 396 ChildSpec = #{id => ProcName,
180 start => {wpool, start_pool, [ProcName, WpoolOpts]},
181 restart => temporary,
182 type => supervisor,
183 modules => [wpool]},
184 396 supervisor:start_child(SupName, ChildSpec).
185
186 -spec stop() -> term().
187 stop() ->
188 101 [stop_pool(PoolName) || PoolName <- get_pools()].
189
190 -spec stop_pool(pool_name()) -> stop_result().
191 stop_pool({PoolType, HostType, Tag}) ->
192 294 stop(PoolType, HostType, Tag).
193
194 -spec stop(pool_type()) -> stop_result().
195 stop(PoolType) ->
196 
:-(
 stop(PoolType, global).
197
198 -spec stop(pool_type(), host_type_or_global()) -> stop_result().
199 stop(PoolType, HostType) ->
200 
:-(
 stop(PoolType, HostType, default).
201
202 -spec stop(pool_type(), host_type_or_global(), tag()) -> stop_result().
203 stop(PoolType, HostType, Tag) ->
204 387 try
205 387 ets:delete(?MODULE, {PoolType, HostType, Tag}),
206 387 call_callback(stop, PoolType, [HostType, Tag]),
207 387 mongoose_wpool_mgr:stop(PoolType, HostType, Tag)
208 catch
209 C:R:S ->
210 
:-(
 ?LOG_ERROR(#{what => pool_stop_failed,
211 pool_type => PoolType, server => HostType, pool_tag => Tag,
212 pool_key => {PoolType, HostType, Tag},
213 
:-(
 class => C, reason => R, stacktrace => S})
214 end.
215
216 -spec is_configured(pool_type()) -> boolean().
217 is_configured(PoolType) ->
218 35 Pools = mongoose_config:get_opt(outgoing_pools),
219 35 lists:any(fun(#{type := Type}) -> Type =:= PoolType end, Pools).
220
221 -spec get_worker(pool_type(), host_type_or_global()) -> worker_result().
222 get_worker(PoolType, HostType) ->
223 
:-(
 get_worker(PoolType, HostType, default).
224
225 -spec get_worker(pool_type(), host_type_or_global(), tag()) -> worker_result().
226 get_worker(PoolType, HostType, Tag) ->
227 1930 case get_pool(PoolType, HostType, Tag) of
228 {ok, #mongoose_wpool{strategy = Strategy} = Pool} ->
229 1930 Worker = wpool_pool:Strategy(make_pool_name(Pool)),
230 1930 {ok, whereis(Worker)};
231 Err ->
232 
:-(
 Err
233 end.
234
235 call(PoolType, Request) ->
236 
:-(
 call(PoolType, global, Request).
237
238 call(PoolType, HostType, Request) ->
239 
:-(
 call(PoolType, HostType, default, Request).
240
241 call(PoolType, HostType, Tag, Request) ->
242 26687 case get_pool(PoolType, HostType, Tag) of
243 {ok, #mongoose_wpool{strategy = Strategy, call_timeout = CallTimeout} = Pool} ->
244 26350 wpool:call(make_pool_name(Pool), Request, Strategy, CallTimeout);
245 Err ->
246 337 Err
247 end.
248
249 call(PoolType, HostType, Tag, HashKey, Request) ->
250 
:-(
 case get_pool(PoolType, HostType, Tag) of
251 {ok, #mongoose_wpool{call_timeout = CallTimeout} = Pool} ->
252 
:-(
 wpool:call(make_pool_name(Pool), Request, {hash_worker, HashKey}, CallTimeout);
253 Err ->
254 
:-(
 Err
255 end.
256
257 send_request(PoolType, Request) ->
258 
:-(
 send_request(PoolType, global, Request).
259
260 send_request(PoolType, HostType, Request) ->
261 
:-(
 send_request(PoolType, HostType, default, Request).
262
263 send_request(PoolType, HostType, Tag, Request) ->
264 
:-(
 case get_pool(PoolType, HostType, Tag) of
265 {ok, #mongoose_wpool{strategy = Strategy, call_timeout = CallTimeout} = Pool} ->
266 
:-(
 wpool_send_request(make_pool_name(Pool), Request, Strategy, CallTimeout);
267 Err ->
268 
:-(
 Err
269 end.
270
271 send_request(PoolType, HostType, Tag, HashKey, Request) ->
272 
:-(
 case get_pool(PoolType, HostType, Tag) of
273 {ok, #mongoose_wpool{call_timeout = CallTimeout} = Pool} ->
274 
:-(
 wpool_send_request(make_pool_name(Pool), Request, {hash_worker, HashKey}, CallTimeout);
275 Err ->
276 
:-(
 Err
277 end.
278
279 wpool_send_request(PoolName, Request, Strategy, Timeout) ->
280 
:-(
 wpool:send_request(PoolName, Request, Strategy, Timeout).
281
282 cast(PoolType, Request) ->
283 
:-(
 cast(PoolType, global, Request).
284
285 cast(PoolType, HostType, Request) ->
286 
:-(
 cast(PoolType, HostType, default, Request).
287
288 cast(PoolType, HostType, Tag, Request) ->
289 98 case get_pool(PoolType, HostType, Tag) of
290 {ok, #mongoose_wpool{strategy = Strategy} = Pool} ->
291 98 wpool:cast(make_pool_name(Pool), Request, Strategy);
292 Err ->
293 
:-(
 Err
294 end.
295
296 cast(PoolType, HostType, Tag, HashKey, Request) ->
297 386 case get_pool(PoolType, HostType, Tag) of
298 {ok, #mongoose_wpool{} = Pool} ->
299 386 wpool:cast(make_pool_name(Pool), Request, {hash_worker, HashKey});
300 Err ->
301 
:-(
 Err
302 end.
303
304 -spec get_pool_settings(pool_type(), host_type_or_global(), tag()) ->
305 #mongoose_wpool{} | undefined.
306 get_pool_settings(PoolType, HostType, Tag) ->
307 248 case get_pool(PoolType, HostType, Tag) of
308 
:-(
 {ok, PoolRec} -> PoolRec;
309 248 {error, {pool_not_started, _}} -> undefined
310 end.
311
312 -spec get_pools() -> [pool_name()].
313 get_pools() ->
314 101 lists:map(fun(#mongoose_wpool{name = Name}) -> Name end, ets:tab2list(?MODULE)).
315
316 stats(PoolType, HostType, Tag) ->
317 
:-(
 wpool:stats(make_pool_name(PoolType, HostType, Tag)).
318
319 -spec make_pool_name(pool_type(), scope(), tag()) -> proc_name().
320 make_pool_name(PoolType, HostType, Tag) when is_atom(HostType) ->
321 966 make_pool_name(PoolType, atom_to_binary(HostType, utf8), Tag);
322 make_pool_name(PoolType, HostType, Tag) when is_binary(HostType) ->
323 1185 binary_to_atom(<<"mongoose_wpool$", (atom_to_binary(PoolType, utf8))/binary, $$,
324 HostType/binary, $$, (atom_to_binary(Tag, utf8))/binary>>, utf8).
325
326 make_pool_name(#mongoose_wpool{atom_name = undefined, name = {PoolType, HostType, Tag}}) ->
327 
:-(
 make_pool_name(PoolType, HostType, Tag);
328 make_pool_name(#mongoose_wpool{atom_name = AtomName}) ->
329 28764 AtomName.
330
331 -spec call_start_callback(pool_type(), list()) -> term().
332 call_start_callback(PoolType, Args) ->
333 396 call_callback(start, PoolType, Args).
334
335 -spec call_callback(callback_fun(), pool_type(), list()) -> term().
336 call_callback(CallbackFun, PoolType, Args) ->
337 1006 try
338 1006 CallbackModule = make_callback_module_name(PoolType),
339 1006 erlang:apply(CallbackModule, CallbackFun, Args)
340 catch E:R:ST ->
341 
:-(
 ?LOG_ERROR(#{what => pool_callback_failed,
342 pool_type => PoolType, callback_function => CallbackFun,
343 
:-(
 error => E, reason => R, stacktrace => ST}),
344 
:-(
 {error, {callback_crashed, CallbackFun, E, R, ST}}
345 end.
346
347 -spec make_callback_module_name(pool_type()) -> module().
348 make_callback_module_name(PoolType) ->
349 1402 Name = "mongoose_wpool_" ++ atom_to_list(PoolType),
350 1402 list_to_atom(Name).
351
352 -spec get_host_type_specific_pools([mongooseim:host_type()]) -> [pool_map_in()].
353 get_host_type_specific_pools(HostTypes) ->
354 122 lists:append([ mongoose_config:get_opt({outgoing_pools, HostType}, [])
355 122 || HostType <- HostTypes ]).
356
357 -spec expand_pools([pool_map_in()], [pool_map_in()], [mongooseim:host_type()]) -> [pool_map()].
358 expand_pools(Pools, PerHostType, HostTypes) ->
359 %% First we select only vhost/host_type specific pools
360 122 HostSpecific = [ {Type, HT, Tag}
361 122 || #{type := Type, scope := HT, tag := Tag} <- PerHostType ],
362 %% Then we expand all pools with `host_type` as scope
363 %% but using host_type specific configs if they were provided
364 122 F = fun(M = #{type := PoolType, scope := host_type, tag := Tag}) ->
365 1 [M#{scope => HostType} || HostType <- HostTypes,
366 1 not lists:member({PoolType, HostType, Tag}, HostSpecific)];
367 323 (Other) -> [Other]
368 end,
369 122 Pools1 = lists:flatmap(F, Pools),
370 122 lists:map(fun prepare_pool_map/1, PerHostType ++ Pools1).
371
372 -spec prepare_pool_map(pool_map_in()) -> pool_map().
373 prepare_pool_map(Pool = #{scope := HT, opts := Opts}) ->
374 %% Rename "scope" field to "host_type" and change wpool opts to a KV list
375 324 Pool1 = maps:remove(scope, Pool),
376 324 Pool1#{host_type => HT, opts => maps:to_list(Opts)}.
377
378 -spec get_unique_types([pool_map_in()]) -> [pool_type()].
379 get_unique_types(Pools) ->
380 122 lists:usort([maps:get(type, Pool) || Pool <- Pools]).
381
382 -spec get_pool(pool_type(), host_type_or_global(), tag()) -> pool_record_result().
383 get_pool(PoolType, HostType, Tag) ->
384 55946 case ets:lookup(?MODULE, {PoolType, HostType, Tag}) of
385 26597 [] when is_binary(HostType) -> get_pool(PoolType, global, Tag);
386 585 [] -> {error, {pool_not_started, {PoolType, HostType, Tag}}};
387 28764 [Pool] -> {ok, Pool}
388 end.
Line Hits Source