stream decoder (#15)

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

@@ -1,9 +1,9 @@
 #pragma once
 
+#include <algorithm>
 #include <cstdint>
 #include <iterator>
 #include <string_view>
-
 #include <exception>
 
 namespace hpack {
@@ -19,9 +19,15 @@ struct protocol_error : std::exception {
   }
 };
 
-}  // namespace hpack
+// thrown if there are not enough data for reading header
+struct incomplete_data_error : hpack::protocol_error {
+  // approx value - how many bytes need to be readen for receiving next part (int or string)
+  size_t required_bytes = 0;
 
-namespace hpack {
+  explicit incomplete_data_error(size_t required_bytes_approx)
+      : hpack::protocol_error("incomplete data"), required_bytes(required_bytes_approx) {
+  }
+};
 
 struct sym_info_t {
   uint32_t bits;
@@ -46,6 +52,8 @@ concept Out = std::output_iterator<T, byte_t>;
 
 namespace noexport {
 
+// caches first byte for avoiding *it = x == push_back,
+// so *it | mask will be into next byte (may be with back_inserter)
 template <typename T>
 struct adapted_output_iterator {
   T base_it;
@@ -100,6 +108,54 @@ Original unadapt(byte_t* ptr) {
   return reinterpret_cast<Original>(ptr);
 }
 
+// standard interface (e.g. for vector) for inserting many values at back
+// Note: ignores fact, that someone can make super-bad type with push_back + insert making something wrong
+template <typename T>
+constexpr inline bool can_insert_many =
+    requires(T& value, const char* p) { value.insert(value.end(), p, p); };
+
+// standard back_insert_iterator rly uses protected field exactly for such accessing
+template <typename C>
+C& access_protected_container(std::back_insert_iterator<C> c) {
+  static_assert(std::is_trivially_copyable_v<decltype(c)>);
+  struct accessor : std::back_insert_iterator<C> {
+    C* get() noexcept {
+      return this->container;
+    }
+  };
+  return *accessor{c}.get();
+}
+
+template <typename C>
+  requires(can_insert_many<C>)
+std::back_insert_iterator<C> do_copy_n_fast(const char* ptr, size_t sz, std::back_insert_iterator<C> it) {
+  auto& c = access_protected_container(it);
+  c.insert(c.end(), ptr, ptr + sz);
+  return it;  // back insert iterator does not change on ++/* etc
+}
+
+template <typename C>
+  requires(can_insert_many<C>)
+adapted_output_iterator<std::back_insert_iterator<C>> do_copy_n_fast(
+    const char* ptr, size_t sz, adapted_output_iterator<std::back_insert_iterator<C>> it) {
+  auto& c = access_protected_container(it.base_it);
+  c.insert(c.end(), ptr, ptr + sz);
+  return it;  // adapted iterator must be unchanged, since base_it unchanged (its back inserter)
+}
+
+// fallback
+template <typename It>
+It do_copy_n_fast(const char* ptr, size_t sz, It it) {
+  return std::copy_n(ptr, sz, std::move(it));
+}
+
+// makes copy_n, but for back_insert iterator makes insert(end, It, It + n)
+// this converts many push_backs into one uninitialized_copy_n
+template <typename It>
+It copy_n_fast(const char* ptr, size_t sz, It it) {
+  return do_copy_n_fast(ptr, sz, std::move(it));
+}
+
 }  // namespace noexport
 
 struct table_entry {

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,96 @@ 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, size_t& approx) {
+    In in = chunk.data();
+    In e = in + chunk.size();
+    assert(in != e);
+    In in_just_before_fail;
+    approx = 0;
+    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& e) {
+      approx = e.required_bytes;
+      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
+  // returns approx count of bytes required for receiving next part of header
+  // or 0 if there are no unhandled data in chunk
+  // e.g. may be used to detect too big string before receiving it
+  template <typename V>
+  size_t feed(std::span<byte_t> chunk, bool last_chunk, V&& visitor) {
+    if (chunk.empty()) [[unlikely]]
+      return 0;
+    size_t approx;
+    if (!incomplete.empty()) {
+      incomplete.insert(incomplete.end(), chunk.begin(), chunk.end());
+      In i = do_feed(incomplete, last_chunk, std::forward<V>(visitor), approx);
+      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), approx);
+      incomplete.assign(i, In(chunk.data()) + chunk.size());
+    }
+    return approx;
+  }
+
+  // returns such value, that `pending_data_size` + `feed` result == almost exact value of bytes which will be
+  // stored until next part of header (not header itself!) will be parsed.
+  // Note, there are only 2 parts of header - name and value
+  [[nodiscard]] size_t pending_data_size() const noexcept {
+    return incomplete.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(2);
     auto i = *in;
     ++in;
     return i;
   };
+  // get first N bits
   UInt I = pull() & prefix_mask;
   if (I < prefix_mask)
     return I;

include/hpack/strings.hpp

@@ -57,7 +57,7 @@ O encode_string(std::string_view str, O _out) {
   if constexpr (!Huffman) {
     *out = 0;  // set H bit to 0
     out = encode_integer(str.size(), 7, out);
-    out = std::copy_n(str.data(), str.size(), out);
+    out = noexport::copy_n_fast(str.data(), str.size(), out);
   } else {
     out = encode_string_huffman(str, out);
   }

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(1);
   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(str_len - std::distance(in, e));
   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();