Chapril
/
xmpp.chapril.org-ejabberd
mirror of https://github.com/processone/ejabberd.git
24
1
Fork 0
Code Releases Activity

* src/ejabberd_s2s.erl: Max number of connections and max number 

of connections per node now can be specified via
max_s2s_connections and max_s2s_connections_per_node acl rules

SVN Revision: 950
This commit is contained in:
Alexey Shchepin 2007-09-25 16:32:18 +00:00
parent 4b2bd22533
commit f82b84158e
2 changed files with 234 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
ChangeLog
View File

@ -1,10 +1,16 @@
2007-09-25 Alexey Shchepin <alexey@process-one.net>
* src/ejabberd_s2s.erl: Max number of connections and max number
of connections per node now can be specified via
max_s2s_connections and max_s2s_connections_per_node acl rules
2007-09-14 Mickael Remond <mremond@process-one.net> 2007-09-14 Mickael Remond <mremond@process-one.net>
* src/ejabberd_s2s_out.erl: Changed to actual p1_fsm behaviour. It * src/ejabberd_s2s_out.erl: Changed to actual p1_fsm behaviour. It
was working correctly as the API of p1_fsm and gen_fsm is strictly was working correctly as the API of p1_fsm and gen_fsm is strictly
the same. the same.
* src/ejabberd_s2_in.erl: Added debug hook for s2s loop (EJAB-358). * src/ejabberd_s2s_in.erl: Added debug hook for s2s loop (EJAB-358).
* src/ejabberd_c2s.erl: Added debug hook for c2s loop (EJAB-358). * src/ejabberd_c2s.erl: Added debug hook for c2s loop (EJAB-358).

408
src/ejabberd_s2s.erl
View File

@ -3,36 +3,37 @@
%%% Author : Alexey Shchepin <alexey@sevcom.net> %%% Author : Alexey Shchepin <alexey@sevcom.net>
%%% Purpose : S2S connections manager %%% Purpose : S2S connections manager
%%% Created : 7 Dec 2002 by Alexey Shchepin <alexey@sevcom.net> %%% Created : 7 Dec 2002 by Alexey Shchepin <alexey@sevcom.net>
%%% Id : $Id: ejabberd_s2s.erl 820 2007-07-19 21:17:13Z mremond $ %%% Id : $Id$
%%%---------------------------------------------------------------------- %%%----------------------------------------------------------------------
-module(ejabberd_s2s). -module(ejabberd_s2s).
-author('alexey@sevcom.net'). -author('alexey@sevcom.net').
-vsn('$Revision: 820 $ '). -vsn('$Revision$ ').
-behaviour(gen_server). -behaviour(gen_server).
%% API %% API
-export([start_link/0, -export([start_link/0,
route/3, route/3,
have_connection/1, have_connection/1,
has_key/2, has_key/2,
try_register/1, try_register/1,
remove_connection/3, remove_connection/3,
dirty_get_connections/0, dirty_get_connections/0,
allow_host/2, allow_host/2,
ctl_process/2 ctl_process/2
]). ]).
%% gen_server callbacks %% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]). terminate/2, code_change/3]).
-include("ejabberd.hrl"). -include("ejabberd.hrl").
-include("jlib.hrl"). -include("jlib.hrl").
-include("ejabberd_ctl.hrl"). -include("ejabberd_ctl.hrl").
-define(DEFAULT_MAX_S2S_CONNEXIONS_NUMBER, 3). -define(DEFAULT_MAX_S2S_CONNECTIONS_NUMBER, 1).
-define(DEFAULT_MAX_S2S_CONNECTIONS_NUMBER_PER_NODE, 1).
-record(s2s, {fromto, pid, key}). -record(s2s, {fromto, pid, key}).
-record(state, {}). -record(state, {}).
@ -49,64 +50,72 @@ start_link() ->
route(From, To, Packet) -> route(From, To, Packet) ->
case catch do_route(From, To, Packet) of case catch do_route(From, To, Packet) of
{'EXIT', Reason} -> {'EXIT', Reason} ->
?ERROR_MSG("~p~nwhen processing: ~p", ?ERROR_MSG("~p~nwhen processing: ~p",
[Reason, {From, To, Packet}]); [Reason, {From, To, Packet}]);
_ -> _ ->
ok ok
end. end.
remove_connection(FromTo, Pid, Key) -> remove_connection(FromTo, Pid, Key) ->
case catch mnesia:dirty_match_object(s2s, {s2s, FromTo, Pid, '_'}) of case catch mnesia:dirty_match_object(s2s, #s2s{fromto = FromTo,
[#s2s{pid = Pid, key = Key}] -> pid = Pid,
F = fun() -> _ = '_'}) of
mnesia:delete_object(#s2s{fromto = FromTo, [#s2s{pid = Pid, key = Key}] ->
pid = Pid, F = fun() ->
key = Key}) mnesia:delete_object(#s2s{fromto = FromTo,
end, pid = Pid,
mnesia:transaction(F); key = Key})
_ -> end,
ok mnesia:transaction(F);
_ ->
ok
end. end.
have_connection(FromTo) -> have_connection(FromTo) ->
case catch mnesia:dirty_read(s2s, FromTo) of case catch mnesia:dirty_read(s2s, FromTo) of
[_] -> [_] ->
true; true;
_ -> _ ->
false false
end. end.
has_key(FromTo, Key) -> has_key(FromTo, Key) ->
case mnesia:dirty_select(s2s, case mnesia:dirty_select(s2s,
[{#s2s{fromto = FromTo, key = Key, _ = '_'}, [{#s2s{fromto = FromTo, key = Key, _ = '_'},
[], [],
['$_']}]) of ['$_']}]) of
[] -> [] ->
false; false;
_ -> _ ->
true true
end. end.
try_register(FromTo) -> try_register(FromTo) ->
Key = randoms:get_string(), Key = randoms:get_string(),
Max_S2S_Connections_Number = max_s2s_connexions_number(element(1, FromTo)), MaxS2SConnectionsNumber = max_s2s_connections_number(FromTo),
MaxS2SConnectionsNumberPerNode =
max_s2s_connections_number_per_node(FromTo),
F = fun() -> F = fun() ->
case mnesia:read({s2s, FromTo}) of L = mnesia:read({s2s, FromTo}),
L when length(L) < Max_S2S_Connections_Number -> NeededConnections = needed_connections_number(
mnesia:write(#s2s{fromto = FromTo, L, MaxS2SConnectionsNumber,
pid = self(), MaxS2SConnectionsNumberPerNode),
key = Key}), if
{key, Key}; NeededConnections > 0 ->
_ -> mnesia:write(#s2s{fromto = FromTo,
false pid = self(),
end key = Key}),
end, {key, Key};
true ->
false
end
end,
case mnesia:transaction(F) of case mnesia:transaction(F) of
{atomic, Res} -> {atomic, Res} ->
Res; Res;
_ -> _ ->
false false
end. end.
dirty_get_connections() -> dirty_get_connections() ->
@ -126,9 +135,8 @@ dirty_get_connections() ->
init([]) -> init([]) ->
update_tables(), update_tables(),
mnesia:create_table(s2s, [{ram_copies, [node()]}, {type, bag}, mnesia:create_table(s2s, [{ram_copies, [node()]}, {type, bag},
{attributes, record_info(fields, s2s)}]), {attributes, record_info(fields, s2s)}]),
mnesia:add_table_copy(s2s, node(), ram_copies), mnesia:add_table_copy(s2s, node(), ram_copies),
mnesia:add_table_index(s2s, key),
mnesia:subscribe(system), mnesia:subscribe(system),
ejabberd_ctl:register_commands( ejabberd_ctl:register_commands(
[{"incoming-s2s-number", "print number of incoming s2s connections on the node"}, [{"incoming-s2s-number", "print number of incoming s2s connections on the node"},
@ -169,11 +177,11 @@ handle_info({mnesia_system_event, {mnesia_down, Node}}, State) ->
{noreply, State}; {noreply, State};
handle_info({route, From, To, Packet}, State) -> handle_info({route, From, To, Packet}, State) ->
case catch do_route(From, To, Packet) of case catch do_route(From, To, Packet) of
{'EXIT', Reason} -> {'EXIT', Reason} ->
?ERROR_MSG("~p~nwhen processing: ~p", ?ERROR_MSG("~p~nwhen processing: ~p",
[Reason, {From, To, Packet}]); [Reason, {From, To, Packet}]);
_ -> _ ->
ok ok
end, end,
{noreply, State}; {noreply, State};
handle_info(_Info, State) -> handle_info(_Info, State) ->
@ -201,125 +209,162 @@ code_change(_OldVsn, State, _Extra) ->
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
clean_table_from_bad_node(Node) -> clean_table_from_bad_node(Node) ->
F = fun() -> F = fun() ->
Es = mnesia:select( Es = mnesia:select(
s2s, s2s,
[{#s2s{pid = '$1', _ = '_'}, [{#s2s{pid = '$1', _ = '_'},
[{'==', {node, '$1'}, Node}], [{'==', {node, '$1'}, Node}],
['$_']}]), ['$_']}]),
lists:foreach(fun(E) -> lists:foreach(fun(E) ->
mnesia:delete_object(E) mnesia:delete_object(E)
end, Es) end, Es)
end, end,
mnesia:transaction(F). mnesia:transaction(F).
do_route(From, To, Packet) -> do_route(From, To, Packet) ->
?DEBUG("s2s manager~n\tfrom ~p~n\tto ~p~n\tpacket ~P~n", ?DEBUG("s2s manager~n\tfrom ~p~n\tto ~p~n\tpacket ~P~n",
[From, To, Packet, 8]), [From, To, Packet, 8]),
case find_connection(From, To) of case find_connection(From, To) of
{atomic, Pid} when pid(Pid) -> {atomic, Pid} when pid(Pid) ->
?DEBUG("sending to process ~p~n", [Pid]), ?DEBUG("sending to process ~p~n", [Pid]),
{xmlelement, Name, Attrs, Els} = Packet, {xmlelement, Name, Attrs, Els} = Packet,
NewAttrs = jlib:replace_from_to_attrs(jlib:jid_to_string(From), NewAttrs = jlib:replace_from_to_attrs(jlib:jid_to_string(From),
jlib:jid_to_string(To), jlib:jid_to_string(To),
Attrs), Attrs),
send_element(Pid, {xmlelement, Name, NewAttrs, Els}), send_element(Pid, {xmlelement, Name, NewAttrs, Els}),
ok; ok;
{aborted, _Reason} -> {aborted, _Reason} ->
case xml:get_tag_attr_s("type", Packet) of case xml:get_tag_attr_s("type", Packet) of
"error" -> ok; "error" -> ok;
"result" -> ok; "result" -> ok;
_ -> _ ->
Err = jlib:make_error_reply( Err = jlib:make_error_reply(
Packet, ?ERR_SERVICE_UNAVAILABLE), Packet, ?ERR_SERVICE_UNAVAILABLE),
ejabberd_router:route(To, From, Err) ejabberd_router:route(To, From, Err)
end, end,
false false
end. end.
find_connection(From, To) -> find_connection(From, To) ->
#jid{lserver = MyServer} = From, #jid{lserver = MyServer} = From,
#jid{lserver = Server} = To, #jid{lserver = Server} = To,
FromTo = {MyServer, Server}, FromTo = {MyServer, Server},
Max_S2S_Connections_Number = max_s2s_connexions_number(MyServer), MaxS2SConnectionsNumber = max_s2s_connections_number(FromTo),
?INFO_MSG("Finding connection for ~p~n", [FromTo]), MaxS2SConnectionsNumberPerNode =
max_s2s_connections_number_per_node(FromTo),
?DEBUG("Finding connection for ~p~n", [FromTo]),
case catch mnesia:dirty_read(s2s, FromTo) of case catch mnesia:dirty_read(s2s, FromTo) of
{'EXIT', Reason} -> {'EXIT', Reason} ->
{aborted, Reason}; {aborted, Reason};
[] -> [] ->
%% We try to establish all the connections if the host is not a %% We try to establish all the connections if the host is not a
%% service and if the s2s host is not blacklisted or %% service and if the s2s host is not blacklisted or
%% is in whitelist: %% is in whitelist:
case not is_service(From, To) andalso allow_host(MyServer, Server) of case not is_service(From, To) andalso allow_host(MyServer, Server) of
true -> true ->
open_several_connections(Max_S2S_Connections_Number, MyServer, NeededConnections = needed_connections_number(
Server, From, FromTo, Max_S2S_Connections_Number); [], MaxS2SConnectionsNumber,
false -> MaxS2SConnectionsNumberPerNode),
{aborted, error} open_several_connections(
end; NeededConnections, MyServer,
L when is_list(L) , length(L) < Max_S2S_Connections_Number -> Server, From, FromTo,
%% We establish the missing connections for this pair. MaxS2SConnectionsNumber, MaxS2SConnectionsNumberPerNode);
open_several_connections(Max_S2S_Connections_Number-length(L), MyServer, false ->
Server, From, FromTo, Max_S2S_Connections_Number); {aborted, error}
L when is_list(L) -> end;
%% We choose a connexion from the pool of opened ones. L when is_list(L) ->
{atomic, choose_connection(From, L)} NeededConnections = needed_connections_number(
L, MaxS2SConnectionsNumber,
MaxS2SConnectionsNumberPerNode),
if
NeededConnections > 0 ->
%% We establish the missing connections for this pair.
open_several_connections(
NeededConnections, MyServer,
Server, From, FromTo,
MaxS2SConnectionsNumber, MaxS2SConnectionsNumberPerNode);
true ->
%% We choose a connexion from the pool of opened ones.
{atomic, choose_connection(From, L)}
end
end. end.
choose_connection(From, Connections) -> choose_connection(From, Connections) ->
%El = lists:nth(random:uniform(length(Connections)), Connections), choose_pid(From, [C#s2s.pid || C <- Connections]).
% use sticky connections based on the full JID of the sender
Pid = case lists:nth(erlang:phash(From, length(Connections)), Connections) of choose_pid(From, Pids) ->
El when is_record(El, s2s) -> Pids1 = case [P || P <- Pids, node(P) == node()] of
El#s2s.pid; [] -> Pids;
P when is_pid(P) -> Ps -> Ps
P end,
end, %% use sticky connections based on the full JID of the sender
?INFO_MSG("Using ejabberd_s2s_out ~p~n", [Pid]), Pid = lists:nth(erlang:phash(From, length(Pids1)), Pids1),
?DEBUG("Using ejabberd_s2s_out ~p~n", [Pid]),
Pid. Pid.
open_several_connections(N, MyServer, Server, From, FromTo, Max_S2S_Connections_Number) -> open_several_connections(N, MyServer, Server, From, FromTo,
Connections_Result = [new_connection(MyServer, Server, From, FromTo, Max_S2S_Connections_Number) MaxS2SConnectionsNumber,
|| _N <- lists:seq(1, N)], MaxS2SConnectionsNumberPerNode) ->
case [PID || {atomic, PID} <- Connections_Result] of ConnectionsResult =
[new_connection(MyServer, Server, From, FromTo,
MaxS2SConnectionsNumber, MaxS2SConnectionsNumberPerNode)
|| _N <- lists:seq(1, N)],
case [PID || {atomic, PID} <- ConnectionsResult] of
[] -> [] ->
hd(Connections_Result); hd(ConnectionsResult);
PIDs -> PIDs ->
{atomic, choose_connection(From, PIDs)} {atomic, choose_pid(From, PIDs)}
end. end.
new_connection(MyServer, Server, From, FromTo, Max_S2S_Connections_Number) -> new_connection(MyServer, Server, From, FromTo,
MaxS2SConnectionsNumber, MaxS2SConnectionsNumberPerNode) ->
Key = randoms:get_string(), Key = randoms:get_string(),
{ok, Pid} = ejabberd_s2s_out:start( {ok, Pid} = ejabberd_s2s_out:start(
MyServer, Server, {new, Key}), MyServer, Server, {new, Key}),
F = fun() -> F = fun() ->
case mnesia:read({s2s, FromTo}) of L = mnesia:read({s2s, FromTo}),
L when length(L) < Max_S2S_Connections_Number -> NeededConnections = needed_connections_number(
mnesia:write(#s2s{fromto = FromTo, L, MaxS2SConnectionsNumber,
pid = Pid, MaxS2SConnectionsNumberPerNode),
key = Key}), if
?INFO_MSG("New s2s connection started ~p~n", [Pid]), NeededConnections > 0 ->
Pid; mnesia:write(#s2s{fromto = FromTo,
L -> pid = Pid,
choose_connection(From, L) key = Key}),
end ?INFO_MSG("New s2s connection started ~p", [Pid]),
end, Pid;
true ->
choose_connection(From, L)
end
end,
TRes = mnesia:transaction(F), TRes = mnesia:transaction(F),
case TRes of case TRes of
{atomic, Pid} -> {atomic, Pid} ->
ejabberd_s2s_out:start_connection(Pid); ejabberd_s2s_out:start_connection(Pid);
_ -> _ ->
ejabberd_s2s_out:stop_connection(Pid) ejabberd_s2s_out:stop_connection(Pid)
end, end,
TRes. TRes.
max_s2s_connexions_number(Host) -> max_s2s_connections_number({From, To}) ->
case ejabberd_config:get_local_option({max_s2s_connexions_number, Host}) of case acl:match_rule(
N when is_integer(N) -> From, max_s2s_connections, jlib:make_jid("", To, "")) of
N; Max when is_integer(Max) -> Max;
_ -> _ -> ?DEFAULT_MAX_S2S_CONNECTIONS_NUMBER
?DEFAULT_MAX_S2S_CONNEXIONS_NUMBER
end. end.
max_s2s_connections_number_per_node({From, To}) ->
case acl:match_rule(
From, max_s2s_connections_per_node, jlib:make_jid("", To, "")) of
Max when is_integer(Max) -> Max;
_ -> ?DEFAULT_MAX_S2S_CONNECTIONS_NUMBER_PER_NODE
end.
needed_connections_number(Ls, MaxS2SConnectionsNumber,
MaxS2SConnectionsNumberPerNode) ->
LocalLs = [L || L <- Ls, node(L#s2s.pid) == node()],
lists:min([MaxS2SConnectionsNumber - length(Ls),
MaxS2SConnectionsNumberPerNode - length(LocalLs)]).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Function: is_service(From, To) -> true | false %% Function: is_service(From, To) -> true | false
%% Description: Return true if the destination must be considered as a %% Description: Return true if the destination must be considered as a
@ -328,12 +373,12 @@ max_s2s_connexions_number(Host) ->
is_service(From, To) -> is_service(From, To) ->
LFromDomain = From#jid.lserver, LFromDomain = From#jid.lserver,
case ejabberd_config:get_local_option({route_subdomains, LFromDomain}) of case ejabberd_config:get_local_option({route_subdomains, LFromDomain}) of
s2s -> % bypass RFC 3920 10.3 s2s -> % bypass RFC 3920 10.3
false; false;
_ -> _ ->
LDstDomain = To#jid.lserver, LDstDomain = To#jid.lserver,
P = fun(Domain) -> is_subdomain(LDstDomain, Domain) end, P = fun(Domain) -> is_subdomain(LDstDomain, Domain) end,
lists:any(P, ?MYHOSTS) lists:any(P, ?MYHOSTS)
end. end.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
@ -361,39 +406,40 @@ ctl_process(Val, _Args) ->
update_tables() -> update_tables() ->
case catch mnesia:table_info(s2s, type) of case catch mnesia:table_info(s2s, type) of
bag -> bag ->
ok; ok;
{'EXIT', _} -> {'EXIT', _} ->
ok; ok;
_ -> _ ->
% XXX TODO convert it ? % XXX TODO convert it ?
mnesia:delete_table(s2s) mnesia:delete_table(s2s)
end, end,
case catch mnesia:table_info(s2s, attributes) of case catch mnesia:table_info(s2s, attributes) of
[fromto, node, key] -> [fromto, node, key] ->
mnesia:transform_table(s2s, ignore, [fromto, pid, key]), mnesia:transform_table(s2s, ignore, [fromto, pid, key]),
mnesia:clear_table(s2s); mnesia:clear_table(s2s);
[fromto, pid, key] -> [fromto, pid, key] ->
ok; ok;
{'EXIT', _} -> {'EXIT', _} ->
ok ok
end, end,
case lists:member(local_s2s, mnesia:system_info(tables)) of case lists:member(local_s2s, mnesia:system_info(tables)) of
true -> true ->
mnesia:delete_table(local_s2s); mnesia:delete_table(local_s2s);
false -> false ->
ok ok
end. end.
%% Check if host is in blacklist or white list %% Check if host is in blacklist or white list
allow_host(MyServer, S2SHost) -> allow_host(MyServer, S2SHost) ->
case ejabberd_config:get_local_option({{s2s_host, S2SHost},MyServer}) of case ejabberd_config:get_local_option({{s2s_host, S2SHost},MyServer}) of
deny -> false; deny -> false;
allow -> true; allow -> true;
_ -> _ ->
case ejabberd_config:get_local_option({s2s_default_policy, MyServer}) of case ejabberd_config:get_local_option({s2s_default_policy, MyServer}) of
deny -> false; deny -> false;
allow -> true; allow -> true;
_ -> true %% The default s2s policy is allow _ -> true %% The default s2s policy is allow
end end
end. end.