Introduce Suspendable Requests to the Event MPM.

Using this basic framework, you can return SUSPENDED from an HTTP Handler,
and then register a callback that is invoked by the MPM at a later time.

This initial version only supports _timers_ as callbacks, but in the future I
would like to add things like wait for socket activity, on a socket specified by
the handler.

Once in a callback, It is then the responsibility of the callback fucntion 
to finish the HTTP Request handling, but this alows you to do cool things like 
a fully async proxy, COMET support, or even rate limiting.

To prove I'm not insane, I've inlcuded an example module, mod_dialup.

You can configure it like this:
<Location "/docs">
  ModemStandard "V.32"
</Location>

And for static files inside that path, you will be rate limited to V.32 speeds, 
aka 9.6 kilobits/second.

Does anyone besides Rüdiger read commit emails :-) ?

I know there are likely huge problems with this, but I would like to see how far
we can push the Event MPM, figure out what to do better, if there is anything, 
and then really dive into the 3.0 development before ApacheCon.

* server/mpm/experimental/event/fdqueue.h:
    (timer_event_t): New structure to hold timer events and callback functions.
    
* server/mpm/experimental/event/fdqueue.c
    (ap_queue_empty): Modify to also look at Timer Ring.

    (ap_queue_init): Initialize Timer Ring.

    (ap_queue_push_timer): New function, pushes a timer event into the queue.

    (ap_queue_pop_something): Renamed function, returns a timer event or
        a socket/pool for a worker thread to run.


* server/mpm/experimental/event/event.c
    (process_socket): If the connection is in SUSPENDED state, don't force it
        into linger mode yet, the callback will have to take care of that.

    (push_timer2worker): New shortcut function, pushes timer event into queue
        for a worker to run.

    (timer_free_ring): New global data structure to recycle memory used by 
        timer events.

    (timer_ring): New global data structure to hold active timer events.

    (g_timer_ring_mtx): Thread mutex to protect timer event data structures.

    (ap_mpm_register_timed_callback): New Function, registers a callback to be
        invoked by the MPM at a later time.

    (listener_thread): Calculate our wakeup time based on the upcoming Event 
        Queue, and after pollset_poll runs, push any Timers that have passed
        onto worker threads to run.
    
    (worker_thread): Call new queue pop method, and if the Timer Event is 
        non-null, invoke the callback.  Once the callback is done, push the
        structure onto the timer_free_ring, to be recycled.

    (child_main): Initialize new mutex and ring structures.


* server/config.c
    (ap_invoke_handler): Allow SUSPENDED aa valid return code from handlers.


* modules/http/http_core.c
    (ap_process_http_async_connection): Don't close the connection when in 
        SUSPENDED state.


* modules/http/http_request.c
    (ap_process_request_after_handler): New function, body pulled from the old,
        ap_process_async_request.  Split to let handlers invoke this so they 
        don't need to know all of the details of finishing a request.

    (ap_process_async_request): If the handler returns SUSPENDED, don't do
        anything but return.


* include/ap_mmn.h: Bump MMN.


* include/ap_mpm.h
    (ap_mpm_register_timed_callback): New function.


* include/httpd.h:
    (SUSPENDED): New return code for handlers.
    (request_rec::invoke_mtx): New mutex to protect callback invokcations
        from being run before the original handler finishes running.
    (conn_state_e): Add a suspended state.


* include/http_request.h
    (ap_process_request_after_handler): New function to make it easier for 
        handlers to finish the HTTP Request.


* modules/test/config.m4: Add mod_dialup to build.


* modules/test/mod_dialup.c: New rate limiting module, requires the Event MPM 
    to work.




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

2 files changed, 310 insertions, 0 deletions

diff --git a/modules/test/config.m4 b/modules/test/config.m4
index 01bc0fa971..9c150f488f 100644
--- a/modules/test/config.m4
+++ b/modules/test/config.m4
@@ -6,4 +6,6 @@ APACHE_MODULE(optional_hook_import, example optional hook importer, , , no)
 APACHE_MODULE(optional_fn_import, example optional function importer, , , no)
 APACHE_MODULE(optional_fn_export, example optional function exporter, , , no)
 
+APACHE_MODULE(dialup, rate limits static files to dialup modem speeds, , , no)
+
 APACHE_MODPATH_FINISH
diff --git a/modules/test/mod_dialup.c b/modules/test/mod_dialup.c
new file mode 100644
index 0000000000..349cd2c6e6
--- /dev/null
+++ b/modules/test/mod_dialup.c
@@ -0,0 +1,308 @@
+/* Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+
+#include "httpd.h"
+#include "util_filter.h"
+#include "http_log.h"
+#include "http_config.h"
+#include "http_request.h"
+
+/* to detect sendfile enabled, we need CORE_PRIVATE. Someone should fix this. */
+#define CORE_PRIVATE
+#include "http_core.h"
+
+
+module AP_MODULE_DECLARE_DATA dialup_module;
+
+#ifndef apr_time_from_msec
+#define apr_time_from_msec(x) (x * 1000)
+#endif
+
+
+typedef struct dialup_dcfg_t {
+    apr_size_t bytes_per_second;
+} dialup_dcfg_t;
+
+typedef struct dialup_baton_t {
+    apr_size_t bytes_per_second;
+    request_rec *r;
+    apr_file_t *fd;
+    apr_bucket_brigade *bb;
+    apr_bucket_brigade *tmpbb;
+} dialup_baton_t;
+
+static int
+dialup_send_pulse(dialup_baton_t *db)
+{
+    int status;
+    apr_off_t len = 0;
+    apr_size_t bytes_sent = 0;
+    
+    while (!APR_BRIGADE_EMPTY(db->bb) && bytes_sent < db->bytes_per_second) {
+        apr_bucket *e;
+
+        if (db->r->connection->aborted) {
+            return HTTP_INTERNAL_SERVER_ERROR;
+        }
+        
+        status = apr_brigade_partition(db->bb, db->bytes_per_second, &e);
+
+        if (status != APR_SUCCESS && status != APR_INCOMPLETE) {
+            /* XXXXXX: Log me. */
+            return HTTP_INTERNAL_SERVER_ERROR;
+        }
+
+        if (e != APR_BRIGADE_SENTINEL(db->bb)) {
+            apr_bucket *f;
+            apr_bucket *b = APR_BUCKET_PREV(e);
+            f = APR_RING_FIRST(&db->bb->list);
+            APR_RING_UNSPLICE(f, b, link);
+            APR_RING_SPLICE_HEAD(&db->tmpbb->list, f, b, apr_bucket, link);
+        }
+        else {
+            APR_BRIGADE_CONCAT(db->tmpbb, db->bb);
+        }
+        
+        e = apr_bucket_flush_create(db->r->connection->bucket_alloc);
+        
+        APR_BRIGADE_INSERT_TAIL(db->tmpbb, e);
+
+        apr_brigade_length(db->tmpbb, 1, &len);
+        bytes_sent += len;
+        status = ap_pass_brigade(db->r->output_filters, db->tmpbb);
+
+        apr_brigade_cleanup(db->tmpbb);
+
+        if (status != OK) {
+            ap_log_rerror(APLOG_MARK, APLOG_ERR, status, db->r,
+                          "dialup: pulse: ap_pass_brigade failed:");
+            return status;
+        }
+    }
+
+    if (APR_BRIGADE_EMPTY(db->bb)) {
+        return DONE;
+    }
+    else {
+        return SUSPENDED;
+    }
+}
+
+void 
+dialup_callback(void *baton)
+{
+    int status;
+    dialup_baton_t *db = (dialup_baton_t *)baton;
+
+    apr_thread_mutex_lock(db->r->invoke_mtx);
+
+    status = dialup_send_pulse(db);
+
+    if (status == SUSPENDED) {
+        ap_mpm_register_timed_callback(apr_time_from_sec(1), dialup_callback, baton);
+    }
+    else if (status == DONE) {
+        apr_thread_mutex_unlock(db->r->invoke_mtx);
+        ap_finalize_request_protocol(db->r);
+        ap_process_request_after_handler(db->r);
+        return;
+    }
+    else {
+        ap_log_rerror(APLOG_MARK, APLOG_ERR, status, db->r,
+                      "dialup: pulse returned: %d", status);
+        db->r->status = HTTP_OK;
+        ap_die(status, db->r);
+    }
+
+    apr_thread_mutex_unlock(db->r->invoke_mtx);
+}
+
+static int
+dialup_handler(request_rec *r)
+{
+    int status;
+    apr_status_t rv;
+
+    /* See core.c, default handler for all of the cases we just decline. */
+    if (r->method_number != M_GET || 
+        r->finfo.filetype == 0 || 
+        r->finfo.filetype == APR_DIR) {
+        return DECLINED;
+    }
+
+    dialup_dcfg_t *dcfg = ap_get_module_config(r->per_dir_config,
+                                               &dialup_module);
+    if (dcfg->bytes_per_second == 0) {
+        return DECLINED;
+    }
+    core_dir_config *ccfg = ap_get_module_config(r->per_dir_config,
+                                                     &core_module);
+    
+    apr_file_t *fd;
+
+    rv = apr_file_open(&fd, r->filename, APR_READ | APR_BINARY
+#if APR_HAS_SENDFILE
+                           | ((ccfg->enable_sendfile == ENABLE_SENDFILE_OFF)
+                              ? 0 : APR_SENDFILE_ENABLED)
+#endif
+                       , 0, r->pool);
+
+    if (rv) {
+        return DECLINED;
+    }
+
+    /* copied from default handler: */
+    ap_update_mtime(r, r->finfo.mtime);
+    ap_set_last_modified(r);
+    ap_set_etag(r);
+    apr_table_setn(r->headers_out, "Accept-Ranges", "bytes");
+    ap_set_content_length(r, r->finfo.size);
+
+    status = ap_meets_conditions(r);
+    if (status != OK) {
+        ap_log_rerror(APLOG_MARK, APLOG_ERR, 0, r,
+                      "dialup: declined, meets conditions, good luck core handler");
+        return DECLINED;
+    }
+
+    apr_bucket_brigade *bb;
+
+    dialup_baton_t *db = apr_palloc(r->pool, sizeof(dialup_baton_t));
+    
+    db->bb = apr_brigade_create(r->pool, r->connection->bucket_alloc);
+    db->tmpbb = apr_brigade_create(r->pool, r->connection->bucket_alloc);
+
+    apr_bucket *e;
+
+    e = apr_brigade_insert_file(db->bb, fd, 0, r->finfo.size, r->pool);
+
+#if APR_HAS_MMAP
+    if (ccfg->enable_mmap == ENABLE_MMAP_OFF) {
+        apr_bucket_file_enable_mmap(e, 0);
+    }
+#endif
+    
+    
+    db->bytes_per_second = dcfg->bytes_per_second;
+    db->r = r;
+    db->fd = fd;
+
+    e = apr_bucket_eos_create(r->connection->bucket_alloc);
+
+    APR_BRIGADE_INSERT_TAIL(db->bb, e);
+
+    status = dialup_send_pulse(db);
+    if (status != SUSPENDED && status != DONE) {
+        ap_log_rerror(APLOG_MARK, APLOG_ERR, 0, r,
+                      "dialup: failed, send pulse");
+        return status;
+    }
+
+    ap_mpm_register_timed_callback(apr_time_from_sec(1), dialup_callback, db);
+
+    return SUSPENDED;
+}
+
+
+
+#ifndef APR_HOOK_ALMOST_LAST
+#define APR_HOOK_ALMOST_LAST (APR_HOOK_REALLY_LAST - 1)
+#endif
+
+static void
+dialup_register_hooks(apr_pool_t *p)
+{
+    ap_hook_handler(dialup_handler, NULL, NULL, APR_HOOK_ALMOST_LAST);
+}
+
+typedef struct modem_speed_t {
+    const char *name;
+    apr_size_t bytes_per_second;
+} modem_speed_t;
+
+#ifndef BITRATE_TO_BYTES
+#define BITRATE_TO_BYTES(x) ((1000 * x)/8)
+#endif
+
+static const modem_speed_t modem_bitrates[] =
+{
+    {"V.21",    BITRATE_TO_BYTES(0.1)},
+    {"V.26bis", BITRATE_TO_BYTES(2.4)},
+    {"V.32",    BITRATE_TO_BYTES(9.6)},
+    {"V.34",    BITRATE_TO_BYTES(28.8)},
+    {"V.92",    BITRATE_TO_BYTES(56.0)},
+    {"i-was-rich-and-got-a-leased-line", BITRATE_TO_BYTES(1500)},
+    {NULL, 0}
+};
+
+static const char *
+cmd_modem_standard(cmd_parms *cmd,
+             void *dconf,
+             const char *input)
+{
+    const modem_speed_t *standard;
+    int i = 0;
+    dialup_dcfg_t *dcfg = (dialup_dcfg_t*)dconf;
+    
+    dcfg->bytes_per_second = 0;
+
+    while (modem_bitrates[i].name != NULL) {
+        standard = &modem_bitrates[i];
+        if (strcasecmp(standard->name, input) == 0) {
+            dcfg->bytes_per_second = standard->bytes_per_second;
+            break;
+        }
+        i++;
+    }
+
+    if (dcfg->bytes_per_second == 0) {
+        return "mod_diaulup: Unkonwn Modem Standard specified.";
+    }
+
+    return NULL;
+}
+
+static void *
+dialup_dcfg_create(apr_pool_t *p, char *dummy)
+{
+    dialup_dcfg_t *cfg = apr_palloc(p, sizeof(dialup_dcfg_t));
+    
+    cfg->bytes_per_second = 0;
+
+    return cfg;
+}
+
+
+static const command_rec dialup_cmds[] =
+{
+    AP_INIT_TAKE1("ModemStandard", cmd_modem_standard, NULL, ACCESS_CONF,
+                  "Modem Standard to.. simulate. "
+                  "Must be one of: 'V.21', 'V.26bis', 'V.32', 'V.34', or 'V.92'"),
+    NULL
+};
+
+module AP_MODULE_DECLARE_DATA dialup_module =
+{
+    STANDARD20_MODULE_STUFF,
+    dialup_dcfg_create,
+    NULL,
+    NULL,
+    NULL,
+    dialup_cmds,
+    dialup_register_hooks
+};