/* 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_log.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_conn.h"

#include "h2_ctx.h"

#include "h2_h2.h"

#include "h2_mplx.h"

#include "h2_ngn_shed.h"

#include "h2_request.h"

#include "h2_stream.h"

#include "h2_session.h"

#include "h2_task.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;

} stream_iter_ctx;

apr_status_t h2_mplx_child_init(apr_pool_t *pool, server_rec *s)

{

    return APR_SUCCESS;

}

#define H2_MPLX_ENTER(m)    \

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

        return rv;\

    } } 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, lock)    \

    if (lock) apr_thread_mutex_lock(m->lock)

#define H2_MPLX_LEAVE_MAYBE(m, lock)    \

    if (lock) apr_thread_mutex_unlock(m->lock)

static void check_data_for(h2_mplx *m, h2_stream *stream, int lock);

static void stream_output_consumed(void *ctx, 

                                   h2_bucket_beam *beam, apr_off_t length)

{

    h2_stream *stream = ctx;

    h2_task *task = stream->task;

    if (length > 0 && task && task->assigned) {

        h2_req_engine_out_consumed(task->assigned, task->c, length); 

    }

}

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

{

    h2_stream *stream = ctx;

    h2_mplx *m = stream->session->mplx;

    apr_atomic_set32(&m->event_pending, 1); 

}

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

{

    h2_stream_in_consumed(ctx, length);

}

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

{

    ap_assert(!stream->task || stream->task->worker_done);

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

    h2_ihash_add(m->spurge, stream);

}

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

{

    ap_assert(stream->state == H2_SS_CLEANUP);

    if (stream->input) {

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

        h2_beam_abort(stream->input);

    }

    if (stream->output) {

        h2_beam_on_produced(stream->output, NULL, NULL);

        h2_beam_leave(stream->output);

    }

    h2_stream_cleanup(stream);

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

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

    h2_ififo_remove(m->readyq, stream->id);

    h2_ihash_add(m->shold, stream);

    if (!stream->task || stream->task->worker_done) {

        stream_joined(m, stream);

    }

    else if (stream->task) {

        stream->task->c->aborted = 1;

        apr_thread_cond_broadcast(m->task_thawed);

    }

}

/**

 * 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_create(conn_rec *c, apr_pool_t *parent, 

                        const h2_config *conf, 

                        h2_workers *workers)

{

    apr_status_t status = APR_SUCCESS;

    apr_allocator_t *allocator;

    apr_thread_mutex_t *mutex;

    h2_mplx *m;

    h2_ctx *ctx = h2_ctx_get(c, 0);

    ap_assert(conf);

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

    if (m) {

        m->id = c->id;

        m->c = c;

        m->s = (ctx? h2_ctx_server_get(ctx) : NULL);

        if (!m->s) {

            m->s = c->base_server;

        }

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

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

         * processing independant 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) {

            return NULL;

        }

        apr_allocator_max_free_set(allocator, ap_max_mem_free);

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

        if (!m->pool) {

            apr_allocator_destroy(allocator);

            return NULL;

        }

        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 (status != APR_SUCCESS) {

            apr_pool_destroy(m->pool);

            return NULL;

        }

        apr_allocator_mutex_set(allocator, mutex);

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

                                         m->pool);

        if (status != APR_SUCCESS) {

            apr_pool_destroy(m->pool);

            return NULL;

        }

        status = apr_thread_cond_create(&m->task_thawed, m->pool);

        if (status != APR_SUCCESS) {

            apr_pool_destroy(m->pool);

            return NULL;

        }

        m->max_streams = h2_config_geti(conf, H2_CONF_MAX_STREAMS);

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

        status = h2_ififo_set_create(&m->readyq, m->pool, m->max_streams);

        if (status != APR_SUCCESS) {

            apr_pool_destroy(m->pool);

            return NULL;

        }

        m->workers = workers;

        m->max_active = workers->max_workers;

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

        m->last_limit_change = m->last_idle_block = apr_time_now();

        m->limit_change_interval = apr_time_from_msec(100);

        m->spare_slaves = apr_array_make(m->pool, 10, sizeof(conn_rec*));

        m->ngn_shed = h2_ngn_shed_create(m->pool, m->c, m->max_streams, 

                                         m->stream_max_mem);

        h2_ngn_shed_set_ctx(m->ngn_shed , m);

    }

    return m;

}

int h2_mplx_shutdown(h2_mplx *m)

{

    int max_stream_started = 0;

    H2_MPLX_ENTER(m);

    max_stream_started = m->max_stream_started;

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

    h2_iq_clear(m->q);

    H2_MPLX_LEAVE(m);

    return max_stream_started;

}

static int input_consumed_signal(h2_mplx *m, h2_stream *stream)

{

    if (stream->input) {

        return h2_beam_report_consumption(stream->input);

    }

    return 0;

}

static int report_consumption_iter(void *ctx, void *val)

{

    h2_stream *stream = val;

    h2_mplx *m = ctx;

    input_consumed_signal(m, stream);

    if (stream->state == H2_SS_CLOSED_L

        && (!stream->task || stream->task->worker_done)) {

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

                      H2_STRM_LOG(APLOGNO(10026), stream, "remote close missing")); 

        nghttp2_submit_rst_stream(stream->session->ngh2, NGHTTP2_FLAG_NONE, 

                                  stream->id, NGHTTP2_NO_ERROR);

    }

    return 1;

}

static int output_consumed_signal(h2_mplx *m, h2_task *task)

{

    if (task->output.beam) {

        return h2_beam_report_consumption(task->output.beam);

    }

    return 0;

}

static int stream_destroy_iter(void *ctx, void *val) 

{   

    h2_mplx *m = ctx;

    h2_stream *stream = val;

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

    ap_assert(stream->state == H2_SS_CLEANUP);

    if (stream->input) {

        /* Process outstanding events before destruction */

        input_consumed_signal(m, stream);    

        h2_beam_log(stream->input, m->c, APLOG_TRACE2, "stream_destroy");

        h2_beam_destroy(stream->input);

        stream->input = NULL;

    }

    if (stream->task) {

        h2_task *task = stream->task;

        conn_rec *slave;

        int reuse_slave = 0;

        stream->task = NULL;

        slave = task->c;

        if (slave) {

            /* On non-serialized requests, the IO logging has not accounted for any

             * meta data send over the network: response headers and h2 frame headers. we

             * counted this on the stream and need to add this now.

             * This is supposed to happen before the EOR bucket triggers the

             * logging of the transaction. *fingers crossed* */

            if (task->request && !task->request->serialize && h2_task_logio_add_bytes_out) {

                apr_off_t unaccounted = stream->out_frame_octets - stream->out_data_octets;

                if (unaccounted > 0) {

                    h2_task_logio_add_bytes_out(slave, unaccounted);

                }

            }

            if (m->s->keep_alive_max == 0 || slave->keepalives < m->s->keep_alive_max) {

                reuse_slave = ((m->spare_slaves->nelts < (m->limit_active * 3 / 2))

                               && !task->rst_error);

            }

            if (reuse_slave && slave->keepalive == AP_CONN_KEEPALIVE) {

                h2_beam_log(task->output.beam, m->c, APLOG_DEBUG, 

                            APLOGNO(03385) "h2_task_destroy, reuse slave");    

                h2_task_destroy(task);

                APR_ARRAY_PUSH(m->spare_slaves, conn_rec*) = slave;

            }

            else {

                h2_beam_log(task->output.beam, m->c, APLOG_TRACE1, 

                            "h2_task_destroy, destroy slave");    

                h2_slave_destroy(slave);

            }

        }

    }

    h2_stream_destroy(stream);

    return 0;

}

static void purge_streams(h2_mplx *m, int lock)

{

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

        H2_MPLX_ENTER_MAYBE(m, lock);

        while (!h2_ihash_iter(m->spurge, stream_destroy_iter, m)) {

            /* repeat until empty */

        }

        H2_MPLX_LEAVE_MAYBE(m, lock);

    }

}

typedef struct {

    h2_mplx_stream_cb *cb;

    void *ctx;

} stream_iter_ctx_t;

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

{

    stream_iter_ctx_t *x = ctx;

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

}

apr_status_t h2_mplx_stream_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, stream_iter_wrap, &x);

    H2_MPLX_LEAVE(m);

    return APR_SUCCESS;

}

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

    h2_mplx *m = ctx;

    h2_stream *stream = val;

    h2_task *task = stream->task;

    if (APLOGctrace1(m->c)) {

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

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

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

                      (long)h2_beam_get_buffered(stream->output));

    }

    if (task) {

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

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

                      "[started=%d/done=%d/frozen=%d]"), 

                      task->request->method, task->request->authority, 

                      task->request->path, task->worker_started, 

                      task->worker_done, task->frozen);

    }

    else {

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

                      H2_STRM_MSG(stream, "->03198: no task"));

    }

    return 1;

}

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

    h2_mplx *m = ctx;

    h2_stream *stream = val;

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

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

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

    return 1;

}

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

    h2_mplx *m = ctx;

    h2_stream *stream = val;

    /* disabled input consumed reporting */

    if (stream->input) {

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

    }

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

    stream_cleanup(m, stream);  

    return 0;

}

void h2_mplx_release_and_join(h2_mplx *m, apr_thread_cond_t *wait)

{

    apr_status_t status;

    int i, wait_secs = 60;

    /* How to shut down a h2 connection:

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

    m->aborted = 1;

    h2_workers_unregister(m->workers, m);

    H2_MPLX_ENTER_ALWAYS(m);

    /* How to shut down a h2 connection:

     * 1. cancel all streams still active */

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

        /* until empty */

    }

    /* 2. terminate ngn_shed, no more streams

     * should be scheduled or in the active set */

    h2_ngn_shed_abort(m->ngn_shed);

    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 tasks from this connection, wait on them finishing

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

     *    Eventually, this has to succeed. */    

    m->join_wait = wait;

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

        status = apr_thread_cond_timedwait(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->c, APLOGNO(03198)

                          "h2_mplx(%ld): waited %d sec for %d tasks", 

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

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

        }

    }

    ap_assert(m->tasks_active == 0);

    m->join_wait = NULL;

    /* 4. close the h2_req_enginge shed */

    h2_ngn_shed_destroy(m->ngn_shed);

    m->ngn_shed = NULL;

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

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

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

                      "h2_mplx(%ld): unexpected %d streams in hold", 

                      m->id, (int)h2_ihash_count(m->shold));

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

    }

    H2_MPLX_LEAVE(m);

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

                  "h2_mplx(%ld): released", m->id);

}

apr_status_t h2_mplx_stream_cleanup(h2_mplx *m, h2_stream *stream)

{

    H2_MPLX_ENTER(m);

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

                  H2_STRM_MSG(stream, "cleanup"));

    stream_cleanup(m, stream);        

    H2_MPLX_LEAVE(m);

    return APR_SUCCESS;

}

h2_stream *h2_mplx_stream_get(h2_mplx *m, int id)

{

    h2_stream *s = NULL;

    H2_MPLX_ENTER_ALWAYS(m);

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

    H2_MPLX_LEAVE(m);

    return s;

}

static void output_produced(void *ctx, h2_bucket_beam *beam, apr_off_t bytes)

{

    h2_stream *stream = ctx;

    h2_mplx *m = stream->session->mplx;

    check_data_for(m, stream, 1);

}

static apr_status_t out_open(h2_mplx *m, int stream_id, h2_bucket_beam *beam)

{

    apr_status_t status = APR_SUCCESS;

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

    if (!stream || !stream->task || m->aborted) {

        return APR_ECONNABORTED;

    }

    ap_assert(stream->output == NULL);

    stream->output = beam;

    if (APLOGctrace2(m->c)) {

        h2_beam_log(beam, m->c, APLOG_TRACE2, "out_open");

    }

    else {

        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, m->c,

                      "h2_mplx(%s): out open", stream->task->id);

    }

    h2_beam_on_consumed(stream->output, NULL, stream_output_consumed, stream);

    h2_beam_on_produced(stream->output, output_produced, stream);

    if (stream->task->output.copy_files) {

        h2_beam_on_file_beam(stream->output, h2_beam_no_files, NULL);

    }

    /* we might see some file buckets in the output, see

     * if we have enough handles reserved. */

    check_data_for(m, stream, 0);

    return status;

}

apr_status_t h2_mplx_out_open(h2_mplx *m, int stream_id, h2_bucket_beam *beam)

{

    apr_status_t status;

    H2_MPLX_ENTER(m);

    if (m->aborted) {

        status = APR_ECONNABORTED;

    }

    else {

        status = out_open(m, stream_id, beam);

    }

    H2_MPLX_LEAVE(m);

    return status;

}

static apr_status_t out_close(h2_mplx *m, h2_task *task)

{

    apr_status_t status = APR_SUCCESS;

    h2_stream *stream;

    if (!task) {

        return APR_ECONNABORTED;

    }

    if (task->c) {

        ++task->c->keepalives;

    }

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

    if (!stream) {

        return APR_ECONNABORTED;

    }

    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, m->c,

                  "h2_mplx(%s): close", task->id);

    status = h2_beam_close(task->output.beam);

    h2_beam_log(task->output.beam, m->c, APLOG_TRACE2, "out_close");

    output_consumed_signal(m, task);

    check_data_for(m, stream, 0);

    return status;

}

apr_status_t h2_mplx_out_trywait(h2_mplx *m, apr_interval_time_t timeout,

                                 apr_thread_cond_t *iowait)

{

    apr_status_t status;

    H2_MPLX_ENTER(m);

    if (m->aborted) {

        status = APR_ECONNABORTED;

    }

    else if (h2_mplx_has_master_events(m)) {

        status = APR_SUCCESS;

    }

    else {

        purge_streams(m, 0);

        h2_ihash_iter(m->streams, report_consumption_iter, m);

        m->added_output = iowait;

        status = apr_thread_cond_timedwait(m->added_output, m->lock, timeout);

        if (APLOGctrace2(m->c)) {

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

                          "h2_mplx(%ld): trywait on data for %f ms)",

                          m->id, timeout/1000.0);

        }

        m->added_output = NULL;

    }

    H2_MPLX_LEAVE(m);

    return status;

}

static void check_data_for(h2_mplx *m, h2_stream *stream, int lock)

{

    if (h2_ififo_push(m->readyq, stream->id) == APR_SUCCESS) {

        apr_atomic_set32(&m->event_pending, 1);

        H2_MPLX_ENTER_MAYBE(m, lock);

        if (m->added_output) {

            apr_thread_cond_signal(m->added_output);

        }

        H2_MPLX_LEAVE_MAYBE(m, lock);

    }

}

apr_status_t h2_mplx_reprioritize(h2_mplx *m, h2_stream_pri_cmp *cmp, void *ctx)

{

    apr_status_t status;

    H2_MPLX_ENTER(m);

    if (m->aborted) {

        status = APR_ECONNABORTED;

    }

    else {

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

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

                      "h2_mplx(%ld): reprioritize tasks", m->id);

        status = APR_SUCCESS;

    }

    H2_MPLX_LEAVE(m);

    return status;

}

static void register_if_needed(h2_mplx *m) 

{

    if (!m->aborted && !m->is_registered && !h2_iq_empty(m->q)) {

        apr_status_t status = h2_workers_register(m->workers, m); 

        if (status == APR_SUCCESS) {

            m->is_registered = 1;

        }

        else {

            ap_log_cerror(APLOG_MARK, APLOG_ERR, status, m->c, APLOGNO(10021)

                          "h2_mplx(%ld): register at workers", m->id);

        }

    }

}

apr_status_t h2_mplx_process(h2_mplx *m, struct h2_stream *stream, 

                             h2_stream_pri_cmp *cmp, void *ctx)

{

    apr_status_t status;

    H2_MPLX_ENTER(m);

    if (m->aborted) {

        status = APR_ECONNABORTED;

    }

    else {

        status = APR_SUCCESS;

        h2_ihash_add(m->streams, stream);

        if (h2_stream_is_ready(stream)) {

            /* already have a response */

            check_data_for(m, stream, 0);

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

                          H2_STRM_MSG(stream, "process, add to readyq")); 

        }

        else {

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

            register_if_needed(m);                

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

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

        }

    }

    H2_MPLX_LEAVE(m);

    return status;

}

static h2_task *next_stream_task(h2_mplx *m)

{

    h2_stream *stream;

    int sid;

    while (!m->aborted && (m->tasks_active < m->limit_active)

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

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

        if (stream) {

            conn_rec *slave, **pslave;

            pslave = (conn_rec **)apr_array_pop(m->spare_slaves);

            if (pslave) {

                slave = *pslave;

                slave->aborted = 0;

            }

            else {

                slave = h2_slave_create(m->c, stream->id, m->pool);

            }

            if (!stream->task) {

                if (sid > m->max_stream_started) {

                    m->max_stream_started = sid;

                }

                if (stream->input) {

                    h2_beam_on_consumed(stream->input, stream_input_ev, 

                                        stream_input_consumed, stream);

                }

                stream->task = h2_task_create(slave, stream->id, 

                                              stream->request, m, stream->input, 

                                              stream->session->s->timeout,

                                              m->stream_max_mem);

                if (!stream->task) {

                    ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_ENOMEM, slave,

                                  H2_STRM_LOG(APLOGNO(02941), stream, 

                                  "create task"));

                    return NULL;

                }

            }

            ++m->tasks_active;

            return stream->task;

        }

    }

    return NULL;

}

apr_status_t h2_mplx_pop_task(h2_mplx *m, h2_task **ptask)

{

    apr_status_t rv = APR_EOF;

    *ptask = NULL;

    ap_assert(m);

    ap_assert(m->lock);

    if (APR_SUCCESS != (rv = apr_thread_mutex_lock(m->lock))) {

        return rv;

    }

    if (m->aborted) {

        rv = APR_EOF;

    }

    else {

        *ptask = next_stream_task(m);

        rv = (*ptask != NULL && !h2_iq_empty(m->q))? APR_EAGAIN : APR_SUCCESS;

    }

    if (APR_EAGAIN != rv) {

        m->is_registered = 0; /* h2_workers will discard this mplx */

    }

    H2_MPLX_LEAVE(m);

    return rv;

}

static void task_done(h2_mplx *m, h2_task *task, h2_req_engine *ngn)

{

    h2_stream *stream;

    if (task->frozen) {

        /* this task was handed over to an engine for processing 

         * and the original worker has finished. That means the 

         * engine may start processing now. */

        h2_task_thaw(task);

        apr_thread_cond_broadcast(m->task_thawed);

        return;

    }

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

                  "h2_mplx(%ld): task(%s) done", m->id, task->id);

    out_close(m, task);

    if (ngn) {

        apr_off_t bytes = 0;

        h2_beam_send(task->output.beam, NULL, APR_NONBLOCK_READ);

        bytes += h2_beam_get_buffered(task->output.beam);

        if (bytes > 0) {

            /* we need to report consumed and current buffered output

             * to the engine. The request will be streamed out or cancelled,

             * no more data is coming from it and the engine should update

             * its calculations before we destroy this information. */

            h2_req_engine_out_consumed(ngn, task->c, bytes);

        }

    }

    if (task->engine) {

        if (!m->aborted && !task->c->aborted 

            && !h2_req_engine_is_shutdown(task->engine)) {

            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c, APLOGNO(10022)

                          "h2_mplx(%ld): task(%s) has not-shutdown "

                          "engine(%s)", m->id, task->id, 

                          h2_req_engine_get_id(task->engine));

        }

        h2_ngn_shed_done_ngn(m->ngn_shed, task->engine);

    }

    task->worker_done = 1;

    task->done_at = apr_time_now();

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

                  "h2_mplx(%s): request done, %f ms elapsed", task->id, 

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

    if (task->started_at > m->last_idle_block) {

        /* this task finished without causing an 'idle block', e.g.

         * a block by flow control.

         */

        if (task->done_at- m->last_limit_change >= m->limit_change_interval

            && m->limit_active < m->max_active) {

            /* Well behaving stream, allow it more workers */

            m->limit_active = H2MIN(m->limit_active * 2, 

                                     m->max_active);

            m->last_limit_change = task->done_at;

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

                          "h2_mplx(%ld): increase worker limit to %d",

                          m->id, m->limit_active);

        }

    }

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

    if (stream) {

        /* stream not done yet. */

        if (!m->aborted && h2_ihash_get(m->sredo, stream->id)) {

            /* reset and schedule again */

            h2_task_redo(task);

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

            h2_iq_add(m->q, stream->id, NULL, NULL);

        }

        else {

            /* stream not cleaned up, stay around */

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

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

            if (stream->input) {

                h2_beam_leave(stream->input);

            }

            /* more data will not arrive, resume the stream */

            check_data_for(m, stream, 0);            

        }

    }

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

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

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

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

        if (stream->input) {

            h2_beam_leave(stream->input);

        }

        stream_joined(m, stream);

    }

    else if ((stream = h2_ihash_get(m->spurge, task->stream_id)) != NULL) {