src: Implement per-frame DATA compression

Currently, nghttpd server only compresses files whose extensions are
one of .html, .js, .css and .txt.  nghttp advertises its support of
per-frame compression in SETTINGS frame.  To implement this feature,
we added 2 public API: nghttp2_session_get_remote_settings() and
nghttp2_gzip_inflate_finished().

lib/includes/nghttp2/nghttp2.h

@@ -2080,6 +2080,14 @@ int32_t nghttp2_session_get_stream_remote_window_size(nghttp2_session* session,
 int nghttp2_session_terminate_session(nghttp2_session *session,
                                       nghttp2_error_code error_code);
 
+/**
+ * @function
+ *
+ * Returns the value of SETTINGS |id| notified by a remote endpoint.
+ */
+uint32_t nghttp2_session_get_remote_settings(nghttp2_session *session,
+                                             nghttp2_settings_id id);
+
 /**
  * @function
  *
@@ -2799,6 +2807,15 @@ int nghttp2_gzip_inflate(nghttp2_gzip *inflater,
                          uint8_t *out, size_t *outlen_ptr,
                          const uint8_t *in, size_t *inlen_ptr);
 
+/**
+ * @function
+ *
+ * Returns nonzero if |inflater| sees the end of deflate stream.
+ * After this function returns nonzero, `nghttp2_gzip_inflate()` with
+ * |inflater| gets to return error.
+ */
+int nghttp2_gzip_inflate_finished(nghttp2_gzip *inflater);
+
 /**
  * @function
  *

lib/nghttp2_gzip.c

@@ -89,3 +89,8 @@ int nghttp2_gzip_inflate(nghttp2_gzip *inflater,
     return 0;
   }
 }
+
+int nghttp2_gzip_inflate_finished(nghttp2_gzip *inflater)
+{
+  return inflater->finished;
+}

lib/nghttp2_session.c

@@ -5779,6 +5779,16 @@ int32_t nghttp2_session_get_stream_remote_window_size(nghttp2_session* session,
   return nghttp2_session_next_data_read(session, stream);
 }
 
+uint32_t nghttp2_session_get_remote_settings(nghttp2_session *session,
+                                             nghttp2_settings_id id)
+{
+  if(id > NGHTTP2_SETTINGS_MAX) {
+    return 0;
+  }
+
+  return session->remote_settings[id];
+}
+
 int nghttp2_session_upgrade(nghttp2_session *session,
                             const uint8_t *settings_payload,
                             size_t settings_payloadlen,

src/HttpServer.cc

@@ -109,7 +109,8 @@ Stream::Stream(Http2Handler *handler, int32_t stream_id)
     rtimer(nullptr),
     wtimer(nullptr),
     stream_id(stream_id),
-    file(-1)
+    file(-1),
+    enable_compression(false)
 {}
 
 Stream::~Stream()
@@ -714,9 +715,13 @@ int Http2Handler::on_connect()
     return r;
   }
   nghttp2_settings_entry entry[4];
-  size_t niv = 1;
+  size_t niv = 2;
+
   entry[0].settings_id = NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS;
   entry[0].value = 100;
+  entry[1].settings_id = NGHTTP2_SETTINGS_COMPRESS_DATA;
+  entry[1].value = 1;
+
   if(sessions_->get_config()->header_table_size >= 0) {
     entry[niv].settings_id = NGHTTP2_SETTINGS_HEADER_TABLE_SIZE;
     entry[niv].value = sessions_->get_config()->header_table_size;
@@ -769,7 +774,7 @@ int Http2Handler::verify_npn_result()
 }
 
 int Http2Handler::submit_file_response(const std::string& status,
-                                       int32_t stream_id,
+                                       Stream *stream,
                                        time_t last_modified,
                                        off_t file_length,
                                        nghttp2_data_provider *data_prd)
@@ -788,8 +793,8 @@ int Http2Handler::submit_file_response(const std::string& status,
     last_modified_str = util::http_date(last_modified);
     nva.push_back(http2::make_nv_ls("last-modified", last_modified_str));
   }
-  return nghttp2_submit_response(session_, stream_id, nva.data(), nva.size(),
+  return nghttp2_submit_response(session_, stream->stream_id,
-                                 data_prd);
+                                 nva.data(), nva.size(), data_prd);
 }
 
 int Http2Handler::submit_response
@@ -920,6 +925,19 @@ void Http2Handler::terminate_session(nghttp2_error_code error_code)
   nghttp2_session_terminate_session(session_, error_code);
 }
 
+void Http2Handler::decide_compression(const std::string& path, Stream *stream)
+{
+  if(nghttp2_session_get_remote_settings
+     (session_, NGHTTP2_SETTINGS_COMPRESS_DATA) == 1 &&
+     (util::endsWith(path, ".html") ||
+      util::endsWith(path, ".js") ||
+      util::endsWith(path, ".css") ||
+      util::endsWith(path, ".txt"))) {
+
+    stream->enable_compression = true;
+  }
+}
+
 ssize_t file_read_callback
 (nghttp2_session *session, int32_t stream_id,
  uint8_t *buf, size_t length, uint32_t *data_flags,
@@ -929,23 +947,52 @@ ssize_t file_read_callback
   auto stream = hd->get_stream(stream_id);
 
   int fd = source->fd;
-  ssize_t r;
+  ssize_t nread;
+  ssize_t rv;
 
-  while((r = read(fd, buf, length)) == -1 && errno == EINTR);
+  // Compressing too small data is not efficient?
-  if(r == -1) {
+  if(length >= 1024 && stream && stream->enable_compression) {
-    if(stream) {
+    uint8_t srcbuf[4096];
-      remove_stream_read_timeout(stream);
+    auto maxread = std::min(length, sizeof(srcbuf));
-      remove_stream_write_timeout(stream);
+
+    while((nread = read(fd, srcbuf, maxread)) == -1 && errno == EINTR);
+    if(nread == -1) {
+      if(stream) {
+        remove_stream_read_timeout(stream);
+        remove_stream_write_timeout(stream);
+      }
+
+      return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
     }
 
-    return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+    if(nread > 0) {
+      rv = deflate_data(buf, length, srcbuf, nread);
+
+      if(rv < 0) {
+        memcpy(buf, srcbuf, nread);
+      } else {
+        nread = rv;
+
+        *data_flags |= NGHTTP2_DATA_FLAG_COMPRESSED;
+      }
+    }
+  } else {
+    while((nread = read(fd, buf, length)) == -1 && errno == EINTR);
+    if(nread == -1) {
+      if(stream) {
+        remove_stream_read_timeout(stream);
+        remove_stream_write_timeout(stream);
+      }
+
+      return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+    }
   }
 
-  if(r == 0) {
+  if(nread == 0) {
     *data_flags |= NGHTTP2_DATA_FLAG_EOF;
   }
 
-  return r;
+  return nread;
 }
 
 namespace {
@@ -1064,7 +1111,9 @@ void prepare_response(Stream *stream, Http2Handler *hd, bool allow_push = true)
       if(last_mod_found && buf.st_mtime <= last_mod) {
         prepare_status_response(stream, hd, STATUS_304);
       } else {
-        hd->submit_file_response(STATUS_200, stream->stream_id, buf.st_mtime,
+        hd->decide_compression(path, stream);
+
+        hd->submit_file_response(STATUS_200, stream, buf.st_mtime,
                                  buf.st_size, &data_prd);
       }
     }

src/HttpServer.h

@@ -88,6 +88,7 @@ struct Stream {
   event *wtimer;
   int32_t stream_id;
   int file;
+  bool enable_compression;
   Stream(Http2Handler *handler, int32_t stream_id);
   ~Stream();
 };
@@ -109,7 +110,7 @@ public:
   int recvcb(uint8_t *buf, size_t len);
 
   int submit_file_response(const std::string& status,
-                           int32_t stream_id,
+                           Stream *stream,
                            time_t last_modified,
                            off_t file_length,
                            nghttp2_data_provider *data_prd);
@@ -139,6 +140,7 @@ public:
   void remove_settings_timer();
   void terminate_session(nghttp2_error_code error_code);
   int tls_handshake();
+  void decide_compression(const std::string& path, Stream *stream);
 private:
   int handle_ssl_temporal_error(int err);
   int tls_write(const uint8_t *data, size_t datalen);

src/app_helper.cc

@@ -43,6 +43,8 @@
 #include <iomanip>
 #include <fstream>
 
+#include <zlib.h>
+
 #include "app_helper.h"
 #include "util.h"
 #include "http2.h"
@@ -97,6 +99,8 @@ const char* strsettingsid(int32_t id)
     return "SETTINGS_MAX_CONCURRENT_STREAMS";
   case NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE:
     return "SETTINGS_INITIAL_WINDOW_SIZE";
+  case NGHTTP2_SETTINGS_COMPRESS_DATA:
+    return "SETTINGS_COMPRESS_DATA";
   default:
     return "UNKNOWN";
   }
@@ -574,4 +578,48 @@ std::chrono::steady_clock::time_point get_time()
   return std::chrono::steady_clock::now();
 }
 
+ssize_t deflate_data(uint8_t *out, size_t outlen,
+                     const uint8_t *in, size_t inlen)
+{
+  int rv;
+  z_stream zst;
+  uint8_t temp_out[8192];
+  auto temp_outlen = sizeof(temp_out);
+
+  zst.next_in = Z_NULL;
+  zst.zalloc = Z_NULL;
+  zst.zfree = Z_NULL;
+  zst.opaque = Z_NULL;
+
+  rv = deflateInit2(&zst, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
+                    31, 9, Z_DEFAULT_STRATEGY);
+
+  if(rv != Z_OK) {
+    return -1;
+  }
+
+  zst.avail_in = inlen;
+  zst.next_in = (uint8_t*)in;
+  zst.avail_out = temp_outlen;
+  zst.next_out = temp_out;
+
+  rv = deflate(&zst, Z_FINISH);
+
+  deflateEnd(&zst);
+
+  if(rv != Z_STREAM_END) {
+    return -1;
+  }
+
+  temp_outlen -= zst.avail_out;
+
+  if(temp_outlen > outlen) {
+    return -1;
+  }
+
+  memcpy(out, temp_out, temp_outlen);
+
+  return temp_outlen;
+}
+
 } // namespace nghttp2

src/app_helper.h

@@ -91,6 +91,9 @@ void set_color_output(bool f);
 // used.
 void set_output(FILE *file);
 
+ssize_t deflate_data(uint8_t *out, size_t outlen,
+                     const uint8_t *in, size_t inlen);
+
 } // namespace nghttp2
 
 #endif // APP_HELPER_H

src/nghttp.cc

@@ -99,6 +99,7 @@ struct Config {
   bool stat;
   bool upgrade;
   bool continuation;
+  bool compress_data;
   Config()
     : output_upper_thres(1024*1024),
       padding(0),
@@ -115,7 +116,8 @@ struct Config {
       get_assets(false),
       stat(false),
       upgrade(false),
-      continuation(false)
+      continuation(false),
+      compress_data(false)
   {
     nghttp2_option_new(&http2_option);
     nghttp2_option_set_peer_max_concurrent_streams
@@ -344,15 +346,21 @@ Config config;
 namespace {
 size_t populate_settings(nghttp2_settings_entry *iv)
 {
-  size_t niv = 2;
+  size_t niv = 3;
+
   iv[0].settings_id = NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS;
   iv[0].value = 100;
+
   iv[1].settings_id = NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;
   if(config.window_bits != -1) {
     iv[1].value = (1 << config.window_bits) - 1;
   } else {
     iv[1].value = NGHTTP2_INITIAL_WINDOW_SIZE;
   }
+
+  iv[2].settings_id = NGHTTP2_SETTINGS_COMPRESS_DATA;
+  iv[2].value = 1;
+
   if(config.header_table_size >= 0) {
     iv[niv].settings_id = NGHTTP2_SETTINGS_HEADER_TABLE_SIZE;
     iv[niv].value = config.header_table_size;
@@ -430,11 +438,13 @@ struct HttpClient {
   event *settings_timerev;
   addrinfo *addrs;
   addrinfo *next_addr;
+  nghttp2_gzip *inflater;
   // The number of completed requests, including failed ones.
   size_t complete;
   // The length of settings_payload
   size_t settings_payloadlen;
   client_state state;
+  int32_t last_inflate_error_stream_id;
   // The HTTP status code of the response message of HTTP Upgrade.
   unsigned int upgrade_response_status_code;
   // true if the response message of HTTP Upgrade request is fully
@@ -454,9 +464,11 @@ struct HttpClient {
       settings_timerev(nullptr),
       addrs(nullptr),
       next_addr(nullptr),
+      inflater(nullptr),
       complete(0),
       settings_payloadlen(0),
       state(STATE_IDLE),
+      last_inflate_error_stream_id(0),
       upgrade_response_status_code(0),
       upgrade_response_complete(false)
   {}
@@ -600,7 +612,7 @@ struct HttpClient {
     ssize_t rv;
     record_handshake_time();
     assert(!reqvec.empty());
-    nghttp2_settings_entry iv[16];
+    nghttp2_settings_entry iv[32];
     size_t niv = populate_settings(iv);
     assert(sizeof(settings_payload) >= 8*niv);
     rv = nghttp2_pack_settings_payload(settings_payload,
@@ -881,6 +893,38 @@ struct HttpClient {
   {
     stat.on_handshake_time = get_time();
   }
+
+  bool check_inflater(int32_t stream_id)
+  {
+    if(inflater == nullptr || last_inflate_error_stream_id == stream_id) {
+      return false;
+    }
+
+    last_inflate_error_stream_id = 0;
+
+    return true;
+  }
+
+  bool reset_inflater()
+  {
+    int rv;
+    nghttp2_gzip *gzip;
+
+    if(inflater) {
+      nghttp2_gzip_inflate_del(inflater);
+      inflater = nullptr;
+    }
+
+    rv = nghttp2_gzip_inflate_new(&gzip);
+
+    if(rv != 0) {
+      return false;
+    }
+
+    inflater = gzip;
+
+    return true;
+  }
 };
 } // namespace
 
@@ -1071,6 +1115,45 @@ int on_data_chunk_recv_callback
         data += tlen;
         len -= tlen;
       }
+    } else if(flags & NGHTTP2_FLAG_COMPRESSED) {
+      if(len == 0 || !client->check_inflater(stream_id)) {
+        return 0;
+      }
+
+      const size_t MAX_OUTLEN = 4096;
+      uint8_t out[MAX_OUTLEN];
+      size_t outlen;
+
+      do {
+        outlen = MAX_OUTLEN;
+        auto tlen = len;
+
+        int rv = nghttp2_gzip_inflate(client->inflater, out, &outlen,
+                                      data, &tlen);
+        if(rv != 0) {
+          goto per_frame_decomp_error;
+        }
+
+        if(!config.null_out) {
+          std::cout.write(reinterpret_cast<const char*>(out), outlen);
+        }
+
+        update_html_parser(client, req, out, outlen, 0);
+
+        data += tlen;
+        len -= tlen;
+
+        if(nghttp2_gzip_inflate_finished(client->inflater)) {
+          // When Z_STREAM_END was reached, remaining input length
+          // must be 0.
+          if(len > 0) {
+            goto per_frame_decomp_error;
+          }
+
+          break;
+        }
+      } while(len > 0 || outlen > 0);
+
     } else {
       if(!config.null_out) {
         std::cout.write(reinterpret_cast<const char*>(data), len);
@@ -1078,6 +1161,21 @@ int on_data_chunk_recv_callback
       update_html_parser(client, req, data, len, 0);
     }
   }
+
+  return 0;
+
+ per_frame_decomp_error:
+  // If per-frame decompression failed, we remember the stream ID so
+  // that subsequent chunk of DATA is ignored.
+  client->last_inflate_error_stream_id = stream_id;
+
+  if(!client->reset_inflater()) {
+    return NGHTTP2_ERR_CALLBACK_FAILURE;
+  }
+
+  nghttp2_submit_rst_stream(session, NGHTTP2_FLAG_NONE, stream_id,
+                            NGHTTP2_INTERNAL_ERROR);
+
   return 0;
 }
 } // namespace
@@ -1280,8 +1378,32 @@ namespace {
 int on_frame_recv_callback2
 (nghttp2_session *session, const nghttp2_frame *frame, void *user_data)
 {
+  int rv = 0;
+
   auto client = get_session(user_data);
   switch(frame->hd.type) {
+  case NGHTTP2_DATA:
+    if(frame->hd.flags & NGHTTP2_FLAG_COMPRESSED) {
+
+      auto inflate_finished = nghttp2_gzip_inflate_finished(client->inflater);
+
+      if(!client->reset_inflater()) {
+        rv = nghttp2_session_terminate_session(session,
+                                               NGHTTP2_INTERNAL_ERROR);
+
+        if(nghttp2_is_fatal(rv)) {
+          rv = NGHTTP2_ERR_CALLBACK_FAILURE;
+        } else {
+          rv = 0;
+        }
+      }
+      // Error if compressed block does not end in frame.
+      if(!inflate_finished) {
+        nghttp2_submit_rst_stream(session, NGHTTP2_FLAG_NONE,
+                                  frame->hd.stream_id, NGHTTP2_PROTOCOL_ERROR);
+      }
+    }
+    break;
   case NGHTTP2_HEADERS: {
     if(frame->headers.cat != NGHTTP2_HCAT_RESPONSE &&
        frame->headers.cat != NGHTTP2_HCAT_PUSH_RESPONSE) {
@@ -1359,7 +1481,7 @@ int on_frame_recv_callback2
   if(config.verbose) {
     verbose_on_frame_recv_callback(session, frame, user_data);
   }
-  return 0;
+  return rv;
 }
 } // namespace
 
@@ -1636,6 +1758,10 @@ int communicate(const std::string& scheme, const std::string& host,
   {
     HttpClient client{callbacks, evbase, ssl_ctx};
 
+    if(!client.reset_inflater()) {
+      goto fin;
+    }
+
     nghttp2_priority_spec pri_spec;
 
     if(config.weight != NGHTTP2_DEFAULT_WEIGHT) {
@@ -1690,19 +1816,47 @@ ssize_t file_read_callback
     (session, stream_id);
   assert(req);
   int fd = source->fd;
-  ssize_t r;
+  ssize_t nread;
-  while((r = pread(fd, buf, length, req->data_offset)) == -1 &&
+  ssize_t rv;
-        errno == EINTR);
+
-  if(r == -1) {
+  // Compressing too small data is not efficient?
-    return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+  if(length >= 1024 && config.compress_data &&
-  } else {
+     nghttp2_session_get_remote_settings
-    if(r == 0) {
+     (session, NGHTTP2_SETTINGS_COMPRESS_DATA) == 1) {
-      *data_flags |= NGHTTP2_DATA_FLAG_EOF;
+    uint8_t srcbuf[4096];
-    } else {
+    auto maxread = std::min(length, sizeof(srcbuf));
-      req->data_offset += r;
+
+    while((nread = read(fd, srcbuf, maxread)) == -1 && errno == EINTR);
+    if(nread == -1) {
+      return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+    }
+
+    if(nread > 0) {
+      rv = deflate_data(buf, length, srcbuf, nread);
+
+      if(rv < 0) {
+        memcpy(buf, srcbuf, nread);
+      } else {
+        nread = rv;
+
+        *data_flags |= NGHTTP2_DATA_FLAG_COMPRESSED;
+      }
+    }
+  } else {
+    while((nread = pread(fd, buf, length, req->data_offset)) == -1 &&
+          errno == EINTR);
+    if(nread == -1) {
+      return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
     }
-    return r;
   }
+
+  if(nread == 0) {
+    *data_flags |= NGHTTP2_DATA_FLAG_EOF;
+  } else {
+    req->data_offset += nread;
+  }
+
+  return nread;
 }
 } // namespace
 
@@ -1837,6 +1991,9 @@ Options:
                      be in PEM format.
   -d, --data=<FILE>  Post FILE to  server. If '-' is  given, data will
                      be read from stdin.
+  -g, --compress-data
+                     When used  with -d option, compress  request body
+                     on the fly using per-frame compression.
   -m, --multiply=<N> Request each URI <N> times.  By default, same URI
                      is not requested twice.   This option disables it
                      too.
@@ -1884,6 +2041,7 @@ int main(int argc, char **argv)
       {"help", no_argument, nullptr, 'h'},
       {"header", required_argument, nullptr, 'H'},
       {"data", required_argument, nullptr, 'd'},
+      {"compress-data", no_argument, nullptr, 'g'},
       {"multiply", required_argument, nullptr, 'm'},
       {"upgrade", no_argument, nullptr, 'u'},
       {"weight", required_argument, nullptr, 'p'},
@@ -1898,8 +2056,8 @@ int main(int argc, char **argv)
       {nullptr, 0, nullptr, 0 }
     };
     int option_index = 0;
-    int c = getopt_long(argc, argv, "M:Oab:c:d:m:np:hH:vst:uw:W:", long_options,
+    int c = getopt_long(argc, argv, "M:Oab:c:d:gm:np:hH:vst:uw:W:",
-                        &option_index);
+                        long_options, &option_index);
     char *end;
     if(c == -1) {
       break;
@@ -2003,6 +2161,9 @@ int main(int argc, char **argv)
     case 'd':
       config.datafile = strcmp("-", optarg) == 0 ? "/dev/stdin" : optarg;
       break;
+    case 'g':
+      config.compress_data = true;
+      break;
     case 'm':
       config.multiply = strtoul(optarg, nullptr, 10);
       break;