fixing unbuffered output handling for h2c

git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/trunk@1711451 13f79535-47bb-0310-9956-ffa450edef68

diff --git a/modules/http2/h2_conn_io.c b/modules/http2/h2_conn_io.c
index 6c194d93067df19fed07f074153f6672a257a1f8..370051b40656e568af5ead6594baaa7161693b5a 100644 (file)
--- a/modules/http2/h2_conn_io.c
+++ b/modules/http2/h2_conn_io.c
@@ -52,28 +52,9 @@ apr_status_t h2_conn_io_init(h2_conn_io *io, conn_rec *c)
     io->input              = apr_brigade_create(c->pool, c->bucket_alloc);
     io->output             = apr_brigade_create(c->pool, c->bucket_alloc);
     io->buflen             = 0;
-    /* That is where we start with, 
-     * see https://issues.apache.org/jira/browse/TS-2503 */
-    io->tls_warmup_size    = h2_config_geti64(cfg, H2_CONF_TLS_WARMUP_SIZE);
-    io->tls_cooldown_usecs = (h2_config_geti(cfg, H2_CONF_TLS_COOLDOWN_SECS) 
-                              * APR_USEC_PER_SEC);
-    io->write_size         = WRITE_SIZE_INITIAL; 
-    io->last_write         = 0;
-    io->buffer_output      = h2_h2_is_tls(c);
-
-    if (APLOGctrace1(c)) {
-        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, io->connection,
-                      "h2_conn_io(%ld): init, buffering=%d, warmup_size=%ld, cd_secs=%f",
-                      io->connection->id, io->buffer_output, (long)io->tls_warmup_size,
-                      ((float)io->tls_cooldown_usecs/APR_USEC_PER_SEC));
-    }
-    /* Currently we buffer only for TLS output. The reason this gives
-     * improved performance is that buckets send to the mod_ssl network
-     * filter will be encrypted in chunks. There is a special filter
-     * that tries to aggregate data, but that does not work well when
-     * bucket sizes alternate between tiny frame headers and large data
-     * chunks.
-     */
+    io->is_tls             = h2_h2_is_tls(c);
+    io->buffer_output      = io->is_tls;
+    
     if (io->buffer_output) {
         io->bufsize = WRITE_BUFFER_SIZE;
         io->buffer = apr_pcalloc(c->pool, io->bufsize);
@@ -82,6 +63,27 @@ apr_status_t h2_conn_io_init(h2_conn_io *io, conn_rec *c)
         io->bufsize = 0;
     }
     
+    if (io->is_tls) {
+        /* That is where we start with, 
+         * see https://issues.apache.org/jira/browse/TS-2503 */
+        io->warmup_size    = h2_config_geti64(cfg, H2_CONF_TLS_WARMUP_SIZE);
+        io->cooldown_usecs = (h2_config_geti(cfg, H2_CONF_TLS_COOLDOWN_SECS) 
+                              * APR_USEC_PER_SEC);
+        io->write_size     = WRITE_SIZE_INITIAL; 
+    }
+    else {
+        io->warmup_size    = 0;
+        io->cooldown_usecs = 0;
+        io->write_size     = io->bufsize;
+    }
+
+    if (APLOGctrace1(c)) {
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, io->connection,
+                      "h2_conn_io(%ld): init, buffering=%d, warmup_size=%ld, cd_secs=%f",
+                      io->connection->id, io->buffer_output, (long)io->warmup_size,
+                      ((float)io->cooldown_usecs/APR_USEC_PER_SEC));
+    }
+
     return APR_SUCCESS;
 }
 
@@ -91,6 +93,11 @@ void h2_conn_io_destroy(h2_conn_io *io)
     io->output = NULL;
 }
 
+int h2_conn_io_is_buffered(h2_conn_io *io)
+{
+    return io->bufsize > 0;
+}
+
 static apr_status_t h2_conn_io_bucket_read(h2_conn_io *io,
                                            apr_read_type_e block,
                                            h2_conn_io_on_read_cb on_read_cb,
@@ -206,10 +213,10 @@ static apr_status_t pass_out(apr_bucket_brigade *bb, void *ctx)
     }
     
     ap_update_child_status(io->connection->sbh, SERVER_BUSY_WRITE, NULL);
-    status = apr_brigade_length(bb, 1, &bblen);
+    status = apr_brigade_length(bb, 0, &bblen);
     if (status == APR_SUCCESS) {
         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, io->connection,
-                      "h2_conn_io(%ld): flush, brigade %ld bytes",
+                      "h2_conn_io(%ld): pass_out brigade %ld bytes",
                       io->connection->id, (long)bblen);
         status = ap_pass_brigade(io->connection->output_filters, bb);
         if (status == APR_SUCCESS) {
@@ -231,8 +238,8 @@ static apr_status_t bucketeer_buffer(h2_conn_io *io) {
     int bcount, i;
 
     if (io->write_size > WRITE_SIZE_INITIAL 
-        && (io->tls_cooldown_usecs > 0)
-        && (apr_time_now() - io->last_write) >= io->tls_cooldown_usecs) {
+        && (io->cooldown_usecs > 0)
+        && (apr_time_now() - io->last_write) >= io->cooldown_usecs) {
         /* long time not written, reset write size */
         io->write_size = WRITE_SIZE_INITIAL;
         io->bytes_written = 0;
@@ -241,7 +248,7 @@ static apr_status_t bucketeer_buffer(h2_conn_io *io) {
                       (long)io->connection->id, (long)io->write_size);
     }
     else if (io->write_size < WRITE_SIZE_MAX 
-             && io->bytes_written >= io->tls_warmup_size) {
+             && io->bytes_written >= io->warmup_size) {
         /* connection is hot, use max size */
         io->write_size = WRITE_SIZE_MAX;
         ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, io->connection,
@@ -272,7 +279,7 @@ apr_status_t h2_conn_io_write(h2_conn_io *io,
     apr_status_t status = APR_SUCCESS;
     io->unflushed = 1;
     
-    if (io->buffer_output) {
+    if (io->bufsize > 0) {
         ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, io->connection,
                       "h2_conn_io: buffering %ld bytes", (long)length);
         while (length > 0 && (status == APR_SUCCESS)) {
@@ -297,12 +304,20 @@ apr_status_t h2_conn_io_write(h2_conn_io *io,
         }
         
     }
+    else if (1) {
+        apr_bucket *b;
+        
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, io->connection,
+                      "h2_conn_io: passing %ld transient bytes to output filters", 
+                      (long)length);
+        b = apr_bucket_transient_create(buf,length, io->output->bucket_alloc);
+        APR_BRIGADE_INSERT_TAIL(io->output, b);
+        status = pass_out(io->output, io);
+    }
     else {
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, io->connection,
+                      "h2_conn_io: writing %ld bytes to brigade", (long)length);
         status = apr_brigade_write(io->output, pass_out, io, buf, length);
-        if (status != APR_SUCCESS) {
-            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, io->connection,
-                          "h2_conn_io: write error");
-        }
     }
     
     return status;

diff --git a/modules/http2/h2_conn_io.h b/modules/http2/h2_conn_io.h
index b76ed39643a6f4e2b4b0e06f8d8f3af5e9de32bd..250fa33ced10d4b271e33c67a85272ed61cfce77 100644 (file)
--- a/modules/http2/h2_conn_io.h
+++ b/modules/http2/h2_conn_io.h
@@ -26,15 +26,16 @@ typedef struct {
     conn_rec *connection;
     apr_bucket_brigade *input;
     apr_bucket_brigade *output;
-    int buffer_output;
+
+    int is_tls;
+    apr_time_t cooldown_usecs;
+    apr_int64_t warmup_size;
+    
     int write_size;
     apr_time_t last_write;
     apr_int64_t bytes_written;
     
-    apr_time_t tls_cooldown_usecs;
-    apr_int64_t tls_warmup_size;
-    
-    
+    int buffer_output;
     char *buffer;
     apr_size_t buflen;
     apr_size_t bufsize;
@@ -42,8 +43,11 @@ typedef struct {
 } h2_conn_io;
 
 apr_status_t h2_conn_io_init(h2_conn_io *io, conn_rec *c);
+
 void h2_conn_io_destroy(h2_conn_io *io);
 
+int h2_conn_io_is_buffered(h2_conn_io *io);
+
 typedef apr_status_t (*h2_conn_io_on_read_cb)(const char *data, apr_size_t len,
                                          apr_size_t *readlen, int *done,
                                          void *puser);

diff --git a/modules/http2/h2_session.c b/modules/http2/h2_session.c
index 37cc6a52aa315a24ef324f11efb51808e008c801..ca3da31dea4cd0224048aa436a1be382edbf2c7f 100644 (file)
--- a/modules/http2/h2_session.c
+++ b/modules/http2/h2_session.c
@@ -249,6 +249,8 @@ static int before_frame_send_cb(nghttp2_session *ngh2,
         case NGHTTP2_GOAWAY:
             session->flush = 1;
             break;
+        case NGHTTP2_DATA:
+            
         default:
             break;
 
@@ -557,6 +559,10 @@ static int on_send_data_cb(nghttp2_session *ngh2,
         return NGHTTP2_ERR_CALLBACK_FAILURE;
     }
     
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,
+                  "h2_stream(%ld-%d): send_data_cb for %ld bytes",
+                  session->id, (int)stream_id, (long)length);
+                  
     status = send_data(session, (const char *)framehd, 9);
     if (status == APR_SUCCESS) {
         if (padlen) {
@@ -1124,6 +1130,27 @@ apr_status_t h2_session_close(h2_session *session)
 
 /* The session wants to send more DATA for the given stream.
  */
+ 
+typedef struct {
+    char *buf;
+    size_t offset;
+    h2_session *session;
+    h2_stream *stream;
+} cpy_ctx;
+
+static apr_status_t copy_buffer(void *ctx, const char *data, apr_size_t len)
+{
+    cpy_ctx *c = ctx;
+    
+    ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, c->session->c,
+                  "h2_stream(%ld-%d): copy %ld bytes for DATA #%ld",
+                  c->session->id, c->stream->id, 
+                  (long)c->stream->data_frames_sent, (long)len);
+    memcpy(c->buf + c->offset, data, len);
+    c->offset += len;
+    return APR_SUCCESS;
+}
+
 static ssize_t stream_data_cb(nghttp2_session *ng2s,
                               int32_t stream_id,
                               uint8_t *buf,
@@ -1153,9 +1180,25 @@ static ssize_t stream_data_cb(nghttp2_session *ng2s,
     
     AP_DEBUG_ASSERT(!h2_stream_is_suspended(stream));
     
-    status = h2_stream_prep_read(stream, &nread, &eos);
-    if (nread) {
-        *data_flags |=  NGHTTP2_DATA_FLAG_NO_COPY;
+    if (h2_conn_io_is_buffered(&session->io)) {
+        status = h2_stream_prep_read(stream, &nread, &eos);
+        if (nread) {
+            *data_flags |=  NGHTTP2_DATA_FLAG_NO_COPY;
+        }
+    }
+    else {
+        cpy_ctx ctx;
+        ctx.buf = (char *)buf;
+        ctx.offset = 0;
+        ctx.session = session;
+        ctx.stream = stream;
+        
+        status = h2_stream_readx(stream, copy_buffer, &ctx, &nread, &eos);
+        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, session->c,
+                      "h2_stream(%ld-%d): read %ld bytes (DATA #%ld)",
+                      session->id, (int)stream_id, (long)nread, 
+                      (long)stream->data_frames_sent);
+        stream->data_frames_sent++;
     }
     
     switch (status) {

diff --git a/modules/http2/h2_stream.c b/modules/http2/h2_stream.c
index f5a7f3fd50c213c6d6c6816ce506dcf50beb7ca8..a2c53a1891c2d5ead7b08781edba642692dcfbf8 100644 (file)
--- a/modules/http2/h2_stream.c
+++ b/modules/http2/h2_stream.c
@@ -112,6 +112,7 @@ apr_status_t h2_stream_set_response(h2_stream *stream, h2_response *response,
             stream->bbout = apr_brigade_create(stream->pool, 
                                                stream->m->c->bucket_alloc);
         }
+        /* TODO: this does not move complete file buckets.*/
         status = h2_util_move(stream->bbout, bb, 16 * 1024, NULL,  
                               "h2_stream_set_response");
     }
@@ -271,8 +272,7 @@ apr_status_t h2_stream_prep_read(h2_stream *stream,
     }
     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);
+                  stream->m->id, stream->id, src, (long)*plen, *peos);
     return status;
 }
 
@@ -280,14 +280,31 @@ apr_status_t h2_stream_readx(h2_stream *stream,
                              h2_io_data_cb *cb, void *ctx,
                              apr_size_t *plen, int *peos)
 {
+    apr_status_t status = APR_SUCCESS;
+    const char *src;
+    
     if (stream->rst_error) {
         return APR_ECONNRESET;
     }
     if (stream->bbout && !APR_BRIGADE_EMPTY(stream->bbout)) {
-        return h2_util_bb_readx(stream->bbout, cb, ctx, plen, peos);
+        src = "stream";
+        status = h2_util_bb_readx(stream->bbout, cb, ctx, plen, peos);
+        if (status == APR_SUCCESS && !*peos && !*plen) {
+            apr_brigade_cleanup(stream->bbout);
+            return h2_stream_readx(stream, cb, ctx, plen, peos);
+        }
     }
-    return h2_mplx_out_readx(stream->m, stream->id, 
-                             cb, ctx, plen, peos);
+    else {
+        src = "mplx";
+        status = h2_mplx_out_readx(stream->m, stream->id, cb, ctx, 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): readx %s, len=%ld eos=%d",
+                  stream->m->id, stream->id, src, (long)*plen, *peos);
+    return status;
 }
 
 

diff --git a/modules/http2/h2_stream.h b/modules/http2/h2_stream.h
index d2c9c3a18419f446e3bb586afc2c08194b972d6c..35348b7a17ec54973a46bf2945183128a164ea84 100644 (file)
--- a/modules/http2/h2_stream.h
+++ b/modules/http2/h2_stream.h
@@ -59,6 +59,7 @@ struct h2_stream {
     
     int aborted;                /* was aborted */
     int suspended;              /* DATA sending has been suspended */
+    apr_size_t data_frames_sent;/* # of DATA frames sent out for this stream */
     
     apr_pool_t *pool;           /* the memory pool for this stream */
     struct h2_request *request; /* the request made in this stream */