switch_apr_queue.c File Reference

#include <switch.h>
#include <fspr.h>
#include <fspr_thread_mutex.h>
#include <fspr_thread_cond.h>

Include dependency graph for switch_apr_queue.c:

Go to the source code of this file.

Data Structures
struct	switch_apr_queue_t

Macros
#define	Q_DBG(x, y)
#define	apr_queue_full(queue)   ((queue)->nelts == (queue)->bounds)
#define	apr_queue_empty(queue)   ((queue)->nelts == 0)

Typedefs
typedef struct switch_apr_queue_t	switch_apr_queue_t

Functions
static fspr_status_t	queue_destroy (void *data)
fspr_status_t	switch_apr_queue_create (switch_apr_queue_t **q, unsigned int queue_capacity, fspr_pool_t *a)
fspr_status_t	switch_apr_queue_push (switch_apr_queue_t *queue, void *data)
fspr_status_t	switch_apr_queue_trypush (switch_apr_queue_t *queue, void *data)
unsigned int	switch_apr_queue_size (switch_apr_queue_t *queue)
fspr_status_t	switch_apr_queue_pop (switch_apr_queue_t *queue, void **data)
fspr_status_t	switch_apr_queue_pop_timeout (switch_apr_queue_t *queue, void **data, fspr_interval_time_t timeout)
fspr_status_t	switch_apr_queue_trypop (switch_apr_queue_t *queue, void **data)
fspr_status_t	switch_apr_queue_interrupt_all (switch_apr_queue_t *queue)
fspr_status_t	switch_apr_queue_term (switch_apr_queue_t *queue)

Macro Definition Documentation

apr_queue_empty

#define apr_queue_empty

(

queue

)

((queue)->nelts == 0)

Detects when the switch_apr_queue_t is empty. This utility function is expected to be called from within critical sections, and is not threadsafe.

Definition at line 66 of file switch_apr_queue.c.

Referenced by switch_apr_queue_pop(), switch_apr_queue_pop_timeout(), and switch_apr_queue_trypop().

apr_queue_full

#define apr_queue_full

(

queue

)

((queue)->nelts == (queue)->bounds)

Detects when the switch_apr_queue_t is full. This utility function is expected to be called from within critical sections, and is not threadsafe.

Definition at line 60 of file switch_apr_queue.c.

Referenced by switch_apr_queue_push(), and switch_apr_queue_trypush().

Q_DBG

#define Q_DBG	(		x,
			y
	)

Definition at line 53 of file switch_apr_queue.c.

Referenced by switch_apr_queue_interrupt_all(), switch_apr_queue_pop(), switch_apr_queue_pop_timeout(), switch_apr_queue_push(), switch_apr_queue_trypop(), and switch_apr_queue_trypush().

Typedef Documentation

switch_apr_queue_t

typedef struct switch_apr_queue_t switch_apr_queue_t

Definition at line 42 of file switch_apr_queue.c.

Function Documentation

queue_destroy()

static fspr_status_t queue_destroy ( void *  data )

static

Callback routine that is called to destroy this switch_apr_queue_t when its pool is destroyed.

Definition at line 72 of file switch_apr_queue.c.

References switch_apr_queue_t::data, switch_apr_queue_t::not_empty, switch_apr_queue_t::not_full, and switch_apr_queue_t::one_big_mutex.

Referenced by switch_apr_queue_create().

{
    switch_apr_queue_t *queue = data;

    /* Ignore errors here, we can't do anything about them anyway. */

    fspr_thread_cond_destroy(queue->not_empty);
    fspr_thread_cond_destroy(queue->not_full);
    fspr_thread_mutex_destroy(queue->one_big_mutex);

    return APR_SUCCESS;
}

switch_apr_queue_create()

fspr_status_t switch_apr_queue_create	(	switch_apr_queue_t **	q,
		unsigned int	queue_capacity,
		fspr_pool_t *	a
	)

Initialize the switch_apr_queue_t.

Definition at line 88 of file switch_apr_queue.c.

References switch_apr_queue_t::bounds, switch_apr_queue_t::data, switch_apr_queue_t::empty_waiters, switch_apr_queue_t::full_waiters, switch_apr_queue_t::in, memset(), switch_apr_queue_t::nelts, switch_apr_queue_t::not_empty, switch_apr_queue_t::not_full, switch_apr_queue_t::one_big_mutex, switch_apr_queue_t::out, queue_destroy(), and switch_apr_queue_t::terminated.

Referenced by switch_queue_create().

{
    fspr_status_t rv;
    switch_apr_queue_t *queue;
    queue = fspr_palloc(a, sizeof(switch_apr_queue_t));
    *q = queue;

    /* nested doesn't work ;( */
    rv = fspr_thread_mutex_create(&queue->one_big_mutex,
                                 APR_THREAD_MUTEX_UNNESTED,
                                 a);
    if (rv != APR_SUCCESS) {
        return rv;
    }

    rv = fspr_thread_cond_create(&queue->not_empty, a);
    if (rv != APR_SUCCESS) {
        return rv;
    }

    rv = fspr_thread_cond_create(&queue->not_full, a);
    if (rv != APR_SUCCESS) {
        return rv;
    }

    /* Set all the data in the queue to NULL */
    queue->data = fspr_palloc(a, queue_capacity * sizeof(void*));
        if (!queue->data) return APR_ENOMEM;
        memset(queue->data, 0, queue_capacity * sizeof(void*));
    queue->bounds = queue_capacity;
    queue->nelts = 0;
    queue->in = 0;
    queue->out = 0;
    queue->terminated = 0;
    queue->full_waiters = 0;
    queue->empty_waiters = 0;

    fspr_pool_cleanup_register(a, queue, queue_destroy, fspr_pool_cleanup_null);

    return APR_SUCCESS;
}

switch_apr_queue_interrupt_all()

fspr_status_t switch_apr_queue_interrupt_all

(

switch_apr_queue_t *

queue

)

Definition at line 406 of file switch_apr_queue.c.

References switch_apr_queue_t::not_empty, switch_apr_queue_t::not_full, switch_apr_queue_t::one_big_mutex, and Q_DBG.

Referenced by switch_apr_queue_term(), and switch_queue_interrupt_all().

{
    fspr_status_t rv;
    Q_DBG("intr all", queue);    
    if ((rv = fspr_thread_mutex_lock(queue->one_big_mutex)) != APR_SUCCESS) {
        return rv;
    }
    fspr_thread_cond_broadcast(queue->not_empty);
    fspr_thread_cond_broadcast(queue->not_full);

    if ((rv = fspr_thread_mutex_unlock(queue->one_big_mutex)) != APR_SUCCESS) {
        return rv;
    }

    return APR_SUCCESS;
}

switch_apr_queue_pop()

fspr_status_t switch_apr_queue_pop	(	switch_apr_queue_t *	queue,
		void **	data
	)

Retrieves the next item from the queue. If there are no items available, it will block until one becomes available. Once retrieved, the item is placed into the address specified by 'data'.

Definition at line 244 of file switch_apr_queue.c.

References apr_queue_empty, switch_apr_queue_t::bounds, switch_apr_queue_t::data, switch_apr_queue_t::empty_waiters, switch_apr_queue_t::full_waiters, if(), switch_apr_queue_t::nelts, switch_apr_queue_t::not_empty, switch_apr_queue_t::not_full, switch_apr_queue_t::one_big_mutex, switch_apr_queue_t::out, Q_DBG, and switch_apr_queue_t::terminated.

Referenced by switch_queue_pop().

{
    fspr_status_t rv;

    if (queue->terminated) {
        return APR_EOF; /* no more elements ever again */
    }

    rv = fspr_thread_mutex_lock(queue->one_big_mutex);
    if (rv != APR_SUCCESS) {
        return rv;
    }

    /* Keep waiting until we wake up and find that the queue is not empty. */
    if (apr_queue_empty(queue)) {
        if (!queue->terminated) {
            queue->empty_waiters++;
            rv = fspr_thread_cond_wait(queue->not_empty, queue->one_big_mutex);
            queue->empty_waiters--;
            if (rv != APR_SUCCESS) {
                fspr_thread_mutex_unlock(queue->one_big_mutex);
                return rv;
            }
        }
        /* If we wake up and it's still empty, then we were interrupted */
        if (apr_queue_empty(queue)) {
            Q_DBG("queue empty (intr)", queue);
            rv = fspr_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;
            }
        }
    } 

    *data = queue->data[queue->out];
    queue->nelts--;

    queue->out = (queue->out + 1) % queue->bounds;
    if (queue->full_waiters) {
        Q_DBG("signal !full", queue);
        rv = fspr_thread_cond_signal(queue->not_full);
        if (rv != APR_SUCCESS) {
            fspr_thread_mutex_unlock(queue->one_big_mutex);
            return rv;
        }
    }

    rv = fspr_thread_mutex_unlock(queue->one_big_mutex);
    return rv;
}

switch_apr_queue_pop_timeout()

fspr_status_t switch_apr_queue_pop_timeout	(	switch_apr_queue_t *	queue,
		void **	data,
		fspr_interval_time_t	timeout
	)

Retrieves the next item from the queue. If there are no items available, it will block until one becomes available, or until timeout is elapsed. Once retrieved, the item is placed into the address specified by'data'.

Definition at line 307 of file switch_apr_queue.c.

References apr_queue_empty, switch_apr_queue_t::bounds, switch_apr_queue_t::data, switch_apr_queue_t::empty_waiters, switch_apr_queue_t::full_waiters, if(), switch_apr_queue_t::nelts, switch_apr_queue_t::not_empty, switch_apr_queue_t::not_full, switch_apr_queue_t::one_big_mutex, switch_apr_queue_t::out, Q_DBG, and switch_apr_queue_t::terminated.

Referenced by switch_queue_pop_timeout().

{
    fspr_status_t rv;

    if (queue->terminated) {
        return APR_EOF; /* no more elements ever again */
    }

    rv = fspr_thread_mutex_lock(queue->one_big_mutex);
    if (rv != APR_SUCCESS) {
        return rv;
    }

    /* Keep waiting until we wake up and find that the queue is not empty. */
    if (apr_queue_empty(queue)) {
        if (!queue->terminated) {
            queue->empty_waiters++;
            rv = fspr_thread_cond_timedwait(queue->not_empty, queue->one_big_mutex, timeout);
            queue->empty_waiters--;
                        /* In the event of a timemout, APR_TIMEUP will be returned */
            if (rv != APR_SUCCESS) {
                fspr_thread_mutex_unlock(queue->one_big_mutex);
                return rv;
            }
        }
        /* If we wake up and it's still empty, then we were interrupted */
        if (apr_queue_empty(queue)) {
            Q_DBG("queue empty (intr)", queue);
            rv = fspr_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;
            }
        }
    } 

    *data = queue->data[queue->out];
    queue->nelts--;

    queue->out = (queue->out + 1) % queue->bounds;
    if (queue->full_waiters) {
        Q_DBG("signal !full", queue);
        rv = fspr_thread_cond_signal(queue->not_full);
        if (rv != APR_SUCCESS) {
            fspr_thread_mutex_unlock(queue->one_big_mutex);
            return rv;
        }
    }

    rv = fspr_thread_mutex_unlock(queue->one_big_mutex);
    return rv;
}

switch_apr_queue_push()

fspr_status_t switch_apr_queue_push	(	switch_apr_queue_t *	queue,
		void *	data
	)

Push new data onto the queue. Blocks if the queue is full. Once the push operation has completed, it signals other threads waiting in apr_queue_pop() that they may continue consuming sockets.

Definition at line 135 of file switch_apr_queue.c.

References apr_queue_full, switch_apr_queue_t::bounds, switch_apr_queue_t::data, switch_apr_queue_t::empty_waiters, switch_apr_queue_t::full_waiters, if(), switch_apr_queue_t::in, switch_apr_queue_t::nelts, switch_apr_queue_t::not_empty, switch_apr_queue_t::not_full, switch_apr_queue_t::one_big_mutex, Q_DBG, and switch_apr_queue_t::terminated.

Referenced by switch_queue_push().

{
    fspr_status_t rv;

    if (queue->terminated) {
        return APR_EOF; /* no more elements ever again */
    }

    rv = fspr_thread_mutex_lock(queue->one_big_mutex);
    if (rv != APR_SUCCESS) {
        return rv;
    }

    if (apr_queue_full(queue)) {
        if (!queue->terminated) {
            queue->full_waiters++;
            rv = fspr_thread_cond_wait(queue->not_full, queue->one_big_mutex);
            queue->full_waiters--;
            if (rv != APR_SUCCESS) {
                fspr_thread_mutex_unlock(queue->one_big_mutex);
                return rv;
            }
        }
        /* If we wake up and it's still empty, then we were interrupted */
        if (apr_queue_full(queue)) {
            Q_DBG("queue full (intr)", queue);
            rv = fspr_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;
            }
        }
    }

    queue->data[queue->in] = data;
    queue->in = (queue->in + 1) % queue->bounds;
    queue->nelts++;

    if (queue->empty_waiters) {
        Q_DBG("sig !empty", queue);
        rv = fspr_thread_cond_signal(queue->not_empty);
        if (rv != APR_SUCCESS) {
            fspr_thread_mutex_unlock(queue->one_big_mutex);
            return rv;
        }
    }

    rv = fspr_thread_mutex_unlock(queue->one_big_mutex);
    return rv;
}

switch_apr_queue_size()

unsigned int switch_apr_queue_size

(

switch_apr_queue_t *

queue

)

not thread safe

Definition at line 234 of file switch_apr_queue.c.

References switch_apr_queue_t::nelts.

Referenced by switch_queue_size().

                                                              {
    return queue->nelts;
}

switch_apr_queue_term()

fspr_status_t switch_apr_queue_term

(

switch_apr_queue_t *

queue

)

Definition at line 423 of file switch_apr_queue.c.

References switch_apr_queue_t::one_big_mutex, switch_apr_queue_interrupt_all(), and switch_apr_queue_t::terminated.

Referenced by switch_queue_term().

{
    fspr_status_t rv;

    if ((rv = fspr_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
     */
    queue->terminated = 1;
    if ((rv = fspr_thread_mutex_unlock(queue->one_big_mutex)) != APR_SUCCESS) {
        return rv;
    }
    return switch_apr_queue_interrupt_all(queue);
}

switch_apr_queue_trypop()

fspr_status_t switch_apr_queue_trypop	(	switch_apr_queue_t *	queue,
		void **	data
	)

Retrieves the next item from the queue. If there are no items available, return APR_EAGAIN. Once retrieved, the item is placed into the address specified by 'data'.

Definition at line 371 of file switch_apr_queue.c.

References apr_queue_empty, switch_apr_queue_t::bounds, switch_apr_queue_t::data, switch_apr_queue_t::full_waiters, if(), switch_apr_queue_t::nelts, switch_apr_queue_t::not_full, switch_apr_queue_t::one_big_mutex, switch_apr_queue_t::out, Q_DBG, and switch_apr_queue_t::terminated.

Referenced by switch_queue_trypop().

{
    fspr_status_t rv;

    if (queue->terminated) {
        return APR_EOF; /* no more elements ever again */
    }

    rv = fspr_thread_mutex_lock(queue->one_big_mutex);
    if (rv != APR_SUCCESS) {
        return rv;
    }

    if (apr_queue_empty(queue)) {
        fspr_thread_mutex_unlock(queue->one_big_mutex);
        return APR_EAGAIN;
    } 

    *data = queue->data[queue->out];
    queue->nelts--;

    queue->out = (queue->out + 1) % queue->bounds;
    if (queue->full_waiters) {
        Q_DBG("signal !full", queue);
        rv = fspr_thread_cond_signal(queue->not_full);
        if (rv != APR_SUCCESS) {
            fspr_thread_mutex_unlock(queue->one_big_mutex);
            return rv;
        }
    }

    rv = fspr_thread_mutex_unlock(queue->one_big_mutex);
    return rv;
}

switch_apr_queue_trypush()

fspr_status_t switch_apr_queue_trypush	(	switch_apr_queue_t *	queue,
		void *	data
	)

Push new data onto the queue. Blocks if the queue is full. Once the push operation has completed, it signals other threads waiting in apr_queue_pop() that they may continue consuming sockets.

Definition at line 196 of file switch_apr_queue.c.

References apr_queue_full, switch_apr_queue_t::bounds, switch_apr_queue_t::data, switch_apr_queue_t::empty_waiters, if(), switch_apr_queue_t::in, switch_apr_queue_t::nelts, switch_apr_queue_t::not_empty, switch_apr_queue_t::one_big_mutex, Q_DBG, and switch_apr_queue_t::terminated.

Referenced by switch_queue_trypush().

{
    fspr_status_t rv;

    if (queue->terminated) {
        return APR_EOF; /* no more elements ever again */
    }

    rv = fspr_thread_mutex_lock(queue->one_big_mutex);
    if (rv != APR_SUCCESS) {
        return rv;
    }

    if (apr_queue_full(queue)) {
        fspr_thread_mutex_unlock(queue->one_big_mutex);
        return APR_EAGAIN;
    }
    
    queue->data[queue->in] = data;
    queue->in = (queue->in + 1) % queue->bounds;
    queue->nelts++;

    if (queue->empty_waiters) {
        Q_DBG("sig !empty", queue);
        rv  = fspr_thread_cond_signal(queue->not_empty);
        if (rv != APR_SUCCESS) {
            fspr_thread_mutex_unlock(queue->one_big_mutex);
            return rv;
        }
    }

    rv = fspr_thread_mutex_unlock(queue->one_big_mutex);
    return rv;
}