Merge branch 'raimo/blocking-sctp-send/OTP-19061' into maint-26

* raimo/blocking-sctp-send/OTP-19061:
  Make gen_sctp:send blocking instead of returning ewouldblock

# Conflicts:
#	erts/preloaded/ebin/prim_inet.beam

Author: Erlang/OTP

erts/emulator/drivers/common/inet_drv.c

@@ -1338,6 +1338,7 @@ static int        packet_inet_init(void);
 static void       packet_inet_stop(ErlDrvData);
 static void       packet_inet_command(ErlDrvData, char*, ErlDrvSizeT);
 static void       packet_inet_drv_input(ErlDrvData data, ErlDrvEvent event);
+static void       packet_inet_drv_output(ErlDrvData data, ErlDrvEvent event);
 static ErlDrvData udp_inet_start(ErlDrvPort, char* command);
 #ifdef HAVE_SCTP
 static ErlDrvData sctp_inet_start(ErlDrvPort, char* command);
@@ -1359,10 +1360,10 @@ static struct erl_drv_entry udp_inet_driver_entry =
     packet_inet_command,
 #ifdef __WIN32__
     packet_inet_event,
-    NULL, 
+    NULL,
 #else
     packet_inet_drv_input,
-    NULL,
+    packet_inet_drv_output,
 #endif
     "udp_inet",
     NULL,
@@ -1394,10 +1395,10 @@ static struct erl_drv_entry sctp_inet_driver_entry =
     packet_inet_command,
 #ifdef __WIN32__
     packet_inet_event,
-    NULL, 
+    NULL,
 #else
     packet_inet_drv_input,
-    NULL,
+    packet_inet_drv_output,
 #endif
     "sctp_inet",
     NULL,
@@ -2590,9 +2591,35 @@ inet_reply_finish(inet_descriptor *desc, ErlDrvTermData *spec, int i) {
 }
 
 /* send:
-**   {inet_reply, S, ok[, CallerTag]}
+**   {inet_reply, S, CallerTag}
 */
-static int inet_reply_ok(inet_descriptor* desc)
+static int inet_reply_caller_ref(inet_descriptor* desc)
+{
+    ErlDrvTermData
+        spec[2*LOAD_ATOM_CNT + LOAD_PORT_CNT + LOAD_TUPLE_CNT +
+             LOAD_EXT_CNT];
+    CallerRef *cref_p = &desc->caller_ref;
+    int i = 0;
+
+    if (is_not_internal_pid(desc->caller)) return 0; /* XXX what value for error? */
+    i = LOAD_ATOM(spec, i, am_inet_reply);
+    i = LOAD_PORT(spec, i, desc->dport);
+    if (cref_p->tag_buf != NULL) {
+        i = LOAD_EXT(spec, i, cref_p->tag_buf, cref_p->tag_len);
+    }
+    else {
+        i = LOAD_ATOM(spec, i, am_undefined);
+    }
+    i = LOAD_TUPLE(spec, i, 3);
+
+    ASSERT(sizeof(spec)/sizeof(*spec) >= i);
+    return erl_drv_send_term(desc->dport, desc->caller, spec, i);
+}
+
+/* send:
+**   {inet_reply, S, Atom[, CallerTag]}
+*/
+static int inet_reply_am(inet_descriptor* desc, ErlDrvTermData am)
 {
     ErlDrvTermData
         spec[2*LOAD_ATOM_CNT + LOAD_PORT_CNT + LOAD_TUPLE_CNT +
@@ -2603,13 +2630,21 @@ static int inet_reply_ok(inet_descriptor* desc)
 
     i = LOAD_ATOM(spec, i, am_inet_reply);
     i = LOAD_PORT(spec, i, desc->dport);
-    i = LOAD_ATOM(spec, i, am_ok);
+    i = LOAD_ATOM(spec, i, am);
     i = LOAD_TUPLE(spec, i, 3);
     ASSERT(i == sizeof(spec)/sizeof(*spec) - LOAD_EXT_CNT);
  done:
     return inet_reply_finish(desc, spec, i);
 }
 
+/* send:
+**   {inet_reply, S, ok[, CallerTag]}
+*/
+static int inet_reply_ok(inet_descriptor* desc)
+{
+    return inet_reply_am(desc, am_ok);
+}
+
 #ifdef HAVE_SCTP
 static int inet_reply_ok_port(inet_descriptor* desc, ErlDrvTermData dport)
 {
@@ -14447,7 +14482,7 @@ static void packet_inet_command(ErlDrvData e, char* buf, ErlDrvSizeT len)
     char* qtr;
     char* xerror;
     ErlDrvSizeT sz;
-    int code;
+    long code;
     inet_address other;
 
     if (! init_caller(&desc->caller, &desc->caller_ref,
@@ -14505,7 +14540,8 @@ static void packet_inet_command(ErlDrvData e, char* buf, ErlDrvSizeT len)
 	/* Now do the actual sending. NB: "flags" in "sendmsg" itself are NOT
 	   used: */
 	code = sock_sendmsg(desc->s, &mhdr, 0);
-	goto check_result_code;
+
+        goto check_result_code;
     }
 #endif
     {
@@ -14585,16 +14621,30 @@ static void packet_inet_command(ErlDrvData e, char* buf, ErlDrvSizeT len)
 
 #ifdef HAVE_SCTP
  check_result_code:
-    /* "code" analysis is the same for both SCTP and UDP cases above: */
+    /* "code" analysis is the same for both SCTP and UDP above,
+     * although ERRNO_BLOCK | EINTR never happens for UDP
+     */
 #endif
     if (IS_SOCKET_ERROR(code)) {
-	int err = sock_errno();
-	inet_reply_error(desc, err);
+        int err = sock_errno();
+        if ((err != ERRNO_BLOCK) && (err != EINTR)) {
+            inet_reply_error(desc, err);
+            return;
+        }
+        else {
+            /* XXX if(! INET_IGNORED(INETP(desc))) */
+            sock_select(desc, (FD_WRITE|FD_CLOSE), 1);
+            set_busy_port(desc->port, 1);
+            /* XXX add_multi_timer(... desc->send_timeout, ...); */
+            inet_reply_caller_ref(desc);
+            return;
+        }
     }
-    else
-	inet_reply_ok(desc);
-    return;
-    
+    else {
+        inet_reply_ok(desc);
+        return;
+    }
+
  return_einval:
     inet_reply_error(desc, EINVAL);
     return;
@@ -14621,6 +14671,17 @@ static void packet_inet_event(ErlDrvData e, ErlDrvEvent event)
     }
 }
 
+#endif /* #ifdef __WIN32__ */
+
+#ifdef HAVE_UDP
+static void packet_inet_drv_output(ErlDrvData e, ErlDrvEvent event)
+{
+    inet_descriptor *desc = INETP((udp_descriptor *) e);
+
+    sock_select(desc, (FD_WRITE|FD_CLOSE), 0);
+    set_busy_port(desc->port, 0);
+    inet_reply_ok(desc);
+}
 #endif
 
 #ifdef HAVE_UDP

erts/preloaded/ebin/prim_inet.beam

@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2000-2023. All Rights Reserved.
+%% Copyright Ericsson AB 2000-2024. 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.
@@ -564,25 +564,49 @@ peeloff(S, AssocId) ->
 %%
 send(S, Data, OptList) when is_port(S), is_list(OptList) ->
     ?DBG_FORMAT("prim_inet:send(~p, _, ~p)~n", [S,OptList]),
-    Mref = monitor(port, S),
-    MrefBin = term_to_binary(Mref, [local]),
-    MrefBinSize = byte_size(MrefBin),
-    MrefBinSize = MrefBinSize band 16#FFFF,
+    send(S, Data, OptList, monitor(port, S), make_ref()).
+
+send(S, Data, OptList, Mref, Sref) ->
+    SrefBin = term_to_binary(Sref, [local]),
+    SrefBinSize = byte_size(SrefBin),
+    SrefBinSize = SrefBinSize band 16#FFFF,
     try
         erlang:port_command(
-          S, [<<MrefBinSize:16,MrefBin/binary>>, Data], OptList)
+          S, [<<SrefBinSize:16,SrefBin/binary>>, Data], OptList)
     of
         false -> % Port busy when nosuspend option was passed
 	    ?DBG_FORMAT("prim_inet:send() -> {error,busy}~n", []),
             {error,busy};
         true ->
             receive
-                {inet_reply,S,Status,Mref} ->
+                {inet_reply,S,Sref} ->
+                    %% This causes a wait even though nosuspend was used.
+                    %% It only happens when the OS send operation returns
+                    %% that it would block, which should only happen
+                    %% for SCTP (seqpacket), never for UDP (dgram)
+                    %%
+                    %% To fix this we probably need to pass down
+                    %% the nosuspend option to inform inet_drv
+                    %% to not use driver_select and send a late second reply.
+                    %%
+                    ?DBG_FORMAT(
+                       "prim_inet:send(~p,,,) Waiting~n",
+                       [S]),
+                    receive
+                        {inet_reply,S,ok,Sref} ->
+                            send(S, Data, OptList, Mref, make_ref());
+                        {'DOWN',Mref,_,_,_Reason} ->
+                            ?DBG_FORMAT(
+                               "prim_inet:send(~p,,,) 'DOWN' ~p~n",
+                               [S,_Reason]),
+                            {error,closed}
+                    end;
+                {inet_reply,S,Status,Sref} ->
                     demonitor(Mref, [flush]),
                     Status;
                 {'DOWN',Mref,_,_,_Reason} ->
                     ?DBG_FORMAT(
-                       "prim_inet:send_recv_reply(~p, _) 'DOWN' ~p~n",
+                       "prim_inet:send(~p,,,) 'DOWN' ~p~n",
                        [S,_Reason]),
                     {error,closed}
             end