diff options
Diffstat (limited to 'lib/diameter/src/transport')
-rw-r--r-- | lib/diameter/src/transport/diameter_sctp.erl | 96 | ||||
-rw-r--r-- | lib/diameter/src/transport/diameter_tcp.erl | 241 |
2 files changed, 189 insertions, 148 deletions
diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl index 6a9f1f940b..4eb3379d59 100644 --- a/lib/diameter/src/transport/diameter_sctp.erl +++ b/lib/diameter/src/transport/diameter_sctp.erl @@ -79,7 +79,7 @@ -type option() :: {sender, boolean()} | sender | {packet, boolean() | raw} - | {message_cb, false | diameter:evaluable()}. + | {message_cb, false | diameter:eval()}. -type uint() :: non_neg_integer(). @@ -102,9 +102,10 @@ streams :: {uint(), uint()} %% {InStream, OutStream} counts | undefined, os = 0 :: uint(), %% next output stream + rotate = 1 :: boolean() | 0 | 1, %% rotate os? packet = true :: boolean() %% legacy transport_data? | raw, - message_cb = false :: false | diameter:evaluable(), + message_cb = false :: false | diameter:eval(), send = false :: pid() | boolean()}). %% sending process %% Monitor process state. @@ -112,7 +113,7 @@ {transport :: pid(), ack = false :: boolean(), socket :: gen_sctp:sctp_socket(), - assoc_id :: gen_sctp:assoc_id()}). %% next output stream + assoc_id :: gen_sctp:assoc_id()}). %% Listener process state. -record(listener, @@ -120,7 +121,7 @@ socket :: gen_sctp:sctp_socket(), service :: pid(), %% service process pending = {0, queue:new()}, - opts :: [[match()] | boolean() | diameter:evaluable()]}). + opts :: [[match()] | boolean() | diameter:eval()]}). %% Field pending implements two queues: the first of transport-to-be %% processes to which an association has been assigned but for which %% diameter hasn't yet spawned a transport process, a short-lived @@ -156,12 +157,7 @@ start(T, Svc, Opts) = Svc, diameter_sctp_sup:start(), %% start supervisors on demand Addrs = Caps#diameter_caps.host_ip_address, - s(T, Addrs, Pid, lists:map(fun ip/1, Opts)). - -ip({ifaddr, A}) -> - {ip, A}; -ip(T) -> - T. + s(T, Addrs, Pid, Opts). %% A listener spawns transports either as a consequence of this call %% when there is not yet an association to assign it, or at comm_up on @@ -354,23 +350,35 @@ l([], Ref, T) -> %% open/3 open(Addrs, Opts, PortNr) -> - {LAs, Os} = addrs(Addrs, Opts), - {LAs, case gen_sctp:open(gen_opts(portnr(Os, PortNr))) of - {ok, Sock} -> - Sock; - {error, Reason} -> - x({open, Reason}) - end}. + case gen_sctp:open(gen_opts(portnr(addrs(Addrs, Opts), PortNr))) of + {ok, Sock} -> + {addrs(Sock), Sock}; + {error, Reason} -> + x({open, Reason}) + end. addrs(Addrs, Opts) -> - case proplists:split(Opts, [ip]) of - {[[]], _} -> - {Addrs, Opts ++ [{ip, A} || A <- Addrs]}; - {[As], Os} -> - LAs = [diameter_lib:ipaddr(A) || {ip, A} <- As], - {LAs, Os ++ [{ip, A} || A <- LAs]} + case lists:mapfoldl(fun ipaddr/2, false, Opts) of + {Os, true} -> + Os; + {_, false} -> + Opts ++ [{ip, A} || A <- Addrs] end. +ipaddr({K,A}, _) + when K == ifaddr; + K == ip -> + {{ip, ipaddr(A)}, true}; +ipaddr(T, B) -> + {T, B}. + +ipaddr(A) + when A == loopback; + A == any -> + A; +ipaddr(A) -> + diameter_lib:ipaddr(A). + portnr(Opts, PortNr) -> case proplists:get_value(port, Opts) of undefined -> @@ -379,6 +387,14 @@ portnr(Opts, PortNr) -> Opts end. +addrs(Sock) -> + case inet:socknames(Sock) of + {ok, As} -> + [A || {A,_} <- As]; + {error, Reason} -> + x({socknames, Reason}) + end. + %% x/1 x(Reason) -> @@ -565,7 +581,7 @@ transition(Msg, S) %% Deferred actions from a message_cb. transition({actions, Dir, Acts}, S) -> - actions(Acts, Dir, S); + setopts(ok, actions(Acts, Dir, S)); %% Request to close the transport connection. transition({diameter, {close, Pid}}, #transport{parent = Pid}) -> @@ -677,11 +693,16 @@ send(#diameter_packet{transport_data = {outstream, SId}} = S) -> send(SId rem OS, Msg, S); -%% ... or not: rotate through all streams. -send(Msg, #transport{streams = {_, OS}, +%% ... or not: rotate when sending on multiple streams ... +send(Msg, #transport{rotate = true, + streams = {_, OS}, os = N} = S) -> - send(N, Msg, S#transport{os = (N + 1) rem OS}). + send(N, Msg, S#transport{os = (N + 1) rem OS}); + +%% ... or send on the only stream available. +send(Msg, S) -> + send(0, Msg, S). %% send/3 @@ -749,7 +770,7 @@ recv({[#sctp_sndrcvinfo{assoc_id = Id}], _Bin} %% Inbound Diameter message. recv({[#sctp_sndrcvinfo{}], Bin} = Msg, S) when is_binary(Bin) -> - message(recv, Msg, S); + message(recv, Msg, recv(S)); recv({_, #sctp_shutdown_event{}}, _) -> stop; @@ -769,6 +790,25 @@ recv({_, #sctp_paddr_change{}}, _) -> recv({_, #sctp_pdapi_event{}}, _) -> ok. +%% recv/1 +%% +%% Start sending unordered after the second reception, so that an +%% outgoing CER/CEA will arrive at the peer before another request. + +recv(#transport{rotate = B} = S) + when is_boolean(B) -> + S; + +recv(#transport{rotate = 0, streams = {_,N}, socket = Sock} = S) -> + ok = inet:setopts(Sock, [{sctp_default_send_param, + #sctp_sndrcvinfo{flags = [unordered]}}]), + S#transport{rotate = 1 < N}; + +recv(#transport{rotate = N} = S) -> + S#transport{rotate = N-1}. + +%% publish/4 + publish(T, Ref, Id, Sock) -> true = diameter_reg:add_new({?MODULE, T, {Ref, {Id, Sock}}}), putr(?INFO_KEY, {gen_sctp, Sock}). %% for info/1 diff --git a/lib/diameter/src/transport/diameter_tcp.erl b/lib/diameter/src/transport/diameter_tcp.erl index a2f393d5d4..a8639baa11 100644 --- a/lib/diameter/src/transport/diameter_tcp.erl +++ b/lib/diameter/src/transport/diameter_tcp.erl @@ -87,8 +87,7 @@ module :: module() | undefined}). -type length() :: 0..16#FFFFFF. %% message length from Diameter header --type size() :: non_neg_integer(). %% accumulated binary size --type frag() :: {length(), size(), binary(), list(binary())} +-type frag() :: maybe_improper_list(length(), binary()) | binary(). -type connect_option() :: {raddr, inet:ip_address()} @@ -111,7 +110,7 @@ -type option() :: {port, non_neg_integer()} | {sender, boolean()} | sender - | {message_cb, false | diameter:evaluable()} + | {message_cb, false | diameter:eval()} | {fragment_timer, 0..16#FFFFFFFF}. %% Accepting/connecting transport process state. @@ -126,7 +125,7 @@ timeout :: infinity | 0..16#FFFFFFFF, %% fragment timeout tref = false :: false | reference(), %% fragment timer reference flush = false :: boolean(), %% flush fragment at timeout? - message_cb :: false | diameter:evaluable(), + message_cb :: false | diameter:eval(), send :: pid() | false}). %% sending process %% The usual transport using gen_tcp can be replaced by anything @@ -143,8 +142,7 @@ -> {ok, pid(), [inet:ip_address()]} when Ref :: diameter:transport_ref(); ({connect, Ref}, #diameter_service{}, [connect_option()]) - -> {ok, pid(), [inet:ip_address()]} - | {ok, pid()} + -> {ok, pid()} when Ref :: diameter:transport_ref(). start({T, Ref}, Svc, Opts) -> @@ -259,22 +257,14 @@ i(#monitor{parent = Pid, transport = TPid} = S) -> i({listen, Ref, {Mod, Opts, Addrs}}) -> [_] = diameter_config:subscribe(Ref, transport), %% assert existence - {[LA, LP], Rest} = proplists:split(Opts, [ip, port]), - LAddrOpt = get_addr(LA, Addrs), - LPort = get_port(LP), - {ok, LSock} = Mod:listen(LPort, gen_opts(LAddrOpt, Rest)), - LAddr = laddr(LAddrOpt, Mod, LSock), + {[LP], Rest} = proplists:split(Opts, [port]), + {ok, LSock} = Mod:listen(get_port(LP), gen_opts(Addrs, Rest)), + {ok, {LAddr, _}} = sockname(Mod, LSock), true = diameter_reg:add_new({?MODULE, listener, {Ref, {LAddr, LSock}}}), proc_lib:init_ack({ok, self(), {LAddr, LSock}}), #listener{socket = LSock, module = Mod}. -laddr([], Mod, Sock) -> - {ok, {Addr, _Port}} = sockname(Mod, Sock), - Addr; -laddr([{ip, Addr}], _, _) -> - Addr. - ssl_opts([]) -> false; ssl_opts([{ssl_options, true}]) -> @@ -309,24 +299,16 @@ init(accept = T, Ref, Mod, Pid, Opts, Addrs, SvcPid) -> Sock; init(connect = T, Ref, Mod, Pid, Opts, Addrs, _SvcPid) -> - {[LA, RA, RP], Rest} = proplists:split(Opts, [ip, raddr, rport]), - LAddrOpt = get_addr(LA, Addrs), + {[RA, RP], Rest} = proplists:split(Opts, [raddr, rport]), RAddr = get_addr(RA), RPort = get_port(RP), - proc_lib:init_ack(init_rc(LAddrOpt)), - Sock = ok(connect(Mod, RAddr, RPort, gen_opts(LAddrOpt, Rest))), + proc_lib:init_ack({ok, self()}), + Sock = ok(connect(Mod, RAddr, RPort, gen_opts(Addrs, Rest))), publish(Mod, T, Ref, Sock), - up(Pid, {RAddr, RPort}, LAddrOpt, Mod, Sock), + up(Pid, {RAddr, RPort}, Mod, Sock), Sock. -init_rc([{ip, Addr}]) -> - {ok, self(), [Addr]}; -init_rc([]) -> - {ok, self()}. - -up(Pid, Remote, [{ip, _Addr}], _, _) -> - diameter_peer:up(Pid, Remote); -up(Pid, Remote, [], Mod, Sock) -> +up(Pid, Remote, Mod, Sock) -> {Addr, _Port} = ok(sockname(Mod, Sock)), diameter_peer:up(Pid, Remote, [Addr]). @@ -383,25 +365,41 @@ l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) -> l([], Ref, T) -> diameter_tcp_sup:start_child({listen, Ref, T}). -%% get_addr/1 +%% addrs/2 +%% +%% Take the first address from the service if several are specified +%% and not address is configured. + +addrs(Addrs, Opts) -> + case lists:mapfoldr(fun ipaddr/2, [], Opts) of + {Os, [_]} -> + Os; + {_, []} -> + Opts ++ [{ip, A} || [A|_] <- [Addrs]]; + {_, As} -> + ?ERROR({invalid_addrs, As, Addrs}) + end. -get_addr(As) -> - diameter_lib:ipaddr(addr(As, [])). +ipaddr({K,A}, As) + when K == ifaddr; + K == ip -> + {{ip, ipaddr(A)}, [A | As]}; +ipaddr(T, B) -> + {T, B}. -%% get_addr/2 +ipaddr(A) + when A == loopback; + A == any -> + A; +ipaddr(A) -> + diameter_lib:ipaddr(A). -get_addr([], []) -> - []; -get_addr(As, Def) -> - [{ip, diameter_lib:ipaddr(addr(As, Def))}]. +%% get_addr/1 -%% Take the first address from the service if several are unspecified. -addr([], [Addr | _]) -> - Addr; -addr([{_, Addr}], _) -> - Addr; -addr(As, Addrs) -> - ?ERROR({invalid_addrs, As, Addrs}). +get_addr([{_, Addr}]) -> + diameter_lib:ipaddr(Addr); +get_addr(Addrs) -> + ?ERROR({invalid_addrs, Addrs}). %% get_port/1 @@ -414,10 +412,15 @@ get_port(Ps) -> %% gen_opts/2 -gen_opts(LAddrOpt, Opts) -> +gen_opts(Addrs, Opts) -> + gen_opts(addrs(Addrs, Opts)). + +%% gen_opts/1 + +gen_opts(Opts) -> {L,_} = proplists:split(Opts, [binary, packet, active]), [[],[],[]] == L orelse ?ERROR({reserved_options, Opts}), - [binary, {packet, 0}, {active, false}] ++ LAddrOpt ++ Opts. + [binary, {packet, 0}, {active, false} | Opts]. %% --------------------------------------------------------------------------- %% # ports/1 @@ -599,11 +602,12 @@ t(T,S) -> %% Incoming packets. transition({P, Sock, Bin}, #transport{socket = Sock, - ssl = B} + ssl = B, + frag = Frag} = S) when P == ssl, true == B; P == tcp -> - recv(Bin, S#transport{active = false}); + recv(acc(Frag, Bin), S); %% Capabilties exchange has decided on whether or not to run over TLS. transition({diameter, {tls, Ref, Type, B}}, #transport{parent = Pid} @@ -640,7 +644,7 @@ transition(Msg, S) %% Deferred actions from a message_cb. transition({actions, Dir, Acts}, S) -> - actions(Acts, Dir, S); + setopts(actions(Acts, Dir, S)); %% Request to close the transport connection. transition({diameter, {close, Pid}}, #transport{parent = Pid, @@ -720,86 +724,77 @@ tls(accept, Sock, Opts) -> %% using Nagle. %% Receive packets until a full message is received, -recv(Bin, #transport{frag = Head} = S) -> - case rcv(Head, Bin) of - {Msg, B} -> %% have a complete message ... - message(recv, Msg, S#transport{frag = B}); - Frag -> %% read more on the socket - start_fragment_timer(setopts(S#transport{frag = Frag, - flush = false})) - end. -%% rcv/2 +recv({Msg, Rest}, S) -> %% have a complete message ... + recv(acc(Rest), message(recv, Msg, S)); + +recv(Frag, #transport{recv = B, + socket = Sock, + module = M} + = S) -> %% or not + B andalso setopts(M, Sock), + start_fragment_timer(S#transport{frag = Frag, + flush = false, + active = B}). -%% No previous fragment. -rcv(<<>>, Bin) -> - rcv(Bin); +%% acc/2 -%% Not even the first four bytes of the header. -rcv(Head, Bin) - when is_binary(Head) -> - rcv(<<Head/binary, Bin/binary>>); +%% Know how many bytes to extract. +acc([Len | Acc], Bin) -> + acc1(Len, <<Acc/binary, Bin/binary>>); -%% Or enough to know how many bytes to extract. -rcv({Len, N, Head, Acc}, Bin) -> - rcv(Len, N + size(Bin), Head, [Bin | Acc]). +%% Or not. +acc(Head, Bin) -> + acc(<<Head/binary, Bin/binary>>). -%% rcv/4 +%% acc1/3 %% Extract a message for which we have all bytes. -rcv(Len, N, Head, Acc) - when Len =< N -> - recv1(Len, bin(Head, Acc)); +acc1(Len, Bin) + when Len =< byte_size(Bin) -> + split_binary(Bin, Len); %% Wait for more packets. -rcv(Len, N, Head, Acc) -> - {Len, N, Head, Acc}. - -%% rcv/1 - -%% Nothing left. -rcv(<<>> = Bin) -> - Bin; - -%% The Message Length isn't even sufficient for a header. Chances are -%% things will go south from here but if we're lucky then the bytes we -%% have extend to an intended message boundary and we can recover by -%% simply receiving them. Make it so. -rcv(<<_:1/binary, Len:24, _/binary>> = Bin) - when Len < 20 -> - {Bin, <<>>}; - -%% Enough bytes to extract a message. -rcv(<<_:1/binary, Len:24, _/binary>> = Bin) - when Len =< size(Bin) -> - recv1(Len, Bin); - -%% Or not: wait for more packets. -rcv(<<_:1/binary, Len:24, _/binary>> = Head) -> - {Len, size(Head), Head, []}; +acc1(Len, Bin) -> + [Len | Bin]. + +%% acc/1 + +%% Don't match on Bin since this results in it being copied at the +%% next append according to the Efficiency Guide. This is also the +%% reason that the Len is extracted and maintained when accumulating +%% messages. The simplest implementation is just to accumulate a +%% binary and match <<_, Len:24, _/binary>> each time the length is +%% required, but the performance of this decays quadratically with the +%% message length, since the binary is then copied with each append of +%% additional bytes from gen_tcp. + +acc(Bin) + when 3 < byte_size(Bin) -> + {Head, _} = split_binary(Bin, 4), + [_,A,B,C] = binary_to_list(Head), + Len = (A bsl 16) bor (B bsl 8) bor C, + if Len < 20 -> + %% Message length isn't sufficient for a Diameter Header. + %% Chances are things will go south from here but if we're + %% lucky then the bytes we have extend to an intended + %% message boundary and we can recover by simply receiving + %% them. Make it so. + {Bin, <<>>}; + true -> + acc1(Len, Bin) + end; %% Not even 4 bytes yet. -rcv(Head) -> - Head. - -%% recv1/2 - -recv1(Len, Bin) -> - <<Msg:Len/binary, Rest/binary>> = Bin, - {Msg, Rest}. - -%% bin/2 - -bin(Head, Acc) -> - list_to_binary([Head | lists:reverse(Acc)]). +acc(Bin) -> + Bin. %% bin/1 -bin({_, _, Head, Acc}) -> - bin(Head, Acc); +bin([_ | Bin]) -> + Bin; -bin(Bin) - when is_binary(Bin) -> +bin(Bin) -> Bin. %% flush/1 @@ -911,14 +906,20 @@ setopts(#transport{socket = Sock, module = M} = S) when B, not A -> - case setopts(M, Sock, [{active, once}]) of - ok -> S#transport{active = true}; - X -> x({setopts, Sock, M, X}) %% possibly on peer disconnect - end; + setopts(M, Sock), + S#transport{active = true}; setopts(S) -> S. +%% setopts/2 + +setopts(M, Sock) -> + case setopts(M, Sock, [{active, once}]) of + ok -> ok; + X -> x({setopts, Sock, M, X}) %% possibly on peer disconnect + end. + %% portnr/2 portnr(gen_tcp, Sock) -> @@ -988,7 +989,7 @@ message(ack, _, #transport{message_cb = false} = S) -> S; message(Dir, Msg, #transport{message_cb = CB} = S) -> - recv(<<>>, actions(cb(CB, Dir, Msg), Dir, S)). + setopts(actions(cb(CB, Dir, Msg), Dir, S)). %% actions/3 |