From 3161ed034bbea40a705303811d7213aff9be17d2 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 28 Apr 2024 01:56:20 +0200 Subject: Merging upstream version 2.4.59. Signed-off-by: Daniel Baumann --- modules/http2/h2_mplx.c | 113 +++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 93 insertions(+), 20 deletions(-) (limited to 'modules/http2/h2_mplx.c') diff --git a/modules/http2/h2_mplx.c b/modules/http2/h2_mplx.c index 99c47ea..2aeea42 100644 --- a/modules/http2/h2_mplx.c +++ b/modules/http2/h2_mplx.c @@ -146,6 +146,7 @@ static void m_stream_cleanup(h2_mplx *m, h2_stream *stream) if (c2_ctx->beam_in) { h2_beam_on_send(c2_ctx->beam_in, NULL, NULL); h2_beam_on_received(c2_ctx->beam_in, NULL, NULL); + h2_beam_on_eagain(c2_ctx->beam_in, NULL, NULL); h2_beam_on_consumed(c2_ctx->beam_in, NULL, NULL); } } @@ -333,7 +334,6 @@ h2_mplx *h2_mplx_c1_create(int child_num, apr_uint32_t id, h2_stream *stream0, apr_pollset_add(m->pollset, &conn_ctx->pfd); } - m->scratch_r = apr_pcalloc(m->pool, sizeof(*m->scratch_r)); m->max_spare_transits = 3; m->c2_transits = apr_array_make(m->pool, (int)m->max_spare_transits, sizeof(h2_c2_transit*)); @@ -394,6 +394,31 @@ apr_status_t h2_mplx_c1_streams_do(h2_mplx *m, h2_mplx_stream_cb *cb, void *ctx) return APR_SUCCESS; } +typedef struct { + int stream_count; + int stream_want_send; +} stream_iter_aws_t; + +static int m_stream_want_send_data(void *ctx, void *stream) +{ + stream_iter_aws_t *x = ctx; + ++x->stream_count; + if (h2_stream_wants_send_data(stream)) + ++x->stream_want_send; + return 1; +} + +int h2_mplx_c1_all_streams_want_send_data(h2_mplx *m) +{ + stream_iter_aws_t x; + x.stream_count = 0; + x.stream_want_send = 0; + H2_MPLX_ENTER(m); + h2_ihash_iter(m->streams, m_stream_want_send_data, &x); + H2_MPLX_LEAVE(m); + return x.stream_count && (x.stream_count == x.stream_want_send); +} + static int m_report_stream_iter(void *ctx, void *val) { h2_mplx *m = ctx; h2_stream *stream = val; @@ -441,6 +466,8 @@ static int m_stream_cancel_iter(void *ctx, void *val) { return 0; } +static void c1_purge_streams(h2_mplx *m); + void h2_mplx_c1_destroy(h2_mplx *m) { apr_status_t status; @@ -509,7 +536,9 @@ void h2_mplx_c1_destroy(h2_mplx *m) h2_ihash_count(m->shold)); h2_ihash_iter(m->shold, m_unexpected_stream_iter, m); } - + + c1_purge_streams(m); + m->c1->aborted = old_aborted; H2_MPLX_LEAVE(m); @@ -542,16 +571,6 @@ const h2_stream *h2_mplx_c2_stream_get(h2_mplx *m, int stream_id) } -static void c1_update_scoreboard(h2_mplx *m, h2_stream *stream) -{ - if (stream->c2) { - m->scratch_r->connection = stream->c2; - m->scratch_r->bytes_sent = stream->out_frame_octets; - ap_increment_counts(m->c1->sbh, m->scratch_r); - m->scratch_r->connection = NULL; - } -} - static void c1_purge_streams(h2_mplx *m) { h2_stream *stream; @@ -561,8 +580,6 @@ static void c1_purge_streams(h2_mplx *m) stream = APR_ARRAY_IDX(m->spurge, i, h2_stream*); ap_assert(stream->state == H2_SS_CLEANUP); - c1_update_scoreboard(m, stream); - if (stream->input) { h2_beam_destroy(stream->input, m->c1); stream->input = NULL; @@ -585,6 +602,15 @@ static void c1_purge_streams(h2_mplx *m) apr_array_clear(m->spurge); } +void h2_mplx_c1_going_keepalive(h2_mplx *m) +{ + H2_MPLX_ENTER_ALWAYS(m); + if (m->spurge->nelts) { + c1_purge_streams(m); + } + H2_MPLX_LEAVE(m); +} + apr_status_t h2_mplx_c1_poll(h2_mplx *m, apr_interval_time_t timeout, stream_ev_callback *on_stream_input, stream_ev_callback *on_stream_output, @@ -653,8 +679,12 @@ static apr_status_t c1_process_stream(h2_mplx *m, if (APLOGctrace1(m->c1)) { const h2_request *r = stream->request; ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c1, - H2_STRM_MSG(stream, "process %s %s://%s%s"), - r->method, r->scheme, r->authority, r->path); + H2_STRM_MSG(stream, "process %s%s%s %s%s%s%s"), + r->protocol? r->protocol : "", + r->protocol? " " : "", + r->method, r->scheme? r->scheme : "", + r->scheme? "://" : "", + r->authority, r->path? r->path: ""); } stream->scheduled = 1; @@ -765,6 +795,19 @@ static void c2_beam_input_read_notify(void *ctx, h2_bucket_beam *beam) } } +static void c2_beam_input_read_eagain(void *ctx, h2_bucket_beam *beam) +{ + conn_rec *c = ctx; + h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(c); + /* installed in the input bucket beams when we use pipes. + * Drain the pipe just before the beam returns APR_EAGAIN. + * A clean state for allowing polling on the pipe to rest + * when the beam is empty */ + if (conn_ctx && conn_ctx->pipe_in[H2_PIPE_OUT]) { + h2_util_drain_pipe(conn_ctx->pipe_in[H2_PIPE_OUT]); + } +} + static void c2_beam_output_write_notify(void *ctx, h2_bucket_beam *beam) { conn_rec *c = ctx; @@ -809,6 +852,9 @@ static apr_status_t c2_setup_io(h2_mplx *m, conn_rec *c2, h2_stream *stream, h2_ c2->pool, c2->pool); if (APR_SUCCESS != rv) goto cleanup; #endif + h2_beam_on_eagain(stream->input, c2_beam_input_read_eagain, c2); + if (!h2_beam_empty(stream->input)) + c2_beam_input_write_notify(c2, stream->input); } cleanup: @@ -915,6 +961,15 @@ static void s_c2_done(h2_mplx *m, conn_rec *c2, h2_conn_ctx_t *conn_ctx) "h2_c2(%s-%d): processing finished without final response", conn_ctx->id, conn_ctx->stream_id); c2->aborted = 1; + if (conn_ctx->beam_out) + h2_beam_abort(conn_ctx->beam_out, c2); + } + else if (!conn_ctx->beam_out || !h2_beam_is_complete(conn_ctx->beam_out)) { + ap_log_cerror(APLOG_MARK, APLOG_TRACE1, conn_ctx->last_err, c2, + "h2_c2(%s-%d): processing finished with incomplete output", + conn_ctx->id, conn_ctx->stream_id); + c2->aborted = 1; + h2_beam_abort(conn_ctx->beam_out, c2); } else if (!c2->aborted) { s_mplx_be_happy(m, c2, conn_ctx); @@ -1064,14 +1119,32 @@ static int reset_is_acceptable(h2_stream *stream) return 1; /* otherwise, be forgiving */ } -apr_status_t h2_mplx_c1_client_rst(h2_mplx *m, int stream_id) +apr_status_t h2_mplx_c1_client_rst(h2_mplx *m, int stream_id, h2_stream *stream) { - h2_stream *stream; apr_status_t status = APR_SUCCESS; + int registered; H2_MPLX_ENTER_ALWAYS(m); - stream = h2_ihash_get(m->streams, stream_id); - if (stream && !reset_is_acceptable(stream)) { + registered = (h2_ihash_get(m->streams, stream_id) != NULL); + if (!stream) { + /* a RST might arrive so late, we have already forgotten + * about it. Seems ok. */ + ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c1, + H2_MPLX_MSG(m, "RST on unknown stream %d"), stream_id); + AP_DEBUG_ASSERT(!registered); + } + else if (!registered) { + /* a RST on a stream that mplx has not been told about, but + * which the session knows. Very early and annoying. */ + ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c1, + H2_STRM_MSG(stream, "very early RST, drop")); + h2_stream_set_monitor(stream, NULL); + h2_stream_rst(stream, H2_ERR_STREAM_CLOSED); + h2_stream_dispatch(stream, H2_SEV_EOS_SENT); + m_stream_cleanup(m, stream); + m_be_annoyed(m); + } + else if (!reset_is_acceptable(stream)) { m_be_annoyed(m); } H2_MPLX_LEAVE(m); -- cgit v1.2.3