httpd-cvs mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j..@apache.org
Subject svn commit: r1688474 [6/21] - in /httpd/httpd/trunk/modules/http2: ./ m4/ mod-h2.xcodeproj/ mod-h2.xcodeproj/project.xcworkspace/ mod-h2.xcodeproj/project.xcworkspace/xcshareddata/ mod-h2.xcodeproj/xcuserdata/ mod-h2.xcodeproj/xcuserdata/sei.xcuserdata...
Date Tue, 30 Jun 2015 15:26:19 GMT
Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_request.c
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_request.c?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_request.c (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_request.c Tue Jun 30 15:26:16 2015
@@ -0,0 +1,176 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <assert.h>
+
+#include <apr_strings.h>
+
+#include <httpd.h>
+#include <http_core.h>
+#include <http_config.h>
+#include <http_log.h>
+
+#include "h2_private.h"
+#include "h2_mplx.h"
+#include "h2_to_h1.h"
+#include "h2_request.h"
+#include "h2_task.h"
+#include "h2_util.h"
+
+
+h2_request *h2_request_create(int id, apr_pool_t *pool, 
+                              apr_bucket_alloc_t *bucket_alloc)
+{
+    h2_request *req = apr_pcalloc(pool, sizeof(h2_request));
+    if (req) {
+        req->id = id;
+        req->pool = pool;
+        req->bucket_alloc = bucket_alloc;
+    }
+    return req;
+}
+
+void h2_request_destroy(h2_request *req)
+{
+    if (req->to_h1) {
+        h2_to_h1_destroy(req->to_h1);
+        req->to_h1 = NULL;
+    }
+}
+
+static apr_status_t insert_request_line(h2_request *req, h2_mplx *m);
+
+apr_status_t h2_request_rwrite(h2_request *req, request_rec *r, h2_mplx *m)
+{
+    req->method = r->method;
+    req->path = r->uri;
+    req->authority = r->hostname;
+    if (!strchr(req->authority, ':') && r->parsed_uri.port_str) {
+        req->authority = apr_psprintf(req->pool, "%s:%s", req->authority,
+                                      r->parsed_uri.port_str);
+    }
+    req->scheme = NULL;
+    
+    
+    apr_status_t status = insert_request_line(req, m);
+    if (status == APR_SUCCESS) {
+        status = h2_to_h1_add_headers(req->to_h1, r->headers_in);
+    }
+
+    ap_log_rerror(APLOG_MARK, APLOG_DEBUG, status, r,
+                  "h2_request(%d): written request %s %s, host=%s",
+                  req->id, req->method, req->path, req->authority);
+    
+    return status;
+}
+
+apr_status_t h2_request_write_header(h2_request *req,
+                                     const char *name, size_t nlen,
+                                     const char *value, size_t vlen,
+                                     h2_mplx *m)
+{
+    apr_status_t status = APR_SUCCESS;
+    
+    if (nlen <= 0) {
+        return status;
+    }
+    
+    if (name[0] == ':') {
+        /* pseudo header, see ch. 8.1.2.3, always should come first */
+        if (req->to_h1) {
+            ap_log_perror(APLOG_MARK, APLOG_ERR, 0, req->pool,
+                          "h2_request(%d): pseudo header after request start",
+                          req->id);
+            return APR_EGENERAL;
+        }
+        
+        if (H2_HEADER_METHOD_LEN == nlen
+            && !strncmp(H2_HEADER_METHOD, name, nlen)) {
+            req->method = apr_pstrndup(req->pool, value, vlen);
+        }
+        else if (H2_HEADER_SCHEME_LEN == nlen
+                 && !strncmp(H2_HEADER_SCHEME, name, nlen)) {
+            req->scheme = apr_pstrndup(req->pool, value, vlen);
+        }
+        else if (H2_HEADER_PATH_LEN == nlen
+                 && !strncmp(H2_HEADER_PATH, name, nlen)) {
+            req->path = apr_pstrndup(req->pool, value, vlen);
+        }
+        else if (H2_HEADER_AUTH_LEN == nlen
+                 && !strncmp(H2_HEADER_AUTH, name, nlen)) {
+            req->authority = apr_pstrndup(req->pool, value, vlen);
+        }
+        else {
+            char buffer[32];
+            memset(buffer, 0, 32);
+            strncpy(buffer, name, (nlen > 31)? 31 : nlen);
+            ap_log_perror(APLOG_MARK, APLOG_WARNING, 0, req->pool,
+                          "h2_request(%d): ignoring unknown pseudo header %s",
+                          req->id, buffer);
+        }
+    }
+    else {
+        /* non-pseudo header, append to work bucket of stream */
+        if (!req->to_h1) {
+            status = insert_request_line(req, m);
+            if (status != APR_SUCCESS) {
+                return status;
+            }
+        }
+        
+        if (status == APR_SUCCESS) {
+            status = h2_to_h1_add_header(req->to_h1,
+                                         name, nlen, value, vlen);
+        }
+    }
+    
+    return status;
+}
+
+apr_status_t h2_request_write_data(h2_request *req,
+                                   const char *data, size_t len)
+{
+    return h2_to_h1_add_data(req->to_h1, data, len);
+}
+
+apr_status_t h2_request_end_headers(h2_request *req, struct h2_mplx *m,
+                                    h2_task *task, int eos)
+{
+    if (!req->to_h1) {
+        apr_status_t status = insert_request_line(req, m);
+        if (status != APR_SUCCESS) {
+            return status;
+        }
+    }
+    return h2_to_h1_end_headers(req->to_h1, task, eos);
+}
+
+apr_status_t h2_request_close(h2_request *req)
+{
+    return h2_to_h1_close(req->to_h1);
+}
+
+static apr_status_t insert_request_line(h2_request *req, h2_mplx *m)
+{
+    req->to_h1 = h2_to_h1_create(req->id, req->pool, req->bucket_alloc, 
+                                 req->method, req->path, req->authority, m);
+    return req->to_h1? APR_SUCCESS : APR_ENOMEM;
+}
+
+apr_status_t h2_request_flush(h2_request *req)
+{
+    return h2_to_h1_flush(req->to_h1);
+}
+

Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_request.h
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_request.h?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_request.h (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_request.h Tue Jun 30 15:26:16 2015
@@ -0,0 +1,67 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef __mod_h2__h2_request__
+#define __mod_h2__h2_request__
+
+/* h2_request is the transformer of HTTP2 streams into HTTP/1.1 internal
+ * format that will be fed to various httpd input filters to finally
+ * become a request_rec to be handled by soemone.
+ *
+ * Ideally, we would make a request_rec without serializing the headers
+ * we have only to make someone else parse them back.
+ */
+struct h2_to_h1;
+struct h2_mplx;
+struct h2_task;
+
+typedef struct h2_request h2_request;
+
+struct h2_request {
+    int id;                 /* http2 stream id */
+    apr_pool_t *pool;
+    apr_bucket_alloc_t *bucket_alloc;
+    struct h2_to_h1 *to_h1; /* Converter to HTTP/1.1 format*/
+    
+    /* pseudo header values, see ch. 8.1.2.3 */
+    const char *method;
+    const char *path;
+    const char *authority;
+    const char *scheme;
+};
+
+h2_request *h2_request_create(int id, apr_pool_t *pool, 
+                              apr_bucket_alloc_t *bucket_alloc);
+void h2_request_destroy(h2_request *req);
+
+apr_status_t h2_request_flush(h2_request *req);
+
+apr_status_t h2_request_write_header(h2_request *req,
+                                     const char *name, size_t nlen,
+                                     const char *value, size_t vlen,
+                                     struct h2_mplx *m);
+
+apr_status_t h2_request_write_data(h2_request *request,
+                                   const char *data, size_t len);
+
+apr_status_t h2_request_end_headers(h2_request *req, struct h2_mplx *m, 
+                                    struct h2_task *task, int eos);
+
+apr_status_t h2_request_close(h2_request *req);
+
+apr_status_t h2_request_rwrite(h2_request *req, request_rec *r,
+                               struct h2_mplx *m);
+
+#endif /* defined(__mod_h2__h2_request__) */

Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_response.c
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_response.c?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_response.c (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_response.c Tue Jun 30 15:26:16 2015
@@ -0,0 +1,235 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <assert.h>
+#include <stdio.h>
+
+#include <apr_strings.h>
+
+#include <httpd.h>
+#include <http_core.h>
+#include <http_log.h>
+
+#include <nghttp2/nghttp2.h>
+
+#include "h2_private.h"
+#include "h2_util.h"
+#include "h2_response.h"
+
+static void convert_header(h2_response *response, apr_table_t *headers,
+                           const char *http_status, request_rec *r);
+static int ignore_header(const char *name) 
+{
+    return (H2_HD_MATCH_LIT_CS("connection", name)
+            || H2_HD_MATCH_LIT_CS("proxy-connection", name)
+            || H2_HD_MATCH_LIT_CS("upgrade", name)
+            || H2_HD_MATCH_LIT_CS("keep-alive", name)
+            || H2_HD_MATCH_LIT_CS("transfer-encoding", name));
+}
+
+h2_response *h2_response_create(int stream_id,
+                                const char *http_status,
+                                apr_array_header_t *hlines,
+                                apr_pool_t *pool)
+{
+    apr_table_t *header;
+    h2_response *response = apr_pcalloc(pool, sizeof(h2_response));
+    if (response == NULL) {
+        return NULL;
+    }
+    
+    response->stream_id = stream_id;
+    response->content_length = -1;
+    
+    if (hlines) {
+        header = apr_table_make(pool, hlines->nelts);        
+        for (int i = 0; i < hlines->nelts; ++i) {
+            char *hline = ((char **)hlines->elts)[i];
+            char *sep = strchr(hline, ':');
+            if (!sep) {
+                ap_log_perror(APLOG_MARK, APLOG_WARNING, APR_EINVAL, pool,
+                              "h2_response(%d): invalid header[%d] '%s'",
+                              response->stream_id, i, (char*)hline);
+                /* not valid format, abort */
+                return NULL;
+            }
+            (*sep++) = '\0';
+            while (*sep == ' ' || *sep == '\t') {
+                ++sep;
+            }
+            if (ignore_header(hline)) {
+                /* never forward, ch. 8.1.2.2 */
+            }
+            else {
+                apr_table_merge(header, hline, sep);
+                if (*sep && H2_HD_MATCH_LIT_CS("content-length", hline)) {
+                    char *end;
+                    response->content_length = apr_strtoi64(sep, &end, 10);
+                    if (sep == end) {
+                        ap_log_perror(APLOG_MARK, APLOG_WARNING, APR_EINVAL, 
+                                      pool, "h2_response(%d): content-length"
+                                      " value not parsed: %s", 
+                                      response->stream_id, sep);
+                        response->content_length = -1;
+                    }
+                }
+            }
+        }
+    }
+    else {
+        header = apr_table_make(pool, 0);        
+    }
+    
+    convert_header(response, header, http_status, NULL);
+    return response->headers? response : NULL;
+}
+
+h2_response *h2_response_rcreate(int stream_id, request_rec *r,
+                                 apr_table_t *header, apr_pool_t *pool)
+{
+    h2_response *response = apr_pcalloc(pool, sizeof(h2_response));
+    if (response == NULL) {
+        return NULL;
+    }
+    
+    response->stream_id = stream_id;
+    response->content_length = -1;
+    convert_header(response, header, apr_psprintf(pool, "%d", r->status), r);
+    
+    return response->headers? response : NULL;
+}
+
+void h2_response_cleanup(h2_response *response)
+{
+    if (response->headers) {
+        if (--response->headers->refs == 0) {
+            free(response->headers);
+        }
+        response->headers = NULL;
+    }
+}
+
+void h2_response_destroy(h2_response *response)
+{
+    h2_response_cleanup(response);
+}
+
+void h2_response_copy(h2_response *to, h2_response *from)
+{
+    h2_response_cleanup(to);
+    *to = *from;
+    if (from->headers) {
+        ++from->headers->refs;
+    }
+}
+
+typedef struct {
+    nghttp2_nv *nv;
+    size_t nvlen;
+    size_t nvstrlen;
+    size_t offset;
+    char *strbuf;
+    h2_response *response;
+    int debug;
+    request_rec *r;
+} nvctx_t;
+
+static int count_headers(void *ctx, const char *key, const char *value)
+{
+    if (!ignore_header(key)) {
+        nvctx_t *nvctx = (nvctx_t*)ctx;
+        nvctx->nvlen++;
+        nvctx->nvstrlen += strlen(key) + strlen(value) + 2;
+    }
+    return 1;
+}
+
+#define NV_ADD_LIT_CS(nv, k, v)     addnv_lit_cs(nv, k, sizeof(k) - 1, v, strlen(v))
+#define NV_ADD_CS_CS(nv, k, v)      addnv_cs_cs(nv, k, strlen(k), v, strlen(v))
+#define NV_BUF_ADD(nv, s, len)      memcpy(nv->strbuf, s, len); \
+s = nv->strbuf; \
+nv->strbuf += len + 1
+
+static void addnv_cs_cs(nvctx_t *ctx, const char *key, size_t key_len,
+                        const char *value, size_t val_len)
+{
+    nghttp2_nv *nv = &ctx->nv[ctx->offset];
+    
+    NV_BUF_ADD(ctx, key, key_len);
+    NV_BUF_ADD(ctx, value, val_len);
+    
+    nv->name = (uint8_t*)key;
+    nv->namelen = key_len;
+    nv->value = (uint8_t*)value;
+    nv->valuelen = val_len;
+    
+    ctx->offset++;
+}
+
+static void addnv_lit_cs(nvctx_t *ctx, const char *key, size_t key_len,
+                         const char *value, size_t val_len)
+{
+    nghttp2_nv *nv = &ctx->nv[ctx->offset];
+    
+    NV_BUF_ADD(ctx, value, val_len);
+    
+    nv->name = (uint8_t*)key;
+    nv->namelen = key_len;
+    nv->value = (uint8_t*)value;
+    nv->valuelen = val_len;
+    
+    ctx->offset++;
+}
+
+static int add_header(void *ctx, const char *key, const char *value)
+{
+    if (!ignore_header(key)) {
+        nvctx_t *nvctx = (nvctx_t*)ctx;
+        if (nvctx->debug) {
+            ap_log_rerror(APLOG_MARK, APLOG_WARNING, APR_EINVAL, 
+                          nvctx->r, "h2_response(%d) header -> %s: %s",
+                          nvctx->response->stream_id, key, value);
+        }
+        NV_ADD_CS_CS(ctx, key, value);
+    }
+    return 1;
+}
+
+static void convert_header(h2_response *response, apr_table_t *headers,
+                           const char *status, request_rec *r)
+{
+    nvctx_t ctx = { NULL, 1, strlen(status) + 1, 0, NULL, 
+        response, r? APLOGrdebug(r) : 0, r };
+    
+    apr_table_do(count_headers, &ctx, headers, NULL);
+    
+    size_t n =  (sizeof(h2_headers)
+                 + (ctx.nvlen * sizeof(nghttp2_nv)) + ctx.nvstrlen); 
+    h2_headers *h = calloc(1, n);
+    if (h) {
+        ctx.nv = (nghttp2_nv*)(h + 1);
+        ctx.strbuf = (char*)&ctx.nv[ctx.nvlen];
+        
+        NV_ADD_LIT_CS(&ctx, ":status", status);
+        apr_table_do(add_header, &ctx, headers, NULL);
+        
+        h->nv = ctx.nv;
+        h->nvlen = ctx.nvlen;
+        h->status = (const char *)ctx.nv[0].value;
+        h->refs = 1;
+        response->headers = h;
+    }
+}
+

Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_response.h
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_response.h?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_response.h (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_response.h Tue Jun 30 15:26:16 2015
@@ -0,0 +1,48 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef __mod_h2__h2_response__
+#define __mod_h2__h2_response__
+
+/* h2_response is just the data belonging the the head of a HTTP response,
+ * suitable prepared to be fed to nghttp2 for response submit. 
+ */
+typedef struct h2_headers {
+    nghttp2_nv *nv;
+    apr_size_t nvlen;
+    const char *status;
+    volatile int refs;
+} h2_headers;
+
+typedef struct h2_response {
+    int stream_id;
+    apr_off_t content_length;
+    h2_headers *headers;
+} h2_response;
+
+h2_response *h2_response_create(int stream_id,
+                                  const char *http_status,
+                                  apr_array_header_t *hlines,
+                                  apr_pool_t *pool);
+
+h2_response *h2_response_rcreate(int stream_id, request_rec *r,
+                                 apr_table_t *header, apr_pool_t *pool);
+
+void h2_response_cleanup(h2_response *response);
+void h2_response_destroy(h2_response *response);
+
+void h2_response_copy(h2_response *to, h2_response *from);
+
+#endif /* defined(__mod_h2__h2_response__) */

Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_session.c
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_session.c?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_session.c (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_session.c Tue Jun 30 15:26:16 2015
@@ -0,0 +1,1225 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <assert.h>
+#include <apr_thread_cond.h>
+#include <apr_base64.h>
+#include <apr_strings.h>
+
+#include <httpd.h>
+#include <http_core.h>
+#include <http_config.h>
+#include <http_log.h>
+
+#include "h2_private.h"
+#include "h2_config.h"
+#include "h2_h2.h"
+#include "h2_mplx.h"
+#include "h2_response.h"
+#include "h2_stream.h"
+#include "h2_stream_set.h"
+#include "h2_from_h1.h"
+#include "h2_task.h"
+#include "h2_session.h"
+#include "h2_util.h"
+#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)
+{
+    switch (rv) {
+        case APR_SUCCESS:
+            return NGHTTP2_NO_ERROR;
+        case APR_EAGAIN:
+        case APR_TIMEUP:
+            return NGHTTP2_ERR_WOULDBLOCK;
+        case APR_EOF:
+            return NGHTTP2_ERR_EOF;
+        default:
+            return NGHTTP2_ERR_PROTO;
+    }
+}
+
+static int stream_open(h2_session *session, int stream_id)
+{
+    if (session->aborted) {
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    
+    h2_stream * stream = h2_mplx_open_io(session->mplx, stream_id);
+    if (stream) {
+        h2_stream_set_add(session->streams, stream);
+        
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                      "h2_session: stream(%ld-%d): opened",
+                      session->id, stream_id);
+        
+        return 0;
+    }
+    
+    ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_ENOMEM, session->c,
+                  "h2_session: stream(%ld-%d): unable to create",
+                  session->id, stream_id);
+    return NGHTTP2_ERR_INVALID_STREAM_ID;
+}
+
+static apr_status_t stream_end_headers(h2_session *session,
+                                       h2_stream *stream, int eos)
+{
+    (void)session;
+    return h2_stream_write_eoh(stream, eos);
+}
+
+static apr_status_t send_data(h2_session *session, const char *data, 
+                              apr_size_t length);
+
+/*
+ * Callback when nghttp2 wants to send bytes back to the client.
+ */
+static ssize_t send_cb(nghttp2_session *ngh2,
+                       const uint8_t *data, size_t length,
+                       int flags, void *userp)
+{
+    h2_session *session = (h2_session *)userp;
+    apr_status_t status = send_data(session, (const char *)data, length);
+    (void)ngh2; (void)flags;
+    
+    if (status == APR_SUCCESS) {
+        return length;
+    }
+    if (status == APR_EAGAIN || status == APR_TIMEUP) {
+        return NGHTTP2_ERR_WOULDBLOCK;
+    }
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c,
+                  "h2_session: send error");
+    return h2_session_status_from_apr_status(status);
+}
+
+static int on_invalid_frame_recv_cb(nghttp2_session *ngh2,
+                                    const nghttp2_frame *frame,
+                                    nghttp2_error error, void *userp)
+{
+    h2_session *session = (h2_session *)userp;
+    (void)ngh2;
+    
+    if (session->aborted) {
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    if (APLOGctrace2(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",
+                      (int)error, buffer);
+    }
+    return 0;
+}
+
+static int on_data_chunk_recv_cb(nghttp2_session *ngh2, uint8_t flags,
+                                 int32_t stream_id,
+                                 const uint8_t *data, size_t len, void *userp)
+{
+    int rv;
+    h2_session *session = (h2_session *)userp;
+    (void)flags;
+    
+    if (session->aborted) {
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    h2_stream * stream = h2_stream_set_get(session->streams, stream_id);
+    if (!stream) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, session->c,
+                      "h2_session:  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);
+        if (nghttp2_is_fatal(rv)) {
+            return NGHTTP2_ERR_CALLBACK_FAILURE;
+        }
+        return 0;
+    }
+    
+    apr_status_t status = h2_stream_write_data(stream, (const char *)data, len);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, session->c,
+                  "h2_stream(%ld-%d): written DATA, length %d",
+                  session->id, stream_id, (int)len);
+    if (status != APR_SUCCESS) {
+        rv = nghttp2_submit_rst_stream(ngh2, NGHTTP2_FLAG_NONE, stream_id,
+                                       NGHTTP2_INTERNAL_ERROR);
+        if (nghttp2_is_fatal(rv)) {
+            return NGHTTP2_ERR_CALLBACK_FAILURE;
+        }
+    }
+    return 0;
+}
+
+static int before_frame_send_cb(nghttp2_session *ngh2,
+                                const nghttp2_frame *frame,
+                                void *userp)
+{
+    h2_session *session = (h2_session *)userp;
+    (void)ngh2;
+
+    if (session->aborted) {
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    if (APLOGctrace2(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): before_frame_send %s", 
+                      session->id, buffer);
+    }
+    return 0;
+}
+
+static int on_frame_send_cb(nghttp2_session *ngh2,
+                            const nghttp2_frame *frame,
+                            void *userp)
+{
+    h2_session *session = (h2_session *)userp;
+    (void)ngh2; (void)frame;
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
+                  "h2_session(%ld): on_frame_send", session->id);
+    return 0;
+}
+
+static int on_frame_not_send_cb(nghttp2_session *ngh2,
+                                const nghttp2_frame *frame,
+                                int lib_error_code, void *userp)
+{
+    h2_session *session = (h2_session *)userp;
+    (void)ngh2;
+    
+    if (APLOGctrace2(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: callback on_frame_not_send error=%d %s",
+                      lib_error_code, buffer);
+    }
+    return 0;
+}
+
+static apr_status_t stream_destroy(h2_session *session, h2_stream *stream) 
+{
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                  "h2_stream(%ld-%d): closing", session->id, (int)stream->id);
+    h2_stream_set_remove(session->streams, stream);
+    return h2_mplx_cleanup_stream(session->mplx, stream);
+}
+
+static int on_stream_close_cb(nghttp2_session *ngh2, int32_t stream_id,
+                              uint32_t error_code, void *userp)
+{
+    h2_session *session = (h2_session *)userp;
+    (void)ngh2;
+    
+    if (session->aborted) {
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    h2_stream *stream = h2_stream_set_get(session->streams, stream_id);
+    if (stream) {
+        stream_destroy(session, stream);
+    }
+    
+    if (error_code) {
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                      "h2_stream(%ld-%d): close error %d",
+                      session->id, (int)stream_id, error_code);
+    }
+    
+    return 0;
+}
+
+static int on_begin_headers_cb(nghttp2_session *ngh2,
+                               const nghttp2_frame *frame, void *userp)
+{
+    (void)ngh2;
+    /* This starts a new stream. */
+    int rv = stream_open((h2_session *)userp, frame->hd.stream_id);
+    if (rv != NGHTTP2_ERR_CALLBACK_FAILURE) {
+      /* on_header_cb or on_frame_recv_cb will dectect that stream
+         does not exist and submit RST_STREAM. */
+      return 0;
+    }
+    return NGHTTP2_ERR_CALLBACK_FAILURE;
+}
+
+static int on_header_cb(nghttp2_session *ngh2, const nghttp2_frame *frame,
+                        const uint8_t *name, size_t namelen,
+                        const uint8_t *value, size_t valuelen,
+                        uint8_t flags,
+                        void *userp)
+{
+    h2_session *session = (h2_session *)userp;
+    (void)ngh2; (void)flags;
+    
+    if (session->aborted) {
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    h2_stream * stream = h2_stream_set_get(session->streams,
+                                           frame->hd.stream_id);
+    if (!stream) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, session->c,
+                      "h2_session:  stream(%ld-%d): on_header for unknown stream",
+                      session->id, (int)frame->hd.stream_id);
+        return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+    }
+    
+    apr_status_t status = h2_stream_write_header(stream,
+                                               (const char *)name, namelen,
+                                               (const char *)value, valuelen);
+    if (status != APR_SUCCESS) {
+        return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+    }
+    return 0;
+}
+
+/**
+ * nghttp2 session has received a complete frame. Most, it uses
+ * for processing of internal state. HEADER and DATA frames however
+ * we need to handle ourself.
+ */
+static int on_frame_recv_cb(nghttp2_session *ng2s,
+                            const nghttp2_frame *frame,
+                            void *userp)
+{
+    int rv;
+    h2_session *session = (h2_session *)userp;
+    if (session->aborted) {
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    apr_status_t status = APR_SUCCESS;
+    
+    ++session->frames_received;
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                  "h2_session(%ld): on_frame_rcv #%ld, type=%d", session->id,
+                  (long)session->frames_received, frame->hd.type);
+    switch (frame->hd.type) {
+        case NGHTTP2_HEADERS: {
+            h2_stream * stream = h2_stream_set_get(session->streams,
+                                                   frame->hd.stream_id);
+            if (stream == NULL) {
+                ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, session->c,
+                              "h2_session:  stream(%ld-%d): HEADERS frame "
+                              "for unknown stream", session->id,
+                              (int)frame->hd.stream_id);
+                rv = nghttp2_submit_rst_stream(ng2s, NGHTTP2_FLAG_NONE,
+                                               frame->hd.stream_id,
+                                               NGHTTP2_INTERNAL_ERROR);
+                if (nghttp2_is_fatal(rv)) {
+                    return NGHTTP2_ERR_CALLBACK_FAILURE;
+                }
+                return 0;
+            }
+
+            int eos = (frame->hd.flags & NGHTTP2_FLAG_END_STREAM);
+            status = stream_end_headers(session, stream, eos);
+
+            break;
+        }
+        case NGHTTP2_DATA: {
+            h2_stream * stream = h2_stream_set_get(session->streams,
+                                                   frame->hd.stream_id);
+            if (stream == NULL) {
+                ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, session->c,
+                              "h2_session:  stream(%ld-%d): DATA frame "
+                              "for unknown stream", session->id,
+                              (int)frame->hd.stream_id);
+                rv = nghttp2_submit_rst_stream(ng2s, NGHTTP2_FLAG_NONE,
+                                               frame->hd.stream_id,
+                                               NGHTTP2_INTERNAL_ERROR);
+                if (nghttp2_is_fatal(rv)) {
+                    return NGHTTP2_ERR_CALLBACK_FAILURE;
+                }
+                return 0;
+            }
+            break;
+        }
+        case NGHTTP2_PRIORITY: {
+            ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                          "h2_session:  stream(%ld-%d): PRIORITY frame "
+                          " weight=%d, dependsOn=%d, exclusive=%d", 
+                          session->id, (int)frame->hd.stream_id,
+                          frame->priority.pri_spec.weight,
+                          frame->priority.pri_spec.stream_id,
+                          frame->priority.pri_spec.exclusive);
+            break;
+        }
+        default:
+            if (APLOGctrace2(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: on_frame_rcv %s", buffer);
+            }
+            break;
+    }
+
+    /* only DATA and HEADERS frame can bear END_STREAM flag.  Other
+       frame types may have other flag which has the same value, so we
+       have to check the frame type first.  */
+    if ((frame->hd.type == NGHTTP2_DATA || frame->hd.type == NGHTTP2_HEADERS) &&
+        frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
+        h2_stream * stream = h2_stream_set_get(session->streams,
+                                               frame->hd.stream_id);
+        if (stream != NULL) {
+            status = h2_stream_write_eos(stream);
+            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c,
+                          "h2_stream(%ld-%d): input closed",
+                          session->id, (int)frame->hd.stream_id);
+        }
+    }
+    
+    if (status != APR_SUCCESS) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, status, session->c,
+                      "h2_session: stream(%ld-%d): error handling frame",
+                      session->id, (int)frame->hd.stream_id);
+        rv = nghttp2_submit_rst_stream(ng2s, NGHTTP2_FLAG_NONE,
+                                       frame->hd.stream_id,
+                                       NGHTTP2_INTERNAL_ERROR);
+        if (nghttp2_is_fatal(rv)) {
+            return NGHTTP2_ERR_CALLBACK_FAILURE;
+        }
+        return 0;
+    }
+    
+    return 0;
+}
+
+static apr_status_t send_data(h2_session *session, const char *data, 
+                              apr_size_t length)
+{
+    return h2_conn_io_write(&session->io, data, length);
+}
+
+#if NGHTTP2_HAS_DATA_CB
+
+static apr_status_t pass_data(void *ctx, 
+                              const char *data, apr_size_t length)
+{
+    return send_data((h2_session*)ctx, data, length);
+}
+
+int on_send_data_cb(nghttp2_session *ngh2, 
+                    nghttp2_frame *frame, 
+                    const uint8_t *framehd, 
+                    size_t length, 
+                    nghttp2_data_source *source, 
+                    void *userp)
+{
+    apr_status_t status = APR_SUCCESS;
+    h2_session *session = (h2_session *)userp;
+    int stream_id = (int)frame->hd.stream_id;
+    const unsigned char padlen = frame->data.padlen;
+    apr_size_t frame_len = 9 + (padlen? 1 : 0) + length + padlen;
+    apr_size_t data_len = length;
+    apr_size_t avail, chunk_len;
+    int rv = 0;
+    int eos;
+    h2_stream *stream;
+    
+    if (session->aborted) {
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    
+    stream = h2_stream_set_get(session->streams, stream_id);
+    if (!stream) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_NOTFOUND, session->c,
+                      "h2_stream(%ld-%d): send_data",
+                      session->id, (int)stream_id);
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    
+    status = send_data(session, (const char *)framehd, 9);
+    if (status == APR_SUCCESS) {
+        if (padlen) {
+            status = send_data(session, (const char *)&padlen, 1);
+        }
+
+        if (status == APR_SUCCESS) {
+            apr_size_t len = length;
+            int eos = 0;
+            status = h2_stream_readx(stream, pass_data, session, 
+                                     &len, &eos);
+            if (status == APR_SUCCESS && len != length) {
+                status = APR_EINVAL;
+            }
+        }
+        
+        if (status == APR_SUCCESS && padlen) {
+            if (padlen) {
+                char pad[256];
+                memset(pad, 0, padlen);
+                status = send_data(session, pad, padlen);
+            }
+        }
+    }
+    
+    if (status == APR_SUCCESS) {
+        return 0;
+    }
+    else if (status != APR_EOF) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, status, session->c,
+                      "h2_stream(%ld-%d): failed send_data_cb",
+                      session->id, (int)stream_id);
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    
+    return h2_session_status_from_apr_status(status);
+}
+
+#endif
+
+
+#define NGH2_SET_CALLBACK(callbacks, name, fn)\
+nghttp2_session_callbacks_set_##name##_callback(callbacks, fn)
+
+static apr_status_t init_callbacks(conn_rec *c, nghttp2_session_callbacks **pcb)
+{
+    int rv = nghttp2_session_callbacks_new(pcb);
+    if (rv != 0) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, c,
+                      "nghttp2_session_callbacks_new: %s",
+                      nghttp2_strerror(rv));
+        return APR_EGENERAL;
+    }
+    
+    NGH2_SET_CALLBACK(*pcb, send, send_cb);
+    NGH2_SET_CALLBACK(*pcb, on_frame_recv, on_frame_recv_cb);
+    NGH2_SET_CALLBACK(*pcb, on_invalid_frame_recv, on_invalid_frame_recv_cb);
+    NGH2_SET_CALLBACK(*pcb, on_data_chunk_recv, on_data_chunk_recv_cb);
+    NGH2_SET_CALLBACK(*pcb, before_frame_send, before_frame_send_cb);
+    NGH2_SET_CALLBACK(*pcb, on_frame_send, on_frame_send_cb);
+    NGH2_SET_CALLBACK(*pcb, on_frame_not_send, on_frame_not_send_cb);
+    NGH2_SET_CALLBACK(*pcb, on_stream_close, on_stream_close_cb);
+    NGH2_SET_CALLBACK(*pcb, on_begin_headers, on_begin_headers_cb);
+    NGH2_SET_CALLBACK(*pcb, on_header, on_header_cb);
+#if NGHTTP2_HAS_DATA_CB
+    NGH2_SET_CALLBACK(*pcb, send_data, on_send_data_cb);
+#endif
+    
+    return APR_SUCCESS;
+}
+
+static h2_session *h2_session_create_int(conn_rec *c,
+                                         request_rec *r,
+                                         h2_config *config, 
+                                         h2_workers *workers)
+{
+    nghttp2_session_callbacks *callbacks = NULL;
+    nghttp2_option *options = NULL;
+
+    apr_pool_t *pool = NULL;
+    apr_status_t status = apr_pool_create(&pool, r? r->pool : c->pool);
+    if (status != APR_SUCCESS) {
+        return NULL;
+    }
+
+    h2_session *session = apr_pcalloc(pool, sizeof(h2_session));
+    if (session) {
+        session->id = c->id;
+        session->c = c;
+        session->r = r;
+        
+        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->pool = pool;
+        
+        status = apr_thread_cond_create(&session->iowait, session->pool);
+        if (status != APR_SUCCESS) {
+            return NULL;
+        }
+        
+        session->streams = h2_stream_set_create(session->pool);
+        
+        session->workers = workers;
+        session->mplx = h2_mplx_create(c, session->pool, workers);
+        
+        h2_conn_io_init(&session->io, c);
+        session->bbtmp = apr_brigade_create(session->pool, c->bucket_alloc);
+        
+        status = init_callbacks(c, &callbacks);
+        if (status != APR_SUCCESS) {
+            ap_log_cerror(APLOG_MARK, APLOG_ERR, status, c,
+                          "nghttp2: error in init_callbacks");
+            h2_session_destroy(session);
+            return NULL;
+        }
+        
+        int rv = nghttp2_option_new(&options);
+        if (rv != 0) {
+            ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_EGENERAL, c,
+                          "nghttp2_option_new: %s", nghttp2_strerror(rv));
+            h2_session_destroy(session);
+            return NULL;
+        }
+
+        nghttp2_option_set_peer_max_concurrent_streams(options, 
+                                                       (uint32_t)session->max_stream_count);
+
+        /* We need to handle window updates ourself, otherwise we
+         * get flooded by nghttp2. */
+        nghttp2_option_set_no_auto_window_update(options, 1);
+        
+        rv = nghttp2_session_server_new2(&session->ngh2, callbacks,
+                                         session, options);
+        nghttp2_session_callbacks_del(callbacks);
+        nghttp2_option_del(options);
+        
+        if (rv != 0) {
+            ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_EGENERAL, c,
+                          "nghttp2_session_server_new: %s",
+                          nghttp2_strerror(rv));
+            h2_session_destroy(session);
+            return NULL;
+        }
+        
+    }
+    return session;
+}
+
+h2_session *h2_session_create(conn_rec *c, h2_config *config, 
+                              h2_workers *workers)
+{
+    return h2_session_create_int(c, NULL, config, workers);
+}
+
+h2_session *h2_session_rcreate(request_rec *r, h2_config *config, 
+                               h2_workers *workers)
+{
+    return h2_session_create_int(r->connection, r, config, workers);
+}
+
+void h2_session_destroy(h2_session *session)
+{
+    AP_DEBUG_ASSERT(session);
+    if (session->mplx) {
+        h2_mplx_release_and_join(session->mplx, session->iowait);
+        session->mplx = NULL;
+    }
+    if (session->streams) {
+        if (h2_stream_set_size(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->ngh2) {
+        nghttp2_session_del(session->ngh2);
+        session->ngh2 = NULL;
+    }
+    h2_conn_io_destroy(&session->io);
+    
+    if (session->iowait) {
+        apr_thread_cond_destroy(session->iowait);
+        session->iowait = NULL;
+    }
+    
+    if (session->pool) {
+        apr_pool_destroy(session->pool);
+    }
+}
+
+apr_status_t h2_session_goaway(h2_session *session, apr_status_t reason)
+{
+    AP_DEBUG_ASSERT(session);
+    apr_status_t status = APR_SUCCESS;
+    if (session->aborted) {
+        return APR_EINVAL;
+    }
+    
+    int rv = 0;
+    if (reason == APR_SUCCESS) {
+        rv = nghttp2_submit_shutdown_notice(session->ngh2);
+    }
+    else {
+        int err = 0;
+        int last_id = nghttp2_session_get_last_proc_stream_id(session->ngh2);
+        rv = nghttp2_submit_goaway(session->ngh2, last_id,
+                                   NGHTTP2_FLAG_NONE, err, NULL, 0);
+    }
+    if (rv != 0) {
+        status = APR_EGENERAL;
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, status, session->c,
+                      "session(%ld): submit goaway: %s",
+                      session->id, nghttp2_strerror(rv));
+    }
+    return status;
+}
+
+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 (reason) {
+                ap_log_cerror(APLOG_MARK, (reason == NGHTTP2_ERR_EOF)?
+                              APLOG_DEBUG : APLOG_INFO, 0, session->c,
+                              "session(%ld): aborting session, reason=%d %s",
+                              session->id, reason, nghttp2_strerror(reason));
+            }
+            nghttp2_session_terminate_session(session->ngh2, reason);
+            nghttp2_submit_goaway(session->ngh2, 0, 0, reason, NULL, 0);
+            nghttp2_session_send(session->ngh2);
+            h2_conn_io_flush(&session->io);
+        }
+        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_EOF:
+                rv = 0;
+                break;
+            case APR_EBADF:
+            case APR_ECONNABORTED:
+                rv = NGHTTP2_ERR_EOF;
+                break;
+            default:
+                break;
+        }
+    }
+    return h2_session_abort_int(session, rv);
+}
+
+apr_status_t h2_session_start(h2_session *session, int *rv)
+{
+    AP_DEBUG_ASSERT(session);
+    /* Start the conversation by submitting our SETTINGS frame */
+    apr_status_t status = APR_SUCCESS;
+    *rv = 0;
+    h2_config *config = h2_config_get(session->c);
+    if (session->r) {
+        /* better for vhost matching */
+        config = h2_config_rget(session->r);
+        
+        /* 'h2c' mode: we should have a 'HTTP2-Settings' header with
+         * base64 encoded client settings. */
+        const char *s = apr_table_get(session->r->headers_in, "HTTP2-Settings");
+        if (!s) {
+            ap_log_rerror(APLOG_MARK, APLOG_ERR, APR_EINVAL, session->r,
+                          "HTTP2-Settings header missing in request");
+            return APR_EINVAL;
+        }
+        unsigned char *cs = NULL;
+        apr_size_t dlen = h2_util_base64url_decode(&cs, s, session->pool);
+        
+        if (APLOGrdebug(session->r)) {
+            char buffer[128];
+            h2_util_hex_dump(buffer, 128, (char*)cs, dlen);
+            ap_log_rerror(APLOG_MARK, APLOG_DEBUG, 0, session->r,
+                          "upgrading h2c session with HTTP2-Settings: %s -> %s (%d)",
+                          s, buffer, (int)dlen);
+        }
+        
+        *rv = nghttp2_session_upgrade(session->ngh2, (uint8_t*)cs, dlen, NULL);
+        if (*rv != 0) {
+            status = APR_EINVAL;
+            ap_log_rerror(APLOG_MARK, APLOG_ERR, status, session->r,
+                          "nghttp2_session_upgrade: %s", nghttp2_strerror(*rv));
+            return status;
+        }
+        
+        /* Now we need to auto-open stream 1 for the request we got. */
+        *rv = stream_open(session, 1);
+        if (*rv != 0) {
+            status = APR_EGENERAL;
+            ap_log_rerror(APLOG_MARK, APLOG_ERR, status, session->r,
+                          "open stream 1: %s", nghttp2_strerror(*rv));
+            return status;
+        }
+        
+        h2_stream * stream = h2_stream_set_get(session->streams, 1);
+        if (stream == NULL) {
+            status = APR_EGENERAL;
+            ap_log_rerror(APLOG_MARK, APLOG_ERR, status, session->r,
+                          "lookup of stream 1");
+            return status;
+        }
+        
+        status = h2_stream_rwrite(stream, session->r);
+        if (status != APR_SUCCESS) {
+            return status;
+        }
+        status = stream_end_headers(session, stream, 1);
+        if (status != APR_SUCCESS) {
+            return status;
+        }
+    }
+
+    nghttp2_settings_entry settings[] = {
+        { NGHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE,
+            h2_config_geti(config, H2_CONF_MAX_HL_SIZE) },
+        { NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE,
+            h2_config_geti(config, H2_CONF_WIN_SIZE) },
+        {NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 
+            (uint32_t)session->max_stream_count }, 
+    };
+    *rv = nghttp2_submit_settings(session->ngh2, NGHTTP2_FLAG_NONE,
+                                 settings,
+                                 sizeof(settings)/sizeof(settings[0]));
+    if (*rv != 0) {
+        status = APR_EGENERAL;
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, status, session->c,
+                      "nghttp2_submit_settings: %s", nghttp2_strerror(*rv));
+    }
+    
+    return status;
+}
+
+static int h2_session_want_write(h2_session *session)
+{
+    return nghttp2_session_want_write(session->ngh2);
+}
+
+typedef struct {
+    h2_session *session;
+    int resume_count;
+} resume_ctx;
+
+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);
+    AP_DEBUG_ASSERT(stream);
+    
+    if (h2_stream_is_suspended(stream)) {
+        if (h2_mplx_out_has_data_for(stream->m, stream->id)) {
+            h2_stream_set_suspended(stream, 0);
+            ++rctx->resume_count;
+            
+            int rv = nghttp2_session_resume_data(session->ngh2, stream->id);
+            ap_log_cerror(APLOG_MARK, nghttp2_is_fatal(rv)?
+                          APLOG_ERR : APLOG_DEBUG, 0, session->c,
+                          "h2_stream(%ld-%d): resuming stream %s",
+                          session->id, stream->id, nghttp2_strerror(rv));
+        }
+    }
+    return 1;
+}
+
+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) {
+        resume_ctx ctx = { session, 0 };
+        /* Resume all streams where we have data in the out queue and
+         * which had been suspended before. */
+        h2_stream_set_iter(session->streams, resume_on_data, &ctx);
+        return ctx.resume_count;
+    }
+    return 0;
+}
+
+static void update_window(void *ctx, int stream_id, apr_size_t bytes_read)
+{
+    h2_session *session = (h2_session*)ctx;
+    nghttp2_session_consume(session->ngh2, stream_id, bytes_read);
+}
+
+static apr_status_t h2_session_update_windows(h2_session *session)
+{
+    return h2_mplx_in_update_windows(session->mplx, update_window, session);
+}
+
+apr_status_t h2_session_write(h2_session *session, apr_interval_time_t timeout)
+{
+    apr_status_t status = APR_EAGAIN;
+    h2_stream *stream = NULL;
+    int flush_output = 0;
+    
+    AP_DEBUG_ASSERT(session);
+    
+    /* Check that any pending window updates are sent. */
+    status = h2_session_update_windows(session);
+    if (status == APR_SUCCESS) {
+        flush_output = 1;
+    }
+    else if (status != APR_EAGAIN) {
+        return status;
+    }
+    
+    if (h2_session_want_write(session)) {
+        status = APR_SUCCESS;
+        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_int(session, rv);
+                status = APR_ECONNABORTED;
+            }
+        }
+        flush_output = 1;
+    }
+    
+    /* If we have responses ready, submit them now. */
+    while ((stream = h2_mplx_next_submit(session->mplx, 
+                                         session->streams)) != NULL) {
+        status = h2_session_handle_response(session, stream);
+        flush_output = 1;
+    }
+    
+    if (h2_session_resume_streams_with_data(session) > 0) {
+        flush_output = 1;
+    }
+    
+    if (!flush_output && timeout > 0 && !h2_session_want_write(session)) {
+        status = h2_mplx_out_trywait(session->mplx, timeout, session->iowait);
+
+        if (status != APR_TIMEUP
+            && h2_session_resume_streams_with_data(session) > 0) {
+            flush_output = 1;
+        }
+        else {
+            /* nothing happened to ongoing streams, do some house-keeping */
+        }
+    }
+    
+    if (h2_session_want_write(session)) {
+        status = APR_SUCCESS;
+        int rv = nghttp2_session_send(session->ngh2);
+        if (rv != 0) {
+            ap_log_cerror( APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                          "h2_session: send2: %s", nghttp2_strerror(rv));
+            if (nghttp2_is_fatal(rv)) {
+                h2_session_abort_int(session, rv);
+                status = APR_ECONNABORTED;
+            }
+        }
+        flush_output = 1;
+    }
+    
+    if (flush_output) {
+        h2_conn_io_flush(&session->io);
+    }
+    
+    return status;
+}
+
+h2_stream *h2_session_get_stream(h2_session *session, int stream_id)
+{
+    AP_DEBUG_ASSERT(session);
+    return h2_stream_set_get(session->streams, stream_id);
+}
+
+/* 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_read(h2_session *session, apr_read_type_e block)
+{
+    AP_DEBUG_ASSERT(session);
+    return h2_conn_io_read(&session->io, block, session_receive, session);
+}
+
+apr_status_t h2_session_close(h2_session *session)
+{
+    AP_DEBUG_ASSERT(session);
+    return h2_conn_io_flush(&session->io);
+}
+
+/* The session wants to send more DATA for the given stream.
+ */
+static ssize_t stream_data_cb(nghttp2_session *ng2s,
+                              int32_t stream_id,
+                              uint8_t *buf,
+                              size_t length,
+                              uint32_t *data_flags,
+                              nghttp2_data_source *source,
+                              void *puser)
+{
+    h2_session *session = (h2_session *)puser;
+    AP_DEBUG_ASSERT(session);
+    (void)ng2s;(void)source;
+    
+    h2_stream *stream = h2_stream_set_get(session->streams, stream_id);
+    if (!stream) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_NOTFOUND, session->c,
+                      "h2_stream(%ld-%d): data requested but stream not found",
+                      session->id, (int)stream_id);
+        return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    
+    AP_DEBUG_ASSERT(!h2_stream_is_suspended(stream));
+    apr_size_t nread = length;
+    int eos = 0;
+    
+#if NGHTTP2_HAS_DATA_CB
+    apr_status_t status = h2_stream_prep_read(stream, &nread, &eos);
+    if (nread) {
+        *data_flags |=  NGHTTP2_DATA_FLAG_NO_COPY;
+    }
+#else
+    /* Try to pop data buckets from our queue for this stream
+     * until we see EOS or the buffer is full.
+     */
+    apr_status_t status = h2_stream_read(stream, (char*)buf, &nread, &eos);
+#endif
+    
+    switch (status) {
+        case APR_SUCCESS:
+            break;
+            
+        case APR_EAGAIN:
+            /* If there is no data available, our session will automatically
+             * suspend this stream and not ask for more data until we resume
+             * it. Remember at our h2_stream that we need to do this.
+             */
+            nread = 0;
+            h2_stream_set_suspended(stream, 1);
+            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                          "h2_stream(%ld-%d): suspending stream",
+                          session->id, (int)stream_id);
+            return NGHTTP2_ERR_DEFERRED;
+            
+        case APR_EOF:
+            nread = 0;
+            eos = 1;
+            break;
+            
+        default:
+            nread = 0;
+            ap_log_cerror(APLOG_MARK, APLOG_ERR, status, session->c,
+                          "h2_stream(%ld-%d): reading data",
+                          session->id, (int)stream_id);
+            return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    
+    if (eos) {
+        *data_flags |= NGHTTP2_DATA_FLAG_EOF;
+    }
+    
+    return (ssize_t)nread;
+}
+
+typedef struct {
+    nghttp2_nv *nv;
+    size_t nvlen;
+    size_t offset;
+} nvctx_t;
+
+static int submit_response(h2_session *session, h2_response *response)
+{
+    nghttp2_data_provider provider = {
+        (nghttp2_data_source) response->stream_id,
+        (nghttp2_data_source_read_callback) stream_data_cb
+    };
+    
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                  "h2_stream(%ld-%d): submitting response %s",
+                  session->id, response->stream_id, response->headers->status);
+    
+    int rv = nghttp2_submit_response(session->ngh2, response->stream_id,
+                                     response->headers->nv, 
+                                     response->headers->nvlen, &provider);
+    
+    if (rv != 0) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, 0, session->c,
+                      "h2_stream(%ld-%d): submit_response: %s",
+                      session->id, response->stream_id, nghttp2_strerror(rv));
+    }
+    else {
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c,
+                      "h2_stream(%ld-%d): submitted response %s, rv=%d",
+                      session->id, response->stream_id, 
+                      response->headers->status, rv);
+    }
+    return rv;
+}
+
+/* Start submitting the response to a stream request. This is possible
+ * once we have all the response headers. The response body will be
+ * read by the session using the callback we supply.
+ */
+apr_status_t h2_session_handle_response(h2_session *session, h2_stream *stream)
+{
+    AP_DEBUG_ASSERT(session);
+    AP_DEBUG_ASSERT(stream);
+    AP_DEBUG_ASSERT(stream->response);
+    
+    apr_status_t status = APR_SUCCESS;
+    int rv = 0;
+    if (stream->response->headers) {
+        rv = submit_response(session, stream->response);
+    }
+    else {
+        rv = nghttp2_submit_rst_stream(session->ngh2, NGHTTP2_FLAG_NONE,
+                                       stream->id, NGHTTP2_PROTOCOL_ERROR);
+    }
+    
+    if (nghttp2_is_fatal(rv)) {
+        status = APR_EGENERAL;
+        h2_session_abort_int(session, rv);
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, status, session->c,
+                      "submit_response: %s", nghttp2_strerror(rv));
+    }
+    return status;
+}
+
+int h2_session_is_done(h2_session *session)
+{
+    AP_DEBUG_ASSERT(session);
+    return (session->aborted
+            || !session->ngh2
+            || (!nghttp2_session_want_read(session->ngh2)
+                && !nghttp2_session_want_write(session->ngh2)));
+}
+
+static int log_stream(void *ctx, h2_stream *stream)
+{
+    h2_session *session = (h2_session *)ctx;
+    AP_DEBUG_ASSERT(session);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                  "h2_stream(%ld-%d): in set, suspended=%d, aborted=%d, "
+                  "has_data=%d",
+                  session->id, stream->id, stream->suspended, stream->aborted,
+                  h2_mplx_out_has_data_for(session->mplx, stream->id));
+    return 1;
+}
+
+void h2_session_log_stats(h2_session *session)
+{
+    AP_DEBUG_ASSERT(session);
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,
+                  "h2_session(%ld): %d open streams",
+                  session->id, (int)h2_stream_set_size(session->streams));
+    h2_stream_set_iter(session->streams, log_stream, session);
+}
+
+static int frame_print(const nghttp2_frame *frame, char *buffer, size_t maxlen)
+{
+    char scratch[128];
+    size_t s_len = sizeof(scratch)/sizeof(scratch[0]);
+    
+    switch (frame->hd.type) {
+        case NGHTTP2_DATA: {
+            return apr_snprintf(buffer, maxlen,
+                                "DATA[length=%d, flags=%d, stream=%d, padlen=%d]",
+                                (int)frame->hd.length, frame->hd.flags,
+                                frame->hd.stream_id, (int)frame->data.padlen);
+        }
+        case NGHTTP2_HEADERS: {
+            return apr_snprintf(buffer, maxlen,
+                                "HEADERS[length=%d, hend=%d, stream=%d, eos=%d]",
+                                (int)frame->hd.length,
+                                !!(frame->hd.flags & NGHTTP2_FLAG_END_HEADERS),
+                                frame->hd.stream_id,
+                                !!(frame->hd.flags & NGHTTP2_FLAG_END_STREAM));
+        }
+        case NGHTTP2_PRIORITY: {
+            return apr_snprintf(buffer, maxlen,
+                                "PRIORITY[length=%d, flags=%d, stream=%d]",
+                                (int)frame->hd.length,
+                                frame->hd.flags, frame->hd.stream_id);
+        }
+        case NGHTTP2_RST_STREAM: {
+            return apr_snprintf(buffer, maxlen,
+                                "RST_STREAM[length=%d, flags=%d, stream=%d]",
+                                (int)frame->hd.length,
+                                frame->hd.flags, frame->hd.stream_id);
+        }
+        case NGHTTP2_SETTINGS: {
+            if (frame->hd.flags & NGHTTP2_FLAG_ACK) {
+                return apr_snprintf(buffer, maxlen,
+                                    "SETTINGS[ack=1, stream=%d]",
+                                    frame->hd.stream_id);
+            }
+            return apr_snprintf(buffer, maxlen,
+                                "SETTINGS[length=%d, stream=%d]",
+                                (int)frame->hd.length, frame->hd.stream_id);
+        }
+        case NGHTTP2_PUSH_PROMISE: {
+            return apr_snprintf(buffer, maxlen,
+                                "PUSH_PROMISE[length=%d, hend=%d, stream=%d]",
+                                (int)frame->hd.length,
+                                !!(frame->hd.flags & NGHTTP2_FLAG_END_HEADERS),
+                                frame->hd.stream_id);
+        }
+        case NGHTTP2_PING: {
+            return apr_snprintf(buffer, maxlen,
+                                "PING[length=%d, ack=%d, stream=%d]",
+                                (int)frame->hd.length,
+                                frame->hd.flags&NGHTTP2_FLAG_ACK,
+                                frame->hd.stream_id);
+        }
+        case NGHTTP2_GOAWAY: {
+            size_t len = (frame->goaway.opaque_data_len < s_len)?
+            frame->goaway.opaque_data_len : s_len-1;
+            memcpy(scratch, frame->goaway.opaque_data, len);
+            scratch[len+1] = '\0';
+            return apr_snprintf(buffer, maxlen, "GOAWAY[error=%d, reason='%s']",
+                                frame->goaway.error_code, scratch);
+        }
+        case NGHTTP2_WINDOW_UPDATE: {
+            return apr_snprintf(buffer, maxlen,
+                                "WINDOW_UPDATE[length=%d, stream=%d]",
+                                (int)frame->hd.length, frame->hd.stream_id);
+        }
+        default:
+            return apr_snprintf(buffer, maxlen,
+                                "FRAME[type=%d, length=%d, flags=%d, stream=%d]",
+                                frame->hd.type, (int)frame->hd.length,
+                                frame->hd.flags, frame->hd.stream_id);
+    }
+}
+

Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_session.h
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_session.h?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_session.h (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_session.h Tue Jun 30 15:26:16 2015
@@ -0,0 +1,137 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef __mod_h2__h2_session__
+#define __mod_h2__h2_session__
+
+#include "h2_conn_io.h"
+
+/**
+ * A HTTP/2 connection, a session with a specific client.
+ * 
+ * h2_session sits on top of a httpd conn_rec* instance and takes complete
+ * control of the connection data. It receives protocol frames from the
+ * client. For new HTTP/2 streams it creates h2_task(s) that are sent
+ * via callback to a dispatcher (see h2_conn.c).
+ * h2_session keeps h2_io's for each ongoing stream which buffer the
+ * payload for that stream.
+ *
+ * New incoming HEADER frames are converted into a h2_stream+h2_task instance
+ * that both represent a HTTP/2 stream, but may have separate lifetimes. This
+ * allows h2_task to be scheduled in other threads without semaphores
+ * all over the place. It allows task memory to be freed independant of
+ * session lifetime and sessions may close down while tasks are still running.
+ *
+ *
+ */
+
+struct apr_thread_mutext_t;
+struct apr_thread_cond_t;
+struct h2_config;
+struct h2_mplx;
+struct h2_response;
+struct h2_session;
+struct h2_stream;
+struct h2_task;
+struct h2_workers;
+
+struct nghttp2_session;
+
+typedef struct h2_session h2_session;
+
+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 */
+    int aborted;                    /* this session is being aborted */
+    apr_size_t frames_received;     /* number of http/2 frames received */
+    apr_size_t max_stream_count;    /* max number of open streams */
+    apr_size_t max_stream_mem;      /* max buffer memory for a single stream */
+    
+    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 */
+    
+    h2_conn_io io;                  /* io on httpd conn filters */
+    struct h2_mplx *mplx;           /* multiplexer for stream data */
+    
+    struct h2_stream_set *streams;  /* streams handled by this session */
+    
+    struct nghttp2_session *ngh2;   /* the nghttp2 session (internal use) */
+    struct h2_workers *workers;     /* for executing stream tasks */
+};
+
+
+/* Create a new h2_session for the given connection (mode 'h2').
+ * The session will apply the configured parameter.
+ */
+h2_session *h2_session_create(conn_rec *c, struct h2_config *cfg, 
+                              struct h2_workers *workers);
+
+/* Create a new h2_session for the given request (mode 'h2c').
+ * The session will apply the configured parameter.
+ */
+h2_session *h2_session_rcreate(request_rec *r, struct h2_config *cfg,
+                               struct h2_workers *workers);
+
+/* Destroy the session and all object it still contains. This will not
+ * destroy h2_task instances that not finished yet. */
+void h2_session_destroy(h2_session *session);
+
+/* Called once at start of session. Performs initial client thingies. */
+apr_status_t h2_session_start(h2_session *session, int *rv);
+
+/* Return != 0 iff session is finished and connection can be closed.
+ */
+int h2_session_is_done(h2_session *session);
+
+/* Called when the session will shutdown after all open streams
+ * are handled. New streams will no longer be accepted. 
+ * Call with reason APR_SUCCESS to initiate a graceful shutdown. */
+apr_status_t h2_session_goaway(h2_session *session, apr_status_t reason);
+
+/* Called when an error occured and the session needs to shut down.
+ * Status indicates the reason of the error. */
+apr_status_t h2_session_abort(h2_session *session, apr_status_t reason, int rv);
+
+/* Called before a session gets destroyed, might flush output etc. */
+apr_status_t h2_session_close(h2_session *session);
+
+/* Read more data from the client connection. Used normally with blocking
+ * APR_NONBLOCK_READ, which will return APR_EAGAIN when no data is available.
+ * Use with APR_BLOCK_READ only when certain that no data needs to be written
+ * while waiting. */
+apr_status_t h2_session_read(h2_session *session, apr_read_type_e block);
+
+/* Write data out to the client, if there is any. Otherwise, wait for
+ * a maximum of timeout micro-seconds and return to the caller. If timeout
+ * occurred, APR_TIMEUP will be returned.
+ */
+apr_status_t h2_session_write(h2_session *session,
+                              apr_interval_time_t timeout);
+
+/* Start submitting the response to a stream request. This is possible
+ * once we have all the response headers. */
+apr_status_t h2_session_handle_response(h2_session *session,
+                                        struct h2_stream *stream);
+
+/* Get the h2_stream for the given stream idenrtifier. */
+struct h2_stream *h2_session_get_stream(h2_session *session, int stream_id);
+
+void h2_session_log_stats(h2_session *session);
+
+#endif /* defined(__mod_h2__h2_session__) */

Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_stream.c
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_stream.c?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_stream.c (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_stream.c Tue Jun 30 15:26:16 2015
@@ -0,0 +1,297 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <assert.h>
+#include <stddef.h>
+
+#define APR_POOL_DEBUG  7
+
+#include <httpd.h>
+#include <http_core.h>
+#include <http_connection.h>
+#include <http_log.h>
+
+#include <nghttp2/nghttp2.h>
+
+#include "h2_private.h"
+#include "h2_conn.h"
+#include "h2_mplx.h"
+#include "h2_request.h"
+#include "h2_response.h"
+#include "h2_stream.h"
+#include "h2_task.h"
+#include "h2_ctx.h"
+#include "h2_task_input.h"
+#include "h2_task.h"
+#include "h2_util.h"
+
+
+static void set_state(h2_stream *stream, h2_stream_state_t state)
+{
+    AP_DEBUG_ASSERT(stream);
+    if (stream->state != state) {
+        stream->state = state;
+    }
+}
+
+h2_stream *h2_stream_create(int id, apr_pool_t *pool, struct h2_mplx *m)
+{
+    h2_stream *stream = apr_pcalloc(pool, sizeof(h2_stream));
+    if (stream != NULL) {
+        stream->id = id;
+        stream->state = H2_STREAM_ST_IDLE;
+        stream->pool = pool;
+        stream->m = m;
+        stream->request = h2_request_create(id, pool, m->c->bucket_alloc);
+        ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, m->c,
+                      "h2_stream(%ld-%d): created", m->id, stream->id);
+    }
+    return stream;
+}
+
+void h2_stream_cleanup(h2_stream *stream)
+{
+    if (stream->request) {
+        h2_request_destroy(stream->request);
+        stream->request = NULL;
+    }
+}
+
+apr_status_t h2_stream_destroy(h2_stream *stream)
+{
+    AP_DEBUG_ASSERT(stream);
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->m->c,
+                  "h2_stream(%ld-%d): destroy", stream->m->id, stream->id);
+    h2_stream_cleanup(stream);
+    
+    if (stream->task) {
+        h2_task_destroy(stream->task);
+        stream->task = NULL;
+    }
+    if (stream->pool) {
+        apr_pool_destroy(stream->pool);
+    }
+    return APR_SUCCESS;
+}
+
+void h2_stream_attach_pool(h2_stream *stream, apr_pool_t *pool)
+{
+    stream->pool = pool;
+}
+
+apr_pool_t *h2_stream_detach_pool(h2_stream *stream)
+{
+    apr_pool_t *pool = stream->pool;
+    stream->pool = NULL;
+    return pool;
+}
+
+void h2_stream_abort(h2_stream *stream)
+{
+    AP_DEBUG_ASSERT(stream);
+    stream->aborted = 1;
+}
+
+apr_status_t h2_stream_set_response(h2_stream *stream, h2_response *response,
+                                    apr_bucket_brigade *bb)
+{
+    stream->response = response;
+    if (bb && !APR_BRIGADE_EMPTY(bb)) {
+        if (!stream->bbout) {
+            stream->bbout = apr_brigade_create(stream->pool, 
+                                               stream->m->c->bucket_alloc);
+        }
+        return h2_util_move(stream->bbout, bb, 16 * 1024, 1, NULL, 
+                            "h2_stream_set_response");
+    }
+    return APR_SUCCESS;
+}
+
+static int set_closed(h2_stream *stream) 
+{
+    switch (stream->state) {
+        case H2_STREAM_ST_CLOSED_INPUT:
+        case H2_STREAM_ST_CLOSED:
+            return 0; /* ignore, idempotent */
+        case H2_STREAM_ST_CLOSED_OUTPUT:
+            /* both closed now */
+            set_state(stream, H2_STREAM_ST_CLOSED);
+            break;
+        default:
+            /* everything else we jump to here */
+            set_state(stream, H2_STREAM_ST_CLOSED_INPUT);
+            break;
+    }
+    return 1;
+}
+
+apr_status_t h2_stream_rwrite(h2_stream *stream, request_rec *r)
+{
+    AP_DEBUG_ASSERT(stream);
+    set_state(stream, H2_STREAM_ST_OPEN);
+    apr_status_t status = h2_request_rwrite(stream->request, r, stream->m);
+    return status;
+}
+
+apr_status_t h2_stream_write_eoh(h2_stream *stream, int eos)
+{
+    AP_DEBUG_ASSERT(stream);
+    /* Seeing the end-of-headers, we have everything we need to 
+     * start processing it.
+     */
+    conn_rec *c = h2_conn_create(stream->m->c, stream->pool);
+    if (c == NULL) {
+        ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_ENOMEM, stream->m->c,
+                      "h2_stream(%ld-%d): create connection",
+                      stream->m->id, stream->id);
+        return APR_ENOMEM;
+    }
+    stream->task = h2_task_create(stream->m->id, stream->id, 
+                                  stream->pool, stream->m, c);
+    
+    apr_status_t status = h2_request_end_headers(stream->request, 
+                                                 stream->m, stream->task, eos);
+    if (status == APR_SUCCESS) {
+        status = h2_mplx_do_task(stream->m, stream->task);
+    }
+    if (eos) {
+        status = h2_stream_write_eos(stream);
+    }
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, stream->m->c,
+                  "h2_stream(%ld-%d): end header, task %s %s (%s)",
+                  stream->m->id, stream->id,
+                  stream->request->method, stream->request->path,
+                  stream->request->authority);
+    
+    return status;
+}
+
+apr_status_t h2_stream_write_eos(h2_stream *stream)
+{
+    AP_DEBUG_ASSERT(stream);
+    ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, stream->m->c,
+                  "h2_stream(%ld-%d): closing input",
+                  stream->m->id, stream->id);
+    if (set_closed(stream)) {
+        return h2_request_close(stream->request);
+    }
+    return APR_SUCCESS;
+}
+
+apr_status_t h2_stream_write_header(h2_stream *stream,
+                                    const char *name, size_t nlen,
+                                    const char *value, size_t vlen)
+{
+    AP_DEBUG_ASSERT(stream);
+    switch (stream->state) {
+        case H2_STREAM_ST_IDLE:
+            set_state(stream, H2_STREAM_ST_OPEN);
+            break;
+        case H2_STREAM_ST_OPEN:
+            break;
+        default:
+            return APR_EINVAL;
+    }
+    return h2_request_write_header(stream->request, name, nlen,
+                                   value, vlen, stream->m);
+}
+
+apr_status_t h2_stream_write_data(h2_stream *stream,
+                                  const char *data, size_t len)
+{
+    AP_DEBUG_ASSERT(stream);
+    AP_DEBUG_ASSERT(stream);
+    switch (stream->state) {
+        case H2_STREAM_ST_OPEN:
+            break;
+        default:
+            return APR_EINVAL;
+    }
+    return h2_request_write_data(stream->request, data, len);
+}
+
+apr_status_t h2_stream_prep_read(h2_stream *stream, 
+                                 apr_size_t *plen, int *peos)
+{
+    apr_status_t status = APR_SUCCESS;
+    const char *src;
+    
+    if (stream->bbout && !APR_BRIGADE_EMPTY(stream->bbout)) {
+        src = "stream";
+        status = h2_util_bb_avail(stream->bbout, plen, peos);
+        if (status == APR_SUCCESS && !*peos && !*plen) {
+            apr_brigade_cleanup(stream->bbout);
+            return h2_stream_prep_read(stream, plen, peos);
+        }
+    }
+    else {
+        src = "mplx";
+        status = h2_mplx_out_readx(stream->m, stream->id, 
+                                   NULL, NULL, plen, peos);
+    }
+    if (status == APR_SUCCESS && !*peos && !*plen) {
+        status = APR_EAGAIN;
+    }
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->m->c,
+                  "h2_stream(%ld-%d): prep_read %s, len=%ld eos=%d",
+                  stream->m->id, stream->id, 
+                  src, (long)*plen, *peos);
+    return status;
+}
+
+apr_status_t h2_stream_readx(h2_stream *stream, 
+                             h2_io_data_cb *cb, void *ctx,
+                             apr_size_t *plen, int *peos)
+{
+    if (stream->bbout && !APR_BRIGADE_EMPTY(stream->bbout)) {
+        return h2_util_bb_readx(stream->bbout, cb, ctx, plen, peos);
+    }
+    return h2_mplx_out_readx(stream->m, stream->id, 
+                             cb, ctx, plen, peos);
+}
+
+
+apr_status_t h2_stream_read(h2_stream *stream, char *buffer, 
+                            apr_size_t *plen, int *peos)
+{
+    apr_status_t status = APR_SUCCESS;
+    const char *src;
+    if (stream->bbout && !APR_BRIGADE_EMPTY(stream->bbout)) {
+        src = "stream";
+        status = h2_util_bb_read(stream->bbout, buffer, plen, peos);
+    }
+    else {
+        src = "mplx";
+        status = h2_mplx_out_read(stream->m, stream->id, buffer, plen, peos);
+    }
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, stream->m->c,
+                  "h2_stream(%ld-%d): read %s, len=%ld eos=%d",
+                  stream->m->id, stream->id, 
+                  src, (long)*plen, *peos);
+    return status;
+}
+
+void h2_stream_set_suspended(h2_stream *stream, int suspended)
+{
+    AP_DEBUG_ASSERT(stream);
+    stream->suspended = !!suspended;
+}
+
+int h2_stream_is_suspended(h2_stream *stream)
+{
+    AP_DEBUG_ASSERT(stream);
+    return stream->suspended;
+}
+

Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_stream.h
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_stream.h?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_stream.h (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_stream.h Tue Jun 30 15:26:16 2015
@@ -0,0 +1,111 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef __mod_h2__h2_stream__
+#define __mod_h2__h2_stream__
+
+/**
+ * A HTTP/2 stream, e.g. a client request+response in HTTP/1.1 terms.
+ * 
+ * Ok, not quite, but close enough, since we do not implement server
+ * pushes yet.
+ *
+ * A stream always belongs to a h2_session, the one managing the
+ * connection to the client. The h2_session writes to the h2_stream,
+ * adding HEADERS and DATA and finally an EOS. When headers are done,
+ * h2_stream can create a h2_task that can be scheduled to fullfill the
+ * request.
+ * 
+ * This response headers are added directly to the h2_mplx of the session,
+ * but the response DATA can be read via h2_stream. Reading data will
+ * never block but return APR_EAGAIN when there currently is no data (and
+ * no eos) in the multiplexer for this stream.
+ */
+#include "h2_io.h"
+
+typedef enum {
+    H2_STREAM_ST_IDLE,
+    H2_STREAM_ST_OPEN,
+    H2_STREAM_ST_RESV_LOCAL,
+    H2_STREAM_ST_RESV_REMOTE,
+    H2_STREAM_ST_CLOSED_INPUT,
+    H2_STREAM_ST_CLOSED_OUTPUT,
+    H2_STREAM_ST_CLOSED,
+} h2_stream_state_t;
+
+struct h2_mplx;
+struct h2_request;
+struct h2_response;
+struct h2_task;
+
+typedef struct h2_stream h2_stream;
+
+struct h2_stream {
+    int id;                     /* http2 stream id */
+    h2_stream_state_t state;    /* http/2 state of this stream */
+    struct h2_mplx *m;          /* the multiplexer to work with */
+    
+    int aborted;                /* was aborted */
+    int suspended;              /* DATA sending has been suspended */
+    
+    apr_pool_t *pool;           /* the memory pool for this stream */
+    struct h2_request *request; /* the request made in this stream */
+    
+    struct h2_task *task;       /* task created for this stream */
+    struct h2_response *response; /* the response, once ready */
+    apr_bucket_brigade *bbout;  /* output DATA */
+};
+
+
+h2_stream *h2_stream_create(int id, apr_pool_t *pool, struct h2_mplx *m);
+
+apr_status_t h2_stream_destroy(h2_stream *stream);
+void h2_stream_cleanup(h2_stream *stream);
+
+apr_pool_t *h2_stream_detach_pool(h2_stream *stream);
+void h2_stream_attach_pool(h2_stream *stream, apr_pool_t *pool);
+
+void h2_stream_abort(h2_stream *stream);
+
+apr_status_t h2_stream_rwrite(h2_stream *stream, request_rec *r);
+
+apr_status_t h2_stream_write_eos(h2_stream *stream);
+
+apr_status_t h2_stream_write_header(h2_stream *stream,
+                                    const char *name, size_t nlen,
+                                    const char *value, size_t vlen);
+
+apr_status_t h2_stream_write_eoh(h2_stream *stream, int eos);
+
+apr_status_t h2_stream_write_data(h2_stream *stream,
+                                  const char *data, size_t len);
+
+apr_status_t h2_stream_set_response(h2_stream *stream, 
+                                    struct h2_response *response,
+                                    apr_bucket_brigade *bb);
+
+apr_status_t h2_stream_read(h2_stream *stream, char *buffer, 
+                            apr_size_t *plen, int *peos);
+
+apr_status_t h2_stream_prep_read(h2_stream *stream, 
+                                 apr_size_t *plen, int *peos);
+
+apr_status_t h2_stream_readx(h2_stream *stream, h2_io_data_cb *cb, 
+                             void *ctx, apr_size_t *plen, int *peos);
+
+void h2_stream_set_suspended(h2_stream *stream, int suspended);
+int h2_stream_is_suspended(h2_stream *stream);
+
+#endif /* defined(__mod_h2__h2_stream__) */

Added: httpd/httpd/trunk/modules/http2/mod_h2/h2_stream_set.c
URL: http://svn.apache.org/viewvc/httpd/httpd/trunk/modules/http2/mod_h2/h2_stream_set.c?rev=1688474&view=auto
==============================================================================
--- httpd/httpd/trunk/modules/http2/mod_h2/h2_stream_set.c (added)
+++ httpd/httpd/trunk/modules/http2/mod_h2/h2_stream_set.c Tue Jun 30 15:26:16 2015
@@ -0,0 +1,158 @@
+/* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <assert.h>
+#include <stddef.h>
+
+#include <apr_strings.h>
+
+#include <httpd.h>
+#include <http_core.h>
+#include <http_connection.h>
+#include <http_log.h>
+
+#include "h2_private.h"
+#include "h2_session.h"
+#include "h2_stream.h"
+#include "h2_task.h"
+#include "h2_stream_set.h"
+
+#define H2_STREAM_IDX(list, i) ((h2_stream**)(list)->elts)[i]
+
+struct h2_stream_set {
+    apr_array_header_t *list;
+};
+
+h2_stream_set *h2_stream_set_create(apr_pool_t *pool)
+{
+    h2_stream_set *sp = apr_pcalloc(pool, sizeof(h2_stream_set));
+    if (sp) {
+        sp->list = apr_array_make(pool, 100, sizeof(h2_stream*));
+        if (!sp->list) {
+            return NULL;
+        }
+    }
+    return sp;
+}
+
+void h2_stream_set_destroy(h2_stream_set *sp)
+{
+    (void)sp;
+}
+
+static int h2_stream_id_cmp(const void *s1, const void *s2)
+{
+    h2_stream **pstream1 = (h2_stream **)s1;
+    h2_stream **pstream2 = (h2_stream **)s2;
+    return (*pstream1)->id - (*pstream2)->id;
+}
+
+h2_stream *h2_stream_set_get(h2_stream_set *sp, int stream_id)
+{
+    /* we keep the array sorted by id, so lookup can be done
+     * by bsearch.
+     */
+    h2_stream key;
+    memset(&key, 0, sizeof(key));
+    key.id = stream_id;
+    h2_stream *pkey = &key;
+    h2_stream **ps = bsearch(&pkey, sp->list->elts, sp->list->nelts, 
+                             sp->list->elt_size, h2_stream_id_cmp);
+    return ps? *ps : NULL;
+}
+
+static void h2_stream_set_sort(h2_stream_set *sp)
+{
+    qsort(sp->list->elts, sp->list->nelts, sp->list->elt_size, 
+          h2_stream_id_cmp);
+}
+
+apr_status_t h2_stream_set_add(h2_stream_set *sp, h2_stream *stream)
+{
+    h2_stream *existing = h2_stream_set_get(sp, stream->id);
+    if (!existing) {
+        APR_ARRAY_PUSH(sp->list, h2_stream*) = stream;
+        /* Normally, streams get added in ascending order if id. We
+         * keep the array sorted, so we just need to check of the newly
+         * appended stream has a lower id than the last one. if not,
+         * sorting is not necessary.
+         */
+        int last = sp->list->nelts - 1;
+        if (last > 0 
+            && (H2_STREAM_IDX(sp->list, last)->id 
+                < H2_STREAM_IDX(sp->list, last-1)->id)) {
+            h2_stream_set_sort(sp);
+        }
+    }
+    return APR_SUCCESS;
+}
+
+h2_stream *h2_stream_set_remove(h2_stream_set *sp, h2_stream *stream)
+{
+    for (int i = 0; i < sp->list->nelts; ++i) {
+        h2_stream *s = H2_STREAM_IDX(sp->list, i);
+        if (s == stream) {
+            --sp->list->nelts;
+            int n = sp->list->nelts - i;
+            if (n > 0) {
+                /* Close the hole in the array by moving the upper
+                 * parts down one step.
+                 */
+                h2_stream **selts = (h2_stream**)sp->list->elts;
+                memmove(selts+i, selts+i+1, n * sizeof(h2_stream*));
+            }
+            return s;
+        }
+    }
+    return NULL;
+}
+
+void h2_stream_set_remove_all(h2_stream_set *sp)
+{
+    sp->list->nelts = 0;
+}
+
+int h2_stream_set_is_empty(h2_stream_set *sp)
+{
+    AP_DEBUG_ASSERT(sp);
+    return sp->list->nelts == 0;
+}
+
+h2_stream *h2_stream_set_find(h2_stream_set *sp,
+                              h2_stream_set_match_fn match, void *ctx)
+{
+    h2_stream *s = NULL;
+    for (int i = 0; !s && i < sp->list->nelts; ++i) {
+        s = match(ctx, H2_STREAM_IDX(sp->list, i));
+    }
+    return s;
+}
+
+void h2_stream_set_iter(h2_stream_set *sp,
+                        h2_stream_set_iter_fn *iter, void *ctx)
+{
+    for (int i = 0; i < sp->list->nelts; ++i) {
+        h2_stream *s = H2_STREAM_IDX(sp->list, i);
+        if (!iter(ctx, s)) {
+            break;
+        }
+    }
+}
+
+apr_size_t h2_stream_set_size(h2_stream_set *sp)
+{
+    return sp->list->nelts;
+}
+



Mime
View raw message