diff options
Diffstat (limited to 'src/http')
| -rw-r--r-- | src/http/modules/ngx_http_event_proxy_handler.c | 2 | ||||
| -rw-r--r-- | src/http/ngx_http.h | 3 | ||||
| -rw-r--r-- | src/http/ngx_http_core_module.c | 6 | ||||
| -rw-r--r-- | src/http/ngx_http_event.c | 106 | ||||
| -rw-r--r-- | src/http/ngx_http_output_filter.c | 229 | ||||
| -rw-r--r-- | src/http/ngx_http_output_filter.h | 1 |
6 files changed, 294 insertions, 53 deletions
diff --git a/src/http/modules/ngx_http_event_proxy_handler.c b/src/http/modules/ngx_http_event_proxy_handler.c index 195e4b3ff..c7f115acf 100644 --- a/src/http/modules/ngx_http_event_proxy_handler.c +++ b/src/http/modules/ngx_http_event_proxy_handler.c @@ -437,7 +437,7 @@ static int ngx_http_proxy_read_response_body(ngx_event_t *ev) #elif (HAVE_KQUEUE) - if (ngx_event_type == NGX_KQUEUE_EVENT) { + if (ngx_event_type == NGX_HAVE_KQUEUE_EVENT) { /* do not allocate new block if there is EOF */ if (ev->eof && ev->available == 0) { left = 1; diff --git a/src/http/ngx_http.h b/src/http/ngx_http.h index 296657679..41bd932eb 100644 --- a/src/http/ngx_http.h +++ b/src/http/ngx_http.h @@ -163,6 +163,9 @@ struct ngx_http_request_s { unsigned unusual_uri:1; /* URI is not started with '/' - "GET http://" */ unsigned complex_uri:1; /* URI with "/." or with "//" (WIN32) */ unsigned path_not_found:1; +#ifdef NGX_EVENT + unsigned write_level_event:1; +#endif int state; char *uri_start; diff --git a/src/http/ngx_http_core_module.c b/src/http/ngx_http_core_module.c index 2f33b5726..072fe3666 100644 --- a/src/http/ngx_http_core_module.c +++ b/src/http/ngx_http_core_module.c @@ -101,10 +101,10 @@ int ngx_http_handler(ngx_http_request_t *r) r->connection->unexpected_eof = 0; - r->lingering_close = 0; - r->keepalive = 0; + r->lingering_close = 1; + r->keepalive = 1; -#if 0 +#if 1 r->filter = NGX_HTTP_FILTER_NEED_IN_MEMORY; #endif diff --git a/src/http/ngx_http_event.c b/src/http/ngx_http_event.c index ea5b3225b..0b69209ba 100644 --- a/src/http/ngx_http_event.c +++ b/src/http/ngx_http_event.c @@ -30,6 +30,7 @@ static int ngx_http_event_request_handler(ngx_http_request_t *r); static int ngx_http_writer(ngx_event_t *ev); static int ngx_http_block_read(ngx_event_t *ev); static int ngx_http_read_discarded_body(ngx_event_t *ev); +static int ngx_http_set_keepalive(ngx_http_request_t *r); static int ngx_http_keepalive_handler(ngx_event_t *ev); static int ngx_http_set_lingering_close(ngx_http_request_t *r); static int ngx_http_lingering_close_handler(ngx_event_t *ev); @@ -112,10 +113,11 @@ int ngx_http_init_connection(ngx_connection_t *c) #endif +/* THINK: should ngx_edge_add_event() be moved to accept part ? */ #if (HAVE_EDGE_EVENT) /* epoll */ if (ngx_event_flags & NGX_HAVE_EDGE_EVENT) { - if (ngx_add_event(ev, NGX_READ_EVENT, NGX_EDGE_EVENT) == NGX_ERROR) { + if (ngx_edge_add_event(ev) == NGX_ERROR) { return NGX_ERROR; } return ngx_http_init_request(ev); @@ -129,7 +131,7 @@ int ngx_http_init_connection(ngx_connection_t *c) return ngx_http_init_request(ev); } -#endif /* HAVE_AIO_EVENT */ +#endif /* select, poll, /dev/poll */ @@ -498,9 +500,9 @@ static int ngx_http_event_request_handler(ngx_http_request_t *r) #else -#if (HAVE_AIO_EVENT) /* aio, iocp */ +#if (HAVE_AIO_EVENT) || (HAVE_EDGE_EVENT) /* aio, iocp, epoll */ - if (ngx_event_flags & NGX_HAVE_AIO_EVENT) { + if (ngx_event_flags & (NGX_HAVE_AIO_EVENT|NGX_HAVE_EDGE_EVENT)) { return rc; } @@ -512,30 +514,14 @@ static int ngx_http_event_request_handler(ngx_http_request_t *r) event = NGX_CLEAR_EVENT; } else { - event = NGX_ONESHOT_EVENT; - } - -#elif (HAVE_EDGE_EVENT) /* epoll */ - - if (ngx_event_flags & NGX_HAVE_EDGE_EVENT) { - event = NGX_EDGE_EVENT; - - } else { - event = NGX_ONESHOT_EVENT; - } - -#elif (HAVE_DEVPOLL_EVENT) /* /dev/poll */ - - if (ngx_event_flags & NGX_HAVE_LEVEL_EVENT) { event = NGX_LEVEL_EVENT; - - } else { - event = NGX_ONESHOT_EVENT; + r->write_level_event = 1; } -#else /* select, poll */ +#else /* select, poll, /dev/poll */ - event = NGX_ONESHOT_EVENT; + event = NGX_LEVEL_EVENT; + r->write_level_event = 1; #endif @@ -571,13 +557,7 @@ static int ngx_http_event_request_handler(ngx_http_request_t *r) /* keepalive */ - r->connection->buffer->pos.mem = r->connection->buffer->last.mem - = r->connection->buffer->start; - rev->event_handler = ngx_http_keepalive_handler; - - ngx_http_close_request(r); - - return NGX_OK; + return ngx_http_set_keepalive(r); } @@ -617,14 +597,6 @@ static int ngx_http_writer(ngx_event_t *ev) ngx_add_timer(ev, timeout); } - /* TODO: /dev/poll, epoll, aio_write */ - - if (ev->oneshot) - if (ngx_add_event(ev, NGX_WRITE_EVENT, NGX_ONESHOT_EVENT) - == NGX_ERROR) { - return ngx_http_close_request(r); - } - return rc; } @@ -646,15 +618,11 @@ static int ngx_http_writer(ngx_event_t *ev) /* keepalive */ - c->buffer->pos.mem = c->buffer->last.mem = c->buffer->start; - c->read->event_handler = ngx_http_keepalive_handler; - - ngx_http_close_request(r); - - return NGX_OK; + return ngx_http_set_keepalive(r); } +/* TODO */ static int ngx_http_block_read(ngx_event_t *ev) { ngx_log_debug(ev->log, "http read blocked"); @@ -693,6 +661,7 @@ static int ngx_http_block_read(ngx_event_t *ev) } +/* TODO */ int ngx_http_discard_body(ngx_http_request_t *r) { ngx_event_t *ev; @@ -716,6 +685,7 @@ int ngx_http_discard_body(ngx_http_request_t *r) } +/* TODO */ static int ngx_http_read_discarded_body(ngx_event_t *ev) { size_t size; @@ -757,6 +727,34 @@ static int ngx_http_read_discarded_body(ngx_event_t *ev) } +/* TODO: if c->read->blocked */ +static int ngx_http_set_keepalive(ngx_http_request_t *r) +{ + ngx_connection_t *c; + + c = (ngx_connection_t *) r->connection; + + c->buffer->pos.mem = c->buffer->last.mem = c->buffer->start; + c->read->event_handler = ngx_http_keepalive_handler; + + if (r->write_level_event) { + if (ngx_del_event(c->write, NGX_WRITE_EVENT, 0) == NGX_ERROR) { + return NGX_ERROR; + } + } + + ngx_http_close_request(r); + +#if (HAVE_AIO_EVENT) /* aio, iocp */ + if (ngx_event_flags & NGX_HAVE_AIO_EVENT) { + return ngx_http_keepalive_handler(c->read); + } +#endif + + return NGX_OK; +} + + static int ngx_http_keepalive_handler(ngx_event_t *ev) { ssize_t n; @@ -796,8 +794,10 @@ static int ngx_http_keepalive_handler(ngx_event_t *ev) static int ngx_http_set_lingering_close(ngx_http_request_t *r) { ngx_event_t *ev; + ngx_connection_t *c; ngx_http_core_loc_conf_t *lcf; + c = r->connection; ev = r->connection->read; lcf = (ngx_http_core_loc_conf_t *) @@ -832,7 +832,21 @@ static int ngx_http_set_lingering_close(ngx_http_request_t *r) return ngx_http_close_request(r); } - return NGX_OK; +#if (HAVE_AIO_EVENT) /* aio, iocp */ + if (ngx_event_flags & NGX_HAVE_AIO_EVENT) { + return ngx_http_lingering_close_handler(ev); + } +#endif + +#if (HAVE_CLEAR_EVENT) || (HAVE_EDGE_EVENT) /* kqueue, epoll */ + if (ngx_event_flags & (NGX_HAVE_CLEAR_EVENT|NGX_HAVE_EDGE_EVENT)) { + return NGX_OK; + } +#endif + + /* select, poll, /dev/poll */ + + return ngx_del_event(c->write, NGX_WRITE_EVENT, 0); } diff --git a/src/http/ngx_http_output_filter.c b/src/http/ngx_http_output_filter.c index 93a7d61c4..2ca32f75e 100644 --- a/src/http/ngx_http_output_filter.c +++ b/src/http/ngx_http_output_filter.c @@ -57,6 +57,224 @@ ngx_module_t ngx_http_output_filter_module = { }; +#if 1 + +#define next_filter ngx_http_output_filter_module_ctx.next_output_body_filter + +#define need_to_copy(r, hunk) \ + (((r->filter & NGX_HTTP_FILTER_NEED_IN_MEMORY) \ + && (hunk->type & NGX_HUNK_FILE)) \ + || ((r->filter & NGX_HTTP_FILTER_NEED_TEMP) \ + && (hunk->type & (NGX_HUNK_MEMORY|NGX_HUNK_MMAP)))) + +int ngx_http_output_filter(ngx_http_request_t *r, ngx_hunk_t *hunk) +{ + int rc; + size_t size; + ngx_chain_t *ce, *pe; + ngx_http_output_filter_ctx_t *ctx; + ngx_http_output_filter_conf_t *conf; + + ctx = (ngx_http_output_filter_ctx_t *) + ngx_http_get_module_ctx(r->main ? r->main : r, + ngx_http_output_filter_module); + + if (ctx == NULL) { + ngx_http_create_ctx(r, ctx, ngx_http_output_filter_module, + sizeof(ngx_http_output_filter_ctx_t)); + } + + /* the incoming chain ctx->in is empty */ + if (ctx->in == NULL) { + + if (hunk == NULL) { + return next_filter(r, NULL); + } + + /* we do not need to copy the incoming hunk to our hunk */ + if (!need_to_copy(r, hunk)) { + ctx->out.hunk = hunk; + ctx->out.next = NULL; + + return next_filter(r, &ctx->out); + } + + /* we need to copy the incoming hunk to our hunk */ + + /* allocate ctx->hunk if it's needed */ + if (ctx->hunk == NULL) { + + conf = (ngx_http_output_filter_conf_t *) + ngx_http_get_module_loc_conf(r->main ? r->main : r, + ngx_http_output_filter_module); + + if (hunk->type & NGX_HUNK_LAST) { + size = hunk->last.mem - hunk->pos.mem; + if (size > conf->hunk_size) { + size = conf->hunk_size; + } + + } else { + size = conf->hunk_size; + } + + ngx_test_null(ctx->hunk, + ngx_create_temp_hunk(r->pool, size, 50, 50), + NGX_ERROR); + + ctx->hunk->type |= NGX_HUNK_RECYCLED; + } + + /* copy the incoming hunk or its part to our hunk + and pass it to the next filter */ + + do { + rc = ngx_http_output_filter_copy_hunk(ctx->hunk, hunk); + + if (rc == NGX_ERROR) { + return rc; + } + +#if (NGX_FILE_AIO_READ) + + if (rc == NGX_AGAIN) { + /* add the incoming hunk to the incoming chain */ + ngx_add_hunk_to_chain(ctx->in, hunk, r->pool, NGX_ERROR); + return rc; + } + +#endif + ctx->out.hunk = ctx->hunk; + ctx->out.next = NULL; + + rc = next_filter(r, &ctx->out); + + if (rc == NGX_ERROR) { + return rc; + } + + if (rc == NGX_AGAIN) { + /* add the incoming hunk to the incoming chain */ + ngx_add_hunk_to_chain(ctx->in, hunk, r->pool, NGX_ERROR); + return rc; + } + + /* NGX_OK */ + + /* set our hunk free */ + ctx->hunk->pos.mem = ctx->hunk->last.mem = ctx->hunk->start; + + /* repeat until we will have copied the whole incoming hunk */ + } while (hunk->pos.mem < hunk->last.mem); + + return NGX_OK; + } + + /* the incoming chain ctx->in is not empty */ + + /* add the incoming hunk to the incoming chain */ + if (hunk) { + for (ce = ctx->in; ce->next; ce = ce->next) { + /* void */ ; + } + + ngx_add_hunk_to_chain(ce->next, hunk, r->pool, NGX_ERROR); + } + + /* our hunk is still busy */ + if (ctx->hunk->pos.mem < ctx->hunk->last.mem) { + rc = next_filter(r, NULL); + + if (rc == NGX_ERROR || rc == NGX_AGAIN) { + return rc; + } + + /* NGX_OK */ + ctx->hunk->pos.mem = ctx->hunk->last.mem = ctx->hunk->start; + } + +#if (NGX_SUPPRESS_WARN) + pe = NULL; +#endif + + /* process the incoming chain ctx->in */ + do { + /* find the hunks that do not need to be copied ... */ + for (ce = ctx->in; ce; ce = ce->next) { + if (need_to_copy(r, ce->hunk)) { + break; + } + pe = ce; + } + + /* ... and pass them to the next filter */ + if (ctx->in != ce) { + + ctx->out.hunk = ctx->in->hunk; + ctx->out.next = ctx->in->next; + ctx->in = ce; + pe->next = NULL; + + rc = next_filter(r, &ctx->out); + + if (rc == NGX_ERROR || rc == NGX_AGAIN) { + return rc; + } + + /* NGX_OK */ + if (ctx->in == NULL) { + return rc; + } + } + + + /* copy the first hunk or its part from the incoming chain ctx->in + to our hunk and pass it to the next filter */ + do { + rc = ngx_http_output_filter_copy_hunk(ctx->hunk, ctx->in->hunk); + + if (rc == NGX_ERROR) { + return rc; + } + +#if (NGX_FILE_AIO_READ) + + if (rc == NGX_AGAIN) { + return rc; + } + +#endif + ctx->out.hunk = ctx->hunk; + ctx->out.next = NULL; + + rc = next_filter(r, &ctx->out); + + if (rc == NGX_ERROR || rc == NGX_AGAIN) { + return rc; + } + + /* NGX_OK */ + + /* set our hunk free */ + ctx->hunk->pos.mem = ctx->hunk->last.mem = ctx->hunk->start; + + /* repeat until we will have copied the whole first hunk from + the incoming chain ctx->in */ + } while (ctx->in->hunk->pos.mem < ctx->in->hunk->last.mem); + + /* delete the completed hunk from the incoming chain */ + ctx->in = ctx->in->next; + + /* repeat until we will have processed the whole incoming chain ctx->in */ + } while (ctx->in); + + return NGX_OK; +} + + +#else + + int ngx_http_output_filter(ngx_http_request_t *r, ngx_hunk_t *hunk) { int rc, once; @@ -106,6 +324,7 @@ int ngx_http_output_filter(ngx_http_request_t *r, ngx_hunk_t *hunk) } else { ctx->out.hunk = ctx->hunk; + /* XXX: should we check hunk type before copy it ? */ rc = ngx_http_output_filter_copy_hunk(ctx->hunk, ctx->in->hunk); #if (NGX_FILE_AIO_READ) if (rc == NGX_AGAIN) { @@ -116,6 +335,8 @@ int ngx_http_output_filter(ngx_http_request_t *r, ngx_hunk_t *hunk) return rc; } + /* NGX_OK */ + /* whole hunk is copied so we send to next filter chain part up to next hunk that need to be copied */ if (ctx->in->hunk->pos.mem == ctx->in->hunk->last.mem) { @@ -242,8 +463,10 @@ int ngx_http_output_filter(ngx_http_request_t *r, ngx_hunk_t *hunk) } } - if (rc == NGX_OK && ctx->hunk) + /* set free our hunk if operation has completed */ + if (rc == NGX_OK && ctx->hunk) { ctx->hunk->pos.mem = ctx->hunk->last.mem = ctx->hunk->start; + } } #if (NGX_SUPPRESS_WARN) @@ -259,7 +482,7 @@ int ngx_http_output_filter(ngx_http_request_t *r, ngx_hunk_t *hunk) } if (rc == NGX_OK) { - if (ctx->hunk) { + if (ctx->hunk) { /* XXX: double upper code ? */ ctx->hunk->pos.mem = ctx->hunk->last.mem = ctx->hunk->start; } #if (NGX_LEVEL_EVENT) @@ -270,6 +493,8 @@ int ngx_http_output_filter(ngx_http_request_t *r, ngx_hunk_t *hunk) return rc; } +#endif + static int ngx_http_output_filter_copy_hunk(ngx_hunk_t *dst, ngx_hunk_t *src) { diff --git a/src/http/ngx_http_output_filter.h b/src/http/ngx_http_output_filter.h index 175550eaa..cb7367b93 100644 --- a/src/http/ngx_http_output_filter.h +++ b/src/http/ngx_http_output_filter.h @@ -21,7 +21,6 @@ typedef struct { ngx_hunk_t *hunk; ngx_chain_t *in; ngx_chain_t out; - unsigned last; } ngx_http_output_filter_ctx_t; |