mod_http2: backport of v1.5.6 plus mod_proxy_http2 dsp support (87e88e59) · Commits · CYBER - Cyber Security / TS 103 523 MSP / TLMSP / TLMSP Apache Httpd

CHANGES

+3 −0

Original line number	Diff line number	Diff line
		@@ -2,6 +2,9 @@

		Changes with Apache 2.4.21

		*) mod_http2: improved event handling for suspended streams, responses
		and window updates. [Stefan Eissing]

		*) mod_proxy_hcheck: Provide for dynamic background health
		checks on reverse proxies associated with BalancerMember
		workers. [Jim Jagielski]

modules/http2/h2_bucket_beam.c

+52 −20

Original line number	Diff line number	Diff line
		@@ -356,12 +356,25 @@ static void h2_beam_emitted(h2_bucket_beam beam, h2_beam_proxy proxy)
		}
		}

		static void report_consumption(h2_bucket_beam *beam)
		static void report_consumption(h2_bucket_beam *beam, int force)
		{
		if (beam->consumed_fn && (beam->received_bytes != beam->reported_bytes)) {
		beam->consumed_fn(beam->consumed_ctx, beam,
		beam->received_bytes - beam->reported_bytes);
		beam->reported_bytes = beam->received_bytes;
		if (force \|\| beam->received_bytes != beam->reported_consumed_bytes) {
		if (beam->consumed_fn) {
		beam->consumed_fn(beam->consumed_ctx, beam, beam->received_bytes
		- beam->reported_consumed_bytes);
		}
		beam->reported_consumed_bytes = beam->received_bytes;
		}
		}

		static void report_production(h2_bucket_beam *beam, int force)
		{
		if (force \|\| beam->sent_bytes != beam->reported_produced_bytes) {
		if (beam->produced_fn) {
		beam->produced_fn(beam->produced_ctx, beam, beam->sent_bytes
		- beam->reported_produced_bytes);
		}
		beam->reported_produced_bytes = beam->sent_bytes;
		}
		}

		@@ -393,7 +406,7 @@ static apr_status_t beam_cleanup(void *data)
		beam_close(beam);
		r_purge_reds(beam);
		h2_blist_cleanup(&beam->red);
		report_consumption(beam);
		report_consumption(beam, 0);
		h2_blist_cleanup(&beam->purge);
		h2_blist_cleanup(&beam->hold);

		@@ -500,7 +513,7 @@ void h2_beam_abort(h2_bucket_beam *beam)
		r_purge_reds(beam);
		h2_blist_cleanup(&beam->red);
		beam->aborted = 1;
		report_consumption(beam);
		report_consumption(beam, 0);
		if (beam->m_cond) {
		apr_thread_cond_broadcast(beam->m_cond);
		}
		@@ -515,7 +528,7 @@ apr_status_t h2_beam_close(h2_bucket_beam *beam)
		if (enter_yellow(beam, &bl) == APR_SUCCESS) {
		r_purge_reds(beam);
		beam_close(beam);
		report_consumption(beam);
		report_consumption(beam, 0);
		leave_yellow(beam, &bl);
		}
		return beam->aborted? APR_ECONNABORTED : APR_SUCCESS;
		@@ -530,7 +543,7 @@ apr_status_t h2_beam_shutdown(h2_bucket_beam *beam, apr_read_type_e block)
		r_purge_reds(beam);
		h2_blist_cleanup(&beam->red);
		beam_close(beam);
		report_consumption(beam);
		report_consumption(beam, 0);

		while (status == APR_SUCCESS
		&& (!H2_BPROXY_LIST_EMPTY(&beam->proxies)
		@@ -693,16 +706,18 @@ apr_status_t h2_beam_send(h2_bucket_beam *beam,
		status = APR_ECONNABORTED;
		}
		else if (red_brigade) {
		int force_report = !APR_BRIGADE_EMPTY(red_brigade);
		while (!APR_BRIGADE_EMPTY(red_brigade)
		&& status == APR_SUCCESS) {
		bred = APR_BRIGADE_FIRST(red_brigade);
		status = append_bucket(beam, bred, block, beam->red_pool, &bl);
		}
		report_production(beam, force_report);
		if (beam->m_cond) {
		apr_thread_cond_broadcast(beam->m_cond);
		}
		}
		report_consumption(beam);
		report_consumption(beam, 0);
		leave_yellow(beam, &bl);
		}
		return status;
		@@ -756,8 +771,8 @@ transfer:

		if (APR_BUCKET_IS_METADATA(bred)) {
		if (APR_BUCKET_IS_EOS(bred)) {
		beam->close_sent = 1;
		bgreen = apr_bucket_eos_create(bb->bucket_alloc);
		beam->close_sent = 1;
		}
		else if (APR_BUCKET_IS_FLUSH(bred)) {
		bgreen = apr_bucket_flush_create(bb->bucket_alloc);
		@@ -835,19 +850,24 @@ transfer:
		}
		}

		if (transferred) {
		status = APR_SUCCESS;
		}
		else if (beam->closed) {
		if (beam->closed
		&& (!beam->green \|\| APR_BRIGADE_EMPTY(beam->green))
		&& H2_BLIST_EMPTY(&beam->red)) {
		/* beam is closed and we have nothing more to receive */
		if (!beam->close_sent) {
		apr_bucket *b = apr_bucket_eos_create(bb->bucket_alloc);
		APR_BRIGADE_INSERT_TAIL(bb, b);
		beam->close_sent = 1;
		++transferred;
		status = APR_SUCCESS;
		}
		else {
		status = APR_EOF;
		}

		if (transferred) {
		status = APR_SUCCESS;
		}
		else if (beam->closed) {
		status = APR_EOF;
		}
		else if (block == APR_BLOCK_READ && bl.mutex && beam->m_cond) {
		status = wait_cond(beam, bl.mutex);
		@@ -866,7 +886,7 @@ leave:
		}

		void h2_beam_on_consumed(h2_bucket_beam *beam,
		h2_beam_consumed_callback cb, void ctx)
		h2_beam_io_callback cb, void ctx)
		{
		h2_beam_lock bl;

		@@ -877,6 +897,18 @@ void h2_beam_on_consumed(h2_bucket_beam *beam,
		}
		}

		void h2_beam_on_produced(h2_bucket_beam *beam,
		h2_beam_io_callback cb, void ctx)
		{
		h2_beam_lock bl;

		if (enter_yellow(beam, &bl) == APR_SUCCESS) {
		beam->produced_fn = cb;
		beam->produced_ctx = ctx;
		leave_yellow(beam, &bl);
		}
		}

		void h2_beam_on_file_beam(h2_bucket_beam *beam,
		h2_beam_can_beam_callback cb, void ctx)
		{

modules/http2/h2_bucket_beam.h

+27 −9

Original line number	Diff line number	Diff line
		@@ -152,7 +152,7 @@ typedef struct h2_bucket_beam h2_bucket_beam;

		typedef apr_status_t h2_beam_mutex_enter(void ctx, h2_beam_lock pbl);

		typedef void h2_beam_consumed_callback(void ctx, h2_bucket_beam beam,
		typedef void h2_beam_io_callback(void ctx, h2_bucket_beam beam,
		apr_off_t bytes);

		typedef struct h2_beam_proxy h2_beam_proxy;
		@@ -174,12 +174,14 @@ struct h2_bucket_beam {
		apr_pool_t *red_pool;

		apr_size_t max_buf_size;
		apr_size_t files_beamed; /* how many file handles have been set aside */
		apr_file_t last_beamed; / last file beamed */
		apr_interval_time_t timeout;

		apr_off_t sent_bytes; /* amount of bytes send */
		apr_off_t received_bytes; /* amount of bytes received */
		apr_off_t reported_bytes; /* amount of bytes reported as consumed */
		apr_size_t buckets_sent;

		apr_size_t buckets_sent; /* # of beam buckets sent */
		apr_size_t files_beamed; /* how many file handles have been set aside */
		apr_file_t last_beamed; / last file beamed */

		unsigned int aborted : 1;
		unsigned int closed : 1;
		@@ -188,10 +190,13 @@ struct h2_bucket_beam {
		void *m_ctx;
		h2_beam_mutex_enter *m_enter;
		struct apr_thread_cond_t *m_cond;
		apr_interval_time_t timeout;

		h2_beam_consumed_callback *consumed_fn;
		apr_off_t reported_consumed_bytes; /* amount of bytes reported as consumed */
		h2_beam_io_callback *consumed_fn;
		void *consumed_ctx;
		apr_off_t reported_produced_bytes; /* amount of bytes reported as produced */
		h2_beam_io_callback *produced_fn;
		void *produced_ctx;
		h2_beam_can_beam_callback *can_beam_fn;
		void *can_beam_ctx;
		};
		@@ -319,7 +324,20 @@ apr_size_t h2_beam_buffer_size_get(h2_bucket_beam *beam);
		* Call from the red side, callbacks invoked on red side.
		*/
		void h2_beam_on_consumed(h2_bucket_beam *beam,
		h2_beam_consumed_callback cb, void ctx);
		h2_beam_io_callback cb, void ctx);

		/**
		* Register a callback to be invoked on the red side with the
		* amount of bytes that have been consumed by the red side, since the
		* last callback invocation or reset.
		* @param beam the beam to set the callback on
		* @param cb the callback or NULL
		* @param ctx the context to use in callback invocation
		*
		* Call from the red side, callbacks invoked on red side.
		*/
		void h2_beam_on_produced(h2_bucket_beam *beam,
		h2_beam_io_callback cb, void ctx);

		void h2_beam_on_file_beam(h2_bucket_beam *beam,
		h2_beam_can_beam_callback cb, void ctx);

modules/http2/h2_mplx.c

+184 −91

Original line number	Diff line number	Diff line
		@@ -97,6 +97,7 @@ static apr_status_t enter_mutex(h2_mplx m, int pacquired)
		return APR_SUCCESS;
		}

		AP_DEBUG_ASSERT(m->lock);
		status = apr_thread_mutex_lock(m->lock);
		*pacquired = (status == APR_SUCCESS);
		if (*pacquired) {
		@@ -282,10 +283,11 @@ h2_mplx h2_mplx_create(conn_rec c, apr_pool_t *parent,
		m->stream_max_mem = h2_config_geti(conf, H2_CONF_STREAM_MAX_MEM);

		m->streams = h2_ihash_create(m->pool, offsetof(h2_stream,id));
		m->sready = h2_ihash_create(m->pool, offsetof(h2_stream,id));
		m->shold = h2_ihash_create(m->pool, offsetof(h2_stream,id));
		m->spurge = h2_ihash_create(m->pool, offsetof(h2_stream,id));
		m->q = h2_iq_create(m->pool, m->max_streams);
		m->sready = h2_ihash_create(m->pool, offsetof(h2_stream,id));
		m->sresume = h2_ihash_create(m->pool, offsetof(h2_stream,id));
		m->tasks = h2_ihash_create(m->pool, offsetof(h2_task,stream_id));

		m->stream_timeout = stream_timeout;
		@@ -323,7 +325,7 @@ apr_uint32_t h2_mplx_shutdown(h2_mplx *m)

		static void input_consumed_signal(h2_mplx m, h2_stream stream)
		{
		if (stream->input) {
		if (stream->input && stream->started) {
		h2_beam_send(stream->input, NULL, 0); /* trigger updates */
		}
		}
		@@ -331,7 +333,8 @@ static void input_consumed_signal(h2_mplx m, h2_stream stream)
		static int output_consumed_signal(h2_mplx m, h2_task task)
		{
		if (task->output.beam && task->worker_started && task->assigned) {
		h2_beam_send(task->output.beam, NULL, 0); /* trigger updates */
		/* trigger updates */
		h2_beam_send(task->output.beam, NULL, APR_NONBLOCK_READ);
		}
		return 0;
		}
		@@ -366,6 +369,7 @@ static void task_destroy(h2_mplx m, h2_task task, int called_from_master)
		if (task->output.beam) {
		m->tx_handles_reserved +=
		h2_beam_get_files_beamed(task->output.beam);
		h2_beam_on_produced(task->output.beam, NULL, NULL);
		}

		slave = task->c;
		@@ -428,6 +432,7 @@ static void stream_done(h2_mplx m, h2_stream stream, int rst_error)
		*/
		h2_iq_remove(m->q, stream->id);
		h2_ihash_remove(m->sready, stream->id);
		h2_ihash_remove(m->sresume, stream->id);
		h2_ihash_remove(m->streams, stream->id);
		if (stream->input) {
		m->tx_handles_reserved += h2_beam_get_files_beamed(stream->input);
		@@ -498,6 +503,17 @@ static int task_abort_connection(void ctx, void val)
		return 1;
		}

		static int report_stream_iter(void ctx, void val) {
		h2_mplx *m = ctx;
		h2_stream *stream = val;
		ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c,
		"h2_mplx(%ld-%d): exists, started=%d, scheduled=%d, "
		"submitted=%d, suspended=%d",
		m->id, stream->id, stream->started, stream->scheduled,
		stream->submitted, stream->suspended);
		return 1;
		}

		apr_status_t h2_mplx_release_and_join(h2_mplx m, apr_thread_cond_t wait)
		{
		apr_status_t status;
		@@ -508,6 +524,17 @@ apr_status_t h2_mplx_release_and_join(h2_mplx m, apr_thread_cond_t wait)
		if ((status = enter_mutex(m, &acquired)) == APR_SUCCESS) {
		int i, wait_secs = 5;

		if (!h2_ihash_empty(m->streams) && APLOGctrace1(m->c)) {
		ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c,
		"h2_mplx(%ld): release_join with %d streams open, "
		"%d streams resume, %d streams ready, %d tasks",
		m->id, (int)h2_ihash_count(m->streams),
		(int)h2_ihash_count(m->sresume),
		(int)h2_ihash_count(m->sready),
		(int)h2_ihash_count(m->tasks));
		h2_ihash_iter(m->streams, report_stream_iter, m);
		}

		/* disable WINDOW_UPDATE callbacks */
		h2_mplx_set_consumed_cb(m, NULL, NULL);

		@@ -581,10 +608,9 @@ apr_status_t h2_mplx_release_and_join(h2_mplx m, apr_thread_cond_t wait)
		purge_streams(m);
		}
		AP_DEBUG_ASSERT(h2_ihash_empty(m->spurge));
		AP_DEBUG_ASSERT(h2_ihash_empty(m->tasks));

		if (!h2_ihash_empty(m->tasks)) {
		ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, m->c, APLOGNO(03056)
		ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c, APLOGNO(03056)
		"h2_mplx(%ld): release_join -> destroy, "
		"%d tasks still present",
		m->id, (int)h2_ihash_count(m->tasks));
		@@ -630,74 +656,6 @@ void h2_mplx_set_consumed_cb(h2_mplx m, h2_mplx_consumed_cb cb, void *ctx)
		m->input_consumed_ctx = ctx;
		}

		static int update_window(void ctx, void val)
		{
		input_consumed_signal(ctx, val);
		return 1;
		}

		apr_status_t h2_mplx_in_update_windows(h2_mplx *m)
		{
		apr_status_t status;
		int acquired;

		AP_DEBUG_ASSERT(m);
		if (m->aborted) {
		return APR_ECONNABORTED;
		}
		if ((status = enter_mutex(m, &acquired)) == APR_SUCCESS) {
		h2_ihash_iter(m->streams, update_window, m);

		ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c,
		"h2_session(%ld): windows updated", m->id);
		status = APR_SUCCESS;
		leave_mutex(m, acquired);
		}
		return status;
		}

		static int stream_iter_first(void ctx, void val)
		{
		h2_stream **pstream = ctx;
		*pstream = val;
		return 0;
		}

		h2_stream h2_mplx_next_submit(h2_mplx m)
		{
		apr_status_t status;
		h2_stream *stream = NULL;
		int acquired;

		AP_DEBUG_ASSERT(m);
		if ((status = enter_mutex(m, &acquired)) == APR_SUCCESS) {
		h2_ihash_iter(m->sready, stream_iter_first, &stream);
		if (stream) {
		h2_task *task = h2_ihash_get(m->tasks, stream->id);
		h2_ihash_remove(m->sready, stream->id);
		if (task) {
		task->submitted = 1;
		if (task->rst_error) {
		h2_stream_rst(stream, task->rst_error);
		}
		else {
		AP_DEBUG_ASSERT(task->response);
		h2_stream_set_response(stream, task->response,
		task->output.beam);
		}
		}
		else {
		/* We have the stream ready without a task. This happens
		* when we fail streams early. A response should already
		* be present. */
		AP_DEBUG_ASSERT(stream->response \|\| stream->rst_error);
		}
		}
		leave_mutex(m, acquired);
		}
		return stream;
		}

		static apr_status_t out_open(h2_mplx m, int stream_id, h2_response response)
		{
		apr_status_t status = APR_SUCCESS;
		@@ -798,6 +756,9 @@ apr_status_t h2_mplx_out_trywait(h2_mplx *m, apr_interval_time_t timeout,
		if (m->aborted) {
		status = APR_ECONNABORTED;
		}
		else if (!h2_ihash_empty(m->sready) \|\| !h2_ihash_empty(m->sresume)) {
		status = APR_SUCCESS;
		}
		else {
		m->added_output = iowait;
		status = apr_thread_cond_timedwait(m->added_output, m->lock, timeout);
		@@ -854,15 +815,15 @@ apr_status_t h2_mplx_process(h2_mplx m, struct h2_stream stream,
		if (m->aborted) {
		status = APR_ECONNABORTED;
		}
		else if (stream->response) {
		else {
		h2_ihash_add(m->streams, stream);
		if (stream->response) {
		/* already have a respone, schedule for submit */
		h2_ihash_add(m->sready, stream);
		}
		else {
		h2_beam_create(&stream->input, stream->pool, stream->id,
		"input", 0);
		h2_ihash_add(m->streams, stream);

		if (!m->need_registration) {
		m->need_registration = h2_iq_empty(m->q);
		}
		@@ -875,6 +836,7 @@ apr_status_t h2_mplx_process(h2_mplx m, struct h2_stream stream,
		"h2_mplx(%ld-%d): process, body=%d",
		m->c->id, stream->id, stream->request->body);
		}
		}
		leave_mutex(m, acquired);
		}
		if (do_registration) {
		@@ -916,6 +878,7 @@ static h2_task pop_task(h2_mplx m)
		if (new_conn) {
		h2_slave_run_pre_connection(slave, ap_get_conn_socket(slave));
		}
		stream->started = 1;
		task->worker_started = 1;
		task->started_at = apr_time_now();
		if (sid > m->max_stream_started) {
		@@ -1045,6 +1008,11 @@ static void task_done(h2_mplx m, h2_task task, h2_req_engine *ngn)
		ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c,
		"h2_mplx(%s): task_done, stream still open",
		task->id);
		if (h2_stream_is_suspended(stream)) {
		/* more data will not arrive, resume the stream */
		h2_ihash_add(m->sresume, stream);
		have_out_data_for(m, stream->id);
		}
		}
		else {
		/* stream done, was it placed in hold? */
		@@ -1354,3 +1322,128 @@ void h2_mplx_req_engine_done(h2_req_engine ngn, conn_rec r_conn)
		}
		}

		/*******************************************************************************
		* mplx master events dispatching
		******************************************************************************/

		static int update_window(void ctx, void val)
		{
		input_consumed_signal(ctx, val);
		return 1;
		}

		apr_status_t h2_mplx_dispatch_master_events(h2_mplx *m,
		stream_ev_callback *on_resume,
		stream_ev_callback *on_response,
		void *on_ctx)
		{
		apr_status_t status;
		int acquired;
		int streams[32];
		h2_stream *stream;
		h2_task *task;
		size_t i, n;

		AP_DEBUG_ASSERT(m);
		if ((status = enter_mutex(m, &acquired)) == APR_SUCCESS) {
		ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, m->c,
		"h2_mplx(%ld): dispatch events", m->id);

		/* update input windows for streams */
		h2_ihash_iter(m->streams, update_window, m);

		if (on_response && !h2_ihash_empty(m->sready)) {
		n = h2_ihash_ishift(m->sready, streams, H2_ALEN(streams));
		for (i = 0; i < n; ++i) {
		stream = h2_ihash_get(m->streams, streams[i]);
		if (!stream) {
		continue;
		}
		ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, m->c,
		"h2_mplx(%ld-%d): on_response",
		m->id, stream->id);
		task = h2_ihash_get(m->tasks, stream->id);
		if (task) {
		task->submitted = 1;
		if (task->rst_error) {
		h2_stream_rst(stream, task->rst_error);
		}
		else {
		AP_DEBUG_ASSERT(task->response);
		h2_stream_set_response(stream, task->response, task->output.beam);
		}
		}
		else {
		/* We have the stream ready without a task. This happens
		* when we fail streams early. A response should already
		* be present. */
		AP_DEBUG_ASSERT(stream->response \|\| stream->rst_error);
		}
		status = on_response(on_ctx, stream->id);
		}
		}

		if (on_resume && !h2_ihash_empty(m->sresume)) {
		n = h2_ihash_ishift(m->sresume, streams, H2_ALEN(streams));
		for (i = 0; i < n; ++i) {
		stream = h2_ihash_get(m->streams, streams[i]);
		if (!stream) {
		continue;
		}
		ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, m->c,
		"h2_mplx(%ld-%d): on_resume",
		m->id, stream->id);
		h2_stream_set_suspended(stream, 0);
		status = on_resume(on_ctx, stream->id);
		}
		}

		leave_mutex(m, acquired);
		}
		return status;
		}

		static void output_produced(void ctx, h2_bucket_beam beam, apr_off_t bytes)
		{
		h2_mplx *m = ctx;
		apr_status_t status;
		h2_stream *stream;
		int acquired;

		AP_DEBUG_ASSERT(m);
		if ((status = enter_mutex(m, &acquired)) == APR_SUCCESS) {
		stream = h2_ihash_get(m->streams, beam->id);
		if (stream && h2_stream_is_suspended(stream)) {
		h2_ihash_add(m->sresume, stream);
		h2_beam_on_produced(beam, NULL, NULL);
		have_out_data_for(m, beam->id);
		}
		leave_mutex(m, acquired);
		}
		}

		apr_status_t h2_mplx_suspend_stream(h2_mplx *m, int stream_id)
		{
		apr_status_t status;
		h2_stream *stream;
		h2_task *task;
		int acquired;

		AP_DEBUG_ASSERT(m);
		if ((status = enter_mutex(m, &acquired)) == APR_SUCCESS) {
		stream = h2_ihash_get(m->streams, stream_id);
		if (stream) {
		h2_stream_set_suspended(stream, 1);
		task = h2_ihash_get(m->tasks, stream->id);
		if (stream->started && (!task \|\| task->worker_done)) {
		h2_ihash_add(m->sresume, stream);
		}
		else {
		/* register callback so that we can resume on new output */
		h2_beam_on_produced(task->output.beam, output_produced, m);
		}
		}
		leave_mutex(m, acquired);
		}
		return status;
		}

modules/http2/h2_mplx.h

+14 −18

Original line number	Diff line number	Diff line
		@@ -73,10 +73,12 @@ struct h2_mplx {
		unsigned int need_registration : 1;

		struct h2_ihash_t streams; / all streams currently processing */
		struct h2_ihash_t sready; / all streams ready for response */
		struct h2_ihash_t shold; / all streams done with task ongoing */
		struct h2_ihash_t spurge; / all streams done, ready for destroy */

		struct h2_iqueue q; / all stream ids that need to be started */
		struct h2_ihash_t sready; / all streams ready for response */
		struct h2_ihash_t sresume; / all streams that can be resumed */

		struct h2_ihash_t tasks; / all tasks started and not destroyed */
		struct h2_ihash_t redo_tasks; / all tasks that need to be redone */
		@@ -214,31 +216,25 @@ apr_status_t h2_mplx_reprioritize(h2_mplx m, h2_stream_pri_cmp cmp, void *ctx)
		*/
		void h2_mplx_set_consumed_cb(h2_mplx m, h2_mplx_consumed_cb cb, void *ctx);

		/*******************************************************************************
		* Input handling of streams.
		******************************************************************************/

		typedef apr_status_t stream_ev_callback(void *ctx, int stream_id);

		/**
		* Invoke the consumed callback for all streams that had bytes read since the
		* last call to this function. If no stream had input data consumed, the
		* callback is not invoked.
		* The consumed callback may also be invoked at other times whenever
		* the need arises.
		* Dispatch events for the master connection, such as
		* - resume: new output data has arrived for a suspended stream
		* - response: the response for a stream is ready
		*/
		apr_status_t h2_mplx_in_update_windows(h2_mplx *m);
		apr_status_t h2_mplx_dispatch_master_events(h2_mplx *m,
		stream_ev_callback *on_resume,
		stream_ev_callback *on_response,
		void *ctx);

		apr_status_t h2_mplx_suspend_stream(h2_mplx *m, int stream_id);

		/*******************************************************************************
		* Output handling of streams.
		******************************************************************************/

		/**
		* Get a stream whose response is ready for submit. Will set response and
		* any out data available in stream.
		* @param m the mplxer to get a response from
		* @param bb the brigade to place any existing repsonse body data into
		*/
		struct h2_stream h2_mplx_next_submit(h2_mplx m);

		/**
		* Opens the output for the given stream with the specified response.
		*/