aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator/beam/erl_async.c
diff options
context:
space:
mode:
Diffstat (limited to 'erts/emulator/beam/erl_async.c')
-rw-r--r--erts/emulator/beam/erl_async.c469
1 files changed, 469 insertions, 0 deletions
diff --git a/erts/emulator/beam/erl_async.c b/erts/emulator/beam/erl_async.c
new file mode 100644
index 0000000000..b090564649
--- /dev/null
+++ b/erts/emulator/beam/erl_async.c
@@ -0,0 +1,469 @@
+/*
+ * %CopyrightBegin%
+ *
+ * Copyright Ericsson AB 2000-2009. 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 HAVE_CONFIG_H
+# include "config.h"
+#endif
+
+#include "sys.h"
+#include "erl_sys_driver.h"
+#include "global.h"
+#include "erl_threads.h"
+
+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;
+ void* async_data;
+ ErlDrvPDL pdl;
+ void (*async_invoke)(void*);
+ void (*async_free)(void*);
+} ErlAsync;
+
+typedef struct {
+ erts_mtx_t mtx;
+ erts_cnd_t cv;
+ erts_tid_t thr;
+ int len;
+#ifndef ERTS_SMP
+ int hndl;
+#endif
+ ErlAsync* head;
+ ErlAsync* tail;
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ int no;
+#endif
+} AsyncQueue;
+
+static erts_smp_spinlock_t async_id_lock;
+static long async_id = 0;
+
+
+#ifndef ERTS_SMP
+
+erts_mtx_t async_ready_mtx;
+static ErlAsync* async_ready_list = NULL;
+
+#endif
+
+/*
+** Initialize worker threads (if supported)
+*/
+
+/* Detach from driver */
+static void async_detach(DE_Handle* dh)
+{
+ /* XXX:PaN what should happen here? we want to unload the driver or??? */
+ return;
+}
+
+
+#ifdef USE_THREADS
+
+static AsyncQueue* async_q;
+
+static void* async_main(void*);
+static void async_add(ErlAsync*, AsyncQueue*);
+
+#ifndef ERTS_SMP
+typedef struct ErtsAsyncReadyCallback_ ErtsAsyncReadyCallback;
+struct ErtsAsyncReadyCallback_ {
+ struct ErtsAsyncReadyCallback_ *next;
+ void (*callback)(void);
+};
+
+static ErtsAsyncReadyCallback *callbacks;
+static int async_handle;
+
+int erts_register_async_ready_callback(void (*funcp)(void))
+{
+ 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;
+}
+#endif
+
+int init_async(int hndl)
+{
+ erts_thr_opts_t thr_opts = ERTS_THR_OPTS_DEFAULT_INITER;
+ AsyncQueue* q;
+ int i;
+
+ 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;
+}
+
+
+int exit_async()
+{
+ int i;
+
+ /* 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]);
+ }
+
+ 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);
+#endif
+ if (async_q)
+ erts_free(ERTS_ALC_T_ASYNC_Q, (void *) async_q);
+ return 0;
+}
+
+
+static void async_add(ErlAsync* a, AsyncQueue* q)
+{
+ /* XXX:PaN Is this still necessary when ports lock drivers? */
+ if (is_internal_port(a->port)) {
+ ERTS_LC_ASSERT(erts_drvportid2port(a->port));
+ /* make sure the driver will stay around */
+ driver_lock_driver(internal_port_index(a->port));
+ }
+
+ erts_mtx_lock(&q->mtx);
+
+ 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);
+}
+
+static ErlAsync* async_get(AsyncQueue* q)
+{
+ ErlAsync* a;
+
+ erts_mtx_lock(&q->mtx);
+ while((a = q->tail) == NULL) {
+ erts_cnd_wait(&q->cv, &q->mtx);
+ }
+#ifdef ERTS_SMP
+ ASSERT(a && q->tail == a);
+#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--;
+ }
+ erts_mtx_unlock(&q->mtx);
+ return a;
+}
+
+
+static int async_del(long id)
+{
+ 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;
+ }
+ }
+ erts_mtx_unlock(&async_q[i].mtx);
+ }
+ return 0;
+}
+
+static void* async_main(void* arg)
+{
+ AsyncQueue* q = (AsyncQueue*) arg;
+
+#ifdef ERTS_ENABLE_LOCK_CHECK
+ {
+ char buf[27];
+ erts_snprintf(&buf[0], 27, "async %d", q->no);
+ erts_lc_set_thread_name(&buf[0]);
+ }
+#endif
+
+ while(1) {
+ ErlAsync* a = async_get(q);
+
+ 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 */
+#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);
+#endif
+ }
+ }
+
+ return NULL;
+}
+
+
+#endif
+
+#ifndef ERTS_SMP
+
+int check_async_ready(void)
+{
+#ifdef USE_THREADS
+ ErtsAsyncReadyCallback *cbs;
+#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);
+ }
+ erts_free(ERTS_ALC_T_ASYNC, (void *) a);
+ a = a_next;
+ }
+#ifdef USE_THREADS
+ for (; cbs; cbs = cbs->next)
+ (*cbs->callback)();
+#endif
+ return count;
+}
+
+#endif
+
+
+/*
+** Schedule async_invoke on a worker thread
+** NOTE will be syncrounous when threads are unsupported
+** return values:
+** 0 completed
+** -1 error
+** N handle value (used with async_cancel)
+** arguments:
+** ix driver index
+** key pointer to secedule queue (NULL means round robin)
+** async_invoke function to run in thread
+** async_data data to pass to invoke function
+** async_free function for relase async_data in case of failure
+*/
+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);
+ long id;
+ unsigned int qix;
+
+
+ if (!prt)
+ return -1;
+
+ ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+
+ a->next = NULL;
+ a->prev = NULL;
+ a->hndl = (DE_Handle*)prt->drv_ptr->handle;
+ a->port = prt->id;
+ a->pdl = NULL;
+ a->async_data = async_data;
+ 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);
+
+ a->async_id = id;
+
+ if (key == NULL) {
+ qix = (erts_async_max_threads > 0)
+ ? (id % erts_async_max_threads) : 0;
+ }
+ else {
+ qix = (erts_async_max_threads > 0) ?
+ (*key % erts_async_max_threads) : 0;
+ *key = qix;
+ }
+#ifdef USE_THREADS
+ if (erts_async_max_threads > 0) {
+ if (prt->port_data_lock) {
+ driver_pdl_inc_refc(prt->port_data_lock);
+ a->pdl = prt->port_data_lock;
+ }
+ async_add(a, &async_q[qix]);
+ return id;
+ }
+#endif
+
+ (*a->async_invoke)(a->async_data);
+
+ if (async_ready(prt, a->async_data)) {
+ if (a->async_free != NULL)
+ (*a->async_free)(a->async_data);
+ }
+ erts_free(ERTS_ALC_T_ASYNC, (void *) a);
+
+ return id;
+}
+
+int driver_async_cancel(unsigned int id)
+{
+#ifdef USE_THREADS
+ if (erts_async_max_threads > 0)
+ return async_del(id);
+#endif
+ return 0;
+}
+
+
+
+
+
generated by cgit v1.2.3 (git 2.25.1) at 2024-06-28 05:07:19 +0000