Add async_ports test

4 files changed, 188 insertions, 0 deletions

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..9e0cc8ffab
--- /dev/null
+++ b/erts/emulator/test/async_ports_SUITE.erl
@@ -0,0 +1,117 @@
+-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).
+
+
+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..17e6a6ee6a
--- /dev/null
+++ b/erts/emulator/test/async_ports_SUITE_data/Makefile.src
@@ -0,0 +1,5 @@
+all: busy_drv@dll@ hard_busy_drv@dll@ soft_busy_drv@dll@ scheduling_drv@dll@
+
+@SHLIB_RULES@
+
+cport@obj@: 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..179cdc4b8b
--- /dev/null
+++ b/erts/emulator/test/async_ports_SUITE_data/cport.c
@@ -0,0 +1,65 @@
+#include <stdlib.h>
+#include <stdio.h>
+#include <malloc.h>
+#include <errno.h>
+#include <string.h>
+#include <unistd.h>
+
+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;
+}
+
+int main(int argc, char **argv) {
+  int sleep_time = atoi(argv[1]);
+  int fn, arg, res;
+  byte *buf = malloc(31457280); // 30 mb
+  int len = 0;
+  while ((len = read_cmd(buf)) > 0) {
+    usleep(sleep_time);
+    write_cmd(buf, len);
+  }
+}