/* 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 <apr_lib.h>

#include <apr_atomic.h>

#include <apr_strings.h>

#include <apr_time.h>

#include <apr_buckets.h>

#include <apr_thread_mutex.h>

#include <apr_thread_cond.h>

#include <httpd.h>

#include <http_protocol.h>

#include <http_log.h>

#include "h2_private.h"

#include "h2_util.h"

#include "h2_bucket_beam.h"

static void h2_beam_emitted(h2_bucket_beam *beam, h2_beam_proxy *proxy);

#define H2_BPROXY_NEXT(e)             APR_RING_NEXT((e), link)

#define H2_BPROXY_PREV(e)             APR_RING_PREV((e), link)

#define H2_BPROXY_REMOVE(e)           APR_RING_REMOVE((e), link)

#define H2_BPROXY_LIST_INIT(b)        APR_RING_INIT(&(b)->list, h2_beam_proxy, link);

#define H2_BPROXY_LIST_SENTINEL(b)    APR_RING_SENTINEL(&(b)->list, h2_beam_proxy, link)

#define H2_BPROXY_LIST_EMPTY(b)       APR_RING_EMPTY(&(b)->list, h2_beam_proxy, link)

#define H2_BPROXY_LIST_FIRST(b)       APR_RING_FIRST(&(b)->list)

#define H2_BPROXY_LIST_LAST(b)	      APR_RING_LAST(&(b)->list)

#define H2_PROXY_BLIST_INSERT_HEAD(b, e) do {				\

	h2_beam_proxy *ap__b = (e);                                        \

	APR_RING_INSERT_HEAD(&(b)->list, ap__b, h2_beam_proxy, link);	\

    } while (0)

#define H2_BPROXY_LIST_INSERT_TAIL(b, e) do {				\

	h2_beam_proxy *ap__b = (e);					\

	APR_RING_INSERT_TAIL(&(b)->list, ap__b, h2_beam_proxy, link);	\

    } while (0)

#define H2_BPROXY_LIST_CONCAT(a, b) do {					\

        APR_RING_CONCAT(&(a)->list, &(b)->list, h2_beam_proxy, link);	\

    } while (0)

#define H2_BPROXY_LIST_PREPEND(a, b) do {					\

        APR_RING_PREPEND(&(a)->list, &(b)->list, h2_beam_proxy, link);	\

    } while (0)

/*******************************************************************************

 * beam bucket with reference to beam and bucket it represents

 ******************************************************************************/

const apr_bucket_type_t h2_bucket_type_beam;

#define H2_BUCKET_IS_BEAM(e)     (e->type == &h2_bucket_type_beam)

struct h2_beam_proxy {

    apr_bucket_refcount refcount;

    APR_RING_ENTRY(h2_beam_proxy) link;

    h2_bucket_beam *beam;

    apr_bucket *bsender;

    apr_size_t n;

};

static const char Dummy = '\0';

static apr_status_t beam_bucket_read(apr_bucket *b, const char **str, 

                                     apr_size_t *len, apr_read_type_e block)

{

    h2_beam_proxy *d = b->data;

    if (d->bsender) {

        const char *data;

        apr_status_t status = apr_bucket_read(d->bsender, &data, len, block);

        if (status == APR_SUCCESS) {

            *str = data + b->start;

            *len = b->length;

        }

        return status;

    }

    *str = &Dummy;

    *len = 0;

    return APR_ECONNRESET;

}

static void beam_bucket_destroy(void *data)

{

    h2_beam_proxy *d = data;

    if (apr_bucket_shared_destroy(d)) {

        /* When the beam gets destroyed before this bucket, it will

         * NULLify its reference here. This is not protected by a mutex,

         * so it will not help with race conditions.

         * But it lets us shut down memory pool with circulare beam

         * references. */

        if (d->beam) {

            h2_beam_emitted(d->beam, d);

        }

        apr_bucket_free(d);

    }

}

static apr_bucket * h2_beam_bucket_make(apr_bucket *b, 

                                        h2_bucket_beam *beam,

                                        apr_bucket *bsender, apr_size_t n)

{

    h2_beam_proxy *d;

    d = apr_bucket_alloc(sizeof(*d), b->list);

    H2_BPROXY_LIST_INSERT_TAIL(&beam->proxies, d);

    d->beam = beam;

    d->bsender = bsender;

    d->n = n;

    b = apr_bucket_shared_make(b, d, 0, bsender? bsender->length : 0);

    b->type = &h2_bucket_type_beam;

    return b;

}

static apr_bucket *h2_beam_bucket_create(h2_bucket_beam *beam,

                                         apr_bucket *bsender,

                                         apr_bucket_alloc_t *list,

                                         apr_size_t n)

{

    apr_bucket *b = apr_bucket_alloc(sizeof(*b), list);

    APR_BUCKET_INIT(b);

    b->free = apr_bucket_free;

    b->list = list;

    return h2_beam_bucket_make(b, beam, bsender, n);

}

const apr_bucket_type_t h2_bucket_type_beam = {

    "BEAM", 5, APR_BUCKET_DATA,

    beam_bucket_destroy,

    beam_bucket_read,

    apr_bucket_setaside_noop,

    apr_bucket_shared_split,

    apr_bucket_shared_copy

};

/*******************************************************************************

 * h2_blist, a brigade without allocations

 ******************************************************************************/

static apr_array_header_t *beamers;

static apr_status_t cleanup_beamers(void *dummy)

{

    (void)dummy;

    beamers = NULL;

    return APR_SUCCESS;

}

void h2_register_bucket_beamer(h2_bucket_beamer *beamer)

{

    if (!beamers) {

        apr_pool_cleanup_register(apr_hook_global_pool, NULL,

                                  cleanup_beamers, apr_pool_cleanup_null);

        beamers = apr_array_make(apr_hook_global_pool, 10, 

                                 sizeof(h2_bucket_beamer*));

    }

    APR_ARRAY_PUSH(beamers, h2_bucket_beamer*) = beamer;

}

static apr_bucket *h2_beam_bucket(h2_bucket_beam *beam, 

                                  apr_bucket_brigade *dest,

                                  const apr_bucket *src)

{

    apr_bucket *b = NULL;

    int i;

    if (beamers) {

        for (i = 0; i < beamers->nelts && b == NULL; ++i) {

            h2_bucket_beamer *beamer;

            beamer = APR_ARRAY_IDX(beamers, i, h2_bucket_beamer*);

            b = beamer(beam, dest, src);

        }

    }

    return b;

}

/*******************************************************************************

 * bucket beam that can transport buckets across threads

 ******************************************************************************/

static void mutex_leave(void *ctx, apr_thread_mutex_t *lock)

{

    apr_thread_mutex_unlock(lock);

}

static apr_status_t mutex_enter(void *ctx, h2_beam_lock *pbl)

{

    h2_bucket_beam *beam = ctx;

    pbl->mutex = beam->lock;

    pbl->leave = mutex_leave;

    return apr_thread_mutex_lock(pbl->mutex);

}

static apr_status_t enter_yellow(h2_bucket_beam *beam, h2_beam_lock *pbl)

{

    return mutex_enter(beam, pbl);

}

static void leave_yellow(h2_bucket_beam *beam, h2_beam_lock *pbl)

{

    if (pbl->leave) {

        pbl->leave(pbl->leave_ctx, pbl->mutex);

    }

}

static apr_off_t bucket_mem_used(apr_bucket *b)

{

    if (APR_BUCKET_IS_FILE(b)) {

        return 0;

    }

    else {

        /* should all have determinate length */

        return b->length;

    }

}

static int report_consumption(h2_bucket_beam *beam, h2_beam_lock *pbl)

{

    int rv = 0;

    apr_off_t len = beam->received_bytes - beam->cons_bytes_reported;

    h2_beam_io_callback *cb = beam->cons_io_cb;

    if (len > 0) {

        if (cb) {

            void *ctx = beam->cons_ctx;

            if (pbl) leave_yellow(beam, pbl);

            cb(ctx, beam, len);

            if (pbl) enter_yellow(beam, pbl);

            rv = 1;

        }

        beam->cons_bytes_reported += len;

    }

    return rv;

}

static void report_prod_io(h2_bucket_beam *beam, int force, h2_beam_lock *pbl)

{

    apr_off_t len = beam->sent_bytes - beam->prod_bytes_reported;

    if (force || len > 0) {

        h2_beam_io_callback *cb = beam->prod_io_cb; 

        if (cb) {

            void *ctx = beam->prod_ctx;

            leave_yellow(beam, pbl);

            cb(ctx, beam, len);

            enter_yellow(beam, pbl);

        }

        beam->prod_bytes_reported += len;

    }

}

static apr_size_t calc_buffered(h2_bucket_beam *beam)

{

    apr_size_t len = 0;

    apr_bucket *b;

    for (b = H2_BLIST_FIRST(&beam->send_list); 

         b != H2_BLIST_SENTINEL(&beam->send_list);

         b = APR_BUCKET_NEXT(b)) {

        if (b->length == ((apr_size_t)-1)) {

            /* do not count */

        }

        else if (APR_BUCKET_IS_FILE(b)) {

            /* if unread, has no real mem footprint. */

        }

        else {

            len += b->length;

        }

    }

    return len;

}

static void r_purge_sent(h2_bucket_beam *beam)

{

    apr_bucket *b;

    /* delete all sender buckets in purge brigade, needs to be called

     * from sender thread only */

    while (!H2_BLIST_EMPTY(&beam->purge_list)) {

        b = H2_BLIST_FIRST(&beam->purge_list);

        apr_bucket_delete(b);

    }

}

static apr_size_t calc_space_left(h2_bucket_beam *beam)

{

    if (beam->max_buf_size > 0) {

        apr_off_t len = calc_buffered(beam);

        return (beam->max_buf_size > len? (beam->max_buf_size - len) : 0);

    }

    return APR_SIZE_MAX;

}

static int buffer_is_empty(h2_bucket_beam *beam)

{

    return ((!beam->recv_buffer || APR_BRIGADE_EMPTY(beam->recv_buffer))

            && H2_BLIST_EMPTY(&beam->send_list));

}

static apr_status_t wait_empty(h2_bucket_beam *beam, apr_read_type_e block,  

                               apr_thread_mutex_t *lock)

{

    apr_status_t rv = APR_SUCCESS;

    while (!buffer_is_empty(beam) && APR_SUCCESS == rv) {

        if (APR_BLOCK_READ != block || !lock) {

            rv = APR_EAGAIN;

        }

        else if (beam->timeout > 0) {

            rv = apr_thread_cond_timedwait(beam->change, lock, beam->timeout);

        }

        else {

            rv = apr_thread_cond_wait(beam->change, lock);

        }

    }

    return rv;

}

static apr_status_t wait_not_empty(h2_bucket_beam *beam, apr_read_type_e block,  

                                   apr_thread_mutex_t *lock)

{

    apr_status_t rv = APR_SUCCESS;

    while (buffer_is_empty(beam) && APR_SUCCESS == rv) {

        if (beam->aborted) {

            rv = APR_ECONNABORTED;

        }

        else if (beam->closed) {

            rv = APR_EOF;

        }

        else if (APR_BLOCK_READ != block || !lock) {

            rv = APR_EAGAIN;

        }

        else if (beam->timeout > 0) {

            rv = apr_thread_cond_timedwait(beam->change, lock, beam->timeout);

        }

        else {

            rv = apr_thread_cond_wait(beam->change, lock);

        }

    }

    return rv;

}

static apr_status_t wait_not_full(h2_bucket_beam *beam, apr_read_type_e block, 

                                  apr_size_t *pspace_left, h2_beam_lock *bl)

{

    apr_status_t rv = APR_SUCCESS;

    apr_size_t left;

    while (0 == (left = calc_space_left(beam)) && APR_SUCCESS == rv) {

        if (beam->aborted) {

            rv = APR_ECONNABORTED;

        }

        else if (block != APR_BLOCK_READ || !bl->mutex) {

            rv = APR_EAGAIN;

        }

        else {

            if (beam->timeout > 0) {

                rv = apr_thread_cond_timedwait(beam->change, bl->mutex, beam->timeout);

            }

            else {

                rv = apr_thread_cond_wait(beam->change, bl->mutex);

            }

        }

    }

    *pspace_left = left;

    return rv;

}

static void h2_beam_emitted(h2_bucket_beam *beam, h2_beam_proxy *proxy)

{

    h2_beam_lock bl;

    apr_bucket *b, *next;

    if (enter_yellow(beam, &bl) == APR_SUCCESS) {

        /* even when beam buckets are split, only the one where

         * refcount drops to 0 will call us */

        H2_BPROXY_REMOVE(proxy);

        /* invoked from receiver thread, the last beam bucket for the send

         * bucket is about to be destroyed.

         * remove it from the hold, where it should be now */

        if (proxy->bsender) {

            for (b = H2_BLIST_FIRST(&beam->hold_list); 

                 b != H2_BLIST_SENTINEL(&beam->hold_list);

                 b = APR_BUCKET_NEXT(b)) {

                 if (b == proxy->bsender) {

                    break;

                 }

            }

            if (b != H2_BLIST_SENTINEL(&beam->hold_list)) {

                /* bucket is in hold as it should be, mark this one

                 * and all before it for purging. We might have placed meta

                 * buckets without a receiver proxy into the hold before it 

                 * and schedule them for purging now */

                for (b = H2_BLIST_FIRST(&beam->hold_list); 

                     b != H2_BLIST_SENTINEL(&beam->hold_list);

                     b = next) {

                    next = APR_BUCKET_NEXT(b);

                    if (b == proxy->bsender) {

                        APR_BUCKET_REMOVE(b);

                        H2_BLIST_INSERT_TAIL(&beam->purge_list, b);

                        break;

                    }

                    else if (APR_BUCKET_IS_METADATA(b)) {

                        APR_BUCKET_REMOVE(b);

                        H2_BLIST_INSERT_TAIL(&beam->purge_list, b);

                    }

                    else {

                        /* another data bucket before this one in hold. this

                         * is normal since DATA buckets need not be destroyed

                         * in order */

                    }

                }

                proxy->bsender = NULL;

            }

            else {

                /* it should be there unless we screwed up */

                ap_log_perror(APLOG_MARK, APLOG_WARNING, 0, beam->send_pool, 

                              APLOGNO(03384) "h2_beam(%d-%s): emitted bucket not "

                              "in hold, n=%d", beam->id, beam->tag, 

                              (int)proxy->n);

                ap_assert(!proxy->bsender);

            }

        }

        /* notify anyone waiting on space to become available */

        if (!bl.mutex) {

            r_purge_sent(beam);

        }

        else {

            apr_thread_cond_broadcast(beam->change);

        }

        leave_yellow(beam, &bl);

    }

}

static void h2_blist_cleanup(h2_blist *bl)

{

    apr_bucket *e;

    while (!H2_BLIST_EMPTY(bl)) {

        e = H2_BLIST_FIRST(bl);

        apr_bucket_delete(e);

    }

}

static apr_status_t beam_close(h2_bucket_beam *beam)

{

    if (!beam->closed) {

        beam->closed = 1;

        apr_thread_cond_broadcast(beam->change);

    }

    return APR_SUCCESS;

}

int h2_beam_is_closed(h2_bucket_beam *beam)

{

    return beam->closed;

}

static int pool_register(h2_bucket_beam *beam, apr_pool_t *pool, 

                         apr_status_t (*cleanup)(void *))

{

    if (pool && pool != beam->pool) {

        apr_pool_pre_cleanup_register(pool, beam, cleanup);

        return 1;

    }

    return 0;

}

static int pool_kill(h2_bucket_beam *beam, apr_pool_t *pool,

                     apr_status_t (*cleanup)(void *)) {

    if (pool && pool != beam->pool) {

        apr_pool_cleanup_kill(pool, beam, cleanup);

        return 1;

    }

    return 0;

}

static apr_status_t beam_recv_cleanup(void *data)

{

    h2_bucket_beam *beam = data;

    /* receiver pool has gone away, clear references */

    beam->recv_buffer = NULL;

    beam->recv_pool = NULL;

    return APR_SUCCESS;

}

static apr_status_t beam_send_cleanup(void *data)

{

    h2_bucket_beam *beam = data;

    /* sender is going away, clear up all references to its memory */

    r_purge_sent(beam);

    h2_blist_cleanup(&beam->send_list);

    report_consumption(beam, NULL);

    while (!H2_BPROXY_LIST_EMPTY(&beam->proxies)) {

        h2_beam_proxy *proxy = H2_BPROXY_LIST_FIRST(&beam->proxies);

        H2_BPROXY_REMOVE(proxy);

        proxy->beam = NULL;

        proxy->bsender = NULL;

    }

    h2_blist_cleanup(&beam->purge_list);

    h2_blist_cleanup(&beam->hold_list);

    beam->send_pool = NULL;

    return APR_SUCCESS;

}

static void beam_set_send_pool(h2_bucket_beam *beam, apr_pool_t *pool) 

{

    if (beam->send_pool != pool) {

        if (beam->send_pool && beam->send_pool != beam->pool) {

            pool_kill(beam, beam->send_pool, beam_send_cleanup);

            beam_send_cleanup(beam);

        }

        beam->send_pool = pool;

        pool_register(beam, beam->send_pool, beam_send_cleanup);

    }

}

static void recv_buffer_cleanup(h2_bucket_beam *beam, h2_beam_lock *bl)

{

    if (beam->recv_buffer && !APR_BRIGADE_EMPTY(beam->recv_buffer)) {

        apr_bucket_brigade *bb = beam->recv_buffer;

        apr_off_t bblen = 0;

        beam->recv_buffer = NULL;

        apr_brigade_length(bb, 0, &bblen);

        beam->received_bytes += bblen;

        /* need to do this unlocked since bucket destroy might 

         * call this beam again. */

        if (bl) leave_yellow(beam, bl);

        apr_brigade_destroy(bb);

        if (bl) enter_yellow(beam, bl);

        apr_thread_cond_broadcast(beam->change);

        if (beam->cons_ev_cb) { 

            beam->cons_ev_cb(beam->cons_ctx, beam);

        }

    }

}

static apr_status_t beam_cleanup(h2_bucket_beam *beam, int from_pool)

{

    apr_status_t status = APR_SUCCESS;

    int safe_send = (beam->owner == H2_BEAM_OWNER_SEND);

    int safe_recv = (beam->owner == H2_BEAM_OWNER_RECV);

    /* 

     * Owner of the beam is going away, depending on which side it owns,

     * cleanup strategies will differ.

     *

     * In general, receiver holds references to memory from sender. 

     * Clean up receiver first, if safe, then cleanup sender, if safe.

     */

     /* When called from pool destroy, io callbacks are disabled */

     if (from_pool) {

         beam->cons_io_cb = NULL;

     }

    /* When modify send is not safe, this means we still have multi-thread

     * protection and the owner is receiving the buckets. If the sending

     * side has not gone away, this means we could have dangling buckets

     * in our lists that never get destroyed. This should not happen. */

    ap_assert(safe_send || !beam->send_pool);

    if (!H2_BLIST_EMPTY(&beam->send_list)) {

        ap_assert(beam->send_pool);

    }

    if (safe_recv) {

        if (beam->recv_pool) {

            pool_kill(beam, beam->recv_pool, beam_recv_cleanup);

            beam->recv_pool = NULL;

        }

        recv_buffer_cleanup(beam, NULL);

    }

    else {

        beam->recv_buffer = NULL;

        beam->recv_pool = NULL;

    }

    if (safe_send && beam->send_pool) {

        pool_kill(beam, beam->send_pool, beam_send_cleanup);

        status = beam_send_cleanup(beam);

    }

    if (safe_recv) {

        ap_assert(H2_BPROXY_LIST_EMPTY(&beam->proxies));

        ap_assert(H2_BLIST_EMPTY(&beam->send_list));

        ap_assert(H2_BLIST_EMPTY(&beam->hold_list));

        ap_assert(H2_BLIST_EMPTY(&beam->purge_list));

    }

    return status;

}

static apr_status_t beam_pool_cleanup(void *data)

{

    return beam_cleanup(data, 1);

}

apr_status_t h2_beam_destroy(h2_bucket_beam *beam)

{

    apr_pool_cleanup_kill(beam->pool, beam, beam_pool_cleanup);

    return beam_cleanup(beam, 0);

}

apr_status_t h2_beam_create(h2_bucket_beam **pbeam, apr_pool_t *pool, 

                            int id, const char *tag, 

                            h2_beam_owner_t owner,

                            apr_size_t max_buf_size,

                            apr_interval_time_t timeout)

{

    h2_bucket_beam *beam;

    apr_status_t rv = APR_SUCCESS;

    beam = apr_pcalloc(pool, sizeof(*beam));

    if (!beam) {

        return APR_ENOMEM;

    }

    beam->id = id;

    beam->tag = tag;

    beam->pool = pool;

    beam->owner = owner;

    H2_BLIST_INIT(&beam->send_list);

    H2_BLIST_INIT(&beam->hold_list);

    H2_BLIST_INIT(&beam->purge_list);

    H2_BPROXY_LIST_INIT(&beam->proxies);

    beam->tx_mem_limits = 1;

    beam->max_buf_size = max_buf_size;

    beam->timeout = timeout;

    rv = apr_thread_mutex_create(&beam->lock, APR_THREAD_MUTEX_DEFAULT, pool);

    if (APR_SUCCESS == rv) {

        rv = apr_thread_cond_create(&beam->change, pool);

        if (APR_SUCCESS == rv) {

            apr_pool_pre_cleanup_register(pool, beam, beam_pool_cleanup);

            *pbeam = beam;

        }

    }

    return rv;

}

void h2_beam_buffer_size_set(h2_bucket_beam *beam, apr_size_t buffer_size)

{

    h2_beam_lock bl;

    if (enter_yellow(beam, &bl) == APR_SUCCESS) {

        beam->max_buf_size = buffer_size;

        leave_yellow(beam, &bl);

    }

}

apr_size_t h2_beam_buffer_size_get(h2_bucket_beam *beam)

{

    h2_beam_lock bl;

    apr_size_t buffer_size = 0;

    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {

        buffer_size = beam->max_buf_size;

        leave_yellow(beam, &bl);

    }

    return buffer_size;

}

void h2_beam_timeout_set(h2_bucket_beam *beam, apr_interval_time_t timeout)

{

    h2_beam_lock bl;

    if (enter_yellow(beam, &bl) == APR_SUCCESS) {

        beam->timeout = timeout;

        leave_yellow(beam, &bl);

    }

}

apr_interval_time_t h2_beam_timeout_get(h2_bucket_beam *beam)

{

    h2_beam_lock bl;

    apr_interval_time_t timeout = 0;

    if (enter_yellow(beam, &bl) == APR_SUCCESS) {

        timeout = beam->timeout;

        leave_yellow(beam, &bl);

    }

    return timeout;

}

void h2_beam_abort(h2_bucket_beam *beam)

{

    h2_beam_lock bl;

    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {

        beam->aborted = 1;

        r_purge_sent(beam);

        h2_blist_cleanup(&beam->send_list);

        report_consumption(beam, &bl);

        apr_thread_cond_broadcast(beam->change);

        leave_yellow(beam, &bl);

    }

}

apr_status_t h2_beam_close(h2_bucket_beam *beam)

{

    h2_beam_lock bl;

    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {

        r_purge_sent(beam);

        beam_close(beam);

        report_consumption(beam, &bl);

        leave_yellow(beam, &bl);

    }

    return beam->aborted? APR_ECONNABORTED : APR_SUCCESS;

}

apr_status_t h2_beam_leave(h2_bucket_beam *beam)

{

    h2_beam_lock bl;

    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {

        recv_buffer_cleanup(beam, &bl);

        beam->aborted = 1;

        beam_close(beam);

        leave_yellow(beam, &bl);

    }

    return APR_SUCCESS;

}

apr_status_t h2_beam_wait_empty(h2_bucket_beam *beam, apr_read_type_e block)

{

    apr_status_t status;

    h2_beam_lock bl;

    if ((status = enter_yellow(beam, &bl)) == APR_SUCCESS) {

        status = wait_empty(beam, block, bl.mutex);

        leave_yellow(beam, &bl);

    }

    return status;

}

static void move_to_hold(h2_bucket_beam *beam, 

                         apr_bucket_brigade *sender_bb)

{

    apr_bucket *b;

    while (sender_bb && !APR_BRIGADE_EMPTY(sender_bb)) {

        b = APR_BRIGADE_FIRST(sender_bb);

        APR_BUCKET_REMOVE(b);

        H2_BLIST_INSERT_TAIL(&beam->send_list, b);

    }

}

static apr_status_t append_bucket(h2_bucket_beam *beam, 

                                  apr_bucket *b,

                                  apr_read_type_e block,

                                  apr_size_t *pspace_left,

                                  h2_beam_lock *pbl)

{

    const char *data;

    apr_size_t len;

    apr_status_t status;

    int can_beam = 0, check_len;

    if (beam->aborted) {

        return APR_ECONNABORTED;

    }

    if (APR_BUCKET_IS_METADATA(b)) {

        if (APR_BUCKET_IS_EOS(b)) {

            beam->closed = 1;

        }

        APR_BUCKET_REMOVE(b);

        H2_BLIST_INSERT_TAIL(&beam->send_list, b);

        return APR_SUCCESS;

    }

    else if (APR_BUCKET_IS_FILE(b)) {

        /* For file buckets the problem is their internal readpool that

         * is used on the first read to allocate buffer/mmap.

         * Since setting aside a file bucket will de-register the

         * file cleanup function from the previous pool, we need to

         * call that only from the sender thread.

         *

         * Currently, we do not handle file bucket with refcount > 1 as

         * the beam is then not in complete control of the file's lifetime.

         * Which results in the bug that a file get closed by the receiver

         * while the sender or the beam still have buckets using it. 

         * 

         * Additionally, we allow callbacks to prevent beaming file

         * handles across. The use case for this is to limit the number 

         * of open file handles and rather use a less efficient beam

         * transport. */

        apr_bucket_file *bf = b->data;

        apr_file_t *fd = bf->fd;

        can_beam = (bf->refcount.refcount == 1);

        if (can_beam && beam->can_beam_fn) {

            can_beam = beam->can_beam_fn(beam->can_beam_ctx, beam, fd);

        }

        check_len = !can_beam;

    }

    else {

        if (b->length == ((apr_size_t)-1)) {

            const char *data;

            status = apr_bucket_read(b, &data, &len, APR_BLOCK_READ);

            if (status != APR_SUCCESS) {

                return status;

            }

        }

        check_len = 1;

    }

    if (check_len) {

        if (b->length > *pspace_left) {

            apr_bucket_split(b, *pspace_left);

        }

        *pspace_left -= b->length;

    }

    /* The fundamental problem is that reading a sender bucket from

     * a receiver thread is a total NO GO, because the bucket might use

     * its pool/bucket_alloc from a foreign thread and that will

     * corrupt. */

    status = APR_ENOTIMPL;

    if (APR_BUCKET_IS_TRANSIENT(b)) {

        /* this takes care of transient buckets and converts them

         * into heap ones. Other bucket types might or might not be

         * affected by this. */

        status = apr_bucket_setaside(b, beam->send_pool);

    }

    else if (APR_BUCKET_IS_HEAP(b)) {

        /* For heap buckets read from a receiver thread is fine. The

         * data will be there and live until the bucket itself is

         * destroyed. */

        status = APR_SUCCESS;

    }

    else if (APR_BUCKET_IS_POOL(b)) {

        /* pool buckets are bastards that register at pool cleanup

         * to morph themselves into heap buckets. That may happen anytime,

         * even after the bucket data pointer has been read. So at

         * any time inside the receiver thread, the pool bucket memory

         * may disappear. yikes. */

        status = apr_bucket_read(b, &data, &len, APR_BLOCK_READ);

        if (status == APR_SUCCESS) {

            apr_bucket_heap_make(b, data, len, NULL);

        }

    }

    else if (APR_BUCKET_IS_FILE(b) && can_beam) {

        status = apr_bucket_setaside(b, beam->send_pool);

    }

    if (status == APR_ENOTIMPL) {

        /* we have no knowledge about the internals of this bucket,

         * but hope that after read, its data stays immutable for the

         * lifetime of the bucket. (see pool bucket handling above for

         * a counter example).

         * We do the read while in the sender thread, so that the bucket may

         * use pools/allocators safely. */

        status = apr_bucket_read(b, &data, &len, APR_BLOCK_READ);

        if (status == APR_SUCCESS) {

            status = apr_bucket_setaside(b, beam->send_pool);

        }

    }

    if (status != APR_SUCCESS && status != APR_ENOTIMPL) {

        return status;