diff options
Diffstat (limited to 'modules/http2/h2_workers.c')
| -rw-r--r-- | modules/http2/h2_workers.c | 383 |
1 files changed, 383 insertions, 0 deletions
diff --git a/modules/http2/h2_workers.c b/modules/http2/h2_workers.c new file mode 100644 index 0000000..699f533 --- /dev/null +++ b/modules/http2/h2_workers.c @@ -0,0 +1,383 @@ +/* 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); +} |