stream decoder

Author: kelbon

CMakeLists.txt

@@ -25,12 +25,11 @@ include(cmake/get_cpm.cmake)
 set(BOOST_INCLUDE_LIBRARIES intrusive)
 CPMAddPackage(
   NAME Boost
-  VERSION 1.84.0
+  VERSION 1.87.0
   URL https://github.com/boostorg/boost/releases/download/boost-1.84.0/boost-1.84.0.tar.xz
   OPTIONS "BOOST_ENABLE_CMAKE ON"
 )
 unset(BOOST_INCLUDE_LIBRARIES)
-find_package(Boost 1.84 COMPONENTS intrusive REQUIRED)
 
 endif()
 

include/hpack/basic_types.hpp

@@ -19,9 +19,11 @@ struct protocol_error : std::exception {
   }
 };
 
-}  // namespace hpack
-
-namespace hpack {
+// thrown if there are not enough data for reading header
+struct incomplete_data_error : hpack::protocol_error {
+  incomplete_data_error() : hpack::protocol_error("incomplete data") {
+  }
+};
 
 struct sym_info_t {
   uint32_t bits;

include/hpack/decoder.hpp

@@ -3,10 +3,16 @@
 #include "hpack/basic_types.hpp"
 #include "hpack/dynamic_table.hpp"
 
+#include <span>
 #include <utility>
 
 namespace hpack {
 
+// helper for `decode_string`. Temporal storage for decoded strings
+//
+// may store string_view (AND NOT OWN IT!), may store huffman decoded string
+// and in this case memory will be allocated and owned
+// tries to reuse memory when new huffman string setted and memory already allocated
 struct decoded_string {
  private:
   const char* data = nullptr;
@@ -15,10 +21,6 @@ struct decoded_string {
   // default -1 for removing ambiguity between 'not allocated' and 'allocated 1 byte' (log2(1) == 0)
   int8_t allocated_sz_log2 = -1;
 
-  friend void decode_string(In&, In, decoded_string&);
-
-  void set_huffman(const char* ptr, size_type len);
-
  public:
   decoded_string() = default;
 
@@ -43,12 +45,18 @@ struct decoded_string {
     return *this;
   }
 
+  void set_huffman(const char* ptr, size_type len);
+  // Note: *this will not own `ptr` memory, only contain a view
+  void set_not_huffman(const char* ptr, size_type len) {
+    reset();
+    data = ptr;
+    sz = len;
+  }
+
   // not huffman encoded string
   decoded_string& operator=(std::string_view str) noexcept {
     assert(std::in_range<size_type>(str.size()));
-    reset();
-    data = str.data();
-    sz = str.size();
+    set_not_huffman(str.data(), str.size());
     return *this;
   }
 
@@ -123,7 +131,6 @@ struct header_view {
   }
 };
 
-// precondition: in != e
 void decode_string(In& in, In e, decoded_string& out);
 
 struct decoder {
@@ -137,17 +144,82 @@ struct decoder {
 
   decoder(decoder&&) = default;
   decoder& operator=(decoder&&) noexcept = default;
+
   /*
    Note: this function ignores special 'cookie' header case
    https://www.rfc-editor.org/rfc/rfc7540#section-8.1.2.5
    and protocol error if decoded header name is not lowercase
   */
-  // precondition: in != e
   void decode_header(In& in, In e, header_view& out);
 
   // returns status code
   // its always first header of response, so 'in' must point to first byte of headers block
   int decode_response_status(In& in, In e);
 };
 
+// eats parts of headers fragment, allowing to parse CONTINUATIONS in HTTP/2 part by part
+struct stream_decoder {
+ private:
+  decoder& dec;
+  std::vector<byte_t> incomplete;
+
+  // returns where first unparsed byte starts
+  template <typename V>
+  In do_feed(std::span<byte_t> chunk, bool last_chunk, V&& visitor) {
+    In in = chunk.data();
+    In e = in + chunk.size();
+    assert(in != e);
+    In in_just_before_fail;
+    try {
+      header_view header;
+      while (in != e) {
+        in_just_before_fail = in;
+
+        dec.decode_header(in, e, header);
+
+        if (header) [[likely]]  // dynamic size update decoded without error
+          visitor(header.name.str(), header.value.str());
+      }
+      // successfully parsed all headers
+      return e;
+    } catch (hpack::incomplete_data_error&) {
+      if (last_chunk)
+        throw;
+    }
+    return in_just_before_fail;
+  }
+
+ public:
+  stream_decoder(decoder& d) noexcept : dec(d) {
+  }
+
+  stream_decoder(stream_decoder&&) = delete;
+  void operator=(stream_decoder&&) = delete;
+
+  // `visitor` should accept two string_views, name and value
+  // optimized for case when each `chunk` >> 1 header
+  template <typename V>
+  void feed(std::span<byte_t> chunk, bool last_chunk, V&& visitor) {
+    if (chunk.empty()) [[unlikely]]
+      return;
+    if (!incomplete.empty()) {
+      incomplete.insert(incomplete.end(), chunk.begin(), chunk.end());
+      In i = do_feed(incomplete, last_chunk, std::forward<V>(visitor));
+      In e = incomplete.data() + incomplete.size();
+      auto sz = e - i;
+      // avoid UB on .assign (iterators into vector itself)
+      memmove(incomplete.data(), i, sz);
+      incomplete.resize(sz);
+    } else {
+      In i = do_feed(chunk, last_chunk, std::forward<V>(visitor));
+      incomplete.assign(i, In(chunk.data()) + chunk.size());
+    }
+  }
+
+  // makes possible start from beginning, forgetting previous `feed` calls
+  void clear() noexcept {
+    incomplete.clear();
+  }
+};
+
 }  // namespace hpack

include/hpack/hpack.hpp

@@ -29,7 +29,7 @@ V decode_headers_block(decoder& dec, std::span<const byte_t> bytes, V visitor) {
   header_view header;
   while (in != e) {
     dec.decode_header(in, e, header);
-    if (header)  // dynamic size update decoded without error
+    if (header) [[likely]]  // dynamic size update decoded without error
       visitor(header.name.str(), header.value.str());
   }
   return visitor;

include/hpack/integers.hpp

@@ -1,7 +1,7 @@
 #pragma once
 
 #include <concepts>
-
+#include <cassert>
 #include "hpack/basic_types.hpp"
 
 namespace hpack {
@@ -49,17 +49,19 @@ O encode_integer(std::type_identity_t<UInt> I, uint8_t N, O _out) noexcept {
   return noexport::unadapt<O>(out);
 }
 
+// precondition: N <= 8
 template <std::unsigned_integral UInt = size_type>
-[[nodiscard]] size_type decode_integer(In& in, In e, uint8_t N) {
+[[nodiscard]] UInt decode_integer(In& in, In e, uint8_t N) {
+  assert(N <= 8);
   const UInt prefix_mask = (1 << N) - 1;
-  // get first N bits
   auto pull = [&] {
     if (in == e)
-      throw HPACK_PROTOCOL_ERROR(invalid integer representation);
+      throw incomplete_data_error();
     auto i = *in;
     ++in;
     return i;
   };
+  // get first N bits
   UInt I = pull() & prefix_mask;
   if (I < prefix_mask)
     return I;

src/decoder.cpp

@@ -132,8 +132,7 @@ static void decode_header_fully_indexed(In& in, In e, dynamic_table_t& dyntab, h
   out = entry;
 }
 
-// header with incremental indexing
-static void decode_header_cache(In& in, In e, dynamic_table_t& dyntab, header_view& out) {
+static void decode_header_incremental_indexing(In& in, In e, dynamic_table_t& dyntab, header_view& out) {
   assert(in != e && *in & 0b0100'0000);
   decode_header_impl(in, e, 6, dyntab, out);
   dyntab.add_entry(out.name.str(), out.value.str());
@@ -159,15 +158,15 @@ static size_type decode_dynamic_table_size_update(In& in, In e) {
 
 void decode_string(In& in, In e, decoded_string& out) {
   if (in == e)
-    throw HPACK_PROTOCOL_ERROR(incorrectly encoded string);
+    throw incomplete_data_error();
   bool is_huffman = *in & 0b1000'0000;
   size_type str_len = decode_integer(in, e, 7);
   if (str_len > std::distance(in, e))
-    throw HPACK_PROTOCOL_ERROR(size of encoded string not equal to data length);
+    throw incomplete_data_error();
   if (is_huffman)
     out.set_huffman((const char*)in, str_len);
   else
-    out = std::string_view((const char*)in, str_len);
+    out.set_not_huffman((const char*)in, str_len);
   in += str_len;
 }
 
@@ -176,7 +175,7 @@ void decoder::decode_header(In& in, In e, header_view& out) {
   if (*in & 0b1000'0000)
     return decode_header_fully_indexed(in, e, dyntab, out);
   if (*in & 0b0100'0000)
-    return decode_header_cache(in, e, dyntab, out);
+    return decode_header_incremental_indexing(in, e, dyntab, out);
   if (*in & 0b0010'0000) {
     dyntab.update_size(decode_dynamic_table_size_update(in, e));
     out.name.reset();

tests/test_hpack.cpp

@@ -479,6 +479,95 @@ TEST(tg_answer) {
   error_if(result != expected);
 }
 
+TEST(tg_answer_parts) {
+  std::vector<hpack::byte_t> bytes = {
+      0x88, 0x76, 0x89, 0xaa, 0x63, 0x55, 0xe5, 0x80, 0xae, 0x17, 0x97, 0x7,  0x61, 0x96, 0xc3, 0x61, 0xbe,
+      0x94, 0x3,  0x8a, 0x6e, 0x2d, 0x6a, 0x8,  0x2,  0x69, 0x40, 0x3b, 0x70, 0xf,  0x5c, 0x13, 0x4a, 0x62,
+      0xd1, 0xbf, 0x5f, 0x8b, 0x1d, 0x75, 0xd0, 0x62, 0xd,  0x26, 0x3d, 0x4c, 0x74, 0x41, 0xea, 0x5c, 0x4,
+      0x31, 0x39, 0x32, 0x36, 0x0,  0x91, 0x42, 0x6c, 0x31, 0x12, 0xb2, 0x6c, 0x1d, 0x48, 0xac, 0xf6, 0x25,
+      0x64, 0x14, 0x96, 0xd8, 0x64, 0xfa, 0xa0, 0xa4, 0x7e, 0x56, 0x1c, 0xc5, 0x81, 0x90, 0xb6, 0xcb, 0x80,
+      0x0,  0x3e, 0xd4, 0x35, 0x44, 0xa2, 0xd9, 0xb,  0xba, 0xd8, 0xef, 0x9e, 0x91, 0x9a, 0xa4, 0x7d, 0xa9,
+      0x5d, 0x85, 0xa0, 0xe3, 0x93, 0x0,  0x93, 0x19, 0x8,  0x54, 0x21, 0x62, 0x1e, 0xa4, 0xd8, 0x7a, 0x16,
+      0x1d, 0x14, 0x1f, 0xc2, 0xc7, 0xb0, 0xd3, 0x1a, 0xaf, 0x1,  0x2a, 0x0,  0x94, 0x19, 0x8,  0x54, 0x21,
+      0x62, 0x1e, 0xa4, 0xd8, 0x7a, 0x16, 0x1d, 0x14, 0x1f, 0xc2, 0xd4, 0x95, 0x33, 0x9e, 0x44, 0x7f, 0x90,
+      0xc5, 0x83, 0x7f, 0xd2, 0x9a, 0xf5, 0x6e, 0xdf, 0xf4, 0xa6, 0xad, 0x7b, 0xf2, 0x6a, 0xd3, 0xbb, 0x0,
+      0x94, 0x19, 0x8,  0x54, 0x21, 0x62, 0x1e, 0xa4, 0xd8, 0x7a, 0x16, 0x2f, 0x9a, 0xce, 0x82, 0xad, 0x39,
+      0x47, 0x21, 0x6c, 0x47, 0xa5, 0xbc, 0x7a, 0x92, 0x5a, 0x92, 0xb6, 0x72, 0xd5, 0x32, 0x67, 0xfa, 0xbc,
+      0x7a, 0x92, 0x5a, 0x92, 0xb6, 0xff, 0x55, 0x97, 0xea, 0xf8, 0xd2, 0x5f, 0xad, 0xc5, 0xb3, 0xb9, 0x6c,
+      0xfa, 0xbc, 0x7a, 0xaa, 0x29, 0x12, 0x63, 0xd5,
+  };
+  hpack::decoder e;
+  headers_t expected{
+      {":status", "200"},
+      {"server", "nginx/1.18.0"},
+      {"date", "Fri, 06 Sep 2024 07:08:24 GMT"},
+      {"content-type", "application/json"},
+      {"content-length", "1926"},
+      {"strict-transport-security", "max-age=31536000; includeSubDomains; preload"},
+      {"access-control-allow-origin", "*"},
+      {"access-control-allow-methods", "GET, POST, OPTIONS"},
+      {"access-control-expose-headers", "Content-Length,Content-Type,Date,Server,Connection"},
+  };
+  headers_t result;
+  hpack::stream_decoder decoder(e);
+  std::span<hpack::byte_t> chunks = bytes;
+  auto vtor = [&](std::string_view name, std::string_view value) {
+    result.push_back({std::string(name), std::string(value)});
+  };
+
+  while (!chunks.empty()) {
+    decoder.feed(chunks.subspan(0, 1), chunks.size() == 1, vtor);
+    chunks = chunks.subspan(1);
+  }
+  error_if(result != expected);
+}
+
+TEST(stream_decoder) {
+  headers_t expected{
+      {":status", "200"},
+      {"server", "nginx/1.18.0"},
+      {"longstr", std::string(15000, 'A')},
+      {"content-type", "application/json"},
+      {"content-length", "1926"},
+      {"strict-transport-security", "max-age=31536000; includeSubDomains; preload"},
+      {"access-control-allow-origin", "*"},
+      {"access-control-allow-methods", "GET, POST, OPTIONS"},
+      {"access-control-expose-headers", "Content-Length,Content-Type,Date,Server,Connection"},
+      {std::string(15000, 'B'), std::string(15000, 'A')},
+  };
+  std::mt19937 g(146);
+  using dist = std::uniform_int_distribution<>;
+  bytes_t bytes;
+  hpack::encoder enc;
+  // encode twice for using indexes
+  auto copy = expected;
+  expected.insert(expected.end(), copy.begin(), copy.end());
+  for (auto& [n, v] : expected) {
+    enc.encode(n, v, std::back_inserter(bytes));
+  }
+  hpack::decoder dec;
+
+  headers_t result;
+  hpack::stream_decoder decoder(dec);
+
+  auto vtor = [&](std::string_view name, std::string_view value) {
+    result.push_back({std::string(name), std::string(value)});
+  };
+  std::span<hpack::byte_t> chunks;
+  for (int i = 0; i < 100; ++i) {
+    chunks = bytes;
+    result.clear();
+    // `decoder` must be empty after each last_chunk decoder.clear();
+    while (!chunks.empty()) {
+      auto sz = dist(0, chunks.size())(g);
+      decoder.feed(chunks.subspan(0, sz), chunks.size() == sz, vtor);
+      chunks = chunks.subspan(sz);
+    }
+    g.discard(10);
+    error_if(result != expected);
+  }
+}
+
 TEST(decode_status) {
   hpack::encoder e;
   hpack::decoder de;
@@ -1013,6 +1102,8 @@ TEST(encode_with_cache) {
 }
 
 int main() {
+  test_stream_decoder();
+  test_tg_answer_parts();
   test_dyntable_move();
   test_encode_status();
   test_encode_with_cache();