Merging upstream version 2.4.59.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 535 insertions, 935 deletions

diff --git a/modules/http2/h2_bucket_beam.c b/modules/http2/h2_bucket_beam.c
index f79cbe3..6978254 100644
--- a/modules/http2/h2_bucket_beam.c
+++ b/modules/http2/h2_bucket_beam.c
@@ -24,261 +24,123 @@
 
 #include <httpd.h>
 #include <http_protocol.h>
+#include <http_request.h>
 #include <http_log.h>
 
 #include "h2_private.h"
+#include "h2_conn_ctx.h"
+#include "h2_headers.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);	\
+#define H2_BLIST_INIT(b)        APR_RING_INIT(&(b)->list, apr_bucket, link);
+#define H2_BLIST_SENTINEL(b)    APR_RING_SENTINEL(&(b)->list, apr_bucket, link)
+#define H2_BLIST_EMPTY(b)       APR_RING_EMPTY(&(b)->list, apr_bucket, link)
+#define H2_BLIST_FIRST(b)       APR_RING_FIRST(&(b)->list)
+#define H2_BLIST_LAST(b)	APR_RING_LAST(&(b)->list)
+#define H2_BLIST_INSERT_HEAD(b, e) do {				\
+	apr_bucket *ap__b = (e);                                        \
+	APR_RING_INSERT_HEAD(&(b)->list, ap__b, apr_bucket, 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);	\
+#define H2_BLIST_INSERT_TAIL(b, e) do {				\
+	apr_bucket *ap__b = (e);					\
+	APR_RING_INSERT_TAIL(&(b)->list, ap__b, apr_bucket, link);	\
     } while (0)
-#define H2_BPROXY_LIST_CONCAT(a, b) do {					\
-        APR_RING_CONCAT(&(a)->list, &(b)->list, h2_beam_proxy, link);	\
+#define H2_BLIST_CONCAT(a, b) do {					\
+        APR_RING_CONCAT(&(a)->list, &(b)->list, apr_bucket, link);	\
     } while (0)
-#define H2_BPROXY_LIST_PREPEND(a, b) do {					\
-        APR_RING_PREPEND(&(a)->list, &(b)->list, h2_beam_proxy, link);	\
+#define H2_BLIST_PREPEND(a, b) do {					\
+        APR_RING_PREPEND(&(a)->list, &(b)->list, apr_bucket, 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 int buffer_is_empty(h2_bucket_beam *beam);
+static apr_off_t get_buffered_data_len(h2_bucket_beam *beam);
 
-static apr_status_t cleanup_beamers(void *dummy)
+static int h2_blist_count(h2_blist *blist)
 {
-    (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);
-}
+    apr_bucket *b;
+    int count = 0;
 
-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);
-}
+    for (b = H2_BLIST_FIRST(blist); b != H2_BLIST_SENTINEL(blist);
+         b = APR_BUCKET_NEXT(b)) {
+        ++count;
+    }
+    return count;
+}
+
+#define H2_BEAM_LOG(beam, c, level, rv, msg, bb) \
+    do { \
+        if (APLOG_C_IS_LEVEL((c),(level))) { \
+            char buffer[4 * 1024]; \
+            apr_size_t len, bmax = sizeof(buffer)/sizeof(buffer[0]); \
+            len = bb? h2_util_bb_print(buffer, bmax, "", "", bb) : 0; \
+            ap_log_cerror(APLOG_MARK, (level), rv, (c), \
+                          "BEAM[%s,%s%sdata=%ld,buckets(send/consumed)=%d/%d]: %s %s", \
+                          (beam)->name, \
+                          (beam)->aborted? "aborted," : "", \
+                          buffer_is_empty(beam)? "empty," : "", \
+                          (long)get_buffered_data_len(beam), \
+                          h2_blist_count(&(beam)->buckets_to_send), \
+                          h2_blist_count(&(beam)->buckets_consumed), \
+                          (msg), len? buffer : ""); \
+        } \
+    } while (0)
 
-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)
+static int bucket_is_mmap(apr_bucket *b)
 {
-    if (pbl->leave) {
-        pbl->leave(pbl->leave_ctx, pbl->mutex);
-    }
+#if APR_HAS_MMAP
+    return APR_BUCKET_IS_MMAP(b);
+#else
+    /* if it is not defined as enabled, it should always be no */
+    return 0;
+#endif
 }
 
 static apr_off_t bucket_mem_used(apr_bucket *b)
 {
-    if (APR_BUCKET_IS_FILE(b)) {
+    if (APR_BUCKET_IS_FILE(b) || bucket_is_mmap(b)) {
         return 0;
     }
     else {
         /* should all have determinate length */
-        return b->length;
+        return (apr_off_t)b->length;
     }
 }
 
-static int report_consumption(h2_bucket_beam *beam, h2_beam_lock *pbl)
+static int report_consumption(h2_bucket_beam *beam, int locked)
 {
     int rv = 0;
-    apr_off_t len = beam->received_bytes - beam->cons_bytes_reported;
+    apr_off_t len = beam->recv_bytes - beam->recv_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);
+            if (locked) apr_thread_mutex_unlock(beam->lock);
             cb(ctx, beam, len);
-            if (pbl) enter_yellow(beam, pbl);
+            if (locked) apr_thread_mutex_lock(beam->lock);
             rv = 1;
         }
-        beam->cons_bytes_reported += len;
+        beam->recv_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);
+    for (b = H2_BLIST_FIRST(&beam->buckets_to_send);
+         b != H2_BLIST_SENTINEL(&beam->buckets_to_send);
          b = APR_BUCKET_NEXT(b)) {
         if (b->length == ((apr_size_t)-1)) {
             /* do not count */
         }
-        else if (APR_BUCKET_IS_FILE(b)) {
+        else if (APR_BUCKET_IS_FILE(b) || bucket_is_mmap(b)) {
             /* if unread, has no real mem footprint. */
         }
         else {
@@ -288,13 +150,30 @@ static apr_size_t calc_buffered(h2_bucket_beam *beam)
     return len;
 }
 
-static void r_purge_sent(h2_bucket_beam *beam)
+static void purge_consumed_buckets(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->buckets_consumed)) {
+        b = H2_BLIST_FIRST(&beam->buckets_consumed);
+        if(AP_BUCKET_IS_EOR(b)) {
+          APR_BUCKET_REMOVE(b);
+          H2_BLIST_INSERT_TAIL(&beam->buckets_eor, b);
+        }
+        else {
+          apr_bucket_delete(b);
+        }
+    }
+}
+
+static void purge_eor_buckets(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);
+    while (!H2_BLIST_EMPTY(&beam->buckets_eor)) {
+        b = H2_BLIST_FIRST(&beam->buckets_eor);
         apr_bucket_delete(b);
     }
 }
@@ -302,7 +181,7 @@ static void r_purge_sent(h2_bucket_beam *beam)
 static apr_size_t calc_space_left(h2_bucket_beam *beam)
 {
     if (beam->max_buf_size > 0) {
-        apr_off_t len = calc_buffered(beam);
+        apr_size_t len = calc_buffered(beam);
         return (beam->max_buf_size > len? (beam->max_buf_size - len) : 0);
     }
     return APR_SIZE_MAX;
@@ -310,31 +189,10 @@ static apr_size_t calc_space_left(h2_bucket_beam *beam)
 
 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));
+    return H2_BLIST_EMPTY(&beam->buckets_to_send);
 }
 
-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)
+static apr_status_t wait_not_empty(h2_bucket_beam *beam, conn_rec *c, apr_read_type_e block)
 {
     apr_status_t rv = APR_SUCCESS;
     
@@ -345,21 +203,24 @@ static apr_status_t wait_not_empty(h2_bucket_beam *beam, apr_read_type_e block,
         else if (beam->closed) {
             rv = APR_EOF;
         }
-        else if (APR_BLOCK_READ != block || !lock) {
+        else if (APR_BLOCK_READ != block) {
             rv = APR_EAGAIN;
         }
         else if (beam->timeout > 0) {
-            rv = apr_thread_cond_timedwait(beam->change, lock, beam->timeout);
+            H2_BEAM_LOG(beam, c, APLOG_TRACE2, rv, "wait_not_empty, timeout", NULL);
+            rv = apr_thread_cond_timedwait(beam->change, beam->lock, beam->timeout);
         }
         else {
-            rv = apr_thread_cond_wait(beam->change, lock);
+            H2_BEAM_LOG(beam, c, APLOG_TRACE2, rv, "wait_not_empty, forever", NULL);
+            rv = apr_thread_cond_wait(beam->change, beam->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)
+static apr_status_t wait_not_full(h2_bucket_beam *beam, conn_rec *c,
+                                  apr_read_type_e block,
+                                  apr_size_t *pspace_left)
 {
     apr_status_t rv = APR_SUCCESS;
     apr_size_t left;
@@ -368,15 +229,17 @@ static apr_status_t wait_not_full(h2_bucket_beam *beam, apr_read_type_e block,
         if (beam->aborted) {
             rv = APR_ECONNABORTED;
         }
-        else if (block != APR_BLOCK_READ || !bl->mutex) {
+        else if (block != APR_BLOCK_READ) {
             rv = APR_EAGAIN;
         }
         else {
             if (beam->timeout > 0) {
-                rv = apr_thread_cond_timedwait(beam->change, bl->mutex, beam->timeout);
+                H2_BEAM_LOG(beam, c, APLOG_TRACE2, rv, "wait_not_full, timeout", NULL);
+                rv = apr_thread_cond_timedwait(beam->change, beam->lock, beam->timeout);
             }
             else {
-                rv = apr_thread_cond_wait(beam->change, bl->mutex);
+                H2_BEAM_LOG(beam, c, APLOG_TRACE2, rv, "wait_not_full, forever", NULL);
+                rv = apr_thread_cond_wait(beam->change, beam->lock);
             }
         }
     }
@@ -384,73 +247,6 @@ static apr_status_t wait_not_full(h2_bucket_beam *beam, apr_read_type_e block,
     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;
@@ -461,335 +257,203 @@ static void h2_blist_cleanup(h2_blist *bl)
     }
 }
 
-static apr_status_t beam_close(h2_bucket_beam *beam)
+static void beam_shutdown(h2_bucket_beam *beam, apr_shutdown_how_e how)
 {
-    if (!beam->closed) {
-        beam->closed = 1;
-        apr_thread_cond_broadcast(beam->change);
+    if (!beam->pool) {
+        /* pool being cleared already */
+        return;
     }
-    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;
+    /* shutdown both receiver and sender? */
+    if (how == APR_SHUTDOWN_READWRITE) {
+        beam->cons_io_cb = NULL;
+        beam->recv_cb = NULL;
+        beam->eagain_cb = NULL;
     }
-    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;
+    /* shutdown sender (or both)? */
+    if (how != APR_SHUTDOWN_READ) {
+        purge_consumed_buckets(beam);
+        h2_blist_cleanup(&beam->buckets_to_send);
     }
-    return 0;
 }
 
-static apr_status_t beam_recv_cleanup(void *data)
+static apr_status_t beam_cleanup(void *data)
 {
     h2_bucket_beam *beam = data;
-    /* receiver pool has gone away, clear references */
-    beam->recv_buffer = NULL;
-    beam->recv_pool = NULL;
+    beam_shutdown(beam, APR_SHUTDOWN_READWRITE);
+    purge_eor_buckets(beam);
+    beam->pool = NULL; /* the pool is clearing now */
     return APR_SUCCESS;
 }
 
-static apr_status_t beam_send_cleanup(void *data)
+apr_status_t h2_beam_destroy(h2_bucket_beam *beam, conn_rec *c)
 {
-    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;
+    if (beam->pool) {
+        H2_BEAM_LOG(beam, c, APLOG_TRACE2, 0, "destroy", NULL);
+        apr_pool_cleanup_run(beam->pool, beam, beam_cleanup);
     }
-    h2_blist_cleanup(&beam->purge_list);
-    h2_blist_cleanup(&beam->hold_list);
-    beam->send_pool = NULL;
+    H2_BEAM_LOG(beam, c, APLOG_TRACE2, 0, "destroyed", 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_status_t h2_beam_create(h2_bucket_beam **pbeam, conn_rec *from,
+                            apr_pool_t *pool, int id, const char *tag,
                             apr_size_t max_buf_size,
                             apr_interval_time_t timeout)
 {
     h2_bucket_beam *beam;
-    apr_status_t rv = APR_SUCCESS;
+    h2_conn_ctx_t *conn_ctx = h2_conn_ctx_get(from);
+    apr_status_t rv;
     
     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->from = from;
+    beam->id = id;
+    beam->name = apr_psprintf(pool, "%s-%d-%s",
+                              conn_ctx->id, id, tag);
+
+    H2_BLIST_INIT(&beam->buckets_to_send);
+    H2_BLIST_INIT(&beam->buckets_consumed);
+    H2_BLIST_INIT(&beam->buckets_eor);
     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;
-        }
-    }
+    if (APR_SUCCESS != rv) goto cleanup;
+    rv = apr_thread_cond_create(&beam->change, pool);
+    if (APR_SUCCESS != rv) goto cleanup;
+    apr_pool_pre_cleanup_register(pool, beam, beam_cleanup);
+
+cleanup:
+    H2_BEAM_LOG(beam, from, APLOG_TRACE2, rv, "created", NULL);
+    *pbeam = (APR_SUCCESS == rv)? beam : NULL;
     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_thread_mutex_lock(beam->lock);
+    beam->max_buf_size = buffer_size;
+    apr_thread_mutex_unlock(beam->lock);
 }
 
-apr_size_t h2_beam_buffer_size_get(h2_bucket_beam *beam)
+void h2_beam_set_copy_files(h2_bucket_beam * beam, int enabled)
 {
-    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;
+    apr_thread_mutex_lock(beam->lock);
+    beam->copy_files = enabled;
+    apr_thread_mutex_unlock(beam->lock);
 }
 
-void h2_beam_timeout_set(h2_bucket_beam *beam, apr_interval_time_t timeout)
+apr_size_t h2_beam_buffer_size_get(h2_bucket_beam *beam)
 {
-    h2_beam_lock bl;
+    apr_size_t buffer_size = 0;
     
-    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
-        beam->timeout = timeout;
-        leave_yellow(beam, &bl);
-    }
+    apr_thread_mutex_lock(beam->lock);
+    buffer_size = beam->max_buf_size;
+    apr_thread_mutex_unlock(beam->lock);
+    return buffer_size;
 }
 
 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);
-    }
+    apr_interval_time_t timeout;
+
+    apr_thread_mutex_lock(beam->lock);
+    timeout = beam->timeout;
+    apr_thread_mutex_unlock(beam->lock);
     return timeout;
 }
 
-void h2_beam_abort(h2_bucket_beam *beam)
+void h2_beam_timeout_set(h2_bucket_beam *beam, apr_interval_time_t timeout)
 {
-    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_thread_mutex_lock(beam->lock);
+    beam->timeout = timeout;
+    apr_thread_mutex_unlock(beam->lock);
 }
 
-apr_status_t h2_beam_close(h2_bucket_beam *beam)
+void h2_beam_abort(h2_bucket_beam *beam, conn_rec *c)
 {
-    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_thread_mutex_lock(beam->lock);
+    beam->aborted = 1;
+    if (c == beam->from) {
+        /* sender aborts */
+        if (beam->send_cb) {
+            beam->send_cb(beam->send_ctx, beam);
+        }
+        if (beam->was_empty_cb && buffer_is_empty(beam)) {
+            beam->was_empty_cb(beam->was_empty_ctx, beam);
+        }
+        /* no more consumption reporting to sender */
+        report_consumption(beam, 1);
+        beam->cons_ctx = NULL;
 
-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);
+        beam_shutdown(beam, APR_SHUTDOWN_WRITE);
     }
-    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);
+    else {
+        /* receiver aborts */
+        beam_shutdown(beam, APR_SHUTDOWN_READ);
     }
-    return status;
+    apr_thread_cond_broadcast(beam->change);
+    apr_thread_mutex_unlock(beam->lock);
 }
 
-static void move_to_hold(h2_bucket_beam *beam, 
-                         apr_bucket_brigade *sender_bb)
+void h2_beam_close(h2_bucket_beam *beam, conn_rec *c)
 {
-    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);
+    apr_thread_mutex_lock(beam->lock);
+    if (!beam->closed) {
+        /* should only be called from sender */
+        ap_assert(c == beam->from);
+        beam->closed = 1;
+        if (beam->send_cb) {
+            beam->send_cb(beam->send_ctx, beam);
+        }
+        if (beam->was_empty_cb && buffer_is_empty(beam)) {
+            beam->was_empty_cb(beam->was_empty_ctx, beam);
+        }
+        apr_thread_cond_broadcast(beam->change);
     }
+    apr_thread_mutex_unlock(beam->lock);
 }
 
-static apr_status_t append_bucket(h2_bucket_beam *beam, 
-                                  apr_bucket *b,
+static apr_status_t append_bucket(h2_bucket_beam *beam,
+                                  apr_bucket_brigade *bb,
                                   apr_read_type_e block,
                                   apr_size_t *pspace_left,
-                                  h2_beam_lock *pbl)
+                                  apr_off_t *pwritten)
 {
+    apr_bucket *b;
     const char *data;
     apr_size_t len;
-    apr_status_t status;
-    int can_beam = 0, check_len;
+    apr_status_t rv = APR_SUCCESS;
+    int can_beam = 0;
     
+    (void)block;
     if (beam->aborted) {
-        return APR_ECONNABORTED;
+        rv = APR_ECONNABORTED;
+        goto cleanup;
     }
-    
+
+    ap_assert(beam->pool);
+
+    b = APR_BRIGADE_FIRST(bb);
     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;
+        apr_bucket_setaside(b, beam->pool);
+        H2_BLIST_INSERT_TAIL(&beam->buckets_to_send, b);
+        goto cleanup;
+    }
+    /* non meta bucket */
+
+    /* in case of indeterminate length, we need to read the bucket,
+     * so that it transforms itself into something stable. */
+    if (b->length == ((apr_size_t)-1)) {
+        rv = apr_bucket_read(b, &data, &len, APR_BLOCK_READ);
+        if (rv != APR_SUCCESS) goto cleanup;
     }
-    else if (APR_BUCKET_IS_FILE(b)) {
+
+    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
@@ -806,478 +470,414 @@ static apr_status_t append_bucket(h2_bucket_beam *beam,
          * 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;
+        can_beam = !beam->copy_files && (bf->refcount.refcount == 1);
     }
-    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;
+    else if (bucket_is_mmap(b)) {
+        can_beam = !beam->copy_files;
     }
-    
-    if (check_len) {
-        if (b->length > *pspace_left) {
-            apr_bucket_split(b, *pspace_left);
-        }
-        *pspace_left -= b->length;
+
+    if (b->length == 0) {
+        apr_bucket_delete(b);
+        rv = APR_SUCCESS;
+        goto cleanup;
     }
 
-    /* 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);
+    if (!*pspace_left) {
+        rv = APR_EAGAIN;
+        goto cleanup;
     }
-    else if (APR_BUCKET_IS_HEAP(b)) {
-        /* For heap buckets read from a receiver thread is fine. The
+
+    /* bucket is accepted and added to beam->buckets_to_send */
+    if (APR_BUCKET_IS_HEAP(b)) {
+        /* For heap buckets, a read from a receiver thread is fine. The
          * data will be there and live until the bucket itself is
          * destroyed. */
-        status = APR_SUCCESS;
+        rv = apr_bucket_setaside(b, beam->pool);
+        if (rv != APR_SUCCESS) goto cleanup;
     }
-    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 (can_beam && (APR_BUCKET_IS_FILE(b) || bucket_is_mmap(b))) {
+        rv = apr_bucket_setaside(b, beam->pool);
+        if (rv != APR_SUCCESS) goto cleanup;
     }
-    else if (APR_BUCKET_IS_FILE(b) && can_beam) {
-        status = apr_bucket_setaside(b, beam->send_pool);
+    else {
+        /* we know of no special shortcut to transfer the bucket to
+         * another pool without copying. So we make it a heap bucket. */
+        apr_bucket *b2;
+
+        rv = apr_bucket_read(b, &data, &len, APR_BLOCK_READ);
+        if (rv != APR_SUCCESS) goto cleanup;
+        /* this allocates and copies data */
+        b2 = apr_bucket_heap_create(data, len, NULL, bb->bucket_alloc);
+        apr_bucket_delete(b);
+        b = b2;
+        APR_BRIGADE_INSERT_HEAD(bb, b);
     }
     
-    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);
-        }
+    APR_BUCKET_REMOVE(b);
+    H2_BLIST_INSERT_TAIL(&beam->buckets_to_send, b);
+    *pwritten += (apr_off_t)b->length;
+    if (b->length > *pspace_left) {
+        *pspace_left = 0;
     }
-    
-    if (status != APR_SUCCESS && status != APR_ENOTIMPL) {
-        return status;
+    else {
+        *pspace_left -= b->length;
     }
-    
-    APR_BUCKET_REMOVE(b);
-    H2_BLIST_INSERT_TAIL(&beam->send_list, b);
-    beam->sent_bytes += b->length;
-
-    return APR_SUCCESS;
-}
 
-void h2_beam_send_from(h2_bucket_beam *beam, apr_pool_t *p)
-{
-    h2_beam_lock bl;
-    /* Called from the sender thread to add buckets to the beam */
-    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
-        r_purge_sent(beam);
-        beam_set_send_pool(beam, p);
-        leave_yellow(beam, &bl);
-    }
+cleanup:
+    return rv;
 }
 
-apr_status_t h2_beam_send(h2_bucket_beam *beam, 
+apr_status_t h2_beam_send(h2_bucket_beam *beam, conn_rec *from,
                           apr_bucket_brigade *sender_bb, 
-                          apr_read_type_e block)
+                          apr_read_type_e block,
+                          apr_off_t *pwritten)
 {
-    apr_bucket *b;
     apr_status_t rv = APR_SUCCESS;
     apr_size_t space_left = 0;
-    h2_beam_lock bl;
+    int was_empty;
+
+    ap_assert(beam->pool);
 
     /* Called from the sender thread to add buckets to the beam */
-    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
-        ap_assert(beam->send_pool);
-        r_purge_sent(beam);
-        
+    apr_thread_mutex_lock(beam->lock);
+    ap_assert(beam->from == from);
+    ap_assert(sender_bb);
+    H2_BEAM_LOG(beam, from, APLOG_TRACE2, rv, "start send", sender_bb);
+    purge_consumed_buckets(beam);
+    *pwritten = 0;
+    was_empty = buffer_is_empty(beam);
+
+    space_left = calc_space_left(beam);
+    while (!APR_BRIGADE_EMPTY(sender_bb) && APR_SUCCESS == rv) {
+        rv = append_bucket(beam, sender_bb, block, &space_left, pwritten);
         if (beam->aborted) {
-            move_to_hold(beam, sender_bb);
-            rv = APR_ECONNABORTED;
-        }
-        else if (sender_bb) {
-            int force_report = !APR_BRIGADE_EMPTY(sender_bb);
-            
-            space_left = calc_space_left(beam);
-            while (!APR_BRIGADE_EMPTY(sender_bb) && APR_SUCCESS == rv) {
-                if (space_left <= 0) {
-                    report_prod_io(beam, force_report, &bl);
-                    r_purge_sent(beam);
-                    rv = wait_not_full(beam, block, &space_left, &bl);
-                    if (APR_SUCCESS != rv) {
-                        break;
-                    }
-                }
-                b = APR_BRIGADE_FIRST(sender_bb);
-                rv = append_bucket(beam, b, block, &space_left, &bl);
+            goto cleanup;
+        }
+        else if (APR_EAGAIN == rv) {
+            /* bucket was not added, as beam buffer has no space left.
+             * Trigger event callbacks, so receiver can know there is something
+             * to receive before we do a conditional wait. */
+            purge_consumed_buckets(beam);
+            if (beam->send_cb) {
+                beam->send_cb(beam->send_ctx, beam);
             }
-            
-            report_prod_io(beam, force_report, &bl);
-            apr_thread_cond_broadcast(beam->change);
+            if (was_empty && beam->was_empty_cb) {
+                beam->was_empty_cb(beam->was_empty_ctx, beam);
+            }
+            rv = wait_not_full(beam, from, block, &space_left);
+            if (APR_SUCCESS != rv) {
+                break;
+            }
+            was_empty = buffer_is_empty(beam);
         }
-        report_consumption(beam, &bl);
-        leave_yellow(beam, &bl);
     }
+
+cleanup:
+    if (beam->send_cb && !buffer_is_empty(beam)) {
+        beam->send_cb(beam->send_ctx, beam);
+    }
+    if (was_empty && beam->was_empty_cb && !buffer_is_empty(beam)) {
+        beam->was_empty_cb(beam->was_empty_ctx, beam);
+    }
+    apr_thread_cond_broadcast(beam->change);
+
+    report_consumption(beam, 1);
+    if (beam->aborted) {
+        rv = APR_ECONNABORTED;
+    }
+    H2_BEAM_LOG(beam, from, APLOG_TRACE2, rv, "end send", sender_bb);
+    if(rv != APR_SUCCESS && !APR_STATUS_IS_EAGAIN(rv) && sender_bb != NULL) {
+        apr_brigade_cleanup(sender_bb);
+    }
+    apr_thread_mutex_unlock(beam->lock);
     return rv;
 }
 
-apr_status_t h2_beam_receive(h2_bucket_beam *beam, 
+apr_status_t h2_beam_receive(h2_bucket_beam *beam,
+                             conn_rec *to,
                              apr_bucket_brigade *bb, 
                              apr_read_type_e block,
                              apr_off_t readbytes)
 {
-    h2_beam_lock bl;
     apr_bucket *bsender, *brecv, *ng;
     int transferred = 0;
-    apr_status_t status = APR_SUCCESS;
+    apr_status_t rv = APR_SUCCESS;
     apr_off_t remain;
-    int transferred_buckets = 0;
-    
-    /* Called from the receiver thread to take buckets from the beam */
-    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
-        if (readbytes <= 0) {
-            readbytes = APR_SIZE_MAX;
-        }
-        remain = readbytes;
-        
+    int consumed_buckets = 0;
+
+    apr_thread_mutex_lock(beam->lock);
+    H2_BEAM_LOG(beam, to, APLOG_TRACE2, 0, "start receive", bb);
+    if (readbytes <= 0) {
+        readbytes = (apr_off_t)APR_SIZE_MAX;
+    }
+    remain = readbytes;
+
 transfer:
-        if (beam->aborted) {
-            recv_buffer_cleanup(beam, &bl);
-            status = APR_ECONNABORTED;
-            goto leave;
-        }
+    if (beam->aborted) {
+        beam_shutdown(beam, APR_SHUTDOWN_READ);
+        rv = APR_ECONNABORTED;
+        goto leave;
+    }
 
-        /* transfer enough buckets from our receiver brigade, if we have one */
-        while (remain >= 0 
-               && beam->recv_buffer 
-               && !APR_BRIGADE_EMPTY(beam->recv_buffer)) {
-               
-            brecv = APR_BRIGADE_FIRST(beam->recv_buffer);
-            if (brecv->length > 0 && remain <= 0) {
-                break;
-            }            
-            APR_BUCKET_REMOVE(brecv);
-            APR_BRIGADE_INSERT_TAIL(bb, brecv);
-            remain -= brecv->length;
-            ++transferred;
+    ap_assert(beam->pool);
+
+    /* transfer from our sender brigade, transforming sender buckets to
+     * receiver ones until we have enough */
+    while (remain >= 0 && !H2_BLIST_EMPTY(&beam->buckets_to_send)) {
+
+        brecv = NULL;
+        bsender = H2_BLIST_FIRST(&beam->buckets_to_send);
+        if (bsender->length > 0 && remain <= 0) {
+            break;
         }
 
-        /* transfer from our sender brigade, transforming sender buckets to
-         * receiver ones until we have enough */
-        while (remain >= 0 && !H2_BLIST_EMPTY(&beam->send_list)) {
-               
-            brecv = NULL;
-            bsender = H2_BLIST_FIRST(&beam->send_list);            
-            if (bsender->length > 0 && remain <= 0) {
-                break;
+        if (APR_BUCKET_IS_METADATA(bsender)) {
+            /* we need a real copy into the receivers bucket_alloc */
+            if (APR_BUCKET_IS_EOS(bsender)) {
+                /* this closes the beam */
+                beam->closed = 1;
+                brecv = apr_bucket_eos_create(bb->bucket_alloc);
             }
-                        
-            if (APR_BUCKET_IS_METADATA(bsender)) {
-                if (APR_BUCKET_IS_EOS(bsender)) {
-                    brecv = apr_bucket_eos_create(bb->bucket_alloc);
-                    beam->close_sent = 1;
-                }
-                else if (APR_BUCKET_IS_FLUSH(bsender)) {
-                    brecv = apr_bucket_flush_create(bb->bucket_alloc);
-                }
-                else if (AP_BUCKET_IS_ERROR(bsender)) {
-                    ap_bucket_error *eb = (ap_bucket_error *)bsender;
-                    brecv = ap_bucket_error_create(eb->status, eb->data,
-                                                    bb->p, bb->bucket_alloc);
-                }
+            else if (APR_BUCKET_IS_FLUSH(bsender)) {
+                brecv = apr_bucket_flush_create(bb->bucket_alloc);
             }
-            else if (bsender->length == 0) {
-                APR_BUCKET_REMOVE(bsender);
-                H2_BLIST_INSERT_TAIL(&beam->hold_list, bsender);
-                continue;
+#if AP_HAS_RESPONSE_BUCKETS
+            else if (AP_BUCKET_IS_RESPONSE(bsender)) {
+                brecv = ap_bucket_response_clone(bsender, bb->p, bb->bucket_alloc);
             }
-            else if (APR_BUCKET_IS_FILE(bsender)) {
-                /* This is set aside into the target brigade pool so that 
-                 * any read operation messes with that pool and not 
-                 * the sender one. */
-                apr_bucket_file *f = (apr_bucket_file *)bsender->data;
-                apr_file_t *fd = f->fd;
-                int setaside = (f->readpool != bb->p);
-                
-                if (setaside) {
-                    status = apr_file_setaside(&fd, fd, bb->p);
-                    if (status != APR_SUCCESS) {
-                        goto leave;
-                    }
-                    ++beam->files_beamed;
-                }
-                ng = apr_brigade_insert_file(bb, fd, bsender->start, bsender->length, 
-                                             bb->p);
-#if APR_HAS_MMAP
-                /* disable mmap handling as this leads to segfaults when
-                 * the underlying file is changed while memory pointer has
-                 * been handed out. See also PR 59348 */
-                apr_bucket_file_enable_mmap(ng, 0);
-#endif
-                APR_BUCKET_REMOVE(bsender);
-                H2_BLIST_INSERT_TAIL(&beam->hold_list, bsender);
-
-                remain -= bsender->length;
-                ++transferred;
-                ++transferred_buckets;
-                continue;
+            else if (AP_BUCKET_IS_REQUEST(bsender)) {
+                brecv = ap_bucket_request_clone(bsender, bb->p, bb->bucket_alloc);
             }
-            else {
-                /* create a "receiver" standin bucket. we took care about the
-                 * underlying sender bucket and its data when we placed it into
-                 * the sender brigade.
-                 * the beam bucket will notify us on destruction that bsender is
-                 * no longer needed. */
-                brecv = h2_beam_bucket_create(beam, bsender, bb->bucket_alloc,
-                                               beam->buckets_sent++);
+            else if (AP_BUCKET_IS_HEADERS(bsender)) {
+                brecv = ap_bucket_headers_clone(bsender, bb->p, bb->bucket_alloc);
             }
-            
-            /* Place the sender bucket into our hold, to be destroyed when no
-             * receiver bucket references it any more. */
-            APR_BUCKET_REMOVE(bsender);
-            H2_BLIST_INSERT_TAIL(&beam->hold_list, bsender);
-            
-            beam->received_bytes += bsender->length;
-            ++transferred_buckets;
-            
-            if (brecv) {
-                APR_BRIGADE_INSERT_TAIL(bb, brecv);
-                remain -= brecv->length;
-                ++transferred;
+#else
+            else if (H2_BUCKET_IS_HEADERS(bsender)) {
+                brecv = h2_bucket_headers_clone(bsender, bb->p, bb->bucket_alloc);
             }
-            else {
-                /* let outside hook determine how bucket is beamed */
-                leave_yellow(beam, &bl);
-                brecv = h2_beam_bucket(beam, bb, bsender);
-                enter_yellow(beam, &bl);
-                
-                while (brecv && brecv != APR_BRIGADE_SENTINEL(bb)) {
-                    ++transferred;
-                    remain -= brecv->length;
-                    brecv = APR_BUCKET_NEXT(brecv);
-                }
+#endif /* AP_HAS_RESPONSE_BUCKETS */
+            else if (AP_BUCKET_IS_ERROR(bsender)) {
+                ap_bucket_error *eb = bsender->data;
+                brecv = ap_bucket_error_create(eb->status, eb->data,
+                                               bb->p, bb->bucket_alloc);
             }
         }
-
-        if (remain < 0) {
-            /* too much, put some back into out recv_buffer */
-            remain = readbytes;
-            for (brecv = APR_BRIGADE_FIRST(bb);
-                 brecv != APR_BRIGADE_SENTINEL(bb);
-                 brecv = APR_BUCKET_NEXT(brecv)) {
-                remain -= (beam->tx_mem_limits? bucket_mem_used(brecv) 
-                           : brecv->length);
-                if (remain < 0) {
-                    apr_bucket_split(brecv, brecv->length+remain);
-                    beam->recv_buffer = apr_brigade_split_ex(bb, 
-                                                             APR_BUCKET_NEXT(brecv), 
-                                                             beam->recv_buffer);
-                    break;
-                }
-            }
+        else if (bsender->length == 0) {
+            /* nop */
         }
-
-        if (beam->closed && buffer_is_empty(beam)) {
-            /* beam is closed and we have nothing more to receive */ 
-            if (!beam->close_sent) {
-                apr_bucket *b = apr_bucket_eos_create(bb->bucket_alloc);
-                APR_BRIGADE_INSERT_TAIL(bb, b);
-                beam->close_sent = 1;
-                ++transferred;
-                status = APR_SUCCESS;
-            }
+#if APR_HAS_MMAP
+        else if (APR_BUCKET_IS_MMAP(bsender)) {
+            apr_bucket_mmap *bmmap = bsender->data;
+            apr_mmap_t *mmap;
+            rv = apr_mmap_dup(&mmap, bmmap->mmap, bb->p);
+            if (rv != APR_SUCCESS) goto leave;
+            brecv = apr_bucket_mmap_create(mmap, bsender->start, bsender->length, bb->bucket_alloc);
         }
-        
-        if (transferred_buckets > 0) {
-           if (beam->cons_ev_cb) { 
-               beam->cons_ev_cb(beam->cons_ctx, beam);
+#endif
+        else if (APR_BUCKET_IS_FILE(bsender)) {
+            /* This is setaside into the target brigade pool so that
+             * any read operation messes with that pool and not
+             * the sender one. */
+            apr_bucket_file *f = (apr_bucket_file *)bsender->data;
+            apr_file_t *fd = f->fd;
+            int setaside = (f->readpool != bb->p);
+
+            if (setaside) {
+                rv = apr_file_setaside(&fd, fd, bb->p);
+                if (rv != APR_SUCCESS) goto leave;
             }
-        }
-        
-        if (transferred) {
-            apr_thread_cond_broadcast(beam->change);
-            status = APR_SUCCESS;
+            ng = apr_brigade_insert_file(bb, fd, bsender->start, (apr_off_t)bsender->length,
+                                         bb->p);
+#if APR_HAS_MMAP
+            /* disable mmap handling as this leads to segfaults when
+             * the underlying file is changed while memory pointer has
+             * been handed out. See also PR 59348 */
+            apr_bucket_file_enable_mmap(ng, 0);
+#endif
+            remain -= bsender->length;
+            ++transferred;
         }
         else {
-            status = wait_not_empty(beam, block, bl.mutex);
-            if (status != APR_SUCCESS) {
-                goto leave;
-            }
-            goto transfer;
+            const char *data;
+            apr_size_t dlen;
+            /* we did that when the bucket was added, so this should
+             * give us the same data as before without changing the bucket
+             * or anything (pool) connected to it. */
+            rv = apr_bucket_read(bsender, &data, &dlen, APR_BLOCK_READ);
+            if (rv != APR_SUCCESS) goto leave;
+            rv = apr_brigade_write(bb, NULL, NULL, data, dlen);
+            if (rv != APR_SUCCESS) goto leave;
+
+            remain -= dlen;
+            ++transferred;
+        }
+
+        if (brecv) {
+            /* we have a proxy that we can give the receiver */
+            APR_BRIGADE_INSERT_TAIL(bb, brecv);
+            remain -= brecv->length;
+            ++transferred;
+        }
+        APR_BUCKET_REMOVE(bsender);
+        H2_BLIST_INSERT_TAIL(&beam->buckets_consumed, bsender);
+        beam->recv_bytes += bsender->length;
+        ++consumed_buckets;
+    }
+
+    if (beam->recv_cb && consumed_buckets > 0) {
+        beam->recv_cb(beam->recv_ctx, beam);
+    }
+
+    if (transferred) {
+        apr_thread_cond_broadcast(beam->change);
+        rv = APR_SUCCESS;
+    }
+    else if (beam->aborted) {
+        rv = APR_ECONNABORTED;
+    }
+    else if (beam->closed) {
+        rv = APR_EOF;
+    }
+    else {
+        rv = wait_not_empty(beam, to, block);
+        if (rv != APR_SUCCESS) {
+            goto leave;
         }
-leave:        
-        leave_yellow(beam, &bl);
+        goto transfer;
+    }
+
+leave:
+    H2_BEAM_LOG(beam, to, APLOG_TRACE2, rv, "end receive", bb);
+    if (rv == APR_EAGAIN && beam->eagain_cb) {
+        beam->eagain_cb(beam->eagain_ctx, beam);
     }
-    return status;
+    apr_thread_mutex_unlock(beam->lock);
+    return rv;
 }
 
 void h2_beam_on_consumed(h2_bucket_beam *beam, 
-                         h2_beam_ev_callback *ev_cb,
                          h2_beam_io_callback *io_cb, void *ctx)
 {
-    h2_beam_lock bl;
-    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
-        beam->cons_ev_cb = ev_cb;
-        beam->cons_io_cb = io_cb;
-        beam->cons_ctx = ctx;
-        leave_yellow(beam, &bl);
-    }
+    apr_thread_mutex_lock(beam->lock);
+    beam->cons_io_cb = io_cb;
+    beam->cons_ctx = ctx;
+    apr_thread_mutex_unlock(beam->lock);
 }
 
-void h2_beam_on_produced(h2_bucket_beam *beam, 
-                         h2_beam_io_callback *io_cb, void *ctx)
+void h2_beam_on_received(h2_bucket_beam *beam,
+                         h2_beam_ev_callback *recv_cb, void *ctx)
 {
-    h2_beam_lock bl;
-    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
-        beam->prod_io_cb = io_cb;
-        beam->prod_ctx = ctx;
-        leave_yellow(beam, &bl);
-    }
+    apr_thread_mutex_lock(beam->lock);
+    beam->recv_cb = recv_cb;
+    beam->recv_ctx = ctx;
+    apr_thread_mutex_unlock(beam->lock);
 }
 
-void h2_beam_on_file_beam(h2_bucket_beam *beam, 
-                          h2_beam_can_beam_callback *cb, void *ctx)
+void h2_beam_on_eagain(h2_bucket_beam *beam,
+                       h2_beam_ev_callback *eagain_cb, void *ctx)
 {
-    h2_beam_lock bl;
-    
-    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
-        beam->can_beam_fn = cb;
-        beam->can_beam_ctx = ctx;
-        leave_yellow(beam, &bl);
-    }
+    apr_thread_mutex_lock(beam->lock);
+    beam->eagain_cb = eagain_cb;
+    beam->eagain_ctx = ctx;
+    apr_thread_mutex_unlock(beam->lock);
 }
 
+void h2_beam_on_send(h2_bucket_beam *beam,
+                     h2_beam_ev_callback *send_cb, void *ctx)
+{
+    apr_thread_mutex_lock(beam->lock);
+    beam->send_cb = send_cb;
+    beam->send_ctx = ctx;
+    apr_thread_mutex_unlock(beam->lock);
+}
 
-apr_off_t h2_beam_get_buffered(h2_bucket_beam *beam)
+void h2_beam_on_was_empty(h2_bucket_beam *beam,
+                          h2_beam_ev_callback *was_empty_cb, void *ctx)
 {
-    apr_bucket *b;
-    apr_off_t l = 0;
-    h2_beam_lock bl;
-    
-    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {
-        for (b = H2_BLIST_FIRST(&beam->send_list); 
-            b != H2_BLIST_SENTINEL(&beam->send_list);
-            b = APR_BUCKET_NEXT(b)) {
-            /* should all have determinate length */
-            l += b->length;
-        }
-        leave_yellow(beam, &bl);
-    }
-    return l;
+    apr_thread_mutex_lock(beam->lock);
+    beam->was_empty_cb = was_empty_cb;
+    beam->was_empty_ctx = ctx;
+    apr_thread_mutex_unlock(beam->lock);
 }
 
-apr_off_t h2_beam_get_mem_used(h2_bucket_beam *beam)
+
+static apr_off_t get_buffered_data_len(h2_bucket_beam *beam)
 {
     apr_bucket *b;
     apr_off_t l = 0;
-    h2_beam_lock bl;
-    
-    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {
-        for (b = H2_BLIST_FIRST(&beam->send_list); 
-            b != H2_BLIST_SENTINEL(&beam->send_list);
-            b = APR_BUCKET_NEXT(b)) {
-            l += bucket_mem_used(b);
-        }
-        leave_yellow(beam, &bl);
+
+    for (b = H2_BLIST_FIRST(&beam->buckets_to_send);
+        b != H2_BLIST_SENTINEL(&beam->buckets_to_send);
+        b = APR_BUCKET_NEXT(b)) {
+        /* should all have determinate length */
+        l += b->length;
     }
     return l;
 }
 
-int h2_beam_empty(h2_bucket_beam *beam)
+apr_off_t h2_beam_get_buffered(h2_bucket_beam *beam)
 {
-    int empty = 1;
-    h2_beam_lock bl;
-    
-    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {
-        empty = (H2_BLIST_EMPTY(&beam->send_list) 
-                 && (!beam->recv_buffer || APR_BRIGADE_EMPTY(beam->recv_buffer)));
-        leave_yellow(beam, &bl);
-    }
-    return empty;
-}
+    apr_off_t l = 0;
 
-int h2_beam_holds_proxies(h2_bucket_beam *beam)
-{
-    int has_proxies = 1;
-    h2_beam_lock bl;
-    
-    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {
-        has_proxies = !H2_BPROXY_LIST_EMPTY(&beam->proxies);
-        leave_yellow(beam, &bl);
-    }
-    return has_proxies;
+    apr_thread_mutex_lock(beam->lock);
+    l = get_buffered_data_len(beam);
+    apr_thread_mutex_unlock(beam->lock);
+    return l;
 }
 
-int h2_beam_was_received(h2_bucket_beam *beam)
+apr_off_t h2_beam_get_mem_used(h2_bucket_beam *beam)
 {
-    int happend = 0;
-    h2_beam_lock bl;
-    
-    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {
-        happend = (beam->received_bytes > 0);
-        leave_yellow(beam, &bl);
-    }
-    return happend;
-}
+    apr_bucket *b;
+    apr_off_t l = 0;
 
-apr_size_t h2_beam_get_files_beamed(h2_bucket_beam *beam)
-{
-    apr_size_t n = 0;
-    h2_beam_lock bl;
-    
-    if (beam && enter_yellow(beam, &bl) == APR_SUCCESS) {
-        n = beam->files_beamed;
-        leave_yellow(beam, &bl);
+    apr_thread_mutex_lock(beam->lock);
+    for (b = H2_BLIST_FIRST(&beam->buckets_to_send);
+        b != H2_BLIST_SENTINEL(&beam->buckets_to_send);
+        b = APR_BUCKET_NEXT(b)) {
+        l += bucket_mem_used(b);
     }
-    return n;
+    apr_thread_mutex_unlock(beam->lock);
+    return l;
 }
 
-int h2_beam_no_files(void *ctx, h2_bucket_beam *beam, apr_file_t *file)
+int h2_beam_empty(h2_bucket_beam *beam)
 {
-    return 0;
+    int empty = 1;
+
+    apr_thread_mutex_lock(beam->lock);
+    empty = buffer_is_empty(beam);
+    apr_thread_mutex_unlock(beam->lock);
+    return empty;
 }
 
 int h2_beam_report_consumption(h2_bucket_beam *beam)
 {
-    h2_beam_lock bl;
     int rv = 0;
-    if (enter_yellow(beam, &bl) == APR_SUCCESS) {
-        rv = report_consumption(beam, &bl);
-        leave_yellow(beam, &bl);
-    }
+
+    apr_thread_mutex_lock(beam->lock);
+    rv = report_consumption(beam, 1);
+    apr_thread_mutex_unlock(beam->lock);
     return rv;
 }
 
-void h2_beam_log(h2_bucket_beam *beam, conn_rec *c, int level, const char *msg)
+int h2_beam_is_complete(h2_bucket_beam *beam)
 {
-    if (beam && APLOG_C_IS_LEVEL(c,level)) {
-        ap_log_cerror(APLOG_MARK, level, 0, c, 
-                      "beam(%ld-%d,%s,closed=%d,aborted=%d,empty=%d,buf=%ld): %s", 
-                      (c->master? c->master->id : c->id), beam->id, beam->tag, 
-                      beam->closed, beam->aborted, h2_beam_empty(beam), 
-                      (long)h2_beam_get_buffered(beam), msg);
+    int rv = 0;
+
+    apr_thread_mutex_lock(beam->lock);
+    if (beam->closed)
+        rv = 1;
+    else {
+        apr_bucket *b;
+        for (b = H2_BLIST_FIRST(&beam->buckets_to_send);
+             b != H2_BLIST_SENTINEL(&beam->buckets_to_send);
+             b = APR_BUCKET_NEXT(b)) {
+            if (APR_BUCKET_IS_EOS(b)) {
+                rv = 1;
+                break;
+            }
+        }
     }
+    apr_thread_mutex_unlock(beam->lock);
+    return rv;
 }
-
-