diff options
Diffstat (limited to '')
| -rw-r--r-- | server/mpm_fdqueue.c | 530 |
1 files changed, 530 insertions, 0 deletions
diff --git a/server/mpm_fdqueue.c b/server/mpm_fdqueue.c new file mode 100644 index 0000000..c812450 --- /dev/null +++ b/server/mpm_fdqueue.c @@ -0,0 +1,530 @@ +/* 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 "mpm_fdqueue.h" + +#if APR_HAS_THREADS + +#include <apr_atomic.h> + +static const apr_uint32_t zero_pt = APR_UINT32_MAX/2; + +struct recycled_pool +{ + apr_pool_t *pool; + struct recycled_pool *next; +}; + +struct fd_queue_info_t +{ + apr_uint32_t volatile idlers; /** + * >= zero_pt: number of idle worker threads + * < zero_pt: number of threads blocked, + * waiting for an idle worker + */ + apr_thread_mutex_t *idlers_mutex; + apr_thread_cond_t *wait_for_idler; + int terminated; + int max_idlers; + int max_recycled_pools; + apr_uint32_t recycled_pools_count; + struct recycled_pool *volatile recycled_pools; +}; + +struct fd_queue_elem_t +{ + apr_socket_t *sd; + void *sd_baton; + apr_pool_t *p; +}; + +static apr_status_t queue_info_cleanup(void *data_) +{ + fd_queue_info_t *qi = data_; + apr_thread_cond_destroy(qi->wait_for_idler); + apr_thread_mutex_destroy(qi->idlers_mutex); + + /* Clean up any pools in the recycled list */ + for (;;) { + struct recycled_pool *first_pool = qi->recycled_pools; + if (first_pool == NULL) { + break; + } + if (apr_atomic_casptr((void *)&qi->recycled_pools, first_pool->next, + first_pool) == first_pool) { + apr_pool_destroy(first_pool->pool); + } + } + + return APR_SUCCESS; +} + +apr_status_t ap_queue_info_create(fd_queue_info_t **queue_info, + apr_pool_t *pool, int max_idlers, + int max_recycled_pools) +{ + apr_status_t rv; + fd_queue_info_t *qi; + + qi = apr_pcalloc(pool, sizeof(*qi)); + + rv = apr_thread_mutex_create(&qi->idlers_mutex, APR_THREAD_MUTEX_DEFAULT, + pool); + if (rv != APR_SUCCESS) { + return rv; + } + rv = apr_thread_cond_create(&qi->wait_for_idler, pool); + if (rv != APR_SUCCESS) { + return rv; + } + qi->recycled_pools = NULL; + qi->max_recycled_pools = max_recycled_pools; + qi->max_idlers = max_idlers; + qi->idlers = zero_pt; + apr_pool_cleanup_register(pool, qi, queue_info_cleanup, + apr_pool_cleanup_null); + + *queue_info = qi; + + return APR_SUCCESS; +} + +apr_status_t ap_queue_info_set_idle(fd_queue_info_t *queue_info, + apr_pool_t *pool_to_recycle) +{ + apr_status_t rv; + + ap_queue_info_push_pool(queue_info, pool_to_recycle); + + /* If other threads are waiting on a worker, wake one up */ + if (apr_atomic_inc32(&queue_info->idlers) < zero_pt) { + rv = apr_thread_mutex_lock(queue_info->idlers_mutex); + if (rv != APR_SUCCESS) { + AP_DEBUG_ASSERT(0); + return rv; + } + rv = apr_thread_cond_signal(queue_info->wait_for_idler); + if (rv != APR_SUCCESS) { + apr_thread_mutex_unlock(queue_info->idlers_mutex); + return rv; + } + rv = apr_thread_mutex_unlock(queue_info->idlers_mutex); + if (rv != APR_SUCCESS) { + return rv; + } + } + + return APR_SUCCESS; +} + +apr_status_t ap_queue_info_try_get_idler(fd_queue_info_t *queue_info) +{ + /* Don't block if there isn't any idle worker. */ + for (;;) { + apr_uint32_t idlers = queue_info->idlers; + if (idlers <= zero_pt) { + return APR_EAGAIN; + } + if (apr_atomic_cas32(&queue_info->idlers, idlers - 1, + idlers) == idlers) { + return APR_SUCCESS; + } + } +} + +apr_status_t ap_queue_info_wait_for_idler(fd_queue_info_t *queue_info, + int *had_to_block) +{ + apr_status_t rv; + + /* Block if there isn't any idle worker. + * apr_atomic_add32(x, -1) does the same as dec32(x), except + * that it returns the previous value (unlike dec32's bool). + */ + if (apr_atomic_add32(&queue_info->idlers, -1) <= zero_pt) { + rv = apr_thread_mutex_lock(queue_info->idlers_mutex); + if (rv != APR_SUCCESS) { + AP_DEBUG_ASSERT(0); + apr_atomic_inc32(&(queue_info->idlers)); /* back out dec */ + return rv; + } + /* Re-check the idle worker count to guard against a + * race condition. Now that we're in the mutex-protected + * region, one of two things may have happened: + * - If the idle worker count is still negative, the + * workers are all still busy, so it's safe to + * block on a condition variable. + * - If the idle worker count is non-negative, then a + * worker has become idle since the first check + * of queue_info->idlers above. It's possible + * that the worker has also signaled the condition + * variable--and if so, the listener missed it + * because it wasn't yet blocked on the condition + * variable. But if the idle worker count is + * now non-negative, it's safe for this function to + * return immediately. + * + * A "negative value" (relative to zero_pt) in + * queue_info->idlers tells how many + * threads are waiting on an idle worker. + */ + if (queue_info->idlers < zero_pt) { + if (had_to_block) { + *had_to_block = 1; + } + rv = apr_thread_cond_wait(queue_info->wait_for_idler, + queue_info->idlers_mutex); + if (rv != APR_SUCCESS) { + AP_DEBUG_ASSERT(0); + apr_thread_mutex_unlock(queue_info->idlers_mutex); + return rv; + } + } + rv = apr_thread_mutex_unlock(queue_info->idlers_mutex); + if (rv != APR_SUCCESS) { + return rv; + } + } + + if (queue_info->terminated) { + return APR_EOF; + } + else { + return APR_SUCCESS; + } +} + +apr_uint32_t ap_queue_info_num_idlers(fd_queue_info_t *queue_info) +{ + apr_uint32_t val; + val = apr_atomic_read32(&queue_info->idlers); + return (val > zero_pt) ? val - zero_pt : 0; +} + +void ap_queue_info_push_pool(fd_queue_info_t *queue_info, + apr_pool_t *pool_to_recycle) +{ + struct recycled_pool *new_recycle; + /* If we have been given a pool to recycle, atomically link + * it into the queue_info's list of recycled pools + */ + if (!pool_to_recycle) + return; + + if (queue_info->max_recycled_pools >= 0) { + apr_uint32_t n = apr_atomic_read32(&queue_info->recycled_pools_count); + if (n >= queue_info->max_recycled_pools) { + apr_pool_destroy(pool_to_recycle); + return; + } + apr_atomic_inc32(&queue_info->recycled_pools_count); + } + + apr_pool_clear(pool_to_recycle); + new_recycle = apr_palloc(pool_to_recycle, sizeof *new_recycle); + new_recycle->pool = pool_to_recycle; + for (;;) { + /* + * Save queue_info->recycled_pool in local variable next because + * new_recycle->next can be changed after apr_atomic_casptr + * function call. For gory details see PR 44402. + */ + struct recycled_pool *next = queue_info->recycled_pools; + new_recycle->next = next; + if (apr_atomic_casptr((void *)&queue_info->recycled_pools, + new_recycle, next) == next) + break; + } +} + +void ap_queue_info_pop_pool(fd_queue_info_t *queue_info, + apr_pool_t **recycled_pool) +{ + /* Atomically pop a pool from the recycled list */ + + /* This function is safe only as long as it is single threaded because + * it reaches into the queue and accesses "next" which can change. + * We are OK today because it is only called from the listener thread. + * cas-based pushes do not have the same limitation - any number can + * happen concurrently with a single cas-based pop. + */ + + *recycled_pool = NULL; + + + /* Atomically pop a pool from the recycled list */ + for (;;) { + struct recycled_pool *first_pool = queue_info->recycled_pools; + if (first_pool == NULL) { + break; + } + if (apr_atomic_casptr((void *)&queue_info->recycled_pools, + first_pool->next, first_pool) == first_pool) { + *recycled_pool = first_pool->pool; + if (queue_info->max_recycled_pools >= 0) + apr_atomic_dec32(&queue_info->recycled_pools_count); + break; + } + } +} + +void ap_queue_info_free_idle_pools(fd_queue_info_t *queue_info) +{ + apr_pool_t *p; + + queue_info->max_recycled_pools = 0; + for (;;) { + ap_queue_info_pop_pool(queue_info, &p); + if (p == NULL) + break; + apr_pool_destroy(p); + } + apr_atomic_set32(&queue_info->recycled_pools_count, 0); +} + + +apr_status_t ap_queue_info_term(fd_queue_info_t *queue_info) +{ + apr_status_t rv; + + rv = apr_thread_mutex_lock(queue_info->idlers_mutex); + if (rv != APR_SUCCESS) { + return rv; + } + + queue_info->terminated = 1; + apr_thread_cond_broadcast(queue_info->wait_for_idler); + + return apr_thread_mutex_unlock(queue_info->idlers_mutex); +} + +/** + * Detects when the fd_queue_t is full. This utility function is expected + * to be called from within critical sections, and is not threadsafe. + */ +#define ap_queue_full(queue) ((queue)->nelts == (queue)->bounds) + +/** + * Detects when the fd_queue_t is empty. This utility function is expected + * to be called from within critical sections, and is not threadsafe. + */ +#define ap_queue_empty(queue) ((queue)->nelts == 0 && \ + APR_RING_EMPTY(&queue->timers, \ + timer_event_t, link)) + +/** + * Callback routine that is called to destroy this + * fd_queue_t when its pool is destroyed. + */ +static apr_status_t ap_queue_destroy(void *data) +{ + fd_queue_t *queue = data; + + /* Ignore errors here, we can't do anything about them anyway. + * XXX: We should at least try to signal an error here, it is + * indicative of a programmer error. -aaron */ + apr_thread_cond_destroy(queue->not_empty); + apr_thread_mutex_destroy(queue->one_big_mutex); + + return APR_SUCCESS; +} + +/** + * Initialize the fd_queue_t. + */ +apr_status_t ap_queue_create(fd_queue_t **pqueue, int capacity, apr_pool_t *p) +{ + apr_status_t rv; + fd_queue_t *queue; + + queue = apr_pcalloc(p, sizeof *queue); + + if ((rv = apr_thread_mutex_create(&queue->one_big_mutex, + APR_THREAD_MUTEX_DEFAULT, + p)) != APR_SUCCESS) { + return rv; + } + if ((rv = apr_thread_cond_create(&queue->not_empty, p)) != APR_SUCCESS) { + return rv; + } + + APR_RING_INIT(&queue->timers, timer_event_t, link); + + queue->data = apr_pcalloc(p, capacity * sizeof(fd_queue_elem_t)); + queue->bounds = capacity; + + apr_pool_cleanup_register(p, queue, ap_queue_destroy, + apr_pool_cleanup_null); + *pqueue = queue; + + return APR_SUCCESS; +} + +/** + * Push a new socket onto the queue. + * + * precondition: ap_queue_info_wait_for_idler has already been called + * to reserve an idle worker thread + */ +apr_status_t ap_queue_push_socket(fd_queue_t *queue, + apr_socket_t *sd, void *sd_baton, + apr_pool_t *p) +{ + fd_queue_elem_t *elem; + apr_status_t rv; + + if ((rv = apr_thread_mutex_lock(queue->one_big_mutex)) != APR_SUCCESS) { + return rv; + } + + AP_DEBUG_ASSERT(!queue->terminated); + AP_DEBUG_ASSERT(!ap_queue_full(queue)); + + elem = &queue->data[queue->in++]; + if (queue->in >= queue->bounds) + queue->in -= queue->bounds; + elem->sd = sd; + elem->sd_baton = sd_baton; + elem->p = p; + queue->nelts++; + + apr_thread_cond_signal(queue->not_empty); + + return apr_thread_mutex_unlock(queue->one_big_mutex); +} + +apr_status_t ap_queue_push_timer(fd_queue_t *queue, timer_event_t *te) +{ + apr_status_t rv; + + if ((rv = apr_thread_mutex_lock(queue->one_big_mutex)) != APR_SUCCESS) { + return rv; + } + + AP_DEBUG_ASSERT(!queue->terminated); + + APR_RING_INSERT_TAIL(&queue->timers, te, timer_event_t, link); + + apr_thread_cond_signal(queue->not_empty); + + return apr_thread_mutex_unlock(queue->one_big_mutex); +} + +/** + * Retrieves the next available socket from the queue. If there are no + * sockets available, it will block until one becomes available. + * Once retrieved, the socket is placed into the address specified by + * 'sd'. + */ +apr_status_t ap_queue_pop_something(fd_queue_t *queue, + apr_socket_t **sd, void **sd_baton, + apr_pool_t **p, timer_event_t **te_out) +{ + fd_queue_elem_t *elem; + timer_event_t *te; + apr_status_t rv; + + if ((rv = apr_thread_mutex_lock(queue->one_big_mutex)) != APR_SUCCESS) { + return rv; + } + + /* Keep waiting until we wake up and find that the queue is not empty. */ + if (ap_queue_empty(queue)) { + if (!queue->terminated) { + apr_thread_cond_wait(queue->not_empty, queue->one_big_mutex); + } + /* If we wake up and it's still empty, then we were interrupted */ + if (ap_queue_empty(queue)) { + rv = apr_thread_mutex_unlock(queue->one_big_mutex); + if (rv != APR_SUCCESS) { + return rv; + } + if (queue->terminated) { + return APR_EOF; /* no more elements ever again */ + } + else { + return APR_EINTR; + } + } + } + + te = NULL; + if (te_out) { + if (!APR_RING_EMPTY(&queue->timers, timer_event_t, link)) { + te = APR_RING_FIRST(&queue->timers); + APR_RING_REMOVE(te, link); + } + *te_out = te; + } + if (!te) { + elem = &queue->data[queue->out++]; + if (queue->out >= queue->bounds) + queue->out -= queue->bounds; + queue->nelts--; + + *sd = elem->sd; + if (sd_baton) { + *sd_baton = elem->sd_baton; + } + *p = elem->p; +#ifdef AP_DEBUG + elem->sd = NULL; + elem->p = NULL; +#endif /* AP_DEBUG */ + } + + return apr_thread_mutex_unlock(queue->one_big_mutex); +} + +static apr_status_t queue_interrupt(fd_queue_t *queue, int all, int term) +{ + apr_status_t rv; + + if ((rv = apr_thread_mutex_lock(queue->one_big_mutex)) != APR_SUCCESS) { + return rv; + } + + /* we must hold one_big_mutex when setting this... otherwise, + * we could end up setting it and waking everybody up just after a + * would-be popper checks it but right before they block + */ + if (term) { + queue->terminated = 1; + } + if (all) + apr_thread_cond_broadcast(queue->not_empty); + else + apr_thread_cond_signal(queue->not_empty); + + return apr_thread_mutex_unlock(queue->one_big_mutex); +} + +apr_status_t ap_queue_interrupt_all(fd_queue_t *queue) +{ + return queue_interrupt(queue, 1, 0); +} + +apr_status_t ap_queue_interrupt_one(fd_queue_t *queue) +{ + return queue_interrupt(queue, 0, 0); +} + +apr_status_t ap_queue_term(fd_queue_t *queue) +{ + return queue_interrupt(queue, 1, 1); +} + +#endif /* APR_HAS_THREADS */ |