Bind listener ports early and start accepting connections later

(EJAB-1334)

It may happen that auth or rdbms client tcp connections bind a local
socket to a port number required by a configered listener. The ejabberd
applications fails to start up and needs to be restarted.

In plain C you would bind(2) the listener port and listen(2) later on.
gen_tcp:listen/2 does not allow to separate these two steps though, so
another way is not to accept connections while start up. OTOH, the
kernel will syn/ack incoming connections and receive data, leaving them
in a buffer for the ejabberd to read from. If this is unwanted, a load
balancer would need to receive data from the ejabberd server before
adding the node to its pool.

This patch binds tcp ports while initializing the ejabberd_listener
process, storing ListenSockets in an ets table. start_listeners/0 will
reuse these ports later on.

src/ejabberd_listener.erl

@@ -49,8 +49,31 @@ start_link() ->
 
 
 init(_) ->
+    ets:new(listen_sockets, [named_table, public]),
+    bind_tcp_ports(),
     {ok, {{one_for_one, 10, 1}, []}}.
 
+bind_tcp_ports() ->
+    case ejabberd_config:get_local_option(listen) of
+	undefined ->
+	    ignore;
+	Ls ->
+	    lists:foreach(
+	      fun({Port, Module, Opts}) ->
+		      bind_tcp_port(Port, Module, Opts)
+	      end, Ls)
+    end.
+
+bind_tcp_port(PortIP, Module, RawOpts) ->
+    {Port, IPT, IPS, IPV, Proto, OptsClean} = parse_listener_portip(PortIP, RawOpts),
+    {_Opts, SockOpts} = prepare_opts(IPT, IPV, OptsClean),
+    case Proto of
+	udp -> ok;
+	_ ->
+	    ListenSocket = listen_tcp(PortIP, Module, SockOpts, Port, IPS),
+	    ets:insert(listen_sockets, {PortIP, ListenSocket})
+    end.
+
 start_listeners() ->
     case ejabberd_config:get_local_option(listen) of
 	undefined ->
@@ -100,15 +123,7 @@ start_dependent(Port, Module, Opts) ->
 
 init(PortIP, Module, RawOpts) ->
     {Port, IPT, IPS, IPV, Proto, OptsClean} = parse_listener_portip(PortIP, RawOpts),
-    %% The first inet|inet6 and the last {ip, _} work,
+    {Opts, SockOpts} = prepare_opts(IPT, IPV, OptsClean),
-    %% so overriding those in Opts
-    Opts = [IPV | OptsClean] ++ [{ip, IPT}],
-    SockOpts = lists:filter(fun({ip, _}) -> true;
-			       (inet6) -> true;
-			       (inet) -> true;
-			       ({backlog, _}) -> true;
-			       (_) -> false
-			    end, Opts),
     if Proto == udp ->
 	    init_udp(PortIP, Module, Opts, SockOpts, Port, IPS);
        true ->
@@ -129,26 +144,39 @@ init_udp(PortIP, Module, Opts, SockOpts, Port, IPS) ->
     end.
 
 init_tcp(PortIP, Module, Opts, SockOpts, Port, IPS) ->
-    SockOpts2 = case erlang:system_info(otp_release) >= "R13B" of
+    ListenSocket = listen_tcp(PortIP, Module, SockOpts, Port, IPS),
-	true -> [{send_timeout_close, true} | SockOpts];
+    %% Inform my parent that this port was opened succesfully
-	false -> SockOpts
+    proc_lib:init_ack({ok, self()}),
-    end,
+    %% And now start accepting connection attempts
-    Res = gen_tcp:listen(Port, [binary,
+    accept(ListenSocket, Module, Opts).
-				{packet, 0},
+
-				{active, false},
+listen_tcp(PortIP, Module, SockOpts, Port, IPS) ->
-				{reuseaddr, true},
+    case ets:lookup(listen_sockets, PortIP) of
-				{nodelay, true},
+	[{PortIP, ListenSocket}] ->
-				{send_timeout, ?TCP_SEND_TIMEOUT},
+	    ?INFO_MSG("Reusing listening port for ~p", [Port]),
-				{keepalive, true} |
+	    ets:delete(listen_sockets, Port),
-				SockOpts2]),
+	    ListenSocket;
-    case Res of
+	_ ->
-	{ok, ListenSocket} ->
+	    SockOpts2 = try erlang:system_info(otp_release) >= "R13B" of
-	    %% Inform my parent that this port was opened succesfully
+			    true -> [{send_timeout_close, true} | SockOpts];
-	    proc_lib:init_ack({ok, self()}),
+			    false -> SockOpts
-	    %% And now start accepting connection attempts
+			catch
-	    accept(ListenSocket, Module, Opts);
+			    _:_ -> []
-	{error, Reason} ->
+			end,
-	    socket_error(Reason, PortIP, Module, SockOpts, Port, IPS)
+	    Res = gen_tcp:listen(Port, [binary,
+					{packet, 0},
+					{active, false},
+					{reuseaddr, true},
+					{nodelay, true},
+					{send_timeout, ?TCP_SEND_TIMEOUT},
+					{keepalive, true} |
+					SockOpts2]),
+	    case Res of
+		{ok, ListenSocket} ->
+		    ListenSocket;
+		{error, Reason} ->
+		    socket_error(Reason, PortIP, Module, SockOpts, Port, IPS)
+	    end
     end.
 
 %% @spec (PortIP, Opts) -> {Port, IPT, IPS, IPV, OptsClean}
@@ -193,6 +221,18 @@ parse_listener_portip(PortIP, Opts) ->
 	  end,
     {Port, IPT, IPS, IPV, Proto, OptsClean}.
 
+prepare_opts(IPT, IPV, OptsClean) ->
+    %% The first inet|inet6 and the last {ip, _} work,
+    %% so overriding those in Opts
+    Opts = [IPV | OptsClean] ++ [{ip, IPT}],
+    SockOpts = lists:filter(fun({ip, _}) -> true;
+			       (inet6) -> true;
+			       (inet) -> true;
+			       ({backlog, _}) -> true;
+			       (_) -> false
+			    end, Opts),
+    {Opts, SockOpts}.
+
 add_proto(Port, Opts) when is_integer(Port) ->
     {Port, get_proto(Opts)};
 add_proto({Port, Proto}, _Opts) when is_atom(Proto) ->