www/apache
1
0
Fork 0
mirror of https://github.com/apache/httpd.git synced 2025-08-07 04:02:58 +03:00
Code Activity

mod_http2: proper parsing and forwarding of multiple or unnaounce 1xx responses

Browse Source 
mod_proxy_http2: improved robustness when main connection gets aborted


git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/trunk@1767553 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Stefan Eissing
2016-11-01 19:40:37 +00:00
parent 86ac177f0c
commit 29d1681025
15 changed files with 118 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
CHANGES
View File

@@ -1,6 +1,14 @@
-*- coding: utf-8 -*-
Changes with Apache 2.5.0
*) mod_http2: unannounced and multiple interim responses (status code < 200)
are parsed and forwarded to client until a final response arrives.
[Stefan Eissing]
*) mod_proxy_http2: improved robustness when main connection is closed early
by resetting all ongoing streams against the backend.
[Stefan Eissing]
*) mod_http2: allocators from slave connections are released earlier, resulting
in less overall memory use on busy, long lived connections.
[Stefan Eissing]

2
docs/log-message-tags/next-number
View File

@@ -1 +1 @@
3472
3473

4
modules/http2/h2.h
View File

@@ -122,15 +122,11 @@ struct h2_request {
const char *scheme;
const char *authority;
const char *path;
apr_table_t *headers;
apr_time_t request_time;
unsigned int chunked : 1; /* iff requst body needs to be forwarded as chunked */
unsigned int serialize : 1; /* iff this request is written in HTTP/1.1 serialization */
unsigned int expect_100 : 1; /* iff we need a 100-continue response */
unsigned int expect_failed : 1; /* iff we are unable to fullfill expects */
};
typedef struct h2_headers h2_headers;

10
modules/http2/h2_from_h1.c
View File

@@ -424,8 +424,13 @@ static apr_status_t pass_response(h2_task *task, ap_filter_t *f,
status = ap_pass_brigade(f->next, parser->tmp);
apr_brigade_cleanup(parser->tmp);
parser->state = H2_RP_DONE;
task->output.parse_response = 0;
/* reset parser for possible next response */
parser->state = H2_RP_STATUS_LINE;
apr_array_clear(parser->hlines);
if (response->status >= 200) {
task->output.sent_response = 1;
}
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, task->c,
APLOGNO(03197) "h2_task(%s): passed response %d",
task->id, response->status);
@@ -486,6 +491,7 @@ apr_status_t h2_from_h1_parse_response(h2_task *task, ap_filter_t *f,
}
else if (line[0] == '\0') {
/* end of headers, pass response onward */
return pass_response(task, f, parser);
}
else {

23
modules/http2/h2_mplx.c
View File

@@ -308,8 +308,7 @@ h2_mplx *h2_mplx_create(conn_rec *c, apr_pool_t *parent,
m->stream_timeout = stream_timeout;
m->workers = workers;
m->workers_max = workers->max_workers;
m->workers_def_limit = 4;
m->workers_limit = m->workers_def_limit;
m->workers_limit = 6; /* the original h1 max parallel connections */
m->last_limit_change = m->last_idle_block = apr_time_now();
m->limit_change_interval = apr_time_from_msec(200);
@@ -568,9 +567,8 @@ apr_status_t h2_mplx_release_and_join(h2_mplx *m, apr_thread_cond_t *wait)
h2_iq_clear(m->q);
purge_streams(m);
/* 3. mark all slave connections as aborted and wakeup all sleeping
* tasks. Mark all still active streams as 'done'. m->streams has to
* be empty afterwards with streams either in
/* 3. wakeup all sleeping tasks. Mark all still active streams as 'done'.
* m->streams has to be empty afterwards with streams either in
* a) m->shold because a task is still active
* b) m->spurge because task is done, or was not started */
h2_ihash_iter(m->tasks, task_abort_connection, m);
@@ -612,8 +610,9 @@ apr_status_t h2_mplx_release_and_join(h2_mplx *m, apr_thread_cond_t *wait)
if (!h2_ihash_empty(m->tasks)) {
/* when we are here, we lost track of the tasks still present.
* this currently happens with mod_proxy_http2 when we shut
* down a h2_req_engine with tasks assigned... */
ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, m->c, APLOGNO(03056)
* down a h2_req_engine with tasks assigned. Since no parallel
* processing is going on any more, we just clean them up. */
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c, APLOGNO(03056)
"h2_mplx(%ld): 3. release_join with %d tasks",
m->id, (int)h2_ihash_count(m->tasks));
h2_ihash_iter(m->tasks, task_print, m);
@@ -977,7 +976,8 @@ static void task_done(h2_mplx *m, h2_task *task, h2_req_engine *ngn)
}
if (task->engine) {
if (!h2_req_engine_is_shutdown(task->engine)) {
if (!m->aborted && !task->c->aborted
&& !h2_req_engine_is_shutdown(task->engine)) {
ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, m->c,
"h2_mplx(%ld): task(%s) has not-shutdown "
"engine(%s)", m->id, task->id,
@@ -1314,7 +1314,8 @@ apr_status_t h2_mplx_req_engine_pull(h2_req_engine *ngn,
return status;
}
void h2_mplx_req_engine_done(h2_req_engine *ngn, conn_rec *r_conn)
void h2_mplx_req_engine_done(h2_req_engine *ngn, conn_rec *r_conn,
apr_status_t status)
{
h2_task *task = h2_ctx_cget_task(r_conn);
@@ -1325,6 +1326,10 @@ void h2_mplx_req_engine_done(h2_req_engine *ngn, conn_rec *r_conn)
if (enter_mutex(m, &acquired) == APR_SUCCESS) {
ngn_out_update_windows(m, ngn);
h2_ngn_shed_done_task(m->ngn_shed, ngn, task);
if (status != APR_SUCCESS && h2_task_can_redo(task)
&& !h2_ihash_get(m->redo_tasks, task->stream_id)) {
h2_ihash_add(m->redo_tasks, task);
}
if (task->engine) {
/* cannot report that as done until engine returns */
}

4
modules/http2/h2_mplx.h
View File

@@ -84,7 +84,6 @@ struct h2_mplx {
int max_stream_started; /* highest stream id that started processing */
int workers_busy; /* # of workers processing on this mplx */
int workers_limit; /* current # of workers limit, dynamic */
int workers_def_limit; /* default # of workers limit */
int workers_max; /* max, hard limit # of workers in a process */
apr_time_t last_idle_block; /* last time, this mplx entered IDLE while
* streams were ready */
@@ -351,6 +350,7 @@ apr_status_t h2_mplx_req_engine_pull(struct h2_req_engine *ngn,
apr_read_type_e block,
int capacity,
request_rec **pr);
void h2_mplx_req_engine_done(struct h2_req_engine *ngn, conn_rec *r_conn);
void h2_mplx_req_engine_done(struct h2_req_engine *ngn, conn_rec *r_conn,
apr_status_t status);
#endif /* defined(__mod_h2__h2_mplx__) */

6
modules/http2/h2_ngn_shed.c
View File

@@ -335,7 +335,7 @@ void h2_ngn_shed_done_ngn(h2_ngn_shed *shed, struct h2_req_engine *ngn)
if (!shed->aborted && !H2_REQ_ENTRIES_EMPTY(&ngn->entries)) {
h2_ngn_entry *entry;
ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, shed->c,
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, shed->c,
"h2_ngn_shed(%ld): exit engine %s (%s), "
"has still requests queued, shutdown=%d,"
"assigned=%ld, live=%ld, finished=%ld",
@@ -347,7 +347,7 @@ void h2_ngn_shed_done_ngn(h2_ngn_shed *shed, struct h2_req_engine *ngn)
entry != H2_REQ_ENTRIES_SENTINEL(&ngn->entries);
entry = H2_NGN_ENTRY_NEXT(entry)) {
h2_task *task = entry->task;
ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, shed->c,
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, shed->c,
"h2_ngn_shed(%ld): engine %s has queued task %s, "
"frozen=%d, aborting",
shed->c->id, ngn->id, task->id, task->frozen);
@@ -356,7 +356,7 @@ void h2_ngn_shed_done_ngn(h2_ngn_shed *shed, struct h2_req_engine *ngn)
}
}
if (!shed->aborted && (ngn->no_assigned > 1 || ngn->no_live > 1)) {
ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, shed->c,
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, shed->c,
"h2_ngn_shed(%ld): exit engine %s (%s), "
"assigned=%ld, live=%ld, finished=%ld",
shed->c->id, ngn->id, ngn->type,

57
modules/http2/h2_proxy_session.c
View File

@@ -1137,13 +1137,13 @@ static void ev_stream_done(h2_proxy_session *session, int stream_id,
if (stream) {
int touched = (stream->data_sent ||
stream_id <= session->last_stream_id);
int complete = (stream->error_code == 0);
apr_status_t status = (stream->error_code == 0)? APR_SUCCESS : APR_EINVAL;
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03364)
"h2_proxy_sesssion(%s): stream(%d) closed "
"(touched=%d, error=%d)",
session->id, stream_id, touched, stream->error_code);
if (!complete) {
if (status != APR_SUCCESS) {
stream->r->status = 500;
}
else if (!stream->data_received) {
@@ -1164,7 +1164,7 @@ static void ev_stream_done(h2_proxy_session *session, int stream_id,
h2_proxy_ihash_remove(session->streams, stream_id);
h2_proxy_iq_remove(session->suspended, stream_id);
if (session->done) {
session->done(session, stream->r, complete, touched);
session->done(session, stream->r, status, touched);
}
}
@@ -1276,6 +1276,21 @@ static void dispatch_event(h2_proxy_session *session, h2_proxys_event_t ev,
}
}
static int send_loop(h2_proxy_session *session)
{
while (nghttp2_session_want_write(session->ngh2)) {
int rv = nghttp2_session_send(session->ngh2);
if (rv < 0 && nghttp2_is_fatal(rv)) {
ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
"h2_proxy_session(%s): write, rv=%d", session->id, rv);
dispatch_event(session, H2_PROXYS_EV_CONN_ERROR, rv, NULL);
break;
}
return 1;
}
return 0;
}
apr_status_t h2_proxy_session_process(h2_proxy_session *session)
{
apr_status_t status;
@@ -1300,16 +1315,7 @@ run_loop:
case H2_PROXYS_ST_BUSY:
case H2_PROXYS_ST_LOCAL_SHUTDOWN:
case H2_PROXYS_ST_REMOTE_SHUTDOWN:
while (nghttp2_session_want_write(session->ngh2)) {
int rv = nghttp2_session_send(session->ngh2);
if (rv < 0 && nghttp2_is_fatal(rv)) {
ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
"h2_proxy_session(%s): write, rv=%d", session->id, rv);
dispatch_event(session, H2_PROXYS_EV_CONN_ERROR, rv, NULL);
break;
}
have_written = 1;
}
have_written = send_loop(session);
if (nghttp2_session_want_read(session->ngh2)) {
status = h2_proxy_session_read(session, 0, 0);
@@ -1386,13 +1392,36 @@ typedef struct {
h2_proxy_request_done *done;
} cleanup_iter_ctx;
static int cancel_iter(void *udata, void *val)
{
cleanup_iter_ctx *ctx = udata;
h2_proxy_stream *stream = val;
nghttp2_submit_rst_stream(ctx->session->ngh2, NGHTTP2_FLAG_NONE,
stream->id, 0);
return 1;
}
void h2_proxy_session_cancel_all(h2_proxy_session *session)
{
if (!h2_proxy_ihash_empty(session->streams)) {
cleanup_iter_ctx ctx;
ctx.session = session;
ctx.done = session->done;
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03366)
"h2_proxy_session(%s): cancel %d streams",
session->id, (int)h2_proxy_ihash_count(session->streams));
h2_proxy_ihash_iter(session->streams, cancel_iter, &ctx);
session_shutdown(session, 0, NULL);
}
}
static int done_iter(void *udata, void *val)
{
cleanup_iter_ctx *ctx = udata;
h2_proxy_stream *stream = val;
int touched = (stream->data_sent ||
stream->id <= ctx->session->last_stream_id);
ctx->done(ctx->session, stream->r, 0, touched);
ctx->done(ctx->session, stream->r, APR_ECONNABORTED, touched);
return 1;
}

4
modules/http2/h2_proxy_session.h
View File

@@ -52,7 +52,7 @@ typedef enum {
typedef struct h2_proxy_session h2_proxy_session;
typedef void h2_proxy_request_done(h2_proxy_session *s, request_rec *r,
int complete, int touched);
apr_status_t status, int touched);
struct h2_proxy_session {
const char *id;
@@ -103,6 +103,8 @@ apr_status_t h2_proxy_session_submit(h2_proxy_session *s, const char *url,
*/
apr_status_t h2_proxy_session_process(h2_proxy_session *s);
void h2_proxy_session_cancel_all(h2_proxy_session *s);
void h2_proxy_session_cleanup(h2_proxy_session *s, h2_proxy_request_done *done);
void h2_proxy_session_update_window(h2_proxy_session *s,

26
modules/http2/h2_request.c
View File

@@ -187,16 +187,6 @@ apr_status_t h2_request_end_headers(h2_request *req, apr_pool_t *pool, int eos)
}
}
s = apr_table_get(req->headers, "Expect");
if (s && s[0]) {
if (ap_cstr_casecmp(s, "100-continue") == 0) {
req->expect_100 = 1;
}
else {
req->expect_failed = 1;
}
}
return APR_SUCCESS;
}
@@ -215,6 +205,7 @@ request_rec *h2_request_create_rec(const h2_request *req, conn_rec *c)
{
int access_status = HTTP_OK;
const char *rpath;
const char *s;
request_rec *r = ap_create_request(c);
@@ -250,12 +241,15 @@ request_rec *h2_request_create_rec(const h2_request *req, conn_rec *c)
/* we may have switched to another server */
r->per_dir_config = r->server->lookup_defaults;
if (req->expect_100) {
r->expecting_100 = 1;
}
else if (req->expect_failed) {
r->status = HTTP_EXPECTATION_FAILED;
ap_send_error_response(r, 0);
s = apr_table_get(r->headers_in, "Expect");
if (s && s[0]) {
if (ap_cstr_casecmp(s, "100-continue") == 0) {
r->expecting_100 = 1;
}
else {
r->status = HTTP_EXPECTATION_FAILED;
ap_send_error_response(r, 0);
}
}
/*

18
modules/http2/h2_task.c
View File

@@ -331,7 +331,7 @@ static apr_status_t h2_filter_slave_in(ap_filter_t* f,
/* Hmm, well. There is mode AP_MODE_EATCRLF, but we chose not
* to support it. Seems to work. */
ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_ENOTIMPL, f->c,
APLOGNO(02942)
APLOGNO(03472)
"h2_slave_in(%s), unsupported READ mode %d",
task->id, mode);
status = APR_ENOTIMPL;
@@ -368,7 +368,7 @@ static apr_status_t h2_filter_parse_h1(ap_filter_t* f, apr_bucket_brigade* bb)
/* There are cases where we need to parse a serialized http/1.1
* response. One example is a 100-continue answer in serialized mode
* or via a mod_proxy setup */
while (task->output.parse_response) {
while (!task->output.sent_response) {
status = h2_from_h1_parse_response(task, f, bb);
ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, f->c,
"h2_task(%s): parsed response", task->id);
@@ -563,23 +563,13 @@ apr_status_t h2_task_do(h2_task *task, apr_thread_t *thread, int worker_id)
task->input.bb = apr_brigade_create(task->pool, task->c->bucket_alloc);
if (task->request->serialize) {
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,
"h2_task(%s): serialize request %s %s, expect-100=%d",
task->id, task->request->method, task->request->path,
task->request->expect_100);
"h2_task(%s): serialize request %s %s",
task->id, task->request->method, task->request->path);
apr_brigade_printf(task->input.bb, NULL,
NULL, "%s %s HTTP/1.1\r\n",
task->request->method, task->request->path);
apr_table_do(input_ser_header, task, task->request->headers, NULL);
apr_brigade_puts(task->input.bb, NULL, NULL, "\r\n");
if (task->request->expect_100) {
/* we are unable to suppress the serialization of the
* intermediate response and need to parse it */
task->output.parse_response = 1;
}
}
if (task->request->expect_100) {
task->output.parse_response = 1;
}
ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, task->c,

1
modules/http2/h2_task.h
View File

@@ -70,7 +70,6 @@ struct h2_task {
unsigned int opened : 1;
unsigned int sent_response : 1;
unsigned int copy_files : 1;
unsigned int parse_response : 1;
struct h2_response_parser *rparser;
apr_bucket_brigade *bb;
} output;

5
modules/http2/mod_http2.c
View File

@@ -166,9 +166,10 @@ static apr_status_t http2_req_engine_pull(h2_req_engine *ngn,
return h2_mplx_req_engine_pull(ngn, block, capacity, pr);
}
static void http2_req_engine_done(h2_req_engine *ngn, conn_rec *r_conn)
static void http2_req_engine_done(h2_req_engine *ngn, conn_rec *r_conn,
apr_status_t status)
{
h2_mplx_req_engine_done(ngn, r_conn);
h2_mplx_req_engine_done(ngn, r_conn, status);
}
/* Runs once per created child process. Perform any process

3
modules/http2/mod_http2.h
View File

@@ -90,5 +90,6 @@ APR_DECLARE_OPTIONAL_FN(apr_status_t,
request_rec **pr));
APR_DECLARE_OPTIONAL_FN(void,
http2_req_engine_done, (h2_req_engine *engine,
conn_rec *rconn));
conn_rec *rconn,
apr_status_t status));
#endif

36
modules/http2/mod_proxy_http2.c
View File

@@ -46,7 +46,8 @@ static apr_status_t (*req_engine_pull)(h2_req_engine *engine,
apr_read_type_e block,
int capacity,
request_rec **pr);
static void (*req_engine_done)(h2_req_engine *engine, conn_rec *r_conn);
static void (*req_engine_done)(h2_req_engine *engine, conn_rec *r_conn,
apr_status_t status);
typedef struct h2_proxy_ctx {
conn_rec *owner;
@@ -270,12 +271,12 @@ static apr_status_t add_request(h2_proxy_session *session, request_rec *r)
}
static void request_done(h2_proxy_session *session, request_rec *r,
int complete, int touched)
apr_status_t status, int touched)
{
h2_proxy_ctx *ctx = session->user_data;
const char *task_id = apr_table_get(r->connection->notes, H2_TASK_ID_NOTE);
if (!complete) {
if (status != APR_SUCCESS) {
if (!touched) {
/* untouched request, need rescheduling */
if (req_engine_push && is_h2 && is_h2(ctx->owner)) {
@@ -292,7 +293,7 @@ static void request_done(h2_proxy_session *session, request_rec *r,
else {
const char *uri;
uri = apr_uri_unparse(r->pool, &r->parsed_uri, 0);
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, r->connection,
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, r->connection,
APLOGNO(03471) "h2_proxy_session(%s): request %s -> %s "
"not complete, was touched",
ctx->engine_id, task_id, uri);
@@ -300,23 +301,15 @@ static void request_done(h2_proxy_session *session, request_rec *r,
}
if (r == ctx->rbase) {
ctx->r_status = complete? APR_SUCCESS : HTTP_GATEWAY_TIME_OUT;
ctx->r_status = (status == APR_SUCCESS)? APR_SUCCESS : HTTP_SERVICE_UNAVAILABLE;
}
if (req_engine_done && ctx->engine) {
if (complete) {
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, r->connection,
APLOGNO(03370)
"h2_proxy_session(%s): finished request %s",
ctx->engine_id, task_id);
}
else {
ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, r->connection,
APLOGNO(03371)
"h2_proxy_session(%s): failed request %s",
ctx->engine_id, task_id);
}
req_engine_done(ctx->engine, r->connection);
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, r->connection,
APLOGNO(03370)
"h2_proxy_session(%s): finished request %s",
ctx->engine_id, task_id);
req_engine_done(ctx->engine, r->connection, status);
}
}
@@ -382,7 +375,12 @@ static apr_status_t proxy_engine_run(h2_proxy_ctx *ctx) {
ap_log_cerror(APLOG_MARK, APLOG_DEBUG, s2, ctx->owner,
APLOGNO(03374) "eng(%s): pull request",
ctx->engine_id);
status = s2;
/* give notice that we're leaving and cancel all ongoing
* streams. */
next_request(ctx, 1);
h2_proxy_session_cancel_all(ctx->session);
h2_proxy_session_process(ctx->session);
status = ctx->r_status = APR_SUCCESS;
break;
}
if (!ctx->next && h2_proxy_ihash_empty(ctx->session->streams)) {