Merge pull request #1955 (Worker pools unification)

Unify pool deps

Author: fenek

big_tests/tests/mam_helper.erl

@@ -1104,9 +1104,10 @@ is_mam_possible(Host) ->
     mongoose_helper:is_odbc_enabled(Host) orelse is_riak_enabled(Host) orelse
     is_cassandra_enabled(Host) orelse is_elasticsearch_enabled(Host).
 
+%% TODO create mongoose_riak:get_status() for cleaner checks, same for cassandra and elasticsearch
 is_riak_enabled(_Host) ->
-    case rpc(mim(), mongoose_riak, get_worker, []) of
-        Pid when is_pid(Pid) ->
+    case catch rpc(mim(), mongoose_riak, list_buckets, [<<"default">>]) of
+        {ok, _} ->
             true;
         _ ->
             false

big_tests/tests/push_pubsub_SUITE.erl

@@ -75,14 +75,16 @@ init_per_testcase(CaseName, Config) ->
     MongoosePushMockPort = setup_mock_rest(),
 
     %% Start HTTP pool
-    HTTPOpts = [{mongoose_push_http, [
+    HTTPOpts = [
         {server, "http://localhost:" ++ integer_to_list(MongoosePushMockPort)}
-    ]}],
-    rpc(mongoose_http_client, start, [HTTPOpts]),
+    ],
+    rpc(mongoose_http_client, start, []),
+    rpc(mongoose_http_client, start_pool, [mongoose_push_http, HTTPOpts]),
     escalus:init_per_testcase(CaseName, Config).
 
 
 end_per_testcase(CaseName, Config) ->
+    rpc(mongoose_http_client, stop_pool, [mongoose_push_http]),
     rpc(mongoose_http_client, stop, []),
     teardown_mock_rest(),
     escalus:end_per_testcase(CaseName, Config).

include/mongoose.hrl

@@ -34,3 +34,5 @@
 %% ---------------------------------
 %% Logging mechanism
 -include("mongoose_logger.hrl").
+
+

include/mongoose_wpool.hrl

@@ -0,0 +1,8 @@
+-author("bartlomiej.gorny@erlang-solutions.com").
+
+%% for worker_pool
+-record(mongoose_worker_pool, {name :: atom(),
+    selection_strategy :: atom(),
+    extra :: term(),
+    request_timeout :: pos_integer(),
+    pool_timeout :: pos_integer() }).

rebar.config

@@ -32,8 +32,6 @@
 {deps,
  [
   {base16, ".*", {git, "https://github.com/goj/base16.git", "f78918e"}},
-  {cuesport, ".*", {git, "https://github.com/esl/cuesport.git", "d82ff25"}},
-  {redo, ".*", {git, "https://github.com/Wallapop/redo.git", "35a8d1c"}},
   {exml, ".*", {git, "https://github.com/esl/exml.git", {tag, "3.0.1"}}},
   {lager, ".*", {git, "https://github.com/erlang-lager/lager.git", {ref, "06f676a"}}},
   {lager_syslog, ".*", {git, "https://github.com/erlang-lager/lager_syslog.git", "3.0.3"}},
@@ -45,11 +43,10 @@
   {jiffy, ".*", {git, "https://github.com/davisp/jiffy.git", "0.14.11"}},
   {fusco, ".*", {git, "https://github.com/esl/fusco.git", {branch, "master"}}},
   {idna, ".*", {git, "https://github.com/benoitc/erlang-idna.git", {tag, "1.2.0"}}},
-  {poolboy, ".*", {git, "https://github.com/devinus/poolboy.git", {tag, "1.5.1"}}},
   {uuid, ".*", {git, "https://github.com/okeuday/uuid.git", {tag, "v1.7.1"}}},
   {fast_tls, ".*", {git, "https://github.com/processone/fast_tls.git", "a166f0e9fe78304e5ca628fd5eff57c850241813"}},
   {lasse, ".*", {git, "https://github.com/inaka/lasse.git", "692eaec"}},
-  {worker_pool, ".*", {git, "https://github.com/inaka/worker_pool.git", {tag, "3.0.0"}}},
+  {worker_pool, ".*", {git, "https://github.com/inaka/worker_pool.git", {branch, "fix-spec-3.0.0"}}},
 
   {riakc, ".*", {git, "https://github.com/basho/riak-erlang-client", "2.5.3"}},
   {cqerl, ".*", {git, "https://github.com/esl/cqerl.git", {ref, "08067ae"}}},

rebar.lock

@@ -31,10 +31,6 @@
   {git,"https://github.com/bszaf/csv.git",
        {ref,"b0b854d8be2d69b0c6eb9bae5591008098973af9"}},
   0},
- {<<"cuesport">>,
-  {git,"https://github.com/esl/cuesport.git",
-       {ref,"d82ff25baf393eb11712721fb1b881b4b9f80176"}},
-  0},
  {<<"eini">>,{pkg,<<"eini">>,<<"1.2.4">>},1},
  {<<"eodbc">>,
   {git,"https://github.com/arcusfelis/eodbc.git",
@@ -63,11 +59,11 @@
   0},
  {<<"exometer_report_graphite">>,
   {git,"https://github.com/esl/exometer_report_graphite.git",
-       {ref,"699ea55fb351dcf6d0a72f304e4f5cede0674f50"}},
+       {ref,"264dd7bcbadbd7febcd43917302251286c88b681"}},
   0},
  {<<"exometer_report_statsd">>,
   {git,"https://github.com/esl/exometer_report_statsd.git",
-       {ref,"1264c0ca1c463a4d945936cff42d956463d32925"}},
+       {ref,"75f6f9ef2b3c0a677f9dba9a4b3d9d77c276a032"}},
   0},
  {<<"fast_tls">>,
   {git,"https://github.com/processone/fast_tls.git",
@@ -76,7 +72,7 @@
  {<<"folsom">>,{pkg,<<"folsom">>,<<"0.8.5">>},1},
  {<<"fusco">>,
   {git,"https://github.com/esl/fusco.git",
-       {ref,"599a413e367a50388c7e8733c32c9dac6992133f"}},
+       {ref,"de08adef32d12f0984d6d1c6fec28be207db2e7e"}},
   0},
  {<<"goldrush">>,{pkg,<<"goldrush">>,<<"0.1.9">>},1},
  {<<"gun">>,
@@ -117,7 +113,7 @@
  {<<"lhttpc">>,{pkg,<<"lhttpc">>,<<"1.5.1">>},1},
  {<<"luerl">>,
   {git,"https://github.com/rvirding/luerl.git",
-       {ref,"a9c35b8faff7c77676c12ca4e1beb60959bf515f"}},
+       {ref,"091fca3d03de5de19efebb2bd37523cdef4bbdb5"}},
   2},
  {<<"lz4">>,
   {git,"https://github.com/szktty/erlang-lz4.git",
@@ -154,19 +150,12 @@
        {ref,"1a29ef3e112ee0a8d8ac53bd6963ccc8f7879343"}},
   0},
  {<<"observer_cli">>,{pkg,<<"observer_cli">>,<<"1.1.0">>},0},
- {<<"p1_utils">>,
-  {git,"https://github.com/processone/p1_utils",
-       {ref,"470539abadf8b69c9647d17eb6c033d39dc7e6d7"}},
-  1},
+ {<<"p1_utils">>,{pkg,<<"p1_utils">>,<<"1.0.10">>},1},
  {<<"pa">>,
   {git,"https://github.com/erszcz/pa.git",
        {ref,"e53c540005f4114a854cfa414b00cbc7e9e65ecb"}},
   0},
  {<<"parse_trans">>,{pkg,<<"parse_trans">>,<<"3.1.0">>},1},
- {<<"poolboy">>,
-  {git,"https://github.com/devinus/poolboy.git",
-       {ref,"3bb48a893ff5598f7c73731ac17545206d259fac"}},
-  0},
  {<<"pooler">>,{pkg,<<"pooler">>,<<"1.5.3">>},1},
  {<<"proper">>,
   {git,"https://github.com/manopapad/proper.git",
@@ -188,10 +177,6 @@
   {git,"https://github.com/ferd/recon.git",
        {ref,"fcc1a7db6d367234171ab24a3d1762f94e57ff22"}},
   0},
- {<<"redo">>,
-  {git,"https://github.com/Wallapop/redo.git",
-       {ref,"35a8d1caa7bdf431826e63665646cf0c8196e688"}},
-  0},
  {<<"riak_pb">>,
   {git,"https://github.com/basho/riak_pb",
        {ref,"08771aba2ce4935b715d32d1b92555efdc3da994"}},
@@ -231,7 +216,7 @@
   0},
  {<<"worker_pool">>,
   {git,"https://github.com/inaka/worker_pool.git",
-       {ref,"86de749ba4a7b253eba075940e44bcc026c2c7ba"}},
+       {ref,"888ef6a376a3797d1a517bbf901019558e6a7150"}},
   0}]}.
 [
 {pkg_hash,[
@@ -250,6 +235,7 @@
  {<<"metrics">>, <<"25F094DEA2CDA98213CECC3AEFF09E940299D950904393B2A29D191C346A8486">>},
  {<<"mimerl">>, <<"993F9B0E084083405ED8252B99460C4F0563E41729AB42D9074FD5E52439BE88">>},
  {<<"observer_cli">>, <<"F3A47C22327424291758167DF3C4600DCF8BF34795A3787612019AA5B86A5C21">>},
+ {<<"p1_utils">>, <<"A6D6927114BAC79CF6468A10824125492034AF7071ADC6ED5EBC4DDB443845D4">>},
  {<<"parse_trans">>, <<"1BAD3B959941CC53FFD6F4769A5D2666F9CDF179B2D62826636497D3FBAD9EC0">>},
  {<<"pooler">>, <<"898CD1FA301FC42D4A8ED598CE139B71CA85B54C16AB161152B5CC5FBDCFA1A8">>},
  {<<"ssl_verify_fun">>, <<"28A4D65B7F59893BC2C7DE786DEC1E1555BD742D336043FE644AE956C3497FBE">>}]}

src/auth/ejabberd_auth_http.erl

@@ -43,17 +43,13 @@
 
 -spec start(binary()) -> ok.
 start(Host) ->
-    AuthHost = ejabberd_auth:get_opt(Host, host),
-    PoolSize = ejabberd_auth:get_opt(Host, connection_pool_size, 10),
-    Opts = ejabberd_auth:get_opt(Host, connection_opts, []),
-    ChildMods = [fusco],
-    ChildMF = {fusco, start_link},
-    ChildArgs = {for_all, [AuthHost, Opts]},
-    ChildSpec = {{ejabberd_auth_http_sup, Host},
-                  {cuesport, start_link,
-                   [pool_name(Host), PoolSize, ChildMods, ChildMF, ChildArgs]},
-                  transient, 2000, supervisor, [cuesport | ChildMods]},
-    ejabberd_sup:start_child(ChildSpec),
+    Opts = [{selection_strategy, ejabberd_auth:get_opt(Host, selection_strategy, random_worker)},
+            {server, ejabberd_auth:get_opt(Host, host)},
+            {path_prefix, ejabberd_auth:get_opt(Host, path_prefix, "")},
+            {pool_size, ejabberd_auth:get_opt(Host, connection_pool_size, 20)},
+            {http_opts, ejabberd_auth:get_opt(Host, connection_opts, [])}
+           ],
+    mongoose_http_client:start_pool(pool_name(Host), Opts),
     ok.
 
 
@@ -220,12 +216,6 @@ remove_user_req(LUser, LServer, Password, Method) ->
 make_req(_, _, LUser, LServer, _) when LUser == error orelse LServer == error ->
     {error, invalid_jid};
 make_req(Method, Path, LUser, LServer, Password) ->
-    PathPrefix = case ejabberd_auth:get_opt(LServer, path_prefix) of
-                     undefined ->
-                         <<"/">>;
-                     Prefix ->
-                         ejabberd_binary:string_to_binary(Prefix)
-                 end,
     LUserE = list_to_binary(http_uri:encode(binary_to_list(LUser))),
     LServerE = list_to_binary(http_uri:encode(binary_to_list(LServer))),
     PasswordE = list_to_binary(http_uri:encode(binary_to_list(Password))),
@@ -237,14 +227,11 @@ make_req(Method, Path, LUser, LServer, Password) ->
                       BasicAuth64 = base64:encode(BasicAuth),
                       [{<<"Authorization">>, <<"Basic ", BasicAuth64/binary>>}]
               end,
-    Connection = cuesport:get_worker(existing_pool_name(LServer)),
 
     ?DEBUG("Making request '~s' for user ~s@~s...", [Path, LUser, LServer]),
-    {ok, {{Code, _Reason}, _RespHeaders, RespBody, _, _}} = case Method of
-        get -> fusco:request(Connection, <<PathPrefix/binary, Path/binary, "?", Query/binary>>,
-                             "GET", Header, "", 2, 5000);
-        post -> fusco:request(Connection, <<PathPrefix/binary, Path/binary>>,
-                              "POST", Header, Query, 2, 5000)
+    {ok, {Code, RespBody}} = case Method of
+        get -> mongoose_http_client:get(pool_name(LServer), <<Path/binary, "?", Query/binary>>, Header);
+        post -> mongoose_http_client:post(pool_name(LServer), Path, Header, Query)
     end,
 
     ?DEBUG("Request result: ~s: ~p", [Code, RespBody]),
@@ -266,10 +253,6 @@ make_req(Method, Path, LUser, LServer, Password) ->
 pool_name(Host) ->
     list_to_atom("ejabberd_auth_http_" ++ binary_to_list(Host)).
 
--spec existing_pool_name(binary()) -> atom().
-existing_pool_name(Host) ->
-    list_to_existing_atom("ejabberd_auth_http_" ++ binary_to_list(Host)).
-
 -spec verify_scram_password(binary(), binary(), binary()) ->
     {ok, boolean()} | {error, bad_request | not_exists}.
 verify_scram_password(LUser, LServer, Password) ->

src/ejabberd_app.erl

@@ -49,6 +49,7 @@ start(normal, _Args) ->
     acl:start(),
     ejabberd_node_id:start(),
     ejabberd_ctl:init(),
+    mongoose_wpool:ensure_started(),
     ejabberd_commands:init(),
     mongoose_commands:init(),
     mongoose_subhosts:init(),

src/ejabberd_redis.erl

@@ -535,7 +535,11 @@ handle_info(_Info, State) ->
 %%--------------------------------------------------------------------
 -spec terminate(_, state()) -> 'ok'.
 terminate(_Reason, _State) ->
-    ejabberd_commands:unregister_commands(commands()),
+    try
+        ejabberd_commands:unregister_commands(commands())
+    catch E:R ->
+        ?ERROR_MSG("Caught error while terminating sm: ~p:~p~n~p", [E, R, erlang:get_stacktrace()])
+    end,
     ok.
 
 %%--------------------------------------------------------------------