diff --git a/CMakeLists.txt b/CMakeLists.txt index 34e51c6..40e0d5a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -407,6 +407,9 @@ if (BYCORF_BUILD_EXAMPLES) endif() if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS) + add_executable(bycorf_received_buffer_queue_check tests/received_buffer_queue_check.cpp) + target_link_libraries(bycorf_received_buffer_queue_check PRIVATE bycorf::core) + add_executable(bycorf_task_transfer_check tests/task_transfer_check.cpp) target_include_directories(bycorf_task_transfer_check PRIVATE include) # Exercise the public compile policy with this build's actual optimization @@ -487,6 +490,9 @@ if (BYCORF_BUILD_TESTS) endforeach() endif() enable_testing() + add_test(NAME bycorf_received_buffer_queue_check COMMAND bycorf_received_buffer_queue_check) + set_tests_properties(bycorf_received_buffer_queue_check PROPERTIES + TIMEOUT 30 LABELS "runtime;net") add_test(NAME bycorf_task_transfer_check COMMAND bycorf_task_transfer_check) set_tests_properties(bycorf_task_transfer_check PROPERTIES TIMEOUT 30 LABELS "runtime;task") diff --git a/include/bycorf/base/inline_ring_queue.h b/include/bycorf/base/inline_ring_queue.h new file mode 100644 index 0000000..e776203 --- /dev/null +++ b/include/bycorf/base/inline_ring_queue.h @@ -0,0 +1,238 @@ +/* + * Copyright (C) 2026 EloqData Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef BYCORF_BASE_INLINE_RING_QUEUE_H_ +#define BYCORF_BASE_INLINE_RING_QUEUE_H_ + +#include +#include +#include +#include +#include +#include +#include + +namespace bycorf { + +// Single-threaded resizable FIFO for trivially copyable values. InlineCapacity +// must be a positive power of two; heap storage starts at max(8, 2 * inline). +// Push/pop and moves may relocate elements and invalidate references. This +// container does not manage resources referenced by its values. +template +class InlineRingQueue { + static_assert( + std::is_trivially_copyable_v && std::is_trivially_destructible_v && + std::is_nothrow_copy_constructible_v, + "InlineRingQueue requires trivial values with nonthrowing copy"); + static_assert(InlineCapacity > 0 && + (InlineCapacity & (InlineCapacity - 1)) == 0 && + InlineCapacity <= (1u << 29), + "InlineCapacity must be a power of two in [1, 2^29]"); + + public: + // User-provided so value-initializing the queue does not zero the unused + // inline slots before their first values are constructed. + InlineRingQueue() noexcept {} + ~InlineRingQueue() { + if (capacity_ > kInlineCapacity) delete[] storage_.data_; + } + InlineRingQueue(const InlineRingQueue&) = delete; + InlineRingQueue& operator=(const InlineRingQueue&) = delete; + InlineRingQueue(InlineRingQueue&& other) noexcept { + *this = std::move(other); + } + InlineRingQueue& operator=(InlineRingQueue&& other) noexcept { + if (this != &other) { + if (other.capacity_ == kInlineCapacity) { + // Inline elements belong to the queue object and must be copied before + // the moved-from queue can be reused or destroyed. + if (capacity_ > kInlineCapacity) { + delete[] storage_.data_; + std::construct_at(&storage_.inline_); + } + other.CopyTo(storage_.inline_.slots_); + head_ = 0; + } else { + if (capacity_ > kInlineCapacity) delete[] storage_.data_; + // Assignment starts the trivial pointer member's lifetime when needed. + storage_.data_ = other.storage_.data_; + head_ = other.head_; + std::construct_at(&other.storage_.inline_); + } + other.head_ = 0; + size_ = std::exchange(other.size_, 0); + capacity_ = std::exchange(other.capacity_, kInlineCapacity); + low_water_pops_ = std::exchange(other.low_water_pops_, 0); + } + return *this; + } + + bool empty() const noexcept { return size_ == 0; } + std::uint32_t size() const noexcept { return size_; } + std::uint32_t capacity() const noexcept { return capacity_; } + T& operator[](std::uint32_t index) noexcept { + assert(index < size_); + return Storage()[(head_ + index) & (capacity_ - 1)].value_; + } + const T& operator[](std::uint32_t index) const noexcept { + assert(index < size_); + return Storage()[(head_ + index) & (capacity_ - 1)].value_; + } + T& front() noexcept { + assert(!empty()); + return Storage()[head_].value_; + } + const T& front() const noexcept { + assert(!empty()); + return Storage()[head_].value_; + } + + void push_back(T value) { + if (size_ == capacity_) Grow(); + std::construct_at(&Storage()[(head_ + size_) & (capacity_ - 1)].value_, + value); + ++size_; + if (capacity_ > kInlineCapacity && size_ > LowWaterLimit()) + low_water_pops_ = 0; + } + + void pop_front() noexcept { + assert(!empty()); + head_ = (head_ + 1) & (capacity_ - 1); + --size_; + // Inline storage cannot shrink; its low-water count stays zero. + if (capacity_ == kInlineCapacity) return; + if (size_ <= LowWaterLimit()) { + // Require a full capacity's worth of low-water pops before halving. + // Brief bursts and ordinary empty/nonempty alternation at the minimum + // capacity do not allocate. Shrinking is best-effort: reads and close + // cleanup must keep working even when allocation fails. + if (++low_water_pops_ == capacity_) Shrink(); + } else { + low_water_pops_ = 0; + } + } + + // Retains heap capacity. Callers must release any resources referenced by + // the values before discarding them. + void clear() noexcept { head_ = size_ = low_water_pops_ = 0; } + + private: + // An empty slot has no active value. Construct only live values, both + // inline and on the heap; a union's empty constructor avoids + // value-initializing every unused slot during growth. Trivial values + // need no destruction when popped, cleared, overwritten, or released with the + // slot array. + union Slot { + T value_; + Slot() noexcept {} + }; + static_assert(std::is_trivially_destructible_v); + + struct InlineStorage { + Slot slots_[InlineCapacity]; + // User-provided to avoid zeroing unused inline slots on activation. + InlineStorage() noexcept {} + }; + union StorageUnion { + InlineStorage inline_; + Slot* data_; + StorageUnion() noexcept : inline_() {} + }; + + Slot* Storage() noexcept { + return capacity_ == kInlineCapacity ? storage_.inline_.slots_ + : storage_.data_; + } + const Slot* Storage() const noexcept { + return capacity_ == kInlineCapacity ? storage_.inline_.slots_ + : storage_.data_; + } + + std::uint32_t LowWaterLimit() const noexcept { + // The smallest heap allocation contracts directly to the inline slot. + // Only count a sustained depth that actually fits the next capacity. + return capacity_ == kMinHeapCapacity ? kInlineCapacity : capacity_ / 4; + } + + // Keep allocation and value relocation out of inlined push/pop callers. + [[gnu::noinline]] void Grow() { + // Allocation failure leaves the source values and ownership intact. + if (capacity_ == kMaxCapacity) throw std::bad_alloc(); + const auto capacity = + capacity_ == kInlineCapacity ? kMinHeapCapacity : capacity_ * 2; + auto* data = new Slot[capacity]; + CopyTo(data); + // Copying cannot throw. Keep the old storage until all live values have + // been copied, then transfer sole ownership to the active pointer member. + if (capacity_ > kInlineCapacity) delete[] storage_.data_; + storage_.data_ = data; + head_ = 0; + capacity_ = capacity; + low_water_pops_ = 0; + } + + [[gnu::noinline]] void Shrink() noexcept { + low_water_pops_ = 0; + const auto capacity = + capacity_ == kMinHeapCapacity ? kInlineCapacity : capacity_ / 2; + if (capacity == kInlineCapacity) { + // Save the owned array before inline slots overwrite the pointer. + // CopyFrom uses the old head/capacity and cannot throw. + auto* data = storage_.data_; + std::construct_at(&storage_.inline_); + CopyFrom(data, storage_.inline_.slots_); + delete[] data; + } else { + auto* data = new (std::nothrow) Slot[capacity]; + if (!data) return; + CopyTo(data); + delete[] storage_.data_; + storage_.data_ = data; + } + head_ = 0; + capacity_ = capacity; + } + + void CopyTo(Slot* destination) const noexcept { + CopyFrom(Storage(), destination); + } + + void CopyFrom(const Slot* source, Slot* destination) const noexcept { + const auto first = std::min(size_, capacity_ - head_); + for (std::uint32_t i = 0; i < first; ++i) + std::construct_at(&destination[i].value_, source[head_ + i].value_); + for (std::uint32_t i = first; i < size_; ++i) + std::construct_at(&destination[i].value_, source[i - first].value_); + } + + static constexpr std::uint32_t kInlineCapacity = InlineCapacity; + static constexpr std::uint32_t kMinHeapCapacity = + std::max(8u, InlineCapacity * 2); + static constexpr std::uint32_t kMaxCapacity = 1u << 30; + // Inline values and heap ownership are mutually exclusive. capacity_ is the + // discriminator and the queue is the sole owner of data_ in heap mode. + // Every transition starts the new union member's lifetime. Slot construction + // and value copying cannot throw, so allocation is the only failure point. + StorageUnion storage_; + std::uint32_t head_ = 0; + std::uint32_t size_ = 0; + std::uint32_t capacity_ = kInlineCapacity; + std::uint32_t low_water_pops_ = 0; +}; + +} // namespace bycorf +#endif // BYCORF_BASE_INLINE_RING_QUEUE_H_ diff --git a/include/bycorf/net/connection.h b/include/bycorf/net/connection.h index 596a2db..10fd73e 100644 --- a/include/bycorf/net/connection.h +++ b/include/bycorf/net/connection.h @@ -20,13 +20,13 @@ #include #include #include -#include #include #include #include #include "absl/cleanup/cleanup.h" #include "absl/status/statusor.h" +#include "bycorf/base/inline_ring_queue.h" namespace bycorf { @@ -73,7 +73,7 @@ struct Connection { // With 64-byte-aligned allocations, these writes touch one cache line // instead of two. This alignment depends on the allocator, not alignof; // recheck actual addresses and generated stores when changing the layout. - // Normal receive completion still writes the deque tail on another line, + // Normal receive completion still writes queue bookkeeping on another line, // so this does not establish fewer lines for the whole receive/GET path. // Smaller sizeof or fewer touched lines alone does not establish a speedup: // require correctness and paired throughput/tail-latency regression tests. @@ -109,7 +109,15 @@ struct Connection { // reclaimed memory. Borrow and release run on the owning worker thread (or // before the first await that could observe retirement). std::uint32_t storage_borrows_ = 0; - std::deque received_buffers_; + // Owner-worker FIFO of receive descriptors, not the network bytes themselves. + // Callers own recycling backend buffers, including before clear(). References + // must not survive push/pop or moving the queue. + // The inline descriptor and mutable counters occupy the first 32 bytes. In + // the audited x86 layout the queue starts at Connection+96, so these fit one + // cache line when the Connection is 64-byte aligned. The heap owner shares + // storage with the inline descriptor. Recheck allocator addresses and + // throughput/tails after layout changes. + InlineRingQueue received_buffers_; std::uint64_t id_ = 0; std::uint64_t generation_ = 0; diff --git a/src/net/tcp_stream.cpp b/src/net/tcp_stream.cpp index 9b3c14f..c185615 100644 --- a/src/net/tcp_stream.cpp +++ b/src/net/tcp_stream.cpp @@ -217,11 +217,13 @@ absl::StatusOr ReadOperation::await_resume() noexcept { std::memcpy(buffer_.data() + copied, chunk.data(), chunk.size()); copied += to_copy; - received.offset_ += static_cast(to_copy); - if (received.offset_ == received.size_) { + if (to_copy == available) { + // The descriptor is about to leave the queue; no offset store is needed. const auto buffer_id = received.buffer_id_; connection_->received_buffers_.pop_front(); connection_->worker_->ReleaseReceivedBuffer(connection_, buffer_id); + } else { + received.offset_ += static_cast(to_copy); } // Do not poll or suspend between buffers. Only already queued data belongs // to this read, and ordinary ReadSome retains its single-buffer behavior. diff --git a/src/runtime/worker.cpp b/src/runtime/worker.cpp index 2d6fc2f..121bd85 100644 --- a/src/runtime/worker.cpp +++ b/src/runtime/worker.cpp @@ -790,11 +790,14 @@ void Worker::DiscardReceivedBuffers(Connection* connection) { if (connection == nullptr) { return; } - while (!connection->received_buffers_.empty()) { - auto received = connection->received_buffers_.front(); - connection->received_buffers_.pop_front(); - backend_.ReleaseRecvBuffer(connection, received.buffer_id_); - } + auto& received = connection->received_buffers_; + const auto count = received.size(); + // ReleaseRecvBuffer only recycles backend storage; it does not mutate this + // queue or resume readers. Clear once to avoid shrinking while discarding. + for (std::uint32_t i = 0; i < count; ++i) { + backend_.ReleaseRecvBuffer(connection, received[i].buffer_id_); + } + received.clear(); } void Worker::CheckIdleConnections() { diff --git a/tests/received_buffer_queue_check.cpp b/tests/received_buffer_queue_check.cpp new file mode 100644 index 0000000..d292551 --- /dev/null +++ b/tests/received_buffer_queue_check.cpp @@ -0,0 +1,305 @@ +/* + * Copyright (C) 2026 EloqData Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include +#include +#include +#include +#include + +#include "bycorf/net/connection.h" + +namespace { +using ReceiveQueue = decltype(bycorf::Connection::received_buffers_); + +void Check(bool ok, + std::source_location location = std::source_location::current()) { + if (!ok) { + std::cerr << "received buffer queue check failed at line " + << location.line() << "\n"; + std::abort(); + } +} +bool Equal(const bycorf::ReceivedBuffer& a, const bycorf::ReceivedBuffer& b) { + return a.buffer_id_ == b.buffer_id_ && a.size_ == b.size_ && + a.offset_ == b.offset_; +} +void Compare(const ReceiveQueue& queue, + const std::deque& reference) { + Check(queue.size() == reference.size()); + for (std::uint32_t i = 0; i < queue.size(); ++i) + Check(Equal(queue[i], reference[i])); +} +void CheckInlineAndHeapMoves() { + for (bool source_heap : {false, true}) { + for (bool destination_heap : {false, true}) { + ReceiveQueue destination; + for (unsigned i = 0; i < (destination_heap ? 4u : 1u); ++i) + destination.push_back({.buffer_id_ = 99}); + std::deque expected; + { + ReceiveQueue source; + for (unsigned i = 0; i < (source_heap ? 4u : 1u); ++i) { + bycorf::ReceivedBuffer value{ + .buffer_id_ = static_cast(i), + .size_ = 100 + i, + .offset_ = i}; + source.push_back(value); + expected.push_back(value); + } + destination = std::move(source); + Check(source.empty() && source.capacity() == 1); + // Reusing and destroying the source must not alias an inline element. + source.push_back({.buffer_id_ = 77, .size_ = 7}); + Compare(destination, expected); + } + Compare(destination, expected); + ReceiveQueue constructed(std::move(destination)); + Check(destination.empty() && destination.capacity() == 1); + destination.push_back({.buffer_id_ = 88}); + Compare(constructed, expected); + auto& same = constructed; + constructed = std::move(same); + Compare(constructed, expected); + } + } + ReceiveQueue small; + for (unsigned cycle = 0; cycle < 100; ++cycle) { + small.push_back({.buffer_id_ = 1}); + Check(small.capacity() == 1); + small.push_back({.buffer_id_ = 2}); + Check(small.capacity() == 8); + small.pop_front(); + Check(small.front().buffer_id_ == 2 && small.capacity() == 8); + small.pop_front(); + Check(small.empty() && small.capacity() == 8); + for (unsigned i = 0; i < 6; ++i) { + small.push_back({.buffer_id_ = 3}); + small.pop_front(); + } + Check(small.empty() && small.capacity() == 1); + } +} +// No default construction is needed, even for over-aligned values. Its copy +// constructor and destructor remain trivial. +struct alignas(32) Value { + Value() = delete; + explicit Value(std::uint64_t v) : value(v) {} + bool operator==(const Value&) const = default; + std::uint64_t value; +}; + +template +void CheckGenericQueue() { + using Queue = bycorf::InlineRingQueue; + auto compare = [](const Queue& q, const std::deque& expected) { + Check(q.size() == expected.size()); + Check(q.capacity() >= q.size()); + if (!q.empty()) Check(q.front() == expected.front()); + for (std::uint32_t i = 0; i < q.size(); ++i) Check(q[i] == expected[i]); + }; + // Exercise wrapped inline and heap sources against both destination modes. + for (bool source_heap : {false, true}) { + for (bool destination_heap : {false, true}) { + Queue destination; + for (unsigned i = 0; i < (destination_heap ? N + 1 : N); ++i) + destination.push_back(T(999)); + std::deque expected; + { + Queue source; + const auto count = source_heap ? N + 1 : N; + for (unsigned i = 0; i < count; ++i) { + source.push_back(T(i)); + expected.push_back(T(i)); + } + for (unsigned i = 0; i < count * 3 + 1; ++i) { + source.pop_front(); + expected.pop_front(); + source.push_back(T(100 + i)); + expected.push_back(T(100 + i)); + } + destination = std::move(source); + Check(source.empty() && source.capacity() == N); + source.push_back(T(777)); + compare(destination, expected); + } + compare(destination, expected); + Queue constructed(std::move(destination)); + Check(destination.empty() && destination.capacity() == N); + destination.push_back(T(888)); + compare(constructed, expected); + auto& same = constructed; + constructed = std::move(same); + compare(constructed, expected); + // Argument aliases a live value when growth relocates the source. + while (constructed.size() < constructed.capacity()) { + constructed.push_back(T(42)); + expected.push_back(T(42)); + } + expected.push_back(constructed.front()); + constructed.push_back(constructed.front()); + compare(constructed, expected); + } + } + // Empty queues may still own heap storage after clear(). Moving them must + // release the destination's old array and leave both queues reusable. + for (bool source_heap : {false, true}) { + for (bool destination_heap : {false, true}) { + Queue source, destination; + if (source_heap) + for (unsigned i = 0; i <= N; ++i) source.push_back(T(i)); + if (destination_heap) + for (unsigned i = 0; i <= N; ++i) destination.push_back(T(i)); + const auto capacity = source.capacity(); + source.clear(); + destination = std::move(source); + Check(destination.empty() && destination.capacity() == capacity); + Check(source.empty() && source.capacity() == N); + source.push_back(T(7)); + destination.push_back(T(8)); + Check(source.front() == T(7) && destination.front() == T(8)); + } + } + Queue queue; + std::deque expected; + std::mt19937 random(20261019 + N); + for (unsigned phase = 0; phase < 8; ++phase) { + for (unsigned step = 0; step < 4000; ++step) { + auto v = random(); + if (queue.empty() || v % 100 < (phase % 2 == 0 ? 60 : 40)) { + queue.push_back(T(v)); + expected.push_back(T(v)); + } else { + queue.pop_front(); + expected.pop_front(); + } + if (step % 113 == 0) compare(queue, expected); + } + compare(queue, expected); + } + while (queue.size() < N * 4) { + queue.push_back(T(42)); + expected.push_back(T(42)); + } + const auto peak = queue.capacity(); + while (queue.size() > N) { + queue.pop_front(); + expected.pop_front(); + } + // Shrink with N live values, including wrapped sources. Pop before push so + // the push never resets the final heap-to-inline low-water count. + for (unsigned i = 0; i < peak * 4; ++i) { + queue.pop_front(); + expected.pop_front(); + queue.push_back(T(i)); + expected.push_back(T(i)); + compare(queue, expected); + } + Check(queue.capacity() == N); + queue.push_back(T(123)); + Check(queue.capacity() > N); + const auto capacity = queue.capacity(); + queue.clear(); + Check(queue.empty() && queue.capacity() == capacity); + for (unsigned i = 0; i < capacity * 2; ++i) { + queue.push_back(T(i)); + Check(queue.front() == T(i)); + queue.pop_front(); + } + Check(queue.capacity() == N); +} + +template +void CheckGenericTypes() { + CheckGenericQueue(); + CheckGenericQueue(); +} +} // namespace + +int main() { + CheckGenericTypes<1>(); + CheckGenericTypes<2>(); + CheckGenericTypes<4>(); + CheckGenericTypes<8>(); + CheckGenericTypes<16>(); + CheckGenericTypes<32>(); + CheckInlineAndHeapMoves(); + ReceiveQueue queue; + Check(queue.empty() && queue.capacity() == 1); + std::deque reference; + std::mt19937 random(20261019); + // Alternate sustained growth and draining, with partial front consumption. + // This exercises wrapped growth and contraction, not just an empty queue. + for (unsigned phase = 0; phase < 12; ++phase) { + for (unsigned step = 0; step < 10000; ++step) { + const auto value = static_cast(random()); + if (queue.empty() || value % 100 < (phase % 2 == 0 ? 60 : 40)) { + bycorf::ReceivedBuffer buffer{ + .buffer_id_ = static_cast(value), + .size_ = value % 65536 + 10, + .offset_ = value % 10}; + queue.push_back(buffer); + reference.push_back(buffer); + } else { + Check(Equal(queue.front(), reference.front())); + queue.pop_front(); + reference.pop_front(); + } + if (!queue.empty() && step % 7 == 0) { + ++queue.front().offset_; + ++reference.front().offset_; + } + if (step % 113 == 0) Compare(queue, reference); + } + Compare(queue, reference); + } + while (queue.size() < 256) { + queue.push_back({.buffer_id_ = 42, .size_ = 100, .offset_ = 7}); + reference.push_back({.buffer_id_ = 42, .size_ = 100, .offset_ = 7}); + } + const auto peak = queue.capacity(); + while (queue.size() > 3) { + queue.pop_front(); + reference.pop_front(); + } + for (unsigned i = 0; i < peak * 3; ++i) { + queue.push_back({.buffer_id_ = 7, .size_ = 200, .offset_ = 11}); + reference.push_back({.buffer_id_ = 7, .size_ = 200, .offset_ = 11}); + queue.pop_front(); + reference.pop_front(); + Compare(queue, reference); + } + Check(queue.capacity() < peak && queue.capacity() >= queue.size()); + auto moved = std::move(queue); + Check(queue.empty() && queue.capacity() == 1); + Compare(moved, reference); + queue.push_back({.buffer_id_ = 99}); + queue = std::move(moved); + Check(moved.empty() && moved.capacity() == 1); + Compare(queue, reference); + queue.clear(); + Check(queue.empty()); + for (unsigned i = 0; i < 10000; ++i) { + queue.push_back({.buffer_id_ = 1, .size_ = 32}); + Check(queue.front().size_ == 32); + queue.pop_front(); + } + Check(queue.capacity() == 1); + std::cout + << "PASS FIFO, wrap, growth, delayed shrink, partial offsets, " + "move, clear, generic types and inline capacities 1/2/4/8/16/32\n"; +} diff --git a/tests/tcp_buffered_read_check.cpp b/tests/tcp_buffered_read_check.cpp index dba06f8..157fbd1 100644 --- a/tests/tcp_buffered_read_check.cpp +++ b/tests/tcp_buffered_read_check.cpp @@ -52,8 +52,10 @@ absl::Status Send(int fd, std::string_view bytes) { std::size_t QueuedBytes(const bycorf::Connection& connection) { std::size_t bytes = 0; - for (const auto& buffer : connection.received_buffers_) + for (std::uint32_t i = 0; i < connection.received_buffers_.size(); ++i) { + const auto buffer = connection.received_buffers_[i]; bytes += buffer.size_ - buffer.offset_; + } return bytes; } @@ -142,9 +144,9 @@ bycorf::Task CheckReads(bycorf::Worker& worker) { status = Check(connection->received_buffers_.size() >= 2, "multiple remaining buffers for partial-error check"); if (!status.ok()) co_return status; - auto& invalid_buffer = connection->received_buffers_[1]; - const auto saved_id = invalid_buffer.buffer_id_; - invalid_buffer.buffer_id_ = std::numeric_limits::max(); + const auto saved_buffer = connection->received_buffers_[1]; + connection->received_buffers_[1].buffer_id_ = + std::numeric_limits::max(); const auto valid_prefix = connection->received_buffers_[0].size_ - connection->received_buffers_[0].offset_; read = co_await stream.ReadSomeBuffered(output); @@ -152,7 +154,9 @@ bycorf::Task CheckReads(bycorf::Worker& worker) { if (prefix_ok) received.append(reinterpret_cast(output.data()), *read); auto failed = co_await stream.ReadSome(output); - invalid_buffer.buffer_id_ = saved_id; + // The preceding valid descriptor was consumed. Reacquire the invalid front + // rather than retaining a reference across a potentially shrinking pop. + if (prefix_ok) connection->received_buffers_.front() = saved_buffer; status = Check(prefix_ok && absl::IsInternal(failed.status()), "partial progress is returned before a later buffer error"); if (!status.ok()) co_return status; @@ -208,8 +212,48 @@ bycorf::Task CheckReads(bycorf::Worker& worker) { "live EOF returns zero after the queue drains"); } +absl::Status CheckCloseDoesNotShrink(bycorf::Worker& worker) { + int sockets[2]; + if (socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + sockets) != 0) + return absl::InternalError("socketpair failed"); + absl::Cleanup close_peer = [&] { close(sockets[1]); }; + bycorf::Connection initial; + initial.worker_ = &worker; + initial.file_.fd_ = sockets[0]; + initial.closed_ = false; + initial.recv_mode_ = bycorf::RecvMode::kOneShot; + auto* connection = worker.AddConnection(std::move(initial)); + bycorf::TcpStream stream(connection); + auto storage = stream.BorrowStorage(); + absl::Cleanup close_stream = [&] { (void)stream.Close(); }; + connection->read_buffer_.resize(1); + auto& queue = connection->received_buffers_; + const bycorf::ReceivedBuffer buffer{.buffer_id_ = 0, .size_ = 1}; + constexpr std::uint32_t capacity = 64; + for (std::uint32_t i = 0; i < capacity; ++i) queue.push_back(buffer); + while (queue.size() > capacity / 4) queue.pop_front(); + // Synthetic one-shot views do not borrow provided buffers. Leave the queue + // one low-water pop before shrinking: the old close loop allocated 32 new + // descriptors here even though it was about to discard every descriptor. + for (std::uint32_t i = 0; i < capacity - 2; ++i) { + queue.pop_front(); + queue.push_back(buffer); + } + auto status = + Check(queue.size() == capacity / 4 && queue.capacity() == capacity, + "close regression starts near the shrink threshold"); + if (!status.ok()) return status; + status = stream.Close(); + if (!status.ok()) return status; + return Check( + connection->closed_ && queue.empty() && queue.capacity() == capacity, + "close discards queued buffers without shrinking storage"); +} + bycorf::Task RunChecks(bycorf::Worker& worker, bool& passed) { - const auto status = co_await CheckReads(worker); + auto status = co_await CheckReads(worker); + if (status.ok()) status = CheckCloseDoesNotShrink(worker); passed = status.ok(); if (!status.ok()) std::cerr << status << '\n'; worker.RequestStop();