1 /* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
7 * http://www.apache.org/licenses/LICENSE-2.0
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
19 #include <apr_atomic.h>
20 #include <apr_thread_mutex.h>
21 #include <apr_thread_cond.h>
22 #include <apr_strings.h>
26 #include <http_core.h>
29 #include "h2_private.h"
30 #include "h2_config.h"
33 #include "h2_io_set.h"
34 #include "h2_response.h"
36 #include "h2_request.h"
37 #include "h2_stream.h"
38 #include "h2_stream_set.h"
40 #include "h2_task_input.h"
41 #include "h2_task_output.h"
42 #include "h2_task_queue.h"
43 #include "h2_workers.h"
46 static int is_aborted(h2_mplx *m, apr_status_t *pstatus) {
49 *pstatus = APR_ECONNABORTED;
55 static void have_out_data_for(h2_mplx *m, int stream_id);
57 static void h2_mplx_destroy(h2_mplx *m)
66 h2_io_set_destroy(m->ready_ios);
70 h2_io_set_destroy(m->stream_ios);
75 apr_thread_mutex_destroy(m->lock);
80 apr_pool_destroy(m->pool);
85 * A h2_mplx needs to be thread-safe *and* if will be called by
86 * the h2_session thread *and* the h2_worker threads. Therefore:
87 * - calls are protected by a mutex lock, m->lock
88 * - the pool needs its own allocator, since apr_allocator_t are
89 * not re-entrant. The separate allocator works without a
90 * separate lock since we already protect h2_mplx itself.
91 * Since HTTP/2 connections can be expected to live longer than
92 * their HTTP/1 cousins, the separate allocator seems to work better
93 * than protecting a shared h2_session one with an own lock.
95 h2_mplx *h2_mplx_create(conn_rec *c, apr_pool_t *parent, h2_workers *workers)
97 apr_status_t status = APR_SUCCESS;
98 h2_config *conf = h2_config_get(c);
99 apr_allocator_t *allocator = NULL;
101 AP_DEBUG_ASSERT(conf);
103 status = apr_allocator_create(&allocator);
104 if (status != APR_SUCCESS) {
108 m = apr_pcalloc(parent, sizeof(h2_mplx));
111 APR_RING_ELEM_INIT(m, link);
112 apr_atomic_set32(&m->refs, 1);
114 apr_pool_create_ex(&m->pool, parent, NULL, allocator);
118 apr_allocator_owner_set(allocator, m->pool);
120 status = apr_thread_mutex_create(&m->lock, APR_THREAD_MUTEX_DEFAULT,
122 if (status != APR_SUCCESS) {
127 m->bucket_alloc = apr_bucket_alloc_create(m->pool);
129 m->q = h2_tq_create(m->id, m->pool);
130 m->stream_ios = h2_io_set_create(m->pool);
131 m->ready_ios = h2_io_set_create(m->pool);
132 m->closed = h2_stream_set_create(m->pool);
133 m->stream_max_mem = h2_config_geti(conf, H2_CONF_STREAM_MAX_MEM);
134 m->workers = workers;
136 m->file_handles_allowed = h2_config_geti(conf, H2_CONF_SESSION_FILES);
141 static void reference(h2_mplx *m)
143 apr_atomic_inc32(&m->refs);
146 static void release(h2_mplx *m)
148 if (!apr_atomic_dec32(&m->refs)) {
150 apr_thread_cond_signal(m->join_wait);
155 void h2_mplx_reference(h2_mplx *m)
159 void h2_mplx_release(h2_mplx *m)
164 static void workers_register(h2_mplx *m) {
165 /* Initially, there was ref count increase for this as well, but
166 * this is not needed, even harmful.
167 * h2_workers is only a hub for all the h2_worker instances.
168 * At the end-of-life of this h2_mplx, we always unregister at
169 * the workers. The thing to manage are all the h2_worker instances
170 * out there. Those may hold a reference to this h2_mplx and we cannot
171 * call them to unregister.
173 * Therefore: ref counting for h2_workers in not needed, ref counting
174 * for h2_worker using this is critical.
176 h2_workers_register(m->workers, m);
179 static void workers_unregister(h2_mplx *m) {
180 h2_workers_unregister(m->workers, m);
183 apr_status_t h2_mplx_release_and_join(h2_mplx *m, apr_thread_cond_t *wait)
186 workers_unregister(m);
188 status = apr_thread_mutex_lock(m->lock);
189 if (APR_SUCCESS == status) {
193 while (apr_atomic_read32(&m->refs) > 0) {
195 ap_log_cerror(APLOG_MARK, (attempts? APLOG_INFO : APLOG_DEBUG),
197 "h2_mplx(%ld): release_join, refs=%d, waiting...",
199 apr_thread_cond_timedwait(wait, m->lock, apr_time_from_sec(10));
200 if (++attempts >= 6) {
201 ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, m->c,
203 "h2_mplx(%ld): join attempts exhausted, refs=%d",
209 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c,
210 "h2_mplx(%ld): release_join -> destroy", m->id);
213 apr_thread_mutex_unlock(m->lock);
219 void h2_mplx_abort(h2_mplx *m)
223 status = apr_thread_mutex_lock(m->lock);
224 if (APR_SUCCESS == status) {
226 h2_io_set_destroy_all(m->stream_ios);
227 apr_thread_mutex_unlock(m->lock);
229 workers_unregister(m);
233 h2_stream *h2_mplx_open_io(h2_mplx *m, int stream_id)
235 h2_stream *stream = NULL;
242 status = apr_thread_mutex_lock(m->lock);
243 if (APR_SUCCESS == status) {
244 apr_pool_t *stream_pool = m->spare_pool;
247 apr_pool_create(&stream_pool, m->pool);
250 m->spare_pool = NULL;
253 stream = h2_stream_create(stream_id, stream_pool, m);
254 stream->state = H2_STREAM_ST_OPEN;
256 io = h2_io_set_get(m->stream_ios, stream_id);
258 io = h2_io_create(stream_id, stream_pool, m->bucket_alloc);
259 h2_io_set_add(m->stream_ios, io);
261 status = io? APR_SUCCESS : APR_ENOMEM;
262 apr_thread_mutex_unlock(m->lock);
267 static void stream_destroy(h2_mplx *m, h2_stream *stream, h2_io *io)
269 apr_pool_t *pool = h2_stream_detach_pool(stream);
271 apr_pool_clear(pool);
273 apr_pool_destroy(m->spare_pool);
275 m->spare_pool = pool;
277 h2_stream_destroy(stream);
279 /* The pool is cleared/destroyed which also closes all
280 * allocated file handles. Give this count back to our
281 * file handle pool. */
282 m->file_handles_allowed += io->files_handles_owned;
283 h2_io_set_remove(m->stream_ios, io);
284 h2_io_set_remove(m->ready_ios, io);
289 apr_status_t h2_mplx_cleanup_stream(h2_mplx *m, h2_stream *stream)
293 status = apr_thread_mutex_lock(m->lock);
294 if (APR_SUCCESS == status) {
295 h2_io *io = h2_io_set_get(m->stream_ios, stream->id);
296 if (!io || io->task_done) {
297 /* No more io or task already done -> cleanup immediately */
298 stream_destroy(m, stream, io);
301 /* Add stream to closed set for cleanup when task is done */
302 h2_stream_set_add(m->closed, stream);
304 apr_thread_mutex_unlock(m->lock);
309 void h2_mplx_task_done(h2_mplx *m, int stream_id)
311 apr_status_t status = apr_thread_mutex_lock(m->lock);
312 if (APR_SUCCESS == status) {
313 h2_stream *stream = h2_stream_set_get(m->closed, stream_id);
314 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
315 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c,
316 "h2_mplx(%ld): task(%d) done", m->id, stream_id);
318 /* stream was already closed by main connection and is in
319 * zombie state. Now that the task is done with it, we
320 * can free its resources. */
321 h2_stream_set_remove(m->closed, stream);
322 stream_destroy(m, stream, io);
325 /* main connection has not finished stream. Mark task as done
326 * so that eventual cleanup can start immediately. */
329 apr_thread_mutex_unlock(m->lock);
333 apr_status_t h2_mplx_in_read(h2_mplx *m, apr_read_type_e block,
334 int stream_id, apr_bucket_brigade *bb,
335 struct apr_thread_cond_t *iowait)
340 return APR_ECONNABORTED;
342 status = apr_thread_mutex_lock(m->lock);
343 if (APR_SUCCESS == status) {
344 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
346 io->input_arrived = iowait;
347 status = h2_io_in_read(io, bb, 0);
348 while (status == APR_EAGAIN
349 && !is_aborted(m, &status)
350 && block == APR_BLOCK_READ) {
351 apr_thread_cond_wait(io->input_arrived, m->lock);
352 status = h2_io_in_read(io, bb, 0);
354 io->input_arrived = NULL;
359 apr_thread_mutex_unlock(m->lock);
364 apr_status_t h2_mplx_in_write(h2_mplx *m, int stream_id,
365 apr_bucket_brigade *bb)
370 return APR_ECONNABORTED;
372 status = apr_thread_mutex_lock(m->lock);
373 if (APR_SUCCESS == status) {
374 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
376 status = h2_io_in_write(io, bb);
377 if (io->input_arrived) {
378 apr_thread_cond_signal(io->input_arrived);
384 apr_thread_mutex_unlock(m->lock);
389 apr_status_t h2_mplx_in_close(h2_mplx *m, int stream_id)
394 return APR_ECONNABORTED;
396 status = apr_thread_mutex_lock(m->lock);
397 if (APR_SUCCESS == status) {
398 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
400 status = h2_io_in_close(io);
401 if (io->input_arrived) {
402 apr_thread_cond_signal(io->input_arrived);
406 status = APR_ECONNABORTED;
408 apr_thread_mutex_unlock(m->lock);
414 h2_mplx_consumed_cb *cb;
419 static int update_window(void *ctx, h2_io *io)
421 if (io->input_consumed) {
422 update_ctx *uctx = (update_ctx*)ctx;
423 uctx->cb(uctx->cb_ctx, io->id, io->input_consumed);
424 io->input_consumed = 0;
425 ++uctx->streams_updated;
430 apr_status_t h2_mplx_in_update_windows(h2_mplx *m,
431 h2_mplx_consumed_cb *cb, void *cb_ctx)
436 return APR_ECONNABORTED;
438 status = apr_thread_mutex_lock(m->lock);
439 if (APR_SUCCESS == status) {
444 ctx.streams_updated = 0;
447 h2_io_set_iter(m->stream_ios, update_window, &ctx);
449 if (ctx.streams_updated) {
450 status = APR_SUCCESS;
452 apr_thread_mutex_unlock(m->lock);
457 apr_status_t h2_mplx_out_readx(h2_mplx *m, int stream_id,
458 h2_io_data_cb *cb, void *ctx,
459 apr_size_t *plen, int *peos)
464 return APR_ECONNABORTED;
466 status = apr_thread_mutex_lock(m->lock);
467 if (APR_SUCCESS == status) {
468 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
470 status = h2_io_out_readx(io, cb, ctx, plen, peos);
471 if (status == APR_SUCCESS && io->output_drained) {
472 apr_thread_cond_signal(io->output_drained);
476 status = APR_ECONNABORTED;
478 apr_thread_mutex_unlock(m->lock);
483 h2_stream *h2_mplx_next_submit(h2_mplx *m, h2_stream_set *streams)
486 h2_stream *stream = NULL;
491 status = apr_thread_mutex_lock(m->lock);
492 if (APR_SUCCESS == status) {
493 h2_io *io = h2_io_set_get_highest_prio(m->ready_ios);
495 h2_response *response = io->response;
497 AP_DEBUG_ASSERT(response);
498 h2_io_set_remove(m->ready_ios, io);
500 stream = h2_stream_set_get(streams, response->stream_id);
502 h2_stream_set_response(stream, response, io->bbout);
503 if (io->output_drained) {
504 apr_thread_cond_signal(io->output_drained);
508 ap_log_cerror(APLOG_MARK, APLOG_WARNING, APR_NOTFOUND, m->c,
509 APLOGNO(02953) "h2_mplx(%ld): stream for response %d",
510 m->id, response->stream_id);
513 apr_thread_mutex_unlock(m->lock);
518 static apr_status_t out_write(h2_mplx *m, h2_io *io,
519 ap_filter_t* f, apr_bucket_brigade *bb,
520 struct apr_thread_cond_t *iowait)
522 apr_status_t status = APR_SUCCESS;
523 /* We check the memory footprint queued for this stream_id
524 * and block if it exceeds our configured limit.
525 * We will not split buckets to enforce the limit to the last
526 * byte. After all, the bucket is already in memory.
528 while (!APR_BRIGADE_EMPTY(bb)
529 && (status == APR_SUCCESS)
530 && !is_aborted(m, &status)) {
532 status = h2_io_out_write(io, bb, m->stream_max_mem,
533 &m->file_handles_allowed);
535 /* Wait for data to drain until there is room again */
536 while (!APR_BRIGADE_EMPTY(bb)
538 && status == APR_SUCCESS
539 && (m->stream_max_mem <= h2_io_out_length(io))
540 && !is_aborted(m, &status)) {
541 io->output_drained = iowait;
543 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,
544 "h2_mplx(%ld-%d): waiting for out drain",
547 apr_thread_cond_wait(io->output_drained, m->lock);
548 io->output_drained = NULL;
551 apr_brigade_cleanup(bb);
555 static apr_status_t out_open(h2_mplx *m, int stream_id, h2_response *response,
556 ap_filter_t* f, apr_bucket_brigade *bb,
557 struct apr_thread_cond_t *iowait)
559 apr_status_t status = APR_SUCCESS;
561 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
564 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, f->c,
565 "h2_mplx(%ld-%d): open response: %s",
566 m->id, stream_id, response->status);
569 io->response = h2_response_copy(io->pool, response);
570 AP_DEBUG_ASSERT(io->response);
571 h2_io_set_add(m->ready_ios, io);
573 status = out_write(m, io, f, bb, iowait);
575 have_out_data_for(m, stream_id);
578 status = APR_ECONNABORTED;
583 apr_status_t h2_mplx_out_open(h2_mplx *m, int stream_id, h2_response *response,
584 ap_filter_t* f, apr_bucket_brigade *bb,
585 struct apr_thread_cond_t *iowait)
590 return APR_ECONNABORTED;
592 status = apr_thread_mutex_lock(m->lock);
593 if (APR_SUCCESS == status) {
594 status = out_open(m, stream_id, response, f, bb, iowait);
596 return APR_ECONNABORTED;
598 apr_thread_mutex_unlock(m->lock);
604 apr_status_t h2_mplx_out_write(h2_mplx *m, int stream_id,
605 ap_filter_t* f, apr_bucket_brigade *bb,
606 struct apr_thread_cond_t *iowait)
611 return APR_ECONNABORTED;
613 status = apr_thread_mutex_lock(m->lock);
614 if (APR_SUCCESS == status) {
616 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
618 status = out_write(m, io, f, bb, iowait);
619 have_out_data_for(m, stream_id);
621 return APR_ECONNABORTED;
625 status = APR_ECONNABORTED;
630 apr_thread_mutex_unlock(m->lock);
636 apr_status_t h2_mplx_out_close(h2_mplx *m, int stream_id)
641 return APR_ECONNABORTED;
643 status = apr_thread_mutex_lock(m->lock);
644 if (APR_SUCCESS == status) {
646 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
648 if (!io->response || !io->response->ngheader) {
649 /* In case a close comes before a response was created,
650 * insert an error one so that our streams can properly
653 h2_response *r = h2_response_create(stream_id,
654 "500", NULL, m->pool);
655 status = out_open(m, stream_id, r, NULL, NULL, NULL);
657 status = h2_io_out_close(io);
658 have_out_data_for(m, stream_id);
660 /* if we were the last output, the whole session might
661 * have gone down in the meantime.
667 status = APR_ECONNABORTED;
670 apr_thread_mutex_unlock(m->lock);
675 int h2_mplx_in_has_eos_for(h2_mplx *m, int stream_id)
683 status = apr_thread_mutex_lock(m->lock);
684 if (APR_SUCCESS == status) {
685 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
687 has_eos = h2_io_in_has_eos_for(io);
689 apr_thread_mutex_unlock(m->lock);
694 int h2_mplx_out_has_data_for(h2_mplx *m, int stream_id)
702 status = apr_thread_mutex_lock(m->lock);
703 if (APR_SUCCESS == status) {
704 h2_io *io = h2_io_set_get(m->stream_ios, stream_id);
706 has_data = h2_io_out_has_data(io);
708 apr_thread_mutex_unlock(m->lock);
713 apr_status_t h2_mplx_out_trywait(h2_mplx *m, apr_interval_time_t timeout,
714 apr_thread_cond_t *iowait)
719 return APR_ECONNABORTED;
721 status = apr_thread_mutex_lock(m->lock);
722 if (APR_SUCCESS == status) {
723 m->added_output = iowait;
724 status = apr_thread_cond_timedwait(m->added_output, m->lock, timeout);
725 if (APLOGctrace2(m->c)) {
726 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c,
727 "h2_mplx(%ld): trywait on data for %f ms)",
728 m->id, timeout/1000.0);
730 m->added_output = NULL;
731 apr_thread_mutex_unlock(m->lock);
736 static void have_out_data_for(h2_mplx *m, int stream_id)
740 if (m->added_output) {
741 apr_thread_cond_signal(m->added_output);
745 apr_status_t h2_mplx_do_task(h2_mplx *m, struct h2_task *task)
750 return APR_ECONNABORTED;
752 status = apr_thread_mutex_lock(m->lock);
753 if (APR_SUCCESS == status) {
754 /* TODO: needs to sort queue by priority */
755 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c,
756 "h2_mplx: do task(%s)", task->id);
757 h2_tq_append(m->q, task);
758 apr_thread_mutex_unlock(m->lock);
764 h2_task *h2_mplx_pop_task(h2_mplx *m, int *has_more)
766 h2_task *task = NULL;
773 status = apr_thread_mutex_lock(m->lock);
774 if (APR_SUCCESS == status) {
775 task = h2_tq_pop_first(m->q);
777 h2_task_set_started(task);
779 *has_more = !h2_tq_empty(m->q);
780 apr_thread_mutex_unlock(m->lock);
785 apr_status_t h2_mplx_create_task(h2_mplx *m, struct h2_stream *stream)
790 return APR_ECONNABORTED;
792 status = apr_thread_mutex_lock(m->lock);
793 if (APR_SUCCESS == status) {
794 conn_rec *c = h2_conn_create(m->c, stream->pool);
795 stream->task = h2_task_create(m->id, stream->id,
798 apr_thread_mutex_unlock(m->lock);