nif_SUITE send3 stress test

2 files changed, 245 insertions, 37 deletions

diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl
index 888bf582d8..f45cfa3e4a 100644
--- a/erts/emulator/test/nif_SUITE.erl
+++ b/erts/emulator/test/nif_SUITE.erl
@@ -30,7 +30,7 @@
 	 fin_per_testcase/2, basic/1, reload/1, upgrade/1, heap_frag/1,
 	 types/1, many_args/1, binaries/1, get_string/1, get_atom/1, api_macros/1,
 	 from_array/1, iolist_as_binary/1, resource/1, resource_binary/1, resource_takeover/1,
-	 threading/1, send/1, send2/1, send_threaded/1, neg/1, is_checks/1,
+	 threading/1, send/1, send2/1, send3/1, send_threaded/1, neg/1, is_checks/1,
 	 get_length/1, make_atom/1, make_string/1]).
 
 -export([many_args_100/100]).
@@ -51,7 +51,7 @@
 all(suite) ->
     [basic, reload, upgrade, heap_frag, types, many_args, binaries, get_string,
      get_atom, api_macros, from_array, iolist_as_binary, resource, resource_binary,
-     resource_takeover, threading, send, send2, send_threaded, neg, is_checks,
+     resource_takeover, threading, send, send2, send3, send_threaded, neg, is_checks,
      get_length, make_atom, make_string].
 
 %%init_per_testcase(_Case, Config) ->
@@ -916,7 +916,164 @@ forwarder(To, N) ->
 
 other_term() ->
     {fun(X,Y) -> X*Y end, make_ref()}.
-    
+
+send3(doc) -> ["Message sending stress test"];
+send3(Config) when is_list(Config) ->
+    %% Let a number of processes send random message blobs between each other
+    %% using enif_send. Kill and spawn new ones randomly to keep a ~constant
+    %% number of workers running.
+    Seed = now(),
+    io:format("seed: ~p\n",[Seed]), 
+    random:seed(Seed),    
+    ets:new(nif_SUITE,[named_table,public]),
+    ?line true = ets:insert(nif_SUITE,{send3,0,0,0,0}),
+    timer:send_after(10000, timeout), % Run for 10 seconds
+    SpawnCnt = send3_controller(0, [], [], 20),
+    ?line [{_,Rcv,SndOk,SndFail,Balance}] = ets:lookup(nif_SUITE,send3),
+    io:format("spawns=~p received=~p, sent=~p send-failure=~p balance=~p\n",
+              [SpawnCnt,Rcv,SndOk,SndFail,Balance]),
+    ets:delete(nif_SUITE).
+
+send3_controller(SpawnCnt, [], _, infinity) ->
+    SpawnCnt;
+send3_controller(SpawnCnt0, Mons0, Pids0, Tick) ->
+    receive 
+        timeout ->
+            io:format("Timeout. Sending 'halt' to ~p\n",[Pids0]),            
+            lists:foreach(fun(P) -> P ! {halt,self()} end, Pids0),
+            lists:foreach(fun(P) -> receive {halted,P} -> ok end end, Pids0),
+            QTot = lists:foldl(fun(P,QSum) ->
+                                 {message_queue_len,QLen} = 
+                                    erlang:process_info(P,message_queue_len),
+                                 QSum + QLen
+                               end, 0, Pids0),
+            io:format("Total queue length ~p\n",[QTot]),            
+            lists:foreach(fun(P) -> P ! die end, Pids0),
+            send3_controller(SpawnCnt0, Mons0, [], infinity);
+        {'DOWN', MonRef, process, _Pid, _} ->
+            Mons1 = lists:delete(MonRef, Mons0),
+            %%io:format("Got DOWN from ~p. Monitors left: ~p\n",[Pid,Mons1]),            
+            send3_controller(SpawnCnt0, Mons1, Pids0, Tick)
+    after Tick -> 
+        Max = 20,
+        N = length(Pids0),
+        PidN = random:uniform(Max),
+        %%io:format("N=~p PidN=~p Pids0=~p\n", [N,PidN,Pids0]), 
+        case PidN > N of
+            true ->
+                {NewPid,Mon} = spawn_opt(fun send3_proc/0, [link,monitor]),                
+                lists:foreach(fun(P) -> P ! {is_born,NewPid} end, Pids0),
+                ?line Balance = ets:lookup_element(nif_SUITE,send3,5),
+                Inject = (Balance =< 0),
+                case Inject of
+                    true ->  ok;
+                    false -> ets:update_element(nif_SUITE,send3,{5,-1})
+                end,
+                NewPid ! {pids,Pids0,Inject},
+                send3_controller(SpawnCnt0+1, [Mon|Mons0], [NewPid|Pids0], Tick);
+            false ->
+                KillPid = lists:nth(PidN,Pids0),
+                KillPid ! die,
+                Pids1 = lists:delete(KillPid, Pids0),
+                lists:foreach(fun(P) -> P ! {is_dead,KillPid} end, Pids1),
+                send3_controller(SpawnCnt0, Mons0, Pids1, Tick)
+        end        
+   end.
+
+send3_proc() ->
+    %%io:format("Process ~p spawned\n",[self()]),
+    send3_proc([self()], {0,0,0}, {1,2,3,4,5}).
+send3_proc(Pids0, Counters={Rcv,SndOk,SndFail}, State0) ->
+    %%io:format("~p: Pids0=~p", [self(), Pids0]),
+    %%timer:sleep(10),
+    receive 
+        {pids, Pids1, Inject} ->
+            %%io:format("~p: got ~p Inject=~p\n", [self(), Pids1, Inject]), 
+            ?line Pids0 = [self()],
+            Pids2 = [self() | Pids1],
+            case Inject of
+                true -> send3_proc_send(Pids2, Counters, State0);
+                false -> send3_proc(Pids2, Counters, State0)
+            end;
+        {is_born, Pid} ->
+            %%io:format("~p: is_born ~p, got ~p\n", [self(), Pid, Pids0]), 
+            send3_proc([Pid | Pids0], Counters, State0);
+        {is_dead, Pid} ->            
+            Pids1 = lists:delete(Pid,Pids0),
+            %%io:format("~p: is_dead ~p, got ~p\n", [self(), Pid, Pids1]),
+            send3_proc(Pids1, Counters, State0);
+        {blob, Blob0} ->
+            %%io:format("~p: blob ~p\n", [self(), Blob0]),
+            State1 = send3_new_state(State0, Blob0),
+            send3_proc_send(Pids0, {Rcv+1,SndOk,SndFail}, State1);
+        die ->
+            %%io:format("Process ~p terminating, stats = ~p\n",[self(),Counters]),
+            {message_queue_len,Dropped} = erlang:process_info(self(),message_queue_len),
+            _R = ets:update_counter(nif_SUITE,send3,
+                               [{2,Rcv},{3,SndOk},{4,SndFail},{5,1-Dropped}]),
+            %%io:format("~p: dies R=~p\n", [self(), R]),
+            ok;
+        {halt,Papa} ->
+            Papa ! {halted,self()},
+            io:format("~p halted\n",[self()]),
+            receive die -> ok end,
+            io:format("~p dying\n",[self()])
+    end.
+
+send3_proc_send(Pids, {Rcv,SndOk,SndFail}, State0) ->
+    To = lists:nth(random:uniform(length(Pids)),Pids),
+    Blob = send3_make_blob(),
+    State1 = send3_new_state(State0,Blob), 
+    case send3_send(To, Blob) of
+        true ->
+            send3_proc(Pids, {Rcv,SndOk+1,SndFail}, State1);
+        false ->
+            send3_proc(Pids, {Rcv,SndOk,SndFail+1}, State1)
+    end.
+
+
+send3_make_blob() ->    
+    case random:uniform(20)-1 of
+        0 -> {term,[]};
+        N ->
+            MsgEnv = alloc_msgenv(), 
+            repeat(N bsr 1,
+                   fun(_) -> grow_blob(MsgEnv,other_term(),random:uniform(1 bsl 20))
+                   end, void),
+            case (N band 1) of
+                0 -> {term,copy_blob(MsgEnv)};
+                1 -> {msgenv,MsgEnv}
+            end
+    end.
+
+send3_send(Pid, Msg) ->
+    %% 90% enif_send and 10% normal bang
+    case random:uniform(10) of
+        1 -> send3_send_bang(Pid,Msg);
+        _ -> send3_send_nif(Pid,Msg)
+    end.
+send3_send_nif(Pid, {term,Blob}) ->
+    %%io:format("~p send term nif\n",[self()]),
+    send_term(Pid, {blob, Blob}) =:= 1;
+send3_send_nif(Pid, {msgenv,MsgEnv}) ->
+    %%io:format("~p send blob nif\n",[self()]),   
+    send3_blob(MsgEnv, Pid, blob) =:= 1.
+
+send3_send_bang(Pid, {term,Blob}) ->
+    %%io:format("~p send term bang\n",[self()]),   
+    Pid ! {blob, Blob},
+    true;
+send3_send_bang(Pid, {msgenv,MsgEnv}) ->
+    %%io:format("~p send blob bang\n",[self()]),   
+    Pid ! {blob, copy_blob(MsgEnv)},
+    true.
+
+send3_new_state(State, Blob) ->
+    case random:uniform(5+2) of
+        N when N =< 5-> setelement(N, State, Blob);
+        _ -> State  % Don't store blob
+    end.
+
 neg(doc) -> ["Negative testing of load_nif"];
 neg(Config) when is_list(Config) ->
     TmpMem = tmpmem(),
@@ -1070,10 +1227,13 @@ send_new_blob(_,_) -> ?nif_stub.
 alloc_msgenv() -> ?nif_stub.
 clear_msgenv(_) -> ?nif_stub.
 grow_blob(_,_) -> ?nif_stub.
+grow_blob(_,_,_) -> ?nif_stub.
 send_blob(_,_) -> ?nif_stub.
+send3_blob(_,_,_) -> ?nif_stub.
 send_blob_thread(_,_,_) -> ?nif_stub.
 join_send_thread(_) -> ?nif_stub.
-
+copy_blob(_) -> ?nif_stub.
+send_term(_,_) -> ?nif_stub.
 
 nif_stub_error(Line) ->
     exit({nif_not_loaded,module,?MODULE,line,Line}).
diff --git a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
index c8cd323b7e..17f644829f 100644
--- a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
+++ b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
@@ -1038,36 +1038,41 @@ static ERL_NIF_TERM make_term_copy(struct make_term_info* mti, int n)
 {
     return enif_make_copy(mti->dst_env, mti->other_term);
 }
+
+typedef ERL_NIF_TERM Make_term_Func(struct make_term_info*, int);
+static Make_term_Func* make_funcs[] = {
+    make_term_binary,
+    make_term_int,
+    make_term_ulong,
+    make_term_double,
+    make_term_atom,
+    make_term_existing_atom,
+    make_term_string,
+    //make_term_ref,
+    make_term_sub_binary,
+    make_term_uint,
+    make_term_long,
+    make_term_tuple0,
+    make_term_list0,
+    make_term_resource,
+    make_term_new_binary,
+    make_term_caller_pid,
+    make_term_tuple,
+    make_term_list,
+    make_term_list_cell,
+    make_term_tuple_from_array,
+    make_term_list_from_array,
+    make_term_garbage,
+    make_term_copy
+};
+static unsigned num_of_make_funcs()
+{
+    return sizeof(make_funcs)/sizeof(*make_funcs);
+}
 static int make_term_n(struct make_term_info* mti, int n, ERL_NIF_TERM* res)
 {
-    typedef ERL_NIF_TERM Make_term_Func(struct make_term_info*, int);
-    static Make_term_Func* funcs[] = {
-	make_term_binary,
-	make_term_int,
-	make_term_ulong,
-	make_term_double,
-	make_term_atom,
-	make_term_existing_atom,
-	make_term_string,
-	//make_term_ref,
-	make_term_sub_binary,
-	make_term_uint,
-	make_term_long,
-	make_term_tuple0,
-	make_term_list0,
-	make_term_resource,
-	make_term_new_binary,
-	make_term_caller_pid,
-	make_term_tuple,
-	make_term_list,
-	make_term_list_cell,
-	make_term_tuple_from_array,
-	make_term_list_from_array,
-	make_term_garbage,
-	make_term_copy
-    };
-    if (n < sizeof(funcs)/sizeof(*funcs)) {
-	*res = funcs[n](mti, n);
+    if (n < num_of_make_funcs()) {
+	*res = make_funcs[n](mti, n);
 	push_term(mti, *res);
 	return 1;
     }
@@ -1167,14 +1172,14 @@ static ERL_NIF_TERM grow_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[
 {
     union { void* vp; struct make_term_info* p; }mti;
     ERL_NIF_TERM term;
-    if (!enif_get_resource(env, argv[0], msgenv_resource_type, &mti.vp)) {
+    if (!enif_get_resource(env, argv[0], msgenv_resource_type, &mti.vp)
+	|| (argc>2 && !enif_get_uint(env,argv[2], &mti.p->n))) {
 	return enif_make_badarg(env);
     }
     mti.p->caller_env = env;
     mti.p->other_term = argv[1];
-    while (!make_term_n(mti.p, mti.p->n++, &term)) {
-	mti.p->n = 0;
-    }
+    mti.p->n %= num_of_make_funcs();
+    make_term_n(mti.p, mti.p->n++, &term);
     mti.p->blob = enif_make_list_cell(mti.p->dst_env, term, mti.p->blob);
     return atom_ok;
 }
@@ -1194,6 +1199,23 @@ static ERL_NIF_TERM send_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[
     return enif_make_tuple3(env, atom_ok, enif_make_int(env,res), copy);
 }
 
+static ERL_NIF_TERM send3_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    union { void* vp; struct make_term_info* p; }mti;
+    ErlNifPid to;
+    ERL_NIF_TERM copy;
+    int res;
+    if (!enif_get_resource(env, argv[0], msgenv_resource_type, &mti.vp)
+	|| !enif_get_local_pid(env, argv[1], &to)) {
+	return enif_make_badarg(env);
+    }
+    mti.p->blob = enif_make_tuple2(mti.p->dst_env, 
+				   enif_make_copy(mti.p->dst_env, argv[2]),
+				   mti.p->blob);
+    res = enif_send(env, &to, mti.p->dst_env, mti.p->blob);
+    return enif_make_int(env,res);
+}
+
 void* threaded_sender(void *arg)
 {
 
@@ -1253,6 +1275,28 @@ static ERL_NIF_TERM join_send_thread(ErlNifEnv* env, int argc, const ERL_NIF_TER
     return enif_make_tuple2(env, atom_ok, enif_make_int(env, mti.p->send_res));    
 }
 
+static ERL_NIF_TERM copy_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    union { void* vp; struct make_term_info* p; }mti;
+    if (!enif_get_resource(env, argv[0], msgenv_resource_type, &mti.vp)) {
+	return enif_make_badarg(env);
+    }
+    return enif_make_copy(env, mti.p->blob);
+}
+
+static ERL_NIF_TERM send_term(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    ErlNifEnv* menv;
+    ErlNifPid pid;
+    int ret;
+    if (!enif_get_local_pid(env, argv[0], &pid)) {
+	return enif_make_badarg(env);
+    }
+    menv = enif_alloc_env();
+    ret = enif_send(env, &pid, menv, enif_make_copy(menv, argv[1]));
+    enif_free_env(menv);
+    return enif_make_int(env, ret);
+}
 
 static ErlNifFunc nif_funcs[] =
 {
@@ -1291,9 +1335,13 @@ static ErlNifFunc nif_funcs[] =
     {"alloc_msgenv", 0, alloc_msgenv},
     {"clear_msgenv", 1, clear_msgenv},
     {"grow_blob", 2, grow_blob},
+    {"grow_blob", 3, grow_blob},
     {"send_blob", 2, send_blob},
+    {"send3_blob", 3, send3_blob},
     {"send_blob_thread", 3, send_blob_thread},
-    {"join_send_thread", 1, join_send_thread}
+    {"join_send_thread", 1, join_send_thread},
+    {"copy_blob", 1, copy_blob},
+    {"send_term", 2, send_term}
 };
 
 ERL_NIF_INIT(nif_SUITE,nif_funcs,load,reload,upgrade,unload)