Allocate the size of stream sequencer buffer block pointer container on demand. This should save 16K bytes memory per stream on the server side, as currently blocks_ in QuicStreamSequencerBuffer starts with 2048 nullptr as soon as there is any data to read. Protected by quic_reloadable_flag_quic_allocate_stream_sequencer_buffer_blocks_on_demand. PiperOrigin-RevId: 341541273 Change-Id: I882d09cc8b41a0874e94b46958b04562fa96e619
diff --git a/quic/core/quic_flags_list.h b/quic/core/quic_flags_list.h index 7b02787..442e5ea 100644 --- a/quic/core/quic_flags_list.h +++ b/quic/core/quic_flags_list.h
@@ -8,6 +8,7 @@ QUIC_FLAG(FLAGS_quic_reloadable_flag_quic_abort_qpack_on_stream_close, true) QUIC_FLAG(FLAGS_quic_reloadable_flag_quic_ack_delay_alarm_granularity, false) QUIC_FLAG(FLAGS_quic_reloadable_flag_quic_add_stream_info_to_idle_close_detail, true) +QUIC_FLAG(FLAGS_quic_reloadable_flag_quic_allocate_stream_sequencer_buffer_blocks_on_demand, false) QUIC_FLAG(FLAGS_quic_reloadable_flag_quic_allow_client_enabled_bbr_v2, false) QUIC_FLAG(FLAGS_quic_reloadable_flag_quic_bbr2_avoid_too_low_probe_bw_cwnd, false) QUIC_FLAG(FLAGS_quic_reloadable_flag_quic_bbr2_fewer_startup_round_trips, false)
diff --git a/quic/core/quic_stream_sequencer_buffer.cc b/quic/core/quic_stream_sequencer_buffer.cc index b4143ef..9208de6 100644 --- a/quic/core/quic_stream_sequencer_buffer.cc +++ b/quic/core/quic_stream_sequencer_buffer.cc
@@ -4,6 +4,9 @@ #include "net/third_party/quiche/src/quic/core/quic_stream_sequencer_buffer.h" +#include <algorithm> +#include <cstddef> +#include <memory> #include <string> #include "absl/strings/string_view.h" @@ -32,9 +35,13 @@ QuicStreamSequencerBuffer::QuicStreamSequencerBuffer(size_t max_capacity_bytes) : max_buffer_capacity_bytes_(max_capacity_bytes), - blocks_count_(CalculateBlockCount(max_capacity_bytes)), + max_blocks_count_(CalculateBlockCount(max_capacity_bytes)), + current_blocks_count_(0u), total_bytes_read_(0), blocks_(nullptr) { + if (allocate_blocks_on_demand_) { + DCHECK_GE(max_blocks_count_, kInitialBlockCount); + } Clear(); } @@ -44,7 +51,9 @@ void QuicStreamSequencerBuffer::Clear() { if (blocks_ != nullptr) { - for (size_t i = 0; i < blocks_count_; ++i) { + size_t blocks_to_clear = + allocate_blocks_on_demand_ ? current_blocks_count_ : max_blocks_count_; + for (size_t i = 0; i < blocks_to_clear; ++i) { if (blocks_[i] != nullptr) { RetireBlock(i); } @@ -66,6 +75,35 @@ return true; } +void QuicStreamSequencerBuffer::MaybeAddMoreBlocks(size_t next_expected_byte) { + if (current_blocks_count_ == max_blocks_count_) { + return; + } + size_t last_byte = next_expected_byte - 1; + size_t num_of_blocks_needed; + // As long as last_byte does not wrap around, its index plus one blocks are + // needed. Otherwise, block_count_ blocks are needed. + if (last_byte < max_buffer_capacity_bytes_) { + num_of_blocks_needed = + std::max(GetBlockIndex(last_byte) + 1, kInitialBlockCount); + } else { + num_of_blocks_needed = max_blocks_count_; + } + if (current_blocks_count_ >= num_of_blocks_needed) { + return; + } + size_t new_block_count_ = + std::clamp(kBlocksGrowthFactor * current_blocks_count_, + num_of_blocks_needed, max_blocks_count_); + auto new_blocks = std::make_unique<BufferBlock*[]>(new_block_count_); + if (blocks_ != nullptr) { + memcpy(new_blocks.get(), blocks_.get(), + current_blocks_count_ * sizeof(BufferBlock*)); + } + blocks_ = std::move(new_blocks); + current_blocks_count_ = new_block_count_; +} + QuicErrorCode QuicStreamSequencerBuffer::OnStreamData( QuicStreamOffset starting_offset, absl::string_view data, @@ -83,6 +121,12 @@ *error_details = "Received data beyond available range."; return QUIC_INTERNAL_ERROR; } + if (allocate_blocks_on_demand_) { + QUIC_RELOADABLE_FLAG_COUNT( + quic_allocate_stream_sequencer_buffer_blocks_on_demand); + MaybeAddMoreBlocks(starting_offset + size); + } + if (bytes_received_.Empty() || starting_offset >= bytes_received_.rbegin()->max() || bytes_received_.IsDisjoint(QuicInterval<QuicStreamOffset>( @@ -150,7 +194,9 @@ while (source_remaining > 0) { const size_t write_block_num = GetBlockIndex(offset); const size_t write_block_offset = GetInBlockOffset(offset); - DCHECK_GT(blocks_count_, write_block_num); + size_t current_blocks_count = + allocate_blocks_on_demand_ ? current_blocks_count_ : max_blocks_count_; + DCHECK_GT(current_blocks_count, write_block_num); size_t block_capacity = GetBlockCapacity(write_block_num); size_t bytes_avail = block_capacity - write_block_offset; @@ -161,19 +207,21 @@ bytes_avail = total_bytes_read_ + max_buffer_capacity_bytes_ - offset; } - if (blocks_ == nullptr) { - blocks_.reset(new BufferBlock*[blocks_count_]()); - for (size_t i = 0; i < blocks_count_; ++i) { - blocks_[i] = nullptr; + if (!allocate_blocks_on_demand_) { + if (blocks_ == nullptr) { + blocks_.reset(new BufferBlock*[max_blocks_count_]()); + for (size_t i = 0; i < max_blocks_count_; ++i) { + blocks_[i] = nullptr; + } } } - if (write_block_num >= blocks_count_) { + if (write_block_num >= current_blocks_count) { *error_details = quiche::QuicheStrCat( "QuicStreamSequencerBuffer error: OnStreamData() exceed array bounds." "write offset = ", offset, " write_block_num = ", write_block_num, - " blocks_count_ = ", blocks_count_); + " current_blocks_count_ = ", current_blocks_count); return false; } if (blocks_ == nullptr) { @@ -307,14 +355,14 @@ // before gap is met or |iov| is filled. For these blocks, one whole block is // a region. int iov_used = 1; - size_t block_idx = (start_block_idx + iov_used) % blocks_count_; + size_t block_idx = (start_block_idx + iov_used) % max_blocks_count_; while (block_idx != end_block_idx && iov_used < iov_len) { DCHECK(nullptr != blocks_[block_idx]); iov[iov_used].iov_base = blocks_[block_idx]->buffer; iov[iov_used].iov_len = GetBlockCapacity(block_idx); QUIC_DVLOG(1) << "Got block with index: " << block_idx; ++iov_used; - block_idx = (start_block_idx + iov_used) % blocks_count_; + block_idx = (start_block_idx + iov_used) % max_blocks_count_; } // Deal with last block if |iov| can hold more. @@ -397,6 +445,7 @@ void QuicStreamSequencerBuffer::ReleaseWholeBuffer() { Clear(); + current_blocks_count_ = 0; blocks_.reset(nullptr); } @@ -472,7 +521,7 @@ } size_t QuicStreamSequencerBuffer::GetBlockCapacity(size_t block_index) const { - if ((block_index + 1) == blocks_count_) { + if ((block_index + 1) == max_blocks_count_) { size_t result = max_buffer_capacity_bytes_ % kBlockSizeBytes; if (result == 0) { // whole block result = kBlockSizeBytes;
diff --git a/quic/core/quic_stream_sequencer_buffer.h b/quic/core/quic_stream_sequencer_buffer.h index 98ffa9c..66acc53 100644 --- a/quic/core/quic_stream_sequencer_buffer.h +++ b/quic/core/quic_stream_sequencer_buffer.h
@@ -82,7 +82,14 @@ // Size of blocks used by this buffer. // Choose 8K to make block large enough to hold multiple frames, each of // which could be up to 1.5 KB. - static const size_t kBlockSizeBytes = 8 * 1024; // 8KB + static constexpr size_t kBlockSizeBytes = 8 * 1024; // 8KB + + // Number of blocks allocated initially. + static constexpr size_t kInitialBlockCount = 8u; + + // How fast block pointers container grow in size. + // Choose 4 to reduce the amount of reallocation. + static constexpr int kBlocksGrowthFactor = 4; // The basic storage block used by this buffer. struct QUIC_EXPORT_PRIVATE BufferBlock { @@ -212,17 +219,29 @@ // Return all received frames as a string. std::string ReceivedFramesDebugString() const; + // Resize blocks_ if more blocks are needed to accomodate bytes before + // next_expected_byte. + void MaybeAddMoreBlocks(size_t next_expected_byte); + // The maximum total capacity of this buffer in byte, as constructed. size_t max_buffer_capacity_bytes_; - // How many blocks this buffer would need when it reaches full capacity. - size_t blocks_count_; + // Number of blocks this buffer would have when it reaches full capacity, + // i.e., maximal number of blocks in blocks_. + size_t max_blocks_count_; + + // Number of blocks this buffer currently has. + size_t current_blocks_count_; // Number of bytes read out of buffer. QuicStreamOffset total_bytes_read_; + // Whether size of blocks_ grows on demand. + bool allocate_blocks_on_demand_ = GetQuicReloadableFlag( + quic_allocate_stream_sequencer_buffer_blocks_on_demand); + // An ordered, variable-length list of blocks, with the length limited - // such that the number of blocks never exceeds blocks_count_. + // such that the number of blocks never exceeds max_blocks_count_. // Each list entry can hold up to kBlockSizeBytes bytes. std::unique_ptr<BufferBlock*[]> blocks_;
diff --git a/quic/core/quic_stream_sequencer_buffer_test.cc b/quic/core/quic_stream_sequencer_buffer_test.cc index f12810b..e6a8914 100644 --- a/quic/core/quic_stream_sequencer_buffer_test.cc +++ b/quic/core/quic_stream_sequencer_buffer_test.cc
@@ -72,10 +72,10 @@ helper_ = std::make_unique<QuicStreamSequencerBufferPeer>((buffer_.get())); } - // Use 2.5 here to make sure the buffer has more than one block and its end - // doesn't align with the end of a block in order to test all the offset - // calculation. - size_t max_capacity_bytes_ = 2.5 * kBlockSizeBytes; + // Use 8.5 here to make sure that the buffer has more than + // QuicStreamSequencerBuffer::kInitialBlockCount block and its end doesn't + // align with the end of a block in order to test all the offset calculation. + size_t max_capacity_bytes_ = 8.5 * kBlockSizeBytes; std::unique_ptr<QuicStreamSequencerBuffer> buffer_; std::unique_ptr<QuicStreamSequencerBufferPeer> helper_; @@ -86,18 +86,18 @@ TEST_F(QuicStreamSequencerBufferTest, InitializeWithMaxRecvWindowSize) { ResetMaxCapacityBytes(16 * 1024 * 1024); // 16MB EXPECT_EQ(2 * 1024u, // 16MB / 8KB = 2K - helper_->block_count()); + helper_->max_blocks_count()); EXPECT_EQ(max_capacity_bytes_, helper_->max_buffer_capacity()); EXPECT_TRUE(helper_->CheckInitialState()); } TEST_F(QuicStreamSequencerBufferTest, InitializationWithDifferentSizes) { - const size_t kCapacity = 2 * QuicStreamSequencerBuffer::kBlockSizeBytes; + const size_t kCapacity = 16 * QuicStreamSequencerBuffer::kBlockSizeBytes; ResetMaxCapacityBytes(kCapacity); EXPECT_EQ(max_capacity_bytes_, helper_->max_buffer_capacity()); EXPECT_TRUE(helper_->CheckInitialState()); - const size_t kCapacity1 = 8 * QuicStreamSequencerBuffer::kBlockSizeBytes; + const size_t kCapacity1 = 32 * QuicStreamSequencerBuffer::kBlockSizeBytes; ResetMaxCapacityBytes(kCapacity1); EXPECT_EQ(kCapacity1, helper_->max_buffer_capacity()); EXPECT_TRUE(helper_->CheckInitialState()); @@ -399,7 +399,7 @@ // Try to write from [max_capacity_bytes_ - 0.5 * kBlockSizeBytes, // max_capacity_bytes_ + 512 + 1). But last bytes exceeds current capacity. source = std::string(0.5 * kBlockSizeBytes + 512 + 1, 'b'); - EXPECT_THAT(buffer_->OnStreamData(2 * kBlockSizeBytes, source, &written_, + EXPECT_THAT(buffer_->OnStreamData(8 * kBlockSizeBytes, source, &written_, &error_details_), IsError(QUIC_INTERNAL_ERROR)); EXPECT_TRUE(helper_->CheckBufferInvariants()); @@ -523,15 +523,15 @@ TEST_F(QuicStreamSequencerBufferTest, GetReadableRegionsWithMultipleIOVsAcrossEnd) { - // Write into [0, 2 * kBlockSizeBytes + 1024) and then read out [0, 1024) + // Write into [0, 8.5 * kBlockSizeBytes - 1024) and then read out [0, 1024) // and then append 1024 + 512 bytes. - std::string source(2.5 * kBlockSizeBytes - 1024, 'a'); + std::string source(8.5 * kBlockSizeBytes - 1024, 'a'); buffer_->OnStreamData(0, source, &written_, &error_details_); char dest[1024]; helper_->Read(dest, 1024); // Write across the end. source = std::string(1024 + 512, 'b'); - buffer_->OnStreamData(2.5 * kBlockSizeBytes - 1024, source, &written_, + buffer_->OnStreamData(8.5 * kBlockSizeBytes - 1024, source, &written_, &error_details_); // Use short iovec's. iovec iovs[2]; @@ -540,11 +540,11 @@ EXPECT_EQ(kBlockSizeBytes - 1024, iovs[0].iov_len); EXPECT_EQ(kBlockSizeBytes, iovs[1].iov_len); // Use long iovec's and wrap the end of buffer. - iovec iovs1[5]; - EXPECT_EQ(4, buffer_->GetReadableRegions(iovs1, 5)); - EXPECT_EQ(0.5 * kBlockSizeBytes, iovs1[2].iov_len); - EXPECT_EQ(512u, iovs1[3].iov_len); - EXPECT_EQ(std::string(512, 'b'), IovecToStringPiece(iovs1[3])); + iovec iovs1[11]; + EXPECT_EQ(10, buffer_->GetReadableRegions(iovs1, 11)); + EXPECT_EQ(0.5 * kBlockSizeBytes, iovs1[8].iov_len); + EXPECT_EQ(512u, iovs1[9].iov_len); + EXPECT_EQ(std::string(512, 'b'), IovecToStringPiece(iovs1[9])); } TEST_F(QuicStreamSequencerBufferTest, GetReadableRegionEmpty) { @@ -795,20 +795,20 @@ } TEST_F(QuicStreamSequencerBufferTest, MarkConsumedAcrossEnd) { - // Write into [0, 2.5 * kBlockSizeBytes - 1024) and then read out [0, 1024) + // Write into [0, 8.5 * kBlockSizeBytes - 1024) and then read out [0, 1024) // and then append 1024 + 512 bytes. - std::string source(2.5 * kBlockSizeBytes - 1024, 'a'); + std::string source(8.5 * kBlockSizeBytes - 1024, 'a'); buffer_->OnStreamData(0, source, &written_, &error_details_); char dest[1024]; helper_->Read(dest, 1024); source = std::string(1024 + 512, 'b'); - buffer_->OnStreamData(2.5 * kBlockSizeBytes - 1024, source, &written_, + buffer_->OnStreamData(8.5 * kBlockSizeBytes - 1024, source, &written_, &error_details_); EXPECT_EQ(1024u, buffer_->BytesConsumed()); - // Consume to the end of 2nd block. - buffer_->MarkConsumed(2 * kBlockSizeBytes - 1024); - EXPECT_EQ(2 * kBlockSizeBytes, buffer_->BytesConsumed()); + // Consume to the end of 8th block. + buffer_->MarkConsumed(8 * kBlockSizeBytes - 1024); + EXPECT_EQ(8 * kBlockSizeBytes, buffer_->BytesConsumed()); // Consume across the physical end of buffer buffer_->MarkConsumed(0.5 * kBlockSizeBytes + 500); EXPECT_EQ(max_capacity_bytes_ + 500, buffer_->BytesConsumed()); @@ -821,7 +821,7 @@ } TEST_F(QuicStreamSequencerBufferTest, FlushBufferedFrames) { - // Write into [0, 2.5 * kBlockSizeBytes - 1024) and then read out [0, 1024). + // Write into [0, 8.5 * kBlockSizeBytes - 1024) and then read out [0, 1024). std::string source(max_capacity_bytes_ - 1024, 'a'); buffer_->OnStreamData(0, source, &written_, &error_details_); char dest[1024]; @@ -869,7 +869,7 @@ void SetUp() override { // Test against a larger capacity then above tests. Also make sure the last // block is partially available to use. - max_capacity_bytes_ = 6.25 * kBlockSizeBytes; + max_capacity_bytes_ = 8.25 * kBlockSizeBytes; // Stream to be buffered should be larger than the capacity to test wrap // around. bytes_to_buffer_ = 2 * max_capacity_bytes_; @@ -1090,6 +1090,50 @@ EXPECT_LE(bytes_to_buffer_, total_bytes_written_); } +TEST_F(QuicStreamSequencerBufferTest, GrowBlockSizeOnDemand) { + SetQuicReloadableFlag(quic_allocate_stream_sequencer_buffer_blocks_on_demand, + true); + max_capacity_bytes_ = 1024 * kBlockSizeBytes; + std::string source_of_one_block(kBlockSizeBytes, 'a'); + Initialize(); + + ASSERT_EQ(helper_->current_blocks_count(), 0u); + + // A minimum of 8 blocks are allocated + buffer_->OnStreamData(0, source_of_one_block, &written_, &error_details_); + ASSERT_EQ(helper_->current_blocks_count(), 8u); + + // Number of blocks doesn't grow if the data is within the capacity. + buffer_->OnStreamData(kBlockSizeBytes * 7, source_of_one_block, &written_, + &error_details_); + ASSERT_EQ(helper_->current_blocks_count(), 8u); + + // Number of blocks grows by a factor of 4 normally. + buffer_->OnStreamData(kBlockSizeBytes * 8, "a", &written_, &error_details_); + ASSERT_EQ(helper_->current_blocks_count(), 32u); + + // Number of blocks grow to the demanded size of 140 instead of 128 since + // that's not enough. + buffer_->OnStreamData(kBlockSizeBytes * 139, source_of_one_block, &written_, + &error_details_); + ASSERT_EQ(helper_->current_blocks_count(), 140u); + + // Number of blocks grows by a factor of 4 normally. + buffer_->OnStreamData(kBlockSizeBytes * 140, source_of_one_block, &written_, + &error_details_); + ASSERT_EQ(helper_->current_blocks_count(), 560u); + + // max_capacity_bytes is reached and number of blocks is capped. + buffer_->OnStreamData(kBlockSizeBytes * 560, source_of_one_block, &written_, + &error_details_); + ASSERT_EQ(helper_->current_blocks_count(), 1024u); + + // max_capacity_bytes is reached and number of blocks is capped. + buffer_->OnStreamData(kBlockSizeBytes * 1025, source_of_one_block, &written_, + &error_details_); + ASSERT_EQ(helper_->current_blocks_count(), 1024u); +} + } // anonymous namespace } // namespace test
diff --git a/quic/test_tools/quic_stream_sequencer_buffer_peer.cc b/quic/test_tools/quic_stream_sequencer_buffer_peer.cc index b3bf224..eaf2c31 100644 --- a/quic/test_tools/quic_stream_sequencer_buffer_peer.cc +++ b/quic/test_tools/quic_stream_sequencer_buffer_peer.cc
@@ -3,6 +3,7 @@ // found in the LICENSE file. #include "net/third_party/quiche/src/quic/test_tools/quic_stream_sequencer_buffer_peer.h" +#include <cstddef> #include "net/third_party/quiche/src/quic/platform/api/quic_flags.h" #include "net/third_party/quiche/src/quic/platform/api/quic_logging.h" @@ -45,7 +46,8 @@ return true; } - size_t count = buffer_->blocks_count_; + size_t count = buffer_->allocate_blocks_on_demand_ ? current_blocks_count() + : max_blocks_count(); for (size_t i = 0; i < count; i++) { if (buffer_->blocks_[i] != nullptr) { return false; @@ -79,10 +81,11 @@ if (!capacity_sane) { QUIC_LOG(ERROR) << "read offset go beyond 1st block"; } - bool block_match_capacity = (buffer_->max_buffer_capacity_bytes_ <= - buffer_->blocks_count_ * kBlockSizeBytes) && - (buffer_->max_buffer_capacity_bytes_ > - (buffer_->blocks_count_ - 1) * kBlockSizeBytes); + bool block_match_capacity = + (buffer_->max_buffer_capacity_bytes_ <= + buffer_->max_blocks_count_ * kBlockSizeBytes) && + (buffer_->max_buffer_capacity_bytes_ > + (buffer_->max_blocks_count_ - 1) * kBlockSizeBytes); if (!capacity_sane) { QUIC_LOG(ERROR) << "block number not match capcaity."; } @@ -143,8 +146,12 @@ return buffer_->blocks_ != nullptr; } -size_t QuicStreamSequencerBufferPeer::block_count() { - return buffer_->blocks_count_; +size_t QuicStreamSequencerBufferPeer::max_blocks_count() { + return buffer_->max_blocks_count_; +} + +size_t QuicStreamSequencerBufferPeer::current_blocks_count() { + return buffer_->current_blocks_count_; } const QuicIntervalSet<QuicStreamOffset>&
diff --git a/quic/test_tools/quic_stream_sequencer_buffer_peer.h b/quic/test_tools/quic_stream_sequencer_buffer_peer.h index da2d054..4cac6f1 100644 --- a/quic/test_tools/quic_stream_sequencer_buffer_peer.h +++ b/quic/test_tools/quic_stream_sequencer_buffer_peer.h
@@ -49,7 +49,9 @@ bool IsBufferAllocated(); - size_t block_count(); + size_t max_blocks_count(); + + size_t current_blocks_count(); const QuicIntervalSet<QuicStreamOffset>& bytes_received();