apr_queue_pop1 [BLOCKED] (nelts = 0) apr_queue_pop2 [BLOCKED] (nelts = 0) apr_queue_pop3 [BLOCKED] (nelts = 0) apr_queue_push1 (nelts = 1) (nelts was 0, so signals) apr_queue_push2 (nelts = 2) apr_queue_pop1 [UNBLOCKS] (nelts = 1) (awake from push1's signal) apr_queue_push3 (nelts = 2) apr_queue_push4 (nelts = 3)
pop2, pop3 as well as any others remained blocked.
Similar logic is used to handle blocked calls to apr_queue_push. My solution was to add two variables to apr_queue_t to track the number of threads waiting for a non-full and non-empty state, and choosing to signal the conditions based on those variables. So far this has worked fine for me. testqueue doesn't seem to catch this because of the way it's written (timing delays betwee calls make this harder to catch)
I've attached a patch. I appreciate any comments, it being my first patch and all.
--- jacob lewallen [EMAIL PROTECTED]
Index: apr_queue.c
===================================================================
RCS file: /home/cvspublic/apr-util/misc/apr_queue.c,v
retrieving revision 1.10
diff -u -u -r1.10 apr_queue.c
--- apr_queue.c 13 Jan 2003 20:15:50 -0000 1.10
+++ apr_queue.c 9 Feb 2003 05:33:57 -0000
@@ -73,7 +73,7 @@
#include "apr_queue.h"
#if APR_HAS_THREADS
-/*
+/*
* define this to get debug messages
*
#define QUEUE_DEBUG
@@ -85,6 +85,8 @@
unsigned int in; /**< next empty location */
unsigned int out; /**< next filled location */
unsigned int bounds;/**< max size of queue */
+ unsigned int full_waiters;
+ unsigned int empty_waiters;
apr_thread_mutex_t *one_big_mutex;
apr_thread_cond_t *not_empty;
apr_thread_cond_t *not_full;
@@ -93,14 +95,14 @@
#ifdef QUEUE_DEBUG
static void Q_DBG(char*msg, apr_queue_t *q) {
- fprintf(stderr, "%ld\t#%d in %d out %d\t%s\n",
+ fprintf(stderr, "%ld\t#%d in %d out %d\t%s\n",
apr_os_thread_current(),
q->nelts, q->in, q->out,
msg
);
}
#else
-#define Q_DBG(x,y)
+#define Q_DBG(x,y)
#endif
/**
@@ -119,7 +121,7 @@
* Callback routine that is called to destroy this
* apr_queue_t when its pool is destroyed.
*/
-static apr_status_t queue_destroy(void *data)
+static apr_status_t queue_destroy(void *data)
{
apr_queue_t *queue = data;
@@ -135,8 +137,8 @@
/**
* Initialize the apr_queue_t.
*/
-APU_DECLARE(apr_status_t) apr_queue_create(apr_queue_t **q,
- unsigned int queue_capacity,
+APU_DECLARE(apr_status_t) apr_queue_create(apr_queue_t **q,
+ unsigned int queue_capacity,
apr_pool_t *a)
{
apr_status_t rv;
@@ -169,6 +171,8 @@
queue->in = 0;
queue->out = 0;
queue->terminated = 0;
+ queue->full_waiters = 0;
+ queue->empty_waiters = 0;
apr_pool_cleanup_register(a, queue, queue_destroy, apr_pool_cleanup_null);
@@ -183,7 +187,6 @@
APU_DECLARE(apr_status_t) apr_queue_push(apr_queue_t *queue, void *data)
{
apr_status_t rv;
- int need_signal = 0;
if (queue->terminated) {
return APR_EOF; /* no more elements ever again */
@@ -196,7 +199,9 @@
if (apr_queue_full(queue)) {
if (!queue->terminated) {
+ queue->full_waiters++;
rv = apr_thread_cond_wait(queue->not_full, queue->one_big_mutex);
+ queue->full_waiters--;
if (rv != APR_SUCCESS) {
apr_thread_mutex_unlock(queue->one_big_mutex);
return rv;
@@ -218,16 +223,11 @@
}
}
- /* if we were empty then signal that we aren't */
- if (apr_queue_empty(queue)) {
- need_signal = 1;
- }
-
queue->data[queue->in] = data;
queue->in = (queue->in + 1) % queue->bounds;
queue->nelts++;
- if (need_signal == 1) {
+ if (queue->empty_waiters) {
Q_DBG("sig !empty", queue);
rv = apr_thread_cond_signal(queue->not_empty);
if (rv != APR_SUCCESS) {
@@ -248,7 +248,6 @@
APU_DECLARE(apr_status_t) apr_queue_trypush(apr_queue_t *queue, void *data)
{
apr_status_t rv;
- int need_signal = 0;
if (queue->terminated) {
return APR_EOF; /* no more elements ever again */
}
@@ -263,16 +262,11 @@
return APR_EAGAIN;
}
- /* if we were empty then signal that we aren't */
- if (apr_queue_empty(queue)) {
- need_signal = 1;
- }
-
queue->data[queue->in] = data;
queue->in = (queue->in + 1) % queue->bounds;
queue->nelts++;
- if (need_signal == 1) {
+ if (queue->empty_waiters) {
Q_DBG("sig !empty", queue);
rv = apr_thread_cond_signal(queue->not_empty);
if (rv != APR_SUCCESS) {
@@ -301,7 +295,6 @@
APU_DECLARE(apr_status_t) apr_queue_pop(apr_queue_t *queue, void **data)
{
apr_status_t rv;
- int need_signal = 0;
if (queue->terminated) {
return APR_EOF; /* no more elements ever again */
@@ -315,7 +308,9 @@
/* 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 = apr_thread_cond_wait(queue->not_empty, queue->one_big_mutex);
+ queue->empty_waiters--;
if (rv != APR_SUCCESS) {
apr_thread_mutex_unlock(queue->one_big_mutex);
return rv;
@@ -335,16 +330,13 @@
return APR_EINTR;
}
}
- }
- if (apr_queue_full(queue)) {
- need_signal = 1;
}
*data = &queue->data[queue->out];
queue->nelts--;
queue->out = (queue->out + 1) % queue->bounds;
- if (need_signal == 1) {
+ if (queue->full_waiters) {
Q_DBG("signal !full", queue);
rv = apr_thread_cond_signal(queue->not_full);
if (rv != APR_SUCCESS) {
@@ -366,7 +358,6 @@
APU_DECLARE(apr_status_t) apr_queue_trypop(apr_queue_t *queue, void **data)
{
apr_status_t rv;
- int need_signal = 0;
if (queue->terminated) {
return APR_EOF; /* no more elements ever again */
@@ -381,16 +372,13 @@
if (apr_queue_empty(queue)) {
rv = apr_thread_mutex_unlock(queue->one_big_mutex);
return APR_EAGAIN;
- }
- if (apr_queue_full(queue)) {
- need_signal = 1;
}
*data = &queue->data[queue->out];
queue->nelts--;
queue->out = (queue->out + 1) % queue->bounds;
- if (need_signal == 1) {
+ if (queue->full_waiters) {
Q_DBG("signal !full", queue);
rv = apr_thread_cond_signal(queue->not_full);
if (rv != APR_SUCCESS) {
@@ -406,7 +394,7 @@
APU_DECLARE(apr_status_t) apr_queue_interrupt_all(apr_queue_t *queue)
{
apr_status_t rv;
- Q_DBG("intr all", queue);
+ Q_DBG("intr all", queue);
if ((rv = apr_thread_mutex_lock(queue->one_big_mutex)) != APR_SUCCESS) {
return rv;
}
@@ -429,7 +417,7 @@
}
/* we must hold one_big_mutex when setting this... otherwise,
- * we could end up setting it and waking everybody up just after a
+ * 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;
