FIX: Fix shared channel

Author: threecgreen

CHANGELOG.md

@@ -1,10 +1,13 @@
 # Changelog
 
-## 0.35.0 - TBD
+## 0.34.1 - TBD
 
 ### Enhancements
 - Added `InstitutionalPrioritization` variant to `MatchingAlgorithm`
 
+### Bug fixes
+- Improved memory usage of historical streaming requests (`TimeseriesGetRange`)
+
 ## 0.34.0 - 2025-04-22
 
 ### Enhancements

cmake/SourcesAndHeaders.cmake

@@ -7,11 +7,12 @@ set(headers
   include/databento/dbn_decoder.hpp
   include/databento/dbn_encoder.hpp
   include/databento/dbn_file_store.hpp
+  include/databento/detail/buffer.hpp
+  include/databento/detail/dbn_buffer_decoder.hpp
   include/databento/detail/http_client.hpp
   include/databento/detail/json_helpers.hpp
   include/databento/detail/scoped_fd.hpp
   include/databento/detail/scoped_thread.hpp
-  include/databento/detail/shared_channel.hpp
   include/databento/detail/tcp_client.hpp
   include/databento/detail/zstd_stream.hpp
   include/databento/enums.hpp
@@ -47,10 +48,11 @@ set(sources
   src/dbn_decoder.cpp
   src/dbn_encoder.cpp
   src/dbn_file_store.cpp
+  src/detail/buffer.cpp
+  src/detail/dbn_buffer_decoder.cpp
   src/detail/http_client.cpp
   src/detail/json_helpers.cpp
   src/detail/scoped_fd.cpp
-  src/detail/shared_channel.cpp
   src/detail/tcp_client.cpp
   src/detail/zstd_stream.cpp
   src/enums.cpp

include/databento/dbn_decoder.hpp

@@ -6,7 +6,6 @@
 #include <string>
 
 #include "databento/dbn.hpp"
-#include "databento/detail/shared_channel.hpp"
 #include "databento/enums.hpp"  // Upgrade Policy
 #include "databento/file_stream.hpp"
 #include "databento/ireadable.hpp"
@@ -18,7 +17,6 @@ namespace databento {
 // handled. Defaults to upgrading DBNv1 data to version 2 (the current version).
 class DbnDecoder {
  public:
-  DbnDecoder(ILogReceiver* log_receiver, detail::SharedChannel channel);
   DbnDecoder(ILogReceiver* log_receiver, InFileStream file_stream);
   DbnDecoder(ILogReceiver* log_receiver, std::unique_ptr<IReadable> input);
   DbnDecoder(ILogReceiver* log_receiver, std::unique_ptr<IReadable> input,
@@ -27,7 +25,8 @@ class DbnDecoder {
   static std::pair<std::uint8_t, std::size_t> DecodeMetadataVersionAndSize(
       const std::byte* buffer, std::size_t size);
   static Metadata DecodeMetadataFields(std::uint8_t version,
-                                       const std::vector<std::byte>& buffer);
+                                       const std::byte* buffer,
+                                       const std::byte* buffer_end);
   // Decodes a record possibly applying upgrading the data according to the
   // given version and upgrade policy. If an upgrade is applied,
   // compat_buffer is modified.
@@ -42,21 +41,17 @@ class DbnDecoder {
   const Record* DecodeRecord();
 
  private:
-  static std::string DecodeSymbol(
-      std::size_t symbol_cstr_len,
-      std::vector<std::byte>::const_iterator& buffer_it);
+  static std::string DecodeSymbol(std::size_t symbol_cstr_len,
+                                  const std::byte*& buffer);
   static std::vector<std::string> DecodeRepeatedSymbol(
-      std::size_t symbol_cstr_len,
-      std::vector<std::byte>::const_iterator& buffer_it,
-      std::vector<std::byte>::const_iterator buffer_end_it);
+      std::size_t symbol_cstr_len, const std::byte*& buffer,
+      const std::byte* buffer_end);
   static std::vector<SymbolMapping> DecodeSymbolMappings(
-      std::size_t symbol_cstr_len,
-      std::vector<std::byte>::const_iterator& buffer_it,
-      std::vector<std::byte>::const_iterator buffer_end_it);
-  static SymbolMapping DecodeSymbolMapping(
-      std::size_t symbol_cstr_len,
-      std::vector<std::byte>::const_iterator& buffer_it,
-      std::vector<std::byte>::const_iterator buffer_end_it);
+      std::size_t symbol_cstr_len, const std::byte*& buffer,
+      const std::byte* buffer_end);
+  static SymbolMapping DecodeSymbolMapping(std::size_t symbol_cstr_len,
+                                           const std::byte*& buffer,
+                                           const std::byte* buffer_end);
   bool DetectCompression();
   std::size_t FillBuffer();
   std::size_t GetReadBufferSize() const;

include/databento/detail/buffer.hpp

@@ -0,0 +1,56 @@
+#pragma once
+
+#include <cstddef>
+#include <memory>
+
+#include "databento/ireadable.hpp"
+#include "databento/iwritable.hpp"
+
+namespace databento::detail {
+class Buffer : public IReadable, public IWritable {
+ public:
+  Buffer() : Buffer(64 * std::size_t{1 << 10}) {}
+  explicit Buffer(std::size_t init_capacity)
+      : buf_{std::make_unique<std::byte[]>(init_capacity)},
+        end_{buf_.get() + init_capacity},
+        read_pos_{buf_.get()},
+        write_pos_{buf_.get()} {}
+
+  size_t Write(const char* data, std::size_t length);
+  size_t Write(const std::byte* data, std::size_t length);
+  void WriteAll(const char* data, std::size_t length);
+  void WriteAll(const std::byte* data, std::size_t length) override;
+
+  std::byte*& WriteBegin() { return write_pos_; }
+  std::byte* WriteEnd() const { return end_; }
+  std::size_t WriteCapacity() const {
+    return static_cast<std::size_t>(end_ - write_pos_);
+  }
+
+  /// Will throw if `length > ReadCapacity()`.
+  void ReadExact(std::byte* buffer, std::size_t length) override;
+  std::size_t ReadSome(std::byte* buffer, std::size_t max_length) override;
+
+  std::byte*& ReadBegin() { return read_pos_; }
+  std::byte* ReadEnd() const { return write_pos_; }
+  std::size_t ReadCapacity() const {
+    return static_cast<std::size_t>(write_pos_ - read_pos_);
+  }
+
+  std::size_t Capacity() const {
+    return static_cast<std::size_t>(end_ - buf_.get());
+  }
+  void Clear() {
+    read_pos_ = buf_.get();
+    write_pos_ = buf_.get();
+  }
+  void Reserve(std::size_t capacity);
+  void Shift();
+
+ private:
+  std::unique_ptr<std::byte[]> buf_;
+  std::byte* end_;
+  std::byte* read_pos_{};
+  std::byte* write_pos_{};
+};
+}  // namespace databento::detail

include/databento/detail/dbn_buffer_decoder.hpp

@@ -0,0 +1,49 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+
+#include "databento/detail/buffer.hpp"
+#include "databento/detail/zstd_stream.hpp"
+#include "databento/ireadable.hpp"
+#include "databento/record.hpp"
+#include "databento/timeseries.hpp"
+
+namespace databento::detail {
+class DbnBufferDecoder {
+ public:
+  // The instance cannot outlive the lifetime of these references.
+  DbnBufferDecoder(const MetadataCallback& metadata_callback,
+                   const RecordCallback& record_callback)
+      : metadata_callback_{metadata_callback},
+        record_callback_{record_callback},
+        zstd_stream_{InitZstdBuffer()} {}
+
+  KeepGoing Process(const char* data, std::size_t length);
+
+ private:
+  enum class DecoderState : std::uint8_t {
+    Init,
+    Metadata,
+    Records,
+  };
+
+  std::unique_ptr<IReadable> InitZstdBuffer() {
+    auto zstd_buffer = std::make_unique<Buffer>();
+    zstd_buffer_ = zstd_buffer.get();
+    return zstd_buffer;
+  }
+
+  const MetadataCallback& metadata_callback_;
+  const RecordCallback& record_callback_;
+  ZstdDecodeStream zstd_stream_;
+  Buffer* zstd_buffer_;
+  Buffer dbn_buffer_{};
+  std::size_t bytes_needed_{};
+  alignas(RecordHeader) std::array<std::byte, kMaxRecordLen> compat_buffer_{};
+  std::uint8_t input_version_{};
+  bool ts_out_;
+  DecoderState state_{DecoderState::Init};
+};
+}  // namespace databento::detail

include/databento/detail/shared_channel.hpp

@@ -9,8 +9,9 @@
 #include <utility>  // pair
 #include <vector>
 
-#include "databento/datetime.hpp"           // UnixNanos
-#include "databento/dbn.hpp"                // Metadata
+#include "databento/datetime.hpp"  // UnixNanos
+#include "databento/dbn.hpp"       // Metadata
+#include "databento/detail/buffer.hpp"
 #include "databento/detail/tcp_client.hpp"  // TcpClient
 #include "databento/enums.hpp"  // Schema, SType, VersionUpgradePolicy
 #include "databento/live_subscription.hpp"
@@ -118,10 +119,7 @@ class LiveBlocking {
   detail::TcpClient client_;
   std::uint32_t sub_counter_{};
   std::vector<LiveSubscription> subscriptions_;
-  // Must be 8-byte aligned for records
-  alignas(RecordHeader) std::array<std::byte, kMaxStrLen> read_buffer_{};
-  std::size_t buffer_size_{};
-  std::size_t buffer_idx_{};
+  detail::Buffer buffer_{};
   // Must be 8-byte aligned for records
   alignas(RecordHeader) std::array<std::byte, kMaxRecordLen> compat_buffer_{};
   std::uint64_t session_id_;