/*

 * Copyright (C) Igor Sysoev

 * Copyright (C) NGINX, Inc.

 */

#include <nxt_main.h>

static nxt_int_t nxt_thread_pool_init(nxt_thread_pool_t *tp);

static void nxt_thread_pool_exit(nxt_task_t *task, void *obj, void *data);

static void nxt_thread_pool_start(void *ctx);

static void nxt_thread_pool_loop(void *ctx);

static void nxt_thread_pool_wait(nxt_thread_pool_t *tp);

nxt_thread_pool_t *

nxt_thread_pool_create(nxt_uint_t max_threads, nxt_nsec_t timeout,

    nxt_thread_pool_init_t init, nxt_event_engine_t *engine,

    nxt_work_handler_t exit)

{

    nxt_thread_pool_t  *tp;

    tp = nxt_zalloc(sizeof(nxt_thread_pool_t));

    if (tp == NULL) {

        return NULL;

    }

    tp->max_threads = max_threads;

    tp->timeout = timeout;

    tp->engine = engine;

    tp->task.thread = engine->task.thread;

    tp->task.log = engine->task.log;

    tp->init = init;

    tp->exit = exit;

    return tp;

}

nxt_int_t

nxt_thread_pool_post(nxt_thread_pool_t *tp, nxt_work_t *work)

{

    nxt_thread_log_debug("thread pool post");

    if (nxt_slow_path(nxt_thread_pool_init(tp) != NXT_OK)) {

        return NXT_ERROR;

    }

    nxt_locked_work_queue_add(&tp->work_queue, work);

    (void) nxt_sem_post(&tp->sem);

    return NXT_OK;

}

static nxt_int_t

nxt_thread_pool_init(nxt_thread_pool_t *tp)

{

    nxt_int_t            ret;

    nxt_thread_link_t    *link;

    nxt_thread_handle_t  handle;

    if (nxt_fast_path(tp->ready)) {

        return NXT_OK;

    }

    if (tp->max_threads == 0) {

        /* The pool is being destroyed. */

        return NXT_ERROR;

    }

    nxt_thread_spin_lock(&tp->work_queue.lock);

    ret = NXT_OK;

    if (!tp->ready) {

        nxt_thread_log_debug("thread pool init");

        (void) nxt_atomic_fetch_add(&tp->threads, 1);

        if (nxt_fast_path(nxt_sem_init(&tp->sem, 0) == NXT_OK)) {

            link = nxt_zalloc(sizeof(nxt_thread_link_t));

            if (nxt_fast_path(link != NULL)) {

                link->start = nxt_thread_pool_start;

                link->work.data = tp;

                if (nxt_thread_create(&handle, link) == NXT_OK) {

                    tp->ready = 1;

                    goto done;

                }

            }

            nxt_sem_destroy(&tp->sem);

        }

        (void) nxt_atomic_fetch_add(&tp->threads, -1);

        ret = NXT_ERROR;

    }

done:

    nxt_thread_spin_unlock(&tp->work_queue.lock);

    return ret;

}

static void

nxt_thread_pool_start(void *ctx)

{

    nxt_thread_t       *thr;

    nxt_thread_pool_t  *tp;

    tp = ctx;

    thr = nxt_thread();

    tp->main = thr->handle;

    tp->task.thread = thr;

    nxt_thread_pool_loop(ctx);

}

static void

nxt_thread_pool_loop(void *ctx)

{

    void                *obj, *data;

    nxt_task_t          *task;

    nxt_thread_t        *thr;

    nxt_thread_pool_t   *tp;

    nxt_work_handler_t  handler;

    tp = ctx;

    thr = nxt_thread();

    if (tp->init != NULL) {

        tp->init();

    }

    for ( ;; ) {

        nxt_thread_pool_wait(tp);

        handler = nxt_locked_work_queue_pop(&tp->work_queue, &task, &obj,

                                            &data);

        if (nxt_fast_path(handler != NULL)) {

            task->thread = thr;

            nxt_log_debug(thr->log, "locked work queue");

            handler(task, obj, data);

        }

        thr->log = &nxt_main_log;

    }

}

static void

nxt_thread_pool_wait(nxt_thread_pool_t *tp)

{

    nxt_err_t            err;

    nxt_thread_t         *thr;

    nxt_atomic_uint_t    waiting, threads;

    nxt_thread_link_t    *link;

    nxt_thread_handle_t  handle;

    thr = nxt_thread();

    nxt_log_debug(thr->log, "thread pool wait");

    (void) nxt_atomic_fetch_add(&tp->waiting, 1);

    for ( ;; ) {

        err = nxt_sem_wait(&tp->sem, tp->timeout);

        if (err == 0) {

            waiting = nxt_atomic_fetch_add(&tp->waiting, -1);

            break;

        }

        if (err == NXT_ETIMEDOUT) {

            if (nxt_thread_handle_equal(thr->handle, tp->main)) {

                continue;

            }

        }

        (void) nxt_atomic_fetch_add(&tp->waiting, -1);

        (void) nxt_atomic_fetch_add(&tp->threads, -1);

        nxt_thread_exit(thr);

        nxt_unreachable();

    }

    nxt_log_debug(thr->log, "thread pool awake, waiting: %A", waiting);

    if (waiting > 1) {

        return;

    }

    do {

        threads = tp->threads;

        if (threads >= tp->max_threads) {

            return;

        }

    } while (!nxt_atomic_cmp_set(&tp->threads, threads, threads + 1));

    link = nxt_zalloc(sizeof(nxt_thread_link_t));

    if (nxt_fast_path(link != NULL)) {

        link->start = nxt_thread_pool_loop;

        link->work.data = tp;

        if (nxt_thread_create(&handle, link) != NXT_OK) {

            (void) nxt_atomic_fetch_add(&tp->threads, -1);

        }

    }

}

void

nxt_thread_pool_destroy(nxt_thread_pool_t *tp)

{

    nxt_thread_t  *thr;

    thr = nxt_thread();

    nxt_log_debug(thr->log, "thread pool destroy: %A", tp->ready);

    if (!tp->ready) {

        nxt_work_queue_add(&thr->engine->fast_work_queue, tp->exit,

                           &tp->engine->task, tp, NULL);

        return;

    }

    if (tp->max_threads != 0) {

        /* Disable new threads creation and mark a pool as being destroyed. */

        tp->max_threads = 0;

        nxt_work_set(&tp->work, nxt_thread_pool_exit, &tp->task, tp, NULL);

        nxt_thread_pool_post(tp, &tp->work);

    }

}

/*

 * A thread handle (pthread_t) is either pointer or integer, so it can be

 * passed as work handler pointer "data" argument.  To convert void pointer

 * to pthread_t and vice versa the source argument should be cast first to

 * uintptr_t type and then to the destination type.

 *

 * If the handle would be a struct it should be stored in thread pool and

 * the thread pool must be freed in the thread pool exit procedure after

 * the last thread of pool will exit.

 */

static void

nxt_thread_pool_exit(nxt_task_t *task, void *obj, void *data)

{

    nxt_thread_t         *thread;

    nxt_thread_pool_t    *tp;

    nxt_atomic_uint_t    threads;

    nxt_thread_handle_t  handle;

    tp = obj;

    thread = task->thread;

    nxt_debug(task, "thread pool exit");

    if (data != NULL) {

        handle = (nxt_thread_handle_t) (uintptr_t) data;

        nxt_thread_wait(handle);

    }

    threads = nxt_atomic_fetch_add(&tp->threads, -1);

    nxt_debug(task, "thread pool threads: %A", threads);

    if (threads > 1) {

        nxt_work_set(&tp->work, nxt_thread_pool_exit, &tp->task, tp,

                     (void *) (uintptr_t) thread->handle);

        nxt_thread_pool_post(tp, &tp->work);

    } else {

        nxt_debug(task, "thread pool destroy");

        nxt_sem_destroy(&tp->sem);

        nxt_work_set(&tp->work, tp->exit, &tp->engine->task, tp,

                     (void *) (uintptr_t) thread->handle);

        nxt_event_engine_post(tp->engine, &tp->work);

        /* The "tp" memory should be freed by tp->exit handler. */

    }

    nxt_thread_exit(thread);

    nxt_unreachable();

}