Implement functionality for delaying thread progress from unmanaged threads

3 files changed, 241 insertions, 3 deletions

diff --git a/erts/emulator/test/driver_SUITE.erl b/erts/emulator/test/driver_SUITE.erl
index 643357263c..9a66257aff 100644
--- a/erts/emulator/test/driver_SUITE.erl
+++ b/erts/emulator/test/driver_SUITE.erl
@@ -77,7 +77,8 @@
 	 thread_mseg_alloc_cache_clean/1,
 	 otp_9302/1,
 	 thr_free_drv/1,
-	 async_blast/1]).
+	 async_blast/1,
+	 thr_msg_blast/1]).
 
 -export([bin_prefix/2]).
 
@@ -147,7 +148,8 @@ all() ->
      thread_mseg_alloc_cache_clean,
      otp_9302,
      thr_free_drv,
-     async_blast].
+     async_blast,
+     thr_msg_blast].
 
 groups() -> 
     [{timer, [],
@@ -2010,7 +2012,64 @@ async_blast(Config) when is_list(Config) ->
     ?line erlang:display({async_blast_time, AsyncBlastTime}),
     ?line ok.
 
+thr_msg_blast_receiver(_Port, N, N) ->
+    ok;
+thr_msg_blast_receiver(Port, N, Max) ->
+    receive
+	{Port, hi} ->
+	    thr_msg_blast_receiver(Port, N+1, Max)
+    end.
+
+thr_msg_blast_receiver_proc(Port, Max, Parent, Done) ->
+    case port_control(Port, 0, "") of
+	"receiver" ->
+	    spawn(fun () ->
+			  thr_msg_blast_receiver_proc(Port, Max+1, Parent, Done)
+		  end),
+	    thr_msg_blast_receiver(Port, 0, Max);
+	"done" ->
+	    Parent ! Done
+    end.
 
+thr_msg_blast(Config) when is_list(Config) ->
+    case erlang:system_info(smp_support) of
+	false ->
+	    {skipped, "Non-SMP emulator; nothing to test..."};
+	true ->
+	    Path = ?config(data_dir, Config),
+	    erl_ddll:start(),
+	    ok = load_driver(Path, thr_msg_blast_drv),
+	    MemBefore = driver_alloc_size(),
+	    Start = os:timestamp(),
+	    Port = open_port({spawn, thr_msg_blast_drv}, []),
+	    true = is_port(Port),
+	    Done = make_ref(),
+	    Me = self(),
+	    spawn(fun () ->
+			  thr_msg_blast_receiver_proc(Port, 1, Me, Done)
+		  end),
+	    receive
+		Done -> ok
+	    end,
+	    ok = thr_msg_blast_receiver(Port, 0, 32*10000),
+	    port_close(Port),
+	    End = os:timestamp(),
+	    receive
+		Garbage ->
+		    ?t:fail({received_garbage, Port, Garbage})
+	    after 2000 ->
+		    ok
+	    end,
+	    MemAfter = driver_alloc_size(),
+	    io:format("MemBefore=~p, MemAfter=~p~n",
+		      [MemBefore, MemAfter]),
+	    ThrMsgBlastTime = timer:now_diff(End,Start)/1000000,
+	    io:format("ThrMsgBlastTime=~p~n", [ThrMsgBlastTime]),
+	    MemBefore = MemAfter,
+	    Res = {thr_msg_blast_time, ThrMsgBlastTime},
+	    erlang:display(Res),
+	    Res
+    end.
 
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %% 		Utilities
diff --git a/erts/emulator/test/driver_SUITE_data/Makefile.src b/erts/emulator/test/driver_SUITE_data/Makefile.src
index 9cc107cc66..b667dff6b6 100644
--- a/erts/emulator/test/driver_SUITE_data/Makefile.src
+++ b/erts/emulator/test/driver_SUITE_data/Makefile.src
@@ -14,7 +14,8 @@ MISC_DRVS =		outputv_drv@dll@ \
 			thr_alloc_drv@dll@ \
 			otp_9302_drv@dll@ \
 			thr_free_drv@dll@ \
-			async_blast_drv@dll@
+			async_blast_drv@dll@ \
+			thr_msg_blast_drv@dll@
 
 SYS_INFO_DRVS = 	sys_info_base_drv@dll@ \
 			sys_info_prev_drv@dll@ \
diff --git a/erts/emulator/test/driver_SUITE_data/thr_msg_blast_drv.c b/erts/emulator/test/driver_SUITE_data/thr_msg_blast_drv.c
new file mode 100644
index 0000000000..1070678d7b
--- /dev/null
+++ b/erts/emulator/test/driver_SUITE_data/thr_msg_blast_drv.c
@@ -0,0 +1,178 @@
+/*
+ * %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%
+ */
+
+#include "erl_driver.h"
+
+#define THR_MSG_BLAST_NO_PROCS 10
+#define THR_MSG_BLAST_NO_SENDS_PER_PROC 10000
+
+#define THR_MSG_BLAST_THREADS 32
+
+static void stop(ErlDrvData drv_data);
+static ErlDrvData start(ErlDrvPort port,
+			char *command);
+static ErlDrvSSizeT control(ErlDrvData drv_data,
+			    unsigned int command,
+			    char *buf, ErlDrvSizeT len,
+			    char **rbuf, ErlDrvSizeT rlen);
+
+static ErlDrvEntry thr_msg_blast_drv_entry = { 
+    NULL /* init */,
+    start,
+    stop,
+    NULL /* output */,
+    NULL /* ready_input */,
+    NULL /* ready_output */,
+    "thr_msg_blast_drv",
+    NULL /* finish */,
+    NULL /* handle */,
+    control,
+    NULL /* 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,
+    ERL_DRV_FLAG_USE_PORT_LOCKING,
+    NULL /* handle2 */,
+    NULL /* handle_monitor */
+};
+
+typedef struct {
+    ErlDrvPort port;
+    ErlDrvTermData td_port;
+    ErlDrvTermData hi;
+    ErlDrvTid tid[THR_MSG_BLAST_THREADS];
+    int no_thrs;
+    ErlDrvTermData proc[THR_MSG_BLAST_NO_PROCS];
+    int no_procs;
+} thr_msg_blast_data_t;
+
+
+DRIVER_INIT(thr_msg_blast_drv)
+{
+    return &thr_msg_blast_drv_entry;
+}
+
+static void stop(ErlDrvData drv_data)
+{
+    int i;
+    thr_msg_blast_data_t *tmbd = (thr_msg_blast_data_t *) drv_data;
+    for (i = 0; i < tmbd->no_thrs; i++)
+	erl_drv_thread_join(tmbd->tid[i], NULL);
+    driver_free((void *) tmbd);
+}
+
+static ErlDrvData start(ErlDrvPort port,
+			char *command)
+{
+    thr_msg_blast_data_t *tmbd;
+
+    tmbd = driver_alloc(sizeof(thr_msg_blast_data_t));
+    if (!tmbd)
+	return ERL_DRV_ERROR_GENERAL;
+
+    tmbd->port = port;
+    tmbd->td_port = driver_mk_port(port);
+    tmbd->hi = driver_mk_atom("hi");
+    tmbd->no_thrs = 0;
+    tmbd->no_procs = 1;
+    tmbd->proc[0] = driver_caller(port);
+
+    return (ErlDrvData) tmbd;
+}
+
+static void *thread(void *);
+
+static ErlDrvSSizeT control(ErlDrvData drv_data,
+			    unsigned int command,
+			    char *buf, ErlDrvSizeT len,
+			    char **rbuf, ErlDrvSizeT rlen)
+{
+    thr_msg_blast_data_t *tmbd = (thr_msg_blast_data_t *) drv_data;
+    char *res_str = "error";
+
+    if (tmbd->no_procs >= THR_MSG_BLAST_NO_PROCS) {
+	int i;
+	for (i = 0; i < tmbd->no_thrs; i++)
+	    erl_drv_thread_join(tmbd->tid[i], NULL);
+	tmbd->no_thrs = 0;
+	res_str = "done";
+    }
+    else {
+
+	tmbd->proc[tmbd->no_procs++] = driver_caller(tmbd->port);
+
+	if (tmbd->no_procs == THR_MSG_BLAST_NO_PROCS) {
+	    for (tmbd->no_thrs = 0;
+		 tmbd->no_thrs < THR_MSG_BLAST_THREADS;
+		 tmbd->no_thrs++) {
+		int res = erl_drv_thread_create("test",
+						&tmbd->tid[tmbd->no_thrs],
+						thread,
+						tmbd,
+						NULL);
+		if (res != 0) {
+		    driver_failure_posix(tmbd->port, res);
+		    goto done;
+		}
+	    }
+	}
+
+	res_str = "receiver";
+    }
+
+ done: {
+	ErlDrvSSizeT res_len = strlen(res_str);
+	if (res_len > rlen) {
+	    char *abuf = driver_alloc(sizeof(char)*res_len);
+	    if (!abuf)
+		return 0;
+	    *rbuf = abuf;
+	}
+
+	memcpy((void *) *rbuf, (void *) res_str, res_len);
+
+	return res_len;
+    }
+}
+
+static void *thread(void *varg)
+{
+    int s, p;
+    thr_msg_blast_data_t *tmbd = (thr_msg_blast_data_t *) varg;
+    ErlDrvTermData spec[] = {
+	ERL_DRV_PORT, tmbd->td_port,
+	ERL_DRV_ATOM, tmbd->hi,
+	ERL_DRV_TUPLE, 2
+    };
+
+    for (s = 0; s < THR_MSG_BLAST_NO_SENDS_PER_PROC; s++) {
+	for (p = 0; p < THR_MSG_BLAST_NO_PROCS; p++) {
+	    int res = driver_send_term(tmbd->port, tmbd->proc[p],
+				       spec, sizeof(spec)/sizeof(spec[0]));
+	    if (p == 0 && res <= 0)
+		abort(); /* Could not send to creator */
+	}
+    }
+    return NULL;
+}