//
// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
// file was obtained (LICENSE.txt).  A copy of the license may also be
// found online at https://opensource.org/licenses/MIT.
//

#include "core/nng_impl.h"

typedef struct nni_taskq_thr nni_taskq_thr;
struct nni_taskq_thr {
	nni_taskq *tqt_tq;
	nni_thr    tqt_thread;
};
struct nni_taskq {
	nni_list       tq_tasks;
	nni_mtx        tq_mtx;
	nni_cv         tq_sched_cv;
	nni_cv         tq_wait_cv;
	nni_taskq_thr *tq_threads;
	int            tq_nthreads;
	bool           tq_run;
};

static nni_taskq *nni_taskq_systq = NULL;

static void
nni_taskq_thread(void *self)
{
	nni_taskq_thr *thr = self;
	nni_taskq *    tq  = thr->tqt_tq;
	nni_task *     task;

        nni_thr_set_name(NULL, "nng:task");

        nni_mtx_lock(&tq->tq_mtx);
	for (;;) {
		if ((task = nni_list_first(&tq->tq_tasks)) != NULL) {

			nni_list_remove(&tq->tq_tasks, task);

			nni_mtx_unlock(&tq->tq_mtx);

			task->task_cb(task->task_arg);

			nni_mtx_lock(&task->task_mtx);
			task->task_busy--;
			if (task->task_busy == 0) {
				nni_cv_wake(&task->task_cv);
			}
			nni_mtx_unlock(&task->task_mtx);

			nni_mtx_lock(&tq->tq_mtx);

			continue;
		}

		if (!tq->tq_run) {
			break;
		}
		nni_cv_wait(&tq->tq_sched_cv);
	}
	nni_mtx_unlock(&tq->tq_mtx);
}

int
nni_taskq_init(nni_taskq **tqp, int nthr)
{
	nni_taskq *tq;

	if ((tq = NNI_ALLOC_STRUCT(tq)) == NULL) {
		return (NNG_ENOMEM);
	}
	if ((tq->tq_threads = NNI_ALLOC_STRUCTS(tq->tq_threads, nthr)) ==
	    NULL) {
		NNI_FREE_STRUCT(tq);
		return (NNG_ENOMEM);
	}
	tq->tq_nthreads = nthr;
	NNI_LIST_INIT(&tq->tq_tasks, nni_task, task_node);

	nni_mtx_init(&tq->tq_mtx);
	nni_cv_init(&tq->tq_sched_cv, &tq->tq_mtx);
	nni_cv_init(&tq->tq_wait_cv, &tq->tq_mtx);

	for (int i = 0; i < nthr; i++) {
		int rv;
		tq->tq_threads[i].tqt_tq = tq;
		rv = nni_thr_init(&tq->tq_threads[i].tqt_thread,
		    nni_taskq_thread, &tq->tq_threads[i]);
		if (rv != 0) {
			nni_taskq_fini(tq);
			return (rv);
		}
	}
	tq->tq_run = true;
	for (int i = 0; i < tq->tq_nthreads; i++) {
		nni_thr_run(&tq->tq_threads[i].tqt_thread);
	}
	*tqp = tq;
	return (0);
}

void
nni_taskq_fini(nni_taskq *tq)
{
	if (tq == NULL) {
		return;
	}
	if (tq->tq_run) {
		nni_mtx_lock(&tq->tq_mtx);
		tq->tq_run = false;
		nni_cv_wake(&tq->tq_sched_cv);
		nni_mtx_unlock(&tq->tq_mtx);
	}
	for (int i = 0; i < tq->tq_nthreads; i++) {
		nni_thr_fini(&tq->tq_threads[i].tqt_thread);
	}
	nni_cv_fini(&tq->tq_wait_cv);
	nni_cv_fini(&tq->tq_sched_cv);
	nni_mtx_fini(&tq->tq_mtx);
	NNI_FREE_STRUCTS(tq->tq_threads, tq->tq_nthreads);
	NNI_FREE_STRUCT(tq);
}

void
nni_task_exec(nni_task *task)
{
	nni_mtx_lock(&task->task_mtx);
	if (task->task_prep) {
		task->task_prep = false;
	} else {
		task->task_busy++;
	}
	nni_mtx_unlock(&task->task_mtx);

	if (task->task_cb != NULL) {
		task->task_cb(task->task_arg);
	}

	nni_mtx_lock(&task->task_mtx);
	task->task_busy--;
	if (task->task_busy == 0) {
		nni_cv_wake(&task->task_cv);
	}
	nni_mtx_unlock(&task->task_mtx);
}

void
nni_task_dispatch(nni_task *task)
{
	nni_taskq *tq = task->task_tq;

	// If there is no callback to perform, then do nothing!
	// The user will be none the wiser.
	if (task->task_cb == NULL) {
		nni_task_exec(task);
		return;
	}
	nni_mtx_lock(&task->task_mtx);
	if (task->task_prep) {
		task->task_prep = false;
	} else {
		task->task_busy++;
	}
	nni_mtx_unlock(&task->task_mtx);

	nni_mtx_lock(&tq->tq_mtx);
	nni_list_append(&tq->tq_tasks, task);
	nni_cv_wake1(&tq->tq_sched_cv); // waking just one waiter is adequate
	nni_mtx_unlock(&tq->tq_mtx);
}

void
nni_task_prep(nni_task *task)
{
	nni_mtx_lock(&task->task_mtx);
	task->task_busy++;
	task->task_prep = true;
	nni_mtx_unlock(&task->task_mtx);
}

void
nni_task_abort(nni_task *task)
{
	// This is called when unscheduling the task.
	nni_mtx_lock(&task->task_mtx);
	if (task->task_prep) {
		task->task_prep = false;
		task->task_busy--;
		if (task->task_busy == 0) {
			nni_cv_wake(&task->task_cv);
		}
	}
	nni_mtx_unlock(&task->task_mtx);
}
void
nni_task_wait(nni_task *task)
{
	nni_mtx_lock(&task->task_mtx);
	while (task->task_busy) {
		nni_cv_wait(&task->task_cv);
	}
	nni_mtx_unlock(&task->task_mtx);
}

void
nni_task_init(nni_task *task, nni_taskq *tq, nni_cb cb, void *arg)
{
	NNI_LIST_NODE_INIT(&task->task_node);
	nni_mtx_init(&task->task_mtx);
	nni_cv_init(&task->task_cv, &task->task_mtx);
	task->task_prep = false;
	task->task_busy = 0;
	task->task_cb   = cb;
	task->task_arg  = arg;
	task->task_tq   = tq != NULL ? tq : nni_taskq_systq;
}

void
nni_task_fini(nni_task *task)
{
	nni_mtx_lock(&task->task_mtx);
	while (task->task_busy) {
		nni_cv_wait(&task->task_cv);
	}
	nni_mtx_unlock(&task->task_mtx);
	nni_cv_fini(&task->task_cv);
	nni_mtx_fini(&task->task_mtx);
}

int
nni_taskq_sys_init(void)
{
	int nthrs;

#ifndef NNG_NUM_TASKQ_THREADS
	nthrs = nni_plat_ncpu() * 2;
#else
	nthrs = NNG_NUM_TASKQ_THREADS;
#endif
#if NNG_MAX_TASKQ_THREADS > 0
	if (nthrs > NNG_MAX_TASKQ_THREADS) {
		nthrs = NNG_MAX_TASKQ_THREADS;
	}
#endif

	return (nni_taskq_init(&nni_taskq_systq, nthrs));
}

void
nni_taskq_sys_fini(void)
{
	nni_taskq_fini(nni_taskq_systq);
	nni_taskq_systq = NULL;
}