Add diameter_dist for ready spawn_opt callbacks

That is, of functions that can be configured as spawn_opt MFAs in
transport configuration.

This commits adds the spawn_local described in the parent commit, and a
route_session that assumes that the local node initiates all sessions
with Session-Id returned by diameter:session_id/1, and handles incoming
requests on the node on which the id in question was returned,
diameter:session_id/1 using node() as optional value in the Session-Id
format.

5 files changed, 227 insertions, 4 deletions

diff --git a/lib/diameter/src/base/diameter_dist.erl b/lib/diameter/src/base/diameter_dist.erl
new file mode 100644
index 0000000000..cef9522c9d
--- /dev/null
+++ b/lib/diameter/src/base/diameter_dist.erl
@@ -0,0 +1,218 @@
+%%
+%% %CopyrightBegin%
+%%
+%% Copyright Ericsson AB 2019. 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.
+%%
+%% %CopyrightEnd%
+%%
+
+-module(diameter_dist).
+
+-behaviour(gen_server).
+
+%%
+%% Implements callbacks that can be configured as a spawn_opt
+%% transport configuration, to be able to distribute incoming Diameter
+%% requests to handler processes (local or remote) in various ways.
+%%
+
+%% spawn_opt callbacks; initial argument constructed in diameter_traffic
+-export([spawn_local/2,
+         spawn_local/1,
+         route_session/2,
+         route_session/1]).
+
+-include_lib("diameter/include/diameter.hrl").
+
+%% server start
+-export([start_link/0]).
+
+%% gen_server callbacks
+-export([init/1,
+         handle_info/2,
+         handle_cast/2,
+         handle_call/3,
+         code_change/3,
+         terminate/2]).
+
+-define(SERVER, ?MODULE).  %% server monitoring node connections
+-define(TABLE, ?MODULE).   %% node() binary -> node() atom
+
+-define(B(A), atom_to_binary(A, utf8)).
+
+%% spawn_local/2
+%%
+%% Callback that is equivalent to an options list. That is, the
+%% following are equivalent when passed as options to
+%% diameter:add_transport/2.
+%%
+%%   {spawn_opt, Opts}
+%%   {spawn_opt, {diameter_dist, spawn_local, [Opts]}}
+
+spawn_local(ReqT, Opts) ->
+    spawn_opt(diameter_traffic, request, [ReqT], Opts).
+
+%% spawn_local/1
+
+spawn_local(ReqT) ->
+    spawn_local(ReqT, []).
+
+%% route_session/2
+%%
+%% Callback that routes requests containing Session-Id AVPs as
+%% returned by diameter:session_id/0 back to the node on which the
+%% function was called. This is only appropriate when sessions are
+%% initiated by the own (typically client) node, and ids have been
+%% returned from diameter:session_id/0.
+
+route_session(ReqT, Opts) ->
+    #diameter_packet{bin = Bin} = element(1, ReqT),
+    Node = node_of_session_id(Bin),
+    spawn_opt(Node, diameter_traffic, request, [ReqT], Opts).
+
+%% route_session/1
+
+route_session(ReqT) ->
+    route_session(ReqT, []).
+
+%% node_of_session_id/1
+%%
+%% Return the node name encoded as optional value in a Session-Id,
+%% assuming the id has been created with diameter:session_id/0.
+%%
+%% node() is returned if a node name can't be extracted for any
+%% reason.
+
+node_of_session_id(<<_Head:20/binary, Avps/binary>>) ->
+    sid_node(Avps);
+
+node_of_session_id(_) ->
+    node().
+
+%% sid_node/1
+
+%% Session-Id = Command Code 263, V-bit = 0.
+sid_node(<<263:32, 0:1, _:7, Len:24, _/binary>> = Bin) ->
+    case Bin of
+        <<Avp:Len/binary>> ->
+            <<_:8/binary, Sid/binary>> = Avp,
+            sid_node(Sid, pattern(), 2);  %% look for the optional value
+        _ ->
+            node()
+    end;
+
+%% Jump to the next AVP. This is potentially costly for a message with
+%% many AVPs and no Session-Id, which an attacker is prone to send.
+%% 8.8 or RFC 6733 says that Session-Id SHOULD (but not MUST) appear
+%% immediately following the Diameter Header, so there is no
+%% guarantee.
+sid_node(<<_:40, Len:24, _/binary>> = Bin) ->
+    Pad = (4 - (Len rem 4)) rem 4,
+    case Bin of
+        <<_:Len/binary, _:Pad/binary, Rest/binary>> ->
+            sid_node(Rest);
+        _ ->
+            node()
+    end.
+
+%% sid_node/2
+
+%% Lookup the node name to ensure we don't convert arbitrary binaries
+%% to atom.
+sid_node(Bin, _, 0) ->
+    case ets:lookup(?TABLE, Bin) of
+        [{_, Node}] ->
+            Node;
+        [] ->
+            node()
+    end;
+
+%% The optional value (if any) of a Session-Id follows the third
+%% semicolon. Searching with binary:match/2 does better than matching,
+%% especially when the pattern is compiled.
+sid_node(Bin, CP, N) ->
+    case binary:match(Bin, CP) of
+        {Offset, 1} ->
+            <<_:Offset/binary, _, Rest/binary>> = Bin,
+            sid_node(Rest, CP, N-1);
+        nomatch ->
+            node()
+    end.
+
+%% pattern/0
+%%
+%% Since this is being called in a watchdog process, compile the
+%% pattern once and maintain it in the process dictionary.
+
+pattern() ->
+    case get(?MODULE) of
+        undefined ->
+            CP = binary:compile_pattern(<<$;>>), %% tuple
+            put(?MODULE, CP),
+            CP;
+        CP ->
+            CP
+    end.
+
+%% ===========================================================================
+
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, _Args = [], _Opts  = []).
+
+%% init/1
+%%
+%% Maintain [node() | nodes()] in a table that maps from binary-valued
+%% names, so we can lookup the corresponding atoms rather than convert
+%% binaries that aren't necessarily node names.
+
+init([]) ->
+    ets:new(?TABLE, [set, named_table]),
+    ok = net_kernel:monitor_nodes(true, [{node_type, all}, nodedown_reason]),
+    ets:insert(?TABLE, [{B,N} || N <- [node() | nodes()],
+                                 B <- [?B(N)]]),
+    {ok, erlang:monotonic_time()}.
+
+%% handle_call/3
+
+handle_call(_, _From, S) ->
+    {reply, nok, S}.
+
+%% handle_cast/2
+
+handle_cast(_, S) ->
+    {noreply, S}.
+
+%% handle_info/2
+
+handle_info({nodeup, Node, _}, S) ->
+    ets:insert(?TABLE, {?B(Node), Node}),
+    {noreply, S};
+
+handle_info({nodedown, Node, _}, S) ->
+    ets:delete(?TABLE, ?B(Node)),
+    {noreply, S};
+
+handle_info(_, S) ->
+    {noreply, S}.
+
+%% terminate/2
+
+terminate(_, _) ->
+    ok.
+
+%% code_change/3
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
diff --git a/lib/diameter/src/base/diameter_misc_sup.erl b/lib/diameter/src/base/diameter_misc_sup.erl
index 343688be23..fec5a41b5c 100644
--- a/lib/diameter/src/base/diameter_misc_sup.erl
+++ b/lib/diameter/src/base/diameter_misc_sup.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2016. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2019. 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.
@@ -35,6 +35,7 @@
                    diameter_stats,     %% statistics counter management
                    diameter_reg,       %% service/property publishing
                    diameter_peer,      %% remote peer manager
+                   diameter_dist,      %% request distribution
                    diameter_config]).  %% configuration/restart
 
 %% start_link/0
diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl
index e9acb5c0e8..b1b797aad8 100644
--- a/lib/diameter/src/base/diameter_traffic.erl
+++ b/lib/diameter/src/base/diameter_traffic.erl
@@ -288,7 +288,9 @@ spawn_request(false, _, _, _, _, _, _) ->  %% no transport
 %% count outstanding requests. Acknowledgement is implicit if the
 %% handler process dies (in a handle_request callback for example).
 spawn_request(AppT, {M,F,A}, Ack, TPid, Pkt, Dict0, RecvData) ->
-    %% Term to pass to request/1 in an appropriate process.
+    %% Term to pass to request/1 in an appropriate process. Module
+    %% diameter_dist implements callbacks, and uses the form of the
+    %% argument tuple constructed below.
     ReqT = {Pkt, AppT, Ack, TPid, Dict0, RecvData},
     apply(M, F, [ReqT | A]);
 
diff --git a/lib/diameter/src/modules.mk b/lib/diameter/src/modules.mk
index bb86de016a..d16292bb88 100644
--- a/lib/diameter/src/modules.mk
+++ b/lib/diameter/src/modules.mk
@@ -1,7 +1,7 @@
 
 # %CopyrightBegin%
 #
-# Copyright Ericsson AB 2010-2017. All Rights Reserved.
+# Copyright Ericsson AB 2010-2019. 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.
@@ -40,6 +40,7 @@ RT_MODULES = \
 	base/diameter_config \
 	base/diameter_config_sup \
 	base/diameter_codec \
+	base/diameter_dist \
 	base/diameter_gen \
 	base/diameter_lib \
 	base/diameter_misc_sup \
diff --git a/lib/diameter/test/diameter_distribution_SUITE.erl b/lib/diameter/test/diameter_distribution_SUITE.erl
index 5146f68ff1..92d5c59797 100644
--- a/lib/diameter/test/diameter_distribution_SUITE.erl
+++ b/lib/diameter/test/diameter_distribution_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2013-2016. All Rights Reserved.
+%% Copyright Ericsson AB 2013-2019. 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.
@@ -76,6 +76,7 @@
          {share_peers, peers()},
          {use_shared_peers, peers()},
          {restrict_connections, false},
+         {spawn_opt, {diameter_dist, spawn_local, []}},
          {sequence, fun sequence/0},
          {application, [{dictionary, ?DICT},
                         {module, ?MODULE},