From 7fbf2c26ac063988818230a0e18a9df48c2fbf2d Mon Sep 17 00:00:00 2001 From: Steve Vinoski Date: Tue, 27 Aug 2013 11:42:00 -0400 Subject: 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. --- erts/emulator/drivers/common/inet_drv.c | 119 +++++++++++++++++++++++++++--- erts/preloaded/src/prim_inet.erl | 20 ++++- lib/kernel/doc/src/gen_sctp.xml | 31 ++++++-- lib/kernel/doc/src/gen_tcp.xml | 6 ++ lib/kernel/doc/src/gen_udp.xml | 17 ++++- lib/kernel/doc/src/inet.xml | 51 +++++++++---- lib/kernel/src/gen_sctp.erl | 2 +- lib/kernel/src/gen_tcp.erl | 2 +- lib/kernel/src/gen_udp.erl | 2 +- lib/kernel/src/inet.erl | 14 +++- lib/kernel/src/inet_int.hrl | 1 + lib/kernel/test/gen_sctp_SUITE.erl | 125 ++++++++++++++++++++++++++++++-- lib/kernel/test/gen_tcp_misc_SUITE.erl | 114 ++++++++++++++++++++++++++++- lib/kernel/test/gen_udp_SUITE.erl | 106 ++++++++++++++++++++++++++- 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; @@ -3336,6 +3360,34 @@ static int packet_binary_message return erl_drv_output_term(desc->dport, spec, i); } +/* +** 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} */ @@ -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, [<>|Acc]); +enc_opt_val([{active,N}|Opts], Acc) when is_integer(N), N < 32768, N >= -32768 -> + enc_opt_val(Opts, [<>|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]} -> + <> = 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 - gen_sctp:open/1,2 or inet:setopts/2, - retrieved using inet:getopts/2, and when calling - gen_sctp:connect/4,5 options can be changed.

+ gen_sctp:open/1,2 + or inet:setopts/2, + retrieved using inet:getopts/2, + and when calling gen_sctp:connect/4,5 + options can be changed.

@@ -507,7 +509,7 @@

Determines the type of data returned from gen_sctp:recv/1,2.

- {active, true|false|once} + {active, true|false|once|N} @@ -524,11 +526,28 @@

If once, 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.

 + +

If active is specified as an integer N 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 inet:setopts/2, the socket's + mode is automatically reset to passive ({active, + false}) mode. When a socket in this active mode transitions to + passive mode, the message {sctp_passive, Socket} is sent + to the controlling process to notify it that if it wants to + receive more data messages from the socket, it must call + inet:setopts/2 to set + the socket back into an active mode.

+ {tos, integer()} 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:

{tcp, Socket, Data} +

If the socket is in {active, N} mode (see + inet:setopts/2 for details) and its message counter + drops to 0, the following message is delivered to indicate that the + socket has transitioned to passive ({active, false}) mode:

+ +{tcp_passive, Socket}

If the socket is closed, the following message is delivered:

{tcp_closed, Socket} 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 @@ inet:setopts/2.

 -

The returned socket Socket is used to send packets - from this port with send/4. When UDP packets arrive at - the opened port, they are delivered as messages:

+

The returned socket Socket is used to send + packets from this port with send/4. 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:

{udp, Socket, IP, InPortNo, Packet} -

Note that arriving UDP packets that are longer than +

If the socket is not in an active mode, data can be + retrieved via the recv/2,3 calls. + Note that arriving UDP packets that are longer than the receive buffer option specifies, might be truncated without warning.

+

When a socket in {active, N} mode (see + inet:setopts/2 for details) transitions to passive + ({active, false}) mode, the controlling process is notified by a + message of the following form:

+ +{udp_passive, Socket}

IP and InPortNo define the address from which Packet came. Packet is a list of bytes if the option list was specified. Packet 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

Sets one or more options for a socket. The following options are available:

- {active, true | false | once} + {active, true | false | once | N}

If the value is true, which is the default, everything received from the socket will be sent as messages to the receiving process. If the value is false (passive mode), the process must explicitly - receive incoming data by calling gen_tcp:recv/2,3 - or gen_udp:recv/2,3 (depending on the type of - socket).

+ receive incoming data by calling + gen_tcp:recv/2,3, + gen_udp:recv/2,3 + or gen_sctp:recv/1,2 + (depending on the type of socket).

If the value is once ({active, once}), one data message from the socket will be sent to the process. To receive one more message, setopts/2 must be called again with the {active, once} option.

-

When using {active, once}, the socket changes - behaviour automatically when data is received. This can - sometimes be confusing in combination with connection - oriented sockets (i.e. gen_tcp) as a socket with - {active, false} behaviour reports closing +

If the value is an integer N 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 setopts/2 must be called again to set the + socket back into an active mode.

+

When using {active, once} or {active, N}, the + socket changes behaviour automatically when data is received. + This can sometimes be confusing in combination with + connection-oriented sockets (i.e. gen_tcp) as a socket + with {active, false} behaviour reports closing differently than a socket with {active, true} behaviour. To make programming easier, a socket where the peer closed and this was detected while in {active, false} mode, will still generate the message - {tcp_closed,Socket} when set to {active, once} or {active, true} mode. It is therefore + {tcp_closed,Socket} when set to {active, once}, + {active, true} or {active, N} mode. It is therefore safe to assume that the message {tcp_closed,Socket}, possibly followed by socket port termination (depending on the exit_on_close option) will eventually appear when a socket changes back and forth between {active, true} and - {active, false} mode. However, + {active, false} mode. However, when peer closing is detected is all up to the underlying TCP/IP stack and protocol.

-

Note that {active,true} mode provides no flow +

Note that {active, true} 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 + {active, N} 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. {active,false} - mode or use of the {active, once} mode provides - flow control; the other side will not be able send + amount of data exchanged is small. {active, false} + mode, use of the {active, once} mode or {active, N} + mode with values of N appropriate for the application + provides flow control; the other side will not be able send faster than the receiver can read.

 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 -- cgit v1.2.3