Index: include/httpd.h =================================================================== --- include/httpd.h (revision 1622873) +++ include/httpd.h (working copy) @@ -1181,6 +1181,12 @@ /** Context under which this connection was suspended */ void *suspended_baton; + + /** Activity marker for this connection */ + unsigned int activity; + + /** Empty bucket brigade */ + apr_bucket_brigade *empty; }; struct conn_slave_rec { Index: include/util_filter.h =================================================================== --- include/util_filter.h (revision 1622873) +++ include/util_filter.h (working copy) @@ -533,6 +533,51 @@ apr_bucket_brigade **b, apr_pool_t *p); /** + * Prepare a bucket brigade to be setaside, creating a dedicated pool to handle + * the lifetime of the setaside brigade. If a non-empty amount of data was set + * aside, the counter indicating whether data exists in the output chain is + * incremented. + * @param f The current filter + * @param buffered_bb The bucket brigade to set aside. This bucket brigade will + * be created if it doesn't exist. + * @param deferred_write_pool Ensure that all data in the brigade lives as long + * as this pool. The pool will be created if it + * doesn't yet exist. + * @param bb The bucket brigade to set aside. This brigade is always empty + * on return + */ +AP_DECLARE(apr_status_t) ap_filter_setaside_brigade(ap_filter_t *f, + apr_bucket_brigade **buffered_bb, + apr_pool_t **deferred_write_pool, + apr_bucket_brigade *bb); + +/** + * Reinstate a brigade setaside earlier, and calculate the amount of data we + * should write based on the presence of flush buckets, size limits on in + * memory buckets, and the number of outstanding requests in the pipeline. If + * data was previously set aside, the counter indicating whether data exists + * in the output chain is decremented. + * @param f The current filter + * @param buffered_bb The bucket brigade to reinstate. May be NULL or empty. + * @param bb The bucket brigade to restore to. + * @param flush_upto If provided, work out the bucket we need to flush up to, + * based on the presence of a flush bucket, size limits on + * in-memory buckets, size limits on the number of requests + * outstanding in the pipeline. + * @return If non zero, there are at least THRESHOLD_MIN_WRITE bytes to write. + */ +AP_DECLARE(int) ap_filter_reinstate_brigade(ap_filter_t *f, + apr_bucket_brigade *buffered_bb, + apr_bucket_brigade *bb, + apr_bucket **flush_upto); + +/** + * Return non zero if the current filter should yield to allow write completion + * to take place. + */ +AP_DECLARE(apr_status_t) ap_filter_should_yield(ap_filter_t *f); + +/** * Flush function for apr_brigade_* calls. This calls ap_pass_brigade * to flush the brigade if the brigade buffer overflows. * @param bb The brigade to flush Index: server/core.c =================================================================== --- server/core.c (revision 1622873) +++ server/core.c (working copy) @@ -4888,6 +4888,7 @@ c->id = id; c->bucket_alloc = alloc; + c->empty = apr_brigade_create(c->pool, c->bucket_alloc); c->clogging_input_filters = 0; Index: server/core_filters.c =================================================================== --- server/core_filters.c (revision 1622873) +++ server/core_filters.c (working copy) @@ -328,11 +328,6 @@ return APR_SUCCESS; } -static void setaside_remaining_output(ap_filter_t *f, - core_output_filter_ctx_t *ctx, - apr_bucket_brigade *bb, - conn_rec *c); - static apr_status_t send_brigade_nonblocking(apr_socket_t *s, apr_bucket_brigade *bb, apr_size_t *bytes_written, @@ -358,33 +353,24 @@ conn_rec *c); #endif -/* XXX: Should these be configurable parameters? */ -#define THRESHOLD_MIN_WRITE 4096 -#define THRESHOLD_MAX_BUFFER 65536 -#define MAX_REQUESTS_IN_PIPELINE 5 - /* Optional function coming from mod_logio, used for logging of output * traffic */ extern APR_OPTIONAL_FN_TYPE(ap_logio_add_bytes_out) *ap__logio_add_bytes_out; -apr_status_t ap_core_output_filter(ap_filter_t *f, apr_bucket_brigade *new_bb) +apr_status_t ap_core_output_filter(ap_filter_t *f, apr_bucket_brigade *bb) { conn_rec *c = f->c; core_net_rec *net = f->ctx; core_output_filter_ctx_t *ctx = net->out_ctx; - apr_bucket_brigade *bb = NULL; - apr_bucket *bucket, *next, *flush_upto = NULL; - apr_size_t bytes_in_brigade, non_file_bytes_in_brigade; - int eor_buckets_in_brigade, morphing_bucket_in_brigade; - apr_status_t rv; + apr_bucket *flush_upto = NULL; + int should_write; + apr_status_t rv, empty; int loglevel = ap_get_conn_module_loglevel(c, APLOG_MODULE_INDEX); /* Fail quickly if the connection has already been aborted. */ if (c->aborted) { - if (new_bb != NULL) { - apr_brigade_cleanup(new_bb); - } + apr_brigade_cleanup(bb); return APR_ECONNABORTED; } @@ -401,29 +387,12 @@ ctx->buffered_bb = apr_brigade_create(c->pool, c->bucket_alloc); } - if (new_bb != NULL) - bb = new_bb; - - if ((ctx->buffered_bb != NULL) && - !APR_BRIGADE_EMPTY(ctx->buffered_bb)) { - if (new_bb != NULL) { - APR_BRIGADE_PREPEND(bb, ctx->buffered_bb); - } - else { - bb = ctx->buffered_bb; - } - c->data_in_output_filters = 0; - } - else if (new_bb == NULL) { - return APR_SUCCESS; - } - /* Scan through the brigade and decide whether to attempt a write, * and how much to write, based on the following rules: * - * 1) The new_bb is null: Do a nonblocking write of as much as + * 1) The bb is empty: Do a nonblocking write of as much as * possible: do a nonblocking write of as much data as possible, - * then save the rest in ctx->buffered_bb. (If new_bb == NULL, + * then save the rest in ctx->buffered_bb. (If bb is empty, * it probably means that the MPM is doing asynchronous write * completion and has just determined that this connection * is writable.) @@ -465,7 +434,17 @@ * then save the rest in ctx->buffered_bb. */ - if (new_bb == NULL) { + empty = APR_BRIGADE_EMPTY(bb); + + if ((ctx->buffered_bb != NULL) && !APR_BRIGADE_EMPTY(ctx->buffered_bb)) { + should_write = ap_filter_reinstate_brigade(f, ctx->buffered_bb, bb, + &flush_upto); + } + else if (empty) { + return APR_SUCCESS; + } + + if (empty) { rv = send_brigade_nonblocking(net->client_socket, bb, &(ctx->bytes_written), c); if (APR_STATUS_IS_EAGAIN(rv)) { @@ -477,76 +456,12 @@ "core_output_filter: writing data to the network"); c->aborted = 1; } - setaside_remaining_output(f, ctx, bb, c); + remove_empty_buckets(bb); + ap_filter_setaside_brigade(f, &(ctx->buffered_bb), + &(ctx->deferred_write_pool), bb); return rv; } - bytes_in_brigade = 0; - non_file_bytes_in_brigade = 0; - eor_buckets_in_brigade = 0; - morphing_bucket_in_brigade = 0; - - for (bucket = APR_BRIGADE_FIRST(bb); bucket != APR_BRIGADE_SENTINEL(bb); - bucket = next) { - next = APR_BUCKET_NEXT(bucket); - - if (!APR_BUCKET_IS_METADATA(bucket)) { - if (bucket->length == (apr_size_t)-1) { - /* - * A setaside of morphing buckets would read everything into - * memory. Instead, we will flush everything up to and - * including this bucket. - */ - morphing_bucket_in_brigade = 1; - } - else { - bytes_in_brigade += bucket->length; - if (!APR_BUCKET_IS_FILE(bucket)) - non_file_bytes_in_brigade += bucket->length; - } - } - else if (AP_BUCKET_IS_EOR(bucket)) { - eor_buckets_in_brigade++; - } - - if (APR_BUCKET_IS_FLUSH(bucket) - || non_file_bytes_in_brigade >= THRESHOLD_MAX_BUFFER - || morphing_bucket_in_brigade - || eor_buckets_in_brigade > MAX_REQUESTS_IN_PIPELINE) { - /* this segment of the brigade MUST be sent before returning. */ - - if (loglevel >= APLOG_TRACE6) { - char *reason = APR_BUCKET_IS_FLUSH(bucket) ? - "FLUSH bucket" : - (non_file_bytes_in_brigade >= THRESHOLD_MAX_BUFFER) ? - "THRESHOLD_MAX_BUFFER" : - morphing_bucket_in_brigade ? "morphing bucket" : - "MAX_REQUESTS_IN_PIPELINE"; - ap_log_cerror(APLOG_MARK, APLOG_TRACE6, 0, c, - "will flush because of %s", reason); - ap_log_cerror(APLOG_MARK, APLOG_TRACE8, 0, c, - "seen in brigade%s: bytes: %" APR_SIZE_T_FMT - ", non-file bytes: %" APR_SIZE_T_FMT ", eor " - "buckets: %d, morphing buckets: %d", - flush_upto == NULL ? " so far" - : " since last flush point", - bytes_in_brigade, - non_file_bytes_in_brigade, - eor_buckets_in_brigade, - morphing_bucket_in_brigade); - } - /* - * Defer the actual blocking write to avoid doing many writes. - */ - flush_upto = next; - - bytes_in_brigade = 0; - non_file_bytes_in_brigade = 0; - eor_buckets_in_brigade = 0; - morphing_bucket_in_brigade = 0; - } - } - if (flush_upto != NULL) { ctx->tmp_flush_bb = apr_brigade_split_ex(bb, flush_upto, ctx->tmp_flush_bb); @@ -571,16 +486,7 @@ APR_BRIGADE_CONCAT(bb, ctx->tmp_flush_bb); } - if (loglevel >= APLOG_TRACE8) { - ap_log_cerror(APLOG_MARK, APLOG_TRACE8, 0, c, - "brigade contains: bytes: %" APR_SIZE_T_FMT - ", non-file bytes: %" APR_SIZE_T_FMT - ", eor buckets: %d, morphing buckets: %d", - bytes_in_brigade, non_file_bytes_in_brigade, - eor_buckets_in_brigade, morphing_bucket_in_brigade); - } - - if (bytes_in_brigade >= THRESHOLD_MIN_WRITE) { + if (should_write) { rv = send_brigade_nonblocking(net->client_socket, bb, &(ctx->bytes_written), c); if ((rv != APR_SUCCESS) && (!APR_STATUS_IS_EAGAIN(rv))) { @@ -598,43 +504,12 @@ } } - setaside_remaining_output(f, ctx, bb, c); + remove_empty_buckets(bb); + ap_filter_setaside_brigade(f, &(ctx->buffered_bb), + &(ctx->deferred_write_pool), bb); return APR_SUCCESS; } -/* - * This function assumes that either ctx->buffered_bb == NULL, or - * ctx->buffered_bb is empty, or ctx->buffered_bb == bb - */ -static void setaside_remaining_output(ap_filter_t *f, - core_output_filter_ctx_t *ctx, - apr_bucket_brigade *bb, - conn_rec *c) -{ - if (bb == NULL) { - return; - } - remove_empty_buckets(bb); - if (!APR_BRIGADE_EMPTY(bb)) { - c->data_in_output_filters = 1; - if (bb != ctx->buffered_bb) { - if (!ctx->deferred_write_pool) { - apr_pool_create(&ctx->deferred_write_pool, c->pool); - apr_pool_tag(ctx->deferred_write_pool, "deferred_write"); - } - ap_save_brigade(f, &(ctx->buffered_bb), &bb, - ctx->deferred_write_pool); - } - } - else if (ctx->deferred_write_pool) { - /* - * There are no more requests in the pipeline. We can just clear the - * pool. - */ - apr_pool_clear(ctx->deferred_write_pool); - } -} - #ifndef APR_MAX_IOVEC_SIZE #define MAX_IOVEC_TO_WRITE 16 #else Index: server/mpm/event/event.c =================================================================== --- server/mpm/event/event.c (revision 1622873) +++ server/mpm/event/event.c (working copy) @@ -1090,10 +1090,10 @@ ap_filter_t *output_filter = c->output_filters; apr_status_t rv; ap_update_child_status_from_conn(sbh, SERVER_BUSY_WRITE, c); - while (output_filter->next != NULL) { - output_filter = output_filter->next; - } - rv = output_filter->frec->filter_func.out_func(output_filter, NULL); + + rv = ap_pass_brigade(output_filter, c->empty); + apr_brigade_cleanup(c->empty); + if (rv != APR_SUCCESS) { ap_log_cerror(APLOG_MARK, APLOG_DEBUG, rv, c, APLOGNO(00470) "network write failure in core output filter"); Index: server/util_filter.c =================================================================== --- server/util_filter.c (revision 1622873) +++ server/util_filter.c (working copy) @@ -24,6 +24,7 @@ #include "http_config.h" #include "http_core.h" #include "http_log.h" +#include "http_request.h" #include "util_filter.h" /* NOTE: Apache's current design doesn't allow a pool to be passed thru, @@ -32,6 +33,11 @@ #define FILTER_POOL apr_hook_global_pool #include "ap_hooks.h" /* for apr_hook_global_pool */ +/* XXX: Should these be configurable parameters? */ +#define THRESHOLD_MIN_WRITE 4096 +#define THRESHOLD_MAX_BUFFER 65536 +#define MAX_REQUESTS_IN_PIPELINE 5 + /* ** This macro returns true/false if a given filter should be inserted BEFORE ** another filter. This will happen when one of: 1) there isn't another @@ -566,6 +572,15 @@ { if (next) { apr_bucket *e; + unsigned int activity; + apr_status_t status; + + /* + * Mark that we have passed this way. + */ + next->c->activity++; + activity = next->c->activity; + if ((e = APR_BRIGADE_LAST(bb)) && APR_BUCKET_IS_EOS(e) && next->r) { /* This is only safe because HTTP_HEADER filter is always in * the filter stack. This ensures that there is ALWAYS a @@ -587,7 +602,24 @@ } } } - return next->frec->filter_func.out_func(next, bb); + status = next->frec->filter_func.out_func(next, bb); + + /* No problems found, and did the brigade not get passed on by a + * filter to the next filter in the chain? Compensate by passing + * the empty brigade to the next filter, so every filter gets a + * turn to write. + */ + while (next->c->data_in_output_filters && APR_SUCCESS == status) { + next = next->next; + if (next && next->c->activity == activity) { + status = next->frec->filter_func.out_func(next, next->c->empty); + } + else { + break; + } + } + + return status; } return AP_NOBODY_WROTE; } @@ -673,6 +705,166 @@ return srv; } +AP_DECLARE(apr_status_t) ap_filter_setaside_brigade(ap_filter_t *f, + apr_bucket_brigade **buffered_bb, + apr_pool_t **deferred_write_pool, + apr_bucket_brigade *bb) +{ + if (bb == NULL) { + return APR_SUCCESS; + } + if (!APR_BRIGADE_EMPTY(bb)) { + if (APR_BRIGADE_EMPTY((*buffered_bb))) { + f->c->data_in_output_filters++; + } + if (bb != *buffered_bb) { + if (!(*deferred_write_pool)) { + apr_pool_create(deferred_write_pool, f->c->pool); + apr_pool_tag(*deferred_write_pool, "deferred_write"); + } + return ap_save_brigade(f, buffered_bb, &bb, + *deferred_write_pool); + } + } + else if (*deferred_write_pool) { + /* + * There are no more requests in the pipeline. We can just clear the + * pool. + */ + apr_pool_clear(*deferred_write_pool); + } + return APR_SUCCESS; +} + +AP_DECLARE(int) ap_filter_reinstate_brigade(ap_filter_t *f, + apr_bucket_brigade *buffered_bb, + apr_bucket_brigade *bb, + apr_bucket **flush_upto) +{ + apr_bucket *bucket, *next; + apr_size_t bytes_in_brigade, non_file_bytes_in_brigade; + int eor_buckets_in_brigade, morphing_bucket_in_brigade; + int loglevel = ap_get_conn_module_loglevel(f->c, APLOG_MODULE_INDEX); + + if ((buffered_bb != NULL) && !APR_BRIGADE_EMPTY(buffered_bb)) { + APR_BRIGADE_PREPEND(bb, buffered_bb); + f->c->data_in_output_filters--; + } + + /* + * Determine if and up to which bucket we need to do a blocking write: + * + * a) The brigade contains a flush bucket: Do a blocking write + * of everything up that point. + * + * b) The request is in CONN_STATE_HANDLER state, and the brigade + * contains at least THRESHOLD_MAX_BUFFER bytes in non-file + * buckets: Do blocking writes until the amount of data in the + * buffer is less than THRESHOLD_MAX_BUFFER. (The point of this + * rule is to provide flow control, in case a handler is + * streaming out lots of data faster than the data can be + * sent to the client.) + * + * c) The request is in CONN_STATE_HANDLER state, and the brigade + * contains at least MAX_REQUESTS_IN_PIPELINE EOR buckets: + * Do blocking writes until less than MAX_REQUESTS_IN_PIPELINE EOR + * buckets are left. (The point of this rule is to prevent too many + * FDs being kept open by pipelined requests, possibly allowing a + * DoS). + * + * d) The brigade contains a morphing bucket: If there was no other + * reason to do a blocking write yet, try reading the bucket. If its + * contents fit into memory before THRESHOLD_MAX_BUFFER is reached, + * everything is fine. Otherwise we need to do a blocking write the + * up to and including the morphing bucket, because ap_save_brigade() + * would read the whole bucket into memory later on. + */ + + bytes_in_brigade = 0; + non_file_bytes_in_brigade = 0; + eor_buckets_in_brigade = 0; + morphing_bucket_in_brigade = 0; + + for (bucket = APR_BRIGADE_FIRST(bb); bucket != APR_BRIGADE_SENTINEL(bb); + bucket = next) { + next = APR_BUCKET_NEXT(bucket); + + if (!APR_BUCKET_IS_METADATA(bucket)) { + if (bucket->length == (apr_size_t)-1) { + /* + * A setaside of morphing buckets would read everything into + * memory. Instead, we will flush everything up to and + * including this bucket. + */ + morphing_bucket_in_brigade = 1; + } + else { + bytes_in_brigade += bucket->length; + if (!APR_BUCKET_IS_FILE(bucket)) + non_file_bytes_in_brigade += bucket->length; + } + } + else if (AP_BUCKET_IS_EOR(bucket)) { + eor_buckets_in_brigade++; + } + + if (APR_BUCKET_IS_FLUSH(bucket) + || non_file_bytes_in_brigade >= THRESHOLD_MAX_BUFFER + || morphing_bucket_in_brigade + || eor_buckets_in_brigade > MAX_REQUESTS_IN_PIPELINE) { + /* this segment of the brigade MUST be sent before returning. */ + + if (loglevel >= APLOG_TRACE6) { + char *reason = APR_BUCKET_IS_FLUSH(bucket) ? + "FLUSH bucket" : + (non_file_bytes_in_brigade >= THRESHOLD_MAX_BUFFER) ? + "THRESHOLD_MAX_BUFFER" : + morphing_bucket_in_brigade ? "morphing bucket" : + "MAX_REQUESTS_IN_PIPELINE"; + ap_log_cerror(APLOG_MARK, APLOG_TRACE6, 0, f->c, + "will flush because of %s", reason); + ap_log_cerror(APLOG_MARK, APLOG_TRACE8, 0, f->c, + "seen in brigade%s: bytes: %" APR_SIZE_T_FMT + ", non-file bytes: %" APR_SIZE_T_FMT ", eor " + "buckets: %d, morphing buckets: %d", + flush_upto == NULL ? " so far" + : " since last flush point", + bytes_in_brigade, + non_file_bytes_in_brigade, + eor_buckets_in_brigade, + morphing_bucket_in_brigade); + } + /* + * Defer the actual blocking write to avoid doing many writes. + */ + if (flush_upto) { + *flush_upto = next; + } + + bytes_in_brigade = 0; + non_file_bytes_in_brigade = 0; + eor_buckets_in_brigade = 0; + morphing_bucket_in_brigade = 0; + } + } + + if (loglevel >= APLOG_TRACE8) { + ap_log_cerror(APLOG_MARK, APLOG_TRACE8, 0, f->c, + "brigade contains: bytes: %" APR_SIZE_T_FMT + ", non-file bytes: %" APR_SIZE_T_FMT + ", eor buckets: %d, morphing buckets: %d", + bytes_in_brigade, non_file_bytes_in_brigade, + eor_buckets_in_brigade, morphing_bucket_in_brigade); + } + + return bytes_in_brigade >= THRESHOLD_MIN_WRITE; +} + +AP_DECLARE(apr_status_t) ap_filter_should_yield(ap_filter_t *f) +{ + return f->c->data_in_output_filters; +} + AP_DECLARE_NONSTD(apr_status_t) ap_filter_flush(apr_bucket_brigade *bb, void *ctx) { Index: modules/ssl/ssl_engine_io.c =================================================================== --- modules/ssl/ssl_engine_io.c (revision 1622873) +++ modules/ssl/ssl_engine_io.c (working copy) @@ -116,6 +116,8 @@ conn_rec *c; apr_bucket_brigade *bb; /* Brigade used as a buffer. */ apr_status_t rc; + apr_bucket_brigade *buffered_bb; + apr_pool_t *deferred_write_pool; } bio_filter_out_ctx_t; static bio_filter_out_ctx_t *bio_filter_out_ctx_new(ssl_filter_ctx_t *filter_ctx, @@ -126,6 +128,8 @@ outctx->filter_ctx = filter_ctx; outctx->c = c; outctx->bb = apr_brigade_create(c->pool, c->bucket_alloc); + outctx->buffered_bb = apr_brigade_create(c->pool, c->bucket_alloc); + outctx->deferred_write_pool = NULL; return outctx; } @@ -1663,20 +1667,24 @@ bio_filter_in_ctx_t *inctx; bio_filter_out_ctx_t *outctx; apr_read_type_e rblock = APR_NONBLOCK_READ; + apr_bucket *flush_upto = NULL; if (f->c->aborted) { apr_brigade_cleanup(bb); return APR_ECONNABORTED; } + inctx = (bio_filter_in_ctx_t *)filter_ctx->pbioRead->ptr; + outctx = (bio_filter_out_ctx_t *)filter_ctx->pbioWrite->ptr; + + /* Reinstate any buffered content */ + ap_filter_reinstate_brigade(f, outctx->buffered_bb, bb, &flush_upto); + if (!filter_ctx->pssl) { /* ssl_filter_io_shutdown was called */ return ap_pass_brigade(f->next, bb); } - inctx = (bio_filter_in_ctx_t *)filter_ctx->pbioRead->ptr; - outctx = (bio_filter_out_ctx_t *)filter_ctx->pbioWrite->ptr; - /* When we are the writer, we must initialize the inctx * mode so that we block for any required ssl input, because * output filtering is always nonblocking. @@ -1691,6 +1699,16 @@ while (!APR_BRIGADE_EMPTY(bb)) { apr_bucket *bucket = APR_BRIGADE_FIRST(bb); + /* if the core has set aside data, back off and try later */ + if (!flush_upto) { + if (ap_filter_should_yield(f)) { + break; + } + } + else if (flush_upto == bucket) { + flush_upto = NULL; + } + /* If it is a flush or EOS, we need to pass this down. * These types do not require translation by OpenSSL. */ @@ -1758,8 +1776,14 @@ break; } } + } + if (APR_STATUS_IS_EOF(status) || (status == APR_SUCCESS)) { + return ap_filter_setaside_brigade(f, &(outctx->buffered_bb), + &(outctx->deferred_write_pool), bb); + } + return status; }