Merge branch 'rickard/r16/port-optimizations/OTP-10336'

* rickard/r16/port-optimizations/OTP-10336:
  Change annotate level for emacs-22 in cerl
  Update etp-commands
  Add documentation on communication in Erlang
  Add support for busy port message queue
  Add driver callback epilogue
  Implement true asynchronous signaling between processes and ports
  Add erl_drv_[send|output]_term
  Move busy port flag
  Use rwlock for driver list
  Optimize management of port tasks
  Improve configuration of process and port tables
  Remove R9 compatibility features
  Use ptab functionality also for ports
  Prepare for use of ptab functionality also for ports
  Atomic port state
  Generalize process table implementation
  Implement functionality for delaying thread progress from unmanaged threads

13 files changed, 374 insertions, 102 deletions

diff --git a/erts/preloaded/ebin/erl_prim_loader.beam b/erts/preloaded/ebin/erl_prim_loader.beam
index e46795e2b1..eb6b5259b5 100644
--- a/erts/preloaded/ebin/erl_prim_loader.beam
+++ b/erts/preloaded/ebin/erl_prim_loader.beam
diff --git a/erts/preloaded/ebin/erlang.beam b/erts/preloaded/ebin/erlang.beam
index bf6d6e871b..074e31fb13 100644
--- a/erts/preloaded/ebin/erlang.beam
+++ b/erts/preloaded/ebin/erlang.beam
diff --git a/erts/preloaded/ebin/erts_internal.beam b/erts/preloaded/ebin/erts_internal.beam
new file mode 100644
index 0000000000..c647fa5d4e
--- /dev/null
+++ b/erts/preloaded/ebin/erts_internal.beam
diff --git a/erts/preloaded/ebin/init.beam b/erts/preloaded/ebin/init.beam
index 1a250932da..4a34268f06 100644
--- a/erts/preloaded/ebin/init.beam
+++ b/erts/preloaded/ebin/init.beam
diff --git a/erts/preloaded/ebin/otp_ring0.beam b/erts/preloaded/ebin/otp_ring0.beam
index 2fa007f683..6fcf515461 100644
--- a/erts/preloaded/ebin/otp_ring0.beam
+++ b/erts/preloaded/ebin/otp_ring0.beam
diff --git a/erts/preloaded/ebin/prim_file.beam b/erts/preloaded/ebin/prim_file.beam
index 19cf856754..6d6d70920e 100644
--- a/erts/preloaded/ebin/prim_file.beam
+++ b/erts/preloaded/ebin/prim_file.beam
diff --git a/erts/preloaded/ebin/prim_inet.beam b/erts/preloaded/ebin/prim_inet.beam
index b2d224e560..071dbb00f5 100644
--- a/erts/preloaded/ebin/prim_inet.beam
+++ b/erts/preloaded/ebin/prim_inet.beam
diff --git a/erts/preloaded/ebin/prim_zip.beam b/erts/preloaded/ebin/prim_zip.beam
index 55697cfe11..c4487cc5fa 100644
--- a/erts/preloaded/ebin/prim_zip.beam
+++ b/erts/preloaded/ebin/prim_zip.beam
diff --git a/erts/preloaded/ebin/zlib.beam b/erts/preloaded/ebin/zlib.beam
index d2f30fd9cd..d58261a57c 100644
--- a/erts/preloaded/ebin/zlib.beam
+++ b/erts/preloaded/ebin/zlib.beam
diff --git a/erts/preloaded/src/Makefile b/erts/preloaded/src/Makefile
index 5bcc2eb6e4..a224b6a5d4 100644
--- a/erts/preloaded/src/Makefile
+++ b/erts/preloaded/src/Makefile
@@ -40,7 +40,8 @@ PRE_LOADED_MODULES = \
 	zlib \
 	prim_zip \
 	otp_ring0 \
-	erlang
+	erlang \
+	erts_internal
 
 RELSYSDIR = $(RELEASE_PATH)/lib/erts-$(VSN)
 # not $(RELEASE_PATH)/erts-$(VSN)/preloaded
diff --git a/erts/preloaded/src/erlang.erl b/erts/preloaded/src/erlang.erl
index 9e814ae54b..51a190b095 100644
--- a/erts/preloaded/src/erlang.erl
+++ b/erts/preloaded/src/erlang.erl
@@ -1075,57 +1075,6 @@ phash2(_Term, _Range) ->
 pid_to_list(_Pid) ->
     erlang:nif_error(undefined).
 
-%% port_close/1
--spec port_close(Port) -> true when
-      Port :: port() | atom().
-port_close(_Port) ->
-    erlang:nif_error(undefined).
-
-%% port_command/2
--spec port_command(Port, Data) -> true when
-      Port :: port() | atom(),
-      Data :: iodata().
-port_command(_Port, _Data) ->
-    erlang:nif_error(undefined).
-
-%% port_command/3
--spec port_command(Port, Data, OptionList) -> boolean() when
-      Port :: port() | atom(),
-      Data :: iodata(),
-      OptionList :: [Option],
-      Option :: force | nosuspend.
-port_command(_Port, _Data, _OptionList) ->
-    erlang:nif_error(undefined).
-
-%% port_connect/2
--spec port_connect(Port, Pid) -> true when
-      Port :: port() | atom(),
-      Pid :: pid().
-port_connect(_Port, _Pid) ->
-    erlang:nif_error(undefined).
-
-%% port_control/3
--spec port_control(Port, Operation, Data) -> Res when
-      Port :: port() | atom(),
-      Operation :: integer(),
-      Data :: iodata(),
-      Res :: string() | binary().
-port_control(_Port, _Operation, _Data) ->
-    erlang:nif_error(undefined).
-
-%% port_get_data/1
--spec erlang:port_get_data(P1) -> term() when
-      P1 :: port() | atom().
-port_get_data(_P1) ->
-    erlang:nif_error(undefined).
-
-%% port_set_data/2
--spec erlang:port_set_data(P1, P2) -> true when
-      P1 :: port() | atom(),
-      P2 :: term().
-port_set_data(_P1, _P2) ->
-    erlang:nif_error(undefined).
-
 %% port_to_list/1
 -spec erlang:port_to_list(Port) -> string() when
       Port :: port().
@@ -1700,60 +1649,11 @@ nodes(_Arg) ->
            | out
            | binary
            | eof
+	   | {parallelism, Boolean :: boolean()}
 	   | hide.
 open_port(_PortName,_PortSettings) ->
     erlang:nif_error(undefined).
 
-%% Shadowed by erl_bif_types: erlang:port_call/2
--spec erlang:port_call(Port, Data) -> term() when
-      Port :: port() | atom(),
-      Data :: term().
-port_call(_Port, _Data) ->
-    erlang:nif_error(undefined).
-
-%% Shadowed by erl_bif_types: erlang:port_call/3
--spec erlang:port_call(Port, Operation, Data) -> term() when
-      Port :: port() | atom(),
-      Operation :: integer(),
-      Data :: term().
-port_call(_Port, _Operation, _Data) ->
-    erlang:nif_error(undefined).
-
--type port_info_item() ::
-      registered_name |
-      id |
-      connected |
-      links |
-      name |
-      input |
-      output |
-      os_pid.
-
--type port_info_result_item() ::
-      {registered_name, RegName :: atom()} |
-      {id, Index :: non_neg_integer()} |
-      {connected, Pid :: pid()} |
-      {links, Pids :: [pid()]} |
-      {name, String :: string()} |
-      {input, Bytes :: non_neg_integer()} |
-      {output, Bytes :: non_neg_integer()} |
-      {os_pid, OsPid :: non_neg_integer() | 'undefined'}.
-
-%% Shadowed by erl_bif_types: erlang:port_info/1
--spec erlang:port_info(Port) -> Result when
-      Port :: port() | atom(),
-      Result :: [port_info_result_item()] | undefined.
-port_info(_Result) ->
-    erlang:nif_error(undefined).
-
-%% Shadowed by erl_bif_types: erlang:port_info/2
--spec erlang:port_info(Port, Item) -> Result when
-      Port :: port() | atom(),
-      Item :: port_info_item(),
-      Result :: port_info_result_item() | undefined.
-port_info(_Result, _Item) ->
-    erlang:nif_error(undefined).
-
 -type priority_level() ::
       low | normal | high | max.
 
@@ -2531,6 +2431,216 @@ suspend_process(P) ->
     end.
 
 %%
+%% Port BIFs
+%%
+%%       Currently all port BIFs calls the corresponding
+%%       erts_internal:port_*() native function which perform
+%%       most of the actual work. These native functions should
+%%       *never* be called directly by other functionality. The
+%%       native functions may be changed, or removed without any
+%%       notice whatsoever!
+%%
+%% IMPORTANT NOTE:
+%%       When the erts_internal:port_*() native functions return
+%%       a reference, they have also internally prepared the
+%%       message queue of the caller for a receive that will
+%%       unconditionally wait for a message containing this
+%%       reference. If the erlang code calling these native
+%%       functions do not do this, subsequent receives will not
+%%       work as expected! That is, it is of *vital importance*
+%%       that the receive is performed as described above!
+%%
+
+-spec port_command(Port, Data) -> 'true' when
+      Port :: port() | atom(),
+      Data :: iodata().
+
+port_command(Port, Data) ->
+    case case erts_internal:port_command(Port, Data, []) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	true -> true;
+	Error -> erlang:error(Error, [Port, Data])
+    end.
+
+-spec port_command(Port, Data, OptionList) -> boolean() when
+      Port :: port() | atom(),
+      Data :: iodata(),
+      Option :: force | nosuspend,
+      OptionList :: [Option].
+
+port_command(Port, Data, Flags) ->
+    case case erts_internal:port_command(Port, Data, Flags) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	Bool when Bool == true; Bool == false -> Bool;
+	Error -> erlang:error(Error, [Port, Data, Flags])
+    end.
+
+-spec port_connect(Port, Pid) -> 'true' when
+      Port :: port() | atom(),
+      Pid :: pid().
+
+port_connect(Port, Pid) ->
+    case case erts_internal:port_connect(Port, Pid) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	true -> true;
+	Error -> erlang:error(Error, [Port, Pid])
+    end.
+
+-spec port_close(Port) -> 'true' when
+      Port :: port() | atom().
+
+port_close(Port) ->
+    case case erts_internal:port_close(Port) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	true -> true;
+	Error -> erlang:error(Error, [Port])
+    end.
+
+-spec port_control(Port, Operation, Data) -> iodata() | binary() when
+      Port :: port() | atom(),
+      Operation :: integer(),
+      Data :: iodata().
+
+port_control(Port, Operation, Data) ->
+    case case erts_internal:port_control(Port, Operation, Data) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	badarg -> erlang:error(badarg, [Port, Operation, Data]);
+	Result -> Result
+    end.
+
+-spec erlang:port_call(Port, Data) -> term() when
+      Port :: port() | atom(),
+      Data :: term().
+
+port_call(Port, Data) ->
+    case case erts_internal:port_call(Port, 0, Data) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	{ok, Result} -> Result;
+	Error -> erlang:error(Error, [Port, Data])
+    end.
+
+-spec erlang:port_call(Port, Operation, Data) -> term() when
+      Port :: port() | atom(),
+      Operation :: integer(),
+      Data :: term().
+
+port_call(Port, Operation, Data) ->
+    case case erts_internal:port_call(Port, Operation, Data) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	{ok, Result} -> Result;
+	Error -> erlang:error(Error, [Port, Operation, Data])
+    end.
+
+-spec erlang:port_info(Port) -> Result when
+      Port :: port() | atom(),
+      ResultItem :: {registered_name, RegisteredName :: atom()}
+		  | {id, Index :: non_neg_integer()}
+		  | {connected, Pid :: pid()}
+		  | {links, Pids :: [pid()]}
+		  | {name, String :: string()}
+		  | {input, Bytes :: non_neg_integer()}
+		  | {output, Bytes :: non_neg_integer()}
+		  | {os_pid, OsPid :: non_neg_integer() | 'undefined'},
+      Result :: [ResultItem] | 'undefined'.
+
+port_info(Port) ->
+    case case erts_internal:port_info(Port) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	badarg -> erlang:error(badarg, [Port]);
+	Result -> Result
+    end.
+
+-spec erlang:port_info(Port, connected) -> {connected, Pid} | 'undefined' when
+      Port :: port() | atom(),
+      Pid :: pid();
+		      (Port, id) -> {id, Index} | 'undefined' when
+      Port :: port() | atom(),
+      Index :: non_neg_integer();
+		      (Port, input) -> {input, Bytes} | 'undefined' when
+      Port :: port() | atom(),
+      Bytes :: non_neg_integer();
+		      (Port, links) -> {links, Pids} | 'undefined' when
+      Port :: port() | atom(),
+      Pids :: [pid()];
+		      (Port, locking) -> {locking, Locking} | 'undefined' when
+      Port :: port() | atom(),
+      Locking :: 'false' | 'port_level' | 'driver_level';
+		      (Port, memory) -> {memory, Bytes} | 'undefined' when
+      Port :: port() | atom(),
+      Bytes :: non_neg_integer();
+		      (Port, monitors) -> {monitors, Monitors} | 'undefined' when
+      Port :: port() | atom(),
+      Monitors :: [{process, pid()}];
+		      (Port, name) -> {name, Name} | 'undefined' when
+      Port :: port() | atom(),
+      Name :: string();
+		      (Port, os_pid) -> {os_pid, OsPid} | 'undefined' when
+      Port :: port() | atom(),
+      OsPid :: non_neg_integer() | 'undefined';
+		      (Port, output) -> {output, Bytes} | 'undefined' when
+      Port :: port() | atom(),
+      Bytes :: non_neg_integer();
+		      (Port, parallelism) -> {parallelism, Boolean} | 'undefined' when
+      Port :: port() | atom(),
+      Boolean :: boolean();
+		      (Port, queue_size) -> {queue_size, Bytes} | 'undefined' when
+      Port :: port() | atom(),
+      Bytes :: non_neg_integer();
+		      (Port, registered_name) -> {registered_name, RegisteredName} | [] | 'undefined' when
+      Port :: port() | atom(),
+      RegisteredName :: atom().
+
+port_info(Port, Item) ->
+    case case erts_internal:port_info(Port, Item) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	badarg -> erlang:error(badarg, [Port, Item]);
+	Result -> Result
+    end.
+
+-spec erlang:port_set_data(Port, Data) -> 'true' when
+      Port :: port() | atom(),
+      Data :: term().
+    
+port_set_data(Port, Data) ->
+    case case erts_internal:port_set_data(Port, Data) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	badarg -> erlang:error(badarg, [Port, Data]);
+	Result -> Result
+    end.
+
+-spec erlang:port_get_data(Port) -> term() when
+      Port :: port() | atom().
+
+port_get_data(Port) ->
+    case case erts_internal:port_get_data(Port) of
+	     Ref when erlang:is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	{ok, Data} -> Data;
+	Error -> erlang:error(Error, [Port])
+    end.
+
+%%
 %% If the emulator wants to perform a distributed command and
 %% a connection is not established to the actual node the following 
 %% functions are called in order to set up the connection and then
diff --git a/erts/preloaded/src/erts_internal.erl b/erts/preloaded/src/erts_internal.erl
new file mode 100644
index 0000000000..f1c83f4518
--- /dev/null
+++ b/erts/preloaded/src/erts_internal.erl
@@ -0,0 +1,155 @@
+%%
+%% %CopyrightBegin%
+%%
+%% Copyright Ericsson AB 2012. All Rights Reserved.
+%%
+%% The contents of this file are subject to the Erlang Public License,
+%% Version 1.1, (the "License"); you may not use this file except in
+%% compliance with the License. You should have received a copy of the
+%% Erlang Public License along with this software. If not, it can be
+%% retrieved online at http://www.erlang.org/.
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and limitations
+%% under the License.
+%%
+%% %CopyrightEnd%
+%%
+
+%%
+%% As the module name imply, this module is here for ERTS internal
+%% functionality. As an application programmer you should *never*
+%% call anything in this module directly. Functions exported by
+%% this module may change behaviour or be removed at any time
+%% without any notice whatsoever. Everything in this module is
+%% intentionally left undocumented, and should remain so.
+%%
+
+-module(erts_internal).
+
+-export([await_port_send_result/3]).
+
+-export([port_command/3, port_connect/2, port_close/1,
+	 port_control/3, port_call/3, port_set_data/2, port_get_data/1,
+	 port_info/1, port_info/2]).
+
+%%
+%% Await result of send to port
+%%
+
+await_port_send_result(Ref, Busy, Ok) ->
+    receive
+	{Ref, false} -> Busy;
+	{Ref, _} -> Ok
+    end.
+
+%%
+%% Statically linked port NIFs
+%%
+
+-spec erts_internal:port_command(Port, Data, OptionList) -> Result when
+      Port :: port() | atom(),
+      Data :: iodata(),
+      OptionList :: [Option],
+      Option :: force | nosuspend,
+      Result :: boolean() | reference() | badarg | notsup.
+port_command(_Port, _Data, _OptionList) ->
+    erlang:nif_error(undefined).
+
+-spec erts_internal:port_connect(Port, Pid) -> Result when
+      Port :: port() | atom(),
+      Pid :: pid(),
+      Result :: true | reference() | badarg.
+port_connect(_Port, _Pid) ->
+    erlang:nif_error(undefined).
+
+-spec erts_internal:port_close(Port) -> Result when
+      Port :: port() | atom(),
+      Result :: true | reference() | badarg.
+port_close(_Port) ->
+    erlang:nif_error(undefined).
+
+-spec erts_internal:port_control(Port, Operation, Data) -> Result when
+      Port :: port() | atom(),
+      Operation :: integer(),
+      Data :: iodata(),
+      Result :: string() | binary() | reference() | badarg.
+port_control(_Port, _Operation, _Data) ->
+    erlang:nif_error(undefined).
+
+-spec erts_internal:port_call(Port, Operation, Data) -> Result when
+      Port :: port() | atom(),
+      Operation :: integer(),
+      Data :: term(),
+      Result :: {ok, term()} | reference() | badarg.
+port_call(_Port, _Operation, _Data) ->
+    erlang:nif_error(undefined).
+
+-spec erts_internal:port_get_data(P1) -> Result when
+      P1 :: port() | atom(),
+      Result :: {ok, term()} | reference() | badarg.
+port_get_data(_P1) ->
+    erlang:nif_error(undefined).
+
+-spec erts_internal:port_set_data(P1, P2) -> Result when
+      P1 :: port() | atom(),
+      P2 :: term(),
+      Result :: true | reference() | badarg.
+port_set_data(_P1, _P2) ->
+    erlang:nif_error(undefined).
+
+-type port_info_1_result_item() ::
+      {registered_name, RegName :: atom()} |
+      {id, Index :: non_neg_integer()} |
+      {connected, Pid :: pid()} |
+      {links, Pids :: [pid()]} |
+      {name, String :: string()} |
+      {input, Bytes :: non_neg_integer()} |
+      {output, Bytes :: non_neg_integer()} |
+      {os_pid, OsPid :: non_neg_integer() | 'undefined'}.
+
+-spec erts_internal:port_info(Port) -> Result when
+      Port :: port() | atom(),
+      Result :: [port_info_1_result_item()] | undefined | reference() | badarg | [].
+port_info(_Result) ->
+    erlang:nif_error(undefined).
+
+-type port_info_2_item() ::
+      registered_name |
+      id |
+      connected |
+      links |
+      name |
+      input |
+      output |
+      os_pid |
+      monitors |
+      memory |
+      parallelism |
+      queue_size |
+      locking.
+
+-type port_info_2_result_item() ::
+      {registered_name, RegName :: atom()} |
+      [] | % No registered name
+      {id, Index :: non_neg_integer()} |
+      {connected, Pid :: pid()} |
+      {links, Pids :: [pid()]} |
+      {name, String :: string()} |
+      {input, Bytes :: non_neg_integer()} |
+      {output, Bytes :: non_neg_integer()} |
+      {os_pid, OsPid :: non_neg_integer() | 'undefined'} |
+      {monitors, Monitors :: [{process, pid()}]} |
+      {memory, MemSz :: non_neg_integer()} |
+      {parallelism, Boolean :: boolean()} |
+      {queue_size, QSz :: non_neg_integer()} |
+      {locking, Locking :: 'false' | 'port_level' | 'driver_level'}.
+
+-spec erts_internal:port_info(Port, Item) -> Result when
+      Port :: port() | atom(),
+      Item :: port_info_2_item(),
+      Result :: port_info_2_result_item() | undefined | reference() | badarg.
+
+port_info(_Result, _Item) ->
+    erlang:nif_error(undefined).
diff --git a/erts/preloaded/src/prim_inet.erl b/erts/preloaded/src/prim_inet.erl
index b9800a05da..21d23159f0 100644
--- a/erts/preloaded/src/prim_inet.erl
+++ b/erts/preloaded/src/prim_inet.erl
@@ -1072,6 +1072,8 @@ enc_opt(deliver)         -> ?INET_LOPT_DELIVER;
 enc_opt(exit_on_close)   -> ?INET_LOPT_EXITONCLOSE;
 enc_opt(high_watermark)  -> ?INET_LOPT_TCP_HIWTRMRK;
 enc_opt(low_watermark)   -> ?INET_LOPT_TCP_LOWTRMRK;
+enc_opt(high_msgq_watermark)  -> ?INET_LOPT_TCP_MSGQ_HIWTRMRK;
+enc_opt(low_msgq_watermark)   -> ?INET_LOPT_TCP_MSGQ_LOWTRMRK;
 enc_opt(send_timeout)    -> ?INET_LOPT_TCP_SEND_TIMEOUT;
 enc_opt(send_timeout_close) -> ?INET_LOPT_TCP_SEND_TIMEOUT_CLOSE;
 enc_opt(delay_send)      -> ?INET_LOPT_TCP_DELAY_SEND;
@@ -1126,6 +1128,8 @@ dec_opt(?INET_LOPT_DELIVER)       -> deliver;
 dec_opt(?INET_LOPT_EXITONCLOSE)   -> exit_on_close;
 dec_opt(?INET_LOPT_TCP_HIWTRMRK)  -> high_watermark;
 dec_opt(?INET_LOPT_TCP_LOWTRMRK)  -> low_watermark;
+dec_opt(?INET_LOPT_TCP_MSGQ_HIWTRMRK)  -> high_msgq_watermark;
+dec_opt(?INET_LOPT_TCP_MSGQ_LOWTRMRK)  -> low_msgq_watermark;
 dec_opt(?INET_LOPT_TCP_SEND_TIMEOUT) -> send_timeout;
 dec_opt(?INET_LOPT_TCP_SEND_TIMEOUT_CLOSE) -> send_timeout_close;
 dec_opt(?INET_LOPT_TCP_DELAY_SEND)   -> delay_send;
@@ -1221,6 +1225,8 @@ type_opt_1(deliver) ->
 type_opt_1(exit_on_close)   -> bool;
 type_opt_1(low_watermark)   -> int;
 type_opt_1(high_watermark)  -> int;
+type_opt_1(low_msgq_watermark)   -> int;
+type_opt_1(high_msgq_watermark)  -> int;
 type_opt_1(send_timeout)    -> time;
 type_opt_1(send_timeout_close) -> bool;
 type_opt_1(delay_send)      -> bool;