[PATCH] Worker fdqueue patch...

[Justin Erenkrantz <je...@ebuilt.com>]

In the course of Aaron showing me how condition variables are supposed
to work, Aaron and I came up with this patch for the fdqueue code in 
the new worker MPM.

We think this makes the code thread-safe (tail == head, push starts, pop
comes - you are screwed, so the push should acquire one_big_mutex), 
but I'll let you guys be the judge of that.  Based on what I know about 
condition variables (which until this morning was nothing), this looks 
to be closer to what we want.

HTH.  -- justin

Index: fdqueue.c
===================================================================
RCS file: /home/cvs/httpd-2.0/server/mpm/worker/fdqueue.c,v
retrieving revision 1.1
diff -u -r1.1 fdqueue.c
--- fdqueue.c	2001/07/30 05:02:53	1.1
+++ fdqueue.c	2001/07/30 21:51:39
@@ -87,11 +87,6 @@
     return FD_QUEUE_SUCCESS;
 }
 
-static int increase_blanks(FDQueue *queue) {
-    queue->blanks++;
-    return FD_QUEUE_SUCCESS;
-}
-
 static apr_status_t ap_queue_destroy(void *data) {
     FDQueue *queue = data;
     /* Ignore errors here, we can't do anything about them anyway */
@@ -117,45 +112,40 @@
     return FD_QUEUE_SUCCESS;
 }
 
+/* Producer */
 int ap_queue_push(FDQueue *queue, apr_socket_t *sd, apr_pool_t *p) {
+    pthread_mutex_lock(&queue->one_big_mutex);
+    while (ap_queue_full(queue))
+        pthread_cond_wait(&queue->not_full, &queue->one_big_mutex);
     queue->data[queue->tail].sd = sd;
     queue->data[queue->tail].p  = p;
     queue->tail = (queue->tail + 1) % queue->bounds;
     queue->blanks--;
+    /* According to POSIX, cond_signal doesn't need the mutex.  This avoids 
+     * lock contention and unnecessary delays. */
+    pthread_mutex_unlock(&queue->one_big_mutex);
     pthread_cond_signal(&queue->not_empty);
-#if 0
-    if (queue->head == (queue->tail + 1) % queue->bounds) {
-#endif
-    if (ap_queue_full(queue)) {
-        pthread_cond_wait(&queue->not_full, &queue->one_big_mutex);
-    }
     return FD_QUEUE_SUCCESS;
 }
-
-apr_status_t ap_queue_pop(FDQueue *queue, apr_socket_t **sd, apr_pool_t **p, int block_if_empty) {
-    increase_blanks(queue);
-    /* We have just removed one from the queue.  By definition, it is
-     * no longer full.  We can ALWAYS signal the listener thread at
-     * this point.  However, the original code didn't do it this way,
-     * so I am leaving the original code in, just commented out.  BTW,
-     * originally, the increase_blanks wasn't in this function either.
-     *
-     if (queue->blanks > 0) {
-     */
-    pthread_cond_signal(&queue->not_full);
 
-    /*    }    */
-    if (queue->head == queue->tail) {
-        if (block_if_empty) {
-            pthread_cond_wait(&queue->not_empty, &queue->one_big_mutex);
-        }
-    } 
-    
+/* Consumer */
+apr_status_t ap_queue_pop(FDQueue *queue, apr_socket_t **sd, apr_pool_t **p) {
+    pthread_mutex_lock(&queue->one_big_mutex);
+    /* Wait until we have one to consume */
+    while (!ap_queue_size(queue))
+        pthread_cond_wait(&queue->not_empty, &queue->one_big_mutex);
+    /* Do the pop. */
     *sd = queue->data[queue->head].sd;
     *p  = queue->data[queue->head].p;
     queue->data[queue->head].sd = NULL;
+    queue->blanks++;
+    /* Advance the head. */
     if (*sd != NULL) {
         queue->head = (queue->head + 1) % queue->bounds;
     }
+    /* According to POSIX, cond_signal doesn't need the mutex.  This avoids 
+     * lock contention and unnecessary delays. */
+    pthread_mutex_unlock(&queue->one_big_mutex);
+    pthread_cond_signal(&queue->not_full);
     return APR_SUCCESS;
 }
Index: fdqueue.h
===================================================================
RCS file: /home/cvs/httpd-2.0/server/mpm/worker/fdqueue.h,v
retrieving revision 1.1
diff -u -r1.1 fdqueue.h
--- fdqueue.h	2001/07/30 05:02:53	1.1
+++ fdqueue.h	2001/07/30 21:51:39
@@ -87,7 +87,7 @@
 
 int ap_queue_init(FDQueue *queue, int queue_size, apr_pool_t *a);
 int ap_queue_push(FDQueue *queue, apr_socket_t *sd, apr_pool_t *p);
-apr_status_t ap_queue_pop(FDQueue *queue, apr_socket_t **sd, apr_pool_t **p, int block_if_empty);
+apr_status_t ap_queue_pop(FDQueue *queue, apr_socket_t **sd, apr_pool_t **p);
 int ap_queue_size(FDQueue *queue);
 int ap_queue_full(FDQueue *queue);
 int ap_block_on_queue(FDQueue *queue);
Index: worker.c
===================================================================
RCS file: /home/cvs/httpd-2.0/server/mpm/worker/worker.c,v
retrieving revision 1.1
diff -u -r1.1 worker.c
--- worker.c	2001/07/30 05:02:53	1.1
+++ worker.c	2001/07/30 21:51:39
@@ -632,7 +632,6 @@
             }
             if (csd != NULL) {
                 ap_queue_push(worker_queue, csd, ptrans);
-                ap_block_on_queue(worker_queue);
             }
         }
         else {
@@ -680,7 +679,7 @@
     free(ti);
 
     while (!workers_may_exit) {
-        ap_queue_pop(worker_queue, &csd, &ptrans, 1);
+        ap_queue_pop(worker_queue, &csd, &ptrans);
         process_socket(ptrans, csd, process_slot, thread_slot);
         requests_this_child--;
         apr_pool_clear(ptrans);