/* 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 <scoreboard.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 int add_for_purge(h2_mplx *m, h2_stream *stream)
{
    int i;
    for (i = 0; i < m->spurge->nelts; ++i) {
        h2_stream *s = APR_ARRAY_IDX(m->spurge, i, h2_stream*);
        if (s == stream)  /* already scheduled for purging */
            return FALSE;
    }
    APR_ARRAY_PUSH(m->spurge, h2_stream *) = stream;
    return TRUE;
}

static void c1c2_stream_joined(h2_mplx *m, h2_stream *stream)
{
    ap_assert(!stream_is_running(stream));
    
    h2_ihash_remove(m->shold, stream->id);
    add_for_purge(m, 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 */
            add_for_purge(m, 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 */
        int added = add_for_purge(m, stream);
        if (added)
            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,
                          H2_STRM_MSG(stream, "cleanup, never started, move to spurge"));
    }
}

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;
    
    H2_MPLX_ENTER(m);

    x.cb = cb;
    x.ctx = ctx;
    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;
    int stream_send_win_exhausted;
} 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_stream_send_win_exh(void *ctx, void *s)
{
    h2_stream *stream = s;
    int win;
    stream_iter_aws_t *x = ctx;
    ++x->stream_count;
    win = nghttp2_session_get_stream_remote_window_size(stream->session->ngh2, stream->id);
    if (win == 0)
      ++x->stream_send_win_exhausted;
    return 1;
}

int h2_mplx_c1_all_streams_send_win_exhausted(h2_mplx *m)
{
    stream_iter_aws_t x;
    x.stream_count = 0;
    x.stream_send_win_exhausted = 0;
    H2_MPLX_ENTER(m);
    h2_ihash_iter(m->streams, m_stream_send_win_exh, &x);
    H2_MPLX_LEAVE(m);
    return x.stream_count && (x.stream_count == x.stream_send_win_exhausted);
}

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,
                        h2_iqueue *ready_to_process,
                        h2_stream_get_fn *get_stream,
                        h2_stream_pri_cmp_fn *stream_pri_cmp,
                        h2_session *session,
                        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;
#if AP_MODULE_MAGIC_AT_LEAST(20211221, 29)
    ap_set_time_process_request(c2->sbh,conn_ctx->started_at,conn_ctx->done_at);
#endif
    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);
                return;
            }
        }

        ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, c2, APLOGNO(03518)
                      "h2_mplx(%s-%d): c2_done, stream not found",
                      conn_ctx->id, conn_ctx->stream_id);
        ap_assert("stream should still be available" == NULL);
    }
}

static void c2_prod_done(void *baton, conn_rec *c2)
{
    h2_mplx *m = baton;
    h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(c2);

    AP_DEBUG_ASSERT(conn_ctx);
    H2_MPLX_ENTER_ALWAYS(m);

    --m->processing_count;
    s_c2_done(m, c2, conn_ctx);
    if (m->join_wait) apr_thread_cond_signal(m->join_wait);

    H2_MPLX_LEAVE(m);
}

static void workers_shutdown(void *baton, int graceful)
{
    h2_mplx *m = baton;

    apr_thread_mutex_lock(m->poll_lock);
    /* time to wakeup and assess what to do */
    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,
                  H2_MPLX_MSG(m, "workers shutdown, waking pollset"));
    m->shutdown = 1;
    if (!graceful) {
        m->aborted = 1;
    }
    apr_pollset_wakeup(m->pollset);
    apr_thread_mutex_unlock(m->poll_lock);
}

/*******************************************************************************
 * h2_mplx DoS protection
 ******************************************************************************/

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

    if (m->processing_limit < m->processing_max
        && conn_ctx->started_at > m->last_mood_change) {
        --m->irritations_since;
        if (m->processing_limit < m->processing_max
            && ((now = apr_time_now()) - m->last_mood_change >= m->mood_update_interval
                || m->irritations_since < -m->processing_limit)) {
            m->processing_limit = H2MIN(m->processing_limit * 2, m->processing_max);
            m->last_mood_change = now;
            m->irritations_since = 0;
            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
                          H2_MPLX_MSG(m, "mood update, increasing worker limit"
                          "to %d, processing %d right now"),
                          m->processing_limit, m->processing_count);
        }
    }
}

static void m_be_annoyed(h2_mplx *m)
{
    apr_time_t now;

    if (m->processing_limit > 2) {
        ++m->irritations_since;
        if (((now = apr_time_now()) - m->last_mood_change >= m->mood_update_interval)
            || (m->irritations_since >= m->processing_limit)) {

            if (m->processing_limit > 16) {
                m->processing_limit = 16;
            }
            else if (m->processing_limit > 8) {
                m->processing_limit = 8;
            }
            else if (m->processing_limit > 4) {
                m->processing_limit = 4;
            }
            else if (m->processing_limit > 2) {
                m->processing_limit = 2;
            }
            m->last_mood_change = now;
            m->irritations_since = 0;
            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, m->c1,
                          H2_MPLX_MSG(m, "mood update, decreasing worker limit "
                          "to %d, processing %d right now"),
                          m->processing_limit, m->processing_count);
        }
    }
}

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

static int reset_is_acceptable(h2_stream *stream)
{
    /* client may terminate a stream via H2 RST_STREAM message at any time.
     * This is annyoing when we have committed resources (e.g. worker threads)
     * to it, so our mood (e.g. willingness to commit resources on this
     * connection in the future) goes down.
     *
     * This is a DoS protection. We do not want to make it too easy for
     * a client to eat up server resources.
     *
     * However: there are cases where a RST_STREAM is the only way to end
     * a request. This includes websockets and server-side-event streams (SSEs).
     * The responses to such requests continue forever otherwise.
     *
     */
    if (stream->rst_error) return 0; /* errored stream. bad. */
    if (!stream_is_running(stream)) return 1;
    if (!(stream->id & 0x01)) return 1; /* stream initiated by us. acceptable. */
    if (!stream->response) return 0; /* no response headers produced yet. bad. */
    if (!stream->out_data_frames) return 0; /* no response body data sent yet. bad. */
    return 1; /* otherwise, be forgiving */
}

apr_status_t h2_mplx_c1_client_rst(h2_mplx *m, int stream_id, h2_stream *stream)
{
    apr_status_t status = APR_SUCCESS;
    int registered;

    H2_MPLX_ENTER_ALWAYS(m);
    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);
    return status;
}

static apr_status_t mplx_pollset_create(h2_mplx *m)
{
    /* stream0 output only */
    return apr_pollset_create(&m->pollset, 1, m->pool,
                              APR_POLLSET_WAKEABLE);
}

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)
{
    apr_status_t rv;
    const apr_pollfd_t *results, *pfd;
    apr_int32_t nresults, i;
    h2_conn_ctx_t *conn_ctx;
    h2_stream *stream;

    /* Make sure we are not called recursively. */
    ap_assert(!m->polling);
    m->polling = 1;
    do {
        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,
                      H2_MPLX_MSG(m, "enter polling timeout=%d"),
                      (int)apr_time_sec(timeout));

        apr_array_clear(m->streams_ev_in);
        apr_array_clear(m->streams_ev_out);

        do {
            /* add streams we started processing in the meantime */
            apr_thread_mutex_lock(m->poll_lock);
            if (!h2_iq_empty(m->streams_input_read)
                || !h2_iq_empty(m->streams_output_written)) {
                while ((i = h2_iq_shift(m->streams_input_read))) {
                    stream = h2_ihash_get(m->streams, i);
                    if (stream) {
                        APR_ARRAY_PUSH(m->streams_ev_in, h2_stream*) = stream;
                    }
                }
                while ((i = h2_iq_shift(m->streams_output_written))) {
                    stream = h2_ihash_get(m->streams, i);
                    if (stream) {
                        APR_ARRAY_PUSH(m->streams_ev_out, h2_stream*) = stream;
                    }
                }
                nresults = 0;
                rv = APR_SUCCESS;
                apr_thread_mutex_unlock(m->poll_lock);
                break;
            }
            apr_thread_mutex_unlock(m->poll_lock);

            H2_MPLX_LEAVE(m);
            rv = apr_pollset_poll(m->pollset, timeout >= 0? timeout : -1, &nresults, &results);
            H2_MPLX_ENTER_ALWAYS(m);
            if (APR_STATUS_IS_EINTR(rv) && m->shutdown) {
                if (!m->aborted) {
                    rv = APR_SUCCESS;
                }
                goto cleanup;
            }
        } while (APR_STATUS_IS_EINTR(rv));

        if (APR_SUCCESS != rv) {
            if (APR_STATUS_IS_TIMEUP(rv)) {
                ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, m->c1,
                              H2_MPLX_MSG(m, "polling timed out "));
            }
            else {
                ap_log_cerror(APLOG_MARK, APLOG_ERR, rv, m->c1, APLOGNO(10310) \
                              H2_MPLX_MSG(m, "polling failed"));
            }
            goto cleanup;
        }

        for (i = 0; i < nresults; i++) {
            pfd = &results[i];
            conn_ctx = pfd->client_data;

            AP_DEBUG_ASSERT(conn_ctx);
            if (conn_ctx->stream_id == 0) {
                if (on_stream_input) {
                    APR_ARRAY_PUSH(m->streams_ev_in, h2_stream*) = m->stream0;
                }
                break;
            }
        }

        if (on_stream_input && m->streams_ev_in->nelts) {
            H2_MPLX_LEAVE(m);
            for (i = 0; i < m->streams_ev_in->nelts; ++i) {
                on_stream_input(on_ctx, APR_ARRAY_IDX(m->streams_ev_in, i, h2_stream*));
            }
            H2_MPLX_ENTER_ALWAYS(m);
        }
        if (on_stream_output && m->streams_ev_out->nelts) {
            H2_MPLX_LEAVE(m);
            for (i = 0; i < m->streams_ev_out->nelts; ++i) {
                on_stream_output(on_ctx, APR_ARRAY_IDX(m->streams_ev_out, i, h2_stream*));
            }
            H2_MPLX_ENTER_ALWAYS(m);
        }
        break;
    } while(1);

cleanup:
    m->polling = 0;
    return rv;
}