Merge branch 'maint'

* maint:
  Fix build of test port program
  Update Makefile.src
  Add async_ports test
  Fix abort of nosuspend-tasks in erts_port_task_schedule()

5 files changed, 220 insertions, 4 deletions

diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index 31d9a1e26e..2fc95ed5d8 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -798,12 +798,13 @@ schedule_port_task_handle_list_free(ErtsPortTaskHandleList *pthlp)
 static ERTS_INLINE void
 abort_nosuspend_task(Port *pp,
 		     ErtsPortTaskType type,
-		     ErtsPortTaskTypeData *tdp)
+		     ErtsPortTaskTypeData *tdp,
+		     int bpq_data)
 {
 
     ASSERT(type == ERTS_PORT_TASK_PROC_SIG);
 
-    if (!pp->sched.taskq.bpq)
+    if (!bpq_data)
 	tdp->psig.callback(NULL,
 			   ERTS_PORT_SFLG_INVALID,
 			   ERTS_PROC2PORT_SIG_ABORT_NOSUSPEND,
@@ -1345,7 +1346,7 @@ erts_port_task_abort_nosuspend_tasks(Port *pp)
 #endif
 	schedule_port_task_handle_list_free(pthlp);
 
-	abort_nosuspend_task(pp, type, &td);
+	abort_nosuspend_task(pp, type, &td, pp->sched.taskq.bpq != NULL);
     }
 }
 
@@ -1525,7 +1526,7 @@ abort_nosuspend:
 	erts_port_dec_refc(pp);
 #endif
 
-    abort_nosuspend_task(pp, ptp->type, &ptp->u.alive.td);
+    abort_nosuspend_task(pp, ptp->type, &ptp->u.alive.td, 0);
 
     ASSERT(ns_pthlp);
     erts_free(ERTS_ALC_T_PT_HNDL_LIST, ns_pthlp);
diff --git a/erts/emulator/test/Makefile b/erts/emulator/test/Makefile
index 0b0568c31a..dfbe47786a 100644
--- a/erts/emulator/test/Makefile
+++ b/erts/emulator/test/Makefile
@@ -31,6 +31,7 @@ MODULES= \
 	a_SUITE \
 	after_SUITE \
 	alloc_SUITE \
+	async_ports_SUITE \
 	beam_SUITE \
 	beam_literals_SUITE \
 	bif_SUITE \
diff --git a/erts/emulator/test/async_ports_SUITE.erl b/erts/emulator/test/async_ports_SUITE.erl
new file mode 100644
index 0000000000..c89b3655ff
--- /dev/null
+++ b/erts/emulator/test/async_ports_SUITE.erl
@@ -0,0 +1,118 @@
+-module(async_ports_SUITE).
+
+-include_lib("common_test/include/ct.hrl").
+
+-compile(export_all).
+
+-define(PACKET_SIZE, (10 * 1024 * 8)).
+-define(CPORT_DELAY, 100).
+-define(TEST_LOOPS_COUNT, 100000).
+-define(SLEEP_BEFORE_CHECK, 1000).
+-define(TEST_PROCS_COUNT, 2).
+-define(TC_TIMETRAP_SECONDS, 10).
+
+suite() -> [{ct_hooks,[ts_install_cth]}].
+
+all() ->
+    [
+     permanent_busy_test
+    ].
+
+permanent_busy_test(Config) ->
+    ct:timetrap({seconds, ?TC_TIMETRAP_SECONDS}),
+    ExePath = filename:join(?config(data_dir, Config), "cport"),
+
+    Self = self(),
+    spawn_link(
+      fun() ->
+              Block = <<0:?PACKET_SIZE>>,
+
+              Port = open_port(ExePath),
+              
+              Testers =
+                  lists:map(
+                    fun(_) ->
+                            erlang:spawn_link(?MODULE, run_loop,
+                                              [Self,
+                                               Port,
+                                               Block,
+                                               ?TEST_LOOPS_COUNT,
+                                               0])
+                    end,
+                    lists:seq(1, ?TEST_PROCS_COUNT)),
+              Self ! {test_info, Port, Testers},
+              endless_flush(Port)
+      end),
+
+    receive
+        {test_info, Port, Testers} ->
+            MaxWaitTime = round(0.7 * ?TC_TIMETRAP_SECONDS * 1000),
+            ct:log("wait testers, maximum ~w mcsec~n", [MaxWaitTime]),
+            ok = wait_testers(MaxWaitTime, Testers),
+            timer:sleep(?SLEEP_BEFORE_CHECK),
+            case erlang:port_command(Port, <<"test">>, [nosuspend]) of
+                false ->
+                    exit(port_dead);
+                true ->
+                    ok
+            end
+    end.
+
+wait_testers(Timeout, Testers) ->
+    lists:foldl(
+      fun(Pid, AccIn) ->
+              StartWait = os:timestamp(),
+              receive
+                  {Pid, port_dead} ->
+                      recalc_timeout(AccIn, StartWait)
+              after AccIn ->
+                      Pid ! stop,
+                      recalc_timeout(AccIn, StartWait)
+              end
+      end, Timeout, Testers),
+    ok.
+
+recalc_timeout(TimeoutIn, WaitStart) ->
+    erlang:max(0, TimeoutIn - round(timer:now_diff(os:timestamp(), WaitStart)) div 1000).
+
+open_port(ExePath) ->
+    erlang:open_port({spawn, ExePath ++ " 100"}, [{packet, 4}, eof, exit_status, use_stdio, binary]).
+
+run_loop(RootProc, Port, Block, CheckLimit, BusyCnt) ->
+    receive
+        stop ->
+            ok
+    after 0 ->
+            case erlang:port_command(Port, Block, [nosuspend]) of
+                true ->
+                    run_loop(RootProc, Port, Block, CheckLimit, 0);
+                false ->
+                    if
+                        BusyCnt + 1 > CheckLimit ->
+                            check_dead(RootProc, Port, Block, CheckLimit);
+                        true ->
+                            run_loop(RootProc, Port, Block, CheckLimit, BusyCnt + 1)
+                    end
+            end
+    end.
+
+check_dead(RootProc, Port, Block, CheckLimit) ->
+    ct:log("~p: check port dead~n", [self()]),
+    timer:sleep(?SLEEP_BEFORE_CHECK),
+    case erlang:port_command(Port, Block, [nosuspend]) of
+        true ->
+            ct:log("not dead~n"),
+            run_loop(RootProc, Port, Block, CheckLimit, 0);
+        false ->
+            ct:log("port dead: ~p~n", [Port]),
+            RootProc ! {self(), port_dead},
+            ok
+    end.
+
+endless_flush(Port) ->
+    receive
+        {Port, {data, _}} ->
+            endless_flush(Port);
+        {Port, SomethingWrong} ->
+            erlang:error({someting_wrong, SomethingWrong})
+    end.
diff --git a/erts/emulator/test/async_ports_SUITE_data/Makefile.src b/erts/emulator/test/async_ports_SUITE_data/Makefile.src
new file mode 100644
index 0000000000..56da3fbe12
--- /dev/null
+++ b/erts/emulator/test/async_ports_SUITE_data/Makefile.src
@@ -0,0 +1,15 @@
+CC = @CC@
+LD = @LD@
+CFLAGS = @CFLAGS@ @DEFS@
+CROSSLDFLAGS = @CROSSLDFLAGS@
+
+PROGS = cport@exe@
+
+
+all: $(PROGS)
+
+cport@exe@: cport@obj@
+	$(LD) $(CROSSLDFLAGS) -o cport cport@obj@ @LIBS@
+
+cport@obj@: cport.c
+	$(CC) -c -o cport@obj@ $(CFLAGS) cport.c
diff --git a/erts/emulator/test/async_ports_SUITE_data/cport.c b/erts/emulator/test/async_ports_SUITE_data/cport.c
new file mode 100644
index 0000000000..033aff382a
--- /dev/null
+++ b/erts/emulator/test/async_ports_SUITE_data/cport.c
@@ -0,0 +1,81 @@
+#include <stdlib.h>
+#include <stdio.h>
+#include <errno.h>
+#include <string.h>
+#ifdef __WIN32__
+#  include "windows.h"
+#  include "winbase.h"
+#else
+#  include <unistd.h>
+#endif
+
+typedef unsigned char byte;
+
+int read_cmd(byte *buf)
+{
+  int len;
+  if (read_exact(buf, 4) != 4)
+      return(-1);
+
+  len = (buf[0] << 24) | (buf[1] << 16) | (buf[2] << 8) | buf[3];
+  return read_exact(buf, len);
+}
+
+int write_cmd(byte *buf, int len)
+{
+  byte li[4];
+  li[0] = (len >> 24) & 0xff;
+  li[1] = (len >> 16) & 0xff;
+  li[2] = (len >> 8)  & 0xff;
+  li[3] = len  & 0xff;
+  write_exact(&li, 4);
+
+  return write_exact(buf, len);
+}
+
+int read_exact(byte *buf, int len)
+{
+  int i, got=0;
+  do {
+    if ((i = read(0, buf+got, len-got)) <= 0)
+      {
+	return(i);
+      }
+    got += i;
+  } while (got<len);
+  return len;
+}
+
+int write_exact(byte *buf, int len)
+{
+  int i, wrote = 0;
+  do {
+    if ((i = write(1, buf+wrote, len-wrote)) < 0)
+      return (i);
+    wrote += i;
+  } while (wrote<len);
+  return len;
+}
+
+byte static_buf[31457280]; // 30 mb
+
+int main(int argc, char **argv) {
+  int sleep_time = atoi(argv[1]);
+  int fn, arg, res;
+  byte *buf = &static_buf[0];
+  int len = 0;
+  if (sleep_time <= 0)
+    sleep_time = 0;
+#ifdef __WIN32__
+  else
+    sleep_time = ((sleep_time - 1) / 1000) + 1; /* Milli seconds */
+#endif
+  while ((len = read_cmd(buf)) > 0) {
+#ifdef __WIN32__
+    Sleep((DWORD) sleep_time);
+#else
+    usleep(sleep_time);
+#endif
+    write_cmd(buf, len);
+  }
+}