Enabled async write completion in the Event MPM

git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/branches/async-dev@312536 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 47 insertions, 11 deletions

diff --git a/CHANGES b/CHANGES
index d2a71f7c07..e122987dba 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,6 +1,8 @@
                                                         -*- coding: utf-8 -*-
 Changes in Apache 2.3.0 async-dev R&D branch
 
+  *) Added asynchronous write completion for Event MPM.  [Brian Pane]
+
   *) For asynchronous MPMs, moved responsibility for write completion from
      the core to the MPM.  [Brian Pane]
 
diff --git a/server/mpm/experimental/event/event.c b/server/mpm/experimental/event/event.c
index e39e569aff..421209835a 100644
--- a/server/mpm/experimental/event/event.c
+++ b/server/mpm/experimental/event/event.c
@@ -164,7 +164,7 @@ static int sick_child_detected;
 
 apr_thread_mutex_t *timeout_mutex;
 APR_RING_HEAD(timeout_head_t, conn_state_t);
-static struct timeout_head_t timeout_head;
+static struct timeout_head_t timeout_head, keepalive_timeout_head;
 
 static apr_pollset_t *event_pollset;
 
@@ -592,6 +592,7 @@ static int process_socket(apr_pool_t * p, apr_socket_t * sock,
         pt->status = 1;
         pt->baton = cs;
         cs->pfd.client_data = pt;
+        APR_RING_ELEM_INIT(cs, timeout_list);
 
         ap_update_vhost_given_ip(c);
 
@@ -621,6 +622,7 @@ static int process_socket(apr_pool_t * p, apr_socket_t * sock,
     else {
         c = cs->c;
         c->sbh = sbh;
+        pt = cs->pfd.client_data;
     }
 
 read_request:
@@ -643,21 +645,30 @@ read_request:
          * rest of the response.  TODO: Hand off this connection to a
          * pollset for asynchronous write completion.
          */
-        apr_bucket_brigade *bb = apr_brigade_create(c->pool, c->bucket_alloc);
-        apr_bucket *b = apr_bucket_flush_create(c->bucket_alloc);
         ap_filter_t *output_filter = c->output_filters;
         apr_status_t rv;
-        APR_BRIGADE_INSERT_HEAD(bb, b);
         while (output_filter->next != NULL) {
             output_filter = output_filter->next;
         }
-        rv = output_filter->frec->filter_func.out_func(output_filter, bb);
+        rv = output_filter->frec->filter_func.out_func(output_filter, NULL);
         if (rv != APR_SUCCESS) {
-            /* XXX log error */
+            ap_log_error(APLOG_MARK, APLOG_WARNING, rv, ap_server_conf,
+                     "core output filter");
             cs->state = CONN_STATE_LINGER;
         }
         else if (c->data_in_output_filters) {
-            cs->state = CONN_STATE_WRITE_COMPLETION;
+            /* Still in WRITE_COMPLETION_STATE:
+             * Set a write timeout for this connection, and let the
+             * event thread poll for writeability.
+             */
+            cs->expiration_time = ap_server_conf->timeout + time_now;
+            apr_thread_mutex_lock(timeout_mutex);
+            APR_RING_INSERT_TAIL(&timeout_head, cs, conn_state_t, timeout_list);
+            apr_thread_mutex_unlock(timeout_mutex);
+            pt->status = 0;
+            cs->pfd.reqevents = APR_POLLOUT | APR_POLLHUP | APR_POLLERR;
+            rc = apr_pollset_add(event_pollset, &cs->pfd);
+            return 1;
         }
         else if (c->keepalive != AP_CONN_KEEPALIVE || c->aborted || 
             ap_graceful_stop_signalled()) {
@@ -693,11 +704,12 @@ read_request:
          */
         cs->expiration_time = ap_server_conf->keep_alive_timeout + time_now;
         apr_thread_mutex_lock(timeout_mutex);
-        APR_RING_INSERT_TAIL(&timeout_head, cs, conn_state_t, timeout_list);
+        APR_RING_INSERT_TAIL(&keepalive_timeout_head, cs, conn_state_t, timeout_list);
         apr_thread_mutex_unlock(timeout_mutex);
 
         pt->status = 0;
-        /* Add work to pollset. These are always read events */
+        /* Add work to pollset. */
+        cs->pfd.reqevents = APR_POLLIN;
         rc = apr_pollset_add(event_pollset, &cs->pfd);
 
         if (rc != APR_SUCCESS) {
@@ -874,6 +886,7 @@ static void *listener_thread(apr_thread_t * thd, void *dummy)
     }
 
     APR_RING_INIT(&timeout_head, conn_state_t, timeout_list);
+    APR_RING_INIT(&keepalive_timeout_head, conn_state_t, timeout_list);
 
     /* Create the main pollset */
     rc = apr_pollset_create(&event_pollset,
@@ -942,6 +955,8 @@ static void *listener_thread(apr_thread_t * thd, void *dummy)
                 case CONN_STATE_CHECK_REQUEST_LINE_READABLE:
                     cs->state = CONN_STATE_READ_REQUEST_LINE;
                     break;
+                case CONN_STATE_WRITE_COMPLETION:
+                    break;
                 default:
                     ap_log_error(APLOG_MARK, APLOG_ERR, rc,
                                  ap_server_conf,
@@ -953,6 +968,7 @@ static void *listener_thread(apr_thread_t * thd, void *dummy)
                 apr_thread_mutex_lock(timeout_mutex);
                 APR_RING_REMOVE(cs, timeout_list);
                 apr_thread_mutex_unlock(timeout_mutex);
+                APR_RING_ELEM_INIT(cs, timeout_list);
 
                 rc = push2worker(out_pfd, event_pollset);
                 if (rc != APR_SUCCESS) {
@@ -1037,9 +1053,10 @@ static void *listener_thread(apr_thread_t * thd, void *dummy)
         /* handle timed out sockets */
         apr_thread_mutex_lock(timeout_mutex);
 
-        cs = APR_RING_FIRST(&timeout_head);
+        /* Step 1: keepalive timeouts */
+        cs = APR_RING_FIRST(&keepalive_timeout_head);
         timeout_time = time_now + TIMEOUT_FUDGE_FACTOR;
-        while (!APR_RING_EMPTY(&timeout_head, conn_state_t, timeout_list)
+        while (!APR_RING_EMPTY(&keepalive_timeout_head, conn_state_t, timeout_list)
                && cs->expiration_time < timeout_time
                && get_worker(&have_idle_worker)) {
 
@@ -1058,8 +1075,25 @@ static void *listener_thread(apr_thread_t * thd, void *dummy)
                  */
             }
             have_idle_worker = 0;
+            cs = APR_RING_FIRST(&keepalive_timeout_head);
+        }
+
+        /* Step 2: write completion timeouts */
+        cs = APR_RING_FIRST(&timeout_head);
+        while (!APR_RING_EMPTY(&timeout_head, conn_state_t, timeout_list)
+               && cs->expiration_time < timeout_time
+               && get_worker(&have_idle_worker)) {
+
+            cs->state = CONN_STATE_LINGER;
+            APR_RING_REMOVE(cs, timeout_list);
+            rc = push2worker(&cs->pfd, event_pollset);
+            if (rc != APR_SUCCESS) {
+                return NULL;
+            }
+            have_idle_worker = 0;
             cs = APR_RING_FIRST(&timeout_head);
         }
+
         apr_thread_mutex_unlock(timeout_mutex);
 
     }     /* listener main loop */