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.

14 files changed, 557 insertions, 53 deletions

diff --git a/erts/emulator/drivers/common/inet_drv.c b/erts/emulator/drivers/common/inet_drv.c
index 60db50e80a..7dcb191a83 100644
--- a/erts/emulator/drivers/common/inet_drv.c
+++ b/erts/emulator/drivers/common/inet_drv.c
@@ -86,6 +86,13 @@
 #endif
 typedef unsigned long long llu_t;
 
+#ifndef INT16_MIN
+#define INT16_MIN (-32768)
+#endif
+#ifndef INT16_MAX
+#define INT16_MAX (32767)
+#endif
+
 #ifdef __WIN32__
 #define  STRNCASECMP strncasecmp
 
@@ -581,6 +588,7 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n)
 #define INET_PASSIVE        0  /* false */
 #define INET_ACTIVE         1  /* true */
 #define INET_ONCE           2  /* true; active once then passive */
+#define INET_MULTI          3  /* true; active N then passive */
 
 /* INET_REQ_GETSTATUS enumeration */
 #define INET_F_OPEN         0x0001
@@ -925,6 +933,7 @@ typedef struct {
     inet_async_op  op_queue[INET_MAX_ASYNC];  /* call queue */
 
     int   active;               /* 0 = passive, 1 = active, 2 = active once */
+    Sint16 active_count;        /* counter for {active,N} */
     int   stype;                /* socket type:
 				    SOCK_STREAM/SOCK_DGRAM/SOCK_SEQPACKET   */
     int   sprotocol;            /* socket protocol:
@@ -1160,22 +1169,36 @@ static int packet_inet_output(udp_descriptor* udesc, HANDLE event);
 static int async_ref = 0;          /* async reference id generator */
 #define NEW_ASYNC_ID() ((async_ref++) & 0xffff)
 
+/* check for transition from active to passive */
+#define INET_CHECK_ACTIVE_TO_PASSIVE(inet)                              \
+    do {                                                                \
+        if ((inet)->active == INET_ONCE)                                \
+            (inet)->active = INET_PASSIVE;                              \
+        else if ((inet)->active == INET_MULTI && --((inet)->active_count) == 0) { \
+            (inet)->active = INET_PASSIVE;                              \
+            packet_passive_message(inet);                               \
+        }                                                               \
+    } while (0)
 
 static ErlDrvTermData am_ok;
 static ErlDrvTermData am_tcp;
 static ErlDrvTermData am_udp;
 static ErlDrvTermData am_error;
+static ErlDrvTermData am_einval;
 static ErlDrvTermData am_inet_async;
 static ErlDrvTermData am_inet_reply;
 static ErlDrvTermData am_timeout;
 static ErlDrvTermData am_closed;
+static ErlDrvTermData am_tcp_passive;
 static ErlDrvTermData am_tcp_closed;
 static ErlDrvTermData am_tcp_error;
+static ErlDrvTermData am_udp_passive;
 static ErlDrvTermData am_udp_error;
 static ErlDrvTermData am_empty_out_q;
 static ErlDrvTermData am_ssl_tls;
 #ifdef HAVE_SCTP
 static ErlDrvTermData am_sctp;
+static ErlDrvTermData am_sctp_passive;
 static ErlDrvTermData am_sctp_error;
 static ErlDrvTermData am_true;
 static ErlDrvTermData am_false;
@@ -1185,6 +1208,7 @@ static ErlDrvTermData am_list;
 static ErlDrvTermData am_binary;
 static ErlDrvTermData am_active;
 static ErlDrvTermData am_once;
+static ErlDrvTermData am_multi;
 static ErlDrvTermData am_buffer;
 static ErlDrvTermData am_linger;
 static ErlDrvTermData am_recbuf;
@@ -3337,6 +3361,34 @@ static int packet_binary_message
 }
 
 /*
+** active mode message: send active-to-passive transition message
+**        {tcp_passive, S} or
+**        {udp_passive, S} or
+**        {sctp_passive, S}
+*/
+ static int packet_passive_message(inet_descriptor* desc)
+ {
+     ErlDrvTermData spec[6];
+     int i = 0;
+
+     DEBUGF(("packet_passive_message(%ld):\r\n", (long)desc->port));
+
+     if (desc->sprotocol == IPPROTO_TCP)
+         i = LOAD_ATOM(spec, i, am_tcp_passive);
+     else {
+#ifdef HAVE_SCTP
+         i = LOAD_ATOM(spec, i, IS_SCTP(desc) ? am_sctp_passive : am_udp_passive);
+#else
+         i = LOAD_ATOM(spec, i, am_udp_passive);
+#endif
+     }
+     i = LOAD_PORT(spec, i, desc->dport);
+     i = LOAD_TUPLE(spec, i, 2);
+     ASSERT(i <= 6);
+     return erl_drv_output_term(desc->dport, spec, i);
+ }
+
+/*
 ** send active message {udp_error|sctp_error, S, Error}
 */
 static int packet_error_message(udp_descriptor* udesc, int err)
@@ -3378,7 +3430,7 @@ static int tcp_reply_data(tcp_descriptor* desc, char* buf, int len)
     int code;
     const char* body = buf;
     int bodylen = len;
-    
+
     packet_get_body(desc->inet.htype, &body, &bodylen);
 
     if (desc->inet.deliver == INET_DELIVER_PORT) {
@@ -3396,8 +3448,7 @@ static int tcp_reply_data(tcp_descriptor* desc, char* buf, int len)
 
     if (code < 0)
 	return code;
-    if (desc->inet.active == INET_ONCE)
-	desc->inet.active = INET_PASSIVE;
+    INET_CHECK_ACTIVE_TO_PASSIVE(INETP(desc));
     return code;
 }
 
@@ -3424,8 +3475,7 @@ tcp_reply_binary_data(tcp_descriptor* desc, ErlDrvBinary* bin, int offs, int len
     }
     if (code < 0)
 	return code;
-    if (desc->inet.active == INET_ONCE)
-	desc->inet.active = INET_PASSIVE;
+    INET_CHECK_ACTIVE_TO_PASSIVE(INETP(desc));
     return code;
 }
 
@@ -3448,8 +3498,7 @@ packet_reply_binary_data(inet_descriptor* desc, unsigned  int hsz,
 	    code = packet_binary_message(desc, bin, offs, len, extra);
 	if (code < 0)
 	    return code;
-	if (desc->active == INET_ONCE)
-	    desc->active = INET_PASSIVE;
+        INET_CHECK_ACTIVE_TO_PASSIVE(desc);
 	return code;
     }
 }
@@ -3494,6 +3543,7 @@ sock_init(void) /* May be called multiple times. */
 #ifdef HAVE_SCTP
 static void inet_init_sctp(void) {
     INIT_ATOM(sctp);
+    INIT_ATOM(sctp_passive);
     INIT_ATOM(sctp_error);
     INIT_ATOM(true);
     INIT_ATOM(false);
@@ -3503,6 +3553,7 @@ static void inet_init_sctp(void) {
     INIT_ATOM(binary);
     INIT_ATOM(active);
     INIT_ATOM(once);
+    INIT_ATOM(multi);
     INIT_ATOM(buffer);
     INIT_ATOM(linger);
     INIT_ATOM(recbuf);
@@ -3628,12 +3679,15 @@ static int inet_init()
     INIT_ATOM(tcp);
     INIT_ATOM(udp);
     INIT_ATOM(error);
+    INIT_ATOM(einval);
     INIT_ATOM(inet_async);
     INIT_ATOM(inet_reply);
     INIT_ATOM(timeout);
     INIT_ATOM(closed);
+    INIT_ATOM(tcp_passive);
     INIT_ATOM(tcp_closed);
     INIT_ATOM(tcp_error);
+    INIT_ATOM(udp_passive);
     INIT_ATOM(udp_error);
     INIT_ATOM(empty_out_q);
     INIT_ATOM(ssl_tls);
@@ -5508,8 +5562,25 @@ static int inet_set_opts(inet_descriptor* desc, char* ptr, int len)
 
 	case INET_LOPT_ACTIVE:
 	    DEBUGF(("inet_set_opts(%ld): s=%d, ACTIVE=%d\r\n",
-		    (long)desc->port, desc->s,ival));
+		    (long)desc->port, desc->s, ival));
 	    desc->active = ival;
+            if (desc->active == INET_MULTI) {
+                long ac = desc->active_count;
+                Sint16 nval = get_int16(ptr);
+                ptr += 2;
+                len -= 2;
+                ac += nval;
+                if (ac > INT16_MAX || ac < INT16_MIN)
+                    return -1;
+                desc->active_count += nval;
+                if (desc->active_count < 0)
+                    desc->active_count = 0;
+                if (desc->active_count == 0) {
+                    desc->active = INET_PASSIVE;
+                    packet_passive_message(desc);
+                }
+            } else
+                desc->active_count = 0;
 	    if ((desc->stype == SOCK_STREAM) && (desc->active != INET_PASSIVE) && 
 		(desc->state == INET_STATE_CLOSED)) {
 		tcp_closed_message((tcp_descriptor *) desc);
@@ -5820,7 +5891,8 @@ static int inet_set_opts(inet_descriptor* desc, char* ptr, int len)
 		/* XXX fprintf(stderr,"desc->htype == %d, old_htype == %d, 
 		   desc->active == %d, old_active == %d\r\n",(int)desc->htype, 
 		   (int) old_htype, (int) desc->active, (int) old_active );*/
-		return 1+(desc->htype == old_htype && desc->active == INET_ONCE);
+		return 1+(desc->htype == old_htype &&
+                          (desc->active == INET_ONCE || desc->active == INET_MULTI));
 	    }
 	    return 0;
 	}
@@ -5953,6 +6025,21 @@ static int sctp_set_opts(inet_descriptor* desc, char* ptr, int len)
 
 	case INET_LOPT_ACTIVE:
 	    desc->active = get_int32(curr);		curr += 4;
+            if (desc->active == INET_MULTI) {
+                long ac = desc->active_count;
+                Sint16 nval = get_int16(curr);          curr += 2;
+		ac += nval;
+                if (ac > INT16_MAX || ac < INT16_MIN)
+                    return -1;
+                desc->active_count += nval;
+                if (desc->active_count < 0)
+                    desc->active_count = 0;
+                if (desc->active_count == 0) {
+                    desc->active = INET_PASSIVE;
+                    packet_passive_message(desc);
+                }
+            } else
+                desc->active_count = 0;
 	    res = 0;
 	    continue;
 
@@ -6475,6 +6562,11 @@ static ErlDrvSSizeT inet_fill_opts(inet_descriptor* desc,
 	case INET_LOPT_ACTIVE:
 	    *ptr++ = opt;
 	    put_int32(desc->active, ptr);
+            if (desc->active == INET_MULTI) {
+                PLACE_FOR(2,ptr);
+                put_int16(desc->active_count, ptr);
+                ptr += 2;
+            }
 	    continue;
 	case INET_LOPT_PACKET:
 	    *ptr++ = opt;
@@ -6847,7 +6939,10 @@ static ErlDrvSSizeT sctp_fill_opts(inet_descriptor* desc,
 	}
 	case INET_LOPT_ACTIVE:
 	{
-	    PLACE_FOR(spec, i, 2*LOAD_ATOM_CNT + LOAD_TUPLE_CNT);
+            if (desc->active == INET_MULTI)
+                PLACE_FOR(spec, i, LOAD_ATOM_CNT + LOAD_INT_CNT + LOAD_TUPLE_CNT);
+            else
+                PLACE_FOR(spec, i, 2*LOAD_ATOM_CNT + LOAD_TUPLE_CNT);
 	    i = LOAD_ATOM (spec, i, am_active);
 	    switch (desc->active)
 	    {
@@ -6860,6 +6955,9 @@ static ErlDrvSSizeT sctp_fill_opts(inet_descriptor* desc,
 		case INET_ONCE   :
 		{ i = LOAD_ATOM (spec, i, am_once);  break; }
 
+                case INET_MULTI  :
+                { i = LOAD_INT(spec, i, desc->active_count); break; }
+
 		default: ASSERT (0);
 	    }
 	    i = LOAD_TUPLE (spec, i, 2);
@@ -7656,6 +7754,7 @@ static ErlDrvData inet_start(ErlDrvPort port, int size, int protocol)
 					  socket */
     desc->deliver = INET_DELIVER_TERM; /* standard term format */
     desc->active  = INET_PASSIVE;      /* start passive */
+    desc->active_count = 0;
     desc->oph = NULL;
     desc->opt = NULL;
 
diff --git a/erts/preloaded/src/prim_inet.erl b/erts/preloaded/src/prim_inet.erl
index fa621681f3..69eed716ff 100644
--- a/erts/preloaded/src/prim_inet.erl
+++ b/erts/preloaded/src/prim_inet.erl
@@ -1237,7 +1237,8 @@ type_opt_1(buffer)          -> int;
 type_opt_1(active) ->
     {enum,[{false, ?INET_PASSIVE}, 
 	   {true, ?INET_ACTIVE}, 
-	   {once, ?INET_ONCE}]};
+	   {once, ?INET_ONCE},
+           {multi, ?INET_MULTI}]};
 type_opt_1(packet) -> 
     {enum,[{0, ?TCP_PB_RAW},
 	   {1, ?TCP_PB_1},
@@ -1716,11 +1717,14 @@ encode_opt_val(Opts) ->
 	Reason -> {error,Reason}
     end.
 
+%% {active, once} and {active, N} are specially optimized because they will
+%% be used for every packet or every N packets, not only once when
+%% initializing the socket.  Measurements show that this optimization is
+%% worthwhile.
 enc_opt_val([{active,once}|Opts], Acc) ->
-    %% Specially optimized because {active,once} will be used for
-    %% every packet, not only once when initializing the socket.
-    %% Measurements show that this optimization is worthwhile.
     enc_opt_val(Opts, [<<?INET_LOPT_ACTIVE:8,?INET_ONCE:32>>|Acc]);
+enc_opt_val([{active,N}|Opts], Acc) when is_integer(N), N < 32768, N >= -32768 ->
+    enc_opt_val(Opts, [<<?INET_LOPT_ACTIVE:8,?INET_MULTI:32,N:16>>|Acc]);
 enc_opt_val([{raw,P,O,B}|Opts], Acc) ->
     enc_opt_val(Opts, Acc, raw, {P,O,B});
 enc_opt_val([{Opt,Val}|Opts], Acc) ->
@@ -1810,6 +1814,14 @@ dec_opt_val([]) -> [].
 dec_opt_val(Buf, raw, Type) ->
     {{P,O,B},T} = dec_value(Type, Buf),
     [{raw,P,O,B}|dec_opt_val(T)];
+dec_opt_val(Buf, active, Type) ->
+    case dec_value(Type, Buf) of
+        {multi,[M0,M1|T]} ->
+            <<N:16>> = list_to_binary([M0,M1]),
+            [{active,N}|dec_opt_val(T)];
+        {Val,T} ->
+            [{active,Val}|dec_opt_val(T)]
+    end;
 dec_opt_val(Buf, Opt, Type) ->
     {Val,T} = dec_value(Type, Buf),
     [{Opt,Val}|dec_opt_val(T)].
diff --git a/lib/kernel/doc/src/gen_sctp.xml b/lib/kernel/doc/src/gen_sctp.xml
index dacea7a239..986260638d 100644
--- a/lib/kernel/doc/src/gen_sctp.xml
+++ b/lib/kernel/doc/src/gen_sctp.xml
@@ -496,9 +496,11 @@
       orthogonal to the sets of TCP, UDP and generic INET options:
       only those options which are explicitly listed below are allowed
       for SCTP sockets. Options can be set on the socket using
-      <c>gen_sctp:open/1,2</c> or <c>inet:setopts/2</c>,
-      retrieved using <c>inet:getopts/2</c>, and when calling
-      <c>gen_sctp:connect/4,5</c> options can be changed.</p>
+      <seealso marker="#open/1"><c>gen_sctp:open/1,2</c></seealso>
+      or <seealso marker="inet#setopts/2"><c>inet:setopts/2</c></seealso>,
+      retrieved using <seealso marker="inet#getopts/2"><c>inet:getopts/2</c></seealso>,
+      and when calling <seealso marker="#connect/4"><c>gen_sctp:connect/4,5</c></seealso>
+      options can be changed.</p>
     <marker id="option-binary"></marker>
     <marker id="option-list"></marker>
     <taglist>
@@ -507,7 +509,7 @@
         <p>Determines the type of data returned from <c>gen_sctp:recv/1,2</c>.</p>
         <marker id="option-active"></marker>
       </item>
-      <tag><c>{active, true|false|once}</c></tag>
+      <tag><c>{active, true|false|once|N}</c></tag>
       <item>
         <list type="bulleted">
           <item>
@@ -524,11 +526,28 @@
           </item>
           <item>
             <p>If <c>once</c>, only one message is automatically placed
-              in the message queue, after that the mode is automatically
-              re-set to passive. This provides flow control as well as
+              in the message queue, and after that the mode is automatically
+              reset to passive. This provides flow control as well as
               the possibility for the receiver to listen for its incoming
               SCTP data interleaved with other inter-process messages.</p>
           </item>
+          <item>
+            <p>If <c>active</c> is specified as an integer <c>N</c> in the
+              range -32768 to 32767 (inclusive), then that number is added to
+              the socket's count of the number of data messages to be
+              delivered to the controlling process. If the result of the
+              addition would be negative, the count is set to 0. Once the
+              count reaches 0, either through the delivery of messages or by
+              being explicitly set with <seealso
+              marker="inet#setopts/2">inet:setopts/2</seealso>, the socket's
+              mode is automatically reset to passive (<c>{active,
+              false}</c>) mode. When a socket in this active mode transitions to
+              passive mode, the message <c>{sctp_passive, Socket}</c> is sent
+              to the controlling process to notify it that if it wants to
+              receive more data messages from the socket, it must call
+              <seealso marker="inet#setopts/2">inet:setopts/2</seealso> to set
+              the socket back into an active mode.</p>
+          </item>
         </list>
       </item>
      <tag><c>{tos, integer()}</c></tag>
diff --git a/lib/kernel/doc/src/gen_tcp.xml b/lib/kernel/doc/src/gen_tcp.xml
index 7afd73f4bf..dbd0d3c815 100644
--- a/lib/kernel/doc/src/gen_tcp.xml
+++ b/lib/kernel/doc/src/gen_tcp.xml
@@ -148,6 +148,12 @@ do_recv(Sock, Bs) ->
           as messages:</p>
         <code type="none">
 {tcp, Socket, Data}</code>
+        <p>If the socket is in <c>{active, N}</c> mode (see <seealso marker="inet#setopts/2">
+          inet:setopts/2</seealso> for details) and its message counter
+          drops to 0, the following message is delivered to indicate that the
+          socket has transitioned to passive (<c>{active, false}</c>) mode:</p>
+        <code type="none">
+{tcp_passive, Socket}</code>
         <p>If the socket is closed, the following message is delivered:</p>
         <code type="none">
 {tcp_closed, Socket}</code>
diff --git a/lib/kernel/doc/src/gen_udp.xml b/lib/kernel/doc/src/gen_udp.xml
index faca31e928..503725fe18 100644
--- a/lib/kernel/doc/src/gen_udp.xml
+++ b/lib/kernel/doc/src/gen_udp.xml
@@ -145,14 +145,23 @@
               <seealso marker="inet#setopts/2">inet:setopts/2</seealso>.</p>
           </item>
         </taglist>
-        <p>The returned socket <c><anno>Socket</anno></c> is used to send packets
-          from this port with <c>send/4</c>. When UDP packets arrive at
-          the opened port, they are delivered as messages:</p>
+        <p>The returned socket <c><anno>Socket</anno></c> is used to send
+          packets from this port with <c>send/4</c>. When UDP packets arrive
+          at the opened port, if the socket is in an active mode the packets
+          are delivered as messages to the controlling process:</p>
         <code type="none">
 {udp, Socket, IP, InPortNo, Packet}</code>
-        <p>Note that arriving UDP packets that are longer than
+        <p>If the socket is not in an active mode, data can be
+          retrieved via the <seealso marker="#recv/2">recv/2,3</seealso> calls.
+          Note that arriving UDP packets that are longer than
           the receive buffer option specifies, might be truncated
           without warning.</p>
+        <p>When a socket in <c>{active, N}</c> mode (see <seealso marker="inet#setopts/2">
+          inet:setopts/2</seealso> for details) transitions to passive
+          (<c>{active, false}</c>) mode, the controlling process is notified by a
+          message of the following form:</p>
+        <code type="none">
+{udp_passive, Socket}</code>
         <p><c>IP</c> and <c>InPortNo</c> define the address from which
           <c>Packet</c> came. <c>Packet</c> is a list of bytes if
           the option <c>list</c> was specified. <c>Packet</c> is a
diff --git a/lib/kernel/doc/src/inet.xml b/lib/kernel/doc/src/inet.xml
index e1b0b34888..fb8bf32978 100644
--- a/lib/kernel/doc/src/inet.xml
+++ b/lib/kernel/doc/src/inet.xml
@@ -456,47 +456,66 @@ fe80::204:acff:fe17:bf38
         <p>Sets one or more options for a socket. The following options
           are available:</p>
         <taglist>
-          <tag><c>{active, true | false | once}</c></tag>
+          <tag><c>{active, true | false | once | N}</c></tag>
           <item>
             <p>If the value is <c>true</c>, which is the default,
               everything received from the socket will be sent as
               messages to the receiving process. If the value is
               <c>false</c> (passive mode), the process must explicitly
-              receive incoming data by calling <c>gen_tcp:recv/2,3</c>
-              or <c>gen_udp:recv/2,3</c> (depending on the type of
-              socket).</p>
+              receive incoming data by calling
+              <seealso marker="gen_tcp#recv/2"><c>gen_tcp:recv/2,3</c></seealso>,
+              <seealso marker="gen_udp#recv/2"><c>gen_udp:recv/2,3</c></seealso>
+              or <seealso marker="gen_sctp#recv/1"><c>gen_sctp:recv/1,2</c></seealso>
+              (depending on the type of socket).</p>
             <p>If the value is <c>once</c> (<c>{active, once}</c>),
               <em>one</em> data message from the socket will be sent
               to the process.  To receive one more message,
               <c>setopts/2</c> must be called again with the
               <c>{active, once}</c> option.</p>
-            <p>When using <c>{active, once}</c>, the socket changes
-              behaviour automatically when data is received. This can
-              sometimes be confusing in combination with connection
-              oriented sockets (i.e. <c>gen_tcp</c>) as a socket with
-              <c>{active, false}</c> behaviour reports closing
+            <p>If the value is an integer <c>N</c> in the range -32768 to 32767
+              (inclusive), the value is added to the socket's count of data
+              messages sent to the controlling process. A socket's default
+              message count is 0. If a negative value is specified and its
+              magnitude is equal to or greater than the socket's current
+              message count, the socket's message count is set to 0. Once
+              the socket's message count reaches 0, either due to sending
+              received data messages to the process or by being explicitly set,
+              the process is then notified by a special message, specific to
+              the type of socket, that the socket has entered passive
+              mode. Once the socket enters passive mode, to receive more
+              messages <c>setopts/2</c> must be called again to set the
+              socket back into an active mode.</p>
+            <p>When using <c>{active, once}</c> or <c>{active, N}</c>, the
+              socket changes behaviour automatically when data is received.
+              This can sometimes be confusing in combination with
+              connection-oriented sockets (i.e. <c>gen_tcp</c>) as a socket
+              with <c>{active, false}</c> behaviour reports closing
               differently than a socket with <c>{active, true}</c>
               behaviour. To make programming easier, a socket where
               the peer closed and this was detected while in
               <c>{active, false}</c> mode, will still generate the
               message
-              <c>{tcp_closed,Socket}</c> when set to <c>{active, once}</c> or <c>{active, true}</c> mode. It is therefore
+              <c>{tcp_closed,Socket}</c> when set to <c>{active, once}</c>,
+              <c>{active, true}</c> or <c>{active, N}</c> mode. It is therefore
               safe to assume that the message
               <c>{tcp_closed,Socket}</c>, possibly followed by socket
               port termination (depending on the <c>exit_on_close</c>
               option) will eventually appear when a socket changes
               back and forth between <c>{active, true}</c> and
-              <c>{active, false}</c> mode. However, 
+              <c>{active, false}</c> mode. However,
               <em>when</em> peer closing is detected is all up to the
               underlying TCP/IP stack and protocol.</p>
-            <p>Note that <c>{active,true}</c> mode provides no flow
+            <p>Note that <c>{active, true}</c> mode provides no flow
               control; a fast sender could easily overflow the
-              receiver with incoming messages. Use active mode only if
+              receiver with incoming messages. The same is true of
+              <c>{active, N}</c> mode while the message count is greater
+              than zero. Use active mode only if
               your high-level protocol provides its own flow control
               (for instance, acknowledging received messages) or the
-              amount of data exchanged is small. <c>{active,false}</c>
-              mode or use of the <c>{active, once}</c> mode provides
-              flow control; the other side will not be able send
+              amount of data exchanged is small. <c>{active, false}</c>
+              mode, use of the <c>{active, once}</c> mode or <c>{active, N}</c>
+              mode with values of <c>N</c> appropriate for the application
+              provides flow control; the other side will not be able send
               faster than the receiver can read.</p>
           </item>
 
diff --git a/lib/kernel/src/gen_sctp.erl b/lib/kernel/src/gen_sctp.erl
index 0d005eb651..23a698f9b2 100644
--- a/lib/kernel/src/gen_sctp.erl
+++ b/lib/kernel/src/gen_sctp.erl
@@ -36,7 +36,7 @@
 
 -type assoc_id() :: term().
 -type option() ::
-        {active, true | false | once} |
+        {active, true | false | once | -32768..32767} |
         {buffer, non_neg_integer()} |
         {dontroute, boolean()} |
         {high_msgq_watermark, pos_integer()} |
diff --git a/lib/kernel/src/gen_tcp.erl b/lib/kernel/src/gen_tcp.erl
index 96026aa428..bc8ffbe5e3 100644
--- a/lib/kernel/src/gen_tcp.erl
+++ b/lib/kernel/src/gen_tcp.erl
@@ -30,7 +30,7 @@
 -include("file.hrl").
 
 -type option() ::
-        {active,          true | false | once} |
+        {active,          true | false | once | -32768..32767} |
         {buffer,          non_neg_integer()} |
         {delay_send,      boolean()} |
         {deliver,         port | term} |
diff --git a/lib/kernel/src/gen_udp.erl b/lib/kernel/src/gen_udp.erl
index e82b11d2ef..70dceb3679 100644
--- a/lib/kernel/src/gen_udp.erl
+++ b/lib/kernel/src/gen_udp.erl
@@ -26,7 +26,7 @@
 -include("inet_int.hrl").
 
 -type option() ::
-        {active,          true | false | once} |
+        {active,          true | false | once | -32768..32767} |
         {add_membership,  {inet:ip_address(), inet:ip_address()}} |
         {broadcast,       boolean()} |
         {buffer,          non_neg_integer()} |
diff --git a/lib/kernel/src/inet.erl b/lib/kernel/src/inet.erl
index 9cfd3819f0..7261d6486f 100644
--- a/lib/kernel/src/inet.erl
+++ b/lib/kernel/src/inet.erl
@@ -667,6 +667,9 @@ con_opt([Opt | Opts], R, As) ->
 		false ->
 		    {error, badarg}
 	    end;
+        {active,N} when is_integer(N), N < 32768, N >= -32768 ->
+            NOpts = lists:keydelete(active, 1, R#connect_opts.opts),
+            con_opt(Opts, R#connect_opts { opts = [{active,N}|NOpts] }, As);
 	{Name,Val} when is_atom(Name) -> con_add(Name, Val, R, Opts, As);
 	_ -> {error, badarg}
     end;
@@ -733,6 +736,9 @@ list_opt([Opt | Opts], R, As) ->
 		false ->
 		    {error, badarg}
 	    end;
+        {active,N} when is_integer(N), N < 32768, N >= -32768 ->
+            NOpts = lists:keydelete(active, 1, R#listen_opts.opts),
+            list_opt(Opts, R#listen_opts { opts = [{active,N}|NOpts] }, As);
 	{Name,Val} when is_atom(Name) -> list_add(Name, Val, R, Opts, As);
 	_ -> {error, badarg}
     end;
@@ -787,6 +793,9 @@ udp_opt([Opt | Opts], R, As) ->
 		false ->
 		    {error, badarg}
 	    end;
+        {active,N} when is_integer(N), N < 32768, N >= -32768 ->
+            NOpts = lists:keydelete(active, 1, R#udp_opts.opts),
+            udp_opt(Opts, R#udp_opts { opts = [{active,N}|NOpts] }, As);
 	{Name,Val} when is_atom(Name) -> udp_add(Name, Val, R, Opts, As);
 	_ -> {error, badarg}
     end;
@@ -805,7 +814,7 @@ udp_add(Name, Val, R, Opts, As) ->
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %  Currently supported options include:
 %  (*) {mode,   list|binary}	 or just list|binary
-%  (*) {active, true|false|once}
+%  (*) {active, true|false|once|N}
 %  (*) {sctp_module, inet_sctp|inet6_sctp} or just inet|inet6
 %  (*) options set via setsockopt.
 %      The full list is below in sctp_options/0 .
@@ -867,6 +876,9 @@ sctp_opt([Opt|Opts], Mod, R, As) ->
 		false ->
 		    {error, badarg}
 	    end;
+        {active,N} when is_integer(N), N < 32768, N >= -32768 ->
+            NOpts = lists:keydelete(active, 1, R#sctp_opts.opts),
+            sctp_opt(Opts, Mod, R#sctp_opts { opts = [{active,N}|NOpts] }, As);
 	{Name,Val}	-> sctp_opt (Opts, Mod, R, As, Name, Val);
 	_ -> {error,badarg}
     end;
diff --git a/lib/kernel/src/inet_int.hrl b/lib/kernel/src/inet_int.hrl
index 18a4a61b2f..024a5fd3b6 100644
--- a/lib/kernel/src/inet_int.hrl
+++ b/lib/kernel/src/inet_int.hrl
@@ -46,6 +46,7 @@
 -define(INET_PASSIVE, 0).
 -define(INET_ACTIVE,  1).
 -define(INET_ONCE,    2). % Active once then passive
+-define(INET_MULTI,   3). % Active N then passive
 
 %% state codes (getstatus, INET_REQ_GETSTATUS)
 -define(INET_F_OPEN,         16#0001).
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