Add testcases for strange port scheduling scenarios

3 files changed, 495 insertions, 2 deletions

diff --git a/erts/emulator/test/busy_port_SUITE.erl b/erts/emulator/test/busy_port_SUITE.erl
index 32e907ca69..a92afef003 100644
--- a/erts/emulator/test/busy_port_SUITE.erl
+++ b/erts/emulator/test/busy_port_SUITE.erl
@@ -26,6 +26,8 @@
 	 no_trap_exit_unlinked/1, trap_exit/1, multiple_writers/1,
 	 hard_busy_driver/1, soft_busy_driver/1]).
 
+-compile(export_all).
+
 -include_lib("test_server/include/test_server.hrl").
 
 %% Internal exports.
@@ -36,7 +38,9 @@ suite() -> [{ct_hooks,[ts_install_cth]}].
 all() -> 
     [io_to_busy, message_order, send_3, system_monitor,
      no_trap_exit, no_trap_exit_unlinked, trap_exit,
-     multiple_writers, hard_busy_driver, soft_busy_driver].
+     multiple_writers, hard_busy_driver, soft_busy_driver,
+     scheduling_delay_busy,scheduling_delay_busy_nosuspend,
+     scheduling_busy_link].
 
 groups() -> 
     [].
@@ -528,6 +532,304 @@ hs_busy_pcmd(Prt, Opts, StartFun, EndFun) ->
 	    EndFun(P, Res, Time)
     end.
 
+scheduling_delay_busy(Config) ->
+    
+    Scenario = 
+	[{1,{spawn,[{var,drvname},undefined]}},
+	 {2,{call,[{var,1},open_port]}},
+	 {3,{spawn,[{var,2},{var,1}]}},
+	 {0,{ack,[{var,1},{busy,1,250}]}},
+	 {0,{cast,[{var,3},{command,2}]}},
+	 [{0,{cast,[{var,3},{command,I}]}} 
+	  || I <- lists:seq(3,50)],
+	 {0,{cast,[{var,3},take_control]}},
+	 {0,{cast,[{var,1},{new_owner,{var,3}}]}},
+	 {0,{cast,[{var,3},close]}},
+	 {0,{timer,sleep,[300]}},
+	 {0,{erlang,port_command,[{var,2},<<$N>>,[force]]}},
+	 [{0,{cast,[{var,1},{command,I}]}} 
+	  || I <- lists:seq(101,127)]
+	 ,{10,{call,[{var,3},get_data]}}
+	 ],
+
+    Validation = [{seq,10,lists:seq(1,50)}],
+
+    port_scheduling(Scenario,Validation,?config(data_dir,Config)).
+
+scheduling_delay_busy_nosuspend(Config) ->
+
+    Scenario = 
+	[{1,{spawn,[{var,drvname},undefined]}},
+	 {2,{call,[{var,1},open_port]}},
+	 {0,{cast,[{var,1},{command,1,100}]}},
+	 {0,{cast,[{var,1},{busy,2}]}},
+	 {10,{call,[{var,1},{command,3,[nosuspend]}]}},
+	 {0,{timer,sleep,[200]}},
+	 {0,{erlang,port_command,[{var,2},<<$N>>,[force]]}},
+	 {0,{cast,[{var,1},close]}},
+	 {20,{call,[{var,1},get_data]}}
+	 ],
+
+    Validation = [{eq,10,nosuspend},{seq,20,[1,2]}],
+
+    port_scheduling(Scenario,Validation,?config(data_dir,Config)).
+
+scheduling_busy_link(Config) ->
+    
+    Scenario = 
+	[{1,{spawn,[{var,drvname},undefined]}},
+	 {2,{call,[{var,1},open_port]}},
+	 {3,{spawn,[{var,2},{var,1}]}},
+	 {0,{cast,[{var,1},unlink]}},
+	 {0,{cast,[{var,1},{busy,1}]}},
+	 {0,{cast,[{var,1},{command,2}]}},
+	 {0,{cast,[{var,1},link]}},
+	 {0,{timer,sleep,[1000]}},
+	 {0,{ack,[{var,3},take_control]}},
+	 {0,{cast,[{var,1},{new_owner,{var,3}}]}},
+	 {0,{cast,[{var,3},close]}},
+	 {10,{call,[{var,3},get_data]}},
+	 {20,{call,[{var,1},get_exit]}}
+	 ],
+
+    Validation = [{seq,10,[1]},
+		  {seq,20,[{'EXIT',noproc}]}],
+
+    port_scheduling(Scenario,Validation,?config(data_dir,Config)).
+
+process_init(DrvName,Owner) ->
+    process_flag(trap_exit,true),
+    process_loop(DrvName,Owner, {[],[]}).
+
+process_loop(DrvName,undefined,Data) when is_list(DrvName) ->
+    process_loop(DrvName,[binary],Data);
+process_loop(DrvName,PortOpts,Data) when is_list(DrvName) ->
+    receive
+	{call,open_port,P} ->
+	    Port = open_port({spawn, DrvName}, PortOpts),
+	    send(P,Port),
+	    process_loop(Port,self(),Data)
+    end;
+process_loop(Port,undefined,Data) ->
+    receive
+	{cast,{new_owner,Pid}} ->
+	    pal("NewOwner: ~p",[Pid]),
+	    process_loop(Port,Pid,Data)
+    end;
+process_loop(Port,Owner,{Data,Exit} = DE) ->
+    receive
+	{Port,connected} ->
+	    pal("Connected",[]),
+	    process_loop(Port,undefined,DE);	
+	{Port,{data,NewData}} ->
+	    pal("Got: ~p",[NewData]),
+	    receive
+		{Port,closed} ->
+		    process_loop(Port,Owner,{Data ++ [NewData],Exit})
+	    after 2000 ->
+		    exit(did_not_get_port_close)
+	    end;
+	{'EXIT',Port,Reason} = Exit ->
+	    pal("Exit: ~p",[Exit]),
+	    process_loop(Port,Owner,{Data, Exit ++ [[{'EXIT',Reason}]]});
+	{'EXIT',_Port,_Reason} = Exit ->
+	    pal("Exit: ~p",[Exit]);
+	{call,Msg,P} ->
+	    case handle_msg(Msg,Port,Owner,DE) of
+		{Reply,NewOwner,NewData} ->
+		    send(P,Reply),
+		    process_loop(Port,NewOwner,NewData);
+		Reply ->
+		    send(P,Reply),
+		    process_loop(Port,Owner,DE)
+	    end;
+	{ack,Msg,P} ->
+	    send(P,ok),
+	    case handle_msg(Msg,Port,Owner,DE) of
+		{_Reply,NewOwner,NewData} ->
+		    process_loop(Port,NewOwner,NewData);
+		_Reply ->
+		    process_loop(Port,Owner,DE)
+	    end;
+	{cast,Msg} when is_atom(Msg) orelse element(1,Msg) /= new_owner ->
+	    case handle_msg(Msg,Port,Owner,DE) of
+		{_Reply,NewOwner,NewData} ->
+		    process_loop(Port,NewOwner,NewData);
+		_ ->
+		    process_loop(Port,Owner,DE)
+	    end
+    end.
+
+handle_msg({busy,Value,Delay},Port,Owner,_Data) ->
+    pal("Long busy: ~p",[Value]),
+    send(Port,{Owner,{command,<<$L,Value:32,(round(Delay/100))>>}});
+handle_msg({busy,Value},Port,Owner,_Data)  ->
+    pal("Busy: ~p",[Value]),
+    send(Port,{Owner,{command,<<$B,Value:32>>}});
+handle_msg({command,Value},Port,Owner,_Data)  ->
+    pal("Short: ~p",[Value]),
+    send(Port,{Owner,{command,<<$C,Value:32>>}});
+handle_msg({command,Value,Delay},Port,Owner,_Data) when is_integer(Delay) ->
+    pal("Long: ~p",[Value]),
+    send(Port,{Owner,{command,<<$D,Value:32,(round(Delay/100))>>}});
+handle_msg({command,Value,Opts},Port,Owner,_Data)  ->
+    pal("Short Opt: ~p",[Value]),
+    send(Port,{Owner,{command,<<$C,Value:32>>}},Opts);
+handle_msg({command,Value,Opts,Delay},Port,Owner,_Data)  ->
+    pal("Long Opt: ~p",[Value]),
+    send(Port,{Owner,{command,<<$D,Value:32,(round(Delay/100))>>}},Opts);
+handle_msg(take_control,Port,Owner,Data)  ->
+    pal("Connect: ~p",[self()]),
+    send(Port,{Owner, {connect, self()}}),
+    {undefined,self(),Data};
+handle_msg(unlink,Port,_Owner,_Data) ->
+    pal("Unlink:",[]),
+    erlang:unlink(Port);
+handle_msg(link,Port,_Owner,_Data) ->
+    pal("Link:",[]),
+    erlang:link(Port);
+handle_msg(close,Port,Owner,_Data)  ->
+    pal("Close",[]),
+    send(Port,{Owner,close});
+handle_msg(get_data,Port,_Owner,{[],_Exit})  ->
+    %% Wait for data if it has not arrived yet
+    receive
+	{Port,{data,Data}} ->
+	    Data
+    after 2000 ->
+	    pal("~p",[erlang:process_info(self())]),
+	    exit(did_not_get_port_data)
+    end;
+handle_msg(get_data,_Port,Owner,{Data,Exit})  ->
+    pal("GetData",[]),
+    {hd(Data),Owner,{tl(Data),Exit}};
+handle_msg(get_exit,Port,_Owner,{_Data,[]})  ->
+    %% Wait for exit if it has not arrived yet
+    receive
+	{'EXIT',Port,Reason} ->
+	    [{'EXIT',Reason}]
+    after 2000 ->
+	    pal("~p",[erlang:process_info(self())]),
+	    exit(did_not_get_port_exit)
+    end;
+handle_msg(get_exit,_Port,Owner,{Data,Exit}) ->
+    {hd(Exit),Owner,{Data,tl(Exit)}}.
+
+    
+
+call(Pid,Msg) ->
+    pal("call(~p,~p)",[Pid,Msg]),
+    send(Pid,{call,Msg,self()}),
+    receive
+	Ret ->
+	    Ret
+    end.
+ack(Pid,Msg) ->
+    pal("ack(~p,~p)",[Pid,Msg]),
+    send(Pid,{ack,Msg,self()}),
+    receive
+	Ret ->
+	    Ret
+    end.
+
+cast(Pid,Msg) ->
+    pal("cast(~p,~p)",[Pid,Msg]),
+    send(Pid,{cast,Msg}).
+
+send(Pid,Msg) ->
+    erlang:send(Pid,Msg).
+send(Prt,Msg,Opts) ->
+    erlang:send(Prt,Msg,Opts).
+
+
+port_scheduling(Scenario,Validation,Path) ->
+    DrvName = "scheduling_drv",
+    erl_ddll:start(),
+    case erl_ddll:load_driver(Path, DrvName) of
+	ok -> ok;
+	{error, Error} ->
+	    io:format("~s\n", [erl_ddll:format_error(Error)]),
+	    ?line ?t:fail()
+    end,
+
+    Data = run_scenario(lists:flatten(Scenario),[{drvname,DrvName}]),
+    ok = validate_scenario(Data,Validation).
+
+
+run_scenario([{V,{Module,Cmd,Args}}|T],Vars) ->
+    Res = run_command(Module,Cmd,
+		      replace_args(Args,Vars)),
+    run_scenario(T,[{V,Res}|Vars]);
+run_scenario([{V,{Cmd,Args}}|T],Vars) ->
+    run_scenario([{V,{?MODULE,Cmd,Args}}|T],Vars);
+run_scenario([],Vars) ->
+    Vars.
+
+run_command(_M,spawn,{Args,Opts}) ->
+    Pid = spawn_opt(fun() -> apply(?MODULE,process_init,Args) end,[link|Opts]),
+    pal("spawn(~p): ~p",[Args,Pid]),
+    Pid;
+run_command(M,spawn,Args) ->
+    run_command(M,spawn,{Args,[]});
+run_command(Mod,Func,Args) ->
+    erlang:display({{Mod,Func,Args},now()}),
+    apply(Mod,Func,Args).
+
+validate_scenario(Data,[{print,Var}|T]) ->
+    pal("Val: ~p",[proplists:get_value(Var,Data)]),
+    validate_scenario(Data,T);
+validate_scenario(Data,[{eq,Var,Value}|T]) ->
+    case proplists:get_value(Var,Data) of
+	Value ->
+	    validate_scenario(Data,T);
+	Else ->
+	    exit({eq_return,Value,Else})
+    end;
+validate_scenario(Data,[{neq,Var,Value}|T]) ->
+    case proplists:get_value(Var,Data) of
+	Value ->
+	    exit({neq_return,Value});
+	_Else ->
+	    validate_scenario(Data,T)
+    end;
+validate_scenario(Data,[{seq,Var,Seq}|T]) ->
+    try
+	validate_sequence(proplists:get_value(Var,Data),Seq)
+    catch _:{validate_sequence,NotFound} ->
+	    exit({validate_sequence,NotFound,Data})
+    end,
+    validate_scenario(Data,T);
+validate_scenario(_,[]) ->
+    ok.
+
+validate_sequence(Data,Validation) when is_binary(Data) ->
+    validate_sequence(binary_to_list(Data),Validation);	    
+validate_sequence([H|R],[H|T]) ->
+    validate_sequence(R,T);
+validate_sequence([_|R],Seq) ->
+    validate_sequence(R,Seq);
+validate_sequence(_,[]) ->
+    ok;
+validate_sequence([],NotFound) ->
+    exit({validate_sequence,NotFound}).
+
+replace_args({var,Var},Vars) ->
+    proplists:get_value(Var,Vars);
+replace_args([H|T],Vars) ->
+    [replace_args(H,Vars)|replace_args(T,Vars)];
+replace_args([],_Vars) ->
+    [];
+replace_args(Tuple,Vars) when is_tuple(Tuple) ->
+    list_to_tuple(replace_args(tuple_to_list(Tuple),Vars));
+replace_args(Else,_Vars) ->
+    Else.
+
+pal(_F,_A) -> ok.
+%pal(Format,Args) ->
+%    ct:pal("~p "++Format,[self()|Args]).
+%    erlang:display(lists:flatten(io_lib:format("~p "++Format,[self()|Args]))).
+			
+
 %%% Utilities.
 
 chk_range(Min, Val, Max) when Min =< Val, Val =< Max ->
diff --git a/erts/emulator/test/busy_port_SUITE_data/Makefile.src b/erts/emulator/test/busy_port_SUITE_data/Makefile.src
index 664909db71..b5fcf25176 100644
--- a/erts/emulator/test/busy_port_SUITE_data/Makefile.src
+++ b/erts/emulator/test/busy_port_SUITE_data/Makefile.src
@@ -17,9 +17,10 @@
 # %CopyrightEnd%
 #
 
-all: busy_drv@dll@ hard_busy_drv@dll@ soft_busy_drv@dll@
+all: busy_drv@dll@ hard_busy_drv@dll@ soft_busy_drv@dll@ scheduling_drv@dll@
 
 @SHLIB_RULES@
 
 hard_busy_drv@obj@: hard_busy_drv.c hs_busy_drv.c
 soft_busy_drv@obj@: soft_busy_drv.c hs_busy_drv.c
+scheduling_drv@obj@: scheduling_drv.c
diff --git a/erts/emulator/test/busy_port_SUITE_data/scheduling_drv.c b/erts/emulator/test/busy_port_SUITE_data/scheduling_drv.c
new file mode 100644
index 0000000000..57be9b6392
--- /dev/null
+++ b/erts/emulator/test/busy_port_SUITE_data/scheduling_drv.c
@@ -0,0 +1,190 @@
+/*
+ * %CopyrightBegin%
+ * 
+ * Copyright Ericsson AB 2009-2011. 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%
+ */
+
+#ifdef __WIN32__
+#include <windows.h>
+#else
+#include <sys/select.h>
+#endif
+#include <errno.h>
+#include <stdio.h>
+#include "erl_driver.h"
+
+#define get_int32(s) ((((unsigned char*) (s))[0] << 24) | \
+                      (((unsigned char*) (s))[1] << 16) | \
+                      (((unsigned char*) (s))[2] << 8)  | \
+                      (((unsigned char*) (s))[3]))
+
+#define ERTS_TEST_SCHEDULING_DRV_NAME "scheduling_drv"
+#define ERTS_TEST_SCHEDULING_DRV_FLAGS  \
+  ERL_DRV_FLAG_USE_PORT_LOCKING | ERL_DRV_FLAG_SOFT_BUSY
+
+ErlDrvData start(ErlDrvPort port, char *command);
+void output(ErlDrvData drv_data, char *buf, ErlDrvSizeT len);
+ErlDrvSSizeT control(ErlDrvData drv_data, unsigned int command, char *buf,
+		     ErlDrvSizeT len, char **rbuf, ErlDrvSizeT rlen);
+void stop(ErlDrvData drv_data);
+void timeout(ErlDrvData drv_data);
+
+static void delay(unsigned ms);
+
+static ErlDrvEntry busy_drv_entry = { 
+    NULL /* init */,
+    start,
+    stop,
+    output,
+    NULL /* ready_input */,
+    NULL /* ready_output */,
+    ERTS_TEST_SCHEDULING_DRV_NAME,
+    NULL /* finish */,
+    NULL /* handle */,
+    control,
+    timeout,
+    NULL /* outputv */,
+    NULL /* ready_async */,
+    NULL /* flush */,
+    NULL /* call */,
+    NULL /* event */,
+    ERL_DRV_EXTENDED_MARKER,
+    ERL_DRV_EXTENDED_MAJOR_VERSION,
+    ERL_DRV_EXTENDED_MINOR_VERSION,
+    ERTS_TEST_SCHEDULING_DRV_FLAGS,
+    NULL /* handle2 */,
+    NULL /* handle_monitor */,
+    NULL /* stop_select */
+};
+
+#define DBG(data,FMT)
+/* #define DBG(data,FMT) printf("0x%.8lx: %s",driver_caller(data->port),FMT); */
+
+typedef struct SchedDrvData {
+  ErlDrvPort port;
+  char data[255];
+  int curr;
+  int use_auto_busy;
+} SchedDrvData;
+
+DRIVER_INIT(busy_drv)
+{
+    return &busy_drv_entry;
+}
+
+ErlDrvData start(ErlDrvPort port, char *command)
+{
+  SchedDrvData *d = driver_alloc(sizeof(SchedDrvData));
+  d->port = port;
+  d->curr = 0;
+  d->use_auto_busy = 0;
+  DBG(d,"start\r\n");
+  return (ErlDrvData) d;
+}
+
+void stop(ErlDrvData drv_data) {
+  SchedDrvData *d = (SchedDrvData*)drv_data;
+  driver_output(d->port,d->data,d->curr);
+  DBG(d,"close\r\n");
+  driver_free(d);
+  return;
+}
+
+void timeout(ErlDrvData drv_data) {
+  SchedDrvData *d = (SchedDrvData*)drv_data;
+  set_busy_port(d->port, 0);
+  DBG(d,"timeout\r\n");
+}
+
+void output(ErlDrvData drv_data, char *buf, ErlDrvSizeT len)
+{
+  int res;
+  unsigned int command = *buf;
+  SchedDrvData *d = (SchedDrvData*)drv_data;
+  
+  switch (command) {
+  case 'B': /* busy */
+    DBG(d,"busy: ");
+    set_busy_port(d->port, 1);
+    break;
+  case 'L': /* busy long call */
+    DBG(d,"long: ");
+    delay(buf[5]*100);
+    set_busy_port(d->port, 1);
+    break;
+  case 'D': /* delay call */
+    DBG(d,"delay: ");
+    delay(buf[5]*100);
+    break;
+  case 'N': /* not busy */
+    DBG(d,"not");
+    set_busy_port(d->port, 0);
+    goto done;
+  case 'C': /* change state */
+    DBG(d,"chang: ");
+    break;
+  case 'G': /* get state */
+    DBG(d,"get : ");
+    driver_output(d->port,d->data,d->curr);
+    return;
+  default:
+    driver_failure_posix((ErlDrvPort) drv_data, EINVAL);
+    break;
+  }
+  if (len > 1) {
+    unsigned int val = get_int32(buf+1);
+    fprintf(stderr,"%u",val);
+    d->data[d->curr++] = val;
+  }
+ done:
+ fprintf(stderr,"\r\n");
+}
+
+ErlDrvSSizeT control(ErlDrvData drv_data, unsigned int command, char *buf,
+		     ErlDrvSizeT len, char **rbuf, ErlDrvSizeT rlen)
+{
+    switch (command) {
+    case 'B': /* busy */
+	set_busy_port((ErlDrvPort) drv_data, 1);
+	break;
+    case 'N': /* not busy */
+	set_busy_port((ErlDrvPort) drv_data, 0);
+	break;
+    default:
+	driver_failure_posix((ErlDrvPort) drv_data, EINVAL);
+	break;
+    }
+    return 0;
+}
+
+
+/*
+ * Delays (sleeps) the given number of milli-seconds.
+ */
+
+static void delay(unsigned ms)
+{
+  fprintf(stderr,"delay(%u)",ms);
+#ifdef __WIN32__
+  Sleep(ms);
+#else
+  struct timeval t;
+  t.tv_sec = ms/1000;
+  t.tv_usec = (ms % 1000) * 1000;
+
+  select(0, NULL, NULL, NULL, &t);
+#endif
+}