From: Paul Querna <pquerna@apache.org>
Date: Sat, 28 Mar 2009 01:02:32 +0000 (+0000)
Subject: If the MPM supports serf in its core, use this to drive the proxied request
X-Git-Tag: 2.3.3~777
X-Git-Url: https://granicus.if.org/sourcecode?a=commitdiff_plain;h=c9c5ae3ce6575e03f19a39132746776743664e31;p=apache

If the MPM supports serf in its core, use this to drive the proxied request
completely async from the original thread.

Right now it uses a 100ms timer to determine when the request is actaully complete,
but I think this can be removed in the long run with a better mechanism.


git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/trunk@759414 13f79535-47bb-0310-9956-ffa450edef68
---

diff --git a/modules/proxy/mod_serf.c b/modules/proxy/mod_serf.c
index 8d53530099..667b07f862 100644
--- a/modules/proxy/mod_serf.c
+++ b/modules/proxy/mod_serf.c
@@ -25,9 +25,10 @@
 #include "serf.h"
 #include "apr_uri.h"
 #include "apr_strings.h"
-
+#include "ap_mpm.h"
 
 module AP_MODULE_DECLARE_DATA serf_module;
+static int mpm_supprts_serf = 0;
 
 typedef struct {
     int on;
@@ -325,13 +326,39 @@ static apr_status_t setup_request(serf_request_t *request,
     return APR_SUCCESS;
 }
 
+static void 
+timed_callback(void *baton)
+{
+    s_baton_t *ctx = baton;
+
+    if (ctx->keep_reading) {
+        ap_mpm_register_timed_callback(apr_time_from_msec(100), timed_callback, baton);
+    }
+    else if (ctx->rstatus) {
+        ap_log_rerror(APLOG_MARK, APLOG_ERR, ctx->rstatus, ctx->r,
+                      "serf: request returned: %d", ctx->rstatus);
+        ctx->r->status = HTTP_OK;
+        ap_die(ctx->rstatus, ctx->r);
+    }
+    else {
+        ap_finalize_request_protocol(ctx->r);
+        ap_process_request_after_handler(ctx->r);
+        return;
+    }
+}
+
+
+#ifndef apr_time_from_msec
+#define apr_time_from_msec(x) (x * 1000)
+#endif
+
 /* TOOD: rewrite drive_serf to make it async */
 static int drive_serf(request_rec *r, serf_config_t *conf)
 {
     apr_status_t rv;
     apr_pool_t *pool = r->pool;
     apr_sockaddr_t *address;
-    s_baton_t baton;
+    s_baton_t *baton = apr_palloc(r->pool, sizeof(s_baton_t));
     /* XXXXX: make persistent/per-process or something.*/
     serf_context_t *serfme;
     serf_connection_t *conn;
@@ -409,49 +436,71 @@ static int drive_serf(request_rec *r, serf_config_t *conf)
         return HTTP_INTERNAL_SERVER_ERROR;
     }
 
-    serfme = serf_context_create(pool);
+    if (mpm_supprts_serf) {
+        serfme = ap_lookup_provider("mpm_serf", "instance", "0");
+        if (!serfme) {
+            ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, r, "mpm lied to us about supporting serf.");
+            return HTTP_INTERNAL_SERVER_ERROR;
+        }
+    }
+    else {
+        serfme = serf_context_create(pool);
+    }
 
-    baton.r = r;
-    baton.conf = conf;
-    baton.bkt_alloc = serf_bucket_allocator_create(pool, NULL, NULL);
-    baton.ssl_ctx = NULL;
-    baton.rstatus = OK;
+    baton->r = r;
+    baton->conf = conf;
+    baton->bkt_alloc = serf_bucket_allocator_create(pool, NULL, NULL);
+    baton->ssl_ctx = NULL;
+    baton->rstatus = OK;
 
-    baton.done_headers = 0;
-    baton.keep_reading = 1;
+    baton->done_headers = 0;
+    baton->keep_reading = 1;
 
     if (strcasecmp(conf->url.scheme, "https") == 0) {
-        baton.want_ssl = 1;
+        baton->want_ssl = 1;
     }
     else {
-        baton.want_ssl = 0;
+        baton->want_ssl = 0;
     }
 
     conn = serf_connection_create(serfme, address,
-                                  conn_setup, &baton,
-                                  closed_connection, &baton,
+                                  conn_setup, baton,
+                                  closed_connection, baton,
                                   pool);
 
     srequest = serf_connection_request_create(conn, setup_request,
-                                              &baton);
+                                              baton);
 
-    do {
-        rv = serf_context_run(serfme, SERF_DURATION_FOREVER, pool);
-
-        /* XXXX: Handle timeouts */
-        if (APR_STATUS_IS_TIMEUP(rv)) {
-            continue;
-        }
+    if (mpm_supprts_serf) {
 
+        rv = ap_mpm_register_timed_callback(apr_time_from_msec(100), timed_callback, baton);
+        
         if (rv != APR_SUCCESS) {
-            ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, r, "serf_context_run()");
+            ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, r, "ap_mpm_register_timed_callback failed.");
             return HTTP_INTERNAL_SERVER_ERROR;       
         }
 
-        serf_debug__closed_conn(baton.bkt_alloc);
-    } while (baton.keep_reading);
-
-    return baton.rstatus;
+        return SUSPENDED;
+    }
+    else {
+        do {
+            rv = serf_context_run(serfme, SERF_DURATION_FOREVER, pool);
+            
+            /* XXXX: Handle timeouts */
+            if (APR_STATUS_IS_TIMEUP(rv)) {
+                continue;
+            }
+            
+            if (rv != APR_SUCCESS) {
+                ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, r, "serf_context_run()");
+                return HTTP_INTERNAL_SERVER_ERROR;       
+            }
+            
+            serf_debug__closed_conn(baton->bkt_alloc);
+        } while (baton->keep_reading);
+        
+        return baton->rstatus;
+    }
 }
 
 static int serf_handler(request_rec *r)
@@ -860,6 +909,13 @@ static const ap_serf_cluster_provider_t builtin_heartbeat =
 
 static void register_hooks(apr_pool_t *p)
 {
+    apr_status_t rv;
+    rv = ap_mpm_query(AP_MPMQ_HAS_SERF, &mpm_supprts_serf);
+
+    if (rv != APR_SUCCESS) {
+        mpm_supprts_serf = 0;
+    }
+    
     ap_register_provider(p, AP_SERF_CLUSTER_PROVIDER,
                          "heartbeat", "0", &builtin_heartbeat);
 
diff --git a/modules/proxy/mod_serf.h b/modules/proxy/mod_serf.h
index 80f40c6b5e..de0e3abe4c 100644
--- a/modules/proxy/mod_serf.h
+++ b/modules/proxy/mod_serf.h
@@ -23,6 +23,14 @@
 
 #include "httpd.h"
 #include "http_config.h"
+#include "ap_config_auto.h"
+#ifdef HAVE_SERF_H
+#include "serf.h"
+#ifndef AP_HAS_SERF
+#define AP_HAS_SERF 1
+#endif
+#endif
+
 #include "ap_provider.h"
 
 #ifndef _MOD_SERF_H_