%%
%% %CopyrightBegin%
%%
%% Copyright Ericsson AB 2017. 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(ssl_tls_dist_ctrl).
-include_lib("kernel/include/dist.hrl").
-include_lib("kernel/include/dist_util.hrl").
-include_lib("kernel/include/net_address.hrl").
-export([start/1, get_socket/1, set_supervisor/2, hs_data_common/1]).
%%% ------------------------------------------------------------
%% In order to avoid issues with lingering signal binaries
%% we enable off-heap message queue data as well as fullsweep
%% after 0. The fullsweeps will be cheap since we have more
%% or less no live data.
common_spawn_opts() ->
[{message_queue_data, off_heap},
{fullsweep_after, 0}].
%%% ------------------------------------------------------------
start(SslSocket) ->
spawn_opt(
fun () ->
setup(SslSocket)
end,
[{priority, max}] ++ common_spawn_opts()).
get_socket(DistCtrl) ->
call(DistCtrl, get_socket).
set_supervisor(DistCtrl, Pid) ->
call(DistCtrl, {set_supervisor, Pid}).
hs_data_common(DistCtrl) ->
TickHandler = call(DistCtrl, tick_handler),
SslSocket = get_socket(DistCtrl),
#hs_data{
f_send =
fun (Ctrl, Packet) when Ctrl == DistCtrl ->
call(Ctrl, {send, Packet})
end,
f_recv =
fun (Ctrl, Length, Timeout) when Ctrl == DistCtrl ->
case call(Ctrl, {recv, Length, Timeout}) of
{ok, Bin} when is_binary(Bin) ->
{ok, binary_to_list(Bin)};
Other ->
Other
end
end,
f_setopts_pre_nodeup =
fun (Ctrl) when Ctrl == DistCtrl ->
call(Ctrl, pre_nodeup)
end,
f_setopts_post_nodeup =
fun (Ctrl) when Ctrl == DistCtrl ->
call(Ctrl, post_nodeup)
end,
f_getll =
fun (Ctrl) when Ctrl == DistCtrl ->
call(Ctrl, getll)
end,
f_handshake_complete =
fun (Ctrl, Node, DHandle) when Ctrl == DistCtrl ->
call(Ctrl, {handshake_complete, Node, DHandle})
end,
f_address =
fun (Ctrl, Node) when Ctrl == DistCtrl ->
case call(Ctrl, {check_address, Node}) of
{error, no_node} ->
%% No '@' or more than one '@' in node name.
?shutdown(no_node);
Res ->
Res
end
end,
mf_setopts =
fun (Ctrl, Opts) when Ctrl == DistCtrl ->
case setopts_filter(Opts) of
[] ->
ssl:setopts(SslSocket, Opts);
Opts1 ->
{error, {badopts,Opts1}}
end
end,
mf_getopts =
fun (Ctrl, Opts) when Ctrl == DistCtrl ->
ssl:getopts(SslSocket, Opts)
end,
mf_getstat =
fun (Ctrl) when Ctrl == DistCtrl ->
case ssl:getstat(
SslSocket, [recv_cnt, send_cnt, send_pend]) of
{ok, Stat} ->
split_stat(Stat,0,0,0);
Error ->
Error
end
end,
mf_tick =
fun (Ctrl) when Ctrl == DistCtrl ->
TickHandler ! tick
end}.
%%% ------------------------------------------------------------
call(DistCtrl, Msg) ->
Ref = erlang:monitor(process, DistCtrl),
DistCtrl ! {Ref, self(), Msg},
receive
{Ref, Res} ->
erlang:demonitor(Ref, [flush]),
Res;
{'DOWN', Ref, process, DistCtrl, Reason} ->
exit({dist_controller_exit, Reason})
end.
setopts_filter(Opts) ->
[Opt || {K,_} = Opt <- Opts,
K =:= active orelse K =:= deliver orelse K =:= packet].
split_stat([{recv_cnt, R}|Stat], _, W, P) ->
split_stat(Stat, R, W, P);
split_stat([{send_cnt, W}|Stat], R, _, P) ->
split_stat(Stat, R, W, P);
split_stat([{send_pend, P}|Stat], R, W, _) ->
split_stat(Stat, R, W, P);
split_stat([], R, W, P) ->
{ok, R, W, P}.
%%% ------------------------------------------------------------
%%% Distribution controller processes
%%% ------------------------------------------------------------
%%
%% There will be five parties working together when the
%% connection is up:
%% - The SSL socket. Providing a TLS connection
%% to the other node.
%% - The output handler. It will dispatch all outgoing
%% traffic from the VM to the gen_tcp socket. This
%% process is registered as distribution controller
%% for this channel with the VM.
%% - The input handler. It will dispatch all incoming
%% traffic from the gen_tcp socket to the VM. This
%% process is also the socket owner and receives
%% incoming traffic using active-N.
%% - The tick handler. Dispatches asynchronous tick
%% requests to the socket. It executes on max priority
%% since it is important to get ticks through to the
%% other end.
%% - The channel supervisor (provided by dist_util). It
%% monitors traffic. Issue tick requests to the tick
%% handler when no outgoing traffic is seen and bring
%% the connection down if no incoming traffic is seen.
%% This process also executes on max priority.
%%
%% These parties are linked togheter so should one
%% of them fail, all of them are terminated and the
%% connection is taken down.
%%
%%
%% The tick handler process writes a tick to the
%% socket when it receives a 'tick' message from
%% the connection supervisor.
%%
%% We are not allowed to block the connection
%% superviser when writing a tick and we also want
%% the tick to go through even during a heavily
%% loaded system. gen_tcp does not have a
%% non-blocking send operation exposed in its API
%% and we don't want to run the distribution
%% controller under high priority. Therefore this
%% separate process with max prio that dispatches
%% ticks.
%%
tick_handler(SslSocket) ->
receive
tick ->
%% May block due to busy port...
sock_send(SslSocket, ""),
flush_ticks(SslSocket);
_ ->
tick_handler(SslSocket)
end.
flush_ticks(SslSocket) ->
receive
tick ->
flush_ticks(SslSocket)
after 0 ->
tick_handler(SslSocket)
end.
setup(SslSocket) ->
TickHandler =
spawn_opt(
fun () ->
tick_handler(SslSocket)
end,
[link, {priority, max}] ++ common_spawn_opts()),
setup_loop(SslSocket, TickHandler, undefined).
%%
%% During the handshake phase we loop in setup().
%% When the connection is up we spawn an input handler and
%% continue as output handler.
%%
setup_loop(SslSocket, TickHandler, Sup) ->
receive
{ssl_closed, SslSocket} ->
exit(connection_closed);
{ssl_error, SslSocket} ->
exit(connection_closed);
{Ref, From, {set_supervisor, Pid}} ->
Res = link(Pid),
From ! {Ref, Res},
setup_loop(SslSocket, TickHandler, Pid);
{Ref, From, tick_handler} ->
From ! {Ref, TickHandler},
setup_loop(SslSocket, TickHandler, Sup);
{Ref, From, get_socket} ->
From ! {Ref, SslSocket},
setup_loop(SslSocket, TickHandler, Sup);
{Ref, From, {send, Packet}} ->
Res = ssl:send(SslSocket, Packet),
From ! {Ref, Res},
setup_loop(SslSocket, TickHandler, Sup);
{Ref, From, {recv, Length, Timeout}} ->
Res = ssl:recv(SslSocket, Length, Timeout),
From ! {Ref, Res},
setup_loop(SslSocket, TickHandler, Sup);
{Ref, From, {check_address, Node}} ->
Res =
case ssl:peername(SslSocket) of
{ok, Address} ->
case inet_tls_dist:split_node(Node) of
false ->
{error, no_node};
Host ->
#net_address{
address=Address, host=Host,
protocol=tls, family=inet}
end
end,
From ! {Ref, Res},
setup_loop(SslSocket, TickHandler, Sup);
{Ref, From, pre_nodeup} ->
Res =
ssl:setopts(
SslSocket,
[{packet, 4}, inet_tls_dist:nodelay()]),
From ! {Ref, Res},
setup_loop(SslSocket, TickHandler, Sup);
{Ref, From, post_nodeup} ->
Res =
ssl:setopts(
SslSocket,
[{packet, 4}, inet_tls_dist:nodelay()]),
From ! {Ref, Res},
setup_loop(SslSocket, TickHandler, Sup);
{Ref, From, getll} ->
From ! {Ref, {ok, self()}},
setup_loop(SslSocket, TickHandler, Sup);
{Ref, From, {handshake_complete, _Node, DHandle}} ->
From ! {Ref, ok},
%% Handshake complete! Begin dispatching traffic...
%%
%% Use a dedicated input process to push the
%% input-output-flow-control-deadlock problem
%% to the SSL implementation.
InputHandler =
spawn_opt(
fun () ->
link(Sup),
ssl:setopts(SslSocket, [{active, once}]),
receive
DHandle ->
input_loop(DHandle, SslSocket)
end
end,
[link] ++ common_spawn_opts()),
ok = ssl:controlling_process(SslSocket, InputHandler),
ok = erlang:dist_ctrl_input_handler(DHandle, InputHandler),
InputHandler ! DHandle,
%%
%% From now on we execute on normal priority
process_flag(priority, normal),
erlang:dist_ctrl_get_data_notification(DHandle),
output_loop(DHandle, SslSocket)
end.
input_loop(DHandle, SslSocket) ->
receive
{ssl_closed, SslSocket} ->
%% Connection to remote node terminated...
exit(connection_closed);
{ssl_error, SslSocket, _Reason} ->
%% Connection to remote node terminated...
exit(connection_closed);
{ssl, SslSocket, Data} ->
%% Incoming data from remote node...
ok = ssl:setopts(SslSocket, [{active, once}]),
try erlang:dist_ctrl_put_data(DHandle, Data)
catch _:_ -> death_row()
end,
input_loop(DHandle, SslSocket);
_ ->
%% Drop garbage message...
input_loop(DHandle, SslSocket)
end.
output_loop(DHandle, SslSocket) ->
receive
dist_data ->
%% Outgoing data from this node...
try send_data(DHandle, SslSocket)
catch _ : _ -> death_row()
end,
output_loop(DHandle, SslSocket);
{send, From, Ref, Data} ->
%% This is for testing only!
%%
%% Needed by some OTP distribution
%% test suites...
sock_send(SslSocket, Data),
From ! {Ref, ok},
output_loop(DHandle, SslSocket);
_ ->
%% Drop garbage message...
output_loop(DHandle, SslSocket)
end.
send_data(DHandle, SslSocket) ->
case erlang:dist_ctrl_get_data(DHandle) of
none ->
erlang:dist_ctrl_get_data_notification(DHandle);
Data ->
sock_send(SslSocket, Data),
send_data(DHandle, SslSocket)
end.
sock_send(SslSocket, Data) ->
try ssl:send(SslSocket, Data) of
ok -> ok;
{error, Reason} ->
death_row({send_error, Reason})
catch
Type:Reason ->
death_row({send_error, {Type, Reason}})
end.
death_row() ->
death_row(connection_closed).
death_row(normal) ->
%% We do not want to exit with normal
%% exit reason since it wont bring down
%% linked processes...
death_row();
death_row(Reason) ->
%% When the connection is on its way down operations
%% begin to fail. We catch the failures and call
%% this function waiting for termination. We should
%% be terminated by one of our links to the other
%% involved parties that began bringing the
%% connection down. By waiting for termination we
%% avoid altering the exit reason for the connection
%% teardown. We however limit the wait to 5 seconds
%% and bring down the connection ourselves if not
%% terminated...
receive after 5000 -> exit(Reason) end.
