Implement true asynchronous signaling between processes and ports

3 files changed, 66 insertions, 50 deletions

diff --git a/erts/emulator/test/busy_port_SUITE.erl b/erts/emulator/test/busy_port_SUITE.erl
index 3a29fd4d68..32e907ca69 100644
--- a/erts/emulator/test/busy_port_SUITE.erl
+++ b/erts/emulator/test/busy_port_SUITE.erl
@@ -148,9 +148,9 @@ message_order(Config) when is_list(Config) ->
 
 send_to_busy_1(Parent) ->
     {Owner, Slave} = get_slave(),
-    Slave ! {Owner, {command, "set_me_busy"}},
-    Slave ! {Owner, {command, "hello"}},
-    Slave ! {Owner, {command, "hello again"}},
+    (catch port_command(Slave, "set_me_busy")),
+    (catch port_command(Slave, "hello")),
+    (catch port_command(Slave, "hello again")),
     receive
 	Message ->
 	    Parent ! {self(), Message}
@@ -193,10 +193,10 @@ system_monitor(Config) when is_list(Config) ->
     ?line Busy = 
 	spawn_link(
 	  fun() ->
-		  Slave ! {Owner,{command,"set busy"}},
+		  (catch port_command(Slave, "set busy")),
 		  receive {Parent,alpha} -> ok end,
-		  Slave ! {Owner,{command,"busy"}},
-		  Slave ! {Owner,{command,"free"}},
+		  (catch port_command(Slave, "busy")),
+		  (catch port_command(Slave, "free")),
 		  Parent ! {self(),alpha},
 		  command(lock),
 		  receive {Parent,beta} -> ok end,
@@ -212,7 +212,7 @@ system_monitor(Config) when is_list(Config) ->
     ?line Void = rec(Void),
     ?line Busy ! {self(), beta},
     ?line {monitor,Owner,busy_port,Slave} = rec(Void),
-    ?line Master ! {Owner, {command, "u"}},
+    ?line port_command(Master, "u"),
     ?line {Busy,beta} = rec(Void),
     ?line Void = rec(Void),
     ?line _NewMonitor = erlang:system_monitor(OldMonitor),
@@ -296,9 +296,9 @@ no_trap_exit_process(ResultTo, Link, Config) ->
 	      linked -> ok;
 	      unlink -> unlink(Slave)
 	  end,
-    ?line Slave ! {self(), {command, "lock port"}},
+    ?line (catch port_command(Slave, "lock port")),
     ?line ResultTo ! {self(), port_created, Slave},
-    ?line Slave ! {self(), {command, "suspend me"}},
+    ?line (catch port_command(Slave, "suspend me")),
     ok.
 
 %% Assuming the following scenario,
@@ -339,9 +339,9 @@ busy_port_exit_process(ResultTo, Config) ->
     ?line load_busy_driver(Config),
     ?line _Master = open_port({spawn, "busy_drv master"}, [eof]),
     ?line Slave = open_port({spawn, "busy_drv slave"}, [eof]),
-    ?line Slave ! {self(), {command, "lock port"}},
+    ?line (catch port_command(Slave, "lock port")),
     ?line ResultTo ! {self(), port_created, Slave},
-    ?line Slave ! {self(), {command, "suspend me"}},
+    ?line (catch port_command(Slave, "suspend me")),
     receive
 	{'EXIT', Slave, die} ->
 	    ResultTo ! {self(), ok};
@@ -383,8 +383,8 @@ multiple_writers(Config) when is_list(Config) ->
 
 quick_writer() ->
     {Owner, Port} = get_slave(),
-    Port ! {Owner, {command, "port to busy"}},
-    Port ! {Owner, {command, "lock me"}},
+    (catch port_command(Port, "port to busy")),
+    (catch port_command(Port, "lock me")),
     ok.
 
 hard_busy_driver(Config) when is_list(Config) ->
@@ -644,11 +644,11 @@ loop(Master, Slave) ->
 	    Pid ! {busy_drv_reply, {self(), Slave}},
 	    loop(Master, Slave);
 	{Pid, unlock} ->
-	    Master ! {self(), {command, "u"}},
+	    port_command(Master, "u"),
 	    Pid ! {busy_drv_reply, ok},
 	    loop(Master, Slave);
 	{Pid, lock} ->
-	    Master ! {self(), {command, "l"}},
+	    port_command(Master, "l"),
 	    Pid ! {busy_drv_reply, ok},
 	    loop(Master, Slave);
 	{Pid, {port_command,Data}} ->
diff --git a/erts/emulator/test/ddll_SUITE.erl b/erts/emulator/test/ddll_SUITE.erl
index 6e15c228cd..4675cab15c 100644
--- a/erts/emulator/test/ddll_SUITE.erl
+++ b/erts/emulator/test/ddll_SUITE.erl
@@ -136,8 +136,8 @@ delayed_unload_with_ports(Config) when is_list(Config) ->
     ?line {ok,pending_driver,Ref} = erl_ddll:try_unload(echo_drv,[{monitor, pending_driver}]),
     ?line ok = receive _ -> false after 0 -> ok end,
     ?line Port ! {self(), close},
-    ?line 1 = erl_ddll:info(echo_drv, port_count),
     ?line ok = receive {Port,closed} -> ok after 1000 -> false end,
+    ?line 1 = erl_ddll:info(echo_drv, port_count),
     ?line Port2 ! {self(), close},
     ?line ok = receive {Port2,closed} -> ok after 1000 -> false end,
     ?line ok = receive {'DOWN', Ref, driver, echo_drv, unloaded} -> ok after 1000 -> false end,
diff --git a/erts/emulator/test/port_SUITE.erl b/erts/emulator/test/port_SUITE.erl
index 0a1ef5a78f..b00e7f0b59 100644
--- a/erts/emulator/test/port_SUITE.erl
+++ b/erts/emulator/test/port_SUITE.erl
@@ -90,7 +90,7 @@
 	 mix_up_ports/1, otp_5112/1, otp_5119/1, otp_6224/1,
 	 exit_status_multi_scheduling_block/1, ports/1,
 	 spawn_driver/1, spawn_executable/1, close_deaf_port/1,
-	 unregister_name/1]).
+	 unregister_name/1, parallelism_option/1]).
 
 -export([]).
 
@@ -114,7 +114,8 @@ all() ->
      stderr_to_stdout, otp_3906, otp_4389, win_massive,
      mix_up_ports, otp_5112, otp_5119,
      exit_status_multi_scheduling_block, ports, spawn_driver,
-     spawn_executable, close_deaf_port, unregister_name].
+     spawn_executable, close_deaf_port, unregister_name,
+     parallelism_option].
 
 groups() -> 
     [{stream, [], [stream_small, stream_big]},
@@ -1364,6 +1365,43 @@ spawn_driver(Config) when is_list(Config) ->
     ?line test_server:timetrap_cancel(Dog),
     ok.
 
+parallelism_option(suite) ->
+    [];
+parallelism_option(doc) ->
+    ["Test parallelism option of open_port"];
+parallelism_option(Config) when is_list(Config) ->
+    ?line Dog = test_server:timetrap(test_server:seconds(10)),
+    ?line Path = ?config(data_dir, Config),
+    ?line ok = load_driver(Path, "echo_drv"),
+    ?line Port = erlang:open_port({spawn_driver, "echo_drv"},
+				  [{parallelism, true}]),
+    ?line {parallelism, true} = erlang:port_info(Port, parallelism),
+    ?line Port ! {self(), {command, "Hello port!"}},
+    ?line receive 
+	      {Port, {data, "Hello port!"}} = Msg1 -> 
+		  io:format("~p~n", [Msg1]),
+		  ok; 
+	      Other ->
+		  test_server:fail({unexpected, Other})
+	  end,
+    ?line Port ! {self(), close},
+    ?line receive {Port, closed} -> ok end,
+
+    ?line Port2 = erlang:open_port({spawn_driver, "echo_drv -Hello port?"}, 
+				   [{parallelism, false}]),
+    ?line {parallelism, false} = erlang:port_info(Port2, parallelism),
+    ?line receive 
+	      {Port2, {data, "Hello port?"}} = Msg2 -> 
+		  io:format("~p~n", [Msg2]),
+		  ok; 
+	      Other2 ->
+		  test_server:fail({unexpected2, Other2})
+	  end,
+    ?line Port2 ! {self(), close},
+    ?line receive {Port2, closed} -> ok end,
+    ?line test_server:timetrap_cancel(Dog),
+    ok.
+
 spawn_executable(suite) ->
     [];
 spawn_executable(doc) ->
@@ -1632,6 +1670,7 @@ otp_5112(Config) when is_list(Config) ->
     ?line ?t:format("Links1: ~p~n",[Links1]),
     ?line true = lists:member(Port, Links1),
     ?line Port ! {self(), {command, ""}},
+    ?line wait_until(fun () -> lists:member(Port, erlang:ports()) == false end),
     ?line {links, Links2} = process_info(self(),links),
     ?line ?t:format("Links2: ~p~n",[Links2]),
     ?line false = lists:member(Port, Links2), %% This used to fail
@@ -1703,38 +1742,8 @@ otp_5119_fill_empty_port_tab(Ports) ->
 		  ?line LastPort
 	  end.
 
--define(DEF_MAX_PORTS, 1024).
-
-max_ports_env() ->
-    ?line case os:getenv("ERL_MAX_PORTS") of
-	      EMP when is_list(EMP) ->
-		  case catch list_to_integer(EMP) of
-		      Int when is_integer(Int) -> ?line Int;
-		      _ -> ?line false
-		  end;
-	      _ -> ?line false
-	  end.
-
 max_ports() ->
-    ?line PreMaxPorts
-	= case max_ports_env() of
-	      Env when is_integer(Env) -> ?line Env;
-	      _ ->
-		  ?line case os:type() of
-			    {unix, _} ->
-				?line UlimStr = string:strip(os:cmd("ulimit -n")
-							     -- "\n"),
-				?line case catch list_to_integer(UlimStr) of
-					  Ulim when is_integer(Ulim) -> ?line Ulim;
-					  _ -> ?line ?DEF_MAX_PORTS
-				      end;
-			    _ -> ?line ?DEF_MAX_PORTS
-			end
-	  end,
-    ?line case PreMaxPorts > ?DEF_MAX_PORTS of
-	      true -> ?line PreMaxPorts;
-	      false -> ?line ?DEF_MAX_PORTS
-	  end.
+    erlang:system_info(port_limit).
 
 port_ix(Port) when is_port(Port) ->
     ?line ["#Port",_,PortIxStr] = string:tokens(erlang:port_to_list(Port),
@@ -2344,5 +2353,12 @@ close_deaf_port_1(N, Cmd) ->
     	_:eagain ->
 	    {comment, "Could not spawn more than " ++ integer_to_list(N) ++ " OS processes."}
     end.
-    
 
+wait_until(Fun) ->
+    case catch Fun() of
+	true ->
+	    ok;
+	_ ->
+	    receive after 100 -> ok end,
+	    wait_until(Fun)
+    end.