OPT: Make Buffer shift explicit in C++

Author: threecgreen

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Enhancements
 - Added new publisher values for Cboe Titanium Cboe Global Indices Feed
+- Made `detail::Buffer` shifts explicit to avoid redundant moves during record decoding
 - Added `Year` to `SplitDuration` enum for yearly historical batch job submissions
 
 ## 0.54.0 - 2026-04-21

include/databento/detail/buffer.hpp

@@ -46,14 +46,7 @@ class Buffer : public IReadable, public IWritable {
   std::byte* ReadEnd() { return write_pos_; }
   const std::byte* ReadBegin() const { return read_pos_; }
   const std::byte* ReadEnd() const { return write_pos_; }
-  // Indicate how many bytes were read
-  void Consume(std::size_t length) {
-    read_pos_ += length;
-    if (static_cast<std::size_t>(read_pos_ - buf_.get()) > (Capacity() / 2)) {
-      Shift();
-    }
-  }
-  void ConsumeNoShift(std::size_t length) { read_pos_ += length; }
+  void Consume(std::size_t length) { read_pos_ += length; }
   std::size_t ReadCapacity() const {
     return static_cast<std::size_t>(write_pos_ - read_pos_);
   }
@@ -65,6 +58,13 @@ class Buffer : public IReadable, public IWritable {
   }
   void Reserve(std::size_t capacity);
   void Shift();
+  // Shifts unread data to offset 0 if writable space is less than `needed`,
+  // reclaiming the consumed prefix. Does not grow the buffer.
+  void ShiftForSpace(std::size_t needed) {
+    if (WriteCapacity() < needed && read_pos_ != buf_.get()) {
+      Shift();
+    }
+  }
 
   friend std::ostream& operator<<(std::ostream& stream, const Buffer& buffer);
 

src/dbn_decoder.cpp

@@ -293,16 +293,14 @@ const databento::Record* DbnDecoder::DecodeRecord() {
     }
   }
   current_record_ = Record{BufferRecordHeader()};
-  buffer_.ConsumeNoShift(current_record_.Size());
+  buffer_.Consume(current_record_.Size());
   current_record_ = DbnDecoder::DecodeRecordCompat(version_, upgrade_policy_, ts_out_,
                                                    &compat_buffer_, current_record_);
   return &current_record_;
 }
 
 size_t DbnDecoder::FillBuffer() {
-  if (buffer_.WriteCapacity() < kMaxRecordLen) {
-    buffer_.Shift();
-  }
+  buffer_.ShiftForSpace(kMaxRecordLen);
   const auto fill_size =
       input_->ReadSome(buffer_.WriteBegin(), buffer_.WriteCapacity());
   buffer_.Fill(fill_size);

src/detail/buffer.cpp

@@ -13,9 +13,7 @@ size_t Buffer::Write(const char* data, std::size_t length) {
   return Write(reinterpret_cast<const std::byte*>(data), length);
 }
 size_t Buffer::Write(const std::byte* data, std::size_t length) {
-  if (length > WriteCapacity()) {
-    Shift();
-  }
+  ShiftForSpace(length);
   const auto write_size = std::min(WriteCapacity(), length);
   std::copy(data, data + write_size, WriteBegin());
   Fill(write_size);
@@ -28,8 +26,8 @@ void Buffer::WriteAll(const char* data, std::size_t length) {
 void Buffer::WriteAll(const std::byte* data, std::size_t length) {
   if (length > Capacity() - ReadCapacity()) {
     Reserve(ReadCapacity() + length);
-  } else if (length >= WriteCapacity()) {
-    Shift();
+  } else {
+    ShiftForSpace(length);
   }
   std::copy(data, data + length, WriteBegin());
   write_pos_ += length;

src/detail/dbn_buffer_decoder.cpp

@@ -10,6 +10,7 @@ using databento::detail::DbnBufferDecoder;
 databento::KeepGoing DbnBufferDecoder::Process(const char* data, std::size_t length) {
   zstd_buffer_->WriteAll(data, length);
   while (true) {
+    dbn_buffer_.ShiftForSpace(kMaxRecordLen);
     const auto read_size =
         zstd_stream_.ReadSome(dbn_buffer_.WriteBegin(), dbn_buffer_.WriteCapacity());
     dbn_buffer_.Fill(read_size);

src/live_blocking.cpp

@@ -482,7 +482,7 @@ databento::IReadable::Result LiveBlocking::FillBuffer() {
 
 databento::IReadable::Result LiveBlocking::FillBuffer(
     std::chrono::milliseconds timeout) {
-  buffer_.Shift();
+  buffer_.ShiftForSpace(kMaxRecordLen);
   const auto read_res =
       connection_.ReadSome(buffer_.WriteBegin(), buffer_.WriteCapacity(), timeout);
   buffer_.Fill(read_res.read_size);
@@ -494,7 +494,7 @@ databento::IReadable::Result LiveBlocking::FillBuffer(
 
 const databento::Record* LiveBlocking::ConsumeBufferedRecord() {
   current_record_ = Record{BufferRecordHeader()};
-  buffer_.ConsumeNoShift(current_record_.Size());
+  buffer_.Consume(current_record_.Size());
   current_record_ = DbnDecoder::DecodeRecordCompat(
       version_, upgrade_policy_, send_ts_out_, &compat_buffer_, current_record_);
   return &current_record_;

tests/src/buffer_tests.cpp

@@ -11,7 +11,7 @@ namespace databento::detail::tests {
 TEST(BufferTests, TestWriteAllPastCapacity) {
   Buffer target{10};
   target.Fill(4);
-  target.ConsumeNoShift(2);
+  target.Consume(2);
   ASSERT_EQ(target.WriteCapacity(), 6);
   ASSERT_EQ(target.ReadCapacity(), 2);
   ASSERT_EQ(target.Capacity(), 10);
@@ -25,7 +25,7 @@ TEST(BufferTests, TestWriteAllPastCapacity) {
 TEST(BufferTests, TestWriteAllShift) {
   Buffer target{20};
   target.WriteAll("TestWriteAllShift", 17);
-  target.ConsumeNoShift(4);
+  target.Consume(4);
   ASSERT_EQ(target.WriteCapacity(), 3);
   ASSERT_EQ(target.ReadCapacity(), 13);
   ASSERT_EQ(target.Capacity(), 20);
@@ -39,7 +39,7 @@ TEST(BufferTests, TestWriteAllShift) {
 TEST(BufferTests, TestWriteRead) {
   Buffer target{10};
   target.Fill(5);
-  target.ConsumeNoShift(5);
+  target.Consume(5);
   const auto write_len = target.Write("BufferTests", 11);
   ASSERT_EQ(write_len, 10);
   std::array<std::byte, 10> read_buf{};
@@ -54,16 +54,42 @@ TEST(BufferTests, TestReserve) {
   ASSERT_EQ(target.ReadCapacity(), 0);
   ASSERT_EQ(target.Capacity(), 120);
   target.WriteAll("TestReserve", 11);
-  target.ConsumeNoShift(4);
+  target.Consume(4);
 }
 
-TEST(BufferTests, TestConsumeShift) {
+TEST(BufferTests, TestConsumeDoesNotShift) {
+  Buffer target{16};
+  target.Fill(12);
+  target.Consume(10);
+  ASSERT_EQ(target.ReadCapacity(), 2);
+  ASSERT_EQ(target.WriteCapacity(), 4);
+}
+
+TEST(BufferTests, TestShiftForSpace) {
   Buffer target{120};
-  target.Fill(120);
-  ASSERT_EQ(target.WriteCapacity(), 0);
-  target.ConsumeNoShift(100);
-  ASSERT_EQ(target.WriteCapacity(), 0);
-  target.Consume(1);
-  ASSERT_EQ(target.WriteCapacity(), 101);
+  target.Fill(40);
+  target.Consume(20);
+  ASSERT_EQ(target.WriteCapacity(), 80);
+  ASSERT_EQ(target.ReadCapacity(), 20);
+  // Writable space is sufficient: no shift
+  target.ShiftForSpace(50);
+  ASSERT_EQ(target.WriteCapacity(), 80);
+  ASSERT_EQ(target.ReadCapacity(), 20);
+  // Writable space is insufficient: reclaim the consumed prefix
+  target.ShiftForSpace(100);
+  ASSERT_EQ(target.WriteCapacity(), 100);
+  ASSERT_EQ(target.ReadCapacity(), 20);
+  // Nothing left to reclaim; shift is a no-op
+  target.ShiftForSpace(1000);
+  ASSERT_EQ(target.WriteCapacity(), 100);
+  ASSERT_EQ(target.ReadCapacity(), 20);
+}
+
+TEST(BufferTests, TestShiftForSpaceNoopWhenUnconsumed) {
+  Buffer target{16};
+  target.Fill(4);
+  target.ShiftForSpace(1000);
+  ASSERT_EQ(target.WriteCapacity(), 12);
+  ASSERT_EQ(target.ReadCapacity(), 4);
 }
 }  // namespace databento::detail::tests