Use generic lock-free queue for async threads

Queues used for communication between async threads and scheduler threads
have been replaced with lock-free queues.

Drivers using the driver_async functionality are not automatically locked
to the system anymore, and can be unloaded as any dynamically linked in
driver.

Scheduling of ready async jobs is now also interleaved in between other
jobs. Previously all ready async jobs was performed at once.

2 files changed, 126 insertions, 1 deletions

diff --git a/erts/emulator/test/driver_SUITE_data/Makefile.src b/erts/emulator/test/driver_SUITE_data/Makefile.src
index 62ab5169c0..dd48f6a0f7 100644
--- a/erts/emulator/test/driver_SUITE_data/Makefile.src
+++ b/erts/emulator/test/driver_SUITE_data/Makefile.src
@@ -13,7 +13,8 @@ MISC_DRVS =		outputv_drv@dll@ \
 			missing_callback_drv@dll@ \
 			thr_alloc_drv@dll@ \
 			otp_9302_drv@dll@ \
-			thr_free_drv@dll@
+			thr_free_drv@dll@ \
+			async_blast_drv@dll@
 
 SYS_INFO_DRVS = 	sys_info_1_0_drv@dll@ \
 			sys_info_1_1_drv@dll@ \
diff --git a/erts/emulator/test/driver_SUITE_data/async_blast_drv.c b/erts/emulator/test/driver_SUITE_data/async_blast_drv.c
new file mode 100644
index 0000000000..3821f7e3dc
--- /dev/null
+++ b/erts/emulator/test/driver_SUITE_data/async_blast_drv.c
@@ -0,0 +1,124 @@
+/*
+ * %CopyrightBegin%
+ *
+ * Copyright Ericsson AB 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%
+ */
+
+#include "erl_driver.h"
+
+#define NO_ASYNC_JOBS 10000
+
+static void stop(ErlDrvData drv_data);
+static ErlDrvData start(ErlDrvPort port,
+			char *command);
+static void output(ErlDrvData drv_data,
+		   char *buf, int len);
+static void ready_async(ErlDrvData drv_data,
+			ErlDrvThreadData thread_data);
+
+static ErlDrvEntry async_blast_drv_entry = { 
+    NULL /* init */,
+    start,
+    stop,
+    output,
+    NULL /* ready_input */,
+    NULL /* ready_output */,
+    "async_blast_drv",
+    NULL /* finish */,
+    NULL /* handle */,
+    NULL /* control */,
+    NULL /* timeout */,
+    NULL /* outputv */,
+    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 caller;    
+    int counter;
+} async_blast_data_t;
+
+
+DRIVER_INIT(async_blast_drv)
+{
+    return &async_blast_drv_entry;
+}
+
+static void stop(ErlDrvData drv_data)
+{
+    driver_free((void *) drv_data);
+}
+
+static ErlDrvData start(ErlDrvPort port,
+			char *command)
+{
+    async_blast_data_t *abd;
+
+    abd = driver_alloc(sizeof(async_blast_data_t));
+    if (!abd)
+	return ERL_DRV_ERROR_GENERAL;
+
+    abd->port = port;
+    abd->counter = 0;
+    return (ErlDrvData) abd;
+}
+
+static void async_invoke(void *data)
+{
+
+}
+#include <stdio.h>
+
+static void ready_async(ErlDrvData drv_data,
+			ErlDrvThreadData thread_data)
+{
+    async_blast_data_t *abd = (async_blast_data_t *) drv_data;
+    if (--abd->counter == 0) {
+	ErlDrvTermData spec[] = {
+	    ERL_DRV_PORT, driver_mk_port(abd->port),
+	    ERL_DRV_ATOM, driver_mk_atom("done"),
+	    ERL_DRV_TUPLE, 2
+	};
+	driver_send_term(abd->port, abd->caller,
+			 spec, sizeof(spec)/sizeof(spec[0]));
+    }
+}
+
+static void output(ErlDrvData drv_data,
+		   char *buf, int len)
+{
+    async_blast_data_t *abd = (async_blast_data_t *) drv_data;
+    if (abd->counter == 0) {
+	int i;
+	abd->caller = driver_caller(abd->port);
+	abd->counter = NO_ASYNC_JOBS;
+	for (i = 0; i < NO_ASYNC_JOBS; i++) {
+	    if (0 > driver_async(abd->port, NULL, async_invoke, NULL, NULL)) {
+		driver_failure_atom(abd->port, "driver_async_failed");
+		break;
+	    }
+	}
+    }
+}