[erts] Make sure nosuspend command signals to ports are handled properly

No-suspend port command signals (i.e. port command signals sent using the
 `erlang:port_command/3` BIF or the `erlang:send/3` BIF with the `nosuspend`
option) were not aborted properly in all scenarios which could leave the
port queue in a busy state indefinitely. Asynchronously sent no-suspend command
signals (i.e, port command signals sent using the `erlang:send/3` BIF with the
`nosuspend` option) could sometimes be delivered even though the port was busy.

Author: rickard-green

erts/emulator/beam/dist.c

@@ -1024,7 +1024,8 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason)
             ERTS_LC_ASSERT(erts_lc_is_port_locked(erts_port_lookup_raw(dep->cid)));
 
             if (erts_port_task_is_scheduled(&dep->dist_cmd))
-                erts_port_task_abort(&dep->dist_cmd);
+                erts_port_task_abort(erts_port_lookup_raw(dep->cid),
+                                     &dep->dist_cmd);
         }
         else {
             ASSERT(is_internal_pid(dep->cid));

erts/emulator/beam/erl_hl_timer.c

@@ -2757,7 +2757,7 @@ erts_cancel_port_timer(Port *c_prt)
     if (tval == ERTS_PTMR_TIMEDOUT) {
 	while (!erts_port_task_is_scheduled(&c_prt->timeout_task))
 	    erts_thr_yield();
-	erts_port_task_abort(&c_prt->timeout_task);
+        erts_port_task_abort(c_prt, &c_prt->timeout_task);
 	erts_atomic_set_nob(&c_prt->common.timer, ERTS_PTMR_NONE);
 	return;
     }

erts/emulator/beam/erl_port.h

@@ -871,6 +871,7 @@ enum {
 #define ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK	ERTS_P2P_SIG_DATA_FLG(5)
 #define ERTS_P2P_SIG_DATA_FLG_SCHED		ERTS_P2P_SIG_DATA_FLG(6)
 #define ERTS_P2P_SIG_DATA_FLG_ASYNC		ERTS_P2P_SIG_DATA_FLG(7)
+#define ERTS_P2P_SIG_DATA_FLG_ASYNC_NOSUSPEND	ERTS_P2P_SIG_DATA_FLG(8)
 
 struct ErtsProc2PortSigData_ {
     int flags;

erts/emulator/beam/erl_port_task.c

@@ -1001,6 +1001,8 @@ enqueue_task(Port *pp,
 	if (ns_pthlp) {
 	    ns_pthlp->u.next = pp->sched.taskq.local.busy.nosuspend;
 	    pp->sched.taskq.local.busy.nosuspend = ns_pthlp;
+            erts_atomic32_read_bor_nob(&pp->sched.flags,
+                                       ERTS_PTS_FLG_HAVE_NS_TASKS);
 	}
 	if (pp->sched.taskq.in.last) {
 	    ASSERT(pp->sched.taskq.in.first);
@@ -1298,7 +1300,7 @@ erts_port_task_tmp_handle_detach(ErtsPortTaskHandle *pthp)
  */
 
 int
-erts_port_task_abort(ErtsPortTaskHandle *pthp)
+erts_port_task_abort(Port *pp, ErtsPortTaskHandle *pthp)
 {
     int res;
     ErtsPortTask *ptp;
@@ -1326,22 +1328,33 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp)
 	    res = - 1; /* Task already aborted, executing, or executed */
 	else {
 	    reset_port_task_handle(pthp);
+            switch (ptp->type) {
+            case ERTS_PORT_TASK_PROC_SIG: {
+                int abort_type = ((ptp->u.alive.flags
+                                   & (ERTS_PT_FLG_NOSUSPEND
+                                      | ERTS_PT_FLG_ASYNC_NOSUSPEND))
+                                  ? ERTS_PROC2PORT_SIG_ABORT_NOSUSPEND
+                                  : ERTS_PROC2PORT_SIG_ABORT);
+                ASSERT(pp);
+                abort_signal_task(pp, abort_type, ptp->type, &ptp->u.alive.td,
+                                  pp->sched.taskq.bpq != NULL);
+                break;
+            }
+            case ERTS_PORT_TASK_INPUT:
+            case ERTS_PORT_TASK_OUTPUT:
 #if ERTS_POLL_USE_SCHEDULER_POLLING
-            if (erts_sched_poll_enabled()) {
-                switch (ptp->type) {
-                case ERTS_PORT_TASK_INPUT:
-                case ERTS_PORT_TASK_OUTPUT:
+                if (erts_sched_poll_enabled()) {
                     if (ptp->u.alive.td.io.is_scheduler_event) {
                         ASSERT(erts_atomic_read_nob(
                                    &erts_port_task_outstanding_io_tasks) > 0);
                         erts_atomic_dec_relb(&erts_port_task_outstanding_io_tasks);
                     }
-                    break;
-                default:
-                    break;
                 }
-            }
 #endif
+                break;
+            default:
+                break;
+            }
 	    res = 0;
 	}
     }
@@ -1369,7 +1382,6 @@ erts_port_task_abort_nosuspend_tasks(Port *pp)
 	ErtsPortTaskHandle *saved_pthp;
 #endif
 	ErtsPortTaskType type;
-	ErtsPortTaskTypeData td;
 	ErtsPortTaskHandle *pthp;
 	ErtsPortTask *ptp;
 	ErtsPortTaskHandleList *pthlp;
@@ -1413,13 +1425,14 @@ erts_port_task_abort_nosuspend_tasks(Port *pp)
 	reset_port_task_handle(pthp);
 
 	type = ptp->type;
-	td = ptp->u.alive.td;
+
+        abort_nosuspend_task(pp, type, &ptp->u.alive.td,
+                             pp->sched.taskq.bpq != NULL);
 
 	if (dhndl != ERTS_THR_PRGR_DHANDLE_MANAGED)
 	    erts_thr_progress_unmanaged_continue(dhndl);
 	schedule_port_task_handle_list_free(pthlp);
 
-	abort_nosuspend_task(pp, type, &td, pp->sched.taskq.bpq != NULL);
     }
 }
 
@@ -1520,8 +1533,6 @@ erts_port_task_schedule(Eterm id,
     }
 
     add_flags = ERTS_PTS_FLG_HAVE_TASKS;
-    if (ns_pthlp)
-	add_flags |= ERTS_PTS_FLG_HAVE_NS_TASKS;
 
     prof_runnable_ports = erts_system_profile_flags.runnable_ports;
     if (prof_runnable_ports)

erts/emulator/beam/erl_port_task.h

@@ -54,6 +54,7 @@ typedef erts_atomic_t ErtsPortTaskHandle;
 #define ERTS_PT_FLG_NOSUSPEND		(1 << 2)
 #define ERTS_PT_FLG_REF			(1 << 3)
 #define ERTS_PT_FLG_BAD_OUTPUT		(1 << 4)
+#define ERTS_PT_FLG_ASYNC_NOSUSPEND     (1 << 5)
 
 typedef enum {
     ERTS_PORT_TASK_INPUT = 0,
@@ -239,7 +240,7 @@ void erts_port_task_init(void);
 void erts_port_task_pre_alloc_init_thread(void);
 
 void erts_port_task_tmp_handle_detach(ErtsPortTaskHandle *);
-int erts_port_task_abort(ErtsPortTaskHandle *);
+int erts_port_task_abort(Port *, ErtsPortTaskHandle *);
 
 void erts_port_task_abort_nosuspend_tasks(Port *);
 

erts/emulator/beam/io.c

@@ -2110,6 +2110,8 @@ erts_port_output(Process *c_p,
 	    ASSERT(esdp);
 	    ns_pthp = &esdp->nosuspend_port_task_handle;
 	    sigdp->flags &= ~ERTS_P2P_SIG_DATA_FLG_NOSUSPEND;
+            sigdp->flags |= ERTS_P2P_SIG_DATA_FLG_ASYNC_NOSUSPEND;
+            task_flags = ERTS_PT_FLG_WAIT_BUSY|ERTS_PT_FLG_ASYNC_NOSUSPEND;
 	}
 	else if (flags & ERTS_P2P_SIG_DATA_FLG_NOSUSPEND)
 	    task_flags = ERTS_PT_FLG_NOSUSPEND;
@@ -2141,7 +2143,7 @@ erts_port_output(Process *c_p,
 	    if (!async_nosuspend)
 		return ERTS_PORT_OP_BUSY_SCHEDULED;
 	    else {
-		if (erts_port_task_abort(ns_pthp) == 0)
+                if (erts_port_task_abort(prt, ns_pthp) == 0)
 		    return ERTS_PORT_OP_BUSY;
 		else
 		    erts_port_task_tmp_handle_detach(ns_pthp);

erts/emulator/sys/common/erl_check_io.c

@@ -573,7 +573,7 @@ static ERTS_INLINE void
 abort_task(Eterm id, ErtsPortTaskHandle *pthp, EventStateType type)
 {
     if (is_not_nil(id) && erts_port_task_is_scheduled(pthp)) {
-	erts_port_task_abort(pthp);
+        erts_port_task_abort(NULL, pthp);
 	ASSERT(erts_is_port_alive(id));
     }
 }

erts/emulator/test/busy_port_SUITE.erl

@@ -28,7 +28,11 @@
          scheduling_delay_busy/1,
          scheduling_delay_busy_nosuspend/1,
          scheduling_busy_link/1,
-         busy_with_signals/1]).
+         busy_with_signals/1,
+         async_par_nosuspend_abort_cleanup/1,
+         async_nopar_nosuspend_abort_cleanup/1,
+         sync_par_nosuspend_abort_cleanup/1,
+         sync_nopar_nosuspend_abort_cleanup/1]).
 
 -include_lib("common_test/include/ct.hrl").
 
@@ -44,7 +48,11 @@ all() ->
      no_trap_exit, no_trap_exit_unlinked, trap_exit,
      multiple_writers, hard_busy_driver, soft_busy_driver,
      scheduling_delay_busy,scheduling_delay_busy_nosuspend,
-     scheduling_busy_link, busy_with_signals].
+     scheduling_busy_link, busy_with_signals,
+     async_par_nosuspend_abort_cleanup,
+     async_nopar_nosuspend_abort_cleanup,
+     sync_par_nosuspend_abort_cleanup,
+     sync_nopar_nosuspend_abort_cleanup].
 
 init_per_testcase(_Case, Config) when is_list(Config) ->
     Killer = spawn(fun() -> killer_loop([]) end),
@@ -844,6 +852,69 @@ flood_with_exit_signals(Pid, N) ->
     exit(Pid, pling),
     flood_with_exit_signals(Pid, N-1).
 
+async_par_nosuspend_abort_cleanup(Config) when is_list(Config) ->
+    nosuspend_abort_cleanup_tests(Config, false, true).
+
+async_nopar_nosuspend_abort_cleanup(Config) when is_list(Config) ->
+    nosuspend_abort_cleanup_tests(Config, false, false).
+
+sync_par_nosuspend_abort_cleanup(Config) when is_list(Config) ->
+    nosuspend_abort_cleanup_tests(Config, true, true).
+
+sync_nopar_nosuspend_abort_cleanup(Config) when is_list(Config) ->
+    nosuspend_abort_cleanup_tests(Config, true, false).
+
+nosuspend_abort_cleanup_tests(Config, Sync, Parallelism) ->
+    Master = self(),
+    process_flag(priority,high),
+    Drv = "nosuspend_test_drv",
+    DataDir = proplists:get_value(data_dir, Config),
+    erl_ddll:start(),
+    case erl_ddll:load_driver(DataDir, Drv) of
+        ok ->
+            ok;
+        {error, Error} ->
+            ct:fail(erl_ddll:format_error(Error))
+    end,
+    Port = open_port({spawn_driver, Drv},
+                     [{busy_limits_msgq, {1024, 2048}},
+                      {parallelism, Parallelism}]),
+    NoSuspFun = if Sync ->
+                        fun () ->
+                                port_command(Port, <<0:(16*8*1024)>>, [nosuspend])
+                        end;
+                   true ->
+                        fun () ->
+                                erlang:send(Port,
+                                            {Master, {command, <<0:(16*8*1024)>>}},
+                                            [nosuspend])
+                        end
+                end,
+    NoSuspers = lists:map(fun (_) -> spawn_link(NoSuspFun) end,
+                          lists:seq(1, 10000)),
+    _= port_control(Port, $B, <<>>),
+    receive after 1000 -> ok end,
+    _= port_control(Port, $N, <<>>),
+    SyncCmd = spawn_link(fun () ->
+                                 port_command(Port, <<"should not block for long">>),
+                                 Master ! {self(), done}
+                         end),
+    receive
+        {SyncCmd, done} -> ok
+    after
+        10000 -> ct:fail(port_command_blocked)
+    end,
+    lists:foreach(fun (P) ->
+                          unlink(P),
+                          exit(P,kill),
+                          false = is_process_alive(P)
+                  end,
+                  [SyncCmd | NoSuspers]),
+    true = erlang:port_close(Port),
+    ok = erl_ddll:unload_driver(Drv),
+    ok = erl_ddll:stop(),
+    ok.
+
 %%% Utilities.
 
 pal(_F,_A) -> ok.

erts/emulator/test/busy_port_SUITE_data/Makefile.src

@@ -18,7 +18,7 @@
 # %CopyrightEnd%
 #
 
-all: busy_drv@dll@ hard_busy_drv@dll@ soft_busy_drv@dll@ scheduling_drv@dll@
+all: busy_drv@dll@ hard_busy_drv@dll@ soft_busy_drv@dll@ scheduling_drv@dll@ nosuspend_test_drv@dll@
 
 @SHLIB_RULES@
 

erts/emulator/test/busy_port_SUITE_data/nosuspend_test_drv.c

@@ -0,0 +1,110 @@
+/*
+ * %CopyrightBegin%
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Copyright Ericsson AB 2026. 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.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * %CopyrightEnd%
+ */
+
+#include <errno.h>
+#include <stdbool.h>
+#include "erl_driver.h"
+
+static void stop(ErlDrvData drv_data);
+static ErlDrvData start(ErlDrvPort port, char *command);
+static void output(ErlDrvData drv_data, char *buf, ErlDrvSizeT len);
+static ErlDrvSSizeT control(ErlDrvData drv_data, unsigned int command, char *buf,
+                            ErlDrvSizeT len, char **rbuf, ErlDrvSizeT rlen);
+
+static ErlDrvEntry nosuspend_test_drv_entry = { 
+    NULL /* init */,
+    start,
+    stop,
+    output,
+    NULL /* ready_input */,
+    NULL /* ready_output */,
+    "nosuspend_test_drv",
+    NULL /* finish */,
+    NULL /* handle */,
+    control,
+    NULL /* timeout */,
+    NULL /* outputv */,
+    NULL /* ready_async */,
+    NULL /* flush */,
+    NULL /* call */,
+    NULL /* event */,
+    ERL_DRV_EXTENDED_MARKER,
+    ERL_DRV_EXTENDED_MAJOR_VERSION,
+    ERL_DRV_EXTENDED_MINOR_VERSION,
+    ERL_DRV_FLAG_USE_PORT_LOCKING,
+    NULL /* handle2 */,
+    NULL /* handle_monitor */,
+    NULL /* stop_select */
+};
+
+typedef struct {
+    ErlDrvPort port;
+    bool is_busy;
+} PortState;
+
+DRIVER_INIT(nosuspend_test_drv)
+{
+    return &nosuspend_test_drv_entry;
+}
+
+static void stop(ErlDrvData drv_data)
+{
+    driver_free(drv_data);
+}
+
+static ErlDrvData start(ErlDrvPort port, char *command)
+{
+    PortState *state = driver_alloc(sizeof(PortState));
+    if (!state)
+        return ERL_DRV_ERROR_GENERAL;
+    state->port = port;
+    state->is_busy = false;
+    return (ErlDrvData) state;
+}
+
+static void output(ErlDrvData drv_data, char *buf, ErlDrvSizeT len)
+{
+    PortState *state = (PortState *) drv_data;
+    if (state->is_busy) {
+        driver_failure_atom(state->port, "got_data_when_busy");
+    }
+}
+
+static ErlDrvSSizeT control(ErlDrvData drv_data, unsigned int command, char *buf,
+                            ErlDrvSizeT len, char **rbuf, ErlDrvSizeT rlen)
+{
+    PortState *state = (PortState *) drv_data;
+    switch (command) {
+    case 'B': /* busy */
+        set_busy_port(state->port, !0);
+        state->is_busy = true;
+        break;
+    case 'N': /* not busy */
+        set_busy_port(state->port, 0);
+        state->is_busy = false;
+        break;
+    default:
+        driver_failure_posix(state->port, EINVAL);
+        break;
+    }
+    return 0;
+}