httpd-cvs mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ic...@apache.org
Subject svn commit: r1725301 [4/6] - in /httpd/httpd/branches/2.4.x: ./ docs/manual/mod/ modules/http2/
Date Mon, 18 Jan 2016 16:22:58 GMT
Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_session.c
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_session.c?rev=1725301&r1=1725300&r2=1725301&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_session.c (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_session.c Mon Jan 18 16:22:57 2016
@@ -22,10 +22,14 @@
 #include <http_core.h>
 #include <http_config.h>
 #include <http_log.h>
+#include <scoreboard.h>
 
 #include "h2_private.h"
+#include "h2_bucket_eoc.h"
 #include "h2_bucket_eos.h"
 #include "h2_config.h"
+#include "h2_ctx.h"
+#include "h2_filter.h"
 #include "h2_h2.h"
 #include "h2_mplx.h"
 #include "h2_push.h"
@@ -40,6 +44,7 @@
 #include "h2_version.h"
 #include "h2_workers.h"
 
+
 static int frame_print(const nghttp2_frame *frame, char *buffer, size_t maxlen);
 
 static int h2_session_status_from_apr_status(apr_status_t rv)
@@ -60,19 +65,23 @@ static void update_window(void *ctx, int
 {
     h2_session *session = (h2_session*)ctx;
     nghttp2_session_consume(session->ngh2, stream_id, bytes_read);
-    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
                   "h2_session(%ld-%d): consumed %ld bytes",
                   session->id, stream_id, (long)bytes_read);
 }
 
+static apr_status_t h2_session_receive(void *ctx, 
+                                       const char *data, apr_size_t len,
+                                       apr_size_t *readlen);
+
+static int is_accepting_streams(h2_session *session); 
+static void dispatch_event(h2_session *session, h2_session_event_t ev, 
+                             int err, const char *msg);
 
 h2_stream *h2_session_open_stream(h2_session *session, int stream_id)
 {
     h2_stream * stream;
     apr_pool_t *stream_pool;
-    if (session->aborted) {
-        return NULL;
-    }
     
     if (session->spare) {
         stream_pool = session->spare;
@@ -87,6 +96,7 @@ h2_stream *h2_session_open_stream(h2_ses
     h2_stream_set_add(session->streams, stream);
     if (H2_STREAM_CLIENT_INITIATED(stream_id)
         && stream_id > session->max_stream_received) {
+        ++session->requests_received;
         session->max_stream_received = stream->id;
     }
     
@@ -166,7 +176,8 @@ static apr_status_t stream_schedule(h2_s
                                     h2_stream *stream, int eos)
 {
     (void)session;
-    return h2_stream_schedule(stream, eos, stream_pri_cmp, session);
+    return h2_stream_schedule(stream, eos, h2_session_push_enabled(session), 
+                              stream_pri_cmp, session);
 }
 
 /*
@@ -200,16 +211,14 @@ static int on_invalid_frame_recv_cb(nght
     h2_session *session = (h2_session *)userp;
     (void)ngh2;
     
-    if (session->aborted) {
-        return NGHTTP2_ERR_CALLBACK_FAILURE;
-    }
-    if (APLOGctrace2(session->c)) {
+    if (APLOGcdebug(session->c)) {
         char buffer[256];
         
         frame_print(frame, buffer, sizeof(buffer)/sizeof(buffer[0]));
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
-                      "h2_session: callback on_invalid_frame_recv error=%d %s",
-                      error, buffer);
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                      "h2_session(%ld): recv unknown FRAME[%s], frames=%ld/%ld (r/s)",
+                      session->id, buffer, (long)session->frames_received,
+                     (long)session->frames_sent);
     }
     return 0;
 }
@@ -224,14 +233,15 @@ static int on_data_chunk_recv_cb(nghttp2
     int rv;
     
     (void)flags;
-    if (session->aborted) {
-        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    if (!is_accepting_streams(session)) {
+        /* ignore */
+        return 0;
     }
     
     stream = h2_session_get_stream(session, stream_id);
     if (!stream) {
         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                      "h2_session:  stream(%ld-%d): on_data_chunk for unknown stream",
+                      "h2_stream(%ld-%d): on_data_chunk for unknown stream",
                       session->id, (int)stream_id);
         rv = nghttp2_submit_rst_stream(ngh2, NGHTTP2_FLAG_NONE, stream_id,
                                        NGHTTP2_INTERNAL_ERROR);
@@ -261,7 +271,7 @@ static apr_status_t stream_release(h2_se
                                    uint32_t error_code) 
 {
     if (!error_code) {
-        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
                       "h2_stream(%ld-%d): handled, closing", 
                       session->id, (int)stream->id);
         if (stream->id > session->max_stream_handled) {
@@ -288,9 +298,6 @@ static int on_stream_close_cb(nghttp2_se
     h2_stream *stream;
     
     (void)ngh2;
-    if (session->aborted) {
-        return NGHTTP2_ERR_CALLBACK_FAILURE;
-    }
     stream = h2_session_get_stream(session, stream_id);
     if (stream) {
         stream_release(session, stream, error_code);
@@ -329,8 +336,9 @@ static int on_header_cb(nghttp2_session
     
     (void)ngh2;
     (void)flags;
-    if (session->aborted) {
-        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    if (!is_accepting_streams(session)) {
+        /* just ignore */
+        return 0;
     }
     
     stream = h2_session_get_stream(session, frame->hd.stream_id);
@@ -364,14 +372,15 @@ static int on_frame_recv_cb(nghttp2_sess
     apr_status_t status = APR_SUCCESS;
     h2_stream *stream;
     
-    if (session->aborted) {
-        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    if (APLOGcdebug(session->c)) {
+        char buffer[256];
+        
+        frame_print(frame, buffer, sizeof(buffer)/sizeof(buffer[0]));
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                      "h2_session(%ld): recv FRAME[%s], frames=%ld/%ld (r/s)",
+                      session->id, buffer, (long)session->frames_received,
+                     (long)session->frames_sent);
     }
-    
-    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
-                  "h2_stream(%ld-%d): on_frame_rcv #%ld, type=%d", 
-                  session->id, frame->hd.stream_id, 
-                  (long)session->frames_received, frame->hd.type);
 
     ++session->frames_received;
     switch (frame->hd.type) {
@@ -435,6 +444,22 @@ static int on_frame_recv_cb(nghttp2_sess
                           session->id, (int)frame->hd.stream_id,
                           frame->window_update.window_size_increment);
             break;
+        case NGHTTP2_RST_STREAM:
+            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                          "h2_session(%ld-%d): RST_STREAM by client, errror=%d",
+                          session->id, (int)frame->hd.stream_id,
+                          (int)frame->rst_stream.error_code);
+            stream = h2_session_get_stream(session, frame->hd.stream_id);
+            if (stream && stream->initiated_on) {
+                ++session->pushes_reset;
+            }
+            else {
+                ++session->streams_reset;
+            }
+            break;
+        case NGHTTP2_GOAWAY:
+            dispatch_event(session, H2_SESSION_EV_REMOTE_GOAWAY, 0, NULL);
+            break;
         default:
             if (APLOGctrace2(session->c)) {
                 char buffer[256];
@@ -490,10 +515,6 @@ static int on_send_data_cb(nghttp2_sessi
     
     (void)ngh2;
     (void)source;
-    if (session->aborted) {
-        return NGHTTP2_ERR_CALLBACK_FAILURE;
-    }
-    
     if (frame->data.padlen > H2_MAX_PADLEN) {
         return NGHTTP2_ERR_PROTO;
     }
@@ -586,9 +607,11 @@ static int on_frame_send_cb(nghttp2_sess
         
         frame_print(frame, buffer, sizeof(buffer)/sizeof(buffer[0]));
         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                      "h2_session(%ld): frame_send %s",
-                      session->id, buffer);
+                      "h2_session(%ld): sent FRAME[%s], frames=%ld/%ld (r/s)",
+                      session->id, buffer, (long)session->frames_received,
+                     (long)session->frames_sent);
     }
+    ++session->frames_sent;
     return 0;
 }
 
@@ -618,10 +641,102 @@ static apr_status_t init_callbacks(conn_
     return APR_SUCCESS;
 }
 
+static void h2_session_cleanup(h2_session *session)
+{
+    AP_DEBUG_ASSERT(session);
+    /* This is an early cleanup of the session that may
+     * discard what is no longer necessary for *new* streams
+     * and general HTTP/2 processing.
+     * At this point, all frames are in transit or somehwere in
+     * our buffers or passed down output filters.
+     * h2 streams might still being written out.
+     */
+    if (session->c) {
+        h2_ctx_clear(session->c);
+    }
+    if (session->ngh2) {
+        nghttp2_session_del(session->ngh2);
+        session->ngh2 = NULL;
+    }
+    if (session->spare) {
+        apr_pool_destroy(session->spare);
+        session->spare = NULL;
+    }
+}
+
+static void h2_session_destroy(h2_session *session)
+{
+    AP_DEBUG_ASSERT(session);
+    h2_session_cleanup(session);
+
+    if (APLOGctrace1(session->c)) {
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                      "h2_session(%ld): destroy, %d streams open",
+                      session->id, (int)h2_stream_set_size(session->streams));
+    }
+    if (session->mplx) {
+        h2_mplx_set_consumed_cb(session->mplx, NULL, NULL);
+        h2_mplx_release_and_join(session->mplx, session->iowait);
+        session->mplx = NULL;
+    }
+    if (session->streams) {
+        h2_stream_set_destroy(session->streams);
+        session->streams = NULL;
+    }
+    if (session->pool) {
+        apr_pool_destroy(session->pool);
+    }
+}
+
+static apr_status_t h2_session_shutdown(h2_session *session, int reason, const char *msg)
+{
+    apr_status_t status = APR_SUCCESS;
+    const char *err = msg;
+    
+    AP_DEBUG_ASSERT(session);
+    if (!err && reason) {
+        err = nghttp2_strerror(reason);
+    }
+    nghttp2_submit_goaway(session->ngh2, NGHTTP2_FLAG_NONE, 
+                          h2_mplx_get_max_stream_started(session->mplx), 
+                          reason, (uint8_t*)err, err? strlen(err):0);
+    status = nghttp2_session_send(session->ngh2);
+    h2_conn_io_flush(&session->io);
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                  "session(%ld): sent GOAWAY, err=%d, msg=%s", 
+                  session->id, reason, err? err : "");
+    dispatch_event(session, H2_SESSION_EV_LOCAL_GOAWAY, reason, err);
+    return status;
+}
+
 static apr_status_t session_pool_cleanup(void *data)
 {
     h2_session *session = data;
+    /* On a controlled connection shutdown, this gets never
+     * called as we deregister and destroy our pool manually.
+     * However when we have an async mpm, and handed it our idle
+     * connection, it will just cleanup once the connection is closed
+     * from the other side (and sometimes even from out side) and
+     * here we arrive then.
+     */
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                  "session(%ld): pool_cleanup", session->id);
     
+    if (session->state != H2_SESSION_ST_DONE 
+        && session->state != H2_SESSION_ST_LOCAL_SHUTDOWN) {
+        /* Not good. The connection is being torn down and we have
+         * not sent a goaway. This is considered a protocol error and
+         * the client has to assume that any streams "in flight" may have
+         * been processed and are not safe to retry.
+         * As clients with idle connection may only learn about a closed
+         * connection when sending the next request, this has the effect
+         * that at least this one request will fail.
+         */
+        ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, session->c,
+                      "session(%ld): connection disappeared without proper "
+                      "goodbye, clients will be confused, should not happen", 
+                      session->id);
+    }
     /* keep us from destroying the pool, since that is already ongoing. */
     session->pool = NULL;
     h2_session_destroy(session);
@@ -668,14 +783,15 @@ static void *session_realloc(void *p, si
 
 static h2_session *h2_session_create_int(conn_rec *c,
                                          request_rec *r,
-                                         const h2_config *config, 
+                                         h2_ctx *ctx, 
                                          h2_workers *workers)
 {
     nghttp2_session_callbacks *callbacks = NULL;
     nghttp2_option *options = NULL;
+    uint32_t n;
 
     apr_pool_t *pool = NULL;
-    apr_status_t status = apr_pool_create(&pool, r? r->pool : c->pool);
+    apr_status_t status = apr_pool_create(&pool, c->pool);
     h2_session *session;
     if (status != APR_SUCCESS) {
         return NULL;
@@ -689,14 +805,25 @@ static h2_session *h2_session_create_int
         session->id = c->id;
         session->c = c;
         session->r = r;
-        session->config = config;
+        session->s = h2_ctx_server_get(ctx);
+        session->config = h2_config_sget(session->s);
+        
+        session->state = H2_SESSION_ST_INIT;
         
         session->pool = pool;
         apr_pool_pre_cleanup_register(pool, session, session_pool_cleanup);
         
-        session->max_stream_count = h2_config_geti(config, H2_CONF_MAX_STREAMS);
-        session->max_stream_mem = h2_config_geti(config, H2_CONF_STREAM_MAX_MEM);
-
+        session->max_stream_count = h2_config_geti(session->config, H2_CONF_MAX_STREAMS);
+        session->max_stream_mem = h2_config_geti(session->config, H2_CONF_STREAM_MAX_MEM);
+        session->timeout_secs = h2_config_geti(session->config, H2_CONF_TIMEOUT_SECS);
+        if (session->timeout_secs <= 0) {
+            session->timeout_secs = apr_time_sec(session->s->timeout);
+        }
+        session->keepalive_secs = h2_config_geti(session->config, H2_CONF_KEEPALIVE_SECS);
+        if (session->keepalive_secs <= 0) {
+            session->keepalive_secs = apr_time_sec(session->s->keep_alive_timeout);
+        }
+        
         status = apr_thread_cond_create(&session->iowait, session->pool);
         if (status != APR_SUCCESS) {
             return NULL;
@@ -705,11 +832,15 @@ static h2_session *h2_session_create_int
         session->streams = h2_stream_set_create(session->pool, session->max_stream_count);
         
         session->workers = workers;
-        session->mplx = h2_mplx_create(c, session->pool, config, workers);
+        session->mplx = h2_mplx_create(c, session->pool, session->config, workers);
         
         h2_mplx_set_consumed_cb(session->mplx, update_window, session);
         
-        h2_conn_io_init(&session->io, c, config, session->pool);
+        /* Install the connection input filter that feeds the session */
+        session->cin = h2_filter_cin_create(session->pool, h2_session_receive, session);
+        ap_add_input_filter("H2_IN", session->cin, r, c);
+
+        h2_conn_io_init(&session->io, c, session->config, session->pool);
         session->bbtmp = apr_brigade_create(session->pool, c->bucket_alloc);
         
         status = init_callbacks(c, &callbacks);
@@ -759,67 +890,33 @@ static h2_session *h2_session_create_int
             h2_session_destroy(session);
             return NULL;
         }
-        
+         
+        n = h2_config_geti(session->config, H2_CONF_PUSH_DIARY_SIZE);
+        session->push_diary = h2_push_diary_create(session->pool, n);
+        
+        if (APLOGcdebug(c)) {
+            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, c,
+                          "session(%ld) created, timeout=%d, keepalive_timeout=%d, "
+                          "max_streams=%d, stream_mem=%d, push_diary(type=%d,N=%d)",
+                          session->id, session->timeout_secs, session->keepalive_secs,
+                          (int)session->max_stream_count, (int)session->max_stream_mem,
+                          session->push_diary->dtype,
+                          (int)session->push_diary->N);
+        }
     }
     return session;
 }
 
-h2_session *h2_session_create(conn_rec *c, const h2_config *config, 
-                              h2_workers *workers)
+h2_session *h2_session_create(conn_rec *c, h2_ctx *ctx, h2_workers *workers)
 {
-    return h2_session_create_int(c, NULL, config, workers);
+    return h2_session_create_int(c, NULL, ctx, workers);
 }
 
-h2_session *h2_session_rcreate(request_rec *r, const h2_config *config, 
-                               h2_workers *workers)
+h2_session *h2_session_rcreate(request_rec *r, h2_ctx *ctx, h2_workers *workers)
 {
-    return h2_session_create_int(r->connection, r, config, workers);
+    return h2_session_create_int(r->connection, r, ctx, workers);
 }
 
-static void h2_session_cleanup(h2_session *session)
-{
-    AP_DEBUG_ASSERT(session);
-    /* This is an early cleanup of the session that may
-     * discard what is no longer necessary for *new* streams
-     * and general HTTP/2 processing.
-     * At this point, all frames are in transit or somehwere in
-     * our buffers or passed down output filters.
-     * h2 streams might still being written out.
-     */
-    if (session->ngh2) {
-        nghttp2_session_del(session->ngh2);
-        session->ngh2 = NULL;
-    }
-    if (session->spare) {
-        apr_pool_destroy(session->spare);
-        session->spare = NULL;
-    }
-}
-
-void h2_session_destroy(h2_session *session)
-{
-    AP_DEBUG_ASSERT(session);
-    h2_session_cleanup(session);
-    
-    if (session->mplx) {
-        h2_mplx_release_and_join(session->mplx, session->iowait);
-        session->mplx = NULL;
-    }
-    if (session->streams) {
-        if (!h2_stream_set_is_empty(session->streams)) {
-            ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
-                          "h2_session(%ld): destroy, %d streams open",
-                          session->id, (int)h2_stream_set_size(session->streams));
-        }
-        h2_stream_set_destroy(session->streams);
-        session->streams = NULL;
-    }
-    if (session->pool) {
-        apr_pool_destroy(session->pool);
-    }
-}
-
-
 void h2_session_eoc_callback(h2_session *session)
 {
     ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
@@ -828,73 +925,7 @@ void h2_session_eoc_callback(h2_session
     h2_session_destroy(session);
 }
 
-static apr_status_t h2_session_abort_int(h2_session *session, int reason)
-{
-    AP_DEBUG_ASSERT(session);
-    if (!session->aborted) {
-        session->aborted = 1;
-        
-        if (session->ngh2) {            
-            if (NGHTTP2_ERR_EOF == reason) {
-                /* This is our way of indication that the connection is
-                 * gone. No use to send any GOAWAY frames. */
-                nghttp2_session_terminate_session(session->ngh2, reason);
-            }
-            else if (!reason) {
-                nghttp2_submit_goaway(session->ngh2, NGHTTP2_FLAG_NONE, 
-                                      session->max_stream_received, 
-                                      reason, NULL, 0);
-                nghttp2_session_send(session->ngh2);
-            }
-            else {
-                const char *err = nghttp2_strerror(reason);
-                
-                ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                              "session(%ld): aborting session, reason=%d %s",
-                              session->id, reason, err);
-                
-                /* The connection might still be there and we shut down
-                 * with GOAWAY and reason information. */
-                nghttp2_submit_goaway(session->ngh2, NGHTTP2_FLAG_NONE, 
-                                      session->max_stream_received, 
-                                      reason, (const uint8_t *)err, 
-                                      strlen(err));
-                nghttp2_session_send(session->ngh2);
-            }
-        }
-        h2_mplx_abort(session->mplx);
-    }
-    return APR_SUCCESS;
-}
-
-apr_status_t h2_session_abort(h2_session *session, apr_status_t reason, int rv)
-{
-    AP_DEBUG_ASSERT(session);
-    if (rv == 0) {
-        rv = NGHTTP2_ERR_PROTO;
-        switch (reason) {
-            case APR_ENOMEM:
-                rv = NGHTTP2_ERR_NOMEM;
-                break;
-            case APR_SUCCESS:                            /* all fine, just... */
-            case APR_EOF:                         /* client closed its end... */
-            case APR_TIMEUP:                          /* got bored waiting... */
-                rv = 0;                            /* ...gracefully shut down */
-                break;
-            case APR_EBADF:        /* connection unusable, terminate silently */
-            default:
-                if (APR_STATUS_IS_ECONNABORTED(reason)
-                    || APR_STATUS_IS_ECONNRESET(reason)
-                    || APR_STATUS_IS_EBADF(reason)) {
-                    rv = NGHTTP2_ERR_EOF;
-                }
-                break;
-        }
-    }
-    return h2_session_abort_int(session, rv);
-}
-
-apr_status_t h2_session_start(h2_session *session, int *rv)
+static apr_status_t h2_session_start(h2_session *session, int *rv)
 {
     apr_status_t status = APR_SUCCESS;
     nghttp2_settings_entry settings[3];
@@ -969,6 +1000,10 @@ apr_status_t h2_session_start(h2_session
         ++slen;
     }
     
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c,
+                  "h2_session(%ld): start, INITIAL_WINDOW_SIZE=%ld, "
+                  "MAX_CONCURRENT_STREAMS=%d", 
+                  session->id, (long)win_size, (int)session->max_stream_count);
     *rv = nghttp2_submit_settings(session->ngh2, NGHTTP2_FLAG_NONE,
                                   settings, slen);
     if (*rv != 0) {
@@ -1004,7 +1039,8 @@ typedef struct {
     int resume_count;
 } resume_ctx;
 
-static int resume_on_data(void *ctx, h2_stream *stream) {
+static int resume_on_data(void *ctx, h2_stream *stream)
+{
     resume_ctx *rctx = (resume_ctx*)ctx;
     h2_session *session = rctx->session;
     AP_DEBUG_ASSERT(session);
@@ -1020,17 +1056,18 @@ static int resume_on_data(void *ctx, h2_
             ap_log_cerror(APLOG_MARK, nghttp2_is_fatal(rv)?
                           APLOG_ERR : APLOG_DEBUG, 0, session->c,
                           APLOGNO(02936) 
-                          "h2_stream(%ld-%d): resuming stream %s",
-                          session->id, stream->id, nghttp2_strerror(rv));
+                          "h2_stream(%ld-%d): resuming %s",
+                          session->id, stream->id, rv? nghttp2_strerror(rv) : "");
         }
     }
     return 1;
 }
 
-static int h2_session_resume_streams_with_data(h2_session *session) {
+static int h2_session_resume_streams_with_data(h2_session *session)
+{
     AP_DEBUG_ASSERT(session);
     if (!h2_stream_set_is_empty(session->streams)
-        && session->mplx && !session->aborted) {
+        && session->mplx && !session->mplx->aborted) {
         resume_ctx ctx;
         
         ctx.session      = session;
@@ -1052,49 +1089,6 @@ h2_stream *h2_session_get_stream(h2_sess
     return session->last_stream;
 }
 
-/* h2_io_on_read_cb implementation that offers the data read
- * directly to the session for consumption.
- */
-static apr_status_t session_receive(const char *data, apr_size_t len,
-                                    apr_size_t *readlen, int *done,
-                                    void *puser)
-{
-    h2_session *session = (h2_session *)puser;
-    AP_DEBUG_ASSERT(session);
-    if (len > 0) {
-        ssize_t n = nghttp2_session_mem_recv(session->ngh2,
-                                             (const uint8_t *)data, len);
-        if (n < 0) {
-            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, APR_EGENERAL,
-                          session->c,
-                          "h2_session: nghttp2_session_mem_recv error %d",
-                          (int)n);
-            if (nghttp2_is_fatal((int)n)) {
-                *done = 1;
-                h2_session_abort_int(session, (int)n);
-                return APR_EGENERAL;
-            }
-        }
-        else {
-            *readlen = n;
-        }
-    }
-    return APR_SUCCESS;
-}
-
-apr_status_t h2_session_close(h2_session *session)
-{
-    AP_DEBUG_ASSERT(session);
-    if (!session->aborted) {
-        h2_session_abort_int(session, 0);
-    }
-    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0,session->c,
-                  "h2_session: closing, writing eoc");
-    
-    h2_session_cleanup(session);
-    return h2_conn_io_close(&session->io, session);           
-}
-
 static ssize_t stream_data_cb(nghttp2_session *ng2s,
                               int32_t stream_id,
                               uint8_t *buf,
@@ -1114,9 +1108,7 @@ static ssize_t stream_data_cb(nghttp2_se
      * to find out how much of the requested length we can send without
      * blocking.
      * Indicate EOS when we encounter it or DEFERRED if the stream
-     * should be suspended.
-     * TODO: for handling of TRAILERS,  the EOF indication needs
-     * to be aware of that.
+     * should be suspended. Beware of trailers.
      */
  
     (void)ng2s;
@@ -1154,7 +1146,7 @@ static ssize_t stream_data_cb(nghttp2_se
             nread = 0;
             h2_stream_set_suspended(stream, 1);
             ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                          "h2_stream(%ld-%d): suspending stream",
+                          "h2_stream(%ld-%d): suspending",
                           session->id, (int)stream_id);
             return NGHTTP2_ERR_DEFERRED;
             
@@ -1208,17 +1200,17 @@ typedef struct {
 static apr_status_t submit_response(h2_session *session, h2_stream *stream)
 {
     apr_status_t status = APR_SUCCESS;
+    h2_response *response = h2_stream_get_response(stream);
     int rv = 0;
     AP_DEBUG_ASSERT(session);
     AP_DEBUG_ASSERT(stream);
-    AP_DEBUG_ASSERT(stream->response || stream->rst_error);
+    AP_DEBUG_ASSERT(response || stream->rst_error);
     
     if (stream->submitted) {
         rv = NGHTTP2_PROTOCOL_ERROR;
     }
-    else if (stream->response && stream->response->headers) {
+    else if (response && response->headers) {
         nghttp2_data_provider provider;
-        h2_response *response = stream->response;
         h2_ngheader *ngh;
         const h2_priority *prio;
         
@@ -1246,7 +1238,6 @@ static apr_status_t submit_response(h2_s
          *    also have the pushed ones as well.
          */
         if (!stream->initiated_on
-            && h2_config_geti(session->config, H2_CONF_PUSH)
             && H2_HTTP_2XX(response->http_status)
             && h2_session_push_enabled(session)) {
             
@@ -1263,7 +1254,6 @@ static apr_status_t submit_response(h2_s
                                         response->headers);
         rv = nghttp2_submit_response(session->ngh2, response->stream_id,
                                      ngh->nv, ngh->nvlen, &provider);
-        
     }
     else {
         int err = H2_STREAM_RST(stream, H2_ERR_PROTOCOL_ERROR);
@@ -1277,10 +1267,16 @@ static apr_status_t submit_response(h2_s
     }
     
     stream->submitted = 1;
+    if (stream->initiated_on) {
+        ++session->pushes_submitted;
+    }
+    else {
+        ++session->responses_submitted;
+    }
 
     if (nghttp2_is_fatal(rv)) {
         status = APR_EGENERAL;
-        h2_session_abort_int(session, rv);
+        dispatch_event(session, H2_SESSION_EV_PROTO_ERROR, rv, nghttp2_strerror(rv));
         ap_log_cerror(APLOG_MARK, APLOG_ERR, status, session->c,
                       APLOGNO(02940) "submit_response: %s", 
                       nghttp2_strerror(rv));
@@ -1300,16 +1296,16 @@ struct h2_stream *h2_session_push(h2_ses
     ngh = h2_util_ngheader_make_req(is->pool, push->req);
     nid = nghttp2_submit_push_promise(session->ngh2, 0, is->id, 
                                       ngh->nv, ngh->nvlen, NULL);
-                                      
     if (nid <= 0) {
         ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
                       "h2_stream(%ld-%d): submitting push promise fail: %s",
                       session->id, is->id, nghttp2_strerror(nid));
         return NULL;
     }
-
+    ++session->pushes_promised;
+    
     ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                  "h2_stream(%ld-%d): promised new stream %d for %s %s on %d",
+                  "h2_stream(%ld-%d): SERVER_PUSH %d for %s %s on %d",
                   session->id, is->id, nid,
                   push->req->method, push->req->path, is->id);
                   
@@ -1318,7 +1314,7 @@ struct h2_stream *h2_session_push(h2_ses
         h2_stream_set_h2_request(stream, is->id, push->req);
         status = stream_schedule(session, stream, 1);
         if (status != APR_SUCCESS) {
-            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c,
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, session->c,
                           "h2_stream(%ld-%d): scheduling push stream",
                           session->id, stream->id);
             h2_stream_cleanup(stream);
@@ -1357,7 +1353,7 @@ apr_status_t h2_session_set_prio(h2_sess
     
     s = nghttp2_session_find_stream(session->ngh2, stream->id);
     if (!s) {
-        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
                       "h2_stream(%ld-%d): lookup of nghttp2_stream failed",
                       session->id, stream->id);
         return APR_EINVAL;
@@ -1407,7 +1403,7 @@ apr_status_t h2_session_set_prio(h2_sess
                 rv = nghttp2_session_change_stream_priority(session->ngh2, id_parent, &ps);
                 if (rv < 0) {
                     ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                                  "h2_stream(%ld-%d): PUSH BEFORE2, weight=%d, "
+                                  "h2_stream(%ld-%d): PUSH BEFORE, weight=%d, "
                                   "depends=%d, returned=%d",
                                   session->id, id_parent, ps.weight, ps.stream_id, rv);
                     return APR_EGENERAL;
@@ -1543,7 +1539,7 @@ static int frame_print(const nghttp2_fra
         }
         default:
             return apr_snprintf(buffer, maxlen,
-                                "FRAME[type=%d, length=%d, flags=%d, stream=%d]",
+                                "type=%d[length=%d, flags=%d, stream=%d]",
                                 frame->hd.type, (int)frame->hd.length,
                                 frame->hd.flags, frame->hd.stream_id);
     }
@@ -1551,216 +1547,597 @@ static int frame_print(const nghttp2_fra
 
 int h2_session_push_enabled(h2_session *session)
 {
-    return nghttp2_session_get_remote_settings(session->ngh2, 
-                                               NGHTTP2_SETTINGS_ENABLE_PUSH);
+    /* iff we can and they can */
+    return (h2_config_geti(session->config, H2_CONF_PUSH)
+            && nghttp2_session_get_remote_settings(session->ngh2, 
+                                                   NGHTTP2_SETTINGS_ENABLE_PUSH));
 }
 
-
-apr_status_t h2_session_process(h2_session *session)
+static apr_status_t h2_session_send(h2_session *session)
 {
-    apr_status_t status = APR_SUCCESS;
-    apr_interval_time_t wait_micros = 0;
-    static const int MAX_WAIT_MICROS = 200 * 1000;
-    int got_streams = 0;
-    h2_stream *stream;
-
-    while (!session->aborted && (nghttp2_session_want_read(session->ngh2)
-                                 || nghttp2_session_want_write(session->ngh2))) {
-        int have_written = 0;
-        int have_read = 0;
-                                 
-        got_streams = !h2_stream_set_is_empty(session->streams);
-        if (got_streams) {            
-            h2_session_resume_streams_with_data(session);
-            
-            if (h2_stream_set_has_unsubmitted(session->streams)) {
-                int unsent_submits = 0;
-                
-                /* If we have responses ready, submit them now. */
-                while ((stream = h2_mplx_next_submit(session->mplx, session->streams))) {
-                    status = submit_response(session, stream);
-                    ++unsent_submits;
-                    
-                    /* Unsent push promises are written immediately, as nghttp2
-                     * 1.5.0 realizes internal stream data structures only on 
-                     * send and we might need them for other submits. 
-                     * Also, to conserve memory, we send at least every 10 submits
-                     * so that nghttp2 does not buffer all outbound items too 
-                     * long.
-                     */
-                    if (status == APR_SUCCESS 
-                        && (session->unsent_promises || unsent_submits > 10)) {
-                        int rv = nghttp2_session_send(session->ngh2);
-                        if (rv != 0) {
-                            ap_log_cerror( APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                                          "h2_session: send: %s", nghttp2_strerror(rv));
-                            if (nghttp2_is_fatal(rv)) {
-                                h2_session_abort(session, status, rv);
-                                goto end_process;
-                            }
-                        }
-                        else {
-                            have_written = 1;
-                            wait_micros = 0;
-                            session->unsent_promises = 0;
-                            unsent_submits = 0;
-                        }
-                    }
-                }
-            }
+    int rv = nghttp2_session_send(session->ngh2);
+    if (rv != 0) {
+        if (nghttp2_is_fatal(rv)) {
+            dispatch_event(session, H2_SESSION_EV_PROTO_ERROR, rv, nghttp2_strerror(rv));
+            return APR_EGENERAL;
         }
-        
-        /* Send data as long as we have it and window sizes allow. We are
-         * a server after all.
-         */
-        if (nghttp2_session_want_write(session->ngh2)) {
-            int rv;
-            
-            rv = nghttp2_session_send(session->ngh2);
-            if (rv != 0) {
-                ap_log_cerror( APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                              "h2_session: send: %s", nghttp2_strerror(rv));
-                if (nghttp2_is_fatal(rv)) {
-                    h2_session_abort(session, status, rv);
-                    goto end_process;
-                }
-            }
-            else {
-                have_written = 1;
-                wait_micros = 0;
-                session->unsent_promises = 0;
+    }
+    
+    session->unsent_promises = 0;
+    session->unsent_submits = 0;
+    
+    return APR_SUCCESS;
+}
+
+static apr_status_t h2_session_receive(void *ctx, const char *data, 
+                                       apr_size_t len, apr_size_t *readlen)
+{
+    h2_session *session = ctx;
+    ssize_t n;
+    
+    if (len > 0) {
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
+                      "h2_session(%ld): feeding %ld bytes to nghttp2",
+                      session->id, (long)len);
+        n = nghttp2_session_mem_recv(session->ngh2, (const uint8_t *)data, len);
+        if (n < 0) {
+            if (nghttp2_is_fatal((int)n)) {
+                dispatch_event(session, H2_SESSION_EV_PROTO_ERROR, (int)n, nghttp2_strerror(n));
+                return APR_EGENERAL;
             }
         }
-        
-        if (wait_micros > 0) {
-            if (APLOGcdebug(session->c)) {
-                ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
-                              "h2_session: wait for data, %ld micros", 
-                              (long)wait_micros);
-            }
-            nghttp2_session_send(session->ngh2);
-            h2_conn_io_flush(&session->io);
-            status = h2_mplx_out_trywait(session->mplx, wait_micros, session->iowait);
-            
-            if (status == APR_TIMEUP) {
-                if (wait_micros < MAX_WAIT_MICROS) {
-                    wait_micros *= 2;
-                }
-            }
+        else {
+            *readlen = n;
+            session->io.bytes_read += n;
         }
-        
-        if (nghttp2_session_want_read(session->ngh2))
-        {
-            /* When we
-             * - and have no streams at all
-             * - or have streams, but none is suspended or needs submit and
-             *   have nothing written on the last try
-             * 
-             * or, the other way around
-             * - have only streams where data can be sent, but could
-             *   not send anything
-             *
-             * then we are waiting on frames from the client (for
-             * example WINDOW_UPDATE or HEADER) and without new frames
-             * from the client, we cannot make any progress,
-             * 
-             * and *then* we can safely do a blocking read.
-             */
-            int may_block = (session->frames_received <= 1);
-            if (!may_block) {
-                if (got_streams) {
-                    may_block = (!have_written 
-                                 && !h2_stream_set_has_unsubmitted(session->streams)
-                                 && !h2_stream_set_has_suspended(session->streams));
-                }
-                else {
-                    may_block = 1;
-                }
-            }
-            
-            if (may_block) {
-                h2_conn_io_flush(&session->io);
-                if (session->c->cs) {
-                    session->c->cs->state = (got_streams? CONN_STATE_HANDLER
-                                             : CONN_STATE_WRITE_COMPLETION);
-                }
-                status = h2_conn_io_read(&session->io, APR_BLOCK_READ, 
-                                         session_receive, session);
-            }
-            else {
-                if (session->c->cs) {
-                    session->c->cs->state = CONN_STATE_HANDLER;
-                }
-                status = h2_conn_io_read(&session->io, APR_NONBLOCK_READ, 
-                                         session_receive, session);
-            }
+    }
+    return APR_SUCCESS;
+}
 
-            switch (status) {
-                case APR_SUCCESS:       /* successful read, reset our idle timers */
-                    have_read = 1;
-                    wait_micros = 0;
-                    break;
-                case APR_EAGAIN:              /* non-blocking read, nothing there */
-                    break;
-                default:
+static apr_status_t h2_session_read(h2_session *session, int block, int loops)
+{
+    apr_status_t status, rstatus = APR_EAGAIN;
+    conn_rec *c = session->c;
+    int i;
+    
+    for (i = 0; i < loops; ++i) {
+        /* H2_IN filter handles all incoming data against the session.
+         * We just pull at the filter chain to make it happen */
+        status = ap_get_brigade(c->input_filters,
+                                session->bbtmp, AP_MODE_READBYTES,
+                                block? APR_BLOCK_READ : APR_NONBLOCK_READ,
+                                APR_BUCKET_BUFF_SIZE);
+        /* get rid of any possible data we do not expect to get */
+        apr_brigade_cleanup(session->bbtmp); 
+
+        switch (status) {
+            case APR_SUCCESS:
+                /* successful read, reset our idle timers */
+                rstatus = APR_SUCCESS;
+                if (block) {
+                    /* successfull blocked read, try unblocked to
+                     * get more. */
+                    block = 0;
+                }
+                break;
+            case APR_EAGAIN:
+                return rstatus;
+            case APR_TIMEUP:
+                return status;
+            default:
+                if (!i) {
+                    /* first attempt failed */
                     if (APR_STATUS_IS_ETIMEDOUT(status)
                         || APR_STATUS_IS_ECONNABORTED(status)
                         || APR_STATUS_IS_ECONNRESET(status)
                         || APR_STATUS_IS_EOF(status)
                         || APR_STATUS_IS_EBADF(status)) {
                         /* common status for a client that has left */
-                        ap_log_cerror( APLOG_MARK, APLOG_DEBUG, status, session->c,
-                                      "h2_session(%ld): terminating",
-                                      session->id);
-                        /* Stolen from mod_reqtimeout to speed up lingering when
-                         * a read timeout happened.
-                         */
-                        apr_table_setn(session->c->notes, "short-lingering-close", "1");
+                        ap_log_cerror( APLOG_MARK, APLOG_TRACE1, status, c,
+                                      "h2_session(%ld): input gone", session->id);
                     }
                     else {
                         /* uncommon status, log on INFO so that we see this */
-                        ap_log_cerror( APLOG_MARK, APLOG_INFO, status, session->c,
+                        ap_log_cerror( APLOG_MARK, APLOG_INFO, status, c,
                                       APLOGNO(02950) 
                                       "h2_session(%ld): error reading, terminating",
                                       session->id);
                     }
-                    h2_session_abort(session, status, 0);
-                    goto end_process;
-            }
+                    return status;
+                }
+                /* subsequent failure after success(es), return initial
+                 * status. */
+                return rstatus;
         }
-        
-        got_streams = !h2_stream_set_is_empty(session->streams);
-        if (got_streams) {            
-            if (session->reprioritize) {
-                h2_mplx_reprioritize(session->mplx, stream_pri_cmp, session);
-                session->reprioritize = 0;
-            }
+        if (!is_accepting_streams(session)) {
+            break;
+        }
+    }
+    return rstatus;
+}
+
+static apr_status_t h2_session_submit(h2_session *session)
+{
+    apr_status_t status = APR_EAGAIN;
+    h2_stream *stream;
+    
+    if (h2_stream_set_has_unsubmitted(session->streams)) {
+        /* If we have responses ready, submit them now. */
+        while ((stream = h2_mplx_next_submit(session->mplx, session->streams))) {
+            status = submit_response(session, stream);
+            ++session->unsent_submits;
             
-            if (!have_read && !have_written) {
-                /* Nothing read or written. That means no data yet ready to 
-                 * be send out. Slowly back off...
-                 */
-                if (wait_micros == 0) {
-                    wait_micros = 10;
+            /* Unsent push promises are written immediately, as nghttp2
+             * 1.5.0 realizes internal stream data structures only on 
+             * send and we might need them for other submits. 
+             * Also, to conserve memory, we send at least every 10 submits
+             * so that nghttp2 does not buffer all outbound items too 
+             * long.
+             */
+            if (status == APR_SUCCESS 
+                && (session->unsent_promises || session->unsent_submits > 10)) {
+                status = h2_session_send(session);
+                if (status != APR_SUCCESS) {
+                    break;
                 }
             }
-            
-            /* Check that any pending window updates are sent. */
-            status = h2_mplx_in_update_windows(session->mplx);
-            if (APR_STATUS_IS_EAGAIN(status)) {
-                status = APR_SUCCESS;
+        }
+    }
+    return status;
+}
+
+static const char *StateNames[] = {
+    "INIT",      /* H2_SESSION_ST_INIT */
+    "DONE",      /* H2_SESSION_ST_DONE */
+    "IDLE",      /* H2_SESSION_ST_IDLE */
+    "BUSY",      /* H2_SESSION_ST_BUSY */
+    "WAIT",      /* H2_SESSION_ST_WAIT */
+    "LSHUTDOWN", /* H2_SESSION_ST_LOCAL_SHUTDOWN */
+    "RSHUTDOWN", /* H2_SESSION_ST_REMOTE_SHUTDOWN */
+};
+
+static const char *state_name(h2_session_state state)
+{
+    if (state >= (sizeof(StateNames)/sizeof(StateNames[0]))) {
+        return "unknown";
+    }
+    return StateNames[state];
+}
+
+static int is_accepting_streams(h2_session *session)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_IDLE:
+        case H2_SESSION_ST_BUSY:
+        case H2_SESSION_ST_WAIT:
+            return 1;
+        default:
+            return 0;
+    }
+}
+
+static void transit(h2_session *session, const char *action, h2_session_state nstate)
+{
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                  "h2_session(%ld): transit [%s] -- %s --> [%s]", session->id,
+                  state_name(session->state), action, state_name(nstate));
+    session->state = nstate;
+}
+
+static void h2_session_ev_init(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_INIT:
+            transit(session, "init", H2_SESSION_ST_BUSY);
+            break;
+
+        default:
+            /* nop */
+            break;
+    }
+}
+
+static void h2_session_ev_local_goaway(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_LOCAL_SHUTDOWN:
+            /* already did that? */
+            break;
+        case H2_SESSION_ST_IDLE:
+        case H2_SESSION_ST_REMOTE_SHUTDOWN:
+            /* all done */
+            transit(session, "local goaway", H2_SESSION_ST_DONE);
+            break;
+        default:
+            transit(session, "local goaway", H2_SESSION_ST_LOCAL_SHUTDOWN);
+            break;
+    }
+}
+
+static void h2_session_ev_remote_goaway(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_REMOTE_SHUTDOWN:
+            /* already received that? */
+            break;
+        case H2_SESSION_ST_IDLE:
+        case H2_SESSION_ST_LOCAL_SHUTDOWN:
+            /* all done */
+            transit(session, "remote goaway", H2_SESSION_ST_DONE);
+            break;
+        default:
+            transit(session, "remote goaway", H2_SESSION_ST_REMOTE_SHUTDOWN);
+            break;
+    }
+}
+
+static void h2_session_ev_conn_error(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_INIT:
+        case H2_SESSION_ST_DONE:
+        case H2_SESSION_ST_LOCAL_SHUTDOWN:
+            /* just leave */
+            transit(session, "conn error", H2_SESSION_ST_DONE);
+            break;
+        
+        default:
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                          "h2_session(%ld): conn error -> shutdown", session->id);
+            h2_session_shutdown(session, arg, msg);
+            break;
+    }
+}
+
+static void h2_session_ev_proto_error(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_DONE:
+        case H2_SESSION_ST_LOCAL_SHUTDOWN:
+            /* just leave */
+            transit(session, "proto error", H2_SESSION_ST_DONE);
+            break;
+        
+        default:
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                          "h2_session(%ld): proto error -> shutdown", session->id);
+            h2_session_shutdown(session, arg, msg);
+            break;
+    }
+}
+
+static void h2_session_ev_conn_timeout(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_LOCAL_SHUTDOWN:
+            transit(session, "conn timeout", H2_SESSION_ST_DONE);
+            break;
+        default:
+            h2_session_shutdown(session, arg, msg);
+            transit(session, "conn timeout", H2_SESSION_ST_DONE);
+            break;
+    }
+}
+
+static void h2_session_ev_no_io(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_BUSY:
+            /* nothing for input and output to do. If we remain
+             * in this state, we go into a tight loop and suck up
+             * CPU cycles. Ideally, we'd like to do a blocking read, but that
+             * is not possible if we have scheduled tasks and wait
+             * for them to produce something. */
+            if (h2_stream_set_is_empty(session->streams)) {
+                /* When we have no streams, no task event are possible,
+                 * switch to blocking reads */
+                transit(session, "no io", H2_SESSION_ST_IDLE);
+            }
+            else if (!h2_stream_set_has_unsubmitted(session->streams)
+                     && !h2_stream_set_has_suspended(session->streams)) {
+                /* none of our streams is waiting for a response or
+                 * new output data from task processing, 
+                 * switch to blocking reads. */
+                transit(session, "no io", H2_SESSION_ST_IDLE);
             }
-            else if (status == APR_SUCCESS) {
-                /* need to flush window updates onto the connection asap */
-                h2_conn_io_flush(&session->io);
+            else {
+                /* Unable to do blocking reads, as we wait on events from
+                 * task processing in other threads. Do a busy wait with
+                 * backoff timer. */
+                transit(session, "no io", H2_SESSION_ST_WAIT);
             }
+            break;
+        default:
+            /* nop */
+            break;
+    }
+}
+
+static void h2_session_ev_wait_timeout(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_WAIT:
+            transit(session, "wait timeout", H2_SESSION_ST_BUSY);
+            break;
+        default:
+            /* nop */
+            break;
+    }
+}
+
+static void h2_session_ev_stream_ready(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_WAIT:
+            transit(session, "stream ready", H2_SESSION_ST_BUSY);
+            break;
+        default:
+            /* nop */
+            break;
+    }
+}
+
+static void h2_session_ev_data_read(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_IDLE:
+            transit(session, "data read", H2_SESSION_ST_BUSY);
+            break;
+            /* fall through */
+        default:
+            /* nop */
+            break;
+    }
+}
+
+static void h2_session_ev_ngh2_done(h2_session *session, int arg, const char *msg)
+{
+    switch (session->state) {
+        case H2_SESSION_ST_DONE:
+            /* nop */
+            break;
+        default:
+            transit(session, "nghttp2 done", H2_SESSION_ST_DONE);
+            break;
+    }
+}
+
+static void dispatch_event(h2_session *session, h2_session_event_t ev, 
+                      int arg, const char *msg)
+{
+    switch (ev) {
+        case H2_SESSION_EV_INIT:
+            h2_session_ev_init(session, arg, msg);
+            break;            
+        case H2_SESSION_EV_LOCAL_GOAWAY:
+            h2_session_ev_local_goaway(session, arg, msg);
+            break;
+        case H2_SESSION_EV_REMOTE_GOAWAY:
+            h2_session_ev_remote_goaway(session, arg, msg);
+            break;
+        case H2_SESSION_EV_CONN_ERROR:
+            h2_session_ev_conn_error(session, arg, msg);
+            break;
+        case H2_SESSION_EV_PROTO_ERROR:
+            h2_session_ev_proto_error(session, arg, msg);
+            break;
+        case H2_SESSION_EV_CONN_TIMEOUT:
+            h2_session_ev_conn_timeout(session, arg, msg);
+            break;
+        case H2_SESSION_EV_NO_IO:
+            h2_session_ev_no_io(session, arg, msg);
+            break;
+        case H2_SESSION_EV_WAIT_TIMEOUT:
+            h2_session_ev_wait_timeout(session, arg, msg);
+            break;
+        case H2_SESSION_EV_STREAM_READY:
+            h2_session_ev_stream_ready(session, arg, msg);
+            break;
+        case H2_SESSION_EV_DATA_READ:
+            h2_session_ev_data_read(session, arg, msg);
+            break;
+        case H2_SESSION_EV_NGH2_DONE:
+            h2_session_ev_ngh2_done(session, arg, msg);
+            break;
+        default:
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                          "h2_session(%ld): unknown event %d", 
+                          session->id, ev);
+            break;
+    }
+    
+    if (session->state == H2_SESSION_ST_DONE) {
+        h2_mplx_abort(session->mplx);
+    }
+}
+
+static const int MAX_WAIT_MICROS = 200 * 1000;
+
+apr_status_t h2_session_process(h2_session *session, int async)
+{
+    apr_status_t status = APR_SUCCESS;
+    conn_rec *c = session->c;
+    int rv, have_written, have_read;
+
+    ap_log_cerror( APLOG_MARK, APLOG_TRACE1, status, c,
+                  "h2_session(%ld): process start, async=%d", session->id, async);
+                  
+    while (1) {
+        have_read = have_written = 0;
+
+        switch (session->state) {
+            case H2_SESSION_ST_INIT:
+                if (!h2_is_acceptable_connection(c, 1)) {
+                    h2_session_shutdown(session, NGHTTP2_INADEQUATE_SECURITY, NULL);
+                } 
+                else {
+                    ap_update_child_status(c->sbh, SERVER_BUSY_READ, NULL);
+                    status = h2_session_start(session, &rv);
+                    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, c,
+                                  "h2_session(%ld): started on %s:%d", session->id,
+                                  session->s->server_hostname,
+                                  c->local_addr->port);
+                    if (status != APR_SUCCESS) {
+                        dispatch_event(session, H2_SESSION_EV_CONN_ERROR, 0, NULL);
+                    }
+                    dispatch_event(session, H2_SESSION_EV_INIT, 0, NULL);
+                }
+                break;
+                
+            case H2_SESSION_ST_IDLE:
+                h2_filter_cin_timeout_set(session->cin, session->keepalive_secs);
+                ap_update_child_status(c->sbh, SERVER_BUSY_KEEPALIVE, NULL);
+                status = h2_session_read(session, 1, 10);
+                if (status == APR_SUCCESS) {
+                    have_read = 1;
+                    dispatch_event(session, H2_SESSION_EV_DATA_READ, 0, NULL);
+                }
+                else if (status == APR_EAGAIN) {
+                    /* nothing to read */
+                }
+                else if (APR_STATUS_IS_TIMEUP(status)) {
+                    dispatch_event(session, H2_SESSION_EV_CONN_TIMEOUT, 0, NULL);
+                    break;
+                }
+                else {
+                    dispatch_event(session, H2_SESSION_EV_CONN_ERROR, 0, NULL);
+                }
+                break;
+                
+            case H2_SESSION_ST_BUSY:
+            case H2_SESSION_ST_LOCAL_SHUTDOWN:
+            case H2_SESSION_ST_REMOTE_SHUTDOWN:
+                if (nghttp2_session_want_read(session->ngh2)) {
+                    ap_update_child_status(c->sbh, SERVER_BUSY_READ, NULL);
+                    h2_filter_cin_timeout_set(session->cin, session->timeout_secs);
+                    status = h2_session_read(session, 0, 10);
+                    if (status == APR_SUCCESS) {
+                        have_read = 1;
+                        dispatch_event(session, H2_SESSION_EV_DATA_READ, 0, NULL);
+                    }
+                    else if (status == APR_EAGAIN) {
+                        /* nothing to read */
+                    }
+                    else if (APR_STATUS_IS_TIMEUP(status)) {
+                        dispatch_event(session, H2_SESSION_EV_CONN_TIMEOUT, 0, NULL);
+                        break;
+                    }
+                    else {
+                        dispatch_event(session, H2_SESSION_EV_CONN_ERROR, 0, NULL);
+                    }
+                }
+                
+                if (!h2_stream_set_is_empty(session->streams)) {
+                    /* resume any streams for which data is available again */
+                    h2_session_resume_streams_with_data(session);
+                    /* Submit any responses/push_promises that are ready */
+                    status = h2_session_submit(session);
+                    if (status == APR_SUCCESS) {
+                        have_written = 1;
+                    }
+                    else if (status != APR_EAGAIN) {
+                        dispatch_event(session, H2_SESSION_EV_CONN_ERROR, 
+                                         H2_ERR_INTERNAL_ERROR, "submit error");
+                        break;
+                    }
+                    /* send out window updates for our inputs */
+                    status = h2_mplx_in_update_windows(session->mplx);
+                    if (status != APR_SUCCESS && status != APR_EAGAIN) {
+                        dispatch_event(session, H2_SESSION_EV_CONN_ERROR, 
+                                         H2_ERR_INTERNAL_ERROR, "window update error");
+                        break;
+                    }
+                }
+                
+                if (nghttp2_session_want_write(session->ngh2)) {
+                    status = h2_session_send(session);
+                    if (status == APR_SUCCESS) {
+                        have_written = 1;
+                    }
+                    else {
+                        dispatch_event(session, H2_SESSION_EV_CONN_ERROR, 
+                                         H2_ERR_INTERNAL_ERROR, "writing");
+                        break;
+                    }
+                }
+                
+                if (have_read || have_written) {
+                    session->wait_us = 0;
+                }
+                else {
+                    dispatch_event(session, H2_SESSION_EV_NO_IO, 0, NULL);
+                }
+                break;
+                
+            case H2_SESSION_ST_WAIT:
+                session->wait_us = H2MAX(session->wait_us, 10);
+                if (APLOGctrace1(c)) {
+                    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, c,
+                                  "h2_session: wait for data, %ld micros", 
+                                  (long)session->wait_us);
+                }
+                
+                ap_log_cerror( APLOG_MARK, APLOG_TRACE2, status, c,
+                              "h2_session(%ld): process -> trywait", session->id);
+                status = h2_mplx_out_trywait(session->mplx, session->wait_us, 
+                                             session->iowait);
+                if (status == APR_SUCCESS) {
+                    dispatch_event(session, H2_SESSION_EV_STREAM_READY, 0, NULL);
+                }
+                else if (status == APR_TIMEUP) {
+                    /* nothing, increase timer for graceful backup */
+                    session->wait_us = H2MIN(session->wait_us*2, MAX_WAIT_MICROS);
+                    dispatch_event(session, H2_SESSION_EV_WAIT_TIMEOUT, 0, NULL);
+                }
+                else {
+                    h2_session_shutdown(session, H2_ERR_INTERNAL_ERROR, "cond wait error");
+                }
+                break;
+                
+            case H2_SESSION_ST_DONE:
+                status = APR_EOF;
+                goto out;
+                
+            default:
+                ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_EGENERAL, c,
+                              "h2_session(%ld): unknown state %d", session->id, session->state);
+                dispatch_event(session, H2_SESSION_EV_PROTO_ERROR, 0, NULL);
+                break;
         }
-        
+
         if (have_written) {
             h2_conn_io_flush(&session->io);
         }
+        else if (!nghttp2_session_want_read(session->ngh2) 
+                 && !nghttp2_session_want_write(session->ngh2)) {
+            dispatch_event(session, H2_SESSION_EV_NGH2_DONE, 0, NULL); 
+        }
+    }
+    
+out:
+    if (have_written) {
+        h2_conn_io_flush(&session->io);
+    }
+    
+    ap_log_cerror( APLOG_MARK, APLOG_TRACE1, status, c,
+                  "h2_session(%ld): [%s] process returns", 
+                  session->id, state_name(session->state));
+
+    if ((session->state != H2_SESSION_ST_DONE)
+        && (APR_STATUS_IS_EOF(status)
+            || APR_STATUS_IS_ECONNRESET(status) 
+            || APR_STATUS_IS_ECONNABORTED(status))) {
+            dispatch_event(session, H2_SESSION_EV_CONN_ERROR, 0, NULL);
+        }
+
+    status = (session->state == H2_SESSION_ST_DONE)? APR_EOF : APR_SUCCESS;
+    if (session->state == H2_SESSION_ST_DONE) {
+        if (!session->eoc_written) {
+            session->eoc_written = 1;
+            h2_conn_io_write_eoc(&session->io, 
+                                 h2_bucket_eoc_create(session->c->bucket_alloc, session));
+        }
     }
     
-end_process:
     return status;
 }

Modified: httpd/httpd/branches/2.4.x/modules/http2/h2_session.h
URL: http://svn.apache.org/viewvc/httpd/httpd/branches/2.4.x/modules/http2/h2_session.h?rev=1725301&r1=1725300&r2=1725301&view=diff
==============================================================================
--- httpd/httpd/branches/2.4.x/modules/http2/h2_session.h (original)
+++ httpd/httpd/branches/2.4.x/modules/http2/h2_session.h Mon Jan 18 16:22:57 2016
@@ -39,10 +39,13 @@
 
 struct apr_thread_mutext_t;
 struct apr_thread_cond_t;
+struct h2_ctx;
 struct h2_config;
+struct h2_filter_cin;
 struct h2_mplx;
 struct h2_priority;
 struct h2_push;
+struct h2_push_diary;
 struct h2_response;
 struct h2_session;
 struct h2_stream;
@@ -51,42 +54,85 @@ struct h2_workers;
 
 struct nghttp2_session;
 
-typedef struct h2_session h2_session;
+typedef enum {
+    H2_SESSION_ST_INIT,             /* send initial SETTINGS, etc. */
+    H2_SESSION_ST_DONE,             /* finished, connection close */
+    H2_SESSION_ST_IDLE,             /* nothing to write, expecting data inc */
+    H2_SESSION_ST_BUSY,             /* read/write without stop */
+    H2_SESSION_ST_WAIT,             /* waiting for tasks reporting back */
+    H2_SESSION_ST_LOCAL_SHUTDOWN,   /* we announced GOAWAY */
+    H2_SESSION_ST_REMOTE_SHUTDOWN,  /* client announced GOAWAY */
+} h2_session_state;
+
+typedef enum {
+    H2_SESSION_EV_INIT,             /* session was initialized */
+    H2_SESSION_EV_LOCAL_GOAWAY,     /* we send a GOAWAY */
+    H2_SESSION_EV_REMOTE_GOAWAY,    /* remote send us a GOAWAY */
+    H2_SESSION_EV_CONN_ERROR,       /* connection error */
+    H2_SESSION_EV_PROTO_ERROR,      /* protocol error */
+    H2_SESSION_EV_CONN_TIMEOUT,     /* connection timeout */
+    H2_SESSION_EV_NO_IO,            /* nothing has been read or written */
+    H2_SESSION_EV_WAIT_TIMEOUT,     /* timeout waiting for tasks */
+    H2_SESSION_EV_STREAM_READY,     /* stream signalled availability of headers/data */
+    H2_SESSION_EV_DATA_READ,        /* connection data has been read */
+    H2_SESSION_EV_NGH2_DONE,        /* nghttp2 wants neither read nor write anything */
+} h2_session_event_t;
 
-struct h2_session {
+typedef struct h2_session {
     long id;                        /* identifier of this session, unique
                                      * inside a httpd process */
     conn_rec *c;                    /* the connection this session serves */
     request_rec *r;                 /* the request that started this in case
                                      * of 'h2c', NULL otherwise */
+    server_rec *s;                  /* server/vhost we're starting on */
     const struct h2_config *config; /* Relevant config for this session */
-    int aborted;                    /* this session is being aborted */
-    int reprioritize;               /* scheduled streams priority needs to 
-                                     * be re-evaluated */
-    int unsent_promises;            /* number of submitted, but not yet sent
-                                     * push promised */
+    
+    h2_session_state state;         /* state session is in */
+    unsigned int reprioritize  : 1; /* scheduled streams priority changed */
+    unsigned int eoc_written   : 1; /* h2 eoc bucket written */
+    apr_interval_time_t  wait_us;   /* timout during BUSY_WAIT state, micro secs */
+    
+    int unsent_submits;             /* number of submitted, but not yet written responses. */
+    int unsent_promises;            /* number of submitted, but not yet written push promised */
+                                         
+    int requests_received;          /* number of http/2 requests received */
+    int responses_submitted;        /* number of http/2 responses submitted */
+    int streams_reset;              /* number of http/2 streams reset by client */
+    int pushes_promised;            /* number of http/2 push promises submitted */
+    int pushes_submitted;           /* number of http/2 pushed responses submitted */
+    int pushes_reset;               /* number of http/2 pushed reset by client */
+    
     apr_size_t frames_received;     /* number of http/2 frames received */
+    apr_size_t frames_sent;         /* number of http/2 frames sent */
+    
+    int max_stream_received;        /* highest stream id created */
+    int max_stream_handled;         /* highest stream id completed */
+    
     apr_size_t max_stream_count;    /* max number of open streams */
     apr_size_t max_stream_mem;      /* max buffer memory for a single stream */
     
+    int timeout_secs;               /* connection timeout (seconds) */
+    int keepalive_secs;             /* connection idle timeout (seconds) */
+    
     apr_pool_t *pool;               /* pool to use in session handling */
     apr_bucket_brigade *bbtmp;      /* brigade for keeping temporary data */
     struct apr_thread_cond_t *iowait; /* our cond when trywaiting for data */
     
+    struct h2_filter_cin *cin;      /* connection input filter context */
     h2_conn_io io;                  /* io on httpd conn filters */
+
     struct h2_mplx *mplx;           /* multiplexer for stream data */
     
     struct h2_stream *last_stream;  /* last stream worked with */
     struct h2_stream_set *streams;  /* streams handled by this session */
     
-    int max_stream_received;        /* highest stream id created */
-    int max_stream_handled;         /* highest stream id handled successfully */
-    
     apr_pool_t *spare;              /* spare stream pool */
     
     struct nghttp2_session *ngh2;   /* the nghttp2 session (internal use) */
     struct h2_workers *workers;     /* for executing stream tasks */
-};
+    
+    struct h2_push_diary *push_diary; /* remember pushes, avoid duplicates */
+} h2_session;
 
 
 /**
@@ -97,7 +143,7 @@ struct h2_session {
  * @param workers the worker pool to use
  * @return the created session
  */
-h2_session *h2_session_create(conn_rec *c, const struct h2_config *cfg, 
+h2_session *h2_session_create(conn_rec *c, struct h2_ctx *ctx, 
                               struct h2_workers *workers);
 
 /**
@@ -108,7 +154,7 @@ h2_session *h2_session_create(conn_rec *
  * @param workers the worker pool to use
  * @return the created session
  */
-h2_session *h2_session_rcreate(request_rec *r, const struct h2_config *cfg,
+h2_session *h2_session_rcreate(request_rec *r, struct h2_ctx *ctx,
                                struct h2_workers *workers);
 
 /**
@@ -117,14 +163,7 @@ h2_session *h2_session_rcreate(request_r
  *
  * @param session the sessionm to process
  */
-apr_status_t h2_session_process(h2_session *session);
-
-/**
- * Destroy the session and all objects it still contains. This will not
- * destroy h2_task instances that have not finished yet. 
- * @param session the session to destroy
- */
-void h2_session_destroy(h2_session *session);
+apr_status_t h2_session_process(h2_session *session, int async);
 
 /**
  * Cleanup the session and all objects it still contains. This will not
@@ -134,27 +173,17 @@ void h2_session_destroy(h2_session *sess
 void h2_session_eoc_callback(h2_session *session);
 
 /**
- * Called once at start of session. 
- * Sets up the session and sends the initial SETTINGS frame.
- * @param session the session to start
- * @param rv error codes in libnghttp2 lingo are returned here
- * @return APR_SUCCESS if all went well
- */
-apr_status_t h2_session_start(h2_session *session, int *rv);
-
-/**
- * Called when an error occured and the session needs to shut down.
- * @param session the session to shut down
- * @param reason  the apache status that caused the shutdown
- * @param rv      the nghttp2 reason for shutdown, set to 0 if you have none.
- *
+ * Called when a serious error occured and the session needs to terminate
+ * without further connection io.
+ * @param session the session to abort
+ * @param reason  the apache status that caused the abort
  */
-apr_status_t h2_session_abort(h2_session *session, apr_status_t reason, int rv);
+void h2_session_abort(h2_session *session, apr_status_t reason);
 
 /**
- * Called before a session gets destroyed, might flush output etc. 
+ * Close and deallocate the given session.
  */
-apr_status_t h2_session_close(h2_session *session);
+void h2_session_close(h2_session *session);
 
 /* Start submitting the response to a stream request. This is possible
  * once we have all the response headers. */



Mime
View raw message