diff options
Diffstat (limited to 'erts')
31 files changed, 6035 insertions, 482 deletions
diff --git a/erts/doc/src/crash_dump.xml b/erts/doc/src/crash_dump.xml index 33d0903622..876834307a 100644 --- a/erts/doc/src/crash_dump.xml +++ b/erts/doc/src/crash_dump.xml @@ -289,7 +289,7 @@ Slogan: <reason></pre> <marker id="memory"></marker> <title>Memory Information</title> <p>Under the tag <em>=memory</em> is shown information similar - to what can be obtainted on a living node with + to what can be obtained on a living node with <seealso marker="erts:erlang#memory/0"> <c>erlang:memory()</c></seealso>.</p> </section> diff --git a/erts/doc/src/driver.xml b/erts/doc/src/driver.xml index 8f31df4cad..a873bf9931 100644 --- a/erts/doc/src/driver.xml +++ b/erts/doc/src/driver.xml @@ -391,7 +391,7 @@ void encode_result(ei_x_buff* x, PGresult* res, PGconn* conn) <p>Before a driver can be called from Erlang, it must be loaded and opened. Loading is done using the <c><![CDATA[erl_ddll]]></c> module (the <c><![CDATA[erl_ddll]]></c> driver that loads dynamic - driver is actually a driver itself). If loading is successfull, + driver is actually a driver itself). If loading is successful, the port can be opened with <c><![CDATA[open_port/2]]></c>. The port name must match the name of the shared library and the name in the driver entry structure.</p> diff --git a/erts/doc/src/erl_ext_dist.xml b/erts/doc/src/erl_ext_dist.xml index 3730f0e8ac..2ba5994557 100644 --- a/erts/doc/src/erl_ext_dist.xml +++ b/erts/doc/src/erl_ext_dist.xml @@ -128,7 +128,7 @@ are deprecated and are only kept for backward compatibility when decoding terms encoded by older nodes.</p> <p>Support for UTF-8 encoded atoms in the external format has been - available since ERTS 5.10 (OTP R16). This abillity allows such old nodes + available since ERTS 5.10 (OTP R16). This ability allows such old nodes to decode, store and encode any Unicode atoms received from a new OTP 20 node.</p> <p>The maximum number of allowed characters in an atom is 255. In the @@ -913,7 +913,7 @@ <tcaption>SMALL_BIG_EXT</tcaption></table> <p> Bignums are stored in unary form with a <c>Sign</c> byte, - that is, 0 if the binum is positive and 1 if it is negative. The + that is, 0 if the bignum is positive and 1 if it is negative. The digits are stored with the least significant byte stored first. To calculate the integer, the following formula can be used: </p> diff --git a/erts/doc/src/erlang.xml b/erts/doc/src/erlang.xml index f41ed87048..0e82ceba7d 100644 --- a/erts/doc/src/erlang.xml +++ b/erts/doc/src/erlang.xml @@ -10129,6 +10129,11 @@ timestamp() -> time stamp flags are remembered, so if two are passed and the one with highest precedence later is disabled, the other one becomes active.</p> + <p>If a match specification (applicable only for <c>call</c>, <c>send</c> + and <c>'receive'</c> tracing) contains a <c>{message}</c> action + function with a non-boolean value, that value is added as an extra + element to the message tuple either in the last position or before + the timestamp (if it is present).</p> <p>Trace messages:</p> <marker id="trace_3_trace_messages"></marker> <taglist> diff --git a/erts/doc/src/notes.xml b/erts/doc/src/notes.xml index 51c1ecc8b1..cfa952f01c 100644 --- a/erts/doc/src/notes.xml +++ b/erts/doc/src/notes.xml @@ -31,6 +31,84 @@ </header> <p>This document describes the changes made to the ERTS application.</p> +<section><title>Erts 10.4.3</title> + + <section><title>Fixed Bugs and Malfunctions</title> + <list> + <item> + <p>Fixed a buffer overflow when + <c>binary_to_existing_atom/2</c> and + <c>list_to_existing_atom/2</c> was used with the + <c>latin1</c> encoding.</p> + <p> + Own Id: OTP-15819 Aux Id: ERL-944 </p> + </item> + <item> + <p> + The runtime system disconnected a connection if it + received an <c>exit/2</c> signal where the recipient was + a process on an old incarnation of the current node. That + is, the receiving node had the same node name, but + another "creation" number. The signal will now just be + dropped since the receiving process no longer exists.</p> + <p> + Own Id: OTP-15867 Aux Id: ERIERL-373 </p> + </item> + </list> + </section> + +</section> + +<section><title>Erts 10.4.2</title> + + <section><title>Fixed Bugs and Malfunctions</title> + <list> + <item> + <p> + Fixed <c>process_info(Pid,reductions)</c> to not + categorically increase reduction count of the measured + process <c>Pid</c>. Repeated reduction measure of an idle + process will most often (but not guaranteed) return the + same value, like it behaved before OTP 21.3.8.</p> + <p> + Own Id: OTP-15865 Aux Id: ERL-964 </p> + </item> + <item> + <p>Fixed an incorrect load-time optimization that could + cause a crash when extracting deeply nested tuple + elements.</p> + <p> + Own Id: OTP-15871 Aux Id: ERIERL-374 </p> + </item> + <item> + <p> + Fix bug causing VM crash when pressing P for "proc info" + in Erlang shell break menu. Bug exists since OTP 22.0.</p> + <p> + Own Id: OTP-15873 Aux Id: ERL-965 </p> + </item> + </list> + </section> + +</section> + +<section><title>Erts 10.4.1</title> + + <section><title>Fixed Bugs and Malfunctions</title> + <list> + <item> + <p>In nested use of <c>try</c>/<c>catch</c>, rethrowing + an exception using <c>erlang:raise/3</c> with a different + class would not always be able to change the class of the + exception.</p> + <p> + Own Id: OTP-15834 Aux Id: ERIERL-367 </p> + </item> + </list> + </section> + +</section> + <section><title>Erts 10.4</title> <section><title>Fixed Bugs and Malfunctions</title> @@ -514,6 +592,95 @@ </section> +<section><title>Erts 10.3.5.3</title> + + <section><title>Fixed Bugs and Malfunctions</title> + <list> + <item> + <p> + If you set <c>{linger,{true,0}}</c> on a <c>gen_tcp</c> + listen socket, accept a connection on that socket, and + then close the accepted socket, now the linger zero + setting is transferred to the accepted socket. Before + this correction that information was lost and the close + behaviour on the accepted socket incorrect.</p> + <p> + Own Id: OTP-15370 Aux Id: ERIERL-353 </p> + </item> + <item> + <p> + Fixed <c>process_info(Pid,reductions)</c> to not + categorically increase reduction count of the measured + process <c>Pid</c>. Repeated reduction measure of an idle + process will most often (but not guaranteed) return the + same value, like it behaved before OTP 21.3.8.</p> + <p> + Own Id: OTP-15865 Aux Id: ERL-964 </p> + </item> + <item> + <p> + The runtime system disconnected a connection if it + received an <c>exit/2</c> signal where the recipient was + a process on an old incarnation of the current node. That + is, the receiving node had the same node name, but + another "creation" number. The signal will now just be + dropped since the receiving process no longer exists.</p> + <p> + Own Id: OTP-15867 Aux Id: ERIERL-373 </p> + </item> + </list> + </section> + + + <section><title>Improvements and New Features</title> + <list> + <item> + <p> + The possibility to send ancillary data, in particular the + TOS field, has been added to <c>gen_udp:send/4,5</c>.</p> + <p> + Own Id: OTP-15747 Aux Id: ERIERL-294 </p> + </item> + </list> + </section> + +</section> + +<section><title>Erts 10.3.5.2</title> + + <section><title>Fixed Bugs and Malfunctions</title> + <list> + <item> + <p>In nested use of <c>try</c>/<c>catch</c>, rethrowing + an exception using <c>erlang:raise/3</c> with a different + class would not always be able to change the class of the + exception.</p> + <p> + Own Id: OTP-15834 Aux Id: ERIERL-367 </p> + </item> + <item> + <p> + Fixed bug in <c>seq_trace:set_token(label,Term)</c> which + could cause VM crash if <c>Term</c> was heap allocated + (not an atom, small integer, local pid or port). Bug + exists since OTP 21.0 when terms other than small + integers were first allowed as labels.</p> + <p> + Own Id: OTP-15849 Aux Id: ERL-700 </p> + </item> + <item> + <p> + Fix <c>seq_trace:print/2</c> not to raise <c>badarg</c> + exception if label is not a small integer. Bug exists + since OTP 21.0.</p> + <p> + Own Id: OTP-15859 Aux Id: ERL-700 </p> + </item> + </list> + </section> + +</section> + <section><title>Erts 10.3.5.1</title> <section><title>Fixed Bugs and Malfunctions</title> @@ -2540,6 +2707,23 @@ </section> +<section><title>Erts 9.3.3.11</title> + + <section><title>Fixed Bugs and Malfunctions</title> + <list> + <item> + <p>Fixed a buffer overflow when + <c>binary_to_existing_atom/2</c> and + <c>list_to_existing_atom/2</c> was used with the + <c>latin1</c> encoding.</p> + <p> + Own Id: OTP-15819 Aux Id: ERL-944 </p> + </item> + </list> + </section> + +</section> + <section><title>Erts 9.3.3.10</title> <section><title>Fixed Bugs and Malfunctions</title> diff --git a/erts/doc/src/socket.xml b/erts/doc/src/socket.xml index 9b34bf1df1..b4e22e86a8 100644 --- a/erts/doc/src/socket.xml +++ b/erts/doc/src/socket.xml @@ -41,6 +41,38 @@ e.g. <seealso marker="#recv/3"><c>recv/3</c></seealso>, has a timeout argument. </p> <note> + <p>Some functions allow for an <i>asynchronous</i> call. + This is achieved by setting the <c>Timeout</c> argument to + <c>nowait</c>. For instance, if calling the + <seealso marker="#recv_async"><c>recv/3</c></seealso> + function with Timeout set to <c>nowait</c> (<c>recv(Sock, 0, nowait)</c>) + when there is actually nothing to read, it will return with + <c>{ok, </c> + <seealso marker="#type-select_info"><c>SelectInfo</c></seealso><c>}</c>. + When data eventually arrives a 'select' message + will be sent to the caller: </p> + <taglist> + <!-- NOTE THAT THE EMPTY TAG IS INTENTIONAL --> + <tag></tag> + <item><c>{'$socket', socket(), select, select_ref()}</c></item> + </taglist> + <p>The caller can now make another + call to the recv function and now expect data.</p> + <p>Note that all other users are <em>locked out</em> until the + 'current user' has called the function (recv in this case).</p> + <p>Another message the user must be prepared for (when making asynchronous + calls) is the <c>abort</c> message:</p> + <taglist> + <!-- NOTE THAT THE EMPTY TAG IS INTENTIONAL --> + <tag></tag> + <item><c>{'$socket', socket(), abort, Info}</c></item> + </taglist> + <p>This message indicates + that the (asynchronous) operation has been aborted. + If, for instance, the socket has been closed (by another process), + <c>Info</c> will be <c>{select_ref(), closed}</c>. </p> + </note> + <note> <p>There is currently <em>no</em> support for Windows. </p> <p>Support for IPv6 has been implemented but <em>not</em> tested. </p> <p>SCTP has only been partly implemented (and not tested). </p> @@ -65,6 +97,21 @@ </desc> </datatype> <datatype> + <name name="select_tag"/> + <desc> + <p>A tag that describes the (select) operation.</p> + </desc> + </datatype> + <datatype> + <name name="select_ref"/> + <desc> + <p>A reference that uniquely identifies the (select) operation.</p> + </desc> + </datatype> + <datatype> + <name name="select_info"/> + </datatype> + <datatype> <name name="ip4_address"/> </datatype> <datatype> @@ -257,7 +304,7 @@ <funcs> <func> <name name="accept" arity="1" since="OTP 22.0"/> - <name name="accept" arity="2" since="OTP 22.0"/> + <name name="accept" arity="2" clause_i="2" since="OTP 22.0"/> <fsummary>Accept a connection on a socket.</fsummary> <desc> <p>Accept a connection on a socket.</p> @@ -269,6 +316,26 @@ </func> <func> + <name name="accept" arity="2" clause_i="1" anchor="accept_async" since="OTP @OTP-15731@"/> + <fsummary>Accept a connection on a socket.</fsummary> + <desc> + <p>Accept a connection on a socket.</p> + + <p>This call is used with connection-based socket types + (<c>stream</c> or <c>seqpacket</c>). It extracs the first pending + connection request for the listen socket and returns the (newly) + connected socket.</p> + + <p>In the case when there is no connections waiting, the function + will return with the <c>SelectInfo</c>. The caller can then await a + select message, <c>{'$socket', Socket, select, Info}</c> (where + <c>Info</c> is the <c>select_ref()</c> from the <c>SelectInfo</c>), + when a client connects (a subsequent call to accept will then return + the socket). </p> + </desc> + </func> + + <func> <name name="bind" arity="2" since="OTP 22.0"/> <fsummary>Bind a name to a socket.</fsummary> <desc> @@ -282,6 +349,18 @@ </func> <func> + <name name="cancel" arity="2" since="OTP @OTP-15731@"/> + <fsummary>Cancel an asynchronous request.</fsummary> + <desc> + <p>Cancel an asynchronous request.</p> + + <p>Call this function in order to cancel a previous + asynchronous call to, e.g. + <seealso marker="#recv/3"><c>recv/3</c></seealso>. </p> + </desc> + </func> + + <func> <name name="close" arity="1" since="OTP 22.0"/> <fsummary>Close a socket.</fsummary> <desc> @@ -302,7 +381,7 @@ <func> <name name="connect" arity="2" since="OTP 22.0"/> - <name name="connect" arity="3" since="OTP 22.0"/> + <name name="connect" arity="3" clause_i="2" since="OTP 22.0"/> <fsummary>Initiate a connection on a socket.</fsummary> <desc> <p>This function connects the socket to the address @@ -311,6 +390,24 @@ </func> <func> + <name name="connect" arity="3" clause_i="1" anchor="connect_async" since="OTP @OTP-15731@"/> + <fsummary>Initiate a connection on a socket.</fsummary> + <desc> + <p>This function connects the socket to the address + specied by the <c>SockAddr</c> argument.</p> + + <p>In the case when its not possible to immediately establish a + connection, the function will return with the + <seealso marker="#type-select_info"><c>SelectInfo</c></seealso>. + The caller can then await a + select message, <c>{'$socket', Socket, select, Info}</c> (where + <c>Info</c> is the <c>select_ref()</c> from the <c>SelectInfo</c>, + a subsequent call to connect will then + establish the connection). </p> + </desc> + </func> + + <func> <name name="getopt" arity="3" clause_i="1" since="OTP 22.0"/> <name name="getopt" arity="3" clause_i="2" since="OTP 22.0"/> <name name="getopt" arity="3" clause_i="3" since="OTP 22.0"/> @@ -406,8 +503,8 @@ <name name="recv" arity="1" since="OTP 22.0"/> <name name="recv" arity="2" since="OTP 22.0"/> <name name="recv" arity="3" clause_i="1" since="OTP 22.0"/> - <name name="recv" arity="3" clause_i="2" since="OTP 22.0"/> - <name name="recv" arity="4" since="OTP 22.0"/> + <name name="recv" arity="3" clause_i="3" since="OTP 22.0"/> + <name name="recv" arity="4" clause_i="2" since="OTP 22.0"/> <fsummary>Receive a message from a socket.</fsummary> <desc> <p>Receive a message from a socket.</p> @@ -418,36 +515,91 @@ </func> <func> + <name name="recv" arity="3" clause_i="2" anchor="recv_async" since="OTP @OTP-15731@"/> + <name name="recv" arity="4" clause_i="1" since="OTP @OTP-15731@"/> + <fsummary>Receive a message from a socket.</fsummary> + <desc> + <p>Receive a message from a socket.</p> + + <p>There is a special case for the argument <c>Length</c>. + If it is set to zero (0), it means "give me everything you + currently have".</p> + + <p>In the case when there is no data waiting, the function + will return with the <c>SelectInfo</c>. The caller can then await a + select message, <c>{'$socket', Socket, select, Info}</c> (where + <c>Info</c> is the <c>select_ref()</c> from the <c>SelectInfo</c>), + when data has arrived (a subsequent call to recv will then return + the data). </p> + <p>Note that if a length (<c>> 0</c>) is specified, and only part + of that amount of data is available, the function will return with + that data <em>and</em> the <c>SelectInfo</c> (if the caller don't + want to wait for the remaining data, it must immediately call + the <seealso marker="#cancel/2"><c>cancel/2</c></seealso> function.)</p> + </desc> + </func> + + <func> <name name="recvfrom" arity="1" since="OTP 22.0"/> <name name="recvfrom" arity="2" since="OTP 22.0"/> - <name name="recvfrom" arity="3" clause_i="1" since="OTP 22.0"/> <name name="recvfrom" arity="3" clause_i="2" since="OTP 22.0"/> <name name="recvfrom" arity="3" clause_i="3" since="OTP 22.0"/> - <name name="recvfrom" arity="4" since="OTP 22.0"/> + <name name="recvfrom" arity="3" clause_i="5" since="OTP 22.0"/> + <name name="recvfrom" arity="4" clause_i="2" since="OTP 22.0"/> <fsummary>Receive a message from a socket.</fsummary> <desc> <p>Receive a message from a socket.</p> <p>This function reads "messages", which means that regardless of - how much we want to read, it returns when we get a message.</p> + how much we want to read, it returns when we get a message + (if the buffer size is to small, the message will be truncated).</p> + <p>The <c>BufSz</c> argument basically defines the size of the + receive buffer. By setting the value to zero (0), the configured + size (setopt with <c>Level</c> = <c>otp</c> and + <c>Key</c> = <c>rcvbuf</c>) is used.</p> + <p>It may be impossible to know what (buffer) size is appropriate + "in advance", and in those cases it may be convenient to use the + (recv) 'peek' flag. When this flag is provided, the message is *not* + "consumed" from the underlying buffers, so another recvfrom call + is needed, possibly with a then adjusted buffer size.</p> + </desc> + </func> + + <func> + <name name="recvfrom" arity="3" clause_i="1" anchor="recvfrom_async" since="OTP @OTP-15731@"/> + <name name="recvfrom" arity="3" clause_i="4" since="OTP @OTP-15731@"/> + <name name="recvfrom" arity="4" clause_i="1" since="OTP @OTP-15731@"/> + <fsummary>Receive a message from a socket.</fsummary> + <desc> + <p>Receive a message from a socket.</p> + <p>This function reads "messages", which means that regardless of + how much we want to read, it returns when we get a message + (if the buffer size is to small, the message will be truncated).</p> <p>The <c>BufSz</c> argument basically defines the size of the receive buffer. By setting the value to zero (0), the configured - size (setopt with <c>Level</c> = <c>otp</c> and <c>Key</c> = <c>rcvbuf</c>) - is used.</p> + size (setopt with <c>Level</c> = <c>otp</c> and + <c>Key</c> = <c>rcvbuf</c>) is used.</p> <p>It may be impossible to know what (buffer) size is appropriate "in advance", and in those cases it may be convenient to use the (recv) 'peek' flag. When this flag is provided, the message is *not* "consumed" from the underlying buffers, so another recvfrom call is needed, possibly with a then adjusted buffer size.</p> + + <p>In the case when there is no data waiting, the function + will return with the <c>SelectInfo</c>. The caller can then await a + select message, <c>{'$socket', Socket, select, Info}</c> (where + <c>Info</c> is the <c>select_ref()</c> from the <c>SelectInfo</c>), + when data has arrived (a subsequent call to recvfrom will then return + the data). </p> </desc> </func> <func> <name name="recvmsg" arity="1" since="OTP 22.0"/> <name name="recvmsg" arity="2" clause_i="1" since="OTP 22.0"/> - <name name="recvmsg" arity="2" clause_i="2" since="OTP 22.0"/> - <name name="recvmsg" arity="3" clause_i="1" since="OTP 22.0"/> + <name name="recvmsg" arity="2" clause_i="3" since="OTP 22.0"/> <name name="recvmsg" arity="3" clause_i="2" since="OTP 22.0"/> - <name name="recvmsg" arity="5" since="OTP 22.0"/> + <name name="recvmsg" arity="3" clause_i="3" since="OTP 22.0"/> + <name name="recvmsg" arity="5" clause_i="2" since="OTP 22.0"/> <fsummary>Receive a message from a socket.</fsummary> <desc> <p>Receive a message from a socket.</p> @@ -461,8 +613,8 @@ <p>The <c>BufSz</c> argument basically defines the size of the receive buffer. By setting the value to zero (0), the configured - size (setopt with <c>Level</c> = <c>otp</c> and <c>Key</c> = <c>rcvbuf</c>) - is used.</p> + size (setopt with <c>Level</c> = <c>otp</c> and + <c>Key</c> = <c>rcvbuf</c>) is used.</p> <p>The <c>CtrlSz</c> argument basically defines the size of the receive buffer for the control messages. @@ -478,21 +630,83 @@ </func> <func> + <name name="recvmsg" arity="2" clause_i="2" anchor="recvmsg_async" since="OTP @OTP-15731@"/> + <name name="recvmsg" arity="3" clause_i="1" since="OTP @OTP-15731@"/> + <name name="recvmsg" arity="5" clause_i="1" since="OTP @OTP-15731@"/> + <fsummary>Receive a message from a socket.</fsummary> + <desc> + <p>Receive a message from a socket.</p> + <p>This function reads "messages", which means that regardless of + how much we want to read, it returns when we get a message.</p> + <p>The message will be delivered in the form of a <c>msghdr()</c>, + which may contain the source address (if socket not connected), + a list of <c>cmsghdr_recv()</c> (depends on what socket options have + been set and what the protocol and platform supports) and + also a set of flags, providing further info about the read. </p> + + <p>The <c>BufSz</c> argument basically defines the size of the + receive buffer. By setting the value to zero (0), the configured + size (setopt with <c>Level</c> = <c>otp</c> and + <c>Key</c> = <c>rcvbuf</c>) is used.</p> + + <p>The <c>CtrlSz</c> argument basically defines the size of the + receive buffer for the control messages. + By setting the value to zero (0), the configured size (setopt + with <c>Level</c> = <c>otp</c>) is used.</p> + + <p>It may be impossible to know what (buffer) size is appropriate + "in advance", and in those cases it may be convenient to use the + (recv) 'peek' flag. When this flag is provided, the message is *not* + "consumed" from the underlying buffers, so another recvmsg call + is needed, possibly with a then adjusted buffer size.</p> + + <p>In the case when there is no data waiting, the function + will return with the <c>SelectInfo</c>. The caller can then await a + select message, <c>{'$socket', Socket, select, Info}</c> (where + <c>Info</c> is the <c>select_ref()</c> from the <c>SelectInfo</c>), + when data has arrived (a subsequent call to recvmsg will then return + the data). </p> + </desc> + </func> + + <func> <name name="send" arity="2" since="OTP 22.0"/> <name name="send" arity="3" clause_i="1" since="OTP 22.0"/> - <name name="send" arity="3" clause_i="2" since="OTP 22.0"/> - <name name="send" arity="4" since="OTP 22.0"/> + <name name="send" arity="3" clause_i="3" since="OTP 22.0"/> + <name name="send" arity="4" clause_i="2" since="OTP 22.0"/> + <fsummary>Send a message on a socket.</fsummary> + <desc> + <p>Send a message on a connected socket.</p> + </desc> + </func> + + <func> + <name name="send" arity="3" clause_i="2" anchor="send_async" since="OTP @OTP-15731@"/> + <name name="send" arity="4" clause_i="1" since="OTP @OTP-15731@"/> <fsummary>Send a message on a socket.</fsummary> <desc> <p>Send a message on a connected socket.</p> + + <p>In the case when there is no room in the (system-) buffers, + the function will return with the <c>SelectInfo</c>. The caller + can then await a select message, + <c>{'$socket', Socket, select, Info}</c> + (where <c>Info</c> is the <c>select_ref()</c> from the + <c>SelectInfo</c>), when there is room for more data (a subsequent + call to send will then send the data). </p> + <p>Note that if not all the data was sent, the function will return + with the remaining data <em>and</em> the <c>SelectInfo</c> + (if the caller don't + want to wait to be able to send the rest, it should immediately call + the <seealso marker="#cancel/2"><c>cancel/2</c></seealso> function.)</p> </desc> </func> <func> <name name="sendmsg" arity="2" since="OTP 22.0"/> <name name="sendmsg" arity="3" clause_i="1" since="OTP 22.0"/> - <name name="sendmsg" arity="3" clause_i="2" since="OTP 22.0"/> - <name name="sendmsg" arity="4" since="OTP 22.0"/> + <name name="sendmsg" arity="3" clause_i="3" since="OTP 22.0"/> + <name name="sendmsg" arity="4" clause_i="2" since="OTP 22.0"/> <fsummary>Send a message on a socket.</fsummary> <desc> <p>Send a message on a socket. The destination, if needed @@ -513,10 +727,39 @@ </func> <func> + <name name="sendmsg" arity="3" clause_i="2" anchor="sendmsg_async" since="OTP @OTP-15731@"/> + <name name="sendmsg" arity="4" clause_i="1" since="OTP @OTP-15731@"/> + <fsummary>Send a message on a socket.</fsummary> + <desc> + <p>Send a message on a socket. The destination, if needed + (socket <em>not</em> connected) is provided in the <c>MsgHdr</c>, + which also contains the message to send, + The <c>MsgHdr</c> may also contain an list of optional <c>cmsghdr_send()</c> + (depends on what the protocol and platform supports).</p> + + <p>Unlike the <seealso marker="#send/2"><c>send</c></seealso> function, + this one sends <em>one message</em>. + This means that if, for whatever reason, its not possible to send the + message in one go, the function will instead return with the + <em>remaining</em> data (<c>{ok, Remaining}</c>). Thereby leaving it + up to the caller to decide what to do (retry with the remaining data + of give up). </p> + + <p>In the case when there is no room in the (system-) buffers, + the function will return with the <c>SelectInfo</c>. The caller + can then await a select message, + <c>{'$socket', Socket, select, Info}</c> + (where <c>Info</c> is the <c>select_ref()</c> from the + <c>SelectInfo</c>), when there is room for more data (a subsequent + call to sendmsg will then send the data). </p> + </desc> + </func> + + <func> <name name="sendto" arity="3" since="OTP 22.0"/> <name name="sendto" arity="4" clause_i="1" since="OTP 22.0"/> - <name name="sendto" arity="4" clause_i="2" since="OTP 22.0"/> - <name name="sendto" arity="5" since="OTP 22.0"/> + <name name="sendto" arity="4" clause_i="3" since="OTP 22.0"/> + <name name="sendto" arity="5" clause_i="2" since="OTP 22.0"/> <fsummary>Send a message on a socket.</fsummary> <desc> <p>Send a message on a socket, to the specified destination.</p> @@ -524,6 +767,23 @@ </func> <func> + <name name="sendto" arity="4" clause_i="2" anchor="sendto_async" since="OTP @OTP-15731@"/> + <name name="sendto" arity="5" clause_i="1" since="OTP @OTP-15731@"/> + <fsummary>Send a message on a socket.</fsummary> + <desc> + <p>Send a message on a socket, to the specified destination.</p> + + <p>In the case when there is no room in the (system-) buffers, + the function will return with the <c>SelectInfo</c>. The caller + can then await a select message, + <c>{'$socket', Socket, select, Info}</c> + (where <c>Info</c> is the <c>select_ref()</c> from the + <c>SelectInfo</c>), when there is room for more data (a subsequent + call to sendto will then send the data). </p> + </desc> + </func> + + <func> <name name="setopt" arity="4" clause_i="1" since="OTP 22.0"/> <name name="setopt" arity="4" clause_i="2" since="OTP 22.0"/> <name name="setopt" arity="4" clause_i="3" since="OTP 22.0"/> diff --git a/erts/doc/src/socket_usage.xml b/erts/doc/src/socket_usage.xml index 4b3872d7e3..7e65bcbf70 100644 --- a/erts/doc/src/socket_usage.xml +++ b/erts/doc/src/socket_usage.xml @@ -34,11 +34,84 @@ <title>Introduction</title> <p>The socket interface (module) is basically an "thin" layer on top of the OS socket interface. It is assumed that, unless you have special needs, - gen_[tcp|udp|sctp] should be sufficent. </p> + gen_[tcp|udp|sctp] should be sufficent (when they become available). </p> <p>Note that just because we have a documented and described option, it does <em>not</em> mean that the OS supports it. So its recommended that the user reads the platform specific documentation for the option used. </p> + <section> + <title>Asynchronous calls</title> + <p>Some functions allow for an <i>asynchronous</i> call + (<seealso marker="socket#accept_async"><c>accept/2</c></seealso>, + <seealso marker="socket#connect_async"><c>connect/3</c></seealso>, + <seealso marker="socket#recv_async"><c>recv/3,4</c></seealso>, + <seealso marker="socket#recvfrom_async"><c>recvfrom/3,4</c></seealso>, + <seealso marker="socket#recvmsg_async"><c>recvmsg/2,3,5</c></seealso>, + <seealso marker="socket#send_async"><c>send/3,4</c></seealso>, + <seealso marker="socket#sendmsg_async"><c>sendmsg/3,4</c></seealso> and + <seealso marker="socket#sendto_async"><c>sendto/4,5</c></seealso>). + This is achieved by setting the <c>Timeout</c> argument to + <c>nowait</c>. For instance, if calling the + <seealso marker="socket#recv_async"><c>recv/3</c></seealso> + function with Timeout set to <c>nowait</c> (i.e. + <c>recv(Sock, 0, nowait)</c>) + when there is actually nothing to read, it will return with + <c>{ok, </c> + <seealso marker="socket#type-select_info"><c>SelectInfo</c></seealso><c>}</c>. + When data eventually arrives a 'select message' + will be sent to the caller:</p> + <taglist> + <!-- NOTE THAT THE EMPTY TAG IS INTENTIONAL --> + <tag></tag> + <item><c>{'$socket', socket(), select, select_ref()}</c></item> + </taglist> + <p>The caller can then make another + call to the recv function and now expect data.</p> + <p>The user must also be prepared to receive an abort message: </p> + <!-- + <list type="ordered"> + <item><c>{'$socket', Sock, abort, Info}</c></item> + </list> + --> + <taglist> + <!-- NOTE THAT THE EMPTY TAG IS INTENTIONAL --> + <tag></tag> + <item><c>{'$socket', socket(), abort, Info}</c></item> + </taglist> + <p>If the operation is aborted + for whatever reason (e.g. if the socket is closed "by someone else"). + The <c>Info</c> part contains the abort reason (in this case that + the socket has been closed <c>Info = {select_ref(), closed}</c>). </p> + <p>Note that all other users are <em>locked out</em> until the + 'current user' has called the function (recv in this case). So either + immediately call the function or + <seealso marker="socket#cancel/2"><c>cancel</c></seealso>. </p> + + <p>The general form of the 'socket' message is: </p> + <taglist> + <!-- NOTE THAT THE EMPTY TAG IS INTENTIONAL --> + <tag></tag> + <item><c>{'$socket', Sock :: socket(), Tag :: atom(), Info :: term()}</c></item> + </taglist> + <p>Where the format of <c>Info</c> is a function of <c>Tag</c>:</p> + <table> + <row> + <cell><em>Tag</em></cell> + <cell><em>Info value type</em></cell> + </row> + <row> + <cell>select</cell> + <cell>select_ref()</cell> + </row> + <row> + <cell>abort</cell> + <cell>{select_ref(), Reason :: term()}</cell> + </row> + <tcaption>socket message info value type</tcaption> + </table> + <p>The <c>select_ref()</c> is the same as was received in the + <seealso marker="socket#type-select_info"><c>SelectInfo</c></seealso>. </p> + </section> </section> <section> diff --git a/erts/doc/src/zlib.xml b/erts/doc/src/zlib.xml index 38229456c9..8a580c8d7b 100644 --- a/erts/doc/src/zlib.xml +++ b/erts/doc/src/zlib.xml @@ -411,7 +411,7 @@ list_to_binary([B1,B2])</pre> <func> <name name="deflateParams" arity="3" since=""/> - <fsummary>Dynamicly update deflate parameters.</fsummary> + <fsummary>Dynamically update deflate parameters.</fsummary> <desc> <p>Dynamically updates the compression level and compression strategy. The interpretation of <c><anno>Level</anno></c> and diff --git a/erts/emulator/beam/break.c b/erts/emulator/beam/break.c index 80e871aaf6..6379e4e04d 100644 --- a/erts/emulator/beam/break.c +++ b/erts/emulator/beam/break.c @@ -84,8 +84,9 @@ process_info(fmtfn_t to, void *to_arg) * they are most likely just created and has invalid data */ if (p->heap != NULL) { - ErtsProcLocks locks = (p == esdp->current_process || - p == esdp->free_process) ? ERTS_PROC_LOCK_MAIN : 0; + ErtsProcLocks locks = ((esdp && (p == esdp->current_process || + p == esdp->free_process)) + ? ERTS_PROC_LOCK_MAIN : 0); print_process_info(to, to_arg, p, locks); } } diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c index ff19ef018e..5e48a553af 100644 --- a/erts/emulator/beam/dist.c +++ b/erts/emulator/beam/dist.c @@ -2055,10 +2055,17 @@ int erts_net_message(Port *prt, token = tuple[4]; } if (is_not_pid(from) - || dep != external_pid_dist_entry(from) - || is_not_internal_pid(to)) { + || dep != external_pid_dist_entry(from)) { goto invalid_message; } + if (is_not_internal_pid(to)) { + if (is_external_pid(to)) { + DistEntry *dep = external_pid_dist_entry(to); + if (dep == erts_this_dist_entry) + break; /* Old incarnation of this node... */ + } + goto invalid_message; + } if (!erts_proc_lookup(to)) { if (ede_hfrag != NULL) { diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c index 2704b99aa4..0339589b79 100644 --- a/erts/emulator/beam/erl_bif_info.c +++ b/erts/emulator/beam/erl_bif_info.c @@ -768,7 +768,7 @@ static ErtsProcessInfoArgs pi_args[] = { {am_memory, 0, ERTS_PI_FLAG_NEED_MSGQ_LEN|ERTS_PI_FLAG_FORCE_SIG_SEND, ERTS_PROC_LOCK_MAIN}, {am_garbage_collection, 3+2 + 3+2 + 3+2 + 3+2 + 3+2 + ERTS_MAX_HEAP_SIZE_MAP_SZ, 0, ERTS_PROC_LOCK_MAIN}, {am_group_leader, 0, 0, ERTS_PROC_LOCK_MAIN}, - {am_reductions, 0, ERTS_PI_FLAG_FORCE_SIG_SEND, ERTS_PROC_LOCK_MAIN}, + {am_reductions, 0, 0, ERTS_PROC_LOCK_MAIN}, {am_priority, 0, 0, 0}, {am_trace, 0, 0, ERTS_PROC_LOCK_MAIN}, {am_binary, 0, ERTS_PI_FLAG_FORCE_SIG_SEND, ERTS_PROC_LOCK_MAIN}, diff --git a/erts/emulator/beam/erl_trace.c b/erts/emulator/beam/erl_trace.c index c85a7df5ec..9c835ac357 100644 --- a/erts/emulator/beam/erl_trace.c +++ b/erts/emulator/beam/erl_trace.c @@ -713,7 +713,9 @@ trace_sched(Process *p, ErtsProcLocks locks, Eterm what) trace_sched_aux(p, locks, what); } -/* Send {trace_ts, Pid, Send, Msg, DestPid, Timestamp} +/* Send {trace_ts, Pid, Send, Msg, DestPid, PamResult, Timestamp} + * or {trace_ts, Pid, Send, Msg, DestPid, Timestamp} + * or {trace, Pid, Send, Msg, DestPid, PamResult} * or {trace, Pid, Send, Msg, DestPid} * * where 'Send' is 'send' or 'send_to_non_existing_process'. @@ -773,7 +775,9 @@ trace_send(Process *p, Eterm to, Eterm msg) erts_match_set_release_result_trace(p, pam_result); } -/* Send {trace_ts, Pid, receive, Msg, Timestamp} +/* Send {trace_ts, Pid, receive, Msg, PamResult, Timestamp} + * or {trace_ts, Pid, receive, Msg, Timestamp} + * or {trace, Pid, receive, Msg, PamResult} * or {trace, Pid, receive, Msg} */ void diff --git a/erts/emulator/beam/ops.tab b/erts/emulator/beam/ops.tab index 10ca74cd60..9d68a4a68b 100644 --- a/erts/emulator/beam/ops.tab +++ b/erts/emulator/beam/ops.tab @@ -1,7 +1,7 @@ # # %CopyrightBegin% # -# Copyright Ericsson AB 1997-2018. All Rights Reserved. +# Copyright Ericsson AB 1997-2019. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -699,13 +699,18 @@ is_tuple NotTupleFail Tuple=x | is_tagged_tuple WrongRecordFail Tuple Arity Atom is_tagged_tuple_ff f? f? rx A a -get_tuple_element Reg=x P1 D1=x | get_tuple_element Reg=x P2 D2=x | \ +get_tuple_element Reg=x P1 D1=x | \ + get_tuple_element Reg=x P2 D2=x | \ get_tuple_element Reg=x P3 D3=x | \ - succ(P1, P2) | succ(P2, P3) | \ - succ(D1, D2) | succ(D2, D3) => i_get_tuple_element3 Reg P1 D1 - -get_tuple_element Reg=x P1 D1=x | get_tuple_element Reg=x P2 D2=x | \ - succ(P1, P2) | succ(D1, D2) => i_get_tuple_element2 Reg P1 D1 + succ(P1, P2) | succ(P2, P3) | succ(D1, D2) | succ(D2, D3) | \ + distinct(D1, Reg) | distinct(D2, Reg) => \ + i_get_tuple_element3 Reg P1 D1 + +get_tuple_element Reg=x P1 D1=x | \ + get_tuple_element Reg=x P2 D2=x | \ + succ(P1, P2) | succ(D1, D2) | \ + distinct(D1, Reg) => \ + i_get_tuple_element2 Reg P1 D1 get_tuple_element Reg=x P1 D1=x | get_tuple_element Reg=x P2 D2=x | \ succ(P1, P2) | distinct(D1, Reg) => i_get_tuple_element2_dst Reg P1 D1 D2 diff --git a/erts/emulator/drivers/common/inet_drv.c b/erts/emulator/drivers/common/inet_drv.c index c93966d24f..311c5fdd6a 100644 --- a/erts/emulator/drivers/common/inet_drv.c +++ b/erts/emulator/drivers/common/inet_drv.c @@ -1,7 +1,7 @@ /* * %CopyrightBegin% * - * Copyright Ericsson AB 1997-2018. All Rights Reserved. + * Copyright Ericsson AB 1997-2019. All Rights Reserved. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -1077,6 +1077,7 @@ typedef struct { long forced_events; /* Mask of events that are forcefully signalled on windows see winsock_event_select for details */ + int err; /* Keeps error code for closed socket */ int send_would_block; /* Last send attempt failed with "WOULDBLOCK" */ #endif ErlDrvPort port; /* the port identifier */ @@ -3379,6 +3380,71 @@ static int udp_parse_ancillary_data(ErlDrvTermData *spec, int i, i = LOAD_NIL(spec, i); return LOAD_LIST(spec, i, n+1); } + +static int compile_ancillary_data(struct msghdr *mhdr, + char *ptr, ErlDrvSizeT anc_len) { + struct cmsghdr *cmsg; + size_t controllen = 0; + cmsg = CMSG_FIRSTHDR(mhdr); + for (;;) { + if (anc_len == 0) { + /* End of options to compile */ + mhdr->msg_controllen = controllen; + return 0; + } + if (cmsg == NULL) { + /* End of destination before end of options */ + return 1; + } + +#define COMPILE_ANCILLARY_DATA_ITEM(Level, Opt, Type, Get, Size) \ + do { \ + if (anc_len < (Size)) return 1; \ + sys_memset(cmsg, '\0', CMSG_SPACE(sizeof(Type))); \ + cmsg->cmsg_level = Level; \ + cmsg->cmsg_type = Opt; \ + cmsg->cmsg_len = CMSG_LEN(sizeof(Type)); \ + *((Type *) CMSG_DATA(cmsg)) = Get(ptr); \ + controllen += CMSG_SPACE(sizeof(Type)); \ + cmsg = CMSG_NXTHDR(mhdr, cmsg); \ + ptr += 4; \ + anc_len -= 4; \ + } while (0) +#define SIZEOF_ANCILLARY_DATA (2 * CMSG_SPACE(sizeof(int))) + /* (IP_TOS | IPV6_TCLASS) + IP_TTL */ + + switch (anc_len--, *ptr++) { + case INET_OPT_TOS: { +#if defined(IPPROTO_IP) && defined(IP_TOS) + COMPILE_ANCILLARY_DATA_ITEM(IPPROTO_IP, IP_TOS, int, get_int32, 4); +#else + return 1; /* Socket option not implemented */ +#endif + break; + } + case INET_OPT_TTL: { +#if defined(IPPROTO_IP) && defined(IP_TTL) + COMPILE_ANCILLARY_DATA_ITEM(IPPROTO_IP, IP_TTL, int, get_int32, 4); +#else + return 1; /* Socket option not implemented */ +#endif + break; + } + case INET_OPT_TCLASS: { +#if defined(IPPROTO_IPV6) && defined(IPV6_TCLASS) + COMPILE_ANCILLARY_DATA_ITEM(IPPROTO_IPV6, IPV6_TCLASS, int, get_int32, 4); +#else + return 1; /* Socket option not implemented */ +#endif + break; + } + default: + /* Unknow socket option */ + return 1; + } +#undef COMPILE_ANCILLARY_DATA_ITEM + } +} #endif /* ifndef __WIN32__ */ /* @@ -9063,6 +9129,7 @@ static ErlDrvData inet_start(ErlDrvPort port, int size, int protocol) desc->event_mask = 0; #ifdef __WIN32__ desc->forced_events = 0; + desc->err = 0; desc->send_would_block = 0; #endif desc->port = port; @@ -9846,10 +9913,8 @@ static tcp_descriptor* tcp_inet_copy(tcp_descriptor* desc,SOCKET s, copy_desc->send_timeout = desc->send_timeout; copy_desc->send_timeout_close = desc->send_timeout_close; - if (desc->tcp_add_flags & TCP_ADDF_SHOW_ECONNRESET) - copy_desc->tcp_add_flags |= TCP_ADDF_SHOW_ECONNRESET; - else - copy_desc->tcp_add_flags &= ~TCP_ADDF_SHOW_ECONNRESET; + copy_desc->tcp_add_flags = desc->tcp_add_flags + & (TCP_ADDF_SHOW_ECONNRESET | TCP_ADDF_LINGER_ZERO); /* The new port will be linked and connected to the original caller */ port = driver_create_port(port, owner, "tcp_inet", (ErlDrvData) copy_desc); @@ -10914,7 +10979,7 @@ static int winsock_event_select(inet_descriptor *desc, int flags, int on) { int save_event_mask = desc->event_mask; - desc->forced_events = 0; + desc->forced_events &= FD_CLOSE; if (on) desc->event_mask |= flags; else @@ -10966,7 +11031,7 @@ static int winsock_event_select(inet_descriptor *desc, int flags, int on) TIMEVAL tmo = {0,0}; FD_SET fds; int ret; - + FD_ZERO(&fds); FD_SET(desc->s,&fds); do_force = (select(desc->s+1,0,&fds,0,&tmo) > 0); @@ -10983,7 +11048,7 @@ static int winsock_event_select(inet_descriptor *desc, int flags, int on) FD_SET fds; int ret; unsigned long arg; - + FD_ZERO(&fds); FD_SET(desc->s,&fds); ret = select(desc->s+1,&fds,0,0,&tmo); @@ -11022,13 +11087,16 @@ static void tcp_inet_event(ErlDrvData e, ErlDrvEvent event) goto error; } - DEBUGF((" => event=%02X, mask=%02X\r\n", - netEv.lNetworkEvents, desc->inet.event_mask)); + DEBUGF((" => event=%02X, mask=%02X, forced=%02X\r\n", + netEv.lNetworkEvents, desc->inet.event_mask, + desc->inet.forced_events)); /* Add the forced events. */ - netEv.lNetworkEvents |= desc->inet.forced_events; + if (desc->inet.forced_events & FD_CLOSE) + netEv.iErrorCode[FD_CLOSE_BIT] = desc->inet.err; + /* * Calling WSAEventSelect() with a mask of 0 doesn't always turn off * all events. To avoid acting on events we don't want, we mask @@ -11048,16 +11116,29 @@ static void tcp_inet_event(ErlDrvData e, ErlDrvEvent event) goto error; } if (netEv.lNetworkEvents & FD_CLOSE) { - /* + + /* We may not get any more FD_CLOSE events so we + keep this event and always signal it from + this moment on. */ + if ((desc->inet.forced_events & FD_CLOSE) == 0) { + desc->inet.forced_events |= FD_CLOSE; + desc->inet.err = netEv.iErrorCode[FD_CLOSE_BIT]; + } + + /* * We must loop to read out the remaining packets (if any). */ for (;;) { - DEBUGF(("Retrying read due to closed port\r\n")); - /* XXX The buffer will be thrown away on error (empty que). - Possible SMP FIXME. */ - if (!desc->inet.active && (desc->inet.opt) == NULL) { - goto error; - } + + /* if passive and no subscribers, break loop */ + if (!desc->inet.active && desc->inet.opt == NULL) { + /* do not trigger close event when socket is + transitioned to passive */ + netEv.lNetworkEvents &= ~FD_CLOSE; + break; + } + + DEBUGF(("Retrying read due to FD_CLOSE\r\n")); if (tcp_inet_input(desc, event) < 0) { goto error; } @@ -11386,7 +11467,7 @@ static int tcp_shutdown_error(tcp_descriptor* desc, int err) static void tcp_inet_delay_send(ErlDrvData data, ErlDrvTermData dummy) { tcp_descriptor *desc = (tcp_descriptor*)data; - (void)tcp_inet_output(desc, INETP(desc)->s); + (void)tcp_inet_output(desc, (HANDLE) INETP(desc)->s); } /* @@ -12553,7 +12634,7 @@ static void packet_inet_timeout(ErlDrvData e) sock_select(desc, FD_READ, 0); async_error_am (desc, am_timeout); } else { - (void)packet_inet_input(udesc, desc->s); + (void)packet_inet_input(udesc, (HANDLE) desc->s); } } @@ -12597,11 +12678,8 @@ static void packet_inet_command(ErlDrvData e, char* buf, ErlDrvSizeT len) char ancd[CMSG_SPACE(sizeof(*sri))]; } cmsg; - if (len < SCTP_GET_SENDPARAMS_LEN) { - inet_reply_error(desc, EINVAL); - return; - } - + if (len < SCTP_GET_SENDPARAMS_LEN) goto return_einval; + /* The ancillary data */ sri = (struct sctp_sndrcvinfo *) (CMSG_DATA(&cmsg.hdr)); /* Get the "sndrcvinfo" from the buffer, advancing the "ptr": */ @@ -12634,28 +12712,85 @@ static void packet_inet_command(ErlDrvData e, char* buf, ErlDrvSizeT len) goto check_result_code; } #endif - /* UDP socket. Even if it is connected, there is an address prefix - here -- ignored for connected sockets: */ - sz = len; - qtr = ptr; - xerror = inet_set_faddress(desc->sfamily, &other, &qtr, &sz); - if (xerror != NULL) { - inet_reply_error_am(desc, driver_mk_atom(xerror)); - return; - } - len -= (qtr - ptr); - ptr = qtr; - /* Now "ptr" is the user data ptr, "len" is data length: */ - inet_output_count(desc, len); - - if (desc->state & INET_F_ACTIVE) { /* connected (ignore address) */ - code = sock_send(desc->s, ptr, len, 0); - } - else { - code = sock_sendto(desc->s, ptr, len, 0, &other.sa, sz); + { + ErlDrvSizeT anc_len; + + /* UDP socket. Even if it is connected, there is an address prefix + here -- ignored for connected sockets: */ + sz = len; + qtr = ptr; + xerror = inet_set_faddress(desc->sfamily, &other, &qtr, &sz); + if (xerror != NULL) { + inet_reply_error_am(desc, driver_mk_atom(xerror)); + return; + } + len -= (qtr - ptr); + ptr = qtr; + + /* Here comes ancillary data */ + if (len < 4) goto return_einval; + anc_len = get_int32(ptr); + len -= 4; ptr += 4; + if (len < anc_len) goto return_einval; + + if (anc_len == 0 && !!0/*XXX-short-circuit-for-testing*/) { + /* Empty ancillary data */ + /* Now "ptr" is the user data ptr, "len" is data length: */ + inet_output_count(desc, len); + if (desc->state & INET_F_ACTIVE) { + /* connected (ignore address) */ + code = sock_send(desc->s, ptr, len, 0); + } + else { + code = sock_sendto(desc->s, ptr, len, 0, &other.sa, sz); + } + } + else { +#ifdef __WIN32__ + goto return_einval; /* Can not send ancillary data on Windows */ +#else + struct iovec iov[1]; + struct msghdr mhdr; + union { /* For ancillary data */ + struct cmsghdr hdr; + char ancd[SIZEOF_ANCILLARY_DATA]; + } cmsg; + sys_memset(&iov, '\0', sizeof(iov)); + sys_memset(&mhdr, '\0', sizeof(mhdr)); + sys_memset(&cmsg, '\0', sizeof(cmsg)); + if (desc->state & INET_F_ACTIVE) { + /* connected (ignore address) */ + mhdr.msg_name = NULL; + mhdr.msg_namelen = 0; + } + else { + mhdr.msg_name = &other; + mhdr.msg_namelen = sz; + } + mhdr.msg_control = cmsg.ancd; + mhdr.msg_controllen = sizeof(cmsg.ancd); + if (compile_ancillary_data(&mhdr, ptr, anc_len) != 0) { + goto return_einval; + } + if (mhdr.msg_controllen == 0) { + /* XXX Testing - only possible for anc_len == 0 */ + mhdr.msg_control = NULL; + } + len -= anc_len; + ptr += anc_len; + /* Now "ptr" is the user data ptr, "len" is data length: */ + iov[0].iov_len = len; + iov[0].iov_base = ptr; + mhdr.msg_iov = iov; + mhdr.msg_iovlen = 1; + mhdr.msg_flags = 0; + inet_output_count(desc, len); + code = sock_sendmsg(desc->s, &mhdr, 0); +#endif + } } -#ifdef HAVE_SCTP +#ifdef HAVE_SCTP check_result_code: /* "code" analysis is the same for both SCTP and UDP cases above: */ #endif @@ -12665,8 +12800,15 @@ static void packet_inet_command(ErlDrvData e, char* buf, ErlDrvSizeT len) } else inet_reply_ok(desc); + return; + + return_einval: + inet_reply_error(desc, EINVAL); + return; } -#endif + +#endif /* HAVE_UDP */ + #ifdef __WIN32__ static void packet_inet_event(ErlDrvData e, ErlDrvEvent event) diff --git a/erts/emulator/hipe/hipe_arm.c b/erts/emulator/hipe/hipe_arm.c index b61939724c..c5e2af0b5e 100644 --- a/erts/emulator/hipe/hipe_arm.c +++ b/erts/emulator/hipe/hipe_arm.c @@ -30,24 +30,39 @@ #include "hipe_native_bif.h" /* nbif_callemu() */ #include "hipe_bif0.h" +#ifndef __has_builtin +# define __has_builtin(x) 0 +#endif + /* Flush dcache and invalidate icache for a range of addresses. */ void hipe_flush_icache_range(void *address, unsigned int nbytes) { -#if defined(__ARM_EABI__) + void* end = (char*)address + nbytes; + +#if ERTS_AT_LEAST_GCC_VSN__(4, 3, 0) || __has_builtin(__builtin___clear_cache) + __builtin___clear_cache(address, end); +#elif defined(__clang__) + void __clear_cache(void *start, void *end); + __clear_cache(address, end); +#elif defined(__linux__) +# if defined(__ARM_EABI__) register unsigned long beg __asm__("r0") = (unsigned long)address; - register unsigned long end __asm__("r1") = (unsigned long)address + nbytes; + register unsigned long end __asm__("r1") = (unsigned long)end; register unsigned long flg __asm__("r2") = 0; register unsigned long scno __asm__("r7") = 0xf0002; __asm__ __volatile__("swi 0" /* sys_cacheflush() */ : "=r"(beg) : "0"(beg), "r"(end), "r"(flg), "r"(scno)); -#else +# else register unsigned long beg __asm__("r0") = (unsigned long)address; - register unsigned long end __asm__("r1") = (unsigned long)address + nbytes; + register unsigned long end __asm__("r1") = (unsigned long)end; register unsigned long flg __asm__("r2") = 0; __asm__ __volatile__("swi 0x9f0002" /* sys_cacheflush() */ : "=r"(beg) : "0"(beg), "r"(end), "r"(flg)); +# endif +#else +# error "Don't know how to flush instruction cache" #endif } diff --git a/erts/emulator/nifs/common/socket_nif.c b/erts/emulator/nifs/common/socket_nif.c index 25bc712949..adecbb3b6e 100644 --- a/erts/emulator/nifs/common/socket_nif.c +++ b/erts/emulator/nifs/common/socket_nif.c @@ -429,6 +429,7 @@ static void (*esock_sctp_freepaddrs)(struct sockaddr *addrs) = NULL; #define SOCKET_STATE_CONNECTING (SOCKET_STATE_OPEN | SOCKET_FLAG_CON) #define SOCKET_STATE_ACCEPTING (SOCKET_STATE_LISTENING | SOCKET_FLAG_ACC) #define SOCKET_STATE_CLOSING (SOCKET_FLAG_CLOSE) +#define SOCKET_STATE_DTOR (0xFFFF) #define IS_CLOSED(d) \ ((d)->state == SOCKET_STATE_CLOSED) @@ -485,6 +486,9 @@ static void (*esock_sctp_freepaddrs)(struct sockaddr *addrs) = NULL; #define SOCKET_OPT_VALUE_TYPE_INT 1 #define SOCKET_OPT_VALUE_TYPE_BOOL 2 +#define ESOCK_DESC_PATTERN_CREATED 0x03030303 +#define ESOCK_DESC_PATTERN_DTOR 0xC0C0C0C0 + typedef union { struct { // 0 = not open, 1 = open @@ -808,6 +812,14 @@ typedef struct { typedef struct { + /* + * +++ This is a way to, possibly, detect memory overrides "and stuff" +++ + * + * We have two patterns. One is set when the descriptor is created (allocated) + * and one is set when the descriptor is dtor'ed. + */ + Uint32 pattern; + /* +++ The actual socket +++ */ SOCKET sock; HANDLE event; @@ -825,6 +837,10 @@ typedef struct { ErlNifPid ctrlPid; ESockMonitor ctrlMon; + /* +++ Connector process +++ */ + ErlNifPid connPid; + ESockMonitor connMon; + /* +++ Write stuff +++ */ ErlNifMutex* writeMtx; ESockRequestor currentWriter; @@ -2377,6 +2393,8 @@ static int socket_setopt(int sock, const void* optVal, const socklen_t optLen); +static BOOLEAN_T is_connector(ErlNifEnv* env, + ESockDescriptor* descP); static BOOLEAN_T verify_is_connected(ESockDescriptor* descP, int* err); static ESockDescriptor* alloc_descriptor(SOCKET sock, HANDLE event); @@ -4895,17 +4913,17 @@ ERL_NIF_TERM nconnect(ErlNifEnv* env, return esock_make_error(env, atom_closed); if (!IS_OPEN(descP)) { - SSDBG( descP, ("SOCKET", "nif_connect -> not open\r\n") ); + SSDBG( descP, ("SOCKET", "nconnect -> not open\r\n") ); return esock_make_error(env, atom_exbadstate); } if (IS_CONNECTED(descP)) { - SSDBG( descP, ("SOCKET", "nif_connect -> already connected\r\n") ); + SSDBG( descP, ("SOCKET", "nconnect -> already connected\r\n") ); return esock_make_error(env, atom_eisconn); } - if (IS_CONNECTING(descP)) { - SSDBG( descP, ("SOCKET", "nif_connect -> already connecting\r\n") ); + if (IS_CONNECTING(descP) && !is_connector(env, descP)) { + SSDBG( descP, ("SOCKET", "nconnect -> already connecting\r\n") ); return esock_make_error(env, esock_atom_einval); } @@ -4919,31 +4937,93 @@ ERL_NIF_TERM nconnect(ErlNifEnv* env, descP->addrLen); save_errno = sock_errno(); - SSDBG( descP, ("SOCKET", "nif_connect -> connect result: %d, %d\r\n", + SSDBG( descP, ("SOCKET", "nconnect -> connect result: %d, %d\r\n", code, save_errno) ); - if (IS_SOCKET_ERROR(code) && - ((save_errno == ERRNO_BLOCK) || /* Winsock2 */ - (save_errno == EINPROGRESS))) { /* Unix & OSE!! */ - ref = MKREF(env); - descP->state = SOCKET_STATE_CONNECTING; - if ((sres = esock_select_write(env, descP->sock, descP, NULL, - sockRef, ref)) < 0) { - res = esock_make_error(env, - MKT2(env, - esock_atom_select_failed, - MKI(env, sres))); - } else { - res = esock_make_ok2(env, ref); + if (IS_SOCKET_ERROR(code)) { + switch (save_errno) { + case ERRNO_BLOCK: /* Winsock2 */ + case EINPROGRESS: /* Unix & OSE!! */ + SSDBG( descP, ("SOCKET", "nconnect -> would block => select\r\n") ); + + ref = MKREF(env); + + if (IS_CONNECTING(descP)) { + /* Glitch */ + res = esock_make_ok2(env, ref); + } else { + + /* First time here */ + + if (enif_self(env, &descP->connPid) == NULL) + return esock_make_error(env, atom_exself); + + if (MONP("nconnect -> conn", + env, descP, + &descP->connPid, + &descP->connMon) != 0) + return esock_make_error(env, atom_exmon); + + descP->state = SOCKET_STATE_CONNECTING; + + if ((sres = esock_select_write(env, descP->sock, descP, NULL, + sockRef, ref)) < 0) { + res = esock_make_error(env, + MKT2(env, + esock_atom_select_failed, + MKI(env, sres))); + } else { + res = esock_make_ok2(env, ref); + } + } + break; + + case EISCONN: + SSDBG( descP, ("SOCKET", "nconnect -> *already* connected\r\n") ); + { + /* This is ***strange*** so make sure */ + int err = 0; + if (!verify_is_connected(descP, &err)) { + descP->state = SOCKET_STATE_OPEN; /* restore state */ + res = esock_make_error_errno(env, err); + } else { + descP->state = SOCKET_STATE_CONNECTED; + /* And just to be on the safe side, reset these */ + enif_set_pid_undefined(&descP->connPid); + DEMONP("nconnect -> connected", + env, descP, &descP->connMon); + descP->isReadable = TRUE; + descP->isWritable = TRUE; + res = esock_atom_ok; + } + } + break; + + default: + SSDBG( descP, ("SOCKET", "nconnect -> other error(1): %d\r\n", + save_errno) ); + res = esock_make_error_errno(env, save_errno); + break; } + } else if (code == 0) { /* ok we are connected */ + SSDBG( descP, ("SOCKET", "nconnect -> connected\r\n") ); + descP->state = SOCKET_STATE_CONNECTED; + enif_set_pid_undefined(&descP->connPid); + DEMONP("nconnect -> connected", env, descP, &descP->connMon); descP->isReadable = TRUE; descP->isWritable = TRUE; res = esock_atom_ok; + } else { + /* Do we really need this case? */ + + SSDBG( descP, ("SOCKET", "nconnect -> other error(2): %d\r\n", + save_errno) ); + res = esock_make_error_errno(env, save_errno); } @@ -4998,7 +5078,7 @@ ERL_NIF_TERM nfinalize_connection(ErlNifEnv* env, { int error; - if (descP->state != SOCKET_STATE_CONNECTING) + if (!IS_CONNECTING(descP)) return esock_make_error(env, atom_enotconn); if (!verify_is_connected(descP, &error)) { @@ -5007,6 +5087,8 @@ ERL_NIF_TERM nfinalize_connection(ErlNifEnv* env, } descP->state = SOCKET_STATE_CONNECTED; + enif_set_pid_undefined(&descP->connPid); + DEMONP("nfinalize_connection -> connected", env, descP, &descP->connMon); descP->isReadable = TRUE; descP->isWritable = TRUE; @@ -5069,6 +5151,29 @@ BOOLEAN_T verify_is_connected(ESockDescriptor* descP, int* err) +/* *** is_connector *** + * Check if the current process is the connector process. + */ +#if !defined(__WIN32__) +static +BOOLEAN_T is_connector(ErlNifEnv* env, + ESockDescriptor* descP) +{ + ErlNifPid caller; + + if (enif_self(env, &caller) == NULL) + return FALSE; + + if (COMPARE_PIDS(&descP->connPid, &caller) == 0) + return TRUE; + else + return FALSE; + +} +#endif + + + /* ---------------------------------------------------------------------- * nif_listen * @@ -16872,8 +16977,10 @@ ESockDescriptor* alloc_descriptor(SOCKET sock, HANDLE event) if ((descP = enif_alloc_resource(sockets, sizeof(ESockDescriptor))) != NULL) { char buf[64]; /* Buffer used for building the mutex name */ - // This needs to be released when the socket is closed! - // descP->env = enif_alloc_env(); + descP->pattern = ESOCK_DESC_PATTERN_CREATED; + + enif_set_pid_undefined(&descP->connPid); + MON_INIT(&descP->connMon); sprintf(buf, "esock[w,%d]", sock); descP->writeMtx = MCREATE(buf); @@ -18214,21 +18321,30 @@ void socket_dtor(ErlNifEnv* env, void* obj) #if !defined(__WIN32__) ESockDescriptor* descP = (ESockDescriptor*) obj; - MDESTROY(descP->writeMtx); - MDESTROY(descP->readMtx); - MDESTROY(descP->accMtx); - MDESTROY(descP->closeMtx); - MDESTROY(descP->cfgMtx); + MDESTROY(descP->writeMtx); descP->writeMtx = NULL; + MDESTROY(descP->readMtx); descP->readMtx = NULL; + MDESTROY(descP->accMtx); descP->accMtx = NULL; + MDESTROY(descP->closeMtx); descP->closeMtx = NULL; + MDESTROY(descP->cfgMtx); descP->cfgMtx = NULL; - if (descP->currentReader.env) + if (descP->currentReader.env) { esock_free_env("dtor reader", descP->currentReader.env); - if (descP->currentWriter.env) + descP->currentReader.env = NULL; + } + if (descP->currentWriter.env) { esock_free_env("dtor writer", descP->currentWriter.env); - if (descP->currentAcceptor.env) + descP->currentWriter.env = NULL; + } + if (descP->currentAcceptor.env) { esock_free_env("dtor acceptor", descP->currentAcceptor.env); + descP->currentAcceptor.env = NULL; + } free_request_queue(&descP->readersQ); free_request_queue(&descP->writersQ); free_request_queue(&descP->acceptorsQ); + + descP->state = SOCKET_STATE_DTOR; + descP->pattern = ESOCK_DESC_PATTERN_DTOR; #endif } @@ -18652,6 +18768,17 @@ void socket_down(ErlNifEnv* env, MON2T(env, mon)); } + } else if (COMPARE_PIDS(&descP->connPid, pid) == 0) { + + /* The connPid is only set during the connection. + * The same goes for the monitor (connMon). + */ + + descP->state = SOCKET_STATE_OPEN; /* restore state */ + enif_set_pid_undefined(&descP->connPid); + DEMONP("socket_down -> connector", + env, descP, &descP->connMon); + } else { /* check all operation queue(s): acceptor, writer and reader. diff --git a/erts/emulator/test/esock_ttest/esock-ttest b/erts/emulator/test/esock_ttest/esock-ttest index cf1d9cd9ab..9adc51fc8b 100755 --- a/erts/emulator/test/esock_ttest/esock-ttest +++ b/erts/emulator/test/esock_ttest/esock-ttest @@ -60,6 +60,9 @@ usage() -> "~n Which domain to use." "~n Only valid for server." "~n Defaults to: inet" + "~n --async Asynchronous mode (Timeout = nowait)" + "~n This option is only valid for transport = sock." + "~n Also, its only used when active =/= false." "~n --active <active> boolean() | once." "~n Valid for both client and server." "~n Defaults to: false" @@ -111,6 +114,7 @@ process_args(Args) -> process_server_args(Args) -> Defaults = #{role => server, domain => inet, + async => false, active => false, transport => {sock, plain}}, process_server_args(Args, Defaults). @@ -124,6 +128,9 @@ process_server_args(["--domain", Domain|Args], State) (Domain =:= "inet6")) -> process_server_args(Args, State#{domain => list_to_atom(Domain)}); +process_server_args(["--async"|Args], State) -> + process_server_args(Args, State#{async => true}); + process_server_args(["--active", Active|Args], State) when ((Active =:= "false") orelse (Active =:= "once") orelse @@ -145,6 +152,7 @@ process_server_args([Arg|_], _State) -> process_client_args(Args) -> Defaults = #{role => client, + async => false, active => false, transport => {sock, plain}, %% Will cause error if not provided @@ -159,10 +167,13 @@ process_client_args(Args) -> process_client_args([], State) -> process_client_args_ensure_max_outstanding(State); +process_client_args(["--async"|Args], State) -> + process_client_args(Args, State#{async => true}); + process_client_args(["--active", Active|Args], State) - when ((Active =:= "false") orelse - (Active =:= "once") orelse - (Active =:= "true")) -> + when (Active =:= "false") orelse + (Active =:= "once") orelse + (Active =:= "true") -> process_client_args(Args, State#{active => list_to_atom(Active)}); process_client_args(["--transport", "gen" | Args], State) -> @@ -280,9 +291,10 @@ exec(#{role := server, end; exec(#{role := server, domain := Domain, + async := Async, active := Active, transport := {sock, Method}}) -> - case socket_test_ttest_tcp_server_socket:start(Method, Domain, Active) of + case socket_test_ttest_tcp_server_socket:start(Method, Domain, Async, Active) of {ok, {Pid, _}} -> MRef = erlang:monitor(process, Pid), receive @@ -323,12 +335,14 @@ exec(#{role := client, end; exec(#{role := client, server := ServerInfo, + async := Async, active := Active, transport := {sock, Method}, msg_id := MsgID, max_outstanding := MaxOutstanding, runtime := RunTime}) -> case socket_test_ttest_tcp_client_socket:start(true, + Async, Method, ServerInfo, Active, diff --git a/erts/emulator/test/esock_ttest/esock-ttest-server-sock b/erts/emulator/test/esock_ttest/esock-ttest-server-sock index 4ec0d335d9..fc87499f09 100755 --- a/erts/emulator/test/esock_ttest/esock-ttest-server-sock +++ b/erts/emulator/test/esock_ttest/esock-ttest-server-sock @@ -24,9 +24,27 @@ EMU=$ERL_TOP/erts/emulator EMU_TEST=$EMU/test ESOCK_TTEST=$EMU_TEST/esock_ttest -if [ $# = 1 ]; then - ACTIVE="--active $1" +# $1 - async - boolean() +# $2 - active - once | boolean() +if [ $# = 2 ]; then + + async=$1 + active=$2 + + if [ $async = true ]; then + ASYNC="--async" + else + ASYNC= + fi + + ACTIVE="--active $active" + +else + echo "<ERROR> Missing args: async and active" + echo "" + exit 1 fi -$ESOCK_TTEST/esock-ttest --server --transport sock $ACTIVE + +$ESOCK_TTEST/esock-ttest --server $ASYNC --transport sock $ACTIVE diff --git a/erts/emulator/test/process_SUITE.erl b/erts/emulator/test/process_SUITE.erl index b530ced566..3684cde8d4 100644 --- a/erts/emulator/test/process_SUITE.erl +++ b/erts/emulator/test/process_SUITE.erl @@ -1098,42 +1098,86 @@ process_info_status_handled_signal(Config) when is_list(Config) -> %% OTP-15709 %% Provoke a bug where process_info(reductions) returned wrong result %% because REDS_IN (def_arg_reg[5]) is read when the process in not running. +%% +%% And a bug where process_info(reductions) on a process which was releasing its +%% main lock during execution could result in negative reduction diffs. process_info_reductions(Config) when is_list(Config) -> - pi_reductions_tester(spawn_link(fun() -> pi_reductions_spinnloop() end)), - pi_reductions_tester(spawn_link(fun() -> pi_reductions_recvloop() end)), + {S1, S2} = case erlang:system_info(schedulers) of + 1 -> {1,1}; + _ -> {1,2} + end, + io:format("Run on schedulers ~p and ~p\n", [S1,S2]), + Boss = self(), + Doer = spawn_opt(fun () -> + pi_reductions_tester(true, 10, fun pi_reductions_spinnloop/0, S2), + pi_reductions_tester(true, 10, fun pi_reductions_recvloop/0, S2), + pi_reductions_tester(false, 100, fun pi_reductions_main_unlocker/0, S2), + Boss ! {self(), done} + end, + [link, {scheduler, S1}]), + + {Doer, done} = receive M -> M end, ok. -pi_reductions_tester(Pid) -> - {_, DiffList} = - lists:foldl(fun(_, {Prev, Acc}) -> - %% Add another item that force sending the request - %% as a signal, like 'current_function'. - PI = process_info(Pid, [reductions, current_function]), - [{reductions,Reds}, {current_function,_}] = PI, - Diff = Reds - Prev, - {Diff, true} = {Diff, (Diff >= 0)}, - {Diff, true} = {Diff, (Diff =< 1000*1000)}, - {Reds, [Diff | Acc]} - end, - {0, []}, - lists:seq(1,10)), +pi_reductions_tester(ForceSignal, MaxCalls, Fun, S2) -> + Pid = spawn_opt(Fun, [link, {scheduler,S2}]), + Extra = case ForceSignal of + true -> + %% Add another item that force sending the request + %% as a signal, like 'current_function'. + [current_function]; + false -> + [] + end, + LoopFun = fun Me(Calls, Prev, Acc0) -> + PI = process_info(Pid, [reductions | Extra]), + [{reductions,Reds} | _] = PI, + Diff = Reds - Prev, + %% Verify we get sane non-negative reduction diffs + {Diff, true} = {Diff, (Diff >= 0)}, + {Diff, true} = {Diff, (Diff =< 1000*1000)}, + Acc1 = [Diff | Acc0], + case Calls >= MaxCalls of + true -> Acc1; + false -> Me(Calls+1, Reds, Acc1) + end + end, + DiffList = LoopFun(0, 0, []), unlink(Pid), exit(Pid,kill), - io:format("Reduction diffs: ~p\n", [DiffList]), + io:format("Reduction diffs: ~p\n", [lists:reverse(DiffList)]), ok. pi_reductions_spinnloop() -> %% 6 args to make use of def_arg_reg[5] which is also used as REDS_IN - pi_reductions_spinnloop(1, atom, "hej", self(), make_ref(), 3.14). + pi_reductions_spinnloop(999*1000, atom, "hej", self(), make_ref(), 3.14). -pi_reductions_spinnloop(A,B,C,D,E,F) -> - pi_reductions_spinnloop(B,C,D,E,F,A). +pi_reductions_spinnloop(N,A,B,C,D,E) when N > 0 -> + pi_reductions_spinnloop(N-1,B,C,D,E,A); +pi_reductions_spinnloop(0,_,_,_,_,_) -> + %% Stop to limit max number of reductions consumed + pi_reductions_recvloop(). pi_reductions_recvloop() -> receive "a free lunch" -> false end. +pi_reductions_main_unlocker() -> + Other = spawn_link(fun() -> receive die -> ok end end), + pi_reductions_main_unlocker_loop(Other). + +pi_reductions_main_unlocker_loop(Other) -> + %% Assumption: register(OtherPid, Name) will unlock main lock of calling + %% process during execution. + register(pi_reductions_main_unlocker, Other), + unregister(pi_reductions_main_unlocker), + + %% Yield in order to increase probability of process_info sometimes probing + %% this process when it's not RUNNING. + erlang:yield(), + pi_reductions_main_unlocker_loop(Other). + %% Tests erlang:bump_reductions/1. bump_reductions(Config) when is_list(Config) -> diff --git a/erts/emulator/test/socket_SUITE.erl b/erts/emulator/test/socket_SUITE.erl index 49b0fcccc2..d8cb1013e9 100644 --- a/erts/emulator/test/socket_SUITE.erl +++ b/erts/emulator/test/socket_SUITE.erl @@ -81,6 +81,24 @@ api_b_sendmsg_and_recvmsg_tcp4/1, api_b_sendmsg_and_recvmsg_tcpL/1, + %% *** API async *** + api_a_connect_tcp4/1, + api_a_sendto_and_recvfrom_udp4/1, + api_a_sendmsg_and_recvmsg_udp4/1, + api_a_send_and_recv_tcp4/1, + api_a_sendmsg_and_recvmsg_tcp4/1, + api_a_recvfrom_cancel_udp4/1, + api_a_recvmsg_cancel_udp4/1, + api_a_accept_cancel_tcp4/1, + api_a_recv_cancel_tcp4/1, + api_a_recvmsg_cancel_tcp4/1, + api_a_mrecvfrom_cancel_udp4/1, + api_a_mrecvmsg_cancel_udp4/1, + api_a_maccept_cancel_tcp4/1, + api_a_mrecv_cancel_tcp4/1, + api_a_mrecvmsg_cancel_tcp4/1, + + %% *** API Options *** api_opt_simple_otp_options/1, api_opt_simple_otp_rcvbuf_option/1, @@ -565,6 +583,7 @@ use_group(Group, Env, Default) -> groups() -> [{api, [], api_cases()}, {api_basic, [], api_basic_cases()}, + {api_async, [], api_async_cases()}, {api_options, [], api_options_cases()}, {api_op_with_timeout, [], api_op_with_timeout_cases()}, {socket_closure, [], socket_closure_cases()}, @@ -639,6 +658,7 @@ groups() -> api_cases() -> [ {group, api_basic}, + {group, api_async}, {group, api_options}, {group, api_op_with_timeout} ]. @@ -659,6 +679,25 @@ api_basic_cases() -> api_b_sendmsg_and_recvmsg_tcpL ]. +api_async_cases() -> + [ + api_a_connect_tcp4, + api_a_sendto_and_recvfrom_udp4, + api_a_sendmsg_and_recvmsg_udp4, + api_a_send_and_recv_tcp4, + api_a_sendmsg_and_recvmsg_tcp4, + api_a_recvfrom_cancel_udp4, + api_a_recvmsg_cancel_udp4, + api_a_accept_cancel_tcp4, + api_a_recv_cancel_tcp4, + api_a_recvmsg_cancel_tcp4, + api_a_mrecvfrom_cancel_udp4, + api_a_mrecvmsg_cancel_udp4, + api_a_maccept_cancel_tcp4, + api_a_mrecv_cancel_tcp4, + api_a_mrecvmsg_cancel_tcp4 + ]. + api_options_cases() -> [ api_opt_simple_otp_options, @@ -1543,11 +1582,12 @@ init_per_group(ttest = _GroupName, Config) -> io:format("init_per_group(~w) -> entry with" "~n Config: ~p" "~n", [_GroupName, Config]), + ttest_manager_start(), case lists:keysearch(esock_test_ttest_runtime, 1, Config) of {value, _} -> Config; false -> - [{esock_test_ttest_runtime, which_ttest_runtime_env()}|Config] + [{esock_test_ttest_runtime, which_ttest_runtime_env()} | Config] end; init_per_group(_GroupName, Config) -> Config. @@ -1556,6 +1596,7 @@ end_per_group(ttest = _GroupName, Config) -> io:format("init_per_group(~w) -> entry with" "~n Config: ~p" "~n", [_GroupName, Config]), + ttest_manager_stop(), lists:keydelete(esock_test_ttest_runtime, 1, Config); end_per_group(_GroupName, Config) -> Config. @@ -1565,16 +1606,16 @@ init_per_testcase(_TC, Config) -> io:format("init_per_testcase(~w) -> entry with" "~n Config: ~p" "~n", [_TC, Config]), - case quiet_mode(Config) of - default -> - ?LOGGER:start(); - Quiet -> - ?LOGGER:start(Quiet) - end, + %% case quiet_mode(Config) of + %% default -> + %% ?LOGGER:start(); + %% Quiet -> + %% ?LOGGER:start(Quiet) + %% end, Config. end_per_testcase(_TC, Config) -> - ?LOGGER:stop(), + %% ?LOGGER:stop(), Config. @@ -2526,11 +2567,7 @@ api_b_send_and_recv_tcp(InitState) -> ?SEV_ANNOUNCE_CONTINUE(Server, accept), ok end}, - #{desc => "sleep", - cmd => fun(_) -> - ?SLEEP(?SECS(1)), - ok - end}, + ?SEV_SLEEP(?SECS(1)), #{desc => "order client to continue (with connect)", cmd => fun(#{client := Client} = _State) -> ?SEV_ANNOUNCE_CONTINUE(Client, connect), @@ -2617,6 +2654,3852 @@ api_b_send_and_recv_tcp(InitState) -> %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically establish a TCP connection via an async connect. + +api_a_connect_tcp4(suite) -> + []; +api_a_connect_tcp4(doc) -> + []; +api_a_connect_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(10)), + tc_try(api_a_connect_tcp4, + fun() -> + Connect = fun(Sock, SockAddr) -> + socket:connect(Sock, SockAddr, nowait) + end, + Send = fun(Sock, Data) -> + socket:send(Sock, Data) + end, + Recv = fun(Sock) -> + socket:recv(Sock) + end, + InitState = #{domain => inet, + connect => Connect, + send => Send, + recv => Recv}, + ok = api_a_connect_tcp(InitState) + end). + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_connect_tcp(InitState) -> + process_flag(trap_exit, true), + ServerSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{lsa => LSA}} + end}, + #{desc => "create listen socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{lsock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{lsock := LSock, lsa := LSA} = State) -> + case socket:bind(LSock, LSA) of + {ok, Port} -> + {ok, State#{lport => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "make listen socket", + cmd => fun(#{lsock := LSock}) -> + socket:listen(LSock) + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, lport := Port}) -> + ?SEV_ANNOUNCE_READY(Tester, init, Port), + ok + end}, + + %% The actual test + #{desc => "await continue (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, accept) + end}, + #{desc => "await connection", + cmd => fun(#{lsock := LSock} = State) -> + case socket:accept(LSock) of + {ok, Sock} -> + ?SEV_IPRINT("accepted: ~n ~p", [Sock]), + {ok, State#{csock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, accept), + ok + end}, + + #{desc => "await continue (recv_req)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv_req) + end}, + #{desc => "recv req", + cmd => fun(#{csock := Sock, recv := Recv}) -> + case Recv(Sock) of + {ok, ?BASIC_REQ} -> + ok; + {ok, UnexpData} -> + {error, {unexpected_data, UnexpData}}; + {error, _} = ERROR -> + %% At the moment there is no way to get + %% status or state for the socket... + ERROR + end + end}, + #{desc => "announce ready (recv_req)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_req), + ok + end}, + #{desc => "await continue (send_rep)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, send_rep) + end}, + #{desc => "send rep", + cmd => fun(#{csock := Sock, send := Send}) -> + Send(Sock, ?BASIC_REP) + end}, + #{desc => "announce ready (send_rep)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, send_rep), + ok + end}, + + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close connection socket", + cmd => fun(#{csock := Sock} = State) -> + ok = socket:close(Sock), + {ok, maps:remove(csock, State)} + end}, + #{desc => "close listen socket", + cmd => fun(#{lsock := Sock} = State) -> + ok = socket:close(Sock), + {ok, maps:remove(lsock, State)} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + ClientSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + {Tester, Port} = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester, server_port => Port}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** The init part *** + #{desc => "which server (local) address", + cmd => fun(#{domain := Domain, server_port := Port} = State) -> + LSA = which_local_socket_addr(Domain), + SSA = LSA#{port => Port}, + {ok, State#{local_sa => LSA, server_sa => SSA}} + end}, + #{desc => "create socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{sock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{sock := Sock, local_sa := LSA} = _State) -> + case socket:bind(Sock, LSA) of + {ok, _Port} -> + ok; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, init), + ok + end}, + + %% *** The actual test *** + #{desc => "await continue (async connect)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, async_connect) + end}, + #{desc => "connect (async) to server", + cmd => fun(#{sock := Sock, + server_sa := SSA, + connect := Connect} = State) -> + case Connect(Sock, SSA) of + ok -> + ?SEV_IPRINT("ok -> " + "unexpected success => SKIP", + []), + {skip, unexpected_success}; + {select, {select_info, ST, SR}} -> + ?SEV_IPRINT("select ->" + "~n tag: ~p" + "~n ref: ~p", [ST, SR]), + {ok, State#{connect_stag => ST, + connect_sref => SR}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (connect select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, connect_select), + ok + end}, + #{desc => "await select message", + cmd => fun(#{sock := Sock, connect_sref := Ref}) -> + receive + {'$socket', Sock, select, Ref} -> + ?SEV_IPRINT("select message ->" + "~n ref: ~p", [Ref]), + ok + after 5000 -> + ?SEV_EPRINT("timeout: " + "~n message queue: ~p", + [mq()]), + {error, timeout} + end + end}, + #{desc => "announce ready (select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, select), + ok + end}, + #{desc => "connect (async) to server", + cmd => fun(#{sock := Sock, server_sa := SSA, connect := Connect}) -> + case Connect(Sock, SSA) of + ok -> + ok; + {select, SelectInfo} -> + {error, {unexpected_select, SelectInfo}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (connect)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, connect), + ok + end}, + #{desc => "get peername", + cmd => fun(#{sock := Sock} = _State) -> + case socket:peername(Sock) of + {ok, SockAddr} -> + ?SEV_IPRINT("Peer Name: ~p", [SockAddr]), + ok; + {error, _} = ERROR -> + ERROR + end + end}, + + #{desc => "await continue (send_req)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, send_req) + end}, + #{desc => "send req", + cmd => fun(#{sock := Sock, send := Send}) -> + Send(Sock, ?BASIC_REQ) + end}, + #{desc => "announce ready (send_req)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, send_req), + ok + end}, + #{desc => "await continue (recv_rep)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv_rep) + end}, + #{desc => "recv rep", + cmd => fun(#{sock := Sock, recv := Recv}) -> + case Recv(Sock) of + {ok, ?BASIC_REP} -> + ok; + {ok, UnexpData} -> + {error, {unexpected_data, UnexpData}}; + {error, _} = ERROR -> + %% At the moment there is no way to get + %% status or state for the socket... + ERROR + end + end}, + #{desc => "announce ready (recv_rep)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_rep), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close socket", + cmd => fun(#{sock := Sock} = State) -> + ok = socket:close(Sock), + State2 = maps:remove(sock, State), + State3 = maps:remove(connect_stag, State2), + State4 = maps:remove(connect_sref, State3), + {ok, State4} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor client", + cmd => fun(#{client := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, Port} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{server_port => Port}} + end}, + + %% Start the client + #{desc => "order client start", + cmd => fun(#{client := Pid, server_port := Port} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Port), + ok + end}, + #{desc => "await client ready (init)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, init) + end}, + + + %% *** The actual test *** + #{desc => "order client to continue (async connect)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Client, async_connect), + ok + end}, + #{desc => "await client ready (connect select)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_AWAIT_READY(Client, client, connect_select) + end}, + + ?SEV_SLEEP(?SECS(1)), + + #{desc => "order server to continue (with accept)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Server, accept), + ok + end}, + #{desc => "await client ready (select)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_AWAIT_READY(Client, client, select) + end}, + #{desc => "await client ready (connect)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_AWAIT_READY(Client, client, connect) + end}, + #{desc => "await server ready (accept)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_AWAIT_READY(Server, server, accept) + end}, + + ?SEV_SLEEP(?SECS(1)), + + #{desc => "order server to recv test req (recv req)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Server, recv_req), + ok + end}, + #{desc => "order client to send test req (send req)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Client, send_req), + ok + end}, + #{desc => "await client ready (send_req)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_AWAIT_READY(Client, client, send_req) + end}, + #{desc => "await server ready (recv_req)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_AWAIT_READY(Server, server, recv_req) + end}, + #{desc => "order client to recv test rep (send rep)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Client, recv_rep), + ok + end}, + #{desc => "order server to send test rep (send rep)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Server, send_rep), + ok + end}, + #{desc => "await server ready (send_rep)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_AWAIT_READY(Server, server, send_rep) + end}, + #{desc => "await client ready (recv_rep)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_AWAIT_READY(Client, client, recv_rep) + end}, + + + %% *** Termination *** + #{desc => "order client to terminate", + cmd => fun(#{client := Client} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Client), + ok + end}, + #{desc => "await client termination", + cmd => fun(#{client := Client} = State) -> + ?SEV_AWAIT_TERMINATION(Client), + State1 = maps:remove(client, State), + {ok, State1} + end}, + #{desc => "order server to terminate", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Server), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Server} = State) -> + ?SEV_AWAIT_TERMINATION(Server), + State1 = maps:remove(server, State), + {ok, State1} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + Server = ?SEV_START("server", ServerSeq, InitState), + + i("start client evaluator"), + Client = ?SEV_START("client", ClientSeq, InitState), + i("await evaluator(s)"), + + i("start tester evaluator"), + TesterInitState = #{server => Server#ev.pid, + client => Client#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + ok = ?SEV_AWAIT_FINISH([Server, Client, Tester]). + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically send and receive on an IPv4 UDP (dgram) socket using +%% sendto and recvfrom. But we try to be async. That is, we use +%% the 'nowait' value for the Timeout argument (and await the eventual +%% select message). Note that we only do this for the recvfrom, +%% since its much more difficult to "arrange" for sendto. +%% +api_a_sendto_and_recvfrom_udp4(suite) -> + []; +api_a_sendto_and_recvfrom_udp4(doc) -> + []; +api_a_sendto_and_recvfrom_udp4(_Config) when is_list(_Config) -> + ?TT(?SECS(5)), + tc_try(api_a_sendto_and_recvfrom_udp4, + fun() -> + Send = fun(Sock, Data, Dest) -> + socket:sendto(Sock, Data, Dest) + end, + Recv = fun(Sock) -> + socket:recvfrom(Sock, 0, nowait) + end, + InitState = #{domain => inet, + send => Send, + recv => Recv}, + ok = api_a_send_and_recv_udp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically send and receive on an IPv4 UDP (dgram) socket using +%% sendto and recvfrom. But we try to be async. That is, we use +%% the 'nowait' value for the Timeout argument (and await the eventual +%% select message). Note that we only do this for the recvmsg, +%% since its much more difficult to "arrange" for sendmsg. +%% +api_a_sendmsg_and_recvmsg_udp4(suite) -> + []; +api_a_sendmsg_and_recvmsg_udp4(doc) -> + []; +api_a_sendmsg_and_recvmsg_udp4(_Config) when is_list(_Config) -> + ?TT(?SECS(5)), + tc_try(api_a_sendmsg_and_recvmsg_udp4, + fun() -> + Send = fun(Sock, Data, Dest) -> + MsgHdr = #{addr => Dest, + %% ctrl => CMsgHdrs, + iov => [Data]}, + socket:sendmsg(Sock, MsgHdr) + end, + Recv = fun(Sock) -> + case socket:recvmsg(Sock, nowait) of + {ok, #{addr := Source, + iov := [Data]}} -> + {ok, {Source, Data}}; + {ok, _} = OK -> + OK; + {select, _} = SELECT -> + SELECT; + {error, _} = ERROR -> + ERROR + end + end, + InitState = #{domain => inet, + send => Send, + recv => Recv}, + ok = api_a_send_and_recv_udp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_send_and_recv_udp(InitState) -> + ServerSeq = + [ + %% *** Wait for start order part *** + #{desc => "await start", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester} = _State) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{local_sa => LSA}} + end}, + #{desc => "create socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, dgram, udp) of + {ok, Sock} -> + {ok, State#{sock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind socket (to local address)", + cmd => fun(#{sock := Sock, local_sa := LSA} = State) -> + case socket:bind(Sock, LSA) of + {ok, Port} -> + {ok, State#{port => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, local_sa := LSA, port := Port}) -> + ServerSA = LSA#{port => Port}, + ?SEV_ANNOUNCE_READY(Tester, init, ServerSA), + ok + end}, + + %% The actual test + #{desc => "await continue (recv)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv) + end}, + #{desc => "try recv request (with nowait, expect select)", + cmd => fun(#{sock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, {select_info, Tag, RecvRef}} -> + ?SEV_IPRINT("expected select: " + "~n Tag: ~p" + "~n Ref: ~p", [Tag, RecvRef]), + {ok, State#{recv_stag => Tag, + recv_sref => RecvRef}}; + {ok, X} -> + {error, {unexpected_succes, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await select message", + cmd => fun(#{sock := Sock, recv_sref := RecvRef}) -> + receive + {'$socket', Sock, select, RecvRef} -> + ok + after 5000 -> + ?SEV_EPRINT("message queue: ~p", [mq()]), + {error, timeout} + end + end}, + #{desc => "announce ready (select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, select), + ok + end}, + #{desc => "now read the data (request)", + cmd => fun(#{sock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {ok, {Src, ?BASIC_REQ}} -> + {ok, State#{req_src => Src}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv request)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_req), + ok + end}, + + #{desc => "await continue (send reply)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, send_reply) + end}, + #{desc => "send reply", + cmd => fun(#{sock := Sock, req_src := Src, send := Send}) -> + Send(Sock, ?BASIC_REP, Src) + end}, + #{desc => "announce ready (send)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, send), + ok + end}, + + %% Termination + #{desc => "await terminate (from tester)", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + State2 = maps:remove(tester, State), + State3 = maps:remove(recv_stag, State2), + State4 = maps:remove(recv_sref, State3), + State5 = maps:remove(req_src, State4), + {ok, State5}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close socket", + cmd => fun(#{sock := Sock} = State) -> + ok = socket:close(Sock), + {ok, maps:remove(sock, State)} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + ClientSeq = + [ + %% *** Wait for start order part *** + #{desc => "await start", + cmd => fun(State) -> + {Tester, ServerSA} = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester, + server_sa => ServerSA}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester} = _State) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{lsa => LSA}} + end}, + #{desc => "open socket", + cmd => fun(#{domain := Domain} = State) -> + Sock = sock_open(Domain, dgram, udp), + SA = sock_sockname(Sock), + {ok, State#{sock => Sock, sa => SA}} + end}, + #{desc => "bind socket (to local address)", + cmd => fun(#{sock := Sock, lsa := LSA}) -> + case socket:bind(Sock, LSA) of + {ok, _Port} -> + ok; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, init), + ok + end}, + + %% The actual test + #{desc => "await continue (send request)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, send_req) + end}, + #{desc => "send request", + cmd => fun(#{sock := Sock, server_sa := Server, send := Send}) -> + Send(Sock, ?BASIC_REQ, Server) + end}, + #{desc => "announce ready (send request)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, send_req), + ok + end}, + #{desc => "await continue (recv)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv) + end}, + #{desc => "try recv reply (with nowait)", + cmd => fun(#{sock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, {select_info, Tag, RecvRef}} -> + {ok, State#{recv_stag => Tag, + recv_sref => RecvRef}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await select message", + cmd => fun(#{sock := Sock, recv_sref := RecvRef}) -> + receive + {'$socket', Sock, select, RecvRef} -> + ok + end + end}, + #{desc => "announce ready (select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, select), + ok + end}, + #{desc => "now read the data (reply)", + cmd => fun(#{sock := Sock, recv := Recv}) -> + case Recv(Sock) of + {ok, {_Src, ?BASIC_REP}} -> + ok; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv reply)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_rep), + ok + end}, + + %% Termination + #{desc => "await terminate (from tester)", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + State2 = maps:remove(tester, State), + State3 = maps:remove(recv_stag, State2), + State4 = maps:remove(recv_sref, State3), + {ok, State4}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close socket", + cmd => fun(#{sock := Sock} = State) -> + ok = socket:close(Sock), + {ok, maps:remove(sock, State)} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor client", + cmd => fun(#{client := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, ServerSA} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{server_sa => ServerSA}} + end}, + + %% Start the client + #{desc => "order client start", + cmd => fun(#{client := Pid, + server_sa := ServerSA} = _State) -> + ?SEV_ANNOUNCE_START(Pid, ServerSA), + ok + end}, + #{desc => "await client ready (init)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, init) + end}, + + %% The actual test + #{desc => "order server continue (recv)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await server ready (recv_select)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, recv_select) + end}, + + #{desc => "order client continue (send request)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, send_req), + ok + end}, + #{desc => "await client ready (send request)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, send_req) + end}, + #{desc => "await server ready (select)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, select) + end}, + #{desc => "await server ready (recv request)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, recv_req) + end}, + + #{desc => "order client continue (recv)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await client ready (recv_select)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, recv_select) + end}, + #{desc => "order server continue (send reply)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, send_reply), + ok + end}, + #{desc => "await server ready (send)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, send) + end}, + #{desc => "await client ready (select)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, select) + end}, + #{desc => "await client ready (recv reply)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, recv_rep) + end}, + + %% Terminations + #{desc => "order client to terminate", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await client termination", + cmd => fun(#{client := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(client, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "order server to terminate", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(server, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + ServerInitState = InitState, + Server = ?SEV_START("server", ServerSeq, ServerInitState), + + i("start client evaluator(s)"), + ClientInitState = InitState, + Client = ?SEV_START("client", ClientSeq, ClientInitState), + + i("start 'tester' evaluator"), + TesterInitState = #{server => Server#ev.pid, + client => Client#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + i("await evaluator"), + ok = ?SEV_AWAIT_FINISH([Server, Client, Tester]). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically send and receive using the "common" functions (send and recv) +%% on an IPv4 TCP (stream) socket. But we try to be async. That is, we use +%% the 'nowait' value for the Timeout argument (and await the eventual +%% select message). Note that we only do this for the recv, +%% since its much more difficult to "arrange" for send. +%% We *also* test async for accept. +api_a_send_and_recv_tcp4(suite) -> + []; +api_a_send_and_recv_tcp4(doc) -> + []; +api_a_send_and_recv_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(10)), + tc_try(api_a_send_and_recv_tcp4, + fun() -> + Send = fun(Sock, Data) -> + socket:send(Sock, Data) + end, + Recv = fun(Sock) -> + socket:recv(Sock, 0, nowait) + end, + InitState = #{domain => inet, + send => Send, + recv => Recv}, + ok = api_a_send_and_recv_tcp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically send and receive using the msg functions (sendmsg and recvmsg) +%% on an IPv4 TCP (stream) socket. But we try to be async. That is, we use +%% the 'nowait' value for the Timeout argument (and await the eventual +%% select message). Note that we only do this for the recvmsg, +%% since its much more difficult to "arrange" for sendmsg. +%% We *also* test async for accept. +api_a_sendmsg_and_recvmsg_tcp4(suite) -> + []; +api_a_sendmsg_and_recvmsg_tcp4(doc) -> + []; +api_a_sendmsg_and_recvmsg_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(10)), + tc_try(api_a_sendmsg_and_recvmsg_tcp4, + fun() -> + Send = fun(Sock, Data) -> + MsgHdr = #{iov => [Data]}, + socket:sendmsg(Sock, MsgHdr) + end, + Recv = fun(Sock) -> + case socket:recvmsg(Sock, nowait) of + {ok, #{addr := undefined, + iov := [Data]}} -> + {ok, Data}; + {ok, _} = OK -> + OK; + {select, _} = SELECT -> + SELECT; + {error, _} = ERROR -> + ERROR + end + end, + InitState = #{domain => inet, + send => Send, + recv => Recv}, + ok = api_a_send_and_recv_tcp(InitState) + end). + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_send_and_recv_tcp(InitState) -> + process_flag(trap_exit, true), + ServerSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{lsa => LSA}} + end}, + #{desc => "create listen socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{lsock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{lsock := LSock, lsa := LSA} = State) -> + case socket:bind(LSock, LSA) of + {ok, Port} -> + {ok, State#{lport => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "make listen socket", + cmd => fun(#{lsock := LSock}) -> + socket:listen(LSock) + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, lport := Port}) -> + ?SEV_ANNOUNCE_READY(Tester, init, Port), + ok + end}, + + %% The actual test + #{desc => "await continue (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, accept) + end}, + #{desc => "await connection (nowait)", + cmd => fun(#{lsock := LSock} = State) -> + case socket:accept(LSock, nowait) of + {select, {select_info, Tag, Ref}} -> + ?SEV_IPRINT("accept select: " + "~n Tag: ~p" + "~n Ref: ~p", [Tag, Ref]), + {ok, State#{accept_stag => Tag, + accept_sref => Ref}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (accept_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, accept_select), + ok + end}, + #{desc => "await select message", + cmd => fun(#{lsock := Sock, accept_sref := Ref}) -> + receive + {'$socket', Sock, select, Ref} -> + ok + end + end}, + #{desc => "announce ready (select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, select), + ok + end}, + #{desc => "await connection (again)", + cmd => fun(#{lsock := LSock} = State) -> + case socket:accept(LSock, nowait) of + {ok, Sock} -> + ?SEV_IPRINT("accepted: " + "~n Sock: ~p", [Sock]), + {ok, State#{csock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, accept), + ok + end}, + + #{desc => "await continue (recv request)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv_req) + end}, + #{desc => "try recv request (with nowait, expect select)", + cmd => fun(#{csock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, {select_info, Tag, Ref}} -> + ?SEV_IPRINT("recv select: " + "~n Tag: ~p" + "~n Ref: ~p", [Tag, Ref]), + {ok, State#{recv_stag => Tag, + recv_sref => Ref}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await select message", + cmd => fun(#{csock := Sock, recv_sref := RecvRef}) -> + receive + {'$socket', Sock, select, RecvRef} -> + ok + end + end}, + #{desc => "announce ready (select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, select), + ok + end}, + #{desc => "now read the data (request)", + cmd => fun(#{csock := Sock, recv := Recv} = _State) -> + case Recv(Sock) of + {ok, ?BASIC_REQ} -> + ok; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv request)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_req), + ok + end}, + + #{desc => "await continue (send reply)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, send_rep) + end}, + #{desc => "send reply", + cmd => fun(#{csock := Sock, send := Send}) -> + Send(Sock, ?BASIC_REP) + end}, + #{desc => "announce ready (send reply)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, send_rep), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close connection socket", + cmd => fun(#{csock := Sock}) -> + socket:close(Sock) + end}, + #{desc => "close listen socket", + cmd => fun(#{lsock := Sock}) -> + socket:close(Sock) + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + ClientSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + {Tester, Port} = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester, server_port => Port}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** The init part *** + #{desc => "which server (local) address", + cmd => fun(#{domain := Domain, server_port := Port} = State) -> + LSA = which_local_socket_addr(Domain), + SSA = LSA#{port => Port}, + {ok, State#{local_sa => LSA, server_sa => SSA}} + end}, + #{desc => "create socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{sock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{sock := Sock, local_sa := LSA} = _State) -> + case socket:bind(Sock, LSA) of + {ok, _Port} -> + ok; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, init), + ok + end}, + + %% *** The actual test *** + #{desc => "await continue (connect)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, connect) + end}, + #{desc => "connect to server", + cmd => fun(#{sock := Sock, server_sa := SSA}) -> + socket:connect(Sock, SSA) + end}, + #{desc => "announce ready (connect)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, connect), + ok + end}, + + #{desc => "await continue (send request)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, send_req) + end}, + #{desc => "send request (to server)", + cmd => fun(#{sock := Sock, send := Send}) -> + ok = Send(Sock, ?BASIC_REQ) + end}, + #{desc => "announce ready (send request)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, send_req), + ok + end}, + + #{desc => "try recv reply (with nowait, expect select)", + cmd => fun(#{sock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, {select_info, Tag, Ref}} -> + ?SEV_IPRINT("recv select: " + "~n Tag: ~p" + "~n Ref: ~p", [Tag, Ref]), + {ok, State#{recv_stag => Tag, + recv_sref => Ref}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await select message", + cmd => fun(#{sock := Sock, recv_sref := RecvRef}) -> + receive + {'$socket', Sock, select, RecvRef} -> + ok + end + end}, + #{desc => "announce ready (select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, select), + ok + end}, + #{desc => "now read the data (reply)", + cmd => fun(#{sock := Sock, recv := Recv}) -> + {ok, ?BASIC_REP} = Recv(Sock), + ok + end}, + #{desc => "announce ready (recv reply)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_rep), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close socket", + cmd => fun(#{sock := Sock}) -> + socket:close(Sock) + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor client", + cmd => fun(#{client := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, Port} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{server_port => Port}} + end}, + + %% Start the client + #{desc => "order client start", + cmd => fun(#{client := Pid, server_port := Port} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Port), + ok + end}, + #{desc => "await client ready (init)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, init) + end}, + + %% *** The actual test *** + #{desc => "order server to continue (with accept)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Server, accept), + ok + end}, + #{desc => "await server ready (accept select)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, accept_select) + end}, + #{desc => "order client to continue (with connect)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, connect), + ok + end}, + #{desc => "await server ready (select)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, select) + end}, + #{desc => "await server ready (accept)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, accept) + end}, + #{desc => "await client ready (connect)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, client, connect) + end}, + + #{desc => "order server to continue (recv request)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv_req), + ok + end}, + #{desc => "await server ready (recv select)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, recv_select) + end}, + #{desc => "order client to continue (send request)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, send_req), + ok + end}, + #{desc => "await client ready (send request)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_AWAIT_READY(Client, client, send_req) + end}, + #{desc => "await server ready (select)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, select) + end}, + #{desc => "await server ready (recv request)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, recv_req) + end}, + + #{desc => "order client to continue (recv reply)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv_rep), + ok + end}, + #{desc => "await client ready (recv select)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, client, recv_select) + end}, + #{desc => "order server to continue (send reply)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, send_rep), + ok + end}, + #{desc => "await server ready (send reply)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, send_rep) + end}, + #{desc => "await client ready (select)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, client, select) + end}, + #{desc => "await client ready (reply recv)", + cmd => fun(#{client := Client} = _State) -> + ?SEV_AWAIT_READY(Client, client, recv_rep) + end}, + + + %% *** Termination *** + #{desc => "order client to terminate", + cmd => fun(#{client := Client} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Client), + ok + end}, + #{desc => "await client termination", + cmd => fun(#{client := Client} = State) -> + ?SEV_AWAIT_TERMINATION(Client), + State1 = maps:remove(client, State), + {ok, State1} + end}, + #{desc => "order server to terminate", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Server), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Server} = State) -> + ?SEV_AWAIT_TERMINATION(Server), + State1 = maps:remove(server, State), + {ok, State1} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + Server = ?SEV_START("server", ServerSeq, InitState), + + i("start client evaluator"), + Client = ?SEV_START("client", ClientSeq, InitState), + + i("start tester evaluator"), + TesterInitState = #{server => Server#ev.pid, + client => Client#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + i("await evaluator(s)"), + ok = ?SEV_AWAIT_FINISH([Server, Client, Tester]). + + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make an async (Timeout = nowait) call to recvfrom, +%% wait some time and then cancel. +%% +api_a_recvfrom_cancel_udp4(suite) -> + []; +api_a_recvfrom_cancel_udp4(doc) -> + []; +api_a_recvfrom_cancel_udp4(_Config) when is_list(_Config) -> + ?TT(?SECS(10)), + tc_try(api_a_recvfrom_cancel_udp4, + fun() -> + Recv = fun(Sock) -> + case socket:recvfrom(Sock, 0, nowait) of + {ok, _} = OK -> + OK; + {select, _} = SELECT -> + SELECT; + {error, _} = ERROR -> + ERROR + end + end, + InitState = #{domain => inet, + recv => Recv}, + ok = api_a_recv_cancel_udp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make an async (Timeout = nowait) call to recvmsg, +%% wait some time and then cancel. +%% +api_a_recvmsg_cancel_udp4(suite) -> + []; +api_a_recvmsg_cancel_udp4(doc) -> + []; +api_a_recvmsg_cancel_udp4(_Config) when is_list(_Config) -> + ?TT(?SECS(10)), + tc_try(api_a_recvmsg_cancel_udp4, + fun() -> + Recv = fun(Sock) -> + case socket:recvmsg(Sock, nowait) of + {ok, _} = OK -> + OK; + {select, _} = SELECT -> + SELECT; + {error, _} = ERROR -> + ERROR + end + end, + InitState = #{domain => inet, + recv => Recv}, + ok = api_a_recv_cancel_udp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_recv_cancel_udp(InitState) -> + ServerSeq = + [ + %% *** Wait for start order part *** + #{desc => "await start", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester} = _State) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{local_sa => LSA}} + end}, + #{desc => "create socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, dgram, udp) of + {ok, Sock} -> + {ok, State#{sock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind socket (to local address)", + cmd => fun(#{sock := Sock, local_sa := LSA} = State) -> + case socket:bind(Sock, LSA) of + {ok, Port} -> + {ok, State#{port => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, local_sa := LSA, port := Port}) -> + ServerSA = LSA#{port => Port}, + ?SEV_ANNOUNCE_READY(Tester, init, ServerSA), + ok + end}, + + %% The actual test + #{desc => "await continue (recv)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv) + end}, + #{desc => "try recv request (with nowait, expect select)", + cmd => fun(#{sock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, SelectInfo} -> + {ok, State#{recv_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await select message (without success)", + cmd => fun(#{sock := Sock}) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}} + after 5000 -> + ok + end + end}, + #{desc => "announce ready (no select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, no_select), + ok + end}, + #{desc => "await continue (cancel)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, cancel) + end}, + #{desc => "cancel", + cmd => fun(#{sock := Sock, recv_select_info := SelectInfo}) -> + ok = socket:cancel(Sock, SelectInfo) + end}, + #{desc => "announce ready (cancel)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, cancel), + ok + end}, + + %% Termination + #{desc => "await terminate (from tester)", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + State2 = maps:remove(tester, State), + State3 = maps:remove(recv_stag, State2), + State4 = maps:remove(recv_sref, State3), + State5 = maps:remove(req_src, State4), + {ok, State5}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close socket", + cmd => fun(#{sock := Sock} = State) -> + ok = socket:close(Sock), + {ok, maps:remove(sock, State)} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, ServerSA} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{server_sa => ServerSA}} + end}, + + %% The actual test + #{desc => "order server continue (recv)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await server ready (recv select)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, recv_select) + end}, + #{desc => "await server ready (no select)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, no_select) + end}, + #{desc => "order server continue (cancel)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, cancel), + ok + end}, + #{desc => "await server ready (cancel)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, cancel) + end}, + + %% Terminations + #{desc => "order server to terminate", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(server, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + ServerInitState = InitState, + Server = ?SEV_START("server", ServerSeq, ServerInitState), + + i("start 'tester' evaluator"), + TesterInitState = #{server => Server#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + i("await evaluator"), + ok = ?SEV_AWAIT_FINISH([Server, Tester]). + + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make an async (Timeout = nowait) call to accept, +%% wait some time and then cancel. +%% +api_a_accept_cancel_tcp4(suite) -> + []; +api_a_accept_cancel_tcp4(doc) -> + []; +api_a_accept_cancel_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(10)), + tc_try(api_a_accept_cancel_tcp4, + fun() -> + Accept = fun(Sock) -> + case socket:accept(Sock, nowait) of + {ok, _} = OK -> + OK; + {select, _} = SELECT -> + SELECT; + {error, _} = ERROR -> + ERROR + end + end, + InitState = #{domain => inet, + accept => Accept}, + ok = api_a_accept_cancel_tcp(InitState) + end). + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_accept_cancel_tcp(InitState) -> + process_flag(trap_exit, true), + ServerSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{lsa => LSA}} + end}, + #{desc => "create listen socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{lsock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{lsock := LSock, lsa := LSA} = State) -> + case socket:bind(LSock, LSA) of + {ok, Port} -> + {ok, State#{lport => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "make listen socket", + cmd => fun(#{lsock := LSock}) -> + socket:listen(LSock) + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, lport := Port}) -> + ?SEV_ANNOUNCE_READY(Tester, init, Port), + ok + end}, + + %% The actual test + #{desc => "await continue (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, accept) + end}, + #{desc => "await connection (nowait)", + cmd => fun(#{lsock := LSock, accept := Accept} = State) -> + case Accept(LSock) of + {select, {select_info, T, R} = SelectInfo} -> + ?SEV_IPRINT("accept select: " + "~n T: ~p" + "~n R: ~p", [T, R]), + {ok, State#{accept_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (accept_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, accept_select), + ok + end}, + #{desc => "await select message (without success)", + cmd => fun(#{lsock := Sock}) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}} + after 5000 -> + ok + end + end}, + #{desc => "announce ready (no select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, no_select), + ok + end}, + #{desc => "await continue (cancel)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, cancel) + end}, + #{desc => "cancel", + cmd => fun(#{lsock := Sock, accept_select_info := SelectInfo}) -> + ok = socket:cancel(Sock, SelectInfo) + end}, + #{desc => "announce ready (cancel)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, cancel), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close listen socket", + cmd => fun(#{lsock := Sock}) -> + socket:close(Sock) + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, Port} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{server_port => Port}} + end}, + + %% *** The actual test *** + #{desc => "order server to continue (with accept)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Server, accept), + ok + end}, + #{desc => "await server ready (accept select)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, accept_select) + end}, + #{desc => "await server ready (no select)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, no_select) + end}, + #{desc => "order server to continue (cancel)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, cancel), + ok + end}, + #{desc => "await server ready (cancel)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, cancel) + end}, + + %% *** Termination *** + #{desc => "order server to terminate", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Server), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Server} = State) -> + ?SEV_AWAIT_TERMINATION(Server), + State1 = maps:remove(server, State), + {ok, State1} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + Server = ?SEV_START("server", ServerSeq, InitState), + + i("start tester evaluator"), + TesterInitState = #{server => Server#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + i("await evaluator(s)"), + ok = ?SEV_AWAIT_FINISH([Server, Tester]). + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make an async (Timeout = nowait) call to recv, +%% wait some time and then cancel. +%% +api_a_recv_cancel_tcp4(suite) -> + []; +api_a_recv_cancel_tcp4(doc) -> + []; +api_a_recv_cancel_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(10)), + tc_try(api_a_recv_cancel_tcp4, + fun() -> + Recv = fun(Sock) -> + socket:recv(Sock, 0, nowait) + end, + InitState = #{domain => inet, + recv => Recv}, + ok = api_a_recv_cancel_tcp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make an async (Timeout = nowait) call to recvmsg, +%% wait some time and then cancel. +%% +api_a_recvmsg_cancel_tcp4(suite) -> + []; +api_a_recvmsg_cancel_tcp4(doc) -> + []; +api_a_recvmsg_cancel_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(10)), + tc_try(api_a_recvmsg_cancel_tcp4, + fun() -> + Recv = fun(Sock) -> + socket:recvmsg(Sock, nowait) + end, + InitState = #{domain => inet, + recv => Recv}, + ok = api_a_recv_cancel_tcp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_recv_cancel_tcp(InitState) -> + process_flag(trap_exit, true), + ServerSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{lsa => LSA}} + end}, + #{desc => "create listen socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{lsock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{lsock := LSock, lsa := LSA} = State) -> + case socket:bind(LSock, LSA) of + {ok, Port} -> + {ok, State#{lport => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "make listen socket", + cmd => fun(#{lsock := LSock}) -> + socket:listen(LSock) + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, lport := Port}) -> + ?SEV_ANNOUNCE_READY(Tester, init, Port), + ok + end}, + + %% The actual test + #{desc => "await continue (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, accept) + end}, + #{desc => "await connection (nowait)", + cmd => fun(#{lsock := LSock} = State) -> + case socket:accept(LSock) of + {ok, CSock} -> + {ok, State#{csock => CSock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, accept), + ok + end}, + + #{desc => "await continue (nowait recv)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv) + end}, + #{desc => "try recv request (with nowait, expect select)", + cmd => fun(#{csock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, {select_info, T, R} = SelectInfo} -> + ?SEV_IPRINT("recv select: " + "~n Tag: ~p" + "~n Ref: ~p", [T, R]), + {ok, State#{recv_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await select message", + cmd => fun(#{csock := Sock}) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}} + after 5000 -> + ok + end + end}, + #{desc => "announce ready (no select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, no_select), + ok + end}, + #{desc => "await continue (cancel)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, cancel) + end}, + #{desc => "cancel", + cmd => fun(#{csock := Sock, recv_select_info := SelectInfo}) -> + ok = socket:cancel(Sock, SelectInfo) + end}, + #{desc => "announce ready (cancel)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, cancel), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close connection socket", + cmd => fun(#{csock := Sock}) -> + socket:close(Sock) + end}, + #{desc => "close listen socket", + cmd => fun(#{lsock := Sock}) -> + socket:close(Sock) + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + ClientSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + {Tester, Port} = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester, server_port => Port}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** The init part *** + #{desc => "which server (local) address", + cmd => fun(#{domain := Domain, server_port := Port} = State) -> + LSA = which_local_socket_addr(Domain), + SSA = LSA#{port => Port}, + {ok, State#{local_sa => LSA, server_sa => SSA}} + end}, + #{desc => "create socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{sock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{sock := Sock, local_sa := LSA} = _State) -> + case socket:bind(Sock, LSA) of + {ok, _Port} -> + ok; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, init), + ok + end}, + + %% *** The actual test *** + #{desc => "await continue (connect)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, connect) + end}, + #{desc => "connect to server", + cmd => fun(#{sock := Sock, server_sa := SSA}) -> + socket:connect(Sock, SSA) + end}, + #{desc => "announce ready (connect)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, connect), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close socket", + cmd => fun(#{sock := Sock}) -> + socket:close(Sock) + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor client", + cmd => fun(#{client := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, Port} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{server_port => Port}} + end}, + + %% Start the client + #{desc => "order client start", + cmd => fun(#{client := Pid, server_port := Port} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Port), + ok + end}, + #{desc => "await client ready (init)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, init) + end}, + + %% *** The actual test *** + #{desc => "order server to continue (with accept)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Server, accept), + ok + end}, + #{desc => "order client to continue (connect)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, connect), + ok + end}, + #{desc => "await client ready (connect)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, client, connect) + end}, + #{desc => "await server ready (accept)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, accept) + end}, + + #{desc => "order server to continue (recv)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await server ready (recv select)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, recv_select) + end}, + #{desc => "await server ready (no select)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, no_select) + end}, + #{desc => "order server to continue (send request)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, cancel), + ok + end}, + #{desc => "await server ready (cancel)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, server, cancel) + end}, + + %% *** Termination *** + #{desc => "order client to terminate", + cmd => fun(#{client := Client} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Client), + ok + end}, + #{desc => "await client termination", + cmd => fun(#{client := Client} = State) -> + ?SEV_AWAIT_TERMINATION(Client), + State1 = maps:remove(client, State), + {ok, State1} + end}, + #{desc => "order server to terminate", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Server), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Server} = State) -> + ?SEV_AWAIT_TERMINATION(Server), + State1 = maps:remove(server, State), + {ok, State1} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + Server = ?SEV_START("server", ServerSeq, InitState), + + i("start client evaluator"), + Client = ?SEV_START("client", ClientSeq, InitState), + + i("start tester evaluator"), + TesterInitState = #{server => Server#ev.pid, + client => Client#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + i("await evaluator(s)"), + ok = ?SEV_AWAIT_FINISH([Server, Client, Tester]). + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make multiple async (Timeout = nowait) call(s) to recvfrom +%% (from *several* processes), wait some time and then cancel. +%% This should result in abort messages to the 'other' processes. +%% +api_a_mrecvfrom_cancel_udp4(suite) -> + []; +api_a_mrecvfrom_cancel_udp4(doc) -> + []; +api_a_mrecvfrom_cancel_udp4(_Config) when is_list(_Config) -> + ?TT(?SECS(20)), + tc_try(api_a_mrecvfrom_cancel_udp4, + fun() -> + Recv = fun(Sock) -> + case socket:recvfrom(Sock, 0, nowait) of + {ok, _} = OK -> + OK; + {select, _} = SELECT -> + SELECT; + {error, _} = ERROR -> + ERROR + end + end, + InitState = #{domain => inet, + recv => Recv}, + ok = api_a_mrecv_cancel_udp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make multiple async (Timeout = nowait) call(s) to recvmsg +%% (from *several* processes), wait some time and then cancel. +%% This should result in abort messages to the 'other' processes. +%% +api_a_mrecvmsg_cancel_udp4(suite) -> + []; +api_a_mrecvmsg_cancel_udp4(doc) -> + []; +api_a_mrecvmsg_cancel_udp4(_Config) when is_list(_Config) -> + ?TT(?SECS(20)), + tc_try(api_a_mrecvmsg_cancel_udp4, + fun() -> + Recv = fun(Sock) -> + case socket:recvmsg(Sock, nowait) of + {ok, _} = OK -> + OK; + {select, _} = SELECT -> + SELECT; + {error, _} = ERROR -> + ERROR + end + end, + InitState = #{domain => inet, + recv => Recv}, + ok = api_a_mrecv_cancel_udp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_mrecv_cancel_udp(InitState) -> + ServerSeq = + [ + %% *** Wait for start order part *** + #{desc => "await start", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester} = _State) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{local_sa => LSA}} + end}, + #{desc => "create socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, dgram, udp) of + {ok, Sock} -> + {ok, State#{sock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind socket (to local address)", + cmd => fun(#{sock := Sock, local_sa := LSA} = State) -> + case socket:bind(Sock, LSA) of + {ok, Port} -> + {ok, State#{port => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, sock := Sock}) -> + ?SEV_ANNOUNCE_READY(Tester, init, Sock), + ok + end}, + + %% The actual test + #{desc => "await continue (recv)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv) + end}, + #{desc => "try recv request (with nowait, expect select)", + cmd => fun(#{sock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, SelectInfo} -> + {ok, State#{recv_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await abort message", + cmd => fun(#{sock := Sock, + recv_select_info := {select_info, _, Ref}} = State) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}}; + {'$socket', Sock, abort, {Ref, closed}} -> + {ok, maps:remove(sock, State)} + after 5000 -> + ?SEV_EPRINT("message queue: ~p", [mq()]), + {error, timeout} + end + end}, + #{desc => "announce ready (abort)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, abort), + ok + end}, + + %% Termination + #{desc => "await terminate (from tester)", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + State2 = maps:remove(tester, State), + State3 = maps:remove(recv_stag, State2), + State4 = maps:remove(recv_sref, State3), + State5 = maps:remove(req_src, State4), + {ok, State5}; + {error, _} = ERROR -> + ERROR + end + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + AltServerSeq = + [ + %% *** Wait for start order part *** + #{desc => "await start", + cmd => fun(State) -> + {Tester, Sock} = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester, sock => Sock}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester} = _State) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, init), + ok + end}, + + %% The actual test + #{desc => "await continue (recv)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv) + end}, + #{desc => "try recv request (with nowait, expect select)", + cmd => fun(#{sock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, SelectInfo} -> + {ok, State#{recv_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await abort message", + cmd => fun(#{sock := Sock, + recv_select_info := {select_info, _, Ref}} = State) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}}; + {'$socket', Sock, abort, {Ref, closed}} -> + {ok, maps:remove(sock, State)} + after 5000 -> + ?SEV_EPRINT("message queue: ~p", [mq()]), + {error, timeout} + end + end}, + #{desc => "announce ready (abort)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, abort), + ok + end}, + + %% Termination + #{desc => "await terminate (from tester)", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + ?SEV_IPRINT("terminating"), + State1 = maps:remove(recv_select_info, State), + State2 = maps:remove(tester, State1), + State3 = maps:remove(sock, State2), + {ok, State3}; + {error, _} = ERROR -> + ERROR + end + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor alt-server 1", + cmd => fun(#{alt_server1 := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor alt-server 2", + cmd => fun(#{alt_server2 := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, Sock} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{sock => Sock}} + end}, + + %% Start the alt-server 1 + #{desc => "order alt-server 1 start", + cmd => fun(#{alt_server1 := Pid, sock := Sock} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Sock), + ok + end}, + #{desc => "await alt-server 1 ready (init)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, alt_server1, init) + end}, + + %% Start the alt-server 2 + #{desc => "order alt-server 2 start", + cmd => fun(#{alt_server2 := Pid, sock := Sock} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Sock), + ok + end}, + #{desc => "await alt-server 2 ready (init)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, alt_server2, init) + end}, + + + %% The actual test + #{desc => "order server continue (recv)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await server ready (recv select)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, recv_select) + end}, + + #{desc => "order alt-server 1 continue (recv)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await alt-server 1 ready (recv select)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server1, recv_select) + end}, + + #{desc => "order alt-server 2 continue (recv)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await alt-server 2 ready (recv select)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server2, recv_select) + end}, + + ?SEV_SLEEP(?SECS(1)), + + #{desc => "close the socket", + cmd => fun(#{sock := Sock} = _State) -> + socket:close(Sock) + end}, + + #{desc => "await server ready (abort)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, abort) + end}, + #{desc => "await alt-server 1 ready (abort)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server1, abort) + end}, + #{desc => "await alt-server 2 ready (abort)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server2, abort) + end}, + + %% Terminations + #{desc => "order alt-server 2 to terminate", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await alt-server 2 termination", + cmd => fun(#{alt_server2 := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(alt_server2, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + #{desc => "order alt-server 1 to terminate", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await alt-server 1 termination", + cmd => fun(#{alt_server1 := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(alt_server1, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + #{desc => "order server to terminate", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(server, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + Server = ?SEV_START("server", ServerSeq, InitState), + + i("start alt-server 1 evaluator"), + AltServer1 = ?SEV_START("alt_server1", AltServerSeq, InitState), + + i("start alt-server 2 evaluator"), + AltServer2 = ?SEV_START("alt_server2", AltServerSeq, InitState), + + i("start 'tester' evaluator"), + TesterInitState = #{server => Server#ev.pid, + alt_server1 => AltServer1#ev.pid, + alt_server2 => AltServer2#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + i("await evaluator(s)"), + ok = ?SEV_AWAIT_FINISH([Server, AltServer1, AltServer2, Tester]). + + + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make multiple async (Timeout = nowait) call(s) to accept +%% (from *several* processes), wait some time and then cancel, +%% This should result in abort messages to the 'other' processes. +%% +api_a_maccept_cancel_tcp4(suite) -> + []; +api_a_maccept_cancel_tcp4(doc) -> + []; +api_a_maccept_cancel_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(20)), + tc_try(api_a_maccept_cancel_tcp4, + fun() -> + Accept = fun(Sock) -> + case socket:accept(Sock, nowait) of + {ok, _} = OK -> + OK; + {select, _} = SELECT -> + SELECT; + {error, _} = ERROR -> + ERROR + end + end, + InitState = #{domain => inet, + accept => Accept}, + ok = api_a_maccept_cancel_tcp(InitState) + end). + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_maccept_cancel_tcp(InitState) -> + process_flag(trap_exit, true), + ServerSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{lsa => LSA}} + end}, + #{desc => "create listen socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{lsock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{lsock := LSock, lsa := LSA} = State) -> + case socket:bind(LSock, LSA) of + {ok, Port} -> + {ok, State#{lport => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "make listen socket", + cmd => fun(#{lsock := LSock}) -> + socket:listen(LSock) + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, lsock := Sock}) -> + ?SEV_ANNOUNCE_READY(Tester, init, Sock), + ok + end}, + + %% The actual test + #{desc => "await continue (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, accept) + end}, + #{desc => "await connection (nowait)", + cmd => fun(#{lsock := LSock, accept := Accept} = State) -> + case Accept(LSock) of + {select, {select_info, T, R} = SelectInfo} -> + ?SEV_IPRINT("accept select: " + "~n T: ~p" + "~n R: ~p", [T, R]), + {ok, State#{accept_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (accept_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, accept_select), + ok + end}, + #{desc => "await select message (without success)", + cmd => fun(#{lsock := Sock, + accept_select_info := {select_info, _, Ref}} = State) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}}; + {'$socket', Sock, abort, {Ref, closed}} -> + {ok, maps:remove(lsock, State)} + after 5000 -> + ?SEV_EPRINT("message queue: ~p", [mq()]), + {error, timeout} + end + end}, + #{desc => "announce ready (abort)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, abort), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + AltServerSeq = + [ + %% *** Wait for start order part *** + #{desc => "await start", + cmd => fun(State) -> + {Tester, Sock} = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester, lsock => Sock}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester} = _State) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, init), + ok + end}, + + %% The actual test + #{desc => "await continue (accept)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, accept) + end}, + #{desc => "try accept request (with nowait, expect select)", + cmd => fun(#{lsock := Sock, accept := Accept} = State) -> + case Accept(Sock) of + {select, SelectInfo} -> + {ok, State#{accept_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (accept_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, accept_select), + ok + end}, + #{desc => "await abort message", + cmd => fun(#{lsock := Sock, + accept_select_info := {select_info, _, Ref}} = State) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}}; + {'$socket', Sock, abort, {Ref, closed}} -> + {ok, maps:remove(sock, State)} + after 5000 -> + ?SEV_EPRINT("message queue: ~p", [mq()]), + {error, timeout} + end + end}, + #{desc => "announce ready (abort)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, abort), + ok + end}, + + %% Termination + #{desc => "await terminate (from tester)", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + ?SEV_IPRINT("terminating"), + State1 = maps:remove(tester, State), + State2 = maps:remove(accept_select_info, State1), + State3 = maps:remove(lsock, State2), + {ok, State3}; + {error, _} = ERROR -> + ERROR + end + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor alt-server 1", + cmd => fun(#{alt_server1 := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor alt-server 2", + cmd => fun(#{alt_server2 := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, Sock} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{sock => Sock}} + end}, + + %% Start the alt-server 1 + #{desc => "order alt-server 1 start", + cmd => fun(#{alt_server1 := Pid, sock := Sock} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Sock), + ok + end}, + #{desc => "await alt-server 1 ready (init)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, alt_server1, init) + end}, + + %% Start the alt-server 2 + #{desc => "order alt-server 2 start", + cmd => fun(#{alt_server2 := Pid, sock := Sock} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Sock), + ok + end}, + #{desc => "await alt-server 2 ready (init)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, alt_server2, init) + end}, + + + %% *** The actual test *** + #{desc => "order server continue (accept)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, accept), + ok + end}, + #{desc => "await server ready (accept select)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, accept_select) + end}, + + #{desc => "order alt-server 1 continue (accept)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, accept), + ok + end}, + #{desc => "await alt-server 1 ready (accept select)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server1, accept_select) + end}, + + #{desc => "order alt-server 2 continue (accept)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, accept), + ok + end}, + #{desc => "await alt-server 2 ready (accept select)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server2, accept_select) + end}, + + ?SEV_SLEEP(?SECS(1)), + + #{desc => "close the socket", + cmd => fun(#{sock := Sock} = _State) -> + socket:close(Sock) + end}, + + #{desc => "await server ready (abort)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, abort) + end}, + #{desc => "await alt-server 1 ready (abort)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server1, abort) + end}, + #{desc => "await alt-server 2 ready (abort)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server2, abort) + end}, + + + %% *** Termination *** + #{desc => "order alt-server 2 to terminate", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await alt-server 2 termination", + cmd => fun(#{alt_server2 := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(alt_server2, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + #{desc => "order alt-server 1 to terminate", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await alt-server 1 termination", + cmd => fun(#{alt_server1 := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(alt_server1, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + #{desc => "order server to terminate", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Server), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Server} = State) -> + ?SEV_AWAIT_TERMINATION(Server), + State1 = maps:remove(server, State), + {ok, State1} + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + Server = ?SEV_START("server", ServerSeq, InitState), + + i("start alt-server 1 evaluator"), + AltServer1 = ?SEV_START("alt_server1", AltServerSeq, InitState), + + i("start alt-server 2 evaluator"), + AltServer2 = ?SEV_START("alt_server2", AltServerSeq, InitState), + + i("start tester evaluator"), + TesterInitState = #{server => Server#ev.pid, + alt_server1 => AltServer1#ev.pid, + alt_server2 => AltServer2#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + i("await evaluator(s)"), + ok = ?SEV_AWAIT_FINISH([Server, AltServer1, AltServer2, Tester]). + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make multiple async (Timeout = nowait) call(s) to recv +%% (from *several* processes), wait some time and then cancel, +%% This should result in abort messages to the 'other' processes. +%% +api_a_mrecv_cancel_tcp4(suite) -> + []; +api_a_mrecv_cancel_tcp4(doc) -> + []; +api_a_mrecv_cancel_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(20)), + tc_try(api_a_mrecv_cancel_tcp4, + fun() -> + Recv = fun(Sock) -> + socket:recv(Sock, 0, nowait) + end, + InitState = #{domain => inet, + recv => Recv}, + ok = api_a_mrecv_cancel_tcp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +%% Basically we make multiple async (Timeout = nowait) call(s) to recvmsg +%% (from *several* processes), wait some time and then cancel, +%% This should result in abort messages to the 'other' processes. +%% +api_a_mrecvmsg_cancel_tcp4(suite) -> + []; +api_a_mrecvmsg_cancel_tcp4(doc) -> + []; +api_a_mrecvmsg_cancel_tcp4(_Config) when is_list(_Config) -> + ?TT(?SECS(20)), + tc_try(api_a_mrecvmsg_cancel_tcp4, + fun() -> + Recv = fun(Sock) -> + socket:recvmsg(Sock, nowait) + end, + InitState = #{domain => inet, + recv => Recv}, + ok = api_a_mrecv_cancel_tcp(InitState) + end). + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + +api_a_mrecv_cancel_tcp(InitState) -> + process_flag(trap_exit, true), + ServerSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + Tester = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** Init part *** + #{desc => "which local address", + cmd => fun(#{domain := Domain} = State) -> + LSA = which_local_socket_addr(Domain), + {ok, State#{lsa => LSA}} + end}, + #{desc => "create listen socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{lsock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{lsock := LSock, lsa := LSA} = State) -> + case socket:bind(LSock, LSA) of + {ok, Port} -> + {ok, State#{lport => Port}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "make listen socket", + cmd => fun(#{lsock := LSock}) -> + socket:listen(LSock) + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester, lport := Port}) -> + ?SEV_ANNOUNCE_READY(Tester, init, Port), + ok + end}, + + %% The actual test + #{desc => "await continue (accept)", + cmd => fun(#{tester := Tester}) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, accept) + end}, + #{desc => "await connection (nowait)", + cmd => fun(#{lsock := LSock} = State) -> + case socket:accept(LSock) of + {ok, CSock} -> + {ok, State#{csock => CSock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (accept)", + cmd => fun(#{tester := Tester, csock := Sock}) -> + ?SEV_ANNOUNCE_READY(Tester, accept, Sock), + ok + end}, + + #{desc => "await continue (nowait recv)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv) + end}, + #{desc => "try recv request (with nowait, expect select)", + cmd => fun(#{csock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, {select_info, T, R} = SelectInfo} -> + ?SEV_IPRINT("recv select: " + "~n Tag: ~p" + "~n Ref: ~p", [T, R]), + {ok, State#{recv_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await select message", + cmd => fun(#{csock := Sock, + recv_select_info := {select_info, _, Ref}} = State) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}}; + {'$socket', Sock, abort, {Ref, closed}} -> + {ok, maps:remove(sock, State)} + after 5000 -> + ok + end + end}, + #{desc => "announce ready (abort)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, abort), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close listen socket", + cmd => fun(#{lsock := Sock}) -> + socket:close(Sock) + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + AltServerSeq = + [ + %% *** Wait for start order part *** + #{desc => "await start", + cmd => fun(State) -> + {Tester, Sock} = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester, sock => Sock}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester} = _State) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, init), + ok + end}, + + %% The actual test + #{desc => "await continue (recv)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, recv) + end}, + #{desc => "try recv request (with nowait, expect select)", + cmd => fun(#{sock := Sock, recv := Recv} = State) -> + case Recv(Sock) of + {select, SelectInfo} -> + {ok, State#{recv_select_info => SelectInfo}}; + {ok, X} -> + {error, {unexpected_select_info, X}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (recv_select)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, recv_select), + ok + end}, + #{desc => "await abort message", + cmd => fun(#{sock := Sock, + recv_select_info := {select_info, _, Ref}} = State) -> + receive + {'$socket', Sock, select, Ref} -> + {error, {unexpected_select, Ref}}; + {'$socket', Sock, abort, {Ref, closed}} -> + {ok, maps:remove(sock, State)} + after 5000 -> + ?SEV_EPRINT("message queue: ~p", [mq()]), + {error, timeout} + end + end}, + #{desc => "announce ready (abort)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, abort), + ok + end}, + + %% Termination + #{desc => "await terminate (from tester)", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + ?SEV_IPRINT("terminating"), + State1 = maps:remove(recv_select_info, State), + State2 = maps:remove(tester, State1), + State3 = maps:remove(sock, State2), + {ok, State3}; + {error, _} = ERROR -> + ERROR + end + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + ClientSeq = + [ + %% *** Wait for start order *** + #{desc => "await start (from tester)", + cmd => fun(State) -> + {Tester, Port} = ?SEV_AWAIT_START(), + {ok, State#{tester => Tester, server_port => Port}} + end}, + #{desc => "monitor tester", + cmd => fun(#{tester := Tester}) -> + _MRef = erlang:monitor(process, Tester), + ok + end}, + + %% *** The init part *** + #{desc => "which server (local) address", + cmd => fun(#{domain := Domain, server_port := Port} = State) -> + LSA = which_local_socket_addr(Domain), + SSA = LSA#{port => Port}, + {ok, State#{local_sa => LSA, server_sa => SSA}} + end}, + #{desc => "create socket", + cmd => fun(#{domain := Domain} = State) -> + case socket:open(Domain, stream, tcp) of + {ok, Sock} -> + {ok, State#{sock => Sock}}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "bind to local address", + cmd => fun(#{sock := Sock, local_sa := LSA} = _State) -> + case socket:bind(Sock, LSA) of + {ok, _Port} -> + ok; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "announce ready (init)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, init), + ok + end}, + + %% *** The actual test *** + #{desc => "await continue (connect)", + cmd => fun(#{tester := Tester} = _State) -> + ?SEV_AWAIT_CONTINUE(Tester, tester, connect) + end}, + #{desc => "connect to server", + cmd => fun(#{sock := Sock, server_sa := SSA}) -> + socket:connect(Sock, SSA) + end}, + #{desc => "announce ready (connect)", + cmd => fun(#{tester := Tester}) -> + ?SEV_ANNOUNCE_READY(Tester, connect), + ok + end}, + + %% *** Termination *** + #{desc => "await terminate", + cmd => fun(#{tester := Tester} = State) -> + case ?SEV_AWAIT_TERMINATE(Tester, tester) of + ok -> + {ok, maps:remove(tester, State)}; + {error, _} = ERROR -> + ERROR + end + end}, + #{desc => "close socket", + cmd => fun(#{sock := Sock}) -> + socket:close(Sock) + end}, + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + TesterSeq = + [ + %% *** Init part *** + #{desc => "monitor server", + cmd => fun(#{server := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor alt-server 1", + cmd => fun(#{alt_server1 := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor alt-server 2", + cmd => fun(#{alt_server2 := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + #{desc => "monitor client", + cmd => fun(#{client := Pid} = _State) -> + _MRef = erlang:monitor(process, Pid), + ok + end}, + + %% Start the server + #{desc => "order server start", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_START(Pid), + ok + end}, + #{desc => "await server ready (init)", + cmd => fun(#{server := Pid} = State) -> + {ok, Port} = ?SEV_AWAIT_READY(Pid, server, init), + {ok, State#{server_port => Port}} + end}, + + %% Start the client + #{desc => "order client start", + cmd => fun(#{client := Pid, server_port := Port} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Port), + ok + end}, + #{desc => "await client ready (init)", + cmd => fun(#{client := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, client, init) + end}, + + #{desc => "order server to continue (with accept)", + cmd => fun(#{server := Server} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Server, accept), + ok + end}, + #{desc => "order client to continue (connect)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, connect), + ok + end}, + #{desc => "await client ready (connect)", + cmd => fun(#{client := Pid} = _State) -> + ?SEV_AWAIT_READY(Pid, client, connect) + end}, + #{desc => "await server ready (accept)", + cmd => fun(#{server := Pid} = State) -> + {ok, Sock} = ?SEV_AWAIT_READY(Pid, server, accept), + {ok, State#{sock => Sock}} + end}, + + %% Start the alt server 1 + #{desc => "order alt-server 1 start", + cmd => fun(#{alt_server1 := Pid, sock := Sock} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Sock), + ok + end}, + #{desc => "await alt-server 1 ready (init)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server1, init) + end}, + + %% Start the alt server 2 + #{desc => "order alt-server 2 start", + cmd => fun(#{alt_server2 := Pid, sock := Sock} = _State) -> + ?SEV_ANNOUNCE_START(Pid, Sock), + ok + end}, + #{desc => "await alt-server 2 ready (init)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server2, init) + end}, + + + %% *** The actual test *** + #{desc => "order server continue (recv)", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await server ready (recv select)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, recv_select) + end}, + + #{desc => "order alt-server 1 continue (recv)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await alt-server 1 ready (recv select)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server1, recv_select) + end}, + + #{desc => "order alt-server 2 continue (recv)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ?SEV_ANNOUNCE_CONTINUE(Pid, recv), + ok + end}, + #{desc => "await alt-server 2 ready (recv select)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server2, recv_select) + end}, + + ?SEV_SLEEP(?SECS(1)), + + #{desc => "close the socket", + cmd => fun(#{sock := Sock} = _State) -> + socket:close(Sock) + end}, + + #{desc => "await server ready (abort)", + cmd => fun(#{server := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, server, abort) + end}, + #{desc => "await alt-server 1 ready (abort)", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server1, abort) + end}, + #{desc => "await alt-server 2 ready (abort)", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ok = ?SEV_AWAIT_READY(Pid, alt_server2, abort) + end}, + + %% Terminations + #{desc => "order client to terminate", + cmd => fun(#{client := Client} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Client), + ok + end}, + #{desc => "await client termination", + cmd => fun(#{client := Client} = State) -> + ?SEV_AWAIT_TERMINATION(Client), + State1 = maps:remove(client, State), + {ok, State1} + end}, + + #{desc => "order alt-server 2 to terminate", + cmd => fun(#{alt_server2 := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await alt-server 2 termination", + cmd => fun(#{alt_server2 := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(alt_server2, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + #{desc => "order alt-server 1 to terminate", + cmd => fun(#{alt_server1 := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await alt-server 1 termination", + cmd => fun(#{alt_server1 := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(alt_server1, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + #{desc => "order server to terminate", + cmd => fun(#{server := Pid} = _State) -> + ?SEV_ANNOUNCE_TERMINATE(Pid), + ok + end}, + #{desc => "await server termination", + cmd => fun(#{server := Pid} = State) -> + case ?SEV_AWAIT_TERMINATION(Pid) of + ok -> + State1 = maps:remove(server, State), + {ok, State1}; + {error, _} = ERROR -> + ERROR + end + end}, + + + %% *** We are done *** + ?SEV_FINISH_NORMAL + ], + + i("start server evaluator"), + Server = ?SEV_START("server", ServerSeq, InitState), + + i("start alt-server 1 evaluator"), + AltServer1 = ?SEV_START("alt_server1", AltServerSeq, InitState), + + i("start alt-server 2 evaluator"), + AltServer2 = ?SEV_START("alt_server2", AltServerSeq, InitState), + + i("start client evaluator"), + Client = ?SEV_START("client", ClientSeq, InitState), + + i("start tester evaluator"), + TesterInitState = #{server => Server#ev.pid, + alt_server1 => AltServer1#ev.pid, + alt_server2 => AltServer2#ev.pid, + client => Client#ev.pid}, + Tester = ?SEV_START("tester", TesterSeq, TesterInitState), + + i("await evaluator(s)"), + ok = ?SEV_AWAIT_FINISH([Server, AltServer1, AltServer2, Client, Tester]). + + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% %% %% API OPTIONS %% @@ -19206,7 +23089,7 @@ which_ttest_runtime_env(false) -> %% ms: milliseconds %% s: seconds (default) %% m: minutes -which_ttest_runtime_env2([$m, $s | MS]) when (length(MS) > 0) -> +which_ttest_runtime_env2([$s, $m | MS]) when (length(MS) > 0) -> convert_time(MS, fun(X) -> X end); which_ttest_runtime_env2([$m | M]) when (length(M) > 0) -> convert_time(M, fun(X) -> ?MINS(X) end); @@ -19631,12 +23514,23 @@ ttest_tcp(InitState) -> %% Present the results #{desc => "present the results", - cmd => fun(#{result := Result} = State) -> + cmd => fun(#{result := Result, + domain := Domain, + server_mod := ServerTrans, + server_active := ServerActive, + client_mod := ClientTrans, + client_active := ClientActive, + msg_id := MsgID} = State) -> case Result of #{status := ok, runtime := RunTime, cnt := Cnt, bcnt := BCnt} -> + ttest_report(Domain, + ServerTrans, ServerActive, + ClientTrans, ClientActive, + MsgID, + RunTime, BCnt, Cnt), ?SEV_IPRINT( "TTest results: " "~n Run Time: ~s" @@ -19729,7 +23623,9 @@ ttest_tcp_server_start(Node, _Domain, gen, Active) -> socket_test_ttest_tcp_server:start_monitor(Node, Transport, Active); ttest_tcp_server_start(Node, Domain, sock, Active) -> TransportMod = socket_test_ttest_tcp_socket, - Transport = {TransportMod, #{domain => Domain, method => plain}}, + Transport = {TransportMod, #{domain => Domain, + async => true, + method => plain}}, socket_test_ttest_tcp_server:start_monitor(Node, Transport, Active). ttest_tcp_server_stop(Pid) -> @@ -19751,7 +23647,9 @@ ttest_tcp_client_start(Node, Domain, sock, ServerInfo, Active, MsgID, MaxOutstanding, RunTime) -> TransportMod = socket_test_ttest_tcp_socket, - Transport = {TransportMod, #{domain => Domain, method => plain}}, + Transport = {TransportMod, #{domain => Domain, + async => true, + method => plain}}, socket_test_ttest_tcp_client:start_monitor(Node, Notify, Transport, @@ -19762,6 +23660,183 @@ ttest_tcp_client_start(Node, %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% +-define(TTEST_MANAGER, esock_ttest_manager). + +-record(ttest_report_id, + {domain :: socket:domain(), + serv_trans :: gen | sock, + serv_active :: once | boolean(), + client_trans :: gen | sock, + client_active :: once | boolean(), + msg_id :: small | medium | large}). + +-record(ttest_report, {id :: #ttest_report_id{}, + time :: non_neg_integer(), + bytes :: non_neg_integer(), + msgs :: non_neg_integer()}). + +-spec ttest_report(Domain :: socket:domain(), + ServTrans :: gen | sock, ServActive :: once | boolean(), + ClientTrans :: gen | sock, ClientActive :: once | boolean(), + MsgID :: 1 | 2 | 3, + RunTime :: non_neg_integer(), + NumBytes :: non_neg_integer(), + NumMsgs :: non_neg_integer()) -> ok. + +ttest_report(Domain, + ServTrans, ServActive, + ClientTrans, ClientActive, + MsgID, + RunTime, + NumBytes, + NumMsgs) -> + ID = #ttest_report_id{domain = Domain, + serv_trans = ServTrans, + serv_active = ServActive, + client_trans = ClientTrans, + client_active = ClientActive, + msg_id = ttest_msg_id_num_to_name(MsgID)}, + Report = #ttest_report{id = ID, + time = RunTime, + bytes = NumBytes, + msgs = NumMsgs}, + %% If we run just one test case, the group init has never been run + %% and therefor the ttest manager is not running (we also don't actually + %% care about collecting reports in that case). + (catch global:send(?TTEST_MANAGER, Report)), + ok. + +ttest_msg_id_num_to_name(1) -> + small; +ttest_msg_id_num_to_name(2) -> + medium; +ttest_msg_id_num_to_name(3) -> + large. + +ttest_manager_start() -> + Self = self(), + {Pid, MRef} = spawn_monitor(fun() -> ttest_manager_init(Self) end), + receive + {ttest_manager_started, Pid} -> + erlang:demonitor(MRef, [flush]), + ok; + {'DOWN', MRef, process, Pid, Reason} -> + exit({failed_starting, ttest_manager, Reason}) + after 5000 -> + exit(Pid, kill), + exit({failed_starting, ttest_manager, timeout}) + end. + +ttest_manager_stop() -> + case global:whereis_name(?TTEST_MANAGER) of + Pid when is_pid(Pid) -> + erlang:monitor(process, Pid), + global:send(?TTEST_MANAGER, stop), + receive + {'DOWN', _MRef, process, Pid, _} -> + ok + after 10000 -> + exit(Pid, kill), + ok + end; + _ -> + ok + end. + +ttest_manager_init(Parent) -> + yes = global:register_name(?TTEST_MANAGER, self()), + ets:new(?TTEST_MANAGER, + [{keypos, #ttest_report.id}, named_table, protected, ordered_set]), + Parent ! {ttest_manager_started, self()}, + ttest_manager_loop(). + +ttest_manager_loop() -> + receive + stop -> + ?LOGGER:format("manager stopping~n", []), + ttest_manager_done(); + + #ttest_report{id = _ID, + time = _RunTime, + bytes = _NumBytes, + msgs = _NumMsgs} = Report -> + true = ets:insert_new(?TTEST_MANAGER, Report), + ttest_manager_loop() + end. + +%% We are supposed to pretty print the result here... +ttest_manager_done() -> + format_reports(inet), + %% format_reports(inet6), + ets:delete(?TTEST_MANAGER), + exit(normal). + +format_reports(Domain) -> + ?LOGGER:format("Domain ~w reports:~n~n", [Domain]), + format_reports(Domain, small), + format_reports(Domain, medium), + format_reports(Domain, large). + +format_reports(Domain, MsgID) when is_atom(MsgID) -> + case which_ttest_reports(Domain, MsgID) of + [] -> + ?LOGGER:format(" No ~w reports~n~n", [MsgID]); + Reports -> + ?LOGGER:format(" ~w reports: ~n", [MsgID]), + lists:foreach(fun(R) -> format_report(R) end, Reports) + end. + +%% This should really be a table like this: +%% +%% client +%% server gen(false) gen(once) gen(true) sock(false) sock(once) sock(true) +%% gen(false) nnn +%% gen(once) nnn +%% gen(true) nnn +%% sock(false) nnn +%% sock(once) nnn +%% sock(true) nnn +%% +format_report(#ttest_report{id = #ttest_report_id{serv_trans = STrans, + serv_active = SActive, + client_trans = CTrans, + client_active = CActive}, + time = RunTime, + bytes = BCnt, + msgs = MCnt}) -> + ?LOGGER:format(" server ~w[~w] - client ~w[~w] => " + "~n Run Time: ~s" + "~n Bytes: ~s" + "~n Messages: ~s" + "~n", [STrans, SActive, CTrans, CActive, + ?TTEST_LIB:format_time(RunTime), + if ((BCnt =:= 0) orelse (RunTime =:= 0)) -> + ?TTEST_LIB:format("~w, ~w", + [BCnt, RunTime]); + true -> + ?TTEST_LIB:format("~p => ~p byte / ms", + [BCnt, BCnt div RunTime]) + end, + if (RunTime =:= 0) -> + "-"; + true -> + ?TTEST_LIB:format("~p => ~p iterations / ms", + [MCnt, MCnt div RunTime]) + end]), + ok. + + +which_ttest_reports(Domain, all) -> + [R || R = #ttest_report{id = #ttest_report_id{domain = D}} <- + ets:tab2list(?TTEST_MANAGER), Domain =:= D]; +which_ttest_reports(Domain, MsgID) -> + [R || R = #ttest_report{id = #ttest_report_id{domain = D, msg_id = MID}} <- + ets:tab2list(?TTEST_MANAGER), (Domain =:= D) andalso (MsgID =:= MID)]. + + + +%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% + %% This mechanism has only one purpose: So that we are able to kill %% the node-starter process if it takes to long. The node-starter %% runs on the local node. @@ -19982,9 +24057,13 @@ which_addr(Domain, [_|IFL]) -> which_addr2(_Domain, []) -> {error, no_address}; -which_addr2(inet = _Domain, [{addr, Addr}|_IFO]) when (size(Addr) =:= 4) -> +which_addr2(inet = _Domain, [{addr, Addr}|_IFO]) + when (size(Addr) =:= 4) andalso (element(1, Addr) =/= 127) -> {ok, Addr}; -which_addr2(inet6 = _Domain, [{addr, Addr}|_IFO]) when (size(Addr) =:= 8) -> +which_addr2(inet6 = _Domain, [{addr, Addr}|_IFO]) + when (size(Addr) =:= 8) andalso + (element(1, Addr) =/= 0) andalso + (element(1, Addr) =/= 16#fe80) -> {ok, Addr}; which_addr2(Domain, [_|IFO]) -> which_addr2(Domain, IFO). diff --git a/erts/emulator/test/socket_test_logger.erl b/erts/emulator/test/socket_test_logger.erl index 26610e9ef3..f5d4c8c7b2 100644 --- a/erts/emulator/test/socket_test_logger.erl +++ b/erts/emulator/test/socket_test_logger.erl @@ -43,7 +43,7 @@ start(Quiet) -> ok; undefined -> Self = self(), - Pid = spawn_link(fun() -> init(Self, Quiet) end), + Pid = spawn(fun() -> init(Self, Quiet) end), yes = global:register_name(?LOGGER, Pid), ok end. diff --git a/erts/emulator/test/socket_test_ttest_tcp_client_socket.erl b/erts/emulator/test/socket_test_ttest_tcp_client_socket.erl index ccace2a560..ca7eff4437 100644 --- a/erts/emulator/test/socket_test_ttest_tcp_client_socket.erl +++ b/erts/emulator/test/socket_test_ttest_tcp_client_socket.erl @@ -21,89 +21,95 @@ -module(socket_test_ttest_tcp_client_socket). -export([ - start/3, start/4, start/6, start/7, + start/4, start/5, start/7, start/8, stop/1 ]). -define(TRANSPORT_MOD, socket_test_ttest_tcp_socket). --define(MOD(D, M), {?TRANSPORT_MOD, #{domain => D, method => M}}). +-define(MOD(D, A, M), {?TRANSPORT_MOD, #{domain => D, + async => A, + method => M}}). -start(Method, ServerInfo, Active) +start(Method, Async, Active, ServerInfo) when is_list(ServerInfo) -> Domain = local, - socket_test_ttest_tcp_client:start_monitor(?MOD(Domain, Method), - ServerInfo, Active); -start(Method, ServerInfo = {Addr, _}, Active) + socket_test_ttest_tcp_client:start_monitor(?MOD(Domain, Async, Method), + Active, ServerInfo); +start(Method, Async, Active, ServerInfo = {Addr, _}) when is_tuple(Addr) andalso (size(Addr) =:= 4) -> Domain = inet, - socket_test_ttest_tcp_client:start_monitor(?MOD(Domain, Method), - ServerInfo, Active); -start(Method, ServerInfo = {Addr, _}, Active) + socket_test_ttest_tcp_client:start_monitor(?MOD(Domain, Async, Method), + Active, ServerInfo); +start(Method, Async, Active, ServerInfo = {Addr, _}) when is_tuple(Addr) andalso (size(Addr) =:= 8) -> Domain = inet6, - socket_test_ttest_tcp_client:start_monitor(?MOD(Domain, Method), - ServerInfo, Active). + socket_test_ttest_tcp_client:start_monitor(?MOD(Domain, Async, Method), + Active, ServerInfo). -start(Method, ServerInfo, Active, MsgID) +start(Method, Async, Active, ServerInfo, MsgID) when is_list(ServerInfo) -> %% This is just a simplification Domain = local, - socket_test_ttest_tcp_client:start(?MOD(Domain, Method), - ServerInfo, Active, MsgID); -start(Method, ServerInfo = {Addr, _}, Active, MsgID) + socket_test_ttest_tcp_client:start(?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID); +start(Method, Async, Active, ServerInfo = {Addr, _}, MsgID) when is_tuple(Addr) andalso (size(Addr) =:= 4) -> %% This is just a simplification Domain = inet, - socket_test_ttest_tcp_client:start(?MOD(Domain, Method), - ServerInfo, Active, MsgID); -start(Method, ServerInfo = {Addr, _}, Active, MsgID) + socket_test_ttest_tcp_client:start(?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID); +start(Method, Async, Active, ServerInfo = {Addr, _}, MsgID) when is_tuple(Addr) andalso (size(Addr) =:= 8) -> Domain = inet6, - socket_test_ttest_tcp_client:start(?MOD(Domain, Method), - ServerInfo, Active, MsgID). + socket_test_ttest_tcp_client:start(?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID). -start(Method, ServerInfo, Active, MsgID, MaxOutstanding, RunTime) +start(Method, Async, Active, ServerInfo, MsgID, MaxOutstanding, RunTime) when is_list(ServerInfo) -> Domain = local, socket_test_ttest_tcp_client:start(false, - ?MOD(Domain, Method), - ServerInfo, Active, + ?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID, MaxOutstanding, RunTime); -start(Method, ServerInfo = {Addr, _}, Active, MsgID, MaxOutstanding, RunTime) +start(Method, Async, Active, ServerInfo = {Addr, _}, + MsgID, MaxOutstanding, RunTime) when is_tuple(Addr) andalso (size(Addr) =:= 4) -> Domain = inet, socket_test_ttest_tcp_client:start(false, - ?MOD(Domain, Method), - ServerInfo, Active, + ?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID, MaxOutstanding, RunTime); -start(Method, ServerInfo = {Addr, _}, Active, MsgID, MaxOutstanding, RunTime) +start(Method, Async, Active, ServerInfo = {Addr, _}, + MsgID, MaxOutstanding, RunTime) when is_tuple(Addr) andalso (size(Addr) =:= 8) -> Domain = inet6, socket_test_ttest_tcp_client:start(false, - ?MOD(Domain, Method), - ServerInfo, Active, + ?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID, MaxOutstanding, RunTime). -start(Quiet, Method, ServerInfo, Active, MsgID, MaxOutstanding, RunTime) +start(Quiet, Async, Active, Method, ServerInfo, MsgID, MaxOutstanding, RunTime) when is_list(ServerInfo) -> Domain = local, socket_test_ttest_tcp_client:start(Quiet, - ?MOD(Domain, Method), - ServerInfo, Active, + ?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID, MaxOutstanding, RunTime); -start(Quiet, Method, ServerInfo = {Addr, _}, Active, MsgID, MaxOutstanding, RunTime) +start(Quiet, Async, Active, Method, ServerInfo = {Addr, _}, + MsgID, MaxOutstanding, RunTime) when is_tuple(Addr) andalso (size(Addr) =:= 4) -> Domain = inet, socket_test_ttest_tcp_client:start(Quiet, - ?MOD(Domain, Method), - ServerInfo, Active, + ?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID, MaxOutstanding, RunTime); -start(Quiet, Method, ServerInfo = {Addr, _}, Active, MsgID, MaxOutstanding, RunTime) +start(Quiet, Async, Active, Method, ServerInfo = {Addr, _}, + MsgID, MaxOutstanding, RunTime) when is_tuple(Addr) andalso (size(Addr) =:= 8) -> Domain = inet6, socket_test_ttest_tcp_client:start(Quiet, - ?MOD(Domain, Method), - ServerInfo, Active, + ?MOD(Domain, Async, Method), + Active, ServerInfo, MsgID, MaxOutstanding, RunTime). stop(Pid) -> diff --git a/erts/emulator/test/socket_test_ttest_tcp_server.erl b/erts/emulator/test/socket_test_ttest_tcp_server.erl index e916fcb93e..27b561d4b7 100644 --- a/erts/emulator/test/socket_test_ttest_tcp_server.erl +++ b/erts/emulator/test/socket_test_ttest_tcp_server.erl @@ -90,7 +90,9 @@ start_monitor(_, Transport, Active) -> start(Transport, Active) -> do_start(self(), Transport, Active). - +%% Note that the Async option is actually only "used" for the +%% socket transport module (it details how to implement the +%% active feature). do_start(Parent, Transport, Active) when is_pid(Parent) andalso (is_atom(Transport) orelse is_tuple(Transport)) andalso @@ -179,42 +181,59 @@ process_transport({Mod, Opts}, _Active) -> server_loop(State) -> - server_loop( server_handle_message( server_accept(State) ) ). + server_loop( server_handle_message( server_accept(State, ?ACC_TIMEOUT), 0) ). -server_accept(#{mod := Mod, - active := Active, - lsock := LSock, - handlers := Handlers} = State) -> - case Mod:accept(LSock, ?ACC_TIMEOUT) of +server_accept(#{mod := Mod, lsock := LSock} = State, Timeout) -> + case Mod:accept(LSock, Timeout) of {ok, Sock} -> - ?I("accepted connection from ~s", - [case Mod:peername(Sock) of - {ok, Peer} -> - format_peername(Peer); - {error, _} -> - "-" - end]), - {Pid, _} = handler_start(), - ?I("handler ~p started -> try transfer socket control", [Pid]), - case Mod:controlling_process(Sock, Pid) of - ok -> - maybe_start_stats_timer(State, Pid), - ?I("server-accept: handler ~p started", [Pid]), - handler_continue(Pid, Mod, Sock, Active), - Handlers2 = [Pid | Handlers], - State#{handlers => Handlers2}; - {error, CPReason} -> - (catch Mod:close(Sock)), - (catch Mod:close(LSock)), - exit({controlling_process, CPReason}) - end; - {error, timeout} -> + server_handle_accepted(State, Sock); + {error, timeout} when (Timeout =/= nowait) -> State; {error, AReason} -> (catch Mod:close(LSock)), exit({accept, AReason}) end. +%% server_accept(#{mod := Mod, +%% lsock := LSock} = State) -> +%% case Mod:accept(LSock, ?ACC_TIMEOUT) of +%% {ok, Sock} -> +%% server_handle_accepted(State, Sock); +%% {error, timeout} -> +%% State; +%% {error, AReason} -> +%% (catch Mod:close(LSock)), +%% exit({accept, AReason}) +%% end. + +server_handle_accepted(#{mod := Mod, + lsock := LSock, + active := Active, + handlers := Handlers} = State, + Sock) -> + ?I("accepted connection from ~s", + [case Mod:peername(Sock) of + {ok, Peer} -> + format_peername(Peer); + {error, _} -> + "-" + end]), + {Pid, _} = handler_start(), + ?I("handler ~p started -> try transfer socket control", [Pid]), + case Mod:controlling_process(Sock, Pid) of + ok -> + maybe_start_stats_timer(State, Pid), + ?I("server-accept: handler ~p started", [Pid]), + handler_continue(Pid, Mod, Sock, Active), + Handlers2 = [Pid | Handlers], + State#{handlers => Handlers2}; + {error, CPReason} -> + (catch Mod:close(Sock)), + (catch Mod:close(LSock)), + exit({controlling_process, CPReason}) + end. + + format_peername({Addr, Port}) -> case inet:gethostbyaddr(Addr) of {ok, #hostent{h_name = N}} -> @@ -237,7 +256,7 @@ start_stats_timer(Time, ProcStr, Pid) -> server_handle_message(#{mod := Mod, lsock := LSock, parent := Parent, - handlers := H} = State) -> + handlers := H} = State, Timeout) -> receive {timeout, _TRef, {stats, Interval, ProcStr, Pid}} -> case server_handle_stats(ProcStr, Pid) of @@ -247,7 +266,7 @@ server_handle_message(#{mod := Mod, ok end, State; - + {?MODULE, Ref, Parent, stop} -> reply(Parent, Ref, ok), lists:foreach(fun(P) -> handler_stop(P) end, H), @@ -257,7 +276,7 @@ server_handle_message(#{mod := Mod, {'DOWN', _MRef, process, Pid, Reason} -> server_handle_down(Pid, Reason, State) - after 0 -> + after Timeout -> State end. @@ -342,15 +361,15 @@ handler_init(Parent) -> ?I("received continue"), reply(Parent, Ref, ok), handler_initial_activation(Mod, Sock, Active), - handler_loop(#{parent => Parent, - mod => Mod, - sock => Sock, - active => Active, - start => ?T(), - mcnt => 0, - bcnt => 0, - last_reply => none, - acc => <<>>}) + handler_loop(#{parent => Parent, + mod => Mod, + sock => Sock, + active => Active, + start => ?T(), + mcnt => 0, + bcnt => 0, + last_reply => none, + acc => <<>>}) after 5000 -> ?I("timeout when message queue: " diff --git a/erts/emulator/test/socket_test_ttest_tcp_server_socket.erl b/erts/emulator/test/socket_test_ttest_tcp_server_socket.erl index d1de230637..4045bf4e4e 100644 --- a/erts/emulator/test/socket_test_ttest_tcp_server_socket.erl +++ b/erts/emulator/test/socket_test_ttest_tcp_server_socket.erl @@ -21,18 +21,26 @@ -module(socket_test_ttest_tcp_server_socket). -export([ - start/3, + start/4, stop/1 ]). -define(TRANSPORT_MOD, socket_test_ttest_tcp_socket). -%% -define(MOD(D, M), {?TRANSPORT_MOD, #{domain => D, +%% -define(MOD(M), {?TRANSPORT_MOD, #{async => false, %% method => M, %% stats_interval => 10000}}). --define(MOD(D, M), {?TRANSPORT_MOD, #{domain => D, method => M}}). +-define(MOD(D,M,A), {?TRANSPORT_MOD, #{domain => D, + async => A, + method => M}}). -start(Method, Domain, Active) -> - socket_test_ttest_tcp_server:start(?MOD(Domain, Method), Active). +start(Method, Domain, Async, Active) -> + socket_test_ttest_tcp_server:start(?MOD(Domain, Method, Async), Active). + %% {ok, {Pid, AddrPort}} -> + %% MRef = erlang:monitor(process, Pid), + %% {ok, {Pid, MRef, AddrPort}}; + %% {error, _} = ERROR -> + %% ERROR + %% end. stop(Pid) -> socket_test_ttest_tcp_server:stop(Pid). diff --git a/erts/emulator/test/socket_test_ttest_tcp_socket.erl b/erts/emulator/test/socket_test_ttest_tcp_socket.erl index cf68bfe591..3aa3b2c504 100644 --- a/erts/emulator/test/socket_test_ttest_tcp_socket.erl +++ b/erts/emulator/test/socket_test_ttest_tcp_socket.erl @@ -57,7 +57,7 @@ %% ========================================================================== -%% This does not really work. Its just a placeholder for the time beeing... +%% This does not really work. Its just a placeholder for the time being... %% getopt(Sock, Opt) when is_atom(Opt) -> %% socket:getopt(Sock, socket, Opt). @@ -68,22 +68,32 @@ %% ========================================================================== -accept(#{sock := LSock, opts := #{method := Method} = Opts}) -> +%% The way we use server async its no point in doing a async accept call +%% (we do never actually run the test with more than one client). +accept(#{sock := LSock, opts := #{async := Async, + method := Method} = Opts}) -> case socket:accept(LSock) of - {ok, Sock} -> + {ok, Sock} -> Self = self(), - Reader = spawn(fun() -> reader_init(Self, Sock, false, Method) end), + Reader = spawn(fun() -> + reader_init(Self, Sock, Async, false, Method) + end), maybe_start_stats_timer(Opts, Reader), {ok, #{sock => Sock, reader => Reader, method => Method}}; {error, _} = ERROR -> ERROR end. -accept(#{sock := LSock, opts := #{method := Method} = Opts}, Timeout) -> +%% If a timeout has been explictly specified, then we do not use +%% async here. We will pass it on to the reader process. +accept(#{sock := LSock, opts := #{async := Async, + method := Method} = Opts}, Timeout) -> case socket:accept(LSock, Timeout) of {ok, Sock} -> Self = self(), - Reader = spawn(fun() -> reader_init(Self, Sock, false, Method) end), + Reader = spawn(fun() -> + reader_init(Self, Sock, Async, false, Method) + end), maybe_start_stats_timer(Opts, Reader), {ok, #{sock => Sock, reader => Reader, method => Method}}; {error, _} = ERROR -> @@ -153,7 +163,8 @@ connect(Addr, Port, #{domain := Domain} = Opts) -> do_connect(LocalSA, ServerSA, Cleanup, Opts#{proto => tcp}). do_connect(LocalSA, ServerSA, Cleanup, #{domain := Domain, - proto := Proto, + proto := Proto, + async := Async, method := Method} = Opts) -> try begin @@ -181,7 +192,9 @@ do_connect(LocalSA, ServerSA, Cleanup, #{domain := Domain, throw({error, {connect, CReason}}) end, Self = self(), - Reader = spawn(fun() -> reader_init(Self, Sock, false, Method) end), + Reader = spawn(fun() -> + reader_init(Self, Sock, Async, false, Method) + end), maybe_start_stats_timer(Opts, Reader), {ok, #{sock => Sock, reader => Reader, method => Method}} end @@ -219,9 +232,9 @@ listen() -> listen(0). listen(Port) when is_integer(Port) -> - listen(Port, #{domain => inet, method => plain}); + listen(Port, #{domain => inet, async => false, method => plain}); listen(Path) when is_list(Path) -> - listen(Path, #{domain => local, method => plain}). + listen(Path, #{domain => local, async => false, method => plain}). listen(0, #{domain := local} = Opts) -> listen(mk_unique_path(), Opts); @@ -241,8 +254,10 @@ listen(Port, #{domain := Domain} = Opts) do_listen(SA, Cleanup, - #{domain := Domain, proto := Proto, method := Method} = Opts) - when (Method =:= plain) orelse (Method =:= msg) -> + #{domain := Domain, proto := Proto, + async := Async, method := Method} = Opts) + when (Method =:= plain) orelse (Method =:= msg) andalso + is_boolean(Async) -> try begin Sock = case socket:open(Domain, stream, Proto) of @@ -339,13 +354,18 @@ sockname(#{sock := Sock}) -> %% ========================================================================== -reader_init(ControllingProcess, Sock, Active, Method) +reader_init(ControllingProcess, Sock, Async, Active, Method) when is_pid(ControllingProcess) andalso + is_boolean(Async) andalso (is_boolean(Active) orelse (Active =:= once)) andalso ((Method =:= plain) orelse (Method =:= msg)) -> + put(verbose, false), MRef = erlang:monitor(process, ControllingProcess), reader_loop(#{ctrl_proc => ControllingProcess, ctrl_proc_mref => MRef, + async => Async, + select_info => undefined, + select_num => 0, % Count the number of select messages active => Active, sock => Sock, method => Method}). @@ -356,11 +376,11 @@ reader_loop(#{active := false, ctrl_proc := Pid} = State) -> receive {?MODULE, stop} -> - exit(normal); + reader_exit(State, stop); {?MODULE, Pid, controlling_process, NewPid} -> - MRef = maps:get(ctrl_proc_mref, State), - erlang:demonitor(MRef, [flush]), + OldMRef = maps:get(ctrl_proc_mref, State), + erlang:demonitor(OldMRef, [flush]), NewMRef = erlang:monitor(process, NewPid), Pid ! {?MODULE, self(), controlling_process}, reader_loop(State#{ctrl_proc => NewPid, @@ -371,10 +391,8 @@ reader_loop(#{active := false, {'DOWN', MRef, process, Pid, Reason} -> case maps:get(ctrl_proc_mref, State) of - MRef when (Reason =:= normal) -> - exit(normal); MRef -> - exit({controlling_process, Reason}); + reader_exit(State, {ctrl_exit, Reason}); _ -> reader_loop(State) end @@ -382,7 +400,8 @@ reader_loop(#{active := false, %% Read *once* and then change to false reader_loop(#{active := once, - sock := Sock, + async := false, + sock := Sock, method := Method, ctrl_proc := Pid} = State) -> case do_recv(Method, Sock) of @@ -392,25 +411,23 @@ reader_loop(#{active := once, {error, timeout} -> receive {?MODULE, stop} -> - exit(normal); + reader_exit(State, stop); {?MODULE, Pid, controlling_process, NewPid} -> - MRef = maps:get(ctrl_proc_mref, State), - erlang:demonitor(MRef, [flush]), - MRef = erlang:monitor(process, NewPid), + OldMRef = maps:get(ctrl_proc_mref, State), + erlang:demonitor(OldMRef, [flush]), + NewMRef = erlang:monitor(process, NewPid), Pid ! {?MODULE, self(), controlling_process}, reader_loop(State#{ctrl_proc => NewPid, - ctrl_proc_mref => MRef}); + ctrl_proc_mref => NewMRef}); {?MODULE, active, NewActive} -> reader_loop(State#{active => NewActive}); {'DOWN', MRef, process, Pid, Reason} -> case maps:get(ctrl_proc_mref, State) of - MRef when (Reason =:= normal) -> - exit(normal); - MRef -> - exit({controlling_process, Reason}); + MRef -> + reader_exit(State, {ctrl_exit, Reason}); _ -> reader_loop(State) end @@ -418,17 +435,86 @@ reader_loop(#{active := once, reader_loop(State) end; - {error, closed} -> + {error, closed} = E1 -> + Pid ! ?CLOSED_MSG(Sock, Method), + reader_exit(State, E1); + + {error, Reason} = E2 -> + Pid ! ?ERROR_MSG(Sock, Method, Reason), + reader_exit(State, E2) + end; +reader_loop(#{active := once, + async := true, + select_info := undefined, + sock := Sock, + method := Method, + ctrl_proc := Pid} = State) -> + case do_recv(Method, Sock, nowait) of + {select, SelectInfo} -> + reader_loop(State#{select_info => SelectInfo}); + {ok, Data} -> + Pid ! ?DATA_MSG(Sock, Method, Data), + reader_loop(State#{active => false}); + + {error, closed} = E1 -> Pid ! ?CLOSED_MSG(Sock, Method), - exit(normal); + reader_exit(State, E1); - {error, Reason} -> + {error, Reason} = E2 -> Pid ! ?ERROR_MSG(Sock, Method, Reason), - exit(Reason) + reader_exit(State, E2) + end; +reader_loop(#{active := once, + async := true, + select_info := {select_info, _, Ref}, + select_num := N, + sock := Sock, + method := Method, + ctrl_proc := Pid} = State) -> + receive + {?MODULE, stop} -> + reader_exit(State, stop); + + {?MODULE, Pid, controlling_process, NewPid} -> + OldMRef = maps:get(ctrl_proc_mref, State), + erlang:demonitor(OldMRef, [flush]), + NewMRef = erlang:monitor(process, NewPid), + Pid ! {?MODULE, self(), controlling_process}, + reader_loop(State#{ctrl_proc => NewPid, + ctrl_proc_mref => NewMRef}); + + {?MODULE, active, NewActive} -> + reader_loop(State#{active => NewActive}); + + {'DOWN', MRef, process, Pid, Reason} -> + case maps:get(ctrl_proc_mref, State) of + MRef -> + reader_exit(State, {ctrl_exit, Reason}); + _ -> + reader_loop(State) + end; + + {'$socket', Sock, select, Ref} -> + case do_recv(Method, Sock, nowait) of + {ok, Data} when is_binary(Data) -> + Pid ! ?DATA_MSG(Sock, Method, Data), + reader_loop(State#{active => false, + select_info => undefined, + select_num => N+1}); + + {error, closed} = E1 -> + Pid ! ?CLOSED_MSG(Sock, Method), + reader_exit(State, E1); + + {error, Reason} = E2 -> + Pid ! ?ERROR_MSG(Sock, Method, Reason), + reader_exit(State, E2) + end end; %% Read and forward data continuously reader_loop(#{active := true, + async := false, sock := Sock, method := Method, ctrl_proc := Pid} = State) -> @@ -439,25 +525,23 @@ reader_loop(#{active := true, {error, timeout} -> receive {?MODULE, stop} -> - exit(normal); + reader_exit(State, stop); {?MODULE, Pid, controlling_process, NewPid} -> - MRef = maps:get(ctrl_proc_mref, State), - erlang:demonitor(MRef, [flush]), - MRef = erlang:monitor(process, NewPid), + OldMRef = maps:get(ctrl_proc_mref, State), + erlang:demonitor(OldMRef, [flush]), + NewMRef = erlang:monitor(process, NewPid), Pid ! {?MODULE, self(), controlling_process}, reader_loop(State#{ctrl_proc => NewPid, - ctrl_proc_mref => MRef}); + ctrl_proc_mref => NewMRef}); {?MODULE, active, NewActive} -> reader_loop(State#{active => NewActive}); {'DOWN', MRef, process, Pid, Reason} -> case maps:get(ctrl_proc_mref, State) of - MRef when (Reason =:= normal) -> - exit(normal); MRef -> - exit({controlling_process, Reason}); + reader_exit(State, {ctrl_exit, Reason}); _ -> reader_loop(State) end @@ -465,27 +549,170 @@ reader_loop(#{active := true, reader_loop(State) end; - {error, closed} -> + {error, closed} = E1 -> Pid ! ?CLOSED_MSG(Sock, Method), - exit(normal); + reader_exit(State, E1); - {error, Reason} -> + {error, Reason} = E2 -> Pid ! ?ERROR_MSG(Sock, Method, Reason), - exit(Reason) + reader_exit(State, E2) + end; +reader_loop(#{active := true, + async := true, + select_info := undefined, + sock := Sock, + method := Method, + ctrl_proc := Pid} = State) -> + case do_recv(Method, Sock) of + {select, SelectInfo} -> + reader_loop(State#{select_info => SelectInfo}); + {ok, Data} -> + Pid ! ?DATA_MSG(Sock, Method, Data), + reader_loop(State); + + {error, closed} = E1 -> + Pid ! ?CLOSED_MSG(Sock, Method), + reader_exit(State, E1); + + {error, Reason} = E2 -> + Pid ! ?ERROR_MSG(Sock, Method, Reason), + reader_exit(State, E2) + end; +reader_loop(#{active := true, + async := true, + select_info := {select_info, _, Ref}, + select_num := N, + sock := Sock, + method := Method, + ctrl_proc := Pid} = State) -> + receive + {?MODULE, stop} -> + reader_exit(State, stop); + + {?MODULE, Pid, controlling_process, NewPid} -> + OldMRef = maps:get(ctrl_proc_mref, State), + erlang:demonitor(OldMRef, [flush]), + NewMRef = erlang:monitor(process, NewPid), + Pid ! {?MODULE, self(), controlling_process}, + reader_loop(State#{ctrl_proc => NewPid, + ctrl_proc_mref => NewMRef}); + + {?MODULE, active, NewActive} -> + reader_loop(State#{active => NewActive}); + + {'DOWN', MRef, process, Pid, Reason} -> + case maps:get(ctrl_proc_mref, State) of + MRef -> + reader_exit(State, {ctrl_exit, Reason}); + _ -> + reader_loop(State) + end; + + {'$socket', Sock, select, Ref} -> + case do_recv(Method, Sock, nowait) of + {ok, Data} when is_binary(Data) -> + Pid ! ?DATA_MSG(Sock, Method, Data), + reader_loop(State#{select_info => undefined, + select_num => N+1}); + + {error, closed} = E1 -> + Pid ! ?CLOSED_MSG(Sock, Method), + reader_exit(State, E1); + + {error, Reason} = E2 -> + Pid ! ?ERROR_MSG(Sock, Method, Reason), + reader_exit(State, E2) + end end. -do_recv(plain, Sock) -> - socket:recv(Sock, 0, ?READER_RECV_TIMEOUT); -do_recv(msg, Sock) -> - case socket:recvmsg(Sock, 0, 0, [], ?READER_RECV_TIMEOUT) of +do_recv(Method, Sock) -> + do_recv(Method, Sock, ?READER_RECV_TIMEOUT). + +do_recv(plain, Sock, Timeout) -> + socket:recv(Sock, 0, Timeout); +do_recv(msg, Sock, Timeout) -> + case socket:recvmsg(Sock, 0, 0, [], Timeout) of {ok, #{iov := [Bin]}} -> {ok, Bin}; + {select, _} = SELECT -> + SELECT; {error, _} = ERROR -> ERROR end. - - + + +reader_exit(#{async := false, active := Active}, stop) -> + vp("reader stopped when active: ~w", [Active]), + exit(normal); +reader_exit(#{async := true, + active := Active, + select_info := SelectInfo, + select_num := N}, stop) -> + vp("reader stopped when active: ~w" + "~n Current select info: ~p" + "~n Number of select messages: ~p", [Active, SelectInfo, N]), + exit(normal); +reader_exit(#{async := false, active := Active}, {ctrl_exit, normal}) -> + vp("reader ctrl exit when active: ~w", [Active]), + exit(normal); +reader_exit(#{async := true, + active := Active, + select_info := SelectInfo, + select_num := N}, {ctrl_exit, normal}) -> + vp("reader ctrl exit when active: ~w" + "~n Current select info: ~p" + "~n Number of select messages: ~p", [Active, SelectInfo, N]), + exit(normal); +reader_exit(#{async := false, active := Active}, {ctrl_exit, Reason}) -> + vp("reader exit when ctrl crash when active: ~w", [Active]), + exit({controlling_process, Reason}); +reader_exit(#{async := true, + active := Active, + select_info := SelectInfo, + select_num := N}, {ctrl_exit, Reason}) -> + vp("reader exit when ctrl crash when active: ~w" + "~n Current select info: ~p" + "~n Number of select messages: ~p", [Active, SelectInfo, N]), + exit({controlling_process, Reason}); +reader_exit(#{async := false, active := Active}, {error, closed}) -> + vp("reader exit when socket closed when active: ~w", [Active]), + exit(normal); +reader_exit(#{async := true, + active := Active, + select_info := SelectInfo, + select_num := N}, {error, closed}) -> + vp("reader exit when socket closed when active: ~w " + "~n Current select info: ~p" + "~n Number of select messages: ~p", [Active, SelectInfo, N]), + exit(normal); +reader_exit(#{async := false, active := Active}, {error, Reason}) -> + vp("reader exit when socket error when active: ~w", [Active]), + exit(Reason); +reader_exit(#{async := true, + active := Active, + select_info := SelectInfo, + select_num := N}, {error, Reason}) -> + vp("reader exit when socket error when active: ~w: " + "~n Current select info: ~p" + "~n Number of select messages: ~p", [Active, SelectInfo, N]), + exit(Reason). + + + + + %% ========================================================================== +vp(F, A) -> + vp(get(verbose), F, A). + +vp(true, F, A) -> + p(F, A); +vp(_, _, _) -> + ok. + +p(F, A) -> + io:format(F ++ "~n", A). + diff --git a/erts/etc/win32/msys_tools/vc/cc.sh b/erts/etc/win32/msys_tools/vc/cc.sh index 2b0482e876..71af14b9b2 100644 --- a/erts/etc/win32/msys_tools/vc/cc.sh +++ b/erts/etc/win32/msys_tools/vc/cc.sh @@ -195,7 +195,6 @@ mkdir $TMPOBJDIR # Compile for x in $SOURCES; do - start_time=`date '+%s'` # Compile each source if [ $LINKING = false ]; then # We should have an output defined, which is a directory @@ -260,16 +259,28 @@ for x in $SOURCES; do else tail -n +2 $ERR_FILE >&2 if test $DEPENDENCIES = true; then - if test `grep -v $x $MSG_FILE | grep -c '#line'` != "0"; then - o=`echo $x | sed 's,.*/,,' | sed 's,\.cp*$,.o,'` - echo -n $o':' -# cat $MSG_FILE | grep '#line' | grep -v $x | awk -F\" '{printf("%s\n",$2)}' | sort -u | grep -v " " | xargs -n 1 win2msys_path.sh | awk '{printf("\\\n %s ",$0)}' - cat $MSG_FILE | grep '#line' | grep -v $x | awk -F\" '{printf("%s\n",$2)}' | sort -u | grep -v " " | sed 's,^\([A-Za-z]\):[\\/]*,/\1/,;s,\\\\*,/,g'| awk '{printf("\\\n %s ",$0)}' - echo - echo - after_sed=`date '+%s'` - echo Made dependencies for $x':' `expr $after_sed '-' $start_time` 's' >&2 - fi + perl -e ' +my $file = "'$x'"; +while (<>) { + next unless /^#line/; + next if /$file/o; + (undef,$_) = split(/\"/); + next if / /; + $all{$_} = 1; +} +foreach (sort keys %all) { + s@^([A-Za-z]):@/$1@; + s@\\\\@/@g; + push @f, "\\\n $_ "; +} +if (@f) { + my $oname = $file; + $oname =~ s@.*/@@; + $oname =~ s@[.]cp*@.o@; + print $oname, ":", @f; + print "\n\n"; + print STDERR "Made dependencies for $file\n"; +}' $MSG_FILE else cat $MSG_FILE fi diff --git a/erts/preloaded/ebin/prim_inet.beam b/erts/preloaded/ebin/prim_inet.beam Binary files differindex b7c4cbe0a8..f67b660a08 100644 --- a/erts/preloaded/ebin/prim_inet.beam +++ b/erts/preloaded/ebin/prim_inet.beam diff --git a/erts/preloaded/ebin/socket.beam b/erts/preloaded/ebin/socket.beam Binary files differindex e37aa81b7c..134b4eac13 100644 --- a/erts/preloaded/ebin/socket.beam +++ b/erts/preloaded/ebin/socket.beam diff --git a/erts/preloaded/src/prim_inet.erl b/erts/preloaded/src/prim_inet.erl index d5abdd2483..374facb2a3 100644 --- a/erts/preloaded/src/prim_inet.erl +++ b/erts/preloaded/src/prim_inet.erl @@ -553,34 +553,49 @@ send(S, Data) -> %% "sendto" is for UDP. IP and Port are set by the caller to 0 if the socket %% is known to be connected. -sendto(S, Addr, _, Data) when is_port(S), tuple_size(Addr) =:= 2 -> - case type_value(set, addr, Addr) of - true -> - ?DBG_FORMAT("prim_inet:sendto(~p, ~p, ~p)~n", [S,Addr,Data]), - try - erlang:port_command(S, [enc_value(set, addr, Addr),Data]) - of - true -> - receive - {inet_reply,S,Reply} -> - ?DBG_FORMAT( - "prim_inet:sendto() -> ~p~n", [Reply]), - Reply - end - catch - error:_ -> - ?DBG_FORMAT( - "prim_inet:sendto() -> {error,einval}~n", []), - {error,einval} - end; - false -> - ?DBG_FORMAT( - "prim_inet:sendto() -> {error,einval}~n", []), - {error,einval} - end; -sendto(S, IP, Port, Data) -> - sendto(S, {IP, Port}, 0, Data). - +sendto(S, {_, _} = Address, AncOpts, Data) + when is_port(S), is_list(AncOpts) -> + case encode_opt_val(AncOpts) of + {ok, AncData} -> + AncDataLen = iolist_size(AncData), + case + type_value(set, addr, Address) andalso + type_value(set, uint32, AncDataLen) + of + true -> + ?DBG_FORMAT("prim_inet:sendto(~p, ~p, ~p, ~p)~n", + [S,Address,AncOpts,Data]), + PortCommandData = + [enc_value(set, addr, Address), + enc_value(set, uint32, AncDataLen), AncData, + Data], + try erlang:port_command(S, PortCommandData) of + true -> + receive + {inet_reply,S,Reply} -> + ?DBG_FORMAT( + "prim_inet:sendto() -> ~p~n", [Reply]), + Reply + end + catch + _:_ -> + ?DBG_FORMAT( + "prim_inet:sendto() -> {error,einval}~n", []), + {error,einval} + end; + false -> + ?DBG_FORMAT( + "prim_inet:sendto() -> {error,einval}~n", []), + {error,einval} + end; + {error,_} -> + ?DBG_FORMAT( + "prim_inet:sendto() -> {error,einval}~n", []), + {error,einval} + end; +sendto(S, IP, Port, Data) + when is_port(S), is_integer(Port) -> + sendto(S, {IP, Port}, [], Data). %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% @@ -1993,15 +2008,15 @@ enc_value_2(addr, {File,_}) when is_list(File); is_binary(File) -> [?INET_AF_LOCAL,iolist_size(File)|File]; %% enc_value_2(addr, {inet,{any,Port}}) -> - [?INET_AF_INET,?int16(Port),0,0,0,0]; + [?INET_AF_INET,?int16(Port)|ip4_to_bytes({0,0,0,0})]; enc_value_2(addr, {inet,{loopback,Port}}) -> - [?INET_AF_INET,?int16(Port),127,0,0,1]; + [?INET_AF_INET,?int16(Port)|ip4_to_bytes({127,0,0,1})]; enc_value_2(addr, {inet,{IP,Port}}) -> [?INET_AF_INET,?int16(Port)|ip4_to_bytes(IP)]; enc_value_2(addr, {inet6,{any,Port}}) -> - [?INET_AF_INET6,?int16(Port),0,0,0,0,0,0,0,0]; + [?INET_AF_INET6,?int16(Port)|ip6_to_bytes({0,0,0,0,0,0,0,0})]; enc_value_2(addr, {inet6,{loopback,Port}}) -> - [?INET_AF_INET6,?int16(Port),0,0,0,0,0,0,0,1]; + [?INET_AF_INET6,?int16(Port)|ip6_to_bytes({0,0,0,0,0,0,0,1})]; enc_value_2(addr, {inet6,{IP,Port}}) -> [?INET_AF_INET6,?int16(Port)|ip6_to_bytes(IP)]; enc_value_2(addr, {local,Addr}) -> @@ -2149,10 +2164,10 @@ enum_name(_, []) -> false. %% encode opt/val REVERSED since options are stored in reverse order %% i.e. the recent options first (we must process old -> new) encode_opt_val(Opts) -> - try - enc_opt_val(Opts, []) + try + {ok, enc_opt_val(Opts, [])} catch - Reason -> {error,Reason} + throw:Reason -> {error,Reason} end. %% {active, once} and {active, N} are specially optimized because they will @@ -2171,17 +2186,21 @@ enc_opt_val([binary|Opts], Acc) -> enc_opt_val(Opts, Acc, mode, binary); enc_opt_val([list|Opts], Acc) -> enc_opt_val(Opts, Acc, mode, list); -enc_opt_val([_|_], _) -> {error,einval}; -enc_opt_val([], Acc) -> {ok,Acc}. +enc_opt_val([_|_], _) -> + throw(einval); +enc_opt_val([], Acc) -> + Acc. enc_opt_val(Opts, Acc, Opt, Val) when is_atom(Opt) -> Type = type_opt(set, Opt), case type_value(set, Type, Val) of true -> enc_opt_val(Opts, [enc_opt(Opt),enc_value(set, Type, Val)|Acc]); - false -> {error,einval} + false -> + throw(einval) end; -enc_opt_val(_, _, _, _) -> {error,einval}. +enc_opt_val(_, _, _, _) -> + throw(einval). diff --git a/erts/preloaded/src/socket.erl b/erts/preloaded/src/socket.erl index 0f0d8f7a02..ae1ffdb4ac 100644 --- a/erts/preloaded/src/socket.erl +++ b/erts/preloaded/src/socket.erl @@ -53,10 +53,16 @@ getopt/3, sockname/1, - peername/1 + peername/1, + + cancel/2 ]). -export_type([ + select_tag/0, + select_ref/0, + select_info/0, + domain/0, type/0, protocol/0, @@ -585,6 +591,17 @@ #{level := integer(), type := integer(), data := binary()}. +-opaque select_tag() :: atom(). +-opaque select_ref() :: reference(). + +-record(select_info, {tag :: select_tag(), ref :: select_ref()}). + +-type select_info() :: #select_info{}. + +-define(SELECT_INFO(T, R), #select_info{tag = T, ref = R}). +-define(SELECT(T, R), {select, ?SELECT_INFO(T, R)}). + + %% This is used in messages sent from the nif-code to erlang processes: %% %% {?SOCKET_TAG, Socket :: socket(), Tag :: atom(), Info :: term()} @@ -1189,18 +1206,24 @@ validate_inet6_addrs(Addrs) -> %% -spec connect(Socket, SockAddr) -> ok | {error, Reason} when - Socket :: socket(), - SockAddr :: sockaddr(), - Reason :: term(). + Socket :: socket(), + SockAddr :: sockaddr(), + Reason :: term(). connect(Socket, SockAddr) -> connect(Socket, SockAddr, infinity). --spec connect(Socket, SockAddr, Timeout) -> ok | {error, Reason} when - Socket :: socket(), - SockAddr :: sockaddr(), - Timeout :: timeout(), - Reason :: term(). +-spec connect(Socket, SockAddr, nowait) -> + ok | {select, SelectInfo} | {error, Reason} when + Socket :: socket(), + SockAddr :: sockaddr(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, SockAddr, Timeout) -> ok | {error, Reason} when + Socket :: socket(), + SockAddr :: sockaddr(), + Timeout :: timeout(), + Reason :: term(). %% <KOLLA> %% Is it possible to connect with family = local for the (dest) sockaddr? @@ -1210,12 +1233,18 @@ connect(_Socket, _SockAddr, Timeout) {error, timeout}; connect(#socket{ref = SockRef}, #{family := Fam} = SockAddr, Timeout) when ((Fam =:= inet) orelse (Fam =:= inet6) orelse (Fam =:= local)) andalso - ((Timeout =:= infinity) orelse is_integer(Timeout)) -> + ((Timeout =:= nowait) orelse + (Timeout =:= infinity) orelse is_integer(Timeout)) -> TS = timestamp(Timeout), case nif_connect(SockRef, SockAddr) of ok -> %% Connected! ok; + + {ok, Ref} when (Timeout =:= nowait) -> + %% Connecting, but the caller does not want to wait... + ?SELECT(connect, Ref); + {ok, Ref} -> %% Connecting... NewTimeout = next_timeout(TS, Timeout), @@ -1268,17 +1297,27 @@ listen(#socket{ref = SockRef}, Backlog) accept(Socket) -> accept(Socket, ?SOCKET_ACCEPT_TIMEOUT_DEFAULT). --spec accept(LSocket, Timeout) -> {ok, Socket} | {error, Reason} when - LSocket :: socket(), - Timeout :: timeout(), - Socket :: socket(), - Reason :: term(). +-spec accept(LSocket, nowait) -> + {ok, Socket} | + {select, SelectInfo} | + {error, Reason} when + LSocket :: socket(), + Socket :: socket(), + SelectInfo :: select_info(), + Reason :: term() + ; (LSocket, Timeout) -> {ok, Socket} | {error, Reason} when + LSocket :: socket(), + Timeout :: timeout(), + Socket :: socket(), + Reason :: term(). %% Do we really need this optimization? accept(_, Timeout) when is_integer(Timeout) andalso (Timeout =< 0) -> {error, timeout}; accept(#socket{ref = LSockRef}, Timeout) - when is_integer(Timeout) orelse (Timeout =:= infinity) -> + when is_integer(Timeout) orelse + (Timeout =:= infinity) orelse + (Timeout =:= nowait) -> do_accept(LSockRef, Timeout). do_accept(LSockRef, Timeout) -> @@ -1289,6 +1328,11 @@ do_accept(LSockRef, Timeout) -> Socket = #socket{ref = SockRef}, {ok, Socket}; + + {error, eagain} when (Timeout =:= nowait) -> + ?SELECT(accept, AccRef); + + {error, eagain} -> %% Each call is non-blocking, but even then it takes %% *some* time, so just to be sure, recalculate before @@ -1327,33 +1371,56 @@ send(Socket, Data) -> send(Socket, Data, ?SOCKET_SEND_FLAGS_DEFAULT, ?SOCKET_SEND_TIMEOUT_DEFAULT). -spec send(Socket, Data, Flags) -> ok | {error, Reason} when - Socket :: socket(), - Data :: iodata(), - Flags :: send_flags(), - Reason :: term() + Socket :: socket(), + Data :: iodata(), + Flags :: send_flags(), + Reason :: term() + ; (Socket, Data, Timeout :: nowait) -> ok | + {select, SelectInfo} | + {ok, {RestData, SelectInfo}} | + {error, Reason} when + Socket :: socket(), + Data :: iodata(), + RestData :: binary(), + SelectInfo :: select_info(), + Reason :: term() ; (Socket, Data, Timeout) -> ok | {error, Reason} when - Socket :: socket(), - Data :: iodata(), - Timeout :: timeout(), - Reason :: term(). + Socket :: socket(), + Data :: iodata(), + Timeout :: timeout(), + Reason :: term(). send(Socket, Data, Flags) when is_list(Flags) -> send(Socket, Data, Flags, ?SOCKET_SEND_TIMEOUT_DEFAULT); send(Socket, Data, Timeout) -> send(Socket, Data, ?SOCKET_SEND_FLAGS_DEFAULT, Timeout). --spec send(Socket, Data, Flags, Timeout) -> ok | {error, Reason} when - Socket :: socket(), - Data :: iodata(), - Flags :: send_flags(), - Timeout :: timeout(), - Reason :: term(). +-spec send(Socket, Data, Flags, nowait) -> ok | + {select, SelectInfo} | + {ok, {RestData, SelectInfo}} | + {error, Reason} when + Socket :: socket(), + Data :: iodata(), + Flags :: send_flags(), + RestData :: binary(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, Data, Flags, Timeout) -> ok | {error, Reason} when + Socket :: socket(), + Data :: iodata(), + Flags :: send_flags(), + Timeout :: timeout(), + Reason :: term(). send(Socket, Data, Flags, Timeout) when is_list(Data) -> Bin = erlang:list_to_binary(Data), send(Socket, Bin, Flags, Timeout); send(#socket{ref = SockRef}, Data, Flags, Timeout) - when is_binary(Data) andalso is_list(Flags) -> + when is_binary(Data) andalso + is_list(Flags) andalso + ((Timeout =:= nowait) orelse + (Timeout =:= infinity) orelse + (is_integer(Timeout) andalso (Timeout > 0))) -> EFlags = enc_send_flags(Flags), do_send(SockRef, Data, EFlags, Timeout). @@ -1363,6 +1430,15 @@ do_send(SockRef, Data, EFlags, Timeout) -> case nif_send(SockRef, SendRef, Data, EFlags) of ok -> ok; + + + {ok, Written} when (Timeout =:= nowait) -> + <<_:Written/binary, Rest/binary>> = Data, + %% We are partially done, but the user don't want to wait (here) + %% for completion + {ok, {Rest, ?SELECT_INFO(send, SendRef)}}; + + {ok, Written} -> NewTimeout = next_timeout(TS, Timeout), %% We are partially done, wait for continuation @@ -1384,6 +1460,12 @@ do_send(SockRef, Data, EFlags, Timeout) -> cancel(SockRef, send, SendRef), {error, {timeout, size(Data)}} end; + + + {error, eagain} when (Timeout =:= nowait) -> + ?SELECT(send, SendRef); + + {error, eagain} -> receive {?SOCKET_TAG, #socket{ref = SockRef}, select, SendRef} -> @@ -1419,17 +1501,25 @@ sendto(Socket, Data, Dest) -> sendto(Socket, Data, Dest, ?SOCKET_SENDTO_FLAGS_DEFAULT). -spec sendto(Socket, Data, Dest, Flags) -> ok | {error, Reason} when - Socket :: socket(), - Data :: binary(), - Dest :: null | sockaddr(), - Flags :: send_flags(), - Reason :: term() + Socket :: socket(), + Data :: binary(), + Dest :: null | sockaddr(), + Flags :: send_flags(), + Reason :: term() + ; (Socket, Data, Dest, Timeout :: nowait) -> ok | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + Data :: iodata(), + Dest :: null | sockaddr(), + SelectInfo :: select_info(), + Reason :: term() ; (Socket, Data, Dest, Timeout) -> ok | {error, Reason} when - Socket :: socket(), - Data :: iodata(), - Dest :: null | sockaddr(), - Timeout :: timeout(), - Reason :: term(). + Socket :: socket(), + Data :: iodata(), + Dest :: null | sockaddr(), + Timeout :: timeout(), + Reason :: term(). sendto(Socket, Data, Dest, Flags) when is_list(Flags) -> sendto(Socket, Data, Dest, Flags, ?SOCKET_SENDTO_TIMEOUT_DEFAULT); @@ -1437,13 +1527,22 @@ sendto(Socket, Data, Dest, Timeout) -> sendto(Socket, Data, Dest, ?SOCKET_SENDTO_FLAGS_DEFAULT, Timeout). --spec sendto(Socket, Data, Dest, Flags, Timeout) -> ok | {error, Reason} when - Socket :: socket(), - Data :: binary(), - Dest :: null | sockaddr(), - Flags :: send_flags(), - Timeout :: timeout(), - Reason :: term(). +-spec sendto(Socket, Data, Dest, Flags, nowait) -> ok | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + Data :: binary(), + Dest :: null | sockaddr(), + Flags :: send_flags(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, Data, Dest, Flags, Timeout) -> ok | {error, Reason} when + Socket :: socket(), + Data :: binary(), + Dest :: null | sockaddr(), + Flags :: send_flags(), + Timeout :: timeout(), + Reason :: term(). sendto(Socket, Data, Dest, Flags, Timeout) when is_list(Data) -> Bin = erlang:list_to_binary(Data), @@ -1452,14 +1551,18 @@ sendto(#socket{ref = SockRef}, Data, Dest, Flags, Timeout) when is_binary(Data) andalso (Dest =:= null) andalso is_list(Flags) andalso - (is_integer(Timeout) orelse (Timeout =:= infinity)) -> + ((Timeout =:= nowait) orelse + (Timeout =:= infinity) orelse + (is_integer(Timeout) andalso (Timeout > 0))) -> EFlags = enc_send_flags(Flags), do_sendto(SockRef, Data, Dest, EFlags, Timeout); sendto(#socket{ref = SockRef}, Data, #{family := Fam} = Dest, Flags, Timeout) when is_binary(Data) andalso ((Fam =:= inet) orelse (Fam =:= inet6) orelse (Fam =:= local)) andalso is_list(Flags) andalso - (is_integer(Timeout) orelse (Timeout =:= infinity)) -> + ((Timeout =:= nowait) orelse + (Timeout =:= infinity) orelse + (is_integer(Timeout) andalso (Timeout > 0))) -> EFlags = enc_send_flags(Flags), do_sendto(SockRef, Data, Dest, EFlags, Timeout). @@ -1471,6 +1574,11 @@ do_sendto(SockRef, Data, Dest, EFlags, Timeout) -> %% We are done ok; + {ok, Written} when (Timeout =:= nowait) -> + <<_:Written/binary, Rest/binary>> = Data, + {ok, {Rest, ?SELECT_INFO(sendto, SendRef)}}; + + {ok, Written} -> %% We are partially done, wait for continuation receive @@ -1492,6 +1600,11 @@ do_sendto(SockRef, Data, Dest, EFlags, Timeout) -> {error, timeout} end; + + {error, eagain} when (Timeout =:= nowait) -> + ?SELECT(sendto, SendRef); + + {error, eagain} -> receive {?SOCKET_TAG, #socket{ref = SockRef}, select, SendRef} -> @@ -1535,11 +1648,18 @@ sendmsg(Socket, MsgHdr) -> MsgHdr :: msghdr(), Flags :: send_flags(), Reason :: term() + ; (Socket, MsgHdr, Timeout :: nowait) -> ok | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + MsgHdr :: msghdr(), + SelectInfo :: select_info(), + Reason :: term() ; (Socket, MsgHdr, Timeout) -> ok | {error, Reason} when - Socket :: socket(), - MsgHdr :: msghdr(), - Timeout :: timeout(), - Reason :: term(). + Socket :: socket(), + MsgHdr :: msghdr(), + Timeout :: timeout(), + Reason :: term(). sendmsg(Socket, MsgHdr, Flags) when is_list(Flags) -> sendmsg(Socket, MsgHdr, Flags, ?SOCKET_SENDMSG_TIMEOUT_DEFAULT); @@ -1548,19 +1668,34 @@ sendmsg(Socket, MsgHdr, Timeout) sendmsg(Socket, MsgHdr, ?SOCKET_SENDMSG_FLAGS_DEFAULT, Timeout). --spec sendmsg(Socket, MsgHdr, Flags, Timeout) -> - ok | {ok, Remaining} | {error, Reason} when - Socket :: socket(), - MsgHdr :: msghdr(), - Flags :: send_flags(), - Timeout :: timeout(), - Remaining :: erlang:iovec(), - Reason :: term(). +-spec sendmsg(Socket, MsgHdr, Flags, nowait) -> + ok | + {ok, Remaining} | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + MsgHdr :: msghdr(), + Flags :: send_flags(), + Remaining :: erlang:iovec(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, MsgHdr, Flags, Timeout) -> + ok | + {ok, Remaining} | + {error, Reason} when + Socket :: socket(), + MsgHdr :: msghdr(), + Flags :: send_flags(), + Timeout :: timeout(), + Remaining :: erlang:iovec(), + Reason :: term(). sendmsg(#socket{ref = SockRef}, #{iov := IOV} = MsgHdr, Flags, Timeout) when is_list(IOV) andalso is_list(Flags) andalso - (is_integer(Timeout) orelse (Timeout =:= infinity)) -> + ((Timeout =:= nowait) orelse + (Timeout =:= infinity) orelse + (is_integer(Timeout) andalso (Timeout > 0))) -> try ensure_msghdr(MsgHdr) of M -> EFlags = enc_send_flags(Flags), @@ -1580,6 +1715,7 @@ do_sendmsg(SockRef, MsgHdr, EFlags, Timeout) -> %% We are done ok; + {ok, Written} when is_integer(Written) andalso (Written > 0) -> %% We should not retry here since the protocol may not %% be able to handle a message being split. Leave it to @@ -1591,6 +1727,11 @@ do_sendmsg(SockRef, MsgHdr, EFlags, Timeout) -> cancel(SockRef, sendmsg, SendRef), {ok, do_sendmsg_rest(maps:get(iov, MsgHdr), Written)}; + + {error, eagain} when (Timeout =:= nowait) -> + ?SELECT(sendmsg, SendRef); + + {error, eagain} -> receive {?SOCKET_TAG, #socket{ref = SockRef}, select, SendRef} -> @@ -1664,13 +1805,24 @@ recv(Socket, Length) -> ?SOCKET_RECV_FLAGS_DEFAULT, ?SOCKET_RECV_TIMEOUT_DEFAULT). --spec recv(Socket, Length, Flags) -> {ok, Data} | {error, Reason} when +-spec recv(Socket, Length, Flags) -> {ok, Data} | + {error, Reason} when Socket :: socket(), Length :: non_neg_integer(), Flags :: recv_flags(), Data :: binary(), Reason :: term() - ; (Socket, Length, Timeout) -> {ok, Data} | {error, Reason} when + ; (Socket, Length, Timeout :: nowait) -> {ok, Data} | + {select, SelectInfo} | + {ok, {Data, SelectInfo}} | + {error, Reason} when + Socket :: socket(), + Length :: non_neg_integer(), + Data :: binary(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, Length, Timeout) -> {ok, Data} | + {error, Reason} when Socket :: socket(), Length :: non_neg_integer(), Timeout :: timeout(), @@ -1682,18 +1834,31 @@ recv(Socket, Length, Flags) when is_list(Flags) -> recv(Socket, Length, Timeout) -> recv(Socket, Length, ?SOCKET_RECV_FLAGS_DEFAULT, Timeout). --spec recv(Socket, Length, Flags, Timeout) -> {ok, Data} | {error, Reason} when - Socket :: socket(), - Length :: non_neg_integer(), - Flags :: recv_flags(), - Timeout :: timeout(), - Data :: binary(), - Reason :: term(). +-spec recv(Socket, Length, Flags, nowait) -> {ok, Data} | + {select, SelectInfo} | + {ok, {Data, SelectInfo}} | + {error, Reason} when + Socket :: socket(), + Length :: non_neg_integer(), + Flags :: recv_flags(), + Data :: binary(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, Length, Flags, Timeout) -> {ok, Data} | + {error, Reason} when + Socket :: socket(), + Length :: non_neg_integer(), + Flags :: recv_flags(), + Timeout :: timeout(), + Data :: binary(), + Reason :: term(). recv(#socket{ref = SockRef}, Length, Flags, Timeout) when (is_integer(Length) andalso (Length >= 0)) andalso is_list(Flags) andalso - (is_integer(Timeout) orelse (Timeout =:= infinity)) -> + (is_integer(Timeout) orelse + (Timeout =:= infinity) orelse + (Timeout =:= nowait)) -> EFlags = enc_recv_flags(Flags), do_recv(SockRef, undefined, Length, EFlags, <<>>, Timeout). @@ -1701,8 +1866,12 @@ recv(#socket{ref = SockRef}, Length, Flags, Timeout) %% with Length = 0. This case makes it neccessary to have a timeout function %% clause since we may never wait for anything (no receive select), and so the %% the only timeout check will be the function clause. +%% Note that the Timeout value of 'nowait' has a special meaning. It means +%% that we will either return with data or with the with {error, NNNN}. In +%% wich case the caller will receive a select message at some later time. do_recv(SockRef, _OldRef, Length, EFlags, Acc, Timeout) - when (Timeout =:= infinity) orelse + when (Timeout =:= nowait) orelse + (Timeout =:= infinity) orelse (is_integer(Timeout) andalso (Timeout > 0)) -> TS = timestamp(Timeout), RecvRef = make_ref(), @@ -1723,6 +1892,15 @@ do_recv(SockRef, _OldRef, Length, EFlags, Acc, Timeout) <<Acc/binary, Bin/binary>>, next_timeout(TS, Timeout)); + + %% Did not get all the user asked for, but the user also + %% specified 'nowait', so deliver what we got and the + %% select info. + {ok, false = _Completed, Bin} when (Timeout =:= nowait) andalso + (size(Acc) =:= 0) -> + {ok, {Bin, ?SELECT_INFO(recv, RecvRef)}}; + + {ok, false = _Completed, Bin} when (size(Acc) =:= 0) -> %% We got the first chunk of it. %% We will be notified (select message) when there @@ -1761,6 +1939,17 @@ do_recv(SockRef, _OldRef, Length, EFlags, Acc, Timeout) {error, {timeout, Acc}} end; + + %% The user does not want to wait! + %% The user will be informed that there is something to read + %% via the select socket message (see below). + + {error, eagain} when (Timeout =:= nowait) andalso (size(Acc) =:= 0) -> + ?SELECT(recv, RecvRef); + {error, eagain} when (Timeout =:= nowait) -> + {ok, {Acc, ?SELECT_INFO(recv, RecvRef)}}; + + %% We return with the accumulated binary (if its non-empty) {error, eagain} when (Length =:= 0) andalso (size(Acc) > 0) -> %% CAN WE REALLY DO THIS? THE NIF HAS SELECTED!! OR? @@ -1851,30 +2040,51 @@ recvfrom(Socket, BufSz) -> ?SOCKET_RECV_FLAGS_DEFAULT, ?SOCKET_RECV_TIMEOUT_DEFAULT). --spec recvfrom(Socket, Flags, Timeout) -> - {ok, {Source, Data}} | {error, Reason} when - Socket :: socket(), - Flags :: recv_flags(), - Timeout :: timeout(), - Source :: sockaddr() | undefined, - Data :: binary(), - Reason :: term() +-spec recvfrom(Socket, Flags, nowait) -> + {ok, {Source, Data}} | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + Flags :: recv_flags(), + Source :: sockaddr() | undefined, + Data :: binary(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, Flags, Timeout) -> + {ok, {Source, Data}} | + {error, Reason} when + Socket :: socket(), + Flags :: recv_flags(), + Timeout :: timeout(), + Source :: sockaddr() | undefined, + Data :: binary(), + Reason :: term() ; (Socket, BufSz, Flags) -> {ok, {Source, Data}} | {error, Reason} when - Socket :: socket(), - BufSz :: non_neg_integer(), - Flags :: recv_flags(), - Source :: sockaddr() | undefined, - Data :: binary(), - Reason :: term() + Socket :: socket(), + BufSz :: non_neg_integer(), + Flags :: recv_flags(), + Source :: sockaddr() | undefined, + Data :: binary(), + Reason :: term() + ; (Socket, BufSz, nowait) -> + {ok, {Source, Data}} | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + BufSz :: non_neg_integer(), + Source :: sockaddr() | undefined, + Data :: binary(), + SelectInfo :: select_info(), + Reason :: term() ; (Socket, BufSz, Timeout) -> {ok, {Source, Data}} | {error, Reason} when - Socket :: socket(), - BufSz :: non_neg_integer(), - Timeout :: timeout(), - Source :: sockaddr() | undefined, - Data :: binary(), - Reason :: term(). + Socket :: socket(), + BufSz :: non_neg_integer(), + Timeout :: timeout(), + Source :: sockaddr() | undefined, + Data :: binary(), + Reason :: term(). recvfrom(Socket, Flags, Timeout) when is_list(Flags) -> recvfrom(Socket, 0, Flags, Timeout); @@ -1883,20 +2093,34 @@ recvfrom(Socket, BufSz, Flags) when is_list(Flags) -> recvfrom(Socket, BufSz, Timeout) -> recvfrom(Socket, BufSz, ?SOCKET_RECV_FLAGS_DEFAULT, Timeout). --spec recvfrom(Socket, BufSz, Flags, Timeout) -> - {ok, {Source, Data}} | {error, Reason} when - Socket :: socket(), - BufSz :: non_neg_integer(), - Flags :: recv_flags(), - Timeout :: timeout(), - Source :: sockaddr() | undefined, - Data :: binary(), - Reason :: term(). +-spec recvfrom(Socket, BufSz, Flags, nowait) -> + {ok, {Source, Data}} | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + BufSz :: non_neg_integer(), + Flags :: recv_flags(), + Source :: sockaddr() | undefined, + Data :: binary(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, BufSz, Flags, Timeout) -> + {ok, {Source, Data}} | + {error, Reason} when + Socket :: socket(), + BufSz :: non_neg_integer(), + Flags :: recv_flags(), + Timeout :: timeout(), + Source :: sockaddr() | undefined, + Data :: binary(), + Reason :: term(). recvfrom(#socket{ref = SockRef}, BufSz, Flags, Timeout) when (is_integer(BufSz) andalso (BufSz >= 0)) andalso is_list(Flags) andalso - (is_integer(Timeout) orelse (Timeout =:= infinity)) -> + (is_integer(Timeout) orelse + (Timeout =:= infinity) orelse + (Timeout =:= nowait)) -> EFlags = enc_recv_flags(Flags), do_recvfrom(SockRef, BufSz, EFlags, Timeout). @@ -1907,6 +2131,11 @@ do_recvfrom(SockRef, BufSz, EFlags, Timeout) -> {ok, {_Source, _NewData}} = OK -> OK; + + {error, eagain} when (Timeout =:= nowait) -> + ?SELECT(recvfrom, RecvRef); + + {error, eagain} -> %% There is nothing just now, but we will be notified when there %% is something to read (a select message). @@ -1947,29 +2176,44 @@ recvmsg(Socket) -> Flags :: recv_flags(), MsgHdr :: msghdr(), Reason :: term() + ; (Socket, Timeout :: nowait) -> {ok, MsgHdr} | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + MsgHdr :: msghdr(), + SelectInfo :: select_info(), + Reason :: term() ; (Socket, Timeout) -> {ok, MsgHdr} | {error, Reason} when - Socket :: socket(), - Timeout :: timeout(), - MsgHdr :: msghdr(), - Reason :: term(). + Socket :: socket(), + Timeout :: timeout(), + MsgHdr :: msghdr(), + Reason :: term(). recvmsg(Socket, Flags) when is_list(Flags) -> recvmsg(Socket, 0, 0, Flags, ?SOCKET_RECV_TIMEOUT_DEFAULT); recvmsg(Socket, Timeout) -> recvmsg(Socket, 0, 0, ?SOCKET_RECV_FLAGS_DEFAULT, Timeout). --spec recvmsg(Socket, Flags, Timeout) -> {ok, MsgHdr} | {error, Reason} when - Socket :: socket(), - Flags :: recv_flags(), - Timeout :: timeout(), - MsgHdr :: msghdr(), - Reason :: term() +-spec recvmsg(Socket, Flags, nowait) -> {ok, MsgHdr} | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + Flags :: recv_flags(), + MsgHdr :: msghdr(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, Flags, Timeout) -> {ok, MsgHdr} | {error, Reason} when + Socket :: socket(), + Flags :: recv_flags(), + Timeout :: timeout(), + MsgHdr :: msghdr(), + Reason :: term() ; (Socket, BufSz, CtrlSz) -> {ok, MsgHdr} | {error, Reason} when - Socket :: socket(), - BufSz :: non_neg_integer(), - CtrlSz :: non_neg_integer(), - MsgHdr :: msghdr(), - Reason :: term(). + Socket :: socket(), + BufSz :: non_neg_integer(), + CtrlSz :: non_neg_integer(), + MsgHdr :: msghdr(), + Reason :: term(). recvmsg(Socket, Flags, Timeout) when is_list(Flags) -> recvmsg(Socket, 0, 0, Flags, Timeout); @@ -1980,20 +2224,34 @@ recvmsg(Socket, BufSz, CtrlSz) when is_integer(BufSz) andalso is_integer(CtrlSz) -spec recvmsg(Socket, BufSz, CtrlSz, - Flags, Timeout) -> {ok, MsgHdr} | {error, Reason} when - Socket :: socket(), - BufSz :: non_neg_integer(), - CtrlSz :: non_neg_integer(), - Flags :: recv_flags(), - Timeout :: timeout(), - MsgHdr :: msghdr(), - Reason :: term(). + Flags, nowait) -> {ok, MsgHdr} | + {select, SelectInfo} | + {error, Reason} when + Socket :: socket(), + BufSz :: non_neg_integer(), + CtrlSz :: non_neg_integer(), + Flags :: recv_flags(), + MsgHdr :: msghdr(), + SelectInfo :: select_info(), + Reason :: term() + ; (Socket, + BufSz, CtrlSz, + Flags, Timeout) -> {ok, MsgHdr} | {error, Reason} when + Socket :: socket(), + BufSz :: non_neg_integer(), + CtrlSz :: non_neg_integer(), + Flags :: recv_flags(), + Timeout :: timeout(), + MsgHdr :: msghdr(), + Reason :: term(). recvmsg(#socket{ref = SockRef}, BufSz, CtrlSz, Flags, Timeout) when (is_integer(BufSz) andalso (BufSz >= 0)) andalso (is_integer(CtrlSz) andalso (CtrlSz >= 0)) andalso is_list(Flags) andalso - (is_integer(Timeout) orelse (Timeout =:= infinity)) -> + (is_integer(Timeout) orelse + (Timeout =:= infinity) orelse + (Timeout =:= nowait)) -> EFlags = enc_recv_flags(Flags), do_recvmsg(SockRef, BufSz, CtrlSz, EFlags, Timeout). @@ -2004,6 +2262,11 @@ do_recvmsg(SockRef, BufSz, CtrlSz, EFlags, Timeout) -> {ok, _MsgHdr} = OK -> OK; + + {error, eagain} when (Timeout =:= nowait) -> + ?SELECT(recvmsg, RecvRef); + + {error, eagain} -> %% There is nothing just now, but we will be notified when there %% is something to read (a select message). @@ -2339,6 +2602,25 @@ peername(#socket{ref = SockRef}) -> nif_peername(SockRef). +%% =========================================================================== +%% +%% cancel - cancel an operation resulting in a select +%% +%% A call to accept, recv/recvfrom/recvmsg and send/sendto/sendmsg +%% can result in a select if they are called with the Timeout argument +%% set to nowait. This is indicated by the return of the select-info. +%% Such a operation can be cancelled by calling this function. +%% + +-spec cancel(Socket, SelectInfo) -> ok | {error, Reason} when + Socket :: socket(), + SelectInfo :: select_info(), + Reason :: term(). + +cancel(#socket{ref = SockRef}, #select_info{tag = Tag, ref = Ref}) -> + cancel(SockRef, Tag, Ref). + + %% =========================================================================== %% @@ -3462,15 +3744,18 @@ flush_select_msgs(SockRef, Ref) -> %% A timestamp in ms +timestamp(nowait = T) -> + T; timestamp(infinity) -> undefined; timestamp(_) -> timestamp(). timestamp() -> - {A,B,C} = os:timestamp(), - A*1000000000+B*1000+(C div 1000). + erlang:monotonic_time(milli_seconds). +next_timeout(_, nowait = Timeout) -> + Timeout; next_timeout(_, infinity = Timeout) -> Timeout; next_timeout(TS, Timeout) -> diff --git a/erts/vsn.mk b/erts/vsn.mk index 224570fb09..40a9685e9d 100644 --- a/erts/vsn.mk +++ b/erts/vsn.mk @@ -18,7 +18,7 @@ # %CopyrightEnd% # -VSN = 10.4 +VSN = 10.4.3 # Port number 4365 in 4.2 # Port number 4366 in 4.3 |