/* Licensed to the Apache Software Foundation (ASF) under one or more

 * contributor license agreements.  See the NOTICE file distributed with

 * this work for additional information regarding copyright ownership.

 * The ASF licenses this file to You under the Apache License, Version 2.0

 * (the "License"); you may not use this file except in compliance with

 * the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

#include <assert.h>

#include <apr_atomic.h>

#include <apr_thread_mutex.h>

#include <apr_thread_cond.h>

#include <mpm_common.h>

#include <httpd.h>

#include <http_core.h>

#include <http_log.h>

#include "h2.h"

#include "h2_private.h"

#include "h2_mplx.h"

#include "h2_task.h"

#include "h2_workers.h"

#include "h2_util.h"

typedef struct h2_slot h2_slot;

struct h2_slot {

    int id;

    h2_slot *next;

    h2_workers *workers;

    int aborted;

    int sticks;

    h2_task *task;

    apr_thread_t *thread;

    apr_thread_mutex_t *lock;

    apr_thread_cond_t *not_idle;

};

static h2_slot *pop_slot(h2_slot **phead) 

{

    /* Atomically pop a slot from the list */

    for (;;) {

        h2_slot *first = *phead;

        if (first == NULL) {

            return NULL;

        }

        if (apr_atomic_casptr((void*)phead, first->next, first) == first) {

            first->next = NULL;

            return first;

        }

    }

}

static void push_slot(h2_slot **phead, h2_slot *slot)

{

    /* Atomically push a slot to the list */

    ap_assert(!slot->next);

    for (;;) {

        h2_slot *next = slot->next = *phead;

        if (apr_atomic_casptr((void*)phead, slot, next) == next) {

            return;

        }

    }

}

static void* APR_THREAD_FUNC slot_run(apr_thread_t *thread, void *wctx);

static apr_status_t activate_slot(h2_workers *workers, h2_slot *slot) 

{

    apr_status_t status;

    slot->workers = workers;

    slot->aborted = 0;

    slot->task = NULL;

    if (!slot->lock) {

        status = apr_thread_mutex_create(&slot->lock,

                                         APR_THREAD_MUTEX_DEFAULT,

                                         workers->pool);

        if (status != APR_SUCCESS) {

            push_slot(&workers->free, slot);

            return status;

        }

    }

    if (!slot->not_idle) {

        status = apr_thread_cond_create(&slot->not_idle, workers->pool);

        if (status != APR_SUCCESS) {

            push_slot(&workers->free, slot);

            return status;

        }

    }

    ap_log_error(APLOG_MARK, APLOG_TRACE2, 0, workers->s,

                 "h2_workers: new thread for slot %d", slot->id); 

    /* thread will either immediately start work or add itself

     * to the idle queue */

    apr_thread_create(&slot->thread, workers->thread_attr, slot_run, slot, 

                      workers->pool);

    if (!slot->thread) {

        push_slot(&workers->free, slot);

        return APR_ENOMEM;

    }

    apr_atomic_inc32(&workers->worker_count);

    return APR_SUCCESS;

}

static apr_status_t add_worker(h2_workers *workers)

{

    h2_slot *slot = pop_slot(&workers->free);

    if (slot) {

        return activate_slot(workers, slot);

    }

    return APR_EAGAIN;

}

static void wake_idle_worker(h2_workers *workers) 

{

    h2_slot *slot = pop_slot(&workers->idle);

    if (slot) {

        apr_thread_mutex_lock(slot->lock);

        apr_thread_cond_signal(slot->not_idle);

        apr_thread_mutex_unlock(slot->lock);

    }

    else if (workers->dynamic) {

        add_worker(workers);

    }

}

static void cleanup_zombies(h2_workers *workers)

{

    h2_slot *slot;

    while ((slot = pop_slot(&workers->zombies))) {

        if (slot->thread) {

            apr_status_t status;

            apr_thread_join(&status, slot->thread);

            slot->thread = NULL;

        }

        apr_atomic_dec32(&workers->worker_count);

        slot->next = NULL;

        push_slot(&workers->free, slot);

    }

}

static apr_status_t slot_pull_task(h2_slot *slot, h2_mplx *m)

{

    apr_status_t rv;

    rv = h2_mplx_pop_task(m, &slot->task);

    if (slot->task) {

        /* Ok, we got something to give back to the worker for execution. 

         * If we still have idle workers, we let the worker be sticky, 

         * e.g. making it poll the task's h2_mplx instance for more work 

         * before asking back here. */

        slot->sticks = slot->workers->max_workers;

        return rv;            

    }

    slot->sticks = 0;

    return APR_EOF;

}

static h2_fifo_op_t mplx_peek(void *head, void *ctx)

{

    h2_mplx *m = head;

    h2_slot *slot = ctx;

    if (slot_pull_task(slot, m) == APR_EAGAIN) {

        wake_idle_worker(slot->workers);

        return H2_FIFO_OP_REPUSH;

    } 

    return H2_FIFO_OP_PULL;

}

/**

 * Get the next task for the given worker. Will block until a task arrives

 * or the max_wait timer expires and more than min workers exist.

 */

static apr_status_t get_next(h2_slot *slot)

{

    h2_workers *workers = slot->workers;

    apr_status_t status;

    slot->task = NULL;

    while (!slot->aborted) {

        if (!slot->task) {

            status = h2_fifo_try_peek(workers->mplxs, mplx_peek, slot);

            if (status == APR_EOF) {

                return status;

            }

        }

        if (slot->task) {

            return APR_SUCCESS;

        }

        cleanup_zombies(workers);

        apr_thread_mutex_lock(slot->lock);

        push_slot(&workers->idle, slot);

        apr_thread_cond_wait(slot->not_idle, slot->lock);

        apr_thread_mutex_unlock(slot->lock);

    }

    return APR_EOF;

}

static void slot_done(h2_slot *slot)

{

    push_slot(&(slot->workers->zombies), slot);

}

static void* APR_THREAD_FUNC slot_run(apr_thread_t *thread, void *wctx)

{

    h2_slot *slot = wctx;

    while (!slot->aborted) {

        /* Get a h2_task from the mplxs queue. */

        get_next(slot);

        while (slot->task) {

            h2_task_do(slot->task, thread, slot->id);

            /* Report the task as done. If stickyness is left, offer the

             * mplx the opportunity to give us back a new task right away.

             */

            if (!slot->aborted && (--slot->sticks > 0)) {

                h2_mplx_task_done(slot->task->mplx, slot->task, &slot->task);

            }

            else {

                h2_mplx_task_done(slot->task->mplx, slot->task, NULL);

                slot->task = NULL;

            }

        }

    }

    slot_done(slot);

    return NULL;

}

static apr_status_t workers_pool_cleanup(void *data)

{

    h2_workers *workers = data;

    h2_slot *slot;

    if (!workers->aborted) {

        workers->aborted = 1;

        /* abort all idle slots */

        for (;;) {

            slot = pop_slot(&workers->idle);

            if (slot) {

                apr_thread_mutex_lock(slot->lock);

                slot->aborted = 1;

                apr_thread_cond_signal(slot->not_idle);

                apr_thread_mutex_unlock(slot->lock);

            }

            else {

                break;

            }

        }

        h2_fifo_term(workers->mplxs);

        h2_fifo_interrupt(workers->mplxs);

        cleanup_zombies(workers);

    }

    return APR_SUCCESS;

}

h2_workers *h2_workers_create(server_rec *s, apr_pool_t *server_pool,

                              int min_workers, int max_workers,

                              int idle_secs)

{

    apr_status_t status;

    h2_workers *workers;

    apr_pool_t *pool;

    int i, n;

    ap_assert(s);

    ap_assert(server_pool);

    /* let's have our own pool that will be parent to all h2_worker

     * instances we create. This happens in various threads, but always

     * guarded by our lock. Without this pool, all subpool creations would

     * happen on the pool handed to us, which we do not guard.

     */

    apr_pool_create(&pool, server_pool);

    apr_pool_tag(pool, "h2_workers");

    workers = apr_pcalloc(pool, sizeof(h2_workers));

    if (!workers) {

        return NULL;

    }

    workers->s = s;

    workers->pool = pool;

    workers->min_workers = min_workers;

    workers->max_workers = max_workers;

    workers->max_idle_secs = (idle_secs > 0)? idle_secs : 10;

    /* FIXME: the fifo set we use here has limited capacity. Once the

     * set is full, connections with new requests do a wait. Unfortunately,

     * we have optimizations in place there that makes such waiting "unfair"

     * in the sense that it may take connections a looong time to get scheduled.

     *

     * Need to rewrite this to use one of our double-linked lists and a mutex

     * to have unlimited capacity and fair scheduling.

     *

     * For now, we just make enough room to have many connections inside one

     * process.

     */

    status = h2_fifo_set_create(&workers->mplxs, pool, 8 * 1024);

    if (status != APR_SUCCESS) {

        return NULL;

    }

    status = apr_threadattr_create(&workers->thread_attr, workers->pool);

    if (status != APR_SUCCESS) {

        return NULL;

    }

    if (ap_thread_stacksize != 0) {

        apr_threadattr_stacksize_set(workers->thread_attr,

                                     ap_thread_stacksize);

        ap_log_error(APLOG_MARK, APLOG_TRACE3, 0, s,

                     "h2_workers: using stacksize=%ld", 

                     (long)ap_thread_stacksize);

    }

    status = apr_thread_mutex_create(&workers->lock,

                                     APR_THREAD_MUTEX_DEFAULT,

                                     workers->pool);

    if (status == APR_SUCCESS) {        

        n = workers->nslots = workers->max_workers;

        workers->slots = apr_pcalloc(workers->pool, n * sizeof(h2_slot));

        if (workers->slots == NULL) {

            workers->nslots = 0;

            status = APR_ENOMEM;

        }

        for (i = 0; i < n; ++i) {

            workers->slots[i].id = i;

        }

    }

    if (status == APR_SUCCESS) {

        /* we activate all for now, TODO: support min_workers again.

         * do this in reverse for vanity reasons so slot 0 will most

         * likely be at head of idle queue. */

        n = workers->max_workers;

        for (i = n-1; i >= 0; --i) {

            status = activate_slot(workers, &workers->slots[i]);

        }

        /* the rest of the slots go on the free list */

        for(i = n; i < workers->nslots; ++i) {

            push_slot(&workers->free, &workers->slots[i]);

        }

        workers->dynamic = (workers->worker_count < workers->max_workers);

    }

    if (status == APR_SUCCESS) {

        apr_pool_pre_cleanup_register(pool, workers, workers_pool_cleanup);    

        return workers;

    }

    return NULL;

}

apr_status_t h2_workers_register(h2_workers *workers, struct h2_mplx *m)

{

    apr_status_t status = h2_fifo_push(workers->mplxs, m);

    wake_idle_worker(workers);

    return status;

}

apr_status_t h2_workers_unregister(h2_workers *workers, struct h2_mplx *m)

{

    return h2_fifo_remove(workers->mplxs, m);

}