Make high_msgq_watermark and low_msgq_watermark generic inet options

9 files changed, 97 insertions, 71 deletions

diff --git a/erts/emulator/drivers/common/inet_drv.c b/erts/emulator/drivers/common/inet_drv.c
index fb128fff7d..0fdd6934c2 100644
--- a/erts/emulator/drivers/common/inet_drv.c
+++ b/erts/emulator/drivers/common/inet_drv.c
@@ -678,8 +678,8 @@ static int my_strncasecmp(const char *s1, const char *s2, size_t n)
 #define INET_LOPT_UDP_READ_PACKETS 33  /* Number of packets to read */
 #define INET_OPT_RAW               34  /* Raw socket options */
 #define INET_LOPT_TCP_SEND_TIMEOUT_CLOSE 35  /* auto-close on send timeout or not */
-#define INET_LOPT_TCP_MSGQ_HIWTRMRK     36  /* set local high watermark */
-#define INET_LOPT_TCP_MSGQ_LOWTRMRK     37  /* set local low watermark */
+#define INET_LOPT_MSGQ_HIWTRMRK     36  /* set local msgq high watermark */
+#define INET_LOPT_MSGQ_LOWTRMRK     37  /* set local msgq low watermark */
 /* SCTP options: a separate range, from 100: */
 #define SCTP_OPT_RTOINFO		100
 #define SCTP_OPT_ASSOCINFO		101
@@ -5474,27 +5474,25 @@ static int inet_set_opts(inet_descriptor* desc, char* ptr, int len)
 	    }
 	    continue;
 
-	case INET_LOPT_TCP_MSGQ_HIWTRMRK:
-	    if (desc->stype == SOCK_STREAM) {
-		ErlDrvSizeT high;
-		if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
-		    || ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
-		    return -1;
-		high = (ErlDrvSizeT) ival;
-		erl_drv_busy_msgq_limits(desc->port, NULL, &high);
-	    }
+	case INET_LOPT_MSGQ_HIWTRMRK: {
+	    ErlDrvSizeT high;
+	    if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
+		|| ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
+		return -1;
+	    high = (ErlDrvSizeT) ival;
+	    erl_drv_busy_msgq_limits(desc->port, NULL, &high);
 	    continue;
+	}
 
-	case INET_LOPT_TCP_MSGQ_LOWTRMRK:
-	    if (desc->stype == SOCK_STREAM) {
-		ErlDrvSizeT low;
-		if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
-		    || ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
-		    return -1;
-		low = (ErlDrvSizeT) ival;
-		erl_drv_busy_msgq_limits(desc->port, &low, NULL);
-	    }
+	case INET_LOPT_MSGQ_LOWTRMRK: {
+	    ErlDrvSizeT low;
+	    if (ival < ERL_DRV_BUSY_MSGQ_LIM_MIN
+		|| ERL_DRV_BUSY_MSGQ_LIM_MAX < ival)
+		return -1;
+	    low = (ErlDrvSizeT) ival;
+	    erl_drv_busy_msgq_limits(desc->port, &low, NULL);
 	    continue;
+	}
 
 	case INET_LOPT_TCP_SEND_TIMEOUT:
 	    if (desc->stype == SOCK_STREAM) {
@@ -6396,31 +6394,23 @@ static ErlDrvSSizeT inet_fill_opts(inet_descriptor* desc,
 	    }
 	    continue;
 
-	case INET_LOPT_TCP_MSGQ_HIWTRMRK:
-	    if (desc->stype == SOCK_STREAM) {
-		ErlDrvSizeT high = ERL_DRV_BUSY_MSGQ_READ_ONLY;
-		*ptr++ = opt;
-		erl_drv_busy_msgq_limits(desc->port, NULL, &high);
-		ival = high > INT_MAX ? INT_MAX : (int) high;
-		put_int32(ival, ptr);
-	    }
-	    else {
-		TRUNCATE_TO(0,ptr);
-	    }
+	case INET_LOPT_MSGQ_HIWTRMRK: {
+	    ErlDrvSizeT high = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+	    *ptr++ = opt;
+	    erl_drv_busy_msgq_limits(desc->port, NULL, &high);
+	    ival = high > INT_MAX ? INT_MAX : (int) high;
+	    put_int32(ival, ptr);
 	    continue;
+	}
 
-	case INET_LOPT_TCP_MSGQ_LOWTRMRK:
-	    if (desc->stype == SOCK_STREAM) {
-		ErlDrvSizeT low = ERL_DRV_BUSY_MSGQ_READ_ONLY;
-		*ptr++ = opt;
-		erl_drv_busy_msgq_limits(desc->port, &low, NULL);
-		ival = low > INT_MAX ? INT_MAX : (int) low;
-		put_int32(ival, ptr);
-	    }
-	    else {
-		TRUNCATE_TO(0,ptr);
-	    }
+	case INET_LOPT_MSGQ_LOWTRMRK: {
+	    ErlDrvSizeT low = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+	    *ptr++ = opt;
+	    erl_drv_busy_msgq_limits(desc->port, &low, NULL);
+	    ival = low > INT_MAX ? INT_MAX : (int) low;
+	    put_int32(ival, ptr);
 	    continue;
+	}
 
 	case INET_LOPT_TCP_SEND_TIMEOUT:
 	    if (desc->stype == SOCK_STREAM) {
@@ -7469,6 +7459,20 @@ static void inet_stop(inet_descriptor* desc)
     FREE(desc);
 }
 
+static void set_default_msgq_limits(ErlDrvPort port)
+{
+    ErlDrvSizeT q_high = INET_HIGH_MSGQ_WATERMARK;
+    ErlDrvSizeT q_low = INET_LOW_MSGQ_WATERMARK;
+    if (q_low < ERL_DRV_BUSY_MSGQ_LIM_MIN)
+	q_low = ERL_DRV_BUSY_MSGQ_LIM_MIN;
+    else if (q_low > ERL_DRV_BUSY_MSGQ_LIM_MAX)
+	q_low = ERL_DRV_BUSY_MSGQ_LIM_MAX;
+    if (q_high < ERL_DRV_BUSY_MSGQ_LIM_MIN)
+	q_high = ERL_DRV_BUSY_MSGQ_LIM_MIN;
+    else if (q_high > ERL_DRV_BUSY_MSGQ_LIM_MAX)
+	q_high = ERL_DRV_BUSY_MSGQ_LIM_MAX;
+    erl_drv_busy_msgq_limits(port, &q_low, &q_high);
+}
 
 /* Allocate descriptor */
 static ErlDrvData inet_start(ErlDrvPort port, int size, int protocol)
@@ -8089,9 +8093,8 @@ static int tcp_inet_init(void)
 
 /* initialize the TCP descriptor */
 
-static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
+static ErlDrvData prep_tcp_inet_start(ErlDrvPort port, char* args)
 {
-    ErlDrvSizeT q_low, q_high;
     tcp_descriptor* desc;
     DEBUGF(("tcp_inet_start(%ld) {\r\n", (long)port));
 
@@ -8101,17 +8104,6 @@ static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
 	return ERL_DRV_ERROR_ERRNO;
     desc->high = INET_HIGH_WATERMARK;
     desc->low  = INET_LOW_WATERMARK;
-    q_high = INET_HIGH_MSGQ_WATERMARK;
-    q_low = INET_LOW_MSGQ_WATERMARK;
-    if (q_low < ERL_DRV_BUSY_MSGQ_LIM_MIN)
-	q_low = ERL_DRV_BUSY_MSGQ_LIM_MIN;
-    else if (q_low > ERL_DRV_BUSY_MSGQ_LIM_MAX)
-	q_low = ERL_DRV_BUSY_MSGQ_LIM_MAX;
-    if (q_high < ERL_DRV_BUSY_MSGQ_LIM_MIN)
-	q_high = ERL_DRV_BUSY_MSGQ_LIM_MIN;
-    else if (q_high > ERL_DRV_BUSY_MSGQ_LIM_MAX)
-	q_high = ERL_DRV_BUSY_MSGQ_LIM_MAX;
-    erl_drv_busy_msgq_limits(port, &q_low, &q_high);
     desc->send_timeout = INET_INFINITY;
     desc->send_timeout_close = 0;
     desc->busy_on_send = 0;
@@ -8128,6 +8120,12 @@ static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
     return (ErlDrvData) desc;
 }
 
+static ErlDrvData tcp_inet_start(ErlDrvPort port, char* args)
+{
+    ErlDrvData data = prep_tcp_inet_start(port, args);
+    set_default_msgq_limits(port);
+    return data;
+}
 /* Copy a descriptor, by creating a new port with same settings
  * as the descriptor desc.
  * return NULL on error (SYSTEM_LIMIT no ports avail)
@@ -8139,7 +8137,7 @@ static tcp_descriptor* tcp_inet_copy(tcp_descriptor* desc,SOCKET s,
     ErlDrvPort port = desc->inet.port;
     tcp_descriptor* copy_desc;
 
-    copy_desc = (tcp_descriptor*) tcp_inet_start(port, NULL);
+    copy_desc = (tcp_descriptor*) prep_tcp_inet_start(port, NULL);
 
     /* Setup event if needed */
     if ((copy_desc->inet.s = s) != INVALID_SOCKET) {
@@ -9862,12 +9860,15 @@ static int should_use_so_bsdcompat(void)
  * as the descriptor desc.
  * return NULL on error (ENFILE no ports avail)
  */
+static ErlDrvData packet_inet_start(ErlDrvPort port, char* args, int protocol);
+
 static udp_descriptor* sctp_inet_copy(udp_descriptor* desc, SOCKET s, int* err)
 {
+    ErlDrvSizeT q_low, q_high;
     ErlDrvPort port = desc->inet.port;
     udp_descriptor* copy_desc;
 
-    copy_desc = (udp_descriptor*) sctp_inet_start(port, NULL);
+    copy_desc = (udp_descriptor*) packet_inet_start(port, NULL, IPPROTO_SCTP);
 
     /* Setup event if needed */
     if ((copy_desc->inet.s = s) != INVALID_SOCKET) {
@@ -9898,9 +9899,17 @@ static udp_descriptor* sctp_inet_copy(udp_descriptor* desc, SOCKET s, int* err)
 	FREE(copy_desc);
 	return NULL;
     }
+
+    /* Read busy msgq limits of parent */
+    q_low = q_high = ERL_DRV_BUSY_MSGQ_READ_ONLY;
+    erl_drv_busy_msgq_limits(desc->inet.port, &q_low, &q_high);
+    /* Write same busy msgq limits to child */
+    erl_drv_busy_msgq_limits(port, &q_low, &q_high);
+
     copy_desc->inet.port = port;
     copy_desc->inet.dport = driver_mk_port(port);
     *err = 0;
+
     return copy_desc;
 }
 #endif
@@ -9933,13 +9942,17 @@ static ErlDrvData packet_inet_start(ErlDrvPort port, char* args, int protocol)
 
 static ErlDrvData udp_inet_start(ErlDrvPort port, char *args)
 {
-    return packet_inet_start(port, args, IPPROTO_UDP);
+    ErlDrvData data = packet_inet_start(port, args, IPPROTO_UDP);
+    set_default_msgq_limits(port);
+    return data;
 }
 
 #ifdef HAVE_SCTP
 static ErlDrvData sctp_inet_start(ErlDrvPort port, char *args)
 {
-    return packet_inet_start(port, args, IPPROTO_SCTP);
+    ErlDrvData data = packet_inet_start(port, args, IPPROTO_SCTP);
+    set_default_msgq_limits(port);
+    return data;
 }
 #endif
 
diff --git a/erts/preloaded/ebin/prim_inet.beam b/erts/preloaded/ebin/prim_inet.beam
index 8c47d8b611..474bb120da 100644
--- a/erts/preloaded/ebin/prim_inet.beam
+++ b/erts/preloaded/ebin/prim_inet.beam
diff --git a/erts/preloaded/src/prim_inet.erl b/erts/preloaded/src/prim_inet.erl
index 21d23159f0..d0e70aa95c 100644
--- a/erts/preloaded/src/prim_inet.erl
+++ b/erts/preloaded/src/prim_inet.erl
@@ -1072,8 +1072,8 @@ enc_opt(deliver)         -> ?INET_LOPT_DELIVER;
 enc_opt(exit_on_close)   -> ?INET_LOPT_EXITONCLOSE;
 enc_opt(high_watermark)  -> ?INET_LOPT_TCP_HIWTRMRK;
 enc_opt(low_watermark)   -> ?INET_LOPT_TCP_LOWTRMRK;
-enc_opt(high_msgq_watermark)  -> ?INET_LOPT_TCP_MSGQ_HIWTRMRK;
-enc_opt(low_msgq_watermark)   -> ?INET_LOPT_TCP_MSGQ_LOWTRMRK;
+enc_opt(high_msgq_watermark)  -> ?INET_LOPT_MSGQ_HIWTRMRK;
+enc_opt(low_msgq_watermark)   -> ?INET_LOPT_MSGQ_LOWTRMRK;
 enc_opt(send_timeout)    -> ?INET_LOPT_TCP_SEND_TIMEOUT;
 enc_opt(send_timeout_close) -> ?INET_LOPT_TCP_SEND_TIMEOUT_CLOSE;
 enc_opt(delay_send)      -> ?INET_LOPT_TCP_DELAY_SEND;
@@ -1128,8 +1128,8 @@ dec_opt(?INET_LOPT_DELIVER)       -> deliver;
 dec_opt(?INET_LOPT_EXITONCLOSE)   -> exit_on_close;
 dec_opt(?INET_LOPT_TCP_HIWTRMRK)  -> high_watermark;
 dec_opt(?INET_LOPT_TCP_LOWTRMRK)  -> low_watermark;
-dec_opt(?INET_LOPT_TCP_MSGQ_HIWTRMRK)  -> high_msgq_watermark;
-dec_opt(?INET_LOPT_TCP_MSGQ_LOWTRMRK)  -> low_msgq_watermark;
+dec_opt(?INET_LOPT_MSGQ_HIWTRMRK)  -> high_msgq_watermark;
+dec_opt(?INET_LOPT_MSGQ_LOWTRMRK)  -> low_msgq_watermark;
 dec_opt(?INET_LOPT_TCP_SEND_TIMEOUT) -> send_timeout;
 dec_opt(?INET_LOPT_TCP_SEND_TIMEOUT_CLOSE) -> send_timeout_close;
 dec_opt(?INET_LOPT_TCP_DELAY_SEND)   -> delay_send;
diff --git a/lib/kernel/doc/src/inet.xml b/lib/kernel/doc/src/inet.xml
index 541500a300..7cd98914d1 100644
--- a/lib/kernel/doc/src/inet.xml
+++ b/lib/kernel/doc/src/inet.xml
@@ -559,7 +559,7 @@ fe80::204:acff:fe17:bf38
               <c>[Byte1,Byte2|Binary]</c>.</p>
           </item>
 
-          <tag><c>{high_msgq_watermark, Size}</c> (TCP/IP sockets)</tag>
+          <tag><c>{high_msgq_watermark, Size}</c></tag>
           <item>
 	    <p>The socket message queue will be set into a busy
 	    state when the amount of data queued on the message
@@ -674,7 +674,7 @@ fe80::204:acff:fe17:bf38
 				  the flushing time-out in seconds.</p>
 		  </item>
 
-          <tag><c>{low_msgq_watermark, Size}</c> (TCP/IP sockets)</tag>
+          <tag><c>{low_msgq_watermark, Size}</c></tag>
           <item>
 	    <p>If the socket message queue is in a busy state, the
 	    socket message queue will be set in a not busy state when
diff --git a/lib/kernel/src/gen_sctp.erl b/lib/kernel/src/gen_sctp.erl
index 74ad192802..acc116df84 100644
--- a/lib/kernel/src/gen_sctp.erl
+++ b/lib/kernel/src/gen_sctp.erl
@@ -39,7 +39,9 @@
         {active, true | false | once} |
         {buffer, non_neg_integer()} |
         {dontroute, boolean()} |
+        {high_msgq_watermark, pos_integer()} |
         {linger, {boolean(), non_neg_integer()}} |
+        {low_msgq_watermark, pos_integer()} |
         {mode, list | binary} | list | binary |
         {priority, non_neg_integer()} |
         {recbuf, non_neg_integer()} |
@@ -68,7 +70,9 @@
         active |
         buffer |
         dontroute |
+        high_msgq_watermark |
         linger |
+        low_msgq_watermark |
         mode |
         priority |
         recbuf |
diff --git a/lib/kernel/src/gen_udp.erl b/lib/kernel/src/gen_udp.erl
index c5a1173575..fb5737f82e 100644
--- a/lib/kernel/src/gen_udp.erl
+++ b/lib/kernel/src/gen_udp.erl
@@ -34,6 +34,8 @@
         {dontroute,       boolean()} |
         {drop_membership, {inet:ip_address(), inet:ip_address()}} |
         {header,          non_neg_integer()} |
+        {high_msgq_watermark, pos_integer()} |
+        {low_msgq_watermark, pos_integer()} |
         {mode,            list | binary} | list | binary |
         {multicast_if,    inet:ip_address()} |
         {multicast_loop,  boolean()} |
@@ -56,6 +58,8 @@
         deliver |
         dontroute |
         header |
+        high_msgq_watermark |
+        low_msgq_watermark |
         mode |
         multicast_if |
         multicast_loop |
diff --git a/lib/kernel/src/inet.erl b/lib/kernel/src/inet.erl
index 9670271b2e..aada1252ad 100644
--- a/lib/kernel/src/inet.erl
+++ b/lib/kernel/src/inet.erl
@@ -712,7 +712,8 @@ udp_options() ->
     [tos, priority, reuseaddr, sndbuf, recbuf, header, active, buffer, mode, 
      deliver, ipv6_v6only,
      broadcast, dontroute, multicast_if, multicast_ttl, multicast_loop,
-     add_membership, drop_membership, read_packets,raw].
+     add_membership, drop_membership, read_packets,raw,
+     high_msgq_watermark, low_msgq_watermark].
 
 
 udp_options(Opts, Family) ->
@@ -766,7 +767,7 @@ udp_add(Name, Val, R, Opts, As) ->
 sctp_options() ->
 [   % The following are generic inet options supported for SCTP sockets:
     mode, active, buffer, tos, priority, dontroute, reuseaddr, linger, sndbuf,
-    recbuf, ipv6_v6only,
+    recbuf, ipv6_v6only, high_msgq_watermark, low_msgq_watermark,
 
     % Other options are SCTP-specific (though they may be similar to their
     % TCP and UDP counter-parts):
diff --git a/lib/kernel/src/inet_int.hrl b/lib/kernel/src/inet_int.hrl
index 000119bc74..a01c733d81 100644
--- a/lib/kernel/src/inet_int.hrl
+++ b/lib/kernel/src/inet_int.hrl
@@ -141,8 +141,8 @@
 -define(INET_LOPT_READ_PACKETS,  33).
 -define(INET_OPT_RAW,            34).
 -define(INET_LOPT_TCP_SEND_TIMEOUT_CLOSE, 35).
--define(INET_LOPT_TCP_MSGQ_HIWTRMRK,  36).
--define(INET_LOPT_TCP_MSGQ_LOWTRMRK,  37).
+-define(INET_LOPT_MSGQ_HIWTRMRK,  36).
+-define(INET_LOPT_MSGQ_LOWTRMRK,  37).
 % Specific SCTP options: separate range:
 -define(SCTP_OPT_RTOINFO,	 	100).
 -define(SCTP_OPT_ASSOCINFO,	 	101).
diff --git a/lib/kernel/test/inet_sockopt_SUITE.erl b/lib/kernel/test/inet_sockopt_SUITE.erl
index 75496ce745..185751fead 100644
--- a/lib/kernel/test/inet_sockopt_SUITE.erl
+++ b/lib/kernel/test/inet_sockopt_SUITE.erl
@@ -772,8 +772,10 @@ all_listen_options() ->
      {mode,list,binary,true,true}, 
      {deliver,term,port,true,true}, 
      {exit_on_close, true, false, true, true},
-     %{high_watermark,4096,8192,true,true}, 
-     %{low_watermark,2048,4096,true,true}, 
+     {high_watermark,4096,8192,true,true}, 
+     {low_watermark,2048,4096,true,true}, 
+     {high_msgq_watermark,4096,8192,true,true}, 
+     {low_msgq_watermark,2048,4096,true,true}, 
      {send_timeout,infinity,1000,true,true},
      {send_timeout_close,false,true,true,true},
      {delay_send,false,true,true,true}, 
@@ -797,6 +799,8 @@ all_connect_options() ->
      {exit_on_close, true, false, true, true},
      {high_watermark,4096,8192,false,true}, 
      {low_watermark,2048,4096,false,true}, 
+     {high_msgq_watermark,4096,8192,true,true}, 
+     {low_msgq_watermark,2048,4096,true,true}, 
      {send_timeout,infinity,1000,true,true},
      {send_timeout_close,false,true,true,true},
      {delay_send,false,true,true,true},