12 files changed, 701 insertions, 344 deletions

diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index 9f0cb681c0..962db8b831 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -201,7 +201,7 @@ type	LINEBUF		STANDARD	SYSTEM		line_buf
 type	IOQ		STANDARD	SYSTEM		io_queue
 type	BITS_BUF	STANDARD	SYSTEM		bits_buf
 type	TMP_DIST_BUF	TEMPORARY	SYSTEM		tmp_dist_buf
-type	ASYNC_Q		LONG_LIVED	SYSTEM		async_queue
+type	ASYNC_DATA	LONG_LIVED	SYSTEM		internal_async_data
 type	ESTACK		TEMPORARY	SYSTEM		estack
 type	PORT_CALL_BUF	TEMPORARY	SYSTEM		port_call_buf
 type	DB_TABLE	ETS		ETS		db_tab
@@ -308,12 +308,6 @@ type	ETHR_STD	STANDARD	SYSTEM		ethread_standard
 type	ETHR_SL		SHORT_LIVED	SYSTEM		ethread_short_lived
 type	ETHR_LL		LONG_LIVED	SYSTEM		ethread_long_lived
 
-+ifnot smp
-
-type	ARCALLBACK	LONG_LIVED	SYSTEM		async_ready_callback
-
-+endif
-
 +endif
 
 +if shared_heap
diff --git a/erts/emulator/beam/erl_async.c b/erts/emulator/beam/erl_async.c
index 91b64411d4..2dc7237f7c 100644
--- a/erts/emulator/beam/erl_async.c
+++ b/erts/emulator/beam/erl_async.c
@@ -24,10 +24,18 @@
 #include "erl_sys_driver.h"
 #include "global.h"
 #include "erl_threads.h"
+#include "erl_thr_queue.h"
+#include "erl_async.h"
+
+#define ERTS_MAX_ASYNC_READY_CALLS_IN_SEQ 20
+
+#define ERTS_ASYNC_PRINT_JOB 0
+
+#if !defined(ERTS_SMP) && defined(USE_THREADS) && !ERTS_USE_ASYNC_READY_Q
+#  error "Need async ready queue in non-smp case"
+#endif
 
 typedef struct _erl_async {
-    struct _erl_async* next;
-    struct _erl_async* prev;
     DE_Handle*         hndl;   /* The DE_Handle is needed when port is gone */
     Eterm              port;
     long               async_id;
@@ -35,345 +43,498 @@ typedef struct _erl_async {
     ErlDrvPDL          pdl;
     void (*async_invoke)(void*);
     void (*async_free)(void*);
-} ErlAsync;
+#if ERTS_USE_ASYNC_READY_Q
+    Uint               sched_id;
+    union {
+	ErtsThrQPrepEnQ_t *prep_enq;
+	ErtsThrQFinDeQ_t   fin_deq;
+    } q;
+#endif
+} ErtsAsync;
+
+#if ERTS_USE_ASYNC_READY_Q
+
+/*
+ * We can do without the enqueue mutex since it isn't needed for
+ * thread safety. Its only purpose is to put async threads to sleep
+ * during a blast of ready async jobs. This in order to reduce
+ * contention on the enqueue end of the async ready queues. During
+ * such a blast without the enqueue mutex much cpu time is consumed
+ * by the async threads without them doing much progress which in turn
+ * slow down progress of scheduler threads.
+ */
+#define ERTS_USE_ASYNC_READY_ENQ_MTX 1
+
+#if ERTS_USE_ASYNC_READY_ENQ_MTX
 
 typedef struct {
-    erts_mtx_t mtx;
-    erts_cnd_t cv;
-    erts_tid_t thr;
-    int   len;
-#ifndef ERTS_SMP
-    int   hndl;
+    erts_mtx_t enq_mtx;
+} ErtsAsyncReadyQXData;
+
 #endif
-    ErlAsync* head;
-    ErlAsync* tail;
-#ifdef ERTS_ENABLE_LOCK_CHECK
-    int no;
+
+typedef struct {
+#if ERTS_USE_ASYNC_READY_ENQ_MTX
+    union {
+	ErtsAsyncReadyQXData data;
+	char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(
+		sizeof(ErtsAsyncReadyQXData))];
+    } x;
 #endif
-} AsyncQueue;
+    ErtsThrQ_t thr_q;
+    ErtsThrQFinDeQ_t fin_deq;
+} ErtsAsyncReadyQ;
 
-static erts_smp_spinlock_t async_id_lock;
-static long async_id = 0;
 
+typedef union {
+    ErtsAsyncReadyQ arq;
+    char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsAsyncReadyQ))];
+} ErtsAlgndAsyncReadyQ;
 
-#ifndef ERTS_SMP
+#endif /* ERTS_USE_ASYNC_READY_Q */
 
-erts_mtx_t async_ready_mtx;
-static ErlAsync* async_ready_list = NULL;
+typedef struct {
+    ErtsThrQ_t thr_q;
+    erts_tid_t thr_id;
+} ErtsAsyncQ;
+
+typedef union {
+    ErtsAsyncQ aq;
+    char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsAsyncQ))];
+} ErtsAlgndAsyncQ;
 
+typedef struct {
+    int no_initialized;
+    erts_mtx_t mtx;
+    erts_cnd_t cnd;
+    erts_atomic_t id;
+} ErtsAsyncInit;
+
+typedef struct {
+    union {
+	ErtsAsyncInit data;
+	char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsAsyncInit))];
+    } init;
+    ErtsAlgndAsyncQ *queue;
+#if ERTS_USE_ASYNC_READY_Q
+    ErtsAlgndAsyncReadyQ *ready_queue;
 #endif
+} ErtsAsyncData;
 
-/*
-** Initialize worker threads (if supported)
-*/
+int erts_async_max_threads; /* Initialized by erl_init.c */
+int erts_async_thread_suggested_stack_size; /* Initialized by erl_init.c */
 
-/* Detach from driver */
-static void async_detach(DE_Handle* dh)
-{
-    return;
-}
+static ErtsAsyncData *async;
 
+#ifndef USE_THREADS
 
-#ifdef USE_THREADS
+void
+erts_init_async(void)
+{
 
-static AsyncQueue* async_q;
+}
 
-static void* async_main(void*);
-static void async_add(ErlAsync*, AsyncQueue*);
+#else
 
-#ifndef ERTS_SMP
-typedef struct ErtsAsyncReadyCallback_ ErtsAsyncReadyCallback;
-struct ErtsAsyncReadyCallback_ {
-    struct ErtsAsyncReadyCallback_ *next;
-    void (*callback)(void);
-};
+static void *async_main(void *);
 
-static ErtsAsyncReadyCallback *callbacks;
-static int async_handle;
+static ERTS_INLINE ErtsAsyncQ *
+async_q(int i)
+{
+    return &async->queue[i].aq;
+}
+
+#if ERTS_USE_ASYNC_READY_Q
 
-int erts_register_async_ready_callback(void (*funcp)(void))
+static ERTS_INLINE ErtsAsyncReadyQ *
+async_ready_q(Uint sched_id)
 {
-    ErtsAsyncReadyCallback *cb = erts_alloc(ERTS_ALC_T_ARCALLBACK,
-					    sizeof(ErtsAsyncReadyCallback));
-    cb->next = callbacks;
-    cb->callback = funcp;
-    erts_mtx_lock(&async_ready_mtx);
-    callbacks = cb;
-    erts_mtx_unlock(&async_ready_mtx);
-    return async_handle;
+    return &async->ready_queue[((int)sched_id)-1].arq;
 }
+
 #endif
 
-int init_async(int hndl)
+void
+erts_init_async(void)
 {
-    erts_thr_opts_t thr_opts = ERTS_THR_OPTS_DEFAULT_INITER;
-    AsyncQueue* q;
-    int i;
+    async = NULL;
+    if (erts_async_max_threads > 0) {
+#if ERTS_USE_ASYNC_READY_Q
+	ErtsThrQInit_t qinit = ERTS_THR_Q_INIT_DEFAULT;
+#endif
+	erts_thr_opts_t thr_opts = ERTS_THR_OPTS_DEFAULT_INITER;
+	char *ptr;
+	size_t tot_size = 0;
+	int i;
+
+	tot_size += ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsAsyncData));
+	tot_size += sizeof(ErtsAlgndAsyncQ)*erts_async_max_threads;
+#if ERTS_USE_ASYNC_READY_Q
+	tot_size += sizeof(ErtsAlgndAsyncReadyQ)*erts_no_schedulers;
+#endif
 
-    thr_opts.detached = 0;
-    thr_opts.suggested_stack_size = erts_async_thread_suggested_stack_size;
-
-#ifndef ERTS_SMP
-    callbacks = NULL;
-    async_handle = hndl;
-    erts_mtx_init(&async_ready_mtx, "async_ready");
-    async_ready_list = NULL;
-#endif
-
-    async_id = 0;
-    erts_smp_spinlock_init(&async_id_lock, "async_id");
-
-    async_q = q = (AsyncQueue*)
-	(erts_async_max_threads
-	 ? erts_alloc(ERTS_ALC_T_ASYNC_Q,
-		      erts_async_max_threads * sizeof(AsyncQueue))
-	 : NULL);
-    for (i = 0; i < erts_async_max_threads; i++) {
-	q->head = NULL;
-	q->tail = NULL;
-	q->len = 0;
-#ifndef ERTS_SMP
-	q->hndl = hndl;
-#endif
-#ifdef ERTS_ENABLE_LOCK_CHECK
-	q->no = i;
-#endif
-	erts_mtx_init(&q->mtx, "asyncq");
-	erts_cnd_init(&q->cv);
-	erts_thr_create(&q->thr, async_main, (void*)q, &thr_opts);
-	q++;
-    }
-    return 0;
-}
+	ptr = erts_alloc_permanent_cache_aligned(ERTS_ALC_T_ASYNC_DATA,
+						 tot_size);
 
+	async = (ErtsAsyncData *) ptr;
+	ptr += ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsAsyncData));
 
-int exit_async()
-{
-    int i;
+	async->init.data.no_initialized = 0;
+	erts_mtx_init(&async->init.data.mtx, "async_init_mtx");
+	erts_cnd_init(&async->init.data.cnd);
+	erts_atomic_init_nob(&async->init.data.id, 0);
 
-    /* terminate threads */
-    for (i = 0; i < erts_async_max_threads; i++) {
-	ErlAsync* a = (ErlAsync*) erts_alloc(ERTS_ALC_T_ASYNC,
-					     sizeof(ErlAsync));
-	a->port = NIL;
-	async_add(a, &async_q[i]);
-    }
+	async->queue = (ErtsAlgndAsyncQ *) ptr;
+	ptr += sizeof(ErtsAlgndAsyncQ)*erts_async_max_threads;
 
-    for (i = 0; i < erts_async_max_threads; i++) {
-	erts_thr_join(async_q[i].thr, NULL);
-	erts_mtx_destroy(&async_q[i].mtx);
-	erts_cnd_destroy(&async_q[i].cv);
-    }
-#ifndef ERTS_SMP
-    erts_mtx_destroy(&async_ready_mtx);
+#if ERTS_USE_ASYNC_READY_Q
+
+	qinit.live.queue = ERTS_THR_Q_LIVE_LONG;
+	qinit.live.objects = ERTS_THR_Q_LIVE_SHORT;
+	qinit.notify = erts_notify_check_async_ready_queue;
+
+	async->ready_queue = (ErtsAlgndAsyncReadyQ *) ptr;
+	ptr += sizeof(ErtsAlgndAsyncReadyQ)*erts_no_schedulers;
+
+	for (i = 1; i <= erts_no_schedulers; i++) {
+	    ErtsAsyncReadyQ *arq = async_ready_q(i);
+#if ERTS_USE_ASYNC_READY_ENQ_MTX
+	    erts_mtx_init(&arq->x.data.enq_mtx, "async_enq_mtx");
 #endif
-    if (async_q)
-	erts_free(ERTS_ALC_T_ASYNC_Q, (void *) async_q);
-    return 0;
+	    erts_thr_q_finalize_dequeue_state_init(&arq->fin_deq);
+	    qinit.arg = (void *) (SWord) i;
+	    erts_thr_q_initialize(&arq->thr_q, &qinit);
+	}
+
+#endif
+
+	/* Create async threads... */
+
+	thr_opts.detached = 0;
+	thr_opts.suggested_stack_size
+	    = erts_async_thread_suggested_stack_size;
+
+	for (i = 0; i < erts_async_max_threads; i++) {
+	    ErtsAsyncQ *aq = async_q(i);
+	    erts_thr_create(&aq->thr_id, async_main, (void*) aq, &thr_opts);
+	}
+
+	/* Wait for async threads to initialize... */
+
+	erts_mtx_lock(&async->init.data.mtx);
+	while (async->init.data.no_initialized != erts_async_max_threads)
+	    erts_cnd_wait(&async->init.data.cnd, &async->init.data.mtx);
+	erts_mtx_unlock(&async->init.data.mtx);
+
+	erts_mtx_destroy(&async->init.data.mtx);
+	erts_cnd_destroy(&async->init.data.cnd);
+
+    }
 }
 
+#if ERTS_USE_ASYNC_READY_Q
 
-static void async_add(ErlAsync* a, AsyncQueue* q)
+void *
+erts_get_async_ready_queue(Uint sched_id)
+{
+    return (void *) async ? async_ready_q(sched_id) : NULL;
+}
+
+#endif
+
+static ERTS_INLINE void async_add(ErtsAsync *a, ErtsAsyncQ* q)
 {
     if (is_internal_port(a->port)) {
-	ERTS_LC_ASSERT(erts_drvportid2port(a->port));
+#if ERTS_USE_ASYNC_READY_Q
+	ErtsAsyncReadyQ *arq = async_ready_q(a->sched_id);
+	a->q.prep_enq = erts_thr_q_prepare_enqueue(&arq->thr_q);
+#endif
 	/* make sure the driver will stay around */
-	driver_lock_driver(internal_port_index(a->port));
+	if (a->hndl)
+	    erts_ddll_reference_referenced_driver(a->hndl);
     }
 
-    erts_mtx_lock(&q->mtx);
+#if ERTS_ASYNC_PRINT_JOB
+    erts_fprintf(stderr, "-> %ld\n", a->async_id);
+#endif
 
-    if (q->len == 0) {
-	q->head = a;
-	q->tail = a;
-	q->len = 1;
-	erts_cnd_signal(&q->cv);
-    }
-    else { /* no need to signal (since the worker is working) */
-	a->next = q->head;
-	q->head->prev = a;
-	q->head = a;
-	q->len++;
-    }
-    erts_mtx_unlock(&q->mtx);
+    erts_thr_q_enqueue(&q->thr_q, a);
 }
 
-static ErlAsync* async_get(AsyncQueue* q)
+static ERTS_INLINE ErtsAsync *async_get(ErtsThrQ_t *q,
+					erts_tse_t *tse,
+					ErtsThrQPrepEnQ_t **prep_enq)
 {
-    ErlAsync* a;
+#if ERTS_USE_ASYNC_READY_Q
+    int saved_fin_deq = 0;
+    ErtsThrQFinDeQ_t fin_deq;
+#endif
 
-    erts_mtx_lock(&q->mtx);
-    while((a = q->tail) == NULL) {
-	erts_cnd_wait(&q->cv, &q->mtx);
-    }
+    while (1) {
+	ErtsAsync *a = (ErtsAsync *) erts_thr_q_dequeue(q);
+	if (a) {
+
+#if ERTS_USE_ASYNC_READY_Q
+	    *prep_enq = a->q.prep_enq;
+	    erts_thr_q_get_finalize_dequeue_data(q, &a->q.fin_deq);
+	    if (saved_fin_deq)
+		erts_thr_q_append_finalize_dequeue_data(&a->q.fin_deq, &fin_deq);
+#endif
+
+	    return a;
+	}
+
+	if (ERTS_THR_Q_DIRTY != erts_thr_q_clean(q)) {
+	    ErtsThrQFinDeQ_t tmp_fin_deq;
+
+	    erts_tse_reset(tse);
+
+#if ERTS_USE_ASYNC_READY_Q
+	chk_fin_deq:
+	    if (erts_thr_q_get_finalize_dequeue_data(q, &tmp_fin_deq)) {
+		if (!saved_fin_deq) {
+		    erts_thr_q_finalize_dequeue_state_init(&fin_deq);
+		    saved_fin_deq = 1;
+		}
+		erts_thr_q_append_finalize_dequeue_data(&fin_deq,
+							&tmp_fin_deq);
+	    }
+#endif
+
+	    switch (erts_thr_q_inspect(q, 1)) {
+	    case ERTS_THR_Q_DIRTY:
+		break;
 #ifdef ERTS_SMP
-    ASSERT(a && q->tail == a);
+	    case ERTS_THR_Q_NEED_THR_PRGR: {
+		ErtsThrPrgrVal prgr = erts_thr_q_need_thr_progress(q);
+		erts_thr_progress_wakeup(NULL, prgr);
+		/*
+		 * We do no dequeue finalizing in hope that a new async
+		 * job will arrive before we are woken due to thread
+		 * progress...
+		 */
+		erts_tse_wait(tse);
+		break;
+	    }
 #endif
-    if (q->head == q->tail) {
-	q->head = q->tail = NULL;
-	q->len = 0;
-    }
-    else {
-	q->tail->prev->next = NULL;
-	q->tail = q->tail->prev;
-	q->len--;
+	    case ERTS_THR_Q_CLEAN:
+
+#if ERTS_USE_ASYNC_READY_Q
+		if (saved_fin_deq) {
+		    if (erts_thr_q_finalize_dequeue(&fin_deq))
+			goto chk_fin_deq;
+		    else
+			saved_fin_deq = 0;
+		}
+#endif
+
+		erts_tse_wait(tse);
+		break;
+
+	    default:
+		ASSERT(0);
+		break;
+	    }
+
+	}
     }
-    erts_mtx_unlock(&q->mtx);
-    return a;
 }
 
-
-static int async_del(long id)
+static ERTS_INLINE void call_async_ready(ErtsAsync *a)
 {
-    int i;
-    /* scan all queue for an entry with async_id == 'id' */
-
-    for (i = 0; i < erts_async_max_threads; i++) {
-	ErlAsync* a;
-	erts_mtx_lock(&async_q[i].mtx);
-	
-	a = async_q[i].head;
-	while(a != NULL) {
-	    if (a->async_id == id) {
-		if (a->prev != NULL)
-		    a->prev->next = a->next;
-		else
-		    async_q[i].head = a->next;
-		if (a->next != NULL)
-		    a->next->prev = a->prev;
-		else
-		    async_q[i].tail = a->prev;
-		async_q[i].len--;
-		erts_mtx_unlock(&async_q[i].mtx);
-		if (a->async_free != NULL)
-		    a->async_free(a->async_data);
-		async_detach(a->hndl);
-		erts_free(ERTS_ALC_T_ASYNC, a);
-		return 1;
-	    }
-	    a = a->next;
+    Port *p = erts_id2port_sflgs(a->port,
+				 NULL,
+				 0,
+				 ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP);
+    if (!p) {
+	if (a->async_free)
+	    a->async_free(a->async_data);
+    }
+    else {
+	if (async_ready(p, a->async_data)) {
+	    if (a->async_free)
+		a->async_free(a->async_data);
 	}
-	erts_mtx_unlock(&async_q[i].mtx);
+	erts_port_release(p);
     }
-    return 0;
+    if (a->hndl)
+	erts_ddll_dereference_driver(a->hndl);
 }
 
-static void* async_main(void* arg)
+static ERTS_INLINE void async_reply(ErtsAsync *a, ErtsThrQPrepEnQ_t *prep_enq)
 {
-    AsyncQueue* q = (AsyncQueue*) arg;
+#if ERTS_USE_ASYNC_READY_Q
+    ErtsAsyncReadyQ *arq;
 
-#ifdef ERTS_ENABLE_LOCK_CHECK
-    {
-	char buf[27];
-	erts_snprintf(&buf[0], 27, "async %d", q->no);
-	erts_lc_set_thread_name(&buf[0]);
-    }
+    if (a->pdl)
+	driver_pdl_dec_refc(a->pdl);
+
+#if ERTS_ASYNC_PRINT_JOB
+    erts_fprintf(stderr, "=>> %ld\n", a->async_id);
 #endif
 
-    while(1) {
-	ErlAsync* a = async_get(q);
+    arq = async_ready_q(a->sched_id);
 
-	if (a->port == NIL) { /* TIME TO DIE SIGNAL */
-	    erts_free(ERTS_ALC_T_ASYNC, (void *) a);
-	    break;
-	}
-	else {
-	    (*a->async_invoke)(a->async_data);
-	    /* Major problem if the code for async_invoke
-	       or async_free is removed during a blocking operation */
+#if ERTS_USE_ASYNC_READY_ENQ_MTX
+	erts_mtx_lock(&arq->x.data.enq_mtx);
+#endif
+
+	erts_thr_q_enqueue_prepared(&arq->thr_q, (void *) a, prep_enq);
+
+#if ERTS_USE_ASYNC_READY_ENQ_MTX
+	erts_mtx_unlock(&arq->x.data.enq_mtx);
+#endif
+
+#else /* ERTS_USE_ASYNC_READY_Q */
+
+	call_async_ready(a);
+	if (a->pdl)
+	    driver_pdl_dec_refc(a->pdl);
+	erts_free(ERTS_ALC_T_ASYNC, (void *) a);
+
+#endif /* ERTS_USE_ASYNC_READY_Q */
+}
+
+
+static void
+async_wakeup(void *vtse)
+{
+    erts_tse_set((erts_tse_t *) vtse);
+}
+
+static erts_tse_t *async_thread_init(ErtsAsyncQ *aq)
+{
+    ErtsThrQInit_t qinit = ERTS_THR_Q_INIT_DEFAULT;
+    erts_tse_t *tse = erts_tse_fetch();
 #ifdef ERTS_SMP
-	    {
-		Port *p;
-		p = erts_id2port_sflgs(a->port,
-				       NULL,
-				       0,
-				       ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP);
-		if (!p) {
-		    if (a->async_free)
-			(*a->async_free)(a->async_data);
-		}
-		else {
-		    if (async_ready(p, a->async_data)) {
-			if (a->async_free)
-			    (*a->async_free)(a->async_data);
-		    }
-		    async_detach(a->hndl);
-		    erts_port_release(p);
-		}
-		if (a->pdl) {
-		    driver_pdl_dec_refc(a->pdl);
-		}
-		erts_free(ERTS_ALC_T_ASYNC, (void *) a);
-	    }
-#else
-	    if (a->pdl) {
-		driver_pdl_dec_refc(a->pdl);
-	    }
-	    erts_mtx_lock(&async_ready_mtx);
-	    a->next = async_ready_list;
-	    async_ready_list = a;
-	    erts_mtx_unlock(&async_ready_mtx);
-	    sys_async_ready(q->hndl);
+    ErtsThrPrgrCallbacks callbacks;
+
+    callbacks.arg = (void *) tse;
+    callbacks.wakeup = async_wakeup;
+    callbacks.prepare_wait = NULL;
+    callbacks.wait = NULL;
+
+    erts_thr_progress_register_unmanaged_thread(&callbacks);
 #endif
-	}
-    }
 
-    return NULL;
+    qinit.live.queue = ERTS_THR_Q_LIVE_LONG;
+    qinit.live.objects = ERTS_THR_Q_LIVE_SHORT;
+    qinit.arg = (void *) tse;
+    qinit.notify = async_wakeup;
+#if ERTS_USE_ASYNC_READY_Q
+    qinit.auto_finalize_dequeue = 0;
+#endif
+
+    erts_thr_q_initialize(&aq->thr_q, &qinit);
+
+    /* Inform main thread that we are done initializing... */
+    erts_mtx_lock(&async->init.data.mtx);
+    async->init.data.no_initialized++;
+    erts_cnd_signal(&async->init.data.cnd);
+    erts_mtx_unlock(&async->init.data.mtx);
+
+    return tse;
 }
 
+static void *async_main(void* arg)
+{
+    ErtsAsyncQ *aq = (ErtsAsyncQ *) arg;
+    erts_tse_t *tse = async_thread_init(aq);
+
+    while (1) {
+	ErtsThrQPrepEnQ_t *prep_enq;
+	ErtsAsync *a = async_get(&aq->thr_q, tse, &prep_enq);
+	if (is_nil(a->port))
+	    break; /* Time to die */
 
+#if ERTS_ASYNC_PRINT_JOB
+	erts_fprintf(stderr, "<- %ld\n", a->async_id);
 #endif
 
-#ifndef ERTS_SMP
+	a->async_invoke(a->async_data);
+
+	async_reply(a, prep_enq);
+    }
+
+    return NULL;
+}
+
+#endif /* USE_THREADS */
 
-int check_async_ready(void)
+void
+erts_exit_flush_async(void)
 {
 #ifdef USE_THREADS
-    ErtsAsyncReadyCallback *cbs;
+    int i;
+    ErtsAsync a;
+    a.port = NIL;
+    /*
+     * Terminate threads in order to flush queues. We do not
+     * bother to clean everything up since we are about to
+     * terminate the runtime system and a cleanup would only
+     * delay the termination.
+     */
+    for (i = 0; i < erts_async_max_threads; i++)
+	async_add(&a, async_q(i));
+    for (i = 0; i < erts_async_max_threads; i++)
+	erts_thr_join(async->queue[i].aq.thr_id, NULL);
 #endif
-    ErlAsync* a;
-    int count = 0;
+}
 
-    erts_mtx_lock(&async_ready_mtx);
-    a = async_ready_list;
-    async_ready_list = NULL;
-#ifdef USE_THREADS
-    cbs = callbacks;
-#endif
-    erts_mtx_unlock(&async_ready_mtx);
-
-    while(a != NULL) {
-	ErlAsync* a_next = a->next;
-	/* Every port not dead */
-	Port *p = erts_id2port_sflgs(a->port,
-				     NULL,
-				     0,
-				     ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP);
-	if (!p) {
-	    if (a->async_free)
-		(*a->async_free)(a->async_data);
-	}
-	else {
-	    count++;
-	    if (async_ready(p, a->async_data)) {
-		if (a->async_free != NULL)
-		    (*a->async_free)(a->async_data);
-	    }
-	    async_detach(a->hndl);
-	    erts_port_release(p);
+#if defined(USE_THREADS) && ERTS_USE_ASYNC_READY_Q
+
+int erts_check_async_ready(void *varq)
+{
+    ErtsAsyncReadyQ *arq = (ErtsAsyncReadyQ *) varq;
+    int res = 1;
+    int i;
+
+    for (i = 0; i < ERTS_MAX_ASYNC_READY_CALLS_IN_SEQ; i++) {
+	ErtsAsync *a = (ErtsAsync *) erts_thr_q_dequeue(&arq->thr_q);
+	if (!a) {
+	    res = 0;
+	    break;
 	}
+
+#if ERTS_ASYNC_PRINT_JOB
+	erts_fprintf(stderr, "<<= %ld\n", a->async_id);
+#endif
+	erts_thr_q_append_finalize_dequeue_data(&arq->fin_deq, &a->q.fin_deq);
+	call_async_ready(a);
 	erts_free(ERTS_ALC_T_ASYNC, (void *) a);
-	a = a_next;
     }
-#ifdef USE_THREADS
-    for (; cbs; cbs = cbs->next)
-	(*cbs->callback)();
-#endif
-    return count;
+
+    erts_thr_q_finalize_dequeue(&arq->fin_deq);
+    
+    return res;
 }
 
+int erts_async_ready_clean(void *varq, void *val)
+{
+    ErtsAsyncReadyQ *arq = (ErtsAsyncReadyQ *) varq;
+    ErtsThrQCleanState_t cstate;
+
+    cstate = erts_thr_q_clean(&arq->thr_q);
+
+    if (erts_thr_q_finalize_dequeue(&arq->fin_deq))
+	return ERTS_ASYNC_READY_DIRTY;
+
+    switch (cstate) {
+    case ERTS_THR_Q_DIRTY:
+	return ERTS_ASYNC_READY_DIRTY;
+#ifdef ERTS_SMP
+    case ERTS_THR_Q_NEED_THR_PRGR:
+	*((ErtsThrPrgrVal *) val)
+	    = erts_thr_q_need_thr_progress(&arq->thr_q);
+	return ERTS_ASYNC_READY_NEED_THR_PRGR;
 #endif
+    case ERTS_THR_Q_CLEAN:
+	break;
+    }
+    return ERTS_ASYNC_READY_CLEAN;
+}
 
+#endif
 
 /*
 ** Schedule async_invoke on a worker thread
@@ -393,19 +554,29 @@ long driver_async(ErlDrvPort ix, unsigned int* key,
 		  void (*async_invoke)(void*), void* async_data,
 		  void (*async_free)(void*))
 {
-    ErlAsync* a = (ErlAsync*) erts_alloc(ERTS_ALC_T_ASYNC, sizeof(ErlAsync));
-    Port* prt = erts_drvport2port(ix);
+    ErtsAsync* a;
+    Port* prt;
     long id;
     unsigned int qix;
+#if ERTS_USE_ASYNC_READY_Q
+    Uint sched_id;
 
+    sched_id = erts_get_scheduler_id();
+    if (!sched_id)
+	sched_id = 1;
+#endif
 
+    prt = erts_drvport2port(ix);
     if (!prt)
 	return -1;
 
     ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
 
-    a->next = NULL;
-    a->prev = NULL;
+    a = (ErtsAsync*) erts_alloc(ERTS_ALC_T_ASYNC, sizeof(ErtsAsync));
+
+#if ERTS_USE_ASYNC_READY_Q
+    a->sched_id = sched_id;
+#endif
     a->hndl = (DE_Handle*)prt->drv_ptr->handle;
     a->port = prt->id;
     a->pdl = NULL;
@@ -413,12 +584,16 @@ long driver_async(ErlDrvPort ix, unsigned int* key,
     a->async_invoke = async_invoke;
     a->async_free = async_free;
 
-    erts_smp_spin_lock(&async_id_lock);
-    async_id = (async_id + 1) & 0x7fffffff;
-    if (async_id == 0)
-	async_id++;
-    id = async_id;
-    erts_smp_spin_unlock(&async_id_lock);
+    if (!async)
+	id = 0;
+    else {
+	do {
+	    id = erts_atomic_inc_read_nob(&async->init.data.id);
+	} while (id == 0);
+	if (id < 0)
+	    id *= -1;
+	ASSERT(id > 0);
+    }
 
     a->async_id = id;
 
@@ -437,7 +612,7 @@ long driver_async(ErlDrvPort ix, unsigned int* key,
 	    driver_pdl_inc_refc(prt->port_data_lock);
 	    a->pdl = prt->port_data_lock;
 	}
-	async_add(a, &async_q[qix]);
+	async_add(a, async_q(qix));
 	return id;
     }
 #endif
@@ -455,10 +630,16 @@ long driver_async(ErlDrvPort ix, unsigned int* key,
 
 int driver_async_cancel(unsigned int id)
 {
-#ifdef USE_THREADS
-    if (erts_async_max_threads > 0)
-	return async_del(id);
-#endif
+    /*
+     * Not supported anymore. Always fail (which is backward
+     * compatible).
+     *
+     * This functionality could be implemented again. However,
+     * it is (and always has been) completely useless since
+     * it doesn't give you any guarantees whatsoever. The user
+     * needs to (and always have had to) synchronize in his/her
+     * own code in order to get any guarantees.
+     */
     return 0;
 }
 
diff --git a/erts/emulator/beam/erl_async.h b/erts/emulator/beam/erl_async.h
new file mode 100644
index 0000000000..95374a8fc9
--- /dev/null
+++ b/erts/emulator/beam/erl_async.h
@@ -0,0 +1,66 @@
+/*
+ * %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%
+ */
+
+#ifndef ERL_ASYNC_H__
+#define ERL_ASYNC_H__
+
+#define ERTS_MAX_NO_OF_ASYNC_THREADS 1024
+extern int erts_async_max_threads;
+#define ERTS_ASYNC_THREAD_MIN_STACK_SIZE 16	/* Kilo words */
+#define ERTS_ASYNC_THREAD_MAX_STACK_SIZE 8192	/* Kilo words */
+extern int erts_async_thread_suggested_stack_size;
+
+#ifdef USE_THREADS
+
+#ifdef ERTS_SMP
+/*
+ * With smp support we can choose to have, or not to
+ * have an async ready queue.
+ */
+#define ERTS_USE_ASYNC_READY_Q 1
+#endif
+
+#ifndef ERTS_SMP
+/* In non-smp case we *need* the async ready queue */
+#  undef ERTS_USE_ASYNC_READY_Q
+#  define ERTS_USE_ASYNC_READY_Q 1
+#endif
+
+#ifndef ERTS_USE_ASYNC_READY_Q
+#  define ERTS_USE_ASYNC_READY_Q 0
+#endif
+
+#if ERTS_USE_ASYNC_READY_Q
+int erts_check_async_ready(void *);
+int erts_async_ready_clean(void *, void *);
+void *erts_get_async_ready_queue(Uint sched_id);
+#define ERTS_ASYNC_READY_CLEAN 0
+#define ERTS_ASYNC_READY_DIRTY 1
+#ifdef ERTS_SMP
+#define ERTS_ASYNC_READY_NEED_THR_PRGR 2
+#endif
+#endif /* ERTS_USE_ASYNC_READY_Q */
+
+#endif /* USE_THREADS */
+
+void erts_init_async(void);
+void erts_exit_flush_async(void);
+
+
+#endif /* ERL_ASYNC_H__ */
diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index 58eb58d1dc..d4d5691f62 100644
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -39,6 +39,7 @@
 #include "dist.h"
 #include "erl_gc.h"
 #include "erl_cpu_topology.h"
+#include "erl_async.h"
 #include "erl_thr_progress.h"
 #ifdef HIPE
 #include "hipe_arch.h"
diff --git a/erts/emulator/beam/erl_driver.h b/erts/emulator/beam/erl_driver.h
index 401967a8de..ae0c9def90 100644
--- a/erts/emulator/beam/erl_driver.h
+++ b/erts/emulator/beam/erl_driver.h
@@ -28,6 +28,14 @@
 #  include "config.h"
 #endif
 
+#define ERL_DRV_DEPRECATED_FUNC
+#ifdef __GNUC__
+#  if __GNUC__ >= 3
+#    undef ERL_DRV_DEPRECATED_FUNC
+#    define ERL_DRV_DEPRECATED_FUNC __attribute__((deprecated))
+#  endif
+#endif
+
 #ifdef SIZEOF_CHAR
 #  define SIZEOF_CHAR_SAVED__ SIZEOF_CHAR
 #  undef SIZEOF_CHAR
@@ -582,8 +590,11 @@ EXTERN long driver_async(ErlDrvPort ix,
 			 void* async_data,
 			 void (*async_free)(void*));
 
-
-EXTERN int driver_async_cancel(unsigned int key);
+/*
+ * driver_async_cancel() is deprecated. It is scheduled for removal
+ * in OTP-R16. For more information see the erl_driver(3) documentation.
+ */
+EXTERN int driver_async_cancel(unsigned int key) ERL_DRV_DEPRECATED_FUNC;
 
 /* Locks the driver in the machine "forever", there is
    no unlock function. Note that this is almost never useful, as an open
diff --git a/erts/emulator/beam/erl_init.c b/erts/emulator/beam/erl_init.c
index 5fe44afdce..98b997583c 100644
--- a/erts/emulator/beam/erl_init.c
+++ b/erts/emulator/beam/erl_init.c
@@ -44,6 +44,7 @@
 #include "erl_cpu_topology.h"
 #include "erl_thr_progress.h"
 #include "erl_thr_queue.h"
+#include "erl_async.h"
 
 #ifdef HIPE
 #include "hipe_mode_switch.h"	/* for hipe_mode_switch_init() */
@@ -101,8 +102,6 @@ int erts_backtrace_depth;	/* How many functions to show in a backtrace
 				 * in error codes.
 				 */
 
-int erts_async_max_threads;  /* number of threads for async support */
-int erts_async_thread_suggested_stack_size;
 erts_smp_atomic32_t erts_max_gen_gcs;
 
 Eterm erts_error_logger_warnings; /* What to map warning logs to, am_error, 
@@ -279,6 +278,7 @@ erl_init(int ncpu)
     erts_init_node_tables();
     init_dist();
     erl_drv_thr_init();
+    erts_init_async();
     init_io();
     init_copy();
     init_load();
@@ -605,6 +605,8 @@ early_init(int *argc, char **argv) /*
     int max_main_threads;
     int max_reader_groups;
     int reader_groups;
+    char envbuf[21]; /* enough for any 64-bit integer */
+    size_t envbufsz;
 
     use_multi_run_queue = 1;
     erts_printf_eterm_func = erts_printf_term;
@@ -676,6 +678,16 @@ early_init(int *argc, char **argv) /*
     schdlrs = no_schedulers;
     schdlrs_onln = no_schedulers_online;
 
+    envbufsz = sizeof(envbuf);
+
+    /* erts_sys_getenv() not initialized yet; need erts_sys_getenv__() */
+    if (erts_sys_getenv__("ERL_THREAD_POOL_SIZE", envbuf, &envbufsz) == 0)
+	erts_async_max_threads = atoi(envbuf);
+    else
+	erts_async_max_threads = 0;
+    if (erts_async_max_threads > ERTS_MAX_NO_OF_ASYNC_THREADS)
+	erts_async_max_threads = ERTS_MAX_NO_OF_ASYNC_THREADS;
+
     if (argc && argv) {
 	int i = 1;
 	while (i < *argc) {
@@ -703,6 +715,20 @@ early_init(int *argc, char **argv) /*
 		    }
 		    break;
 		}
+		case 'A': {
+		    /* set number of threads in thread pool */
+		    char *arg = get_arg(argv[i]+2, argv[i+1], &i);
+		    if (((erts_async_max_threads = atoi(arg)) < 0) ||
+			(erts_async_max_threads > ERTS_MAX_NO_OF_ASYNC_THREADS)) {
+			erts_fprintf(stderr,
+				     "bad number of async threads %s\n",
+				     arg);
+			erts_usage();
+			VERBOSE(DEBUG_SYSTEM, ("using %d async-threads\n",
+					       erts_async_max_threads));
+		    }
+		    break;
+		}
 		case 'S' : {
 		    int tot, onln;
 		    char *arg = get_arg(argv[i]+2, argv[i+1], &i);
@@ -783,9 +809,12 @@ early_init(int *argc, char **argv) /*
      * ** Aux thread (see erl_process.c)
      * ** Sys message dispatcher thread (see erl_trace.c)
      *
-     * * No unmanaged threads that need to register.
+     * * Unmanaged threads that need to register:
+     * ** Async threads (see erl_async.c)
      */
-    erts_thr_progress_init(no_schedulers, no_schedulers+1, 0);
+    erts_thr_progress_init(no_schedulers,
+			   no_schedulers+2,
+			   erts_async_max_threads);
 #endif
     erts_thr_q_init();
     erts_init_utils();
@@ -867,7 +896,6 @@ erl_start(int argc, char **argv)
     int have_break_handler = 1;
     char envbuf[21]; /* enough for any 64-bit integer */
     size_t envbufsz;
-    int async_max_threads = erts_async_max_threads;
     int ncpu = early_init(&argc, argv);
 
     envbufsz = sizeof(envbuf);
@@ -883,11 +911,6 @@ erl_start(int argc, char **argv)
 				  (erts_aint32_t) max_gen_gcs);
     }
 
-    envbufsz = sizeof(envbuf);
-    if (erts_sys_getenv("ERL_THREAD_POOL_SIZE", envbuf, &envbufsz) == 0) {
-	async_max_threads = atoi(envbuf);
-    }
-
 #if (defined(__APPLE__) && defined(__MACH__)) || defined(__DARWIN__)
     /*
      * The default stack size on MacOS X is too small for pcre.
@@ -1315,17 +1338,8 @@ erl_start(int argc, char **argv)
 	    break;
 	}
 
-	case 'A':
-	    /* set number of threads in thread pool */
-	    arg = get_arg(argv[i]+2, argv[i+1], &i);
-	    if (((async_max_threads = atoi(arg)) < 0) ||
-		(async_max_threads > ERTS_MAX_NO_OF_ASYNC_THREADS)) {
-		erts_fprintf(stderr, "bad number of async threads %s\n", arg);
-		erts_usage();
-	    }
-
-	    VERBOSE(DEBUG_SYSTEM, ("using %d async-threads\n",
-				   async_max_threads));
+	case 'A': /* Was handled in early init just read past it */
+	    (void) get_arg(argv[i]+2, argv[i+1], &i);
 	    break;
 
 	case 'a':
@@ -1414,10 +1428,6 @@ erl_start(int argc, char **argv)
 	i++;
     }
 
-#ifdef USE_THREADS
-    erts_async_max_threads = async_max_threads;
-#endif
-
     /* Delayed check of +P flag */
     if (erts_max_processes < ERTS_MIN_PROCESSES
 	|| erts_max_processes > ERTS_MAX_PROCESSES
@@ -1463,6 +1473,10 @@ erl_start(int argc, char **argv)
     erts_sys_main_thread(); /* May or may not return! */
 #else
     erts_thr_set_main_status(1, 1);
+#if ERTS_USE_ASYNC_READY_Q
+    erts_get_scheduler_data()->aux_work_data.async_ready.queue
+	= erts_get_async_ready_queue(1);
+#endif
     set_main_stack_size();
     process_main();
 #endif
@@ -1535,14 +1549,7 @@ system_cleanup(int exit_code)
     erts_cleanup_incgc();
 #endif
 
-#if defined(USE_THREADS)
-    exit_async();
-#endif
-
-    /*
-     * A lot more cleaning could/should have been done...
-     */
-
+    erts_exit_flush_async();
 }
 
 /*
diff --git a/erts/emulator/beam/erl_lock_check.c b/erts/emulator/beam/erl_lock_check.c
index 633be0ef58..44da6b6c51 100644
--- a/erts/emulator/beam/erl_lock_check.c
+++ b/erts/emulator/beam/erl_lock_check.c
@@ -110,10 +110,6 @@ static erts_lc_lock_order_t erts_lock_order[] = {
     {	"fun_tab",				NULL			},
     {	"environ",				NULL			},
 #endif
-    {	"asyncq",				"address"		},
-#ifndef ERTS_SMP
-    {	"async_ready",				NULL			},
-#endif
     {	"efile_drv",				"address"		},
 #if defined(ENABLE_CHILD_WAITER_THREAD) || defined(ERTS_SMP)
     {	"child_status",				NULL			},
@@ -138,6 +134,7 @@ static erts_lc_lock_order_t erts_lock_order[] = {
     {	"alcu_init_atoms",			NULL			},
     {	"mseg_init_atoms",			NULL			},
     {	"drv_tsd",				NULL			},
+    {	"async_enq_mtx",			NULL			},
 #ifdef ERTS_SMP
     {	"sys_msg_q", 				NULL			},
     {	"atom_tab",				NULL			},
@@ -173,12 +170,12 @@ static erts_lc_lock_order_t erts_lock_order[] = {
     {	"timeofday",				NULL			},
     {	"breakpoints",				NULL			},
     {	"pollsets_lock",			NULL			},
-    {	"async_id",				NULL			},
     {	"pix_lock",				"address"		},
     {	"run_queues_lists",			NULL			},
     {	"sched_stat",				NULL			},
     {	"run_queue_sleep_list",			"address"		},
 #endif
+    {	"async_init_mtx",			NULL			},
 #ifdef ERTS_SMP
     {	"proc_lck_qs_alloc",			NULL 			},
 #endif
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 4292522fba..9db2e874c6 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -41,6 +41,7 @@
 #include "erl_cpu_topology.h"
 #include "erl_thr_progress.h"
 #include "erl_thr_queue.h"
+#include "erl_async.h"
 
 #define ERTS_RUNQ_CHECK_BALANCE_REDS_PER_SCHED (2000*CONTEXT_REDS)
 #define ERTS_RUNQ_CALL_CHECK_BALANCE_REDS \
@@ -364,6 +365,12 @@ dbg_chk_aux_work_val(erts_aint32_t value)
 #ifdef ERTS_SSI_AUX_WORK_MISC_THR_PRGR
     valid |= ERTS_SSI_AUX_WORK_MISC_THR_PRGR;
 #endif
+#ifdef ERTS_SSI_AUX_WORK_ASYNC_READY
+    valid |= ERTS_SSI_AUX_WORK_ASYNC_READY;
+#endif
+#ifdef ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN
+    valid |= ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN;
+#endif
 
 #ifdef ERTS_SSI_AUX_WORK_FIX_ALLOC_LOWER_LIM
     valid |= ERTS_SSI_AUX_WORK_FIX_ALLOC_LOWER_LIM;
@@ -870,6 +877,73 @@ erts_schedule_multi_misc_aux_work(int ignore_self,
    }
 }
 
+#if ERTS_USE_ASYNC_READY_Q
+
+void
+erts_notify_check_async_ready_queue(void *vno)
+{
+    int ix = ((int) (SWord) vno) -1;
+    set_aux_work_flags_wakeup_nob(ERTS_SCHED_SLEEP_INFO_IX(ix),
+				  ERTS_SSI_AUX_WORK_ASYNC_READY);
+}
+
+static erts_aint32_t
+handle_async_ready(ErtsAuxWorkData *awdp,
+		   erts_aint32_t aux_work)
+{
+    ErtsSchedulerSleepInfo *ssi = awdp->ssi;
+    unset_aux_work_flags(ssi, ERTS_SSI_AUX_WORK_ASYNC_READY);
+    if (erts_check_async_ready(awdp->async_ready.queue)) {
+	if (set_aux_work_flags(ssi, ERTS_SSI_AUX_WORK_ASYNC_READY)
+	    & ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN) {
+	    unset_aux_work_flags(ssi, ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN);
+	    aux_work &= ~ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN;
+	}
+	return aux_work;
+    }
+#ifdef ERTS_SMP
+    awdp->async_ready.need_thr_prgr = 0;
+#endif
+    set_aux_work_flags(ssi, ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN);
+    return ((aux_work & ~ERTS_SSI_AUX_WORK_ASYNC_READY)
+	    | ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN);
+}
+
+static erts_aint32_t
+handle_async_ready_clean(ErtsAuxWorkData *awdp,
+			 erts_aint32_t aux_work)
+{
+    void *thr_prgr_p;
+
+#ifdef ERTS_SMP
+    if (awdp->async_ready.need_thr_prgr
+	&& !erts_thr_progress_has_reached(awdp->misc.thr_prgr)) {
+	return aux_work & ~ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN;
+    }
+
+    awdp->async_ready.need_thr_prgr = 0;
+    thr_prgr_p = (void *) &awdp->async_ready.thr_prgr;
+#else
+    thr_prgr_p = NULL;
+#endif
+
+    switch (erts_async_ready_clean(awdp->async_ready.queue, thr_prgr_p)) {
+    case ERTS_ASYNC_READY_CLEAN:
+	unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN);
+	return aux_work & ~ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN;
+#ifdef ERTS_SMP
+    case ERTS_ASYNC_READY_NEED_THR_PRGR:
+	erts_thr_progress_wakeup(awdp->esdp,
+				 awdp->async_ready.thr_prgr);
+	awdp->async_ready.need_thr_prgr = 1;
+#endif
+    default:
+	return aux_work;
+    }
+}
+
+#endif
+
 static erts_aint32_t
 handle_fix_alloc(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
 {
@@ -1108,6 +1182,16 @@ handle_aux_work(ErtsAuxWorkData *awdp, erts_aint32_t aux_work)
 	aux_work = handle_misc_aux_work(awdp, aux_work);
 	ERTS_DBG_CHK_AUX_WORK_VAL(aux_work);
     }
+#if ERTS_USE_ASYNC_READY_Q
+    if (aux_work & ERTS_SSI_AUX_WORK_ASYNC_READY) {
+	aux_work = handle_async_ready(awdp, aux_work);
+	ERTS_DBG_CHK_AUX_WORK_VAL(aux_work);
+    }
+    if (aux_work & ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN) {
+	aux_work = handle_async_ready_clean(awdp, aux_work);
+	ERTS_DBG_CHK_AUX_WORK_VAL(aux_work);
+    }
+#endif
 #ifdef ERTS_SMP_SCHEDULERS_NEED_TO_CHECK_CHILDREN
     if (aux_work & ERTS_SSI_AUX_WORK_CHECK_CHILDREN) {
 	aux_work = handle_check_children(awdp, aux_work);
@@ -3237,6 +3321,13 @@ init_aux_work_data(ErtsAuxWorkData *awdp, ErtsSchedulerData *esdp)
     awdp->dd.completed_callback = NULL;
     awdp->dd.completed_arg = NULL;
 #endif
+#ifdef ERTS_USE_ASYNC_READY_Q
+#ifdef ERTS_SMP
+    awdp->async_ready.need_thr_prgr = 0;
+    awdp->async_ready.thr_prgr = ERTS_THR_PRGR_VAL_WAITING;
+#endif
+    awdp->async_ready.queue = NULL;
+#endif
 }
 
 void
@@ -4448,6 +4539,9 @@ sched_thread_func(void *vesdp)
 #if HAVE_ERTS_MSEG
     erts_mseg_late_init();
 #endif
+#if ERTS_USE_ASYNC_READY_Q
+    esdp->aux_work_data.async_ready.queue = erts_get_async_ready_queue(no);
+#endif
 
     erts_sched_init_check_cpu_bind(esdp);
 
diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 895f5ae3c0..0c8204b4ce 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -54,6 +54,7 @@ typedef struct process Process;
 #include "erl_atom_table.h"
 #include "external.h"
 #include "erl_mseg.h"
+#include "erl_async.h"
 
 #ifdef HIPE
 #include "hipe_process.h"
@@ -256,11 +257,13 @@ typedef enum {
 #endif
 #define ERTS_SSI_AUX_WORK_FIX_ALLOC_LOWER_LIM	(((erts_aint32_t) 1) << 4)
 #define ERTS_SSI_AUX_WORK_FIX_ALLOC_DEALLOC	(((erts_aint32_t) 1) << 5)
+#define ERTS_SSI_AUX_WORK_ASYNC_READY		(((erts_aint32_t) 1) << 6)
+#define ERTS_SSI_AUX_WORK_ASYNC_READY_CLEAN	(((erts_aint32_t) 1) << 7)
 #ifdef ERTS_SMP
-#define ERTS_SSI_AUX_WORK_DD			(((erts_aint32_t) 1) << 6)
-#define ERTS_SSI_AUX_WORK_DD_THR_PRGR		(((erts_aint32_t) 1) << 7)
+#define ERTS_SSI_AUX_WORK_DD			(((erts_aint32_t) 1) << 8)
+#define ERTS_SSI_AUX_WORK_DD_THR_PRGR		(((erts_aint32_t) 1) << 9)
 #endif
-#define ERTS_SSI_AUX_WORK_MSEG_CACHE_CHECK	(((erts_aint32_t) 1) << 8)
+#define ERTS_SSI_AUX_WORK_MSEG_CACHE_CHECK	(((erts_aint32_t) 1) << 10)
 
 #if !HAVE_ERTS_MSEG
 #  undef ERTS_SSI_AUX_WORK_MSEG_CACHE_CHECK
@@ -418,6 +421,15 @@ typedef struct {
 	void (*completed_arg)(void *);
     } dd;
 #endif
+#ifdef ERTS_USE_ASYNC_READY_Q
+    struct {
+#ifdef ERTS_SMP
+	int need_thr_prgr;
+	ErtsThrPrgrVal thr_prgr;
+#endif
+	void *queue;
+    } async_ready;
+#endif
 } ErtsAuxWorkData;
 
 struct ErtsSchedulerData_ {
@@ -1101,6 +1113,9 @@ void erts_start_schedulers(void);
 void erts_alloc_notify_delayed_dealloc(int);
 void erts_smp_notify_check_children_needed(void);
 #endif
+#if ERTS_USE_ASYNC_READY_Q
+void erts_notify_check_async_ready_queue(void *);
+#endif
 void erts_schedule_misc_aux_work(int sched_id,
 				 void (*func)(void *),
 				 void *arg);
diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h
index 6687e02485..ba0b96870e 100644
--- a/erts/emulator/beam/global.h
+++ b/erts/emulator/beam/global.h
@@ -41,12 +41,6 @@
 typedef struct port Port;
 #include "erl_port_task.h"
 
-#define ERTS_MAX_NO_OF_ASYNC_THREADS 1024
-extern int erts_async_max_threads;
-#define ERTS_ASYNC_THREAD_MIN_STACK_SIZE 16	/* Kilo words */
-#define ERTS_ASYNC_THREAD_MAX_STACK_SIZE 8192	/* Kilo words */
-extern int erts_async_thread_suggested_stack_size;
-
 typedef struct erts_driver_t_ erts_driver_t;
 
 #define SMALL_IO_QUEUE 5   /* Number of fixed elements */
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 151c776a3d..fff720634d 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -42,6 +42,7 @@
 #include "erl_bits.h"
 #include "erl_version.h"
 #include "error.h"
+#include "erl_async.h"
 
 extern ErlDrvEntry fd_driver_entry;
 extern ErlDrvEntry vanilla_driver_entry;
@@ -4579,7 +4580,10 @@ int driver_lock_driver(ErlDrvPort ix)
 
     erts_smp_mtx_lock(&erts_driver_list_lock);
 
-    if (prt == NULL) return -1;
+    if (prt == NULL) {
+	erts_smp_mtx_unlock(&erts_driver_list_lock);
+	return -1;
+    }
 
     ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
     if ((dh = (DE_Handle*)prt->drv_ptr->handle ) == NULL) {
diff --git a/erts/emulator/beam/sys.h b/erts/emulator/beam/sys.h
index b63fe98f27..f9cbcc5892 100644
--- a/erts/emulator/beam/sys.h
+++ b/erts/emulator/beam/sys.h
@@ -475,15 +475,6 @@ __decl_noreturn void __noreturn erl_exit(int n, char*, ...);
 #define ERTS_ABORT_EXIT	(INT_MIN + 1)	/* no crash dump; only abort() */
 #define ERTS_DUMP_EXIT	(127)		/* crash dump; then exit() */
 
-
-#ifndef ERTS_SMP
-int check_async_ready(void);
-#ifdef USE_THREADS
-void sys_async_ready(int hndl);
-int erts_register_async_ready_callback(void (*funcp)(void));
-#endif
-#endif
-
 Eterm erts_check_io_info(void *p);
 
 /* Size of misc memory allocated from system dependent code */
@@ -671,6 +662,8 @@ int erts_sys_putenv(char *key_value, int sep_ix);
    *size), a value > 0 if value buffer is too small (*size is set to needed
    size), and a value < 0 on failure. */
 int erts_sys_getenv(char *key, char *value, size_t *size);
+/* erts_sys_getenv__() is only allowed to be used in early init phase */
+int erts_sys_getenv__(char *key, char *value, size_t *size);
 
 /* Easier to use, but not as efficient, environment functions */
 char *erts_read_env(char *key);