add {active,N} socket option for TCP, UDP, and SCTP

Add the {active,N} socket option, where N is an integer in the range
-32768..32767, to allow a caller to specify the number of data messages to
be delivered to the controlling process. Once the socket's delivered
message count either reaches 0 or is explicitly set to 0 with
inet:setopts/2 or by including {active,0} as an option when the socket is
created, the socket transitions to passive ({active, false}) mode and the
socket's controlling process receives a message to inform it of the
transition. TCP sockets receive {tcp_passive,Socket}, UDP sockets receive
{udp_passive,Socket} and SCTP sockets receive {sctp_passive,Socket}.

The socket's delivered message counter defaults to 0, but it can be set
using {active,N} via any gen_tcp, gen_udp, or gen_sctp function that takes
socket options as arguments, or via inet:setopts/2. New N values are added
to the socket's current counter value, and negative numbers can be used to
reduce the counter value. Specifying a number that would cause the socket's
counter value to go above 32767 causes an einval error. If a negative
number is specified such that the counter value would become negative, the
socket's counter value is set to 0 and the socket transitions to passive
mode. If the counter value is already 0 and inet:setopts(Socket,
[{active,0}]) is specified, the counter value remains at 0 but the
appropriate passive mode transition message is generated for the socket.

This commit contains a modified preloaded prim_inet.beam due to changes in
prim_inet.erl.

Add tests for {active,N} mode for TCP, UDP, and SCTP sockets.

Add documentation for {active,N} mode for inet, gen_tcp, gen_udp, and
gen_sctp.

3 files changed, 336 insertions, 9 deletions

diff --git a/lib/kernel/test/gen_sctp_SUITE.erl b/lib/kernel/test/gen_sctp_SUITE.erl
index e89cb44797..6dad52395b 100644
--- a/lib/kernel/test/gen_sctp_SUITE.erl
+++ b/lib/kernel/test/gen_sctp_SUITE.erl
@@ -35,8 +35,9 @@
     open_unihoming_ipv6_socket/1,
     open_multihoming_ipv6_socket/1,
     open_multihoming_ipv4_and_ipv6_socket/1,
-    basic_stream/1, xfer_stream_min/1, peeloff_active_once/1,
-    peeloff_active_true/1, buffers/1]).
+    basic_stream/1, xfer_stream_min/1, active_n/1,
+    peeloff_active_once/1, peeloff_active_true/1, peeloff_active_n/1,
+    buffers/1]).
 
 suite() -> [{ct_hooks,[ts_install_cth]}].
 
@@ -46,9 +47,9 @@ all() ->
      open_multihoming_ipv4_socket,
      open_unihoming_ipv6_socket,
      open_multihoming_ipv6_socket,
-     open_multihoming_ipv4_and_ipv6_socket,
+     open_multihoming_ipv4_and_ipv6_socket, active_n,
      basic_stream, xfer_stream_min, peeloff_active_once,
-     peeloff_active_true, buffers].
+     peeloff_active_true, peeloff_active_n, buffers].
 
 groups() -> 
     [].
@@ -767,6 +768,106 @@ implicit_inet6(S1, Addr) ->
 	  end,
     ?line ok = gen_sctp:close(S2).
 
+active_n(doc) ->
+    "Verify {active,N} socket management";
+active_n(suite) ->
+    [];
+active_n(Config) when is_list(Config) ->
+    N = 3,
+    S1 = ok(gen_sctp:open([{active,N}])),
+    [{active,N}] = ok(inet:getopts(S1, [active])),
+    ok = inet:setopts(S1, [{active,-N}]),
+    receive
+        {sctp_passive, S1} -> ok
+    after
+        5000 ->
+            exit({error,sctp_passive_failure})
+    end,
+    [{active,false}] = ok(inet:getopts(S1, [active])),
+    ok = inet:setopts(S1, [{active,0}]),
+    receive
+        {sctp_passive, S1} -> ok
+    after
+        5000 ->
+            exit({error,sctp_passive_failure})
+    end,
+    ok = inet:setopts(S1, [{active,32767}]),
+    {error,einval} = inet:setopts(S1, [{active,1}]),
+    {error,einval} = inet:setopts(S1, [{active,-32769}]),
+    ok = inet:setopts(S1, [{active,-32768}]),
+    receive
+        {sctp_passive, S1} -> ok
+    after
+        5000 ->
+            exit({error,sctp_passive_failure})
+    end,
+    [{active,false}] = ok(inet:getopts(S1, [active])),
+    ok = inet:setopts(S1, [{active,N}]),
+    ok = inet:setopts(S1, [{active,true}]),
+    [{active,true}] = ok(inet:getopts(S1, [active])),
+    receive
+        _ -> exit({error,active_n})
+    after
+        0 ->
+            ok
+    end,
+    ok = inet:setopts(S1, [{active,N}]),
+    ok = inet:setopts(S1, [{active,once}]),
+    [{active,once}] = ok(inet:getopts(S1, [active])),
+    receive
+        _ -> exit({error,active_n})
+    after
+        0 ->
+            ok
+    end,
+    {error,einval} = inet:setopts(S1, [{active,32768}]),
+    ok = inet:setopts(S1, [{active,false}]),
+    [{active,false}] = ok(inet:getopts(S1, [active])),
+    ok = gen_sctp:listen(S1, true),
+    S1Port = ok(inet:port(S1)),
+    S2 = ok(gen_sctp:open(0, [{active,false}])),
+    Assoc = ok(gen_sctp:connect(S2, "localhost", S1Port, [])),
+    ok = inet:setopts(S1, [{active,N}]),
+    [{active,N}] = ok(inet:getopts(S1, [active])),
+    LoopFun = fun(Count, Count, _Fn) ->
+		      receive
+			  {sctp_passive,S1} ->
+			      ok
+		      after
+			  5000 ->
+			      exit({error,timeout})
+		      end;
+		 (I, Count, Fn) ->
+		      Msg = list_to_binary("message "++integer_to_list(I)),
+		      ok = gen_sctp:send(S2, Assoc, 0, Msg),
+		      receive
+			  {sctp,S1,_,_,{[SR],Msg}} when is_record(SR, sctp_sndrcvinfo) ->
+			      Fn(I+1, Count, Fn);
+			  {sctp,S1,_,_,_} ->
+			      %% ignore non-data messages
+			      ok = inet:setopts(S1, [{active,1}]),
+			      Fn(I, Count, Fn);
+			  Other ->
+			      exit({unexpected, Other})
+		      after
+			  5000 ->
+			      exit({error,timeout})
+		      end
+	      end,
+    ok = LoopFun(1, N, LoopFun),
+    S3 = ok(gen_sctp:open([{active,0}])),
+    receive
+        {sctp_passive,S3} ->
+            [{active,false}] = ok(inet:getopts(S3, [active]))
+    after
+        5000 ->
+            exit({error,udp_passive})
+    end,
+    ok = gen_sctp:close(S3),
+    ok = gen_sctp:close(S2),
+    ok = gen_sctp:close(S1),
+    ok.
+
 basic_stream(doc) ->
     "Hello world stream socket";
 basic_stream(suite) ->
@@ -941,6 +1042,14 @@ peeloff_active_true(suite) ->
 peeloff_active_true(Config) ->
     peeloff(Config, [{active,true}]).
 
+peeloff_active_n(doc) ->
+    "Peel off an SCTP stream socket ({active,N})";
+peeloff_active_n(suite) ->
+    [];
+
+peeloff_active_n(Config) ->
+    peeloff(Config, [{active,1}]).
+
 peeloff(Config, SockOpts) when is_list(Config) ->
     ?line Addr = {127,0,0,1},
     ?line Stream = 0,
@@ -1519,7 +1628,13 @@ s_loop(Socket, Timeout, Parent, Handler, State) ->
     end.
 
 again(Socket) ->
-    inet:setopts(Socket, [{active,once}]).
+    receive
+	{sctp_passive,Socket} ->
+	    [{active, false}] = ok(inet:getopts(Socket, [active])),
+	    ok = inet:setopts(Socket,[{active,1}])
+    after 0 ->
+	    ok = inet:setopts(Socket, [{active,once}])
+    end.
 
 gb_push(Key, Val, GBT) ->
     case gb_trees:lookup(Key, GBT) of
diff --git a/lib/kernel/test/gen_tcp_misc_SUITE.erl b/lib/kernel/test/gen_tcp_misc_SUITE.erl
index ee271fbdfa..ee8bfcceb1 100644
--- a/lib/kernel/test/gen_tcp_misc_SUITE.erl
+++ b/lib/kernel/test/gen_tcp_misc_SUITE.erl
@@ -25,7 +25,7 @@
 -export([all/0, suite/0,groups/0,init_per_suite/1, end_per_suite/1, 
 	 init_per_group/2,end_per_group/2, 
 	 controlling_process/1, controlling_process_self/1,
-	 no_accept/1, close_with_pending_output/1,
+	 no_accept/1, close_with_pending_output/1, active_n/1,
 	 data_before_close/1, iter_max_socks/1, get_status/1,
 	 passive_sockets/1, accept_closed_by_other_process/1,
 	 init_per_testcase/2, end_per_testcase/2,
@@ -70,7 +70,7 @@ suite() -> [{ct_hooks,[ts_install_cth]}].
 all() -> 
     [controlling_process, controlling_process_self, no_accept,
      close_with_pending_output, data_before_close,
-     iter_max_socks, passive_sockets,
+     iter_max_socks, passive_sockets, active_n,
      accept_closed_by_other_process, otp_3924, closed_socket,
      shutdown_active, shutdown_passive, shutdown_pending,
      default_options, http_bad_packet, busy_send,
@@ -407,6 +407,114 @@ send_loop(Sock, Data, Left) ->
     ok = gen_tcp:send(Sock, Data),
     send_loop(Sock, Data, Left-1).
 
+%% Test {active,N} option
+active_n(doc) ->
+    ["Verify operation of the {active,N} option."];
+active_n(suite) -> [];
+active_n(Config) when is_list(Config) ->
+    N = 3,
+    LS = ok(gen_tcp:listen(0, [{active,N}])),
+    [{active,N}] = ok(inet:getopts(LS, [active])),
+    ok = inet:setopts(LS, [{active,-N}]),
+    receive
+        {tcp_passive, LS} -> ok
+    after
+        5000 ->
+            exit({error,tcp_passive_failure})
+    end,
+    [{active,false}] = ok(inet:getopts(LS, [active])),
+    ok = inet:setopts(LS, [{active,0}]),
+    receive
+        {tcp_passive, LS} -> ok
+    after
+        5000 ->
+            exit({error,tcp_passive_failure})
+    end,
+    ok = inet:setopts(LS, [{active,32767}]),
+    {error,einval} = inet:setopts(LS, [{active,1}]),
+    {error,einval} = inet:setopts(LS, [{active,-32769}]),
+    ok = inet:setopts(LS, [{active,-32768}]),
+    receive
+        {tcp_passive, LS} -> ok
+    after
+        5000 ->
+            exit({error,tcp_passive_failure})
+    end,
+    [{active,false}] = ok(inet:getopts(LS, [active])),
+    ok = inet:setopts(LS, [{active,N}]),
+    ok = inet:setopts(LS, [{active,true}]),
+    [{active,true}] = ok(inet:getopts(LS, [active])),
+    receive
+        _ -> exit({error,active_n})
+    after
+        0 ->
+            ok
+    end,
+    ok = inet:setopts(LS, [{active,N}]),
+    ok = inet:setopts(LS, [{active,once}]),
+    [{active,once}] = ok(inet:getopts(LS, [active])),
+    receive
+        _ -> exit({error,active_n})
+    after
+        0 ->
+            ok
+    end,
+    {error,einval} = inet:setopts(LS, [{active,32768}]),
+    ok = inet:setopts(LS, [{active,false}]),
+    [{active,false}] = ok(inet:getopts(LS, [active])),
+    Port = ok(inet:port(LS)),
+    C = ok(gen_tcp:connect("localhost", Port, [{active,N}])),
+    [{active,N}] = ok(inet:getopts(C, [active])),
+    S = ok(gen_tcp:accept(LS)),
+    ok = inet:setopts(S, [{active,N}]),
+    [{active,N}] = ok(inet:getopts(S, [active])),
+    repeat(3,
+           fun(I) ->
+                   Msg = "message "++integer_to_list(I),
+                   ok = gen_tcp:send(C, Msg),
+                   receive
+                       {tcp,S,Msg} ->
+                           ok = gen_tcp:send(S, Msg)
+                   after
+                       5000 ->
+                           exit({error,timeout})
+                   end,
+                   receive
+                       {tcp,C,Msg} ->
+                           ok
+                   after
+                       5000 ->
+                           exit({error,timeout})
+                   end
+           end),
+    receive
+        {tcp_passive,S} ->
+            [{active,false}] = ok(inet:getopts(S, [active]))
+    after
+        5000 ->
+            exit({error,tcp_passive})
+    end,
+    receive
+        {tcp_passive,C} ->
+            [{active,false}] = ok(inet:getopts(C, [active]))
+    after
+        5000 ->
+            exit({error,tcp_passive})
+    end,
+    LS2 = ok(gen_tcp:listen(0, [{active,0}])),
+    receive
+        {tcp_passive,LS2} ->
+            [{active,false}] = ok(inet:getopts(LS2, [active]))
+    after
+        5000 ->
+            exit({error,tcp_passive})
+    end,
+    ok = gen_tcp:close(LS2),
+    ok = gen_tcp:close(C),
+    ok = gen_tcp:close(S),
+    ok = gen_tcp:close(LS),
+    ok.
+
 -define(OTP_3924_MAX_DELAY, 100).
 %% Taken out of the blue, but on intra host connections
 %% I expect propagation of a close to be quite fast
@@ -2659,3 +2767,5 @@ oct_aloop(S,X,Times) ->
 	    gen_tcp:close(S),
 	    closed
     end.
+
+ok({ok,V}) -> V.
diff --git a/lib/kernel/test/gen_udp_SUITE.erl b/lib/kernel/test/gen_udp_SUITE.erl
index 4cc6018614..6bb41999c5 100644
--- a/lib/kernel/test/gen_udp_SUITE.erl
+++ b/lib/kernel/test/gen_udp_SUITE.erl
@@ -33,7 +33,7 @@
 	 init_per_group/2,end_per_group/2]).
 -export([init_per_testcase/2, end_per_testcase/2]).
 
--export([send_to_closed/1, 
+-export([send_to_closed/1, active_n/1,
 	 buffer_size/1, binary_passive_recv/1, bad_address/1,
 	 read_packets/1, open_fd/1, connect/1, implicit_inet6/1]).
 
@@ -42,7 +42,7 @@ suite() -> [{ct_hooks,[ts_install_cth]}].
 all() -> 
     [send_to_closed, buffer_size, binary_passive_recv,
      bad_address, read_packets, open_fd, connect,
-     implicit_inet6].
+     implicit_inet6, active_n].
 
 groups() -> 
     [].
@@ -465,6 +465,108 @@ open_fd(Config) when is_list(Config) ->
 	    ?t:fail(io_lib:format("~w", [flush()]))
     end.
 
+active_n(Config) when is_list(Config) ->
+    N = 3,
+    S1 = ok(gen_udp:open(0, [{active,N}])),
+    [{active,N}] = ok(inet:getopts(S1, [active])),
+    ok = inet:setopts(S1, [{active,-N}]),
+    receive
+        {udp_passive, S1} -> ok
+    after
+        5000 ->
+            exit({error,udp_passive_failure})
+    end,
+    [{active,false}] = ok(inet:getopts(S1, [active])),
+    ok = inet:setopts(S1, [{active,0}]),
+    receive
+        {udp_passive, S1} -> ok
+    after
+        5000 ->
+            exit({error,udp_passive_failure})
+    end,
+    ok = inet:setopts(S1, [{active,32767}]),
+    {error,einval} = inet:setopts(S1, [{active,1}]),
+    {error,einval} = inet:setopts(S1, [{active,-32769}]),
+    ok = inet:setopts(S1, [{active,-32768}]),
+    receive
+        {udp_passive, S1} -> ok
+    after
+        5000 ->
+            exit({error,udp_passive_failure})
+    end,
+    [{active,false}] = ok(inet:getopts(S1, [active])),
+    ok = inet:setopts(S1, [{active,N}]),
+    ok = inet:setopts(S1, [{active,true}]),
+    [{active,true}] = ok(inet:getopts(S1, [active])),
+    receive
+        _ -> exit({error,active_n})
+    after
+        0 ->
+            ok
+    end,
+    ok = inet:setopts(S1, [{active,N}]),
+    ok = inet:setopts(S1, [{active,once}]),
+    [{active,once}] = ok(inet:getopts(S1, [active])),
+    receive
+        _ -> exit({error,active_n})
+    after
+        0 ->
+            ok
+    end,
+    {error,einval} = inet:setopts(S1, [{active,32768}]),
+    ok = inet:setopts(S1, [{active,false}]),
+    [{active,false}] = ok(inet:getopts(S1, [active])),
+    S1Port = ok(inet:port(S1)),
+    S2 = ok(gen_udp:open(0, [{active,N}])),
+    S2Port = ok(inet:port(S2)),
+    [{active,N}] = ok(inet:getopts(S2, [active])),
+    ok = inet:setopts(S1, [{active,N}]),
+    [{active,N}] = ok(inet:getopts(S1, [active])),
+    lists:foreach(
+      fun(I) ->
+              Msg = "message "++integer_to_list(I),
+              ok = gen_udp:send(S2, "localhost", S1Port, Msg),
+              receive
+                  {udp,S1,_,S2Port,Msg} ->
+                      ok = gen_udp:send(S1, "localhost", S2Port, Msg)
+              after
+                  5000 ->
+                      exit({error,timeout})
+              end,
+              receive
+                  {udp,S2,_,S1Port,Msg} ->
+                      ok
+              after
+                  5000 ->
+                      exit({error,timeout})
+              end
+      end, lists:seq(1,N)),
+    receive
+        {udp_passive,S1} ->
+            [{active,false}] = ok(inet:getopts(S1, [active]))
+    after
+        5000 ->
+            exit({error,udp_passive})
+    end,
+    receive
+        {udp_passive,S2} ->
+            [{active,false}] = ok(inet:getopts(S2, [active]))
+    after
+        5000 ->
+            exit({error,udp_passive})
+    end,
+    S3 = ok(gen_udp:open(0, [{active,0}])),
+    receive
+        {udp_passive,S3} ->
+            [{active,false}] = ok(inet:getopts(S3, [active]))
+    after
+        5000 ->
+            exit({error,udp_passive})
+    end,
+    ok = gen_udp:close(S3),
+    ok = gen_udp:close(S2),
+    ok = gen_udp:close(S1),
+    ok.
 
 %
 % Utils