core: Extend support for setting aside data from the network input filter

to any connection or request input filter.


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

CHANGES

@@ -1,6 +1,9 @@
                                                          -*- coding: utf-8 -*-
 Changes with Apache 2.5.0
 
+  *) core: Extend support for setting aside data from the network input filter
+     to any connection or request input filter. [Graham Leggett]
+
   *) mod_ssl: Add "no_crl_for_cert_ok" flag to SSLCARevocationCheck directive
      to opt-in previous behaviour (2.2) with CRLs verification when checking
      certificate(s) with no corresponding CRL.  [Yann Ylavic]

include/ap_mmn.h

@@ -507,14 +507,17 @@
  * 20150222.12 (2.5.0-dev) Add complete_connection hook,
  *                         ap_filter_complete_connection().
  * 20150222.13 (2.5.0-dev) Add ap_create_request().
+ * 20160312.0 (2.5.0-dev)  Rename complete_connection to output_pending,
+ *                         add ap_filter_input_pending(),
+ *                         ap_filter_prepare_brigade(), ap_filter_direction_e
  */
 
 #define MODULE_MAGIC_COOKIE 0x41503235UL /* "AP25" */
 
 #ifndef MODULE_MAGIC_NUMBER_MAJOR
-#define MODULE_MAGIC_NUMBER_MAJOR 20150222
+#define MODULE_MAGIC_NUMBER_MAJOR 20160312
 #endif
-#define MODULE_MAGIC_NUMBER_MINOR 13                 /* 0...n */
+#define MODULE_MAGIC_NUMBER_MINOR 0                 /* 0...n */
 
 /**
  * Determine if the server's current MODULE_MAGIC_NUMBER is at least a

include/httpd.h

@@ -1151,7 +1151,7 @@ struct conn_rec {
     struct apr_bucket_alloc_t *bucket_alloc;
     /** The current state of this connection; may be NULL if not used by MPM */
     conn_state_t *cs;
-    /** Is there data pending in the input filters? */
+    /** No longer used, replaced with ap_filter_input_pending() */
     int data_in_input_filters;
     /** No longer used, replaced with ap_filter_should_yield() */
     int data_in_output_filters;

include/mpm_common.h

@@ -465,7 +465,15 @@ AP_DECLARE_HOOK(const char *,mpm_get_name,(void))
  * should end gracefully, or a positive error if we should begin to linger.
  * @ingroup hooks
  */
-AP_DECLARE_HOOK(int, complete_connection, (conn_rec *c))
+AP_DECLARE_HOOK(int, output_pending, (conn_rec *c))
+
+/**
+ * Hook called to determine whether any data is pending in the input filters.
+ * @param c The current connection
+ * @return OK if we can read without blocking, DECLINED if a read would block.
+ * @ingroup hooks
+ */
+AP_DECLARE_HOOK(int, input_pending, (conn_rec *c))
 
 /**
  * Notification that connection handling is suspending (disassociating from the

include/util_filter.h

@@ -183,6 +183,17 @@ typedef enum {
     AP_FTYPE_NETWORK     = 60
 } ap_filter_type;
 
+/**
+ * These flags indicate whether the given filter is an input filter or an
+ * output filter.
+ */
+typedef enum {
+    /** Input filters */
+    AP_FILTER_INPUT     = 1,
+    /** Output filters */
+    AP_FILTER_OUTPUT    = 2,
+} ap_filter_direction_e;
+
 /**
  * This is the request-time context structure for an installed filter (in
  * the output filter chain). It provides the callback to use for filtering,
@@ -247,6 +258,9 @@ struct ap_filter_rec_t {
 
     /** Protocol flags for this filter */
     unsigned int proto_flags;
+
+    /** Whether the filter is an input or output filter */
+    ap_filter_direction_e direction;
 };
 
 /**
@@ -542,6 +556,19 @@ AP_DECLARE(apr_status_t) ap_save_brigade(ap_filter_t *f,
                                          apr_bucket_brigade **save_to,
                                          apr_bucket_brigade **b, apr_pool_t *p);
 
+/**
+ * Prepare the filter to allow brigades to be set aside. This can be used
+ * within an input filter to allocate space to set aside data in the input
+ * filters, or can be used within an output filter by being called via
+ * ap_filter_setaside_brigade().
+ * @param f The current filter
+ * @param pool The pool that was used to create the brigade. In a request
+ * filter this will be the request pool, in a connection filter this will
+ * be the connection pool.
+ * @returns OK if a brigade was created, DECLINED otherwise.
+ */
+AP_DECLARE(int) ap_filter_prepare_brigade(ap_filter_t *f, apr_pool_t **p);
+
 /**
  * Prepare a bucket brigade to be setaside, creating a dedicated pool if
  * necessary within the filter to handle the lifetime of the setaside brigade.
@@ -599,7 +626,18 @@ AP_DECLARE(int) ap_filter_should_yield(ap_filter_t *f);
  * If some unwritten data remains, this function returns OK. If any
  * attempt to write data failed, this functions returns a positive integer.
  */
-AP_DECLARE(int) ap_filter_complete_connection(conn_rec *c);
+AP_DECLARE(int) ap_filter_output_pending(conn_rec *c);
+
+/**
+ * This function determines whether there is pending data in the input
+ * filters. Pending data is data that has been read from the underlying
+ * socket but not yet returned to the application.
+ *
+ * @param c The connection.
+ * @return If no pending data remains, this function returns DECLINED.
+ * If some pending data remains, this function returns OK.
+ */
+AP_DECLARE(int) ap_filter_input_pending(conn_rec *c);
 
 /**
  * Flush function for apr_brigade_* calls.  This calls ap_pass_brigade

modules/http/http_core.c

@@ -254,6 +254,7 @@ static int ap_process_http_connection(conn_rec *c)
 
 static int http_create_request(request_rec *r)
 {
+    /* FIXME: we must only add these filters if we are an HTTP request */
     if (!r->main && !r->prev) {
         ap_add_output_filter_handle(ap_byterange_filter_handle,
                                     NULL, r, r->connection);

modules/http/http_request.c

@@ -39,6 +39,7 @@
 #include "http_protocol.h"
 #include "http_log.h"
 #include "http_main.h"
+#include "mpm_common.h"
 #include "util_filter.h"
 #include "util_charset.h"
 #include "scoreboard.h"
@@ -236,7 +237,6 @@ static void check_pipeline(conn_rec *c, apr_bucket_brigade *bb)
     apr_size_t cr = 0;
     char buf[2];
 
-    c->data_in_input_filters = 0;
     while (c->keepalive != AP_CONN_CLOSE && !c->aborted) {
         apr_size_t len = cr + 1;
 
@@ -276,7 +276,6 @@ static void check_pipeline(conn_rec *c, apr_bucket_brigade *bb)
                  * where this possible failure comes from (metadata,
                  * morphed EOF socket => empty bucket? debug only here).
                  */
-                c->data_in_input_filters = 1;
                 log_level = APLOG_DEBUG;
             }
             ap_log_cerror(APLOG_MARK, log_level, rv, c, APLOGNO(02968)
@@ -295,7 +294,6 @@ static void check_pipeline(conn_rec *c, apr_bucket_brigade *bb)
                 num_blank_lines--;
             }
             else {
-                c->data_in_input_filters = 1;
                 break;
             }
         }
@@ -308,7 +306,6 @@ static void check_pipeline(conn_rec *c, apr_bucket_brigade *bb)
                 cr = 1;
             }
             else {
-                c->data_in_input_filters = 1;
                 break;
             }
         }
@@ -452,7 +449,7 @@ AP_DECLARE(void) ap_process_request(request_rec *r)
 
     ap_process_async_request(r);
 
-    if (!c->data_in_input_filters) {
+    if (ap_run_input_pending(c) != OK) {
         bb = apr_brigade_create(c->pool, c->bucket_alloc);
         b = apr_bucket_flush_create(c->bucket_alloc);
         APR_BRIGADE_INSERT_HEAD(bb, b);

server/core.c

@@ -5559,8 +5559,8 @@ static void register_hooks(apr_pool_t *p)
     ap_hook_open_htaccess(ap_open_htaccess, NULL, NULL, APR_HOOK_REALLY_LAST);
     ap_hook_optional_fn_retrieve(core_optional_fn_retrieve, NULL, NULL,
                                  APR_HOOK_MIDDLE);
-    ap_hook_complete_connection(ap_filter_complete_connection, NULL, NULL,
+    ap_hook_output_pending(ap_filter_output_pending, NULL, NULL,
-                           APR_HOOK_MIDDLE);
+            APR_HOOK_MIDDLE);
 
     /* register the core's insert_filter hook and register core-provided
      * filters

server/core_filters.c

@@ -84,7 +84,6 @@ struct core_output_filter_ctx {
 };
 
 struct core_filter_ctx {
-    apr_bucket_brigade *b;
     apr_bucket_brigade *tmpbb;
 };
 
@@ -116,19 +115,19 @@ apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
     if (!ctx)
     {
         net->in_ctx = ctx = apr_palloc(f->c->pool, sizeof(*ctx));
-        ctx->b = apr_brigade_create(f->c->pool, f->c->bucket_alloc);
+        ap_filter_prepare_brigade(f, NULL);
         ctx->tmpbb = apr_brigade_create(f->c->pool, f->c->bucket_alloc);
         /* seed the brigade with the client socket. */
-        rv = ap_run_insert_network_bucket(f->c, ctx->b, net->client_socket);
+        rv = ap_run_insert_network_bucket(f->c, f->bb, net->client_socket);
         if (rv != APR_SUCCESS)
             return rv;
     }
-    else if (APR_BRIGADE_EMPTY(ctx->b)) {
+    else if (APR_BRIGADE_EMPTY(f->bb)) {
         return APR_EOF;
     }
 
     /* ### This is bad. */
-    BRIGADE_NORMALIZE(ctx->b);
+    BRIGADE_NORMALIZE(f->bb);
 
     /* check for empty brigade again *AFTER* BRIGADE_NORMALIZE()
      * If we have lost our socket bucket (see above), we are EOF.
@@ -136,13 +135,13 @@ apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
      * Ideally, this should be returning SUCCESS with EOS bucket, but
      * some higher-up APIs (spec. read_request_line via ap_rgetline)
      * want an error code. */
-    if (APR_BRIGADE_EMPTY(ctx->b)) {
+    if (APR_BRIGADE_EMPTY(f->bb)) {
         return APR_EOF;
     }
 
     if (mode == AP_MODE_GETLINE) {
         /* we are reading a single LF line, e.g. the HTTP headers */
-        rv = apr_brigade_split_line(b, ctx->b, block, HUGE_STRING_LEN);
+        rv = apr_brigade_split_line(b, f->bb, block, HUGE_STRING_LEN);
         /* We should treat EAGAIN here the same as we do for EOF (brigade is
          * empty).  We do this by returning whatever we have read.  This may
          * or may not be bogus, but is consistent (for now) with EOF logic.
@@ -170,10 +169,10 @@ apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
          * mean that there is another request, just a blank line.
          */
         while (1) {
-            if (APR_BRIGADE_EMPTY(ctx->b))
+            if (APR_BRIGADE_EMPTY(f->bb))
                 return APR_EOF;
 
-            e = APR_BRIGADE_FIRST(ctx->b);
+            e = APR_BRIGADE_FIRST(f->bb);
 
             rv = apr_bucket_read(e, &str, &len, APR_NONBLOCK_READ);
 
@@ -212,7 +211,7 @@ apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
         apr_bucket *e;
 
         /* Tack on any buckets that were set aside. */
-        APR_BRIGADE_CONCAT(b, ctx->b);
+        APR_BRIGADE_CONCAT(b, f->bb);
 
         /* Since we've just added all potential buckets (which will most
          * likely simply be the socket bucket) we know this is the end,
@@ -230,7 +229,7 @@ apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
 
         AP_DEBUG_ASSERT(readbytes > 0);
 
-        e = APR_BRIGADE_FIRST(ctx->b);
+        e = APR_BRIGADE_FIRST(f->bb);
         rv = apr_bucket_read(e, &str, &len, block);
 
         if (APR_STATUS_IS_EAGAIN(rv) && block == APR_NONBLOCK_READ) {
@@ -267,7 +266,7 @@ apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
             /* We already registered the data in e in len */
             e = APR_BUCKET_NEXT(e);
             while ((len < readbytes) && (rv == APR_SUCCESS)
-                   && (e != APR_BRIGADE_SENTINEL(ctx->b))) {
+                   && (e != APR_BRIGADE_SENTINEL(f->bb))) {
                 /* Check for the availability of buckets with known length */
                 if (e->length != -1) {
                     len += e->length;
@@ -295,22 +294,22 @@ apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
             readbytes = len;
         }
 
-        rv = apr_brigade_partition(ctx->b, readbytes, &e);
+        rv = apr_brigade_partition(f->bb, readbytes, &e);
         if (rv != APR_SUCCESS) {
             return rv;
         }
 
         /* Must do move before CONCAT */
-        ctx->tmpbb = apr_brigade_split_ex(ctx->b, e, ctx->tmpbb);
+        ctx->tmpbb = apr_brigade_split_ex(f->bb, e, ctx->tmpbb);
 
         if (mode == AP_MODE_READBYTES) {
-            APR_BRIGADE_CONCAT(b, ctx->b);
+            APR_BRIGADE_CONCAT(b, f->bb);
         }
         else if (mode == AP_MODE_SPECULATIVE) {
             apr_bucket *copy_bucket;
 
-            for (e = APR_BRIGADE_FIRST(ctx->b);
+            for (e = APR_BRIGADE_FIRST(f->bb);
-                 e != APR_BRIGADE_SENTINEL(ctx->b);
+                 e != APR_BRIGADE_SENTINEL(f->bb);
                  e = APR_BUCKET_NEXT(e))
             {
                 rv = apr_bucket_copy(e, &copy_bucket);
@@ -322,7 +321,7 @@ apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
         }
 
         /* Take what was originally there and place it back on ctx->b */
-        APR_BRIGADE_CONCAT(ctx->b, ctx->tmpbb);
+        APR_BRIGADE_CONCAT(f->bb, ctx->tmpbb);
     }
     return APR_SUCCESS;
 }

server/mpm/event/event.c

@@ -1151,7 +1151,7 @@ read_request:
 
         ap_update_child_status_from_conn(sbh, SERVER_BUSY_WRITE, c);
 
-        not_complete_yet = ap_run_complete_connection(c);
+        not_complete_yet = ap_run_output_pending(c);
 
         if (not_complete_yet > OK) {
             cs->pub.state = CONN_STATE_LINGER;
@@ -1177,7 +1177,7 @@ read_request:
             listener_may_exit) {
             cs->pub.state = CONN_STATE_LINGER;
         }
-        else if (c->data_in_input_filters) {
+        else if (ap_run_input_pending(c) == OK) {
             cs->pub.state = CONN_STATE_READ_REQUEST_LINE;
             goto read_request;
         }

server/mpm/motorz/motorz.c

@@ -402,7 +402,7 @@ read_request:
 
             ap_update_child_status_from_conn(scon->sbh, SERVER_BUSY_WRITE, c);
 
-            not_complete_yet = ap_run_complete_connection(c);
+            not_complete_yet = ap_run_output_pending(c);
 
             if (not_complete_yet > OK) {
                 scon->cs.state = CONN_STATE_LINGER;
@@ -433,7 +433,7 @@ read_request:
             else if (c->keepalive != AP_CONN_KEEPALIVE || c->aborted) {
                 scon->cs.state = CONN_STATE_LINGER;
             }
-            else if (c->data_in_input_filters) {
+            else if (ap_run_input_pending(c) == OK) {
                 scon->cs.state = CONN_STATE_READ_REQUEST_LINE;
                 goto read_request;
             }

server/mpm/simple/simple_io.c

@@ -96,7 +96,7 @@ static apr_status_t simple_io_process(simple_conn_t * scon)
             int not_complete_yet;
 
             ap_update_child_status_from_conn(c->sbh, SERVER_BUSY_WRITE, c);
-            not_complete_yet = ap_run_complete_connection(c);
+            not_complete_yet = ap_run_output_pending(c);
 
             if (not_complete_yet > OK) {
                 scon->cs.state = CONN_STATE_LINGER;
@@ -133,7 +133,7 @@ static apr_status_t simple_io_process(simple_conn_t * scon)
             else if (c->keepalive != AP_CONN_KEEPALIVE || c->aborted) {
                 scon->cs.state = CONN_STATE_LINGER;
             }
-            else if (c->data_in_input_filters) {
+            else if (ap_run_input_pending(c) == OK) {
                 scon->cs.state = CONN_STATE_READ_REQUEST_LINE;
             }
             else {

server/mpm_common.c

@@ -75,7 +75,8 @@
     APR_HOOK_LINK(mpm_resume_suspended) \
     APR_HOOK_LINK(end_generation) \
     APR_HOOK_LINK(child_status) \
-    APR_HOOK_LINK(complete_connection) \
+    APR_HOOK_LINK(output_pending) \
+    APR_HOOK_LINK(input_pending) \
     APR_HOOK_LINK(suspend_connection) \
     APR_HOOK_LINK(resume_connection)
 
@@ -118,7 +119,9 @@ AP_IMPLEMENT_HOOK_RUN_FIRST(apr_status_t, mpm_register_socket_callback_timeout,
 AP_IMPLEMENT_HOOK_RUN_FIRST(apr_status_t, mpm_unregister_socket_callback,
                             (apr_socket_t **s, apr_pool_t *p),
                             (s, p), APR_ENOTIMPL)
-AP_IMPLEMENT_HOOK_RUN_FIRST(int, complete_connection,
+AP_IMPLEMENT_HOOK_RUN_FIRST(int, output_pending,
+                            (conn_rec *c), (c), DECLINED)
+AP_IMPLEMENT_HOOK_RUN_FIRST(int, input_pending,
                             (conn_rec *c), (c), DECLINED)
 
 AP_IMPLEMENT_HOOK_VOID(end_generation,

server/util_filter.c

@@ -209,6 +209,7 @@ static ap_filter_rec_t *register_filter(const char *name,
                             ap_filter_func filter_func,
                             ap_init_filter_func filter_init,
                             ap_filter_type ftype,
+                            ap_filter_direction_e direction,
                             filter_trie_node **reg_filter_set)
 {
     ap_filter_rec_t *frec;
@@ -242,6 +243,7 @@ static ap_filter_rec_t *register_filter(const char *name,
     frec->filter_func = filter_func;
     frec->filter_init_func = filter_init;
     frec->ftype = ftype;
+    frec->direction = direction;
 
     apr_pool_cleanup_register(FILTER_POOL, NULL, filter_cleanup,
                               apr_pool_cleanup_null);
@@ -255,7 +257,7 @@ AP_DECLARE(ap_filter_rec_t *) ap_register_input_filter(const char *name,
 {
     ap_filter_func f;
     f.in_func = filter_func;
-    return register_filter(name, f, filter_init, ftype,
+    return register_filter(name, f, filter_init, ftype, AP_FILTER_INPUT,
                            &registered_input_filters);
 }
 
@@ -278,7 +280,7 @@ AP_DECLARE(ap_filter_rec_t *) ap_register_output_filter_protocol(
     ap_filter_rec_t* ret ;
     ap_filter_func f;
     f.out_func = filter_func;
-    ret = register_filter(name, f, filter_init, ftype,
+    ret = register_filter(name, f, filter_init, ftype, AP_FILTER_OUTPUT,
                           &registered_output_filters);
     ret->proto_flags = proto_flags ;
     return ret ;
@@ -702,6 +704,33 @@ static apr_status_t filters_cleanup(void *data)
     return APR_SUCCESS;
 }
 
+AP_DECLARE(int) ap_filter_prepare_brigade(ap_filter_t *f, apr_pool_t **p)
+{
+    apr_pool_t *pool;
+    ap_filter_t **key;
+
+    if (!f->bb) {
+
+        pool = f->r ? f->r->pool : f->c->pool;
+
+        key = apr_palloc(pool, sizeof(ap_filter_t **));
+        *key = f;
+        apr_hash_set(f->c->filters, key, sizeof(ap_filter_t **), f);
+
+        f->bb = apr_brigade_create(pool, f->c->bucket_alloc);
+
+        apr_pool_pre_cleanup_register(pool, key, filters_cleanup);
+
+        if (p) {
+            *p = pool;
+        }
+
+        return OK;
+    }
+
+    return DECLINED;
+}
+
 AP_DECLARE(apr_status_t) ap_filter_setaside_brigade(ap_filter_t *f,
         apr_bucket_brigade *bb)
 {
@@ -716,24 +745,11 @@ AP_DECLARE(apr_status_t) ap_filter_setaside_brigade(ap_filter_t *f,
     }
 
     if (!APR_BRIGADE_EMPTY(bb)) {
-        apr_pool_t *pool;
+        apr_pool_t *pool = NULL;
         /*
          * Set aside the brigade bb within f->bb.
          */
-        if (!f->bb) {
+        ap_filter_prepare_brigade(f, &pool);
-            ap_filter_t **key;
-
-            pool = f->r ? f->r->pool : f->c->pool;
-
-            key = apr_palloc(pool, sizeof(ap_filter_t **));
-            *key = f;
-            apr_hash_set(f->c->filters, key, sizeof(ap_filter_t **), f);
-
-            f->bb = apr_brigade_create(pool, f->c->bucket_alloc);
-
-            apr_pool_pre_cleanup_register(pool, key, filters_cleanup);
-
-        }
 
         /* decide what pool we setaside to, request pool or deferred pool? */
         if (f->r) {
@@ -953,7 +969,7 @@ AP_DECLARE(int) ap_filter_should_yield(ap_filter_t *f)
     return 0;
 }
 
-AP_DECLARE(int) ap_filter_complete_connection(conn_rec *c)
+AP_DECLARE(int) ap_filter_output_pending(conn_rec *c)
 {
     apr_hash_index_t *rindex;
     int data_in_output_filters = DECLINED;
@@ -962,7 +978,8 @@ AP_DECLARE(int) ap_filter_complete_connection(conn_rec *c)
     while (rindex) {
         ap_filter_t *f = apr_hash_this_val(rindex);
 
-        if (!APR_BRIGADE_EMPTY(f->bb)) {
+        if (f->frec->direction == AP_FILTER_OUTPUT && f->bb
+                && !APR_BRIGADE_EMPTY(f->bb)) {
 
             apr_status_t rv;
 
@@ -986,6 +1003,33 @@ AP_DECLARE(int) ap_filter_complete_connection(conn_rec *c)
     return data_in_output_filters;
 }
 
+AP_DECLARE(int) ap_filter_input_pending(conn_rec *c)
+{
+    apr_hash_index_t *rindex;
+
+    rindex = apr_hash_first(NULL, c->filters);
+    while (rindex) {
+        ap_filter_t *f = apr_hash_this_val(rindex);
+
+        if (f->frec->direction == AP_FILTER_INPUT && f->bb) {
+            apr_bucket *e = APR_BRIGADE_FIRST(f->bb);
+
+            /* if there is at least one non-morphing bucket
+             * in place, then we have data pending
+             */
+            if (e != APR_BRIGADE_SENTINEL(f->bb)
+                    && e->length != (apr_size_t)(-1)) {
+                return OK;
+            }
+
+        }
+
+        rindex = apr_hash_next(rindex);
+    }
+
+    return DECLINED;
+}
+
 AP_DECLARE_NONSTD(apr_status_t) ap_filter_flush(apr_bucket_brigade *bb,
                                                 void *ctx)
 {