/* Licensed to the Apache Software Foundation (ASF) under one or more

* contributor license agreements. See the NOTICE file distributed with

* this work for additional information regarding copyright ownership.

* The ASF licenses this file to You under the Apache License, Version 2.0

* (the "License"); you may not use this file except in compliance with

* the License. You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

#include <assert.h>

#include <stddef.h>

#include <stdlib.h>

#include <apr_atomic.h>

#include <apr_thread_mutex.h>

#include <apr_thread_cond.h>

#include <apr_strings.h>

#include <apr_time.h>

#include <httpd.h>

#include <http_core.h>

#include <http_connection.h>

#include <http_log.h>

#include <http_protocol.h>

#include <mpm_common.h>

#include "mod_http2.h"

#include "h2.h"

#include "h2_private.h"

#include "h2_bucket_beam.h"

#include "h2_config.h"

#include "h2_c1.h"

#include "h2_conn_ctx.h"

#include "h2_protocol.h"

#include "h2_mplx.h"

#include "h2_request.h"

#include "h2_stream.h"

#include "h2_session.h"

#include "h2_c2.h"

#include "h2_workers.h"

#include "h2_util.h"

/* utility for iterating over ihash stream sets */

typedef struct {

h2_mplx *m;

h2_stream *stream;

apr_time_t now;

apr_size_t count;

} stream_iter_ctx;

static conn_rec *c2_prod_next(void *baton, int *phas_more);

static void c2_prod_done(void *baton, conn_rec *c2);

static void workers_shutdown(void *baton, int graceful);

static void s_mplx_be_happy(h2_mplx *m, conn_rec *c, h2_conn_ctx_t *conn_ctx);

static void m_be_annoyed(h2_mplx *m);

static apr_status_t mplx_pollset_create(h2_mplx *m);

static apr_status_t mplx_pollset_poll(h2_mplx *m, apr_interval_time_t timeout,

stream_ev_callback *on_stream_input,

stream_ev_callback *on_stream_output,

void *on_ctx);

static apr_pool_t *pchild;

/* APR callback invoked if allocation fails. */

static int abort_on_oom(int retcode)

{

ap_abort_on_oom();

return retcode; /* unreachable, hopefully. */

}

apr_status_t h2_mplx_c1_child_init(apr_pool_t *pool, server_rec *s)

{

pchild = pool;

return APR_SUCCESS;

}

#define H2_MPLX_ENTER(m) \

do { apr_status_t rv_lock; if ((rv_lock = apr_thread_mutex_lock(m->lock)) != APR_SUCCESS) {\

return rv_lock;\

} } while(0)

#define H2_MPLX_LEAVE(m) \

apr_thread_mutex_unlock(m->lock)

#define H2_MPLX_ENTER_ALWAYS(m) \

apr_thread_mutex_lock(m->lock)

#define H2_MPLX_ENTER_MAYBE(m, dolock) \

if (dolock) apr_thread_mutex_lock(m->lock)

#define H2_MPLX_LEAVE_MAYBE(m, dolock) \

if (dolock) apr_thread_mutex_unlock(m->lock)

static void c1_input_consumed(void *ctx, h2_bucket_beam *beam, apr_off_t length)

{

h2_stream_in_consumed(ctx, length);

}

static int stream_is_running(h2_stream *stream)

{

h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(stream->c2);

return conn_ctx && apr_atomic_read32(&conn_ctx->started) != 0

&& apr_atomic_read32(&conn_ctx->done) == 0;

}

int h2_mplx_c1_stream_is_running(h2_mplx *m, h2_stream *stream)

{

int rv;

H2_MPLX_ENTER(m);

rv = stream_is_running(stream);

H2_MPLX_LEAVE(m);

return rv;

}

static void c1c2_stream_joined(h2_mplx *m, h2_stream *stream)

{

ap_assert(!stream_is_running(stream));

h2_ihash_remove(m->shold, stream->id);

APR_ARRAY_PUSH(m->spurge, h2_stream *) = stream;

}

static void m_stream_cleanup(h2_mplx *m, h2_stream *stream)

{

h2_conn_ctx_t *c2_ctx = h2_conn_ctx_get(stream->c2);

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,

H2_STRM_MSG(stream, "cleanup, unsubscribing from beam events"));

if (c2_ctx) {

if (c2_ctx->beam_out) {

h2_beam_on_was_empty(c2_ctx->beam_out, NULL, NULL);

}

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);

}

}

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,

H2_STRM_MSG(stream, "cleanup, removing from registries"));

ap_assert(stream->state == H2_SS_CLEANUP);

h2_stream_cleanup(stream);

h2_ihash_remove(m->streams, stream->id);

h2_iq_remove(m->q, stream->id);

if (c2_ctx) {

if (!stream_is_running(stream)) {

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,

H2_STRM_MSG(stream, "cleanup, c2 is done, move to spurge"));

/* processing has finished */

APR_ARRAY_PUSH(m->spurge, h2_stream *) = stream;

}

else {

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,

H2_STRM_MSG(stream, "cleanup, c2 is running, abort"));

/* c2 is still running */

h2_c2_abort(stream->c2, m->c1);

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,

H2_STRM_MSG(stream, "cleanup, c2 is done, move to shold"));

h2_ihash_add(m->shold, stream);

}

}

else {

/* never started */

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,

H2_STRM_MSG(stream, "cleanup, never started, move to spurge"));

APR_ARRAY_PUSH(m->spurge, h2_stream *) = stream;

}

}

static h2_c2_transit *c2_transit_create(h2_mplx *m)

{

apr_allocator_t *allocator;

apr_pool_t *ptrans;

h2_c2_transit *transit;

apr_status_t rv;

/* We create a pool with its own allocator to be used for

* processing a request. This is the only way to have the processing

* independent of its parent pool in the sense that it can work in

* another thread.

*/

rv = apr_allocator_create(&allocator);

if (rv == APR_SUCCESS) {

apr_allocator_max_free_set(allocator, ap_max_mem_free);

rv = apr_pool_create_ex(&ptrans, m->pool, NULL, allocator);

}

if (rv != APR_SUCCESS) {

/* maybe the log goes through, maybe not. */

ap_log_cerror(APLOG_MARK, APLOG_ERR, rv, m->c1,

APLOGNO(10004) "h2_mplx: create transit pool");

ap_abort_on_oom();

return NULL; /* should never be reached. */

}

apr_allocator_owner_set(allocator, ptrans);

apr_pool_abort_set(abort_on_oom, ptrans);

apr_pool_tag(ptrans, "h2_c2_transit");

transit = apr_pcalloc(ptrans, sizeof(*transit));

transit->pool = ptrans;

transit->bucket_alloc = apr_bucket_alloc_create(ptrans);

return transit;

}

static void c2_transit_destroy(h2_c2_transit *transit)

{

apr_pool_destroy(transit->pool);

}

static h2_c2_transit *c2_transit_get(h2_mplx *m)

{

h2_c2_transit **ptransit = apr_array_pop(m->c2_transits);

if (ptransit) {

return *ptransit;

}

return c2_transit_create(m);

}

static void c2_transit_recycle(h2_mplx *m, h2_c2_transit *transit)

{

if (m->c2_transits->nelts >= APR_INT32_MAX ||

(apr_uint32_t)m->c2_transits->nelts >= m->max_spare_transits) {

c2_transit_destroy(transit);

}

else {

APR_ARRAY_PUSH(m->c2_transits, h2_c2_transit*) = transit;

}

}

/**

* A h2_mplx needs to be thread-safe *and* if will be called by

* the h2_session thread *and* the h2_worker threads. Therefore:

* - calls are protected by a mutex lock, m->lock

* - the pool needs its own allocator, since apr_allocator_t are

* not re-entrant. The separate allocator works without a

* separate lock since we already protect h2_mplx itself.

* Since HTTP/2 connections can be expected to live longer than

* their HTTP/1 cousins, the separate allocator seems to work better

* than protecting a shared h2_session one with an own lock.

*/

h2_mplx *h2_mplx_c1_create(int child_num, apr_uint32_t id, h2_stream *stream0,

server_rec *s, apr_pool_t *parent,

h2_workers *workers)

{

h2_conn_ctx_t *conn_ctx;

apr_status_t status = APR_SUCCESS;

apr_allocator_t *allocator;

apr_thread_mutex_t *mutex = NULL;

h2_mplx *m = NULL;

m = apr_pcalloc(parent, sizeof(h2_mplx));

m->stream0 = stream0;

m->c1 = stream0->c2;

m->s = s;

m->child_num = child_num;

m->id = id;

/* We create a pool with its own allocator to be used for

* processing secondary connections. This is the only way to have the

* processing independent of its parent pool in the sense that it

* can work in another thread. Also, the new allocator needs its own

* mutex to synchronize sub-pools.

*/

status = apr_allocator_create(&allocator);

if (status != APR_SUCCESS) {

allocator = NULL;

goto failure;

}

apr_allocator_max_free_set(allocator, ap_max_mem_free);

apr_pool_create_ex(&m->pool, parent, NULL, allocator);

if (!m->pool) goto failure;

apr_pool_tag(m->pool, "h2_mplx");

apr_allocator_owner_set(allocator, m->pool);

status = apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_DEFAULT,

m->pool);

if (APR_SUCCESS != status) goto failure;

apr_allocator_mutex_set(allocator, mutex);

status = apr_thread_mutex_create(&m->lock, APR_THREAD_MUTEX_DEFAULT,

m->pool);

if (APR_SUCCESS != status) goto failure;

m->max_streams = h2_config_sgeti(s, H2_CONF_MAX_STREAMS);

m->stream_max_mem = h2_config_sgeti(s, H2_CONF_STREAM_MAX_MEM);

m->streams = h2_ihash_create(m->pool, offsetof(h2_stream,id));

m->shold = h2_ihash_create(m->pool, offsetof(h2_stream,id));

m->spurge = apr_array_make(m->pool, 10, sizeof(h2_stream*));

m->q = h2_iq_create(m->pool, m->max_streams);

m->workers = workers;

m->processing_max = H2MIN(h2_workers_get_max_workers(workers), m->max_streams);

m->processing_limit = 6; /* the original h1 max parallel connections */

m->last_mood_change = apr_time_now();

m->mood_update_interval = apr_time_from_msec(100);

status = mplx_pollset_create(m);

if (APR_SUCCESS != status) {

ap_log_cerror(APLOG_MARK, APLOG_ERR, status, m->c1, APLOGNO(10308)

"nghttp2: could not create pollset");

goto failure;

}

m->streams_ev_in = apr_array_make(m->pool, 10, sizeof(h2_stream*));

m->streams_ev_out = apr_array_make(m->pool, 10, sizeof(h2_stream*));

m->streams_input_read = h2_iq_create(m->pool, 10);

m->streams_output_written = h2_iq_create(m->pool, 10);

status = apr_thread_mutex_create(&m->poll_lock, APR_THREAD_MUTEX_DEFAULT,

m->pool);

if (APR_SUCCESS != status) goto failure;

conn_ctx = h2_conn_ctx_get(m->c1);

if (conn_ctx->pfd.reqevents) {

apr_pollset_add(m->pollset, &conn_ctx->pfd);

}

m->max_spare_transits = 3;

m->c2_transits = apr_array_make(m->pool, (int)m->max_spare_transits,

sizeof(h2_c2_transit*));

m->producer = h2_workers_register(workers, m->pool,

apr_psprintf(m->pool, "h2-%u",

(unsigned int)m->id),

c2_prod_next, c2_prod_done,

workers_shutdown, m);

return m;

failure:

if (m->pool) {

apr_pool_destroy(m->pool);

}

else if (allocator) {

apr_allocator_destroy(allocator);

}

return NULL;

}

int h2_mplx_c1_shutdown(h2_mplx *m)

{

int max_stream_id_started = 0;

H2_MPLX_ENTER(m);

max_stream_id_started = m->max_stream_id_started;

/* Clear schedule queue, disabling existing streams from starting */

h2_iq_clear(m->q);

H2_MPLX_LEAVE(m);

return max_stream_id_started;

}

typedef struct {

h2_mplx_stream_cb *cb;

void *ctx;

} stream_iter_ctx_t;

static int m_stream_iter_wrap(void *ctx, void *stream)

{

stream_iter_ctx_t *x = ctx;

return x->cb(stream, x->ctx);

}

apr_status_t h2_mplx_c1_streams_do(h2_mplx *m, h2_mplx_stream_cb *cb, void *ctx)

{

stream_iter_ctx_t x;

x.cb = cb;

x.ctx = ctx;

H2_MPLX_ENTER(m);

h2_ihash_iter(m->streams, m_stream_iter_wrap, &x);

H2_MPLX_LEAVE(m);

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;

h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(stream->c2);

ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, m->c1,

H2_STRM_MSG(stream, "started=%d, scheduled=%d, ready=%d, out_buffer=%ld"),

!!stream->c2, stream->scheduled, h2_stream_is_ready(stream),

(long)(stream->output? h2_beam_get_buffered(stream->output) : -1));

if (conn_ctx) {

ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c1, /* NO APLOGNO */

H2_STRM_MSG(stream, "->03198: %s %s %s"

"[started=%u/done=%u]"),

conn_ctx->request->method, conn_ctx->request->authority,

conn_ctx->request->path,

apr_atomic_read32(&conn_ctx->started),

apr_atomic_read32(&conn_ctx->done));

}

else {

ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c1, /* NO APLOGNO */

H2_STRM_MSG(stream, "->03198: not started"));

}

return 1;

}

static int m_unexpected_stream_iter(void *ctx, void *val) {

h2_mplx *m = ctx;

h2_stream *stream = val;

ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, m->c1, /* NO APLOGNO */

H2_STRM_MSG(stream, "unexpected, started=%d, scheduled=%d, ready=%d"),

!!stream->c2, stream->scheduled, h2_stream_is_ready(stream));

return 1;

}

static int m_stream_cancel_iter(void *ctx, void *val) {

h2_mplx *m = ctx;

h2_stream *stream = val;

/* take over event monitoring */

h2_stream_set_monitor(stream, NULL);

/* Reset, should transit to CLOSED state */

h2_stream_rst(stream, H2_ERR_NO_ERROR);

/* All connection data has been sent, simulate cleanup */

h2_stream_dispatch(stream, H2_SEV_EOS_SENT);

m_stream_cleanup(m, stream);

return 0;

}

static void c1_purge_streams(h2_mplx *m);

void h2_mplx_c1_destroy(h2_mplx *m)

{

apr_status_t status;

unsigned int i, wait_secs = 60;

int old_aborted;

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,

H2_MPLX_MSG(m, "start release"));

/* How to shut down a h2 connection:

* 0. abort and tell the workers that no more work will come from us */

m->shutdown = m->aborted = 1;

H2_MPLX_ENTER_ALWAYS(m);

/* While really terminating any c2 connections, treat the master

* connection as aborted. It's not as if we could send any more data

* at this point. */

old_aborted = m->c1->aborted;

m->c1->aborted = 1;

/* How to shut down a h2 connection:

* 1. cancel all streams still active */

ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c1,

H2_MPLX_MSG(m, "release, %u/%u/%d streams (total/hold/purge), %d streams"),

h2_ihash_count(m->streams),

h2_ihash_count(m->shold),

m->spurge->nelts, m->processing_count);

while (!h2_ihash_iter(m->streams, m_stream_cancel_iter, m)) {

/* until empty */

}

/* 2. no more streams should be scheduled or in the active set */

ap_assert(h2_ihash_empty(m->streams));

ap_assert(h2_iq_empty(m->q));

/* 3. while workers are busy on this connection, meaning they

* are processing streams from this connection, wait on them finishing

* in order to wake us and let us check again.

* Eventually, this has to succeed. */

if (!m->join_wait) {

apr_thread_cond_create(&m->join_wait, m->pool);

}

for (i = 0; h2_ihash_count(m->shold) > 0; ++i) {

status = apr_thread_cond_timedwait(m->join_wait, m->lock, apr_time_from_sec(wait_secs));

if (APR_STATUS_IS_TIMEUP(status)) {

/* This can happen if we have very long running requests

* that do not time out on IO. */

ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c1, APLOGNO(03198)

H2_MPLX_MSG(m, "waited %u sec for %u streams"),

i*wait_secs, h2_ihash_count(m->shold));

h2_ihash_iter(m->shold, m_report_stream_iter, m);

}

}

H2_MPLX_LEAVE(m);

h2_workers_join(m->workers, m->producer);

H2_MPLX_ENTER_ALWAYS(m);

/* 4. With all workers done, all streams should be in spurge */

ap_assert(m->processing_count == 0);

if (!h2_ihash_empty(m->shold)) {

ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, m->c1, APLOGNO(03516)

H2_MPLX_MSG(m, "unexpected %u streams in hold"),

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);

ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c1,

H2_MPLX_MSG(m, "released"));

}

apr_status_t h2_mplx_c1_stream_cleanup(h2_mplx *m, h2_stream *stream,

unsigned int *pstream_count)

{

H2_MPLX_ENTER(m);

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,

H2_STRM_MSG(stream, "cleanup"));

m_stream_cleanup(m, stream);

*pstream_count = h2_ihash_count(m->streams);

H2_MPLX_LEAVE(m);

return APR_SUCCESS;

}

const h2_stream *h2_mplx_c2_stream_get(h2_mplx *m, int stream_id)

{

h2_stream *s = NULL;

H2_MPLX_ENTER_ALWAYS(m);

s = h2_ihash_get(m->streams, stream_id);

H2_MPLX_LEAVE(m);

return s;

}

static void c1_purge_streams(h2_mplx *m)

{

h2_stream *stream;

int i;

for (i = 0; i < m->spurge->nelts; ++i) {

stream = APR_ARRAY_IDX(m->spurge, i, h2_stream*);

ap_assert(stream->state == H2_SS_CLEANUP);

if (stream->input) {

h2_beam_destroy(stream->input, m->c1);

stream->input = NULL;

}

if (stream->c2) {

conn_rec *c2 = stream->c2;

h2_conn_ctx_t *c2_ctx = h2_conn_ctx_get(c2);

h2_c2_transit *transit;

stream->c2 = NULL;

ap_assert(c2_ctx);

transit = c2_ctx->transit;

h2_c2_destroy(c2); /* c2_ctx is gone as well */

if (transit) {

c2_transit_recycle(m, transit);

}

}

h2_stream_destroy(stream);

}

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,

void *on_ctx)

{

apr_status_t rv;

H2_MPLX_ENTER(m);

if (m->aborted) {

rv = APR_ECONNABORTED;

goto cleanup;

}

/* Purge (destroy) streams outside of pollset processing.

* Streams that are registered in the pollset, will be removed

* when they are destroyed, but the pollset works on copies

* of these registrations. So, if we destroy streams while

* processing pollset events, we might access freed memory.

*/

if (m->spurge->nelts) {

c1_purge_streams(m);

}

rv = mplx_pollset_poll(m, timeout, on_stream_input, on_stream_output, on_ctx);

cleanup:

H2_MPLX_LEAVE(m);

return rv;

}

apr_status_t h2_mplx_c1_reprioritize(h2_mplx *m, h2_stream_pri_cmp_fn *cmp,

h2_session *session)

{

apr_status_t status;

H2_MPLX_ENTER(m);

if (m->aborted) {

status = APR_ECONNABORTED;

}

else {

h2_iq_sort(m->q, cmp, session);

ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c1,

H2_MPLX_MSG(m, "reprioritize streams"));

status = APR_SUCCESS;

}

H2_MPLX_LEAVE(m);

return status;

}

static apr_status_t c1_process_stream(h2_mplx *m,

h2_stream *stream,

h2_stream_pri_cmp_fn *cmp,

h2_session *session)

{

apr_status_t rv = APR_SUCCESS;

if (m->aborted) {

rv = APR_ECONNABORTED;

goto cleanup;

}

if (!stream->request) {

rv = APR_EINVAL;

goto cleanup;

}

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%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;

h2_ihash_add(m->streams, stream);

if (h2_stream_is_ready(stream)) {

/* already have a response */

ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c1,

H2_STRM_MSG(stream, "process, ready already"));

}

else {

/* last chance to set anything up before stream is processed

* by worker threads. */

rv = h2_stream_prepare_processing(stream);

if (APR_SUCCESS != rv) goto cleanup;

h2_iq_add(m->q, stream->id, cmp, session);

ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c1,

H2_STRM_MSG(stream, "process, added to q"));

}

cleanup:

return rv;

}

void h2_mplx_c1_process(h2_mplx *m,

unsigned int *pstream_count)

{

apr_status_t rv;

int sid;

H2_MPLX_ENTER_ALWAYS(m);

while ((sid = h2_iq_shift(ready_to_process)) > 0) {

h2_stream *stream = get_stream(session, sid);

if (stream) {

ap_assert(!stream->scheduled);

rv = c1_process_stream(session->mplx, stream, stream_pri_cmp, session);

if (APR_SUCCESS != rv) {

h2_stream_rst(stream, H2_ERR_INTERNAL_ERROR);

}

}

else {

ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c1,

H2_MPLX_MSG(m, "stream %d not found to process"), sid);

}

}

if ((m->processing_count < m->processing_limit) && !h2_iq_empty(m->q)) {

H2_MPLX_LEAVE(m);

rv = h2_workers_activate(m->workers, m->producer);

H2_MPLX_ENTER_ALWAYS(m);

if (rv != APR_SUCCESS) {

ap_log_cerror(APLOG_MARK, APLOG_ERR, rv, m->c1, APLOGNO(10021)

H2_MPLX_MSG(m, "activate at workers"));

}

}

*pstream_count = h2_ihash_count(m->streams);

#if APR_POOL_DEBUG

do {

apr_size_t mem_g, mem_m, mem_s, mem_c1;

mem_g = pchild? apr_pool_num_bytes(pchild, 1) : 0;

mem_m = apr_pool_num_bytes(m->pool, 1);

mem_s = apr_pool_num_bytes(session->pool, 1);

mem_c1 = apr_pool_num_bytes(m->c1->pool, 1);

ap_log_cerror(APLOG_MARK, APLOG_INFO, 0, m->c1,

H2_MPLX_MSG(m, "child mem=%ld, mplx mem=%ld, session mem=%ld, c1=%ld"),

(long)mem_g, (long)mem_m, (long)mem_s, (long)mem_c1);

} while (0);

#endif

H2_MPLX_LEAVE(m);

}

static void c2_beam_input_write_notify(void *ctx, h2_bucket_beam *beam)

{

conn_rec *c = ctx;

h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(c);

(void)beam;

if (conn_ctx && conn_ctx->stream_id && conn_ctx->pipe_in[H2_PIPE_IN]) {

apr_file_putc(1, conn_ctx->pipe_in[H2_PIPE_IN]);

}

}

static void add_stream_poll_event(h2_mplx *m, int stream_id, h2_iqueue *q)

{

apr_thread_mutex_lock(m->poll_lock);

if (h2_iq_append(q, stream_id) && h2_iq_count(q) == 1) {

/* newly added first */

apr_pollset_wakeup(m->pollset);

}

apr_thread_mutex_unlock(m->poll_lock);

}

static void c2_beam_input_read_notify(void *ctx, h2_bucket_beam *beam)

{

conn_rec *c = ctx;

h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(c);

if (conn_ctx && conn_ctx->stream_id) {

add_stream_poll_event(conn_ctx->mplx, conn_ctx->stream_id,

conn_ctx->mplx->streams_input_read);

}

}

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;

h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(c);

if (conn_ctx && conn_ctx->stream_id) {

add_stream_poll_event(conn_ctx->mplx, conn_ctx->stream_id,

conn_ctx->mplx->streams_output_written);

}

}

static apr_status_t c2_setup_io(h2_mplx *m, conn_rec *c2, h2_stream *stream, h2_c2_transit *transit)

{

h2_conn_ctx_t *conn_ctx;

apr_status_t rv = APR_SUCCESS;

const char *action = "init";

rv = h2_conn_ctx_init_for_c2(&conn_ctx, c2, m, stream, transit);

if (APR_SUCCESS != rv) goto cleanup;

if (!conn_ctx->beam_out) {

action = "create output beam";

rv = h2_beam_create(&conn_ctx->beam_out, c2, conn_ctx->req_pool,

stream->id, "output", 0, c2->base_server->timeout);

if (APR_SUCCESS != rv) goto cleanup;

h2_beam_buffer_size_set(conn_ctx->beam_out, m->stream_max_mem);

h2_beam_on_was_empty(conn_ctx->beam_out, c2_beam_output_write_notify, c2);

}

memset(&conn_ctx->pipe_in, 0, sizeof(conn_ctx->pipe_in));

if (stream->input) {

conn_ctx->beam_in = stream->input;

h2_beam_on_send(stream->input, c2_beam_input_write_notify, c2);

h2_beam_on_received(stream->input, c2_beam_input_read_notify, c2);

h2_beam_on_consumed(stream->input, c1_input_consumed, stream);

#if H2_USE_PIPES

action = "create input write pipe";

rv = apr_file_pipe_create_pools(&conn_ctx->pipe_in[H2_PIPE_OUT],

&conn_ctx->pipe_in[H2_PIPE_IN],

APR_READ_BLOCK,

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:

stream->output = (APR_SUCCESS == rv)? conn_ctx->beam_out : NULL;

if (APR_SUCCESS != rv) {

ap_log_cerror(APLOG_MARK, APLOG_ERR, rv, c2,

H2_STRM_LOG(APLOGNO(10309), stream,

"error %s"), action);

}

return rv;

}

static conn_rec *s_next_c2(h2_mplx *m)

{

h2_stream *stream = NULL;

apr_status_t rv = APR_SUCCESS;

apr_uint32_t sid;

conn_rec *c2 = NULL;

h2_c2_transit *transit = NULL;

while (!m->aborted && !stream && (m->processing_count < m->processing_limit)

&& (sid = h2_iq_shift(m->q)) > 0) {

stream = h2_ihash_get(m->streams, sid);

}

if (!stream) {

if (m->processing_count >= m->processing_limit && !h2_iq_empty(m->q)) {

ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c1,

H2_MPLX_MSG(m, "delaying request processing. "

"Current limit is %d and %d workers are in use."),

m->processing_limit, m->processing_count);

}

goto cleanup;

}

if (sid > m->max_stream_id_started) {

m->max_stream_id_started = sid;

}

transit = c2_transit_get(m);

#if AP_HAS_RESPONSE_BUCKETS

c2 = ap_create_secondary_connection(transit->pool, m->c1, transit->bucket_alloc);

#else

c2 = h2_c2_create(m->c1, transit->pool, transit->bucket_alloc);

#endif

if (!c2) goto cleanup;

ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, m->c1,

H2_STRM_MSG(stream, "created new c2"));

rv = c2_setup_io(m, c2, stream, transit);

if (APR_SUCCESS != rv) goto cleanup;

stream->c2 = c2;

++m->processing_count;

cleanup:

if (APR_SUCCESS != rv && c2) {

h2_c2_destroy(c2);

c2 = NULL;

}

if (transit && !c2) {

c2_transit_recycle(m, transit);

}

return c2;

}

static conn_rec *c2_prod_next(void *baton, int *phas_more)

{

h2_mplx *m = baton;

conn_rec *c = NULL;

H2_MPLX_ENTER_ALWAYS(m);

if (!m->aborted) {

c = s_next_c2(m);

*phas_more = (c != NULL && !h2_iq_empty(m->q));

}

H2_MPLX_LEAVE(m);

return c;

}

static void s_c2_done(h2_mplx *m, conn_rec *c2, h2_conn_ctx_t *conn_ctx)

{

h2_stream *stream;

ap_assert(conn_ctx);

ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c2,

"h2_mplx(%s-%d): c2 done", conn_ctx->id, conn_ctx->stream_id);

AP_DEBUG_ASSERT(apr_atomic_read32(&conn_ctx->done) == 0);

apr_atomic_set32(&conn_ctx->done, 1);

conn_ctx->done_at = apr_time_now();

++c2->keepalives;

/* From here on, the final handling of c2 is done by c1 processing.

* Which means we can give it c1's scoreboard handle for updates. */

c2->sbh = m->c1->sbh;

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c2,

"h2_mplx(%s-%d): request done, %f ms elapsed",

conn_ctx->id, conn_ctx->stream_id,

(conn_ctx->done_at - conn_ctx->started_at) / 1000.0);

if (!conn_ctx->has_final_response) {

ap_log_cerror(APLOG_MARK, APLOG_TRACE1, conn_ctx->last_err, c2,

"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);

}

stream = h2_ihash_get(m->streams, conn_ctx->stream_id);

if (stream) {

/* stream not done yet. trigger a potential polling on the output

* since nothing more will happening here. */

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c2,

H2_STRM_MSG(stream, "c2_done, stream open"));

c2_beam_output_write_notify(c2, NULL);

}

else if ((stream = h2_ihash_get(m->shold, conn_ctx->stream_id)) != NULL) {

/* stream is done, was just waiting for this. */

ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c2,

H2_STRM_MSG(stream, "c2_done, in hold"));

c1c2_stream_joined(m, stream);

}

else {

int i;

for (i = 0; i < m->spurge->nelts; ++i) {

if (stream == APR_ARRAY_IDX(m->spurge, i, h2_stream*)) {

ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, c2,

H2_STRM_LOG(APLOGNO(03517), stream, "already in spurge"));

ap_assert("stream should not be in spurge" == NULL);