Implement true asynchronous signaling between processes and ports

3 files changed, 215 insertions, 1 deletions

diff --git a/erts/preloaded/src/Makefile b/erts/preloaded/src/Makefile
index 145638802f..5502ad588b 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 1cbce5b80b..7d0c93c8d3 100644
--- a/erts/preloaded/src/erlang.erl
+++ b/erts/preloaded/src/erlang.erl
@@ -46,6 +46,10 @@
 -export([gather_sched_wall_time_result/1,
 	 await_sched_wall_time_modifications/2]).
 
+-export([port_command/2, port_command/3, port_connect/2, port_close/1,
+	 port_control/3, port_call/2, port_call/3, port_info/1, port_info/2,
+	 port_set_data/2, port_get_data/1]).
+
 -deprecated([hash/2]).
 
 % Get rid of autoimports of spawn to avoid clashes with ourselves.
@@ -406,6 +410,174 @@ 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 erlang: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 is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	true -> true;
+	Error -> erlang:error(Error, [Port, Data])
+    end.
+
+-spec erlang:port_command(Port, Data, [Flag]) -> boolean() when
+      Port :: port() | atom(),
+      Data :: iodata(),
+      Flag :: force | nosuspend.
+
+port_command(Port, Data, Flags) ->
+    case case erts_internal:port_command(Port, Data, Flags) of
+	     Ref when 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 erlang: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 is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	true -> true;
+	Error -> erlang:error(Error, [Port, Pid])
+    end.
+
+
+-spec erlang:port_close(Port) -> 'true' when
+      Port :: port() | atom().
+
+port_close(Port) ->
+    case case erts_internal:port_close(Port) of
+	     Ref when is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	true -> true;
+	Error -> erlang:error(Error, [Port])
+    end.
+
+-spec erlang:port_control(Port, Command, Data) -> iodata() | binary() when
+      Port :: port() | atom(),
+      Command :: integer(),
+      Data :: iodata().
+
+port_control(Port, Command, Data) ->
+    case case erts_internal:port_control(Port, Command, Data) of
+	     Ref when is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	badarg -> erlang:error(badarg, [Port, Command, 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 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, Command, Data) -> term() when
+      Port :: port() | atom(),
+      Command :: integer(),
+      Data :: term().
+
+port_call(Port, Command, Data) ->
+    case case erts_internal:port_call(Port, Command, Data) of
+	     Ref when is_reference(Ref) -> receive {Ref, Res} -> Res end;
+	     Res -> Res
+	 end of
+	{ok, Result} -> Result;
+	Error -> erlang:error(Error, [Port, Command, Data])
+    end.
+
+-spec erlang:port_info(Port) -> [{Item, Info}] when
+      Port :: port() | atom(),
+      Item :: 'registered_name' | 'id' | 'connected' | 'links' | 'name' | 'input' | 'output',
+      Info :: term().
+
+port_info(Port) ->
+    case case erts_internal:port_info(Port) of
+	     Ref when 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, Item) -> [{Item, Info}] when
+      Port :: port() | atom(),
+      Item :: 'registered_name' | 'id' | 'connected' | 'links' | 'monitors' | 'name' | 'input' | 'output' | 'memory' | 'queue_size' | 'locking',
+      Info :: term().
+
+port_info(Port, Item) ->
+    case case erts_internal:port_info(Port, Item) of
+	     Ref when 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 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 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..8bd7e95f03
--- /dev/null
+++ b/erts/preloaded/src/erts_internal.erl
@@ -0,0 +1,41 @@
+%%
+%% %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]).
+
+%%
+%% Await result of send to port
+%%
+
+await_port_send_result(Ref, Busy, Ok) ->
+    receive
+	{Ref, false} -> Busy;
+	{Ref, _} -> Ok
+    end.