/*
 * %CopyrightBegin%
 *
 * Copyright Ericsson AB 2000-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%
 */
#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)
{
    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)
{
    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;
	    }
	    a = a->next;
	}
	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;
}