Adding upstream version 2.4.59.upstream/2.4.59upstream

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

1 files changed, 499 insertions, 256 deletions

diff --git a/modules/http2/h2_workers.c b/modules/http2/h2_workers.c
index 699f533..e7e2039 100644
--- a/modules/http2/h2_workers.c
+++ b/modules/http2/h2_workers.c
@@ -15,285 +15,440 @@
  */
 
 #include <assert.h>
-#include <apr_atomic.h>
+#include <apr_ring.h>
 #include <apr_thread_mutex.h>
 #include <apr_thread_cond.h>
 
 #include <mpm_common.h>
 #include <httpd.h>
+#include <http_connection.h>
 #include <http_core.h>
 #include <http_log.h>
+#include <http_protocol.h>
 
 #include "h2.h"
 #include "h2_private.h"
 #include "h2_mplx.h"
-#include "h2_task.h"
+#include "h2_c2.h"
 #include "h2_workers.h"
 #include "h2_util.h"
 
+typedef enum {
+    PROD_IDLE,
+    PROD_ACTIVE,
+    PROD_JOINED,
+} prod_state_t;
+
+struct ap_conn_producer_t {
+    APR_RING_ENTRY(ap_conn_producer_t) link;
+    const char *name;
+    void *baton;
+    ap_conn_producer_next *fn_next;
+    ap_conn_producer_done *fn_done;
+    ap_conn_producer_shutdown *fn_shutdown;
+    volatile prod_state_t state;
+    volatile int conns_active;
+};
+
+
+typedef enum {
+    H2_SLOT_FREE,
+    H2_SLOT_RUN,
+    H2_SLOT_ZOMBIE,
+} h2_slot_state_t;
+
 typedef struct h2_slot h2_slot;
 struct h2_slot {
-    int id;
-    h2_slot *next;
+    APR_RING_ENTRY(h2_slot) link;
+    apr_uint32_t id;
+    apr_pool_t *pool;
+    h2_slot_state_t state;
+    volatile int should_shutdown;
+    volatile int is_idle;
     h2_workers *workers;
-    int aborted;
-    int sticks;
-    h2_task *task;
+    ap_conn_producer_t *prod;
     apr_thread_t *thread;
-    apr_thread_mutex_t *lock;
-    apr_thread_cond_t *not_idle;
+    struct apr_thread_cond_t *more_work;
+    int activations;
 };
 
-static h2_slot *pop_slot(h2_slot **phead) 
-{
-    /* Atomically pop a slot from the list */
-    for (;;) {
-        h2_slot *first = *phead;
-        if (first == NULL) {
-            return NULL;
-        }
-        if (apr_atomic_casptr((void*)phead, first->next, first) == first) {
-            first->next = NULL;
-            return first;
-        }
-    }
-}
+struct h2_workers {
+    server_rec *s;
+    apr_pool_t *pool;
+
+    apr_uint32_t max_slots;
+    apr_uint32_t min_active;
+    volatile apr_time_t idle_limit;
+    volatile int aborted;
+    volatile int shutdown;
+    int dynamic;
+
+    volatile apr_uint32_t active_slots;
+    volatile apr_uint32_t idle_slots;
+
+    apr_threadattr_t *thread_attr;
+    h2_slot *slots;
+
+    APR_RING_HEAD(h2_slots_free, h2_slot) free;
+    APR_RING_HEAD(h2_slots_idle, h2_slot) idle;
+    APR_RING_HEAD(h2_slots_busy, h2_slot) busy;
+    APR_RING_HEAD(h2_slots_zombie, h2_slot) zombie;
+
+    APR_RING_HEAD(ap_conn_producer_active, ap_conn_producer_t) prod_active;
+    APR_RING_HEAD(ap_conn_producer_idle, ap_conn_producer_t) prod_idle;
+
+    struct apr_thread_mutex_t *lock;
+    struct apr_thread_cond_t *prod_done;
+    struct apr_thread_cond_t *all_done;
+};
 
-static void push_slot(h2_slot **phead, h2_slot *slot)
-{
-    /* Atomically push a slot to the list */
-    ap_assert(!slot->next);
-    for (;;) {
-        h2_slot *next = slot->next = *phead;
-        if (apr_atomic_casptr((void*)phead, slot, next) == next) {
-            return;
-        }
-    }
-}
 
 static void* APR_THREAD_FUNC slot_run(apr_thread_t *thread, void *wctx);
 
-static apr_status_t activate_slot(h2_workers *workers, h2_slot *slot) 
+static apr_status_t activate_slot(h2_workers *workers)
 {
-    apr_status_t status;
-    
-    slot->workers = workers;
-    slot->aborted = 0;
-    slot->task = NULL;
-
-    if (!slot->lock) {
-        status = apr_thread_mutex_create(&slot->lock,
-                                         APR_THREAD_MUTEX_DEFAULT,
-                                         workers->pool);
-        if (status != APR_SUCCESS) {
-            push_slot(&workers->free, slot);
-            return status;
-        }
-    }
+    h2_slot *slot;
+    apr_pool_t *pool;
+    apr_status_t rv;
 
-    if (!slot->not_idle) {
-        status = apr_thread_cond_create(&slot->not_idle, workers->pool);
-        if (status != APR_SUCCESS) {
-            push_slot(&workers->free, slot);
-            return status;
-        }
+    if (APR_RING_EMPTY(&workers->free, h2_slot, link)) {
+        return APR_EAGAIN;
     }
-    
-    ap_log_error(APLOG_MARK, APLOG_TRACE2, 0, workers->s,
-                 "h2_workers: new thread for slot %d", slot->id); 
-    /* thread will either immediately start work or add itself
-     * to the idle queue */
-    apr_thread_create(&slot->thread, workers->thread_attr, slot_run, slot, 
-                      workers->pool);
-    if (!slot->thread) {
-        push_slot(&workers->free, slot);
-        return APR_ENOMEM;
-    }
-    
-    apr_atomic_inc32(&workers->worker_count);
-    return APR_SUCCESS;
-}
+    slot = APR_RING_FIRST(&workers->free);
+    ap_assert(slot->state == H2_SLOT_FREE);
+    APR_RING_REMOVE(slot, link);
 
-static apr_status_t add_worker(h2_workers *workers)
-{
-    h2_slot *slot = pop_slot(&workers->free);
-    if (slot) {
-        return activate_slot(workers, slot);
-    }
-    return APR_EAGAIN;
-}
+    ap_log_error(APLOG_MARK, APLOG_TRACE3, 0, workers->s,
+                 "h2_workers: activate slot %d", slot->id);
 
-static void wake_idle_worker(h2_workers *workers) 
-{
-    h2_slot *slot = pop_slot(&workers->idle);
-    if (slot) {
-        apr_thread_mutex_lock(slot->lock);
-        apr_thread_cond_signal(slot->not_idle);
-        apr_thread_mutex_unlock(slot->lock);
-    }
-    else if (workers->dynamic) {
-        add_worker(workers);
+    slot->state = H2_SLOT_RUN;
+    slot->should_shutdown = 0;
+    slot->is_idle = 0;
+    slot->pool = NULL;
+    ++workers->active_slots;
+    rv = apr_pool_create(&pool, workers->pool);
+    if (APR_SUCCESS != rv) goto cleanup;
+    apr_pool_tag(pool, "h2_worker_slot");
+    slot->pool = pool;
+
+    rv = ap_thread_create(&slot->thread, workers->thread_attr,
+                          slot_run, slot, slot->pool);
+
+cleanup:
+    if (rv != APR_SUCCESS) {
+        AP_DEBUG_ASSERT(0);
+        slot->state = H2_SLOT_FREE;
+        if (slot->pool) {
+            apr_pool_destroy(slot->pool);
+            slot->pool = NULL;
+        }
+        APR_RING_INSERT_TAIL(&workers->free, slot, h2_slot, link);
+        --workers->active_slots;
     }
+    return rv;
 }
 
-static void cleanup_zombies(h2_workers *workers)
+static void join_zombies(h2_workers *workers)
 {
     h2_slot *slot;
-    while ((slot = pop_slot(&workers->zombies))) {
-        if (slot->thread) {
-            apr_status_t status;
-            apr_thread_join(&status, slot->thread);
-            slot->thread = NULL;
+    apr_status_t status;
+
+    while (!APR_RING_EMPTY(&workers->zombie, h2_slot, link)) {
+        slot = APR_RING_FIRST(&workers->zombie);
+        APR_RING_REMOVE(slot, link);
+        ap_assert(slot->state == H2_SLOT_ZOMBIE);
+        ap_assert(slot->thread != NULL);
+
+        apr_thread_mutex_unlock(workers->lock);
+        apr_thread_join(&status, slot->thread);
+        apr_thread_mutex_lock(workers->lock);
+
+        slot->thread = NULL;
+        slot->state = H2_SLOT_FREE;
+        if (slot->pool) {
+            apr_pool_destroy(slot->pool);
+            slot->pool = NULL;
         }
-        apr_atomic_dec32(&workers->worker_count);
-        slot->next = NULL;
-        push_slot(&workers->free, slot);
+        APR_RING_INSERT_TAIL(&workers->free, slot, h2_slot, link);
     }
 }
 
-static apr_status_t slot_pull_task(h2_slot *slot, h2_mplx *m)
+static void wake_idle_worker(h2_workers *workers, ap_conn_producer_t *prod)
 {
-    apr_status_t rv;
-    
-    rv = h2_mplx_pop_task(m, &slot->task);
-    if (slot->task) {
-        /* Ok, we got something to give back to the worker for execution. 
-         * If we still have idle workers, we let the worker be sticky, 
-         * e.g. making it poll the task's h2_mplx instance for more work 
-         * before asking back here. */
-        slot->sticks = slot->workers->max_workers;
-        return rv;            
+    if (!APR_RING_EMPTY(&workers->idle, h2_slot, link)) {
+        h2_slot *slot;
+        for (slot = APR_RING_FIRST(&workers->idle);
+             slot != APR_RING_SENTINEL(&workers->idle, h2_slot, link);
+             slot = APR_RING_NEXT(slot, link)) {
+             if (slot->is_idle && !slot->should_shutdown) {
+                apr_thread_cond_signal(slot->more_work);
+                slot->is_idle = 0;
+                return;
+             }
+        }
+    }
+    if (workers->dynamic && !workers->shutdown
+        && (workers->active_slots < workers->max_slots)) {
+        activate_slot(workers);
     }
-    slot->sticks = 0;
-    return APR_EOF;
-}
-
-static h2_fifo_op_t mplx_peek(void *head, void *ctx)
-{
-    h2_mplx *m = head;
-    h2_slot *slot = ctx;
-    
-    if (slot_pull_task(slot, m) == APR_EAGAIN) {
-        wake_idle_worker(slot->workers);
-        return H2_FIFO_OP_REPUSH;
-    } 
-    return H2_FIFO_OP_PULL;
 }
 
 /**
- * Get the next task for the given worker. Will block until a task arrives
- * or the max_wait timer expires and more than min workers exist.
+ * Get the next connection to work on.
  */
-static apr_status_t get_next(h2_slot *slot)
+static conn_rec *get_next(h2_slot *slot)
 {
     h2_workers *workers = slot->workers;
-    apr_status_t status;
-    
-    slot->task = NULL;
-    while (!slot->aborted) {
-        if (!slot->task) {
-            status = h2_fifo_try_peek(workers->mplxs, mplx_peek, slot);
-            if (status == APR_EOF) {
-                return status;
-            }
+    conn_rec *c = NULL;
+    ap_conn_producer_t *prod;
+    int has_more;
+
+    slot->prod = NULL;
+    if (!APR_RING_EMPTY(&workers->prod_active, ap_conn_producer_t, link)) {
+        slot->prod = prod = APR_RING_FIRST(&workers->prod_active);
+        APR_RING_REMOVE(prod, link);
+        AP_DEBUG_ASSERT(PROD_ACTIVE == prod->state);
+
+        c = prod->fn_next(prod->baton, &has_more);
+        if (c && has_more) {
+            APR_RING_INSERT_TAIL(&workers->prod_active, prod, ap_conn_producer_t, link);
+            wake_idle_worker(workers, slot->prod);
         }
-        
-        if (slot->task) {
-            return APR_SUCCESS;
+        else {
+            prod->state = PROD_IDLE;
+            APR_RING_INSERT_TAIL(&workers->prod_idle, prod, ap_conn_producer_t, link);
+        }
+        if (c) {
+            ++prod->conns_active;
         }
-        
-        cleanup_zombies(workers);
-
-        apr_thread_mutex_lock(slot->lock);
-        push_slot(&workers->idle, slot);
-        apr_thread_cond_wait(slot->not_idle, slot->lock);
-        apr_thread_mutex_unlock(slot->lock);
     }
-    return APR_EOF;
-}
 
-static void slot_done(h2_slot *slot)
-{
-    push_slot(&(slot->workers->zombies), slot);
+    return c;
 }
 
-
 static void* APR_THREAD_FUNC slot_run(apr_thread_t *thread, void *wctx)
 {
     h2_slot *slot = wctx;
-    
-    while (!slot->aborted) {
-
-        /* Get a h2_task from the mplxs queue. */
-        get_next(slot);
-        while (slot->task) {
-        
-            h2_task_do(slot->task, thread, slot->id);
-            
-            /* Report the task as done. If stickyness is left, offer the
-             * mplx the opportunity to give us back a new task right away.
-             */
-            if (!slot->aborted && (--slot->sticks > 0)) {
-                h2_mplx_task_done(slot->task->mplx, slot->task, &slot->task);
-            }
-            else {
-                h2_mplx_task_done(slot->task->mplx, slot->task, NULL);
-                slot->task = NULL;
+    h2_workers *workers = slot->workers;
+    conn_rec *c;
+    apr_status_t rv;
+
+    apr_thread_mutex_lock(workers->lock);
+    slot->state = H2_SLOT_RUN;
+    ++slot->activations;
+    APR_RING_ELEM_INIT(slot, link);
+    for(;;) {
+        if (APR_RING_NEXT(slot, link) != slot) {
+            /* slot is part of the idle ring from the last loop */
+            APR_RING_REMOVE(slot, link);
+            --workers->idle_slots;
+        }
+        slot->is_idle = 0;
+
+        if (!workers->aborted && !slot->should_shutdown) {
+            APR_RING_INSERT_TAIL(&workers->busy, slot, h2_slot, link);
+            do {
+                c = get_next(slot);
+                if (!c) {
+                    break;
+                }
+                apr_thread_mutex_unlock(workers->lock);
+                /* See the discussion at <https://github.com/icing/mod_h2/issues/195>
+                 *
+                 * Each conn_rec->id is supposed to be unique at a point in time. Since
+                 * some modules (and maybe external code) uses this id as an identifier
+                 * for the request_rec they handle, it needs to be unique for secondary
+                 * connections also.
+                 *
+                 * The MPM module assigns the connection ids and mod_unique_id is using
+                 * that one to generate identifier for requests. While the implementation
+                 * works for HTTP/1.x, the parallel execution of several requests per
+                 * connection will generate duplicate identifiers on load.
+                 *
+                 * The original implementation for secondary connection identifiers used
+                 * to shift the master connection id up and assign the stream id to the
+                 * lower bits. This was cramped on 32 bit systems, but on 64bit there was
+                 * enough space.
+                 *
+                 * As issue 195 showed, mod_unique_id only uses the lower 32 bit of the
+                 * connection id, even on 64bit systems. Therefore collisions in request ids.
+                 *
+                 * The way master connection ids are generated, there is some space "at the
+                 * top" of the lower 32 bits on allmost all systems. If you have a setup
+                 * with 64k threads per child and 255 child processes, you live on the edge.
+                 *
+                 * The new implementation shifts 8 bits and XORs in the worker
+                 * id. This will experience collisions with > 256 h2 workers and heavy
+                 * load still. There seems to be no way to solve this in all possible
+                 * configurations by mod_h2 alone.
+                 */
+                if (c->master) {
+                    c->id = (c->master->id << 8)^slot->id;
+                }
+                c->current_thread = thread;
+                AP_DEBUG_ASSERT(slot->prod);
+
+#if AP_HAS_RESPONSE_BUCKETS
+                ap_process_connection(c, ap_get_conn_socket(c));
+#else
+                h2_c2_process(c, thread, slot->id);
+#endif
+                slot->prod->fn_done(slot->prod->baton, c);
+
+                apr_thread_mutex_lock(workers->lock);
+                if (--slot->prod->conns_active <= 0) {
+                    apr_thread_cond_broadcast(workers->prod_done);
+                }
+                if (slot->prod->state == PROD_IDLE) {
+                    APR_RING_REMOVE(slot->prod, link);
+                    slot->prod->state = PROD_ACTIVE;
+                    APR_RING_INSERT_TAIL(&workers->prod_active, slot->prod, ap_conn_producer_t, link);
+                }
+
+            } while (!workers->aborted && !slot->should_shutdown);
+            APR_RING_REMOVE(slot, link); /* no longer busy */
+        }
+
+        if (workers->aborted || slot->should_shutdown) {
+            break;
+        }
+
+        join_zombies(workers);
+
+        /* we are idle */
+        APR_RING_INSERT_TAIL(&workers->idle, slot, h2_slot, link);
+        ++workers->idle_slots;
+        slot->is_idle = 1;
+        if (slot->id >= workers->min_active && workers->idle_limit > 0) {
+            rv = apr_thread_cond_timedwait(slot->more_work, workers->lock,
+                                           workers->idle_limit);
+            if (APR_TIMEUP == rv) {
+                APR_RING_REMOVE(slot, link);
+                --workers->idle_slots;
+                ap_log_error(APLOG_MARK, APLOG_TRACE1, 0, workers->s,
+                             "h2_workers: idle timeout slot %d in state %d (%d activations)",
+                             slot->id, slot->state, slot->activations);
+                break;
             }
         }
+        else {
+            apr_thread_cond_wait(slot->more_work, workers->lock);
+        }
+    }
+
+    ap_log_error(APLOG_MARK, APLOG_TRACE3, 0, workers->s,
+                 "h2_workers: terminate slot %d in state %d (%d activations)",
+                 slot->id, slot->state, slot->activations);
+    slot->is_idle = 0;
+    slot->state = H2_SLOT_ZOMBIE;
+    slot->should_shutdown = 0;
+    APR_RING_INSERT_TAIL(&workers->zombie, slot, h2_slot, link);
+    --workers->active_slots;
+    if (workers->active_slots <= 0) {
+        apr_thread_cond_broadcast(workers->all_done);
     }
+    apr_thread_mutex_unlock(workers->lock);
 
-    slot_done(slot);
+    apr_thread_exit(thread, APR_SUCCESS);
     return NULL;
 }
 
+static void wake_all_idles(h2_workers *workers)
+{
+    h2_slot *slot;
+    for (slot = APR_RING_FIRST(&workers->idle);
+         slot != APR_RING_SENTINEL(&workers->idle, h2_slot, link);
+         slot = APR_RING_NEXT(slot, link))
+    {
+        apr_thread_cond_signal(slot->more_work);
+    }
+}
+
 static apr_status_t workers_pool_cleanup(void *data)
 {
     h2_workers *workers = data;
-    h2_slot *slot;
-    
-    if (!workers->aborted) {
-        workers->aborted = 1;
-        /* abort all idle slots */
-        for (;;) {
-            slot = pop_slot(&workers->idle);
-            if (slot) {
-                apr_thread_mutex_lock(slot->lock);
-                slot->aborted = 1;
-                apr_thread_cond_signal(slot->not_idle);
-                apr_thread_mutex_unlock(slot->lock);
-            }
-            else {
-                break;
-            }
-        }
+    apr_time_t end, timeout = apr_time_from_sec(1);
+    apr_status_t rv;
+    int n = 0, wait_sec = 5;
 
-        h2_fifo_term(workers->mplxs);
-        h2_fifo_interrupt(workers->mplxs);
+    ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, workers->s,
+                 "h2_workers: cleanup %d workers (%d idle)",
+                 workers->active_slots, workers->idle_slots);
+    apr_thread_mutex_lock(workers->lock);
+    workers->shutdown = 1;
+    workers->aborted = 1;
+    wake_all_idles(workers);
+    apr_thread_mutex_unlock(workers->lock);
+
+    /* wait for all the workers to become zombies and join them.
+     * this gets called after the mpm shuts down and all connections
+     * have either been handled (graceful) or we are forced exiting
+     * (ungrateful). Either way, we show limited patience. */
+    end = apr_time_now() + apr_time_from_sec(wait_sec);
+    while (apr_time_now() < end) {
+        apr_thread_mutex_lock(workers->lock);
+        if (!(n = workers->active_slots)) {
+            apr_thread_mutex_unlock(workers->lock);
+            break;
+        }
+        wake_all_idles(workers);
+        rv = apr_thread_cond_timedwait(workers->all_done, workers->lock, timeout);
+        apr_thread_mutex_unlock(workers->lock);
 
-        cleanup_zombies(workers);
+        if (APR_TIMEUP == rv) {
+            ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, workers->s,
+                         APLOGNO(10290) "h2_workers: waiting for workers to close, "
+                         "still seeing %d workers (%d idle) living",
+                         workers->active_slots, workers->idle_slots);
+        }
+    }
+    if (n) {
+        ap_log_error(APLOG_MARK, APLOG_WARNING, 0, workers->s,
+                     APLOGNO(10291) "h2_workers: cleanup, %d workers (%d idle) "
+                     "did not exit after %d seconds.",
+                     n, workers->idle_slots, wait_sec);
     }
+    ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, workers->s,
+                 "h2_workers: cleanup all workers terminated");
+    apr_thread_mutex_lock(workers->lock);
+    join_zombies(workers);
+    apr_thread_mutex_unlock(workers->lock);
+    ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, workers->s,
+                 "h2_workers: cleanup zombie workers joined");
+
     return APR_SUCCESS;
 }
 
-h2_workers *h2_workers_create(server_rec *s, apr_pool_t *server_pool,
-                              int min_workers, int max_workers,
-                              int idle_secs)
+h2_workers *h2_workers_create(server_rec *s, apr_pool_t *pchild,
+                              int max_slots, int min_active,
+                              apr_time_t idle_limit)
 {
-    apr_status_t status;
+    apr_status_t rv;
     h2_workers *workers;
     apr_pool_t *pool;
-    int i, n;
+    apr_allocator_t *allocator;
+    int locked = 0;
+    apr_uint32_t i;
 
     ap_assert(s);
-    ap_assert(server_pool);
+    ap_assert(pchild);
+    ap_assert(idle_limit > 0);
 
     /* let's have our own pool that will be parent to all h2_worker
      * instances we create. This happens in various threads, but always
      * guarded by our lock. Without this pool, all subpool creations would
      * happen on the pool handed to us, which we do not guard.
      */
-    apr_pool_create(&pool, server_pool);
+    rv = apr_allocator_create(&allocator);
+    if (rv != APR_SUCCESS) {
+        goto cleanup;
+    }
+    rv = apr_pool_create_ex(&pool, pchild, NULL, allocator);
+    if (rv != APR_SUCCESS) {
+        apr_allocator_destroy(allocator);
+        goto cleanup;
+    }
+    apr_allocator_owner_set(allocator, pool);
     apr_pool_tag(pool, "h2_workers");
     workers = apr_pcalloc(pool, sizeof(h2_workers));
     if (!workers) {
@@ -302,31 +457,27 @@ h2_workers *h2_workers_create(server_rec *s, apr_pool_t *server_pool,
     
     workers->s = s;
     workers->pool = pool;
-    workers->min_workers = min_workers;
-    workers->max_workers = max_workers;
-    workers->max_idle_secs = (idle_secs > 0)? idle_secs : 10;
-
-    /* FIXME: the fifo set we use here has limited capacity. Once the
-     * set is full, connections with new requests do a wait. Unfortunately,
-     * we have optimizations in place there that makes such waiting "unfair"
-     * in the sense that it may take connections a looong time to get scheduled.
-     *
-     * Need to rewrite this to use one of our double-linked lists and a mutex
-     * to have unlimited capacity and fair scheduling.
-     *
-     * For now, we just make enough room to have many connections inside one
-     * process.
-     */
-    status = h2_fifo_set_create(&workers->mplxs, pool, 8 * 1024);
-    if (status != APR_SUCCESS) {
-        return NULL;
-    }
-    
-    status = apr_threadattr_create(&workers->thread_attr, workers->pool);
-    if (status != APR_SUCCESS) {
-        return NULL;
-    }
-    
+    workers->min_active = min_active;
+    workers->max_slots = max_slots;
+    workers->idle_limit = idle_limit;
+    workers->dynamic = (workers->min_active < workers->max_slots);
+
+    ap_log_error(APLOG_MARK, APLOG_INFO, 0, s,
+                 "h2_workers: created with min=%d max=%d idle_ms=%d",
+                 workers->min_active, workers->max_slots,
+                 (int)apr_time_as_msec(idle_limit));
+
+    APR_RING_INIT(&workers->idle, h2_slot, link);
+    APR_RING_INIT(&workers->busy, h2_slot, link);
+    APR_RING_INIT(&workers->free, h2_slot, link);
+    APR_RING_INIT(&workers->zombie, h2_slot, link);
+
+    APR_RING_INIT(&workers->prod_active, ap_conn_producer_t, link);
+    APR_RING_INIT(&workers->prod_idle, ap_conn_producer_t, link);
+
+    rv = apr_threadattr_create(&workers->thread_attr, workers->pool);
+    if (rv != APR_SUCCESS) goto cleanup;
+
     if (ap_thread_stacksize != 0) {
         apr_threadattr_stacksize_set(workers->thread_attr,
                                      ap_thread_stacksize);
@@ -335,49 +486,141 @@ h2_workers *h2_workers_create(server_rec *s, apr_pool_t *server_pool,
                      (long)ap_thread_stacksize);
     }
     
-    status = apr_thread_mutex_create(&workers->lock,
-                                     APR_THREAD_MUTEX_DEFAULT,
-                                     workers->pool);
-    if (status == APR_SUCCESS) {        
-        n = workers->nslots = workers->max_workers;
-        workers->slots = apr_pcalloc(workers->pool, n * sizeof(h2_slot));
-        if (workers->slots == NULL) {
-            workers->nslots = 0;
-            status = APR_ENOMEM;
-        }
-        for (i = 0; i < n; ++i) {
-            workers->slots[i].id = i;
-        }
+    rv = apr_thread_mutex_create(&workers->lock,
+                                 APR_THREAD_MUTEX_DEFAULT,
+                                 workers->pool);
+    if (rv != APR_SUCCESS) goto cleanup;
+    rv = apr_thread_cond_create(&workers->all_done, workers->pool);
+    if (rv != APR_SUCCESS) goto cleanup;
+    rv = apr_thread_cond_create(&workers->prod_done, workers->pool);
+    if (rv != APR_SUCCESS) goto cleanup;
+
+    apr_thread_mutex_lock(workers->lock);
+    locked = 1;
+
+    /* create the slots and put them on the free list */
+    workers->slots = apr_pcalloc(workers->pool, workers->max_slots * sizeof(h2_slot));
+
+    for (i = 0; i < workers->max_slots; ++i) {
+        workers->slots[i].id = i;
+        workers->slots[i].state = H2_SLOT_FREE;
+        workers->slots[i].workers = workers;
+        APR_RING_ELEM_INIT(&workers->slots[i], link);
+        APR_RING_INSERT_TAIL(&workers->free, &workers->slots[i], h2_slot, link);
+        rv = apr_thread_cond_create(&workers->slots[i].more_work, workers->pool);
+        if (rv != APR_SUCCESS) goto cleanup;
     }
-    if (status == APR_SUCCESS) {
-        /* we activate all for now, TODO: support min_workers again.
-         * do this in reverse for vanity reasons so slot 0 will most
-         * likely be at head of idle queue. */
-        n = workers->max_workers;
-        for (i = n-1; i >= 0; --i) {
-            status = activate_slot(workers, &workers->slots[i]);
-        }
-        /* the rest of the slots go on the free list */
-        for(i = n; i < workers->nslots; ++i) {
-            push_slot(&workers->free, &workers->slots[i]);
-        }
-        workers->dynamic = (workers->worker_count < workers->max_workers);
+
+    /* activate the min amount of workers */
+    for (i = 0; i < workers->min_active; ++i) {
+        rv = activate_slot(workers);
+        if (rv != APR_SUCCESS) goto cleanup;
     }
-    if (status == APR_SUCCESS) {
-        apr_pool_pre_cleanup_register(pool, workers, workers_pool_cleanup);    
+
+cleanup:
+    if (locked) {
+        apr_thread_mutex_unlock(workers->lock);
+    }
+    if (rv == APR_SUCCESS) {
+        /* Stop/join the workers threads when the MPM child exits (pchild is
+         * destroyed), and as a pre_cleanup of pchild thus before the threads
+         * pools (children of workers->pool) so that they are not destroyed
+         * before/under us.
+         */
+        apr_pool_pre_cleanup_register(pchild, workers, workers_pool_cleanup);    
         return workers;
     }
+    ap_log_error(APLOG_MARK, APLOG_DEBUG, rv, s,
+                 "h2_workers: errors initializing");
     return NULL;
 }
 
-apr_status_t h2_workers_register(h2_workers *workers, struct h2_mplx *m)
+apr_uint32_t h2_workers_get_max_workers(h2_workers *workers)
 {
-    apr_status_t status = h2_fifo_push(workers->mplxs, m);
-    wake_idle_worker(workers);
-    return status;
+    return workers->max_slots;
 }
 
-apr_status_t h2_workers_unregister(h2_workers *workers, struct h2_mplx *m)
+void h2_workers_shutdown(h2_workers *workers, int graceful)
 {
-    return h2_fifo_remove(workers->mplxs, m);
+    ap_conn_producer_t *prod;
+
+    apr_thread_mutex_lock(workers->lock);
+    ap_log_error(APLOG_MARK, APLOG_TRACE1, 0, workers->s,
+                 "h2_workers: shutdown graceful=%d", graceful);
+    workers->shutdown = 1;
+    workers->idle_limit = apr_time_from_sec(1);
+    wake_all_idles(workers);
+    for (prod = APR_RING_FIRST(&workers->prod_idle);
+        prod != APR_RING_SENTINEL(&workers->prod_idle, ap_conn_producer_t, link);
+        prod = APR_RING_NEXT(prod, link)) {
+        if (prod->fn_shutdown) {
+            prod->fn_shutdown(prod->baton, graceful);
+        }
+    }
+    apr_thread_mutex_unlock(workers->lock);
+}
+
+ap_conn_producer_t *h2_workers_register(h2_workers *workers,
+                                        apr_pool_t *producer_pool,
+                                        const char *name,
+                                        ap_conn_producer_next *fn_next,
+                                        ap_conn_producer_done *fn_done,
+                                        ap_conn_producer_shutdown *fn_shutdown,
+                                        void *baton)
+{
+    ap_conn_producer_t *prod;
+
+    prod = apr_pcalloc(producer_pool, sizeof(*prod));
+    APR_RING_ELEM_INIT(prod, link);
+    prod->name = name;
+    prod->fn_next = fn_next;
+    prod->fn_done = fn_done;
+    prod->fn_shutdown = fn_shutdown;
+    prod->baton = baton;
+
+    apr_thread_mutex_lock(workers->lock);
+    prod->state = PROD_IDLE;
+    APR_RING_INSERT_TAIL(&workers->prod_idle, prod, ap_conn_producer_t, link);
+    apr_thread_mutex_unlock(workers->lock);
+
+    return prod;
+}
+
+apr_status_t h2_workers_join(h2_workers *workers, ap_conn_producer_t *prod)
+{
+    apr_status_t rv = APR_SUCCESS;
+
+    apr_thread_mutex_lock(workers->lock);
+    if (PROD_JOINED == prod->state) {
+        AP_DEBUG_ASSERT(APR_RING_NEXT(prod, link) == prod); /* should be in no ring */
+        rv = APR_EINVAL;
+    }
+    else {
+        AP_DEBUG_ASSERT(PROD_ACTIVE == prod->state || PROD_IDLE == prod->state);
+        APR_RING_REMOVE(prod, link);
+        prod->state = PROD_JOINED; /* prevent further activations */
+        while (prod->conns_active > 0) {
+            apr_thread_cond_wait(workers->prod_done, workers->lock);
+        }
+        APR_RING_ELEM_INIT(prod, link); /* make it link to itself */
+    }
+    apr_thread_mutex_unlock(workers->lock);
+    return rv;
+}
+
+apr_status_t h2_workers_activate(h2_workers *workers, ap_conn_producer_t *prod)
+{
+    apr_status_t rv = APR_SUCCESS;
+    apr_thread_mutex_lock(workers->lock);
+    if (PROD_IDLE == prod->state) {
+        APR_RING_REMOVE(prod, link);
+        prod->state = PROD_ACTIVE;
+        APR_RING_INSERT_TAIL(&workers->prod_active, prod, ap_conn_producer_t, link);
+        wake_idle_worker(workers, prod);
+    }
+    else if (PROD_JOINED == prod->state) {
+        rv = APR_EINVAL;
+    }
+    apr_thread_mutex_unlock(workers->lock);
+    return rv;
 }