From 6da7a471d13b8ab1069906ff46abd8e543c6ff57 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 11:27:56 +0800 Subject: [PATCH 01/10] perf: use a growable ring for received buffer descriptors --- CMakeLists.txt | 6 + include/bycorf/net/connection.h | 12 +- include/bycorf/net/received_buffer_queue.h | 134 +++++++++++++++++++++ tests/received_buffer_queue_check.cpp | 108 +++++++++++++++++ tests/tcp_buffered_read_check.cpp | 14 ++- 5 files changed, 260 insertions(+), 14 deletions(-) create mode 100644 include/bycorf/net/received_buffer_queue.h create mode 100644 tests/received_buffer_queue_check.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 34e51c6..52e3f41 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_include_directories(bycorf_received_buffer_queue_check PRIVATE include) + 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/net/connection.h b/include/bycorf/net/connection.h index 596a2db..f8bf7d2 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/net/received_buffer_queue.h" namespace bycorf { @@ -59,12 +59,6 @@ struct RegisteredFile { bool is_fixed_ = false; }; -struct ReceivedBuffer { - std::uint16_t buffer_id_ = 0; - std::uint32_t size_ = 0; - std::uint32_t offset_ = 0; -}; - struct Connection { using PeerDisconnectCallback = void (*)(void*) noexcept; @@ -73,7 +67,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 +103,7 @@ 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_; + ReceivedBufferQueue received_buffers_; std::uint64_t id_ = 0; std::uint64_t generation_ = 0; diff --git a/include/bycorf/net/received_buffer_queue.h b/include/bycorf/net/received_buffer_queue.h new file mode 100644 index 0000000..6896aca --- /dev/null +++ b/include/bycorf/net/received_buffer_queue.h @@ -0,0 +1,134 @@ +/* + * 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_NET_RECEIVED_BUFFER_QUEUE_H_ +#define BYCORF_NET_RECEIVED_BUFFER_QUEUE_H_ + +#include +#include +#include +#include +#include +#include + +namespace bycorf { + +struct ReceivedBuffer { + std::uint16_t buffer_id_ = 0; + std::uint32_t size_ = 0; + std::uint32_t offset_ = 0; +}; + +// Owner-worker FIFO of receive descriptors, not the network bytes themselves. +// Resizing preserves buffer IDs and partial-read offsets; callers still own +// recycling the referenced backend buffers. Never retain an element reference +// across push_back/pop_front: either operation can relocate the descriptors. +class ReceivedBufferQueue { + public: + ReceivedBufferQueue() = default; + ReceivedBufferQueue(const ReceivedBufferQueue&) = delete; + ReceivedBufferQueue& operator=(const ReceivedBufferQueue&) = delete; + ReceivedBufferQueue(ReceivedBufferQueue&& other) noexcept { + *this = std::move(other); + } + ReceivedBufferQueue& operator=(ReceivedBufferQueue&& other) noexcept { + if (this != &other) { + data_ = std::move(other.data_); + head_ = std::exchange(other.head_, 0); + size_ = std::exchange(other.size_, 0); + capacity_ = std::exchange(other.capacity_, 0); + 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_; } + ReceivedBuffer& operator[](std::uint32_t index) noexcept { + assert(index < size_); + return data_[(head_ + index) & (capacity_ - 1)]; + } + const ReceivedBuffer& operator[](std::uint32_t index) const noexcept { + assert(index < size_); + return data_[(head_ + index) & (capacity_ - 1)]; + } + ReceivedBuffer& front() noexcept { return (*this)[0]; } + const ReceivedBuffer& front() const noexcept { return (*this)[0]; } + + void push_back(ReceivedBuffer value) { + if (size_ == capacity_) { + // Bound both doubling and head + index arithmetic. Allocation failure + // leaves every queued descriptor intact, as with deque::push_back. + if (capacity_ == kMaxCapacity) throw std::bad_alloc(); + const auto capacity = capacity_ == 0 ? kMinCapacity : capacity_ * 2; + Reallocate(std::make_unique(capacity), capacity); + } + data_[(head_ + size_) & (capacity_ - 1)] = value; + ++size_; + if (size_ > capacity_ / 4) low_water_pops_ = 0; + } + + void pop_front() noexcept { + assert(!empty()); + head_ = (head_ + 1) & (capacity_ - 1); + --size_; + if (capacity_ > kMinCapacity && size_ <= capacity_ / 4) { + // 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_) { + low_water_pops_ = 0; + const auto capacity = capacity_ / 2; + std::unique_ptr data(new (std::nothrow) + ReceivedBuffer[capacity]); + if (data) Reallocate(std::move(data), capacity); + } + } else { + low_water_pops_ = 0; + } + } + + // Like deque::clear, this discards descriptors without returning backend + // buffers. Production close paths must recycle each buffer before clearing. + void clear() noexcept { head_ = size_ = low_water_pops_ = 0; } + + private: + void Reallocate(std::unique_ptr data, + std::uint32_t capacity) noexcept { + if (size_ != 0) { + const auto first = std::min(size_, capacity_ - head_); + std::copy_n(data_.get() + head_, first, data.get()); + std::copy_n(data_.get(), size_ - first, data.get() + first); + } + data_ = std::move(data); + head_ = 0; + capacity_ = capacity; + low_water_pops_ = 0; + } + + static constexpr std::uint32_t kMinCapacity = 8; + static constexpr std::uint32_t kMaxCapacity = 1u << 30; + std::unique_ptr data_; + std::uint32_t head_ = 0; + std::uint32_t size_ = 0; + std::uint32_t capacity_ = 0; + std::uint32_t low_water_pops_ = 0; +}; + +} // namespace bycorf +#endif // BYCORF_NET_RECEIVED_BUFFER_QUEUE_H_ diff --git a/tests/received_buffer_queue_check.cpp b/tests/received_buffer_queue_check.cpp new file mode 100644 index 0000000..2d9d0e7 --- /dev/null +++ b/tests/received_buffer_queue_check.cpp @@ -0,0 +1,108 @@ +/* + * 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 "bycorf/net/received_buffer_queue.h" + +namespace { +void Check(bool ok) { + if (!ok) { + std::cerr << "received buffer queue check failed\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 bycorf::ReceivedBufferQueue& 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])); +} +} // namespace + +int main() { + bycorf::ReceivedBufferQueue queue; + Check(queue.empty() && queue.capacity() == 0); + 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 = 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() == 0); + Compare(moved, reference); + queue.push_back({.buffer_id_ = 99}); + queue = std::move(moved); + Check(moved.empty() && moved.capacity() == 0); + 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() == 8); + std::cout << "PASS FIFO, wrap, growth, delayed shrink, partial offsets, " + "move, clear\n"; +} diff --git a/tests/tcp_buffered_read_check.cpp b/tests/tcp_buffered_read_check.cpp index dba06f8..9898229 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; From 0d1335cdaeb1087523097aca20ce3903bd23dd4e Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 11:58:08 +0800 Subject: [PATCH 02/10] test: normalize random descriptor values to uint32_t --- tests/received_buffer_queue_check.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/received_buffer_queue_check.cpp b/tests/received_buffer_queue_check.cpp index 2d9d0e7..d38814c 100644 --- a/tests/received_buffer_queue_check.cpp +++ b/tests/received_buffer_queue_check.cpp @@ -50,7 +50,7 @@ int main() { // 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 = random(); + const auto value = static_cast(random()); if (queue.empty() || value % 100 < (phase % 2 == 0 ? 60 : 40)) { bycorf::ReceivedBuffer buffer{ .buffer_id_ = static_cast(value), From 34b789584aa119b90e7bb1a3d07f1bac2b6e9e1e Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 14:23:28 +0800 Subject: [PATCH 03/10] perf: index receive queue front directly by head --- include/bycorf/net/received_buffer_queue.h | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/include/bycorf/net/received_buffer_queue.h b/include/bycorf/net/received_buffer_queue.h index 6896aca..2b273cd 100644 --- a/include/bycorf/net/received_buffer_queue.h +++ b/include/bycorf/net/received_buffer_queue.h @@ -66,8 +66,14 @@ class ReceivedBufferQueue { assert(index < size_); return data_[(head_ + index) & (capacity_ - 1)]; } - ReceivedBuffer& front() noexcept { return (*this)[0]; } - const ReceivedBuffer& front() const noexcept { return (*this)[0]; } + ReceivedBuffer& front() noexcept { + assert(!empty()); + return data_[head_]; + } + const ReceivedBuffer& front() const noexcept { + assert(!empty()); + return data_[head_]; + } void push_back(ReceivedBuffer value) { if (size_ == capacity_) { From da5efcd7f74f58be5a0a06dc8f0ade02032cf5af Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 14:33:27 +0800 Subject: [PATCH 04/10] perf: outline receive queue allocation and relocation paths --- include/bycorf/net/received_buffer_queue.h | 34 +++++++++++++--------- 1 file changed, 20 insertions(+), 14 deletions(-) diff --git a/include/bycorf/net/received_buffer_queue.h b/include/bycorf/net/received_buffer_queue.h index 2b273cd..c7efbbf 100644 --- a/include/bycorf/net/received_buffer_queue.h +++ b/include/bycorf/net/received_buffer_queue.h @@ -76,13 +76,7 @@ class ReceivedBufferQueue { } void push_back(ReceivedBuffer value) { - if (size_ == capacity_) { - // Bound both doubling and head + index arithmetic. Allocation failure - // leaves every queued descriptor intact, as with deque::push_back. - if (capacity_ == kMaxCapacity) throw std::bad_alloc(); - const auto capacity = capacity_ == 0 ? kMinCapacity : capacity_ * 2; - Reallocate(std::make_unique(capacity), capacity); - } + if (size_ == capacity_) Grow(); data_[(head_ + size_) & (capacity_ - 1)] = value; ++size_; if (size_ > capacity_ / 4) low_water_pops_ = 0; @@ -97,13 +91,7 @@ class ReceivedBufferQueue { // 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_) { - low_water_pops_ = 0; - const auto capacity = capacity_ / 2; - std::unique_ptr data(new (std::nothrow) - ReceivedBuffer[capacity]); - if (data) Reallocate(std::move(data), capacity); - } + if (++low_water_pops_ == capacity_) Shrink(); } else { low_water_pops_ = 0; } @@ -114,6 +102,24 @@ class ReceivedBufferQueue { void clear() noexcept { head_ = size_ = low_water_pops_ = 0; } private: + // Keep allocation and descriptor relocation out of inlined push/pop callers. + // These paths are rare; inlining them bloats the ordinary receive hot path. + [[gnu::noinline]] void Grow() { + // Bound both doubling and head + index arithmetic. Allocation failure + // leaves every queued descriptor intact, as with deque::push_back. + if (capacity_ == kMaxCapacity) throw std::bad_alloc(); + const auto capacity = capacity_ == 0 ? kMinCapacity : capacity_ * 2; + Reallocate(std::make_unique(capacity), capacity); + } + + [[gnu::noinline]] void Shrink() noexcept { + low_water_pops_ = 0; + const auto capacity = capacity_ / 2; + std::unique_ptr data(new (std::nothrow) + ReceivedBuffer[capacity]); + if (data) Reallocate(std::move(data), capacity); + } + void Reallocate(std::unique_ptr data, std::uint32_t capacity) noexcept { if (size_ != 0) { From c53adc0c86b0abcbc32675df597703eeed3876aa Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 14:43:01 +0800 Subject: [PATCH 05/10] perf: avoid redundant receive updates and resize during close --- src/net/tcp_stream.cpp | 6 +++-- src/runtime/worker.cpp | 13 ++++++---- tests/tcp_buffered_read_check.cpp | 42 ++++++++++++++++++++++++++++++- 3 files changed, 53 insertions(+), 8 deletions(-) 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/tcp_buffered_read_check.cpp b/tests/tcp_buffered_read_check.cpp index 9898229..157fbd1 100644 --- a/tests/tcp_buffered_read_check.cpp +++ b/tests/tcp_buffered_read_check.cpp @@ -212,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(); From 2453e417731e076b0cd84b714da3a8947550e70a Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 15:51:35 +0800 Subject: [PATCH 06/10] Keep one receive descriptor inline without initializing unused slots --- include/bycorf/net/received_buffer_queue.h | 120 +++++++++++++++------ tests/received_buffer_queue_check.cpp | 66 ++++++++++-- 2 files changed, 148 insertions(+), 38 deletions(-) diff --git a/include/bycorf/net/received_buffer_queue.h b/include/bycorf/net/received_buffer_queue.h index c7efbbf..793e26e 100644 --- a/include/bycorf/net/received_buffer_queue.h +++ b/include/bycorf/net/received_buffer_queue.h @@ -22,6 +22,7 @@ #include #include #include +#include #include namespace bycorf { @@ -38,7 +39,9 @@ struct ReceivedBuffer { // across push_back/pop_front: either operation can relocate the descriptors. class ReceivedBufferQueue { public: - ReceivedBufferQueue() = default; + // User-provided so value-initializing the queue does not zero the unused + // inline slot before its first descriptor is constructed. + ReceivedBufferQueue() noexcept {} ReceivedBufferQueue(const ReceivedBufferQueue&) = delete; ReceivedBufferQueue& operator=(const ReceivedBufferQueue&) = delete; ReceivedBufferQueue(ReceivedBufferQueue&& other) noexcept { @@ -46,10 +49,18 @@ class ReceivedBufferQueue { } ReceivedBufferQueue& operator=(ReceivedBufferQueue&& other) noexcept { if (this != &other) { - data_ = std::move(other.data_); + 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. + data_.reset(); + if (other.size_ != 0) + std::construct_at(&inline_.value_, other.inline_.value_); + } else { + data_ = std::move(other.data_); + } head_ = std::exchange(other.head_, 0); size_ = std::exchange(other.size_, 0); - capacity_ = std::exchange(other.capacity_, 0); + capacity_ = std::exchange(other.capacity_, kInlineCapacity); low_water_pops_ = std::exchange(other.low_water_pops_, 0); } return *this; @@ -60,33 +71,37 @@ class ReceivedBufferQueue { std::uint32_t capacity() const noexcept { return capacity_; } ReceivedBuffer& operator[](std::uint32_t index) noexcept { assert(index < size_); - return data_[(head_ + index) & (capacity_ - 1)]; + return Storage()[(head_ + index) & (capacity_ - 1)].value_; } const ReceivedBuffer& operator[](std::uint32_t index) const noexcept { assert(index < size_); - return data_[(head_ + index) & (capacity_ - 1)]; + return Storage()[(head_ + index) & (capacity_ - 1)].value_; } ReceivedBuffer& front() noexcept { assert(!empty()); - return data_[head_]; + return Storage()[head_].value_; } const ReceivedBuffer& front() const noexcept { assert(!empty()); - return data_[head_]; + return Storage()[head_].value_; } void push_back(ReceivedBuffer value) { if (size_ == capacity_) Grow(); - data_[(head_ + size_) & (capacity_ - 1)] = value; + std::construct_at(&Storage()[(head_ + size_) & (capacity_ - 1)].value_, + value); ++size_; - if (size_ > capacity_ / 4) low_water_pops_ = 0; + if (capacity_ > kInlineCapacity && size_ > LowWaterLimit()) + low_water_pops_ = 0; } void pop_front() noexcept { assert(!empty()); head_ = (head_ + 1) & (capacity_ - 1); --size_; - if (capacity_ > kMinCapacity && size_ <= capacity_ / 4) { + // 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 @@ -102,44 +117,85 @@ class ReceivedBufferQueue { void clear() noexcept { head_ = size_ = low_water_pops_ = 0; } private: + // An empty slot has no active value. Construct only live descriptors, both + // inline and on the heap; a union's empty constructor avoids + // value-initializing every unused slot during growth. Trivial descriptors + // need no destruction when popped, cleared, overwritten, or released with the + // slot array. + union Slot { + ReceivedBuffer value_; + Slot() noexcept {} + }; + static_assert(std::is_trivially_copyable_v); + static_assert(std::is_trivially_destructible_v); + + Slot* Storage() noexcept { + return capacity_ == kInlineCapacity ? &inline_ : data_.get(); + } + const Slot* Storage() const noexcept { + return capacity_ == kInlineCapacity ? &inline_ : data_.get(); + } + + 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 descriptor relocation out of inlined push/pop callers. - // These paths are rare; inlining them bloats the ordinary receive hot path. [[gnu::noinline]] void Grow() { - // Bound both doubling and head + index arithmetic. Allocation failure - // leaves every queued descriptor intact, as with deque::push_back. + // Allocation failure leaves the source descriptors and ownership intact. if (capacity_ == kMaxCapacity) throw std::bad_alloc(); - const auto capacity = capacity_ == 0 ? kMinCapacity : capacity_ * 2; - Reallocate(std::make_unique(capacity), capacity); + const auto capacity = + capacity_ == kInlineCapacity ? kMinHeapCapacity : capacity_ * 2; + auto data = std::make_unique(capacity); + CopyTo(data.get()); + data_ = std::move(data); + head_ = 0; + capacity_ = capacity; + low_water_pops_ = 0; } [[gnu::noinline]] void Shrink() noexcept { low_water_pops_ = 0; - const auto capacity = capacity_ / 2; - std::unique_ptr data(new (std::nothrow) - ReceivedBuffer[capacity]); - if (data) Reallocate(std::move(data), capacity); - } - - void Reallocate(std::unique_ptr data, - std::uint32_t capacity) noexcept { - if (size_ != 0) { - const auto first = std::min(size_, capacity_ - head_); - std::copy_n(data_.get() + head_, first, data.get()); - std::copy_n(data_.get(), size_ - first, data.get() + first); + const auto capacity = + capacity_ == kMinHeapCapacity ? kInlineCapacity : capacity_ / 2; + if (capacity == kInlineCapacity) { + CopyTo(&inline_); + data_.reset(); + } else { + std::unique_ptr data(new (std::nothrow) Slot[capacity]); + if (!data) return; + CopyTo(data.get()); + data_ = std::move(data); } - data_ = std::move(data); head_ = 0; capacity_ = capacity; - low_water_pops_ = 0; } - static constexpr std::uint32_t kMinCapacity = 8; + void CopyTo(Slot* destination) const noexcept { + const auto* source = Storage(); + 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 = 1; + static constexpr std::uint32_t kMinHeapCapacity = 8; static constexpr std::uint32_t kMaxCapacity = 1u << 30; - std::unique_ptr data_; + // The inline descriptor and frequently written counters occupy 28 bytes. + // Keep the heap pointer after them so the usual depth-one path need not + // access it. In the audited x86 layout, this queue starts at Connection+96: + // these 28 bytes fit one cache line if the Connection is 64-byte aligned. + // Recheck allocator addresses and throughput/tails after layout changes. + Slot inline_; std::uint32_t head_ = 0; std::uint32_t size_ = 0; - std::uint32_t capacity_ = 0; + std::uint32_t capacity_ = kInlineCapacity; std::uint32_t low_water_pops_ = 0; + std::unique_ptr data_; }; } // namespace bycorf diff --git a/tests/received_buffer_queue_check.cpp b/tests/received_buffer_queue_check.cpp index d38814c..9ec1df4 100644 --- a/tests/received_buffer_queue_check.cpp +++ b/tests/received_buffer_queue_check.cpp @@ -18,14 +18,17 @@ #include #include #include +#include #include #include "bycorf/net/received_buffer_queue.h" namespace { -void Check(bool ok) { +void Check(bool ok, + std::source_location location = std::source_location::current()) { if (!ok) { - std::cerr << "received buffer queue check failed\n"; + std::cerr << "received buffer queue check failed at line " + << location.line() << "\n"; std::abort(); } } @@ -39,11 +42,62 @@ void Compare(const bycorf::ReceivedBufferQueue& queue, 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}) { + bycorf::ReceivedBufferQueue destination; + for (unsigned i = 0; i < (destination_heap ? 4u : 1u); ++i) + destination.push_back({.buffer_id_ = 99}); + std::deque expected; + { + bycorf::ReceivedBufferQueue 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); + bycorf::ReceivedBufferQueue 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); + } + } + bycorf::ReceivedBufferQueue 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); + } +} } // namespace int main() { + CheckInlineAndHeapMoves(); bycorf::ReceivedBufferQueue queue; - Check(queue.empty() && queue.capacity() == 0); + Check(queue.empty() && queue.capacity() == 1); std::deque reference; std::mt19937 random(20261019); // Alternate sustained growth and draining, with partial front consumption. @@ -89,11 +143,11 @@ int main() { } Check(queue.capacity() < peak && queue.capacity() >= queue.size()); auto moved = std::move(queue); - Check(queue.empty() && queue.capacity() == 0); + Check(queue.empty() && queue.capacity() == 1); Compare(moved, reference); queue.push_back({.buffer_id_ = 99}); queue = std::move(moved); - Check(moved.empty() && moved.capacity() == 0); + Check(moved.empty() && moved.capacity() == 1); Compare(queue, reference); queue.clear(); Check(queue.empty()); @@ -102,7 +156,7 @@ int main() { Check(queue.front().size_ == 32); queue.pop_front(); } - Check(queue.capacity() == 8); + Check(queue.capacity() == 1); std::cout << "PASS FIFO, wrap, growth, delayed shrink, partial offsets, " "move, clear\n"; } From a37ed7d19885275472b707461dc2eddf4dccfbcd Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 16:08:47 +0800 Subject: [PATCH 07/10] Extract receive ring into a template with configurable inline capacity --- include/bycorf/base/inline_ring_queue.h | 203 +++++++++++++++++++++ include/bycorf/net/received_buffer_queue.h | 178 +----------------- tests/received_buffer_queue_check.cpp | 126 ++++++++++++- 3 files changed, 336 insertions(+), 171 deletions(-) create mode 100644 include/bycorf/base/inline_ring_queue.h diff --git a/include/bycorf/base/inline_ring_queue.h b/include/bycorf/base/inline_ring_queue.h new file mode 100644 index 0000000..24f4602 --- /dev/null +++ b/include/bycorf/base/inline_ring_queue.h @@ -0,0 +1,203 @@ +/* + * 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(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. + data_.reset(); + other.CopyTo(inline_); + head_ = 0; + } else { + data_ = std::move(other.data_); + head_ = other.head_; + } + 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); + + Slot* Storage() noexcept { + return capacity_ == kInlineCapacity ? inline_ : data_.get(); + } + const Slot* Storage() const noexcept { + return capacity_ == kInlineCapacity ? inline_ : data_.get(); + } + + 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 = std::make_unique(capacity); + CopyTo(data.get()); + data_ = std::move(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) { + CopyTo(inline_); + data_.reset(); + } else { + std::unique_ptr data(new (std::nothrow) Slot[capacity]); + if (!data) return; + CopyTo(data.get()); + data_ = std::move(data); + } + head_ = 0; + capacity_ = capacity; + } + + void CopyTo(Slot* destination) const noexcept { + const auto* source = Storage(); + 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; + // Keep inline values and mutable counters ahead of the cold heap pointer. + Slot inline_[InlineCapacity]; + std::uint32_t head_ = 0; + std::uint32_t size_ = 0; + std::uint32_t capacity_ = kInlineCapacity; + std::uint32_t low_water_pops_ = 0; + std::unique_ptr data_; +}; + +} // namespace bycorf +#endif // BYCORF_BASE_INLINE_RING_QUEUE_H_ diff --git a/include/bycorf/net/received_buffer_queue.h b/include/bycorf/net/received_buffer_queue.h index 793e26e..f84f325 100644 --- a/include/bycorf/net/received_buffer_queue.h +++ b/include/bycorf/net/received_buffer_queue.h @@ -17,13 +17,9 @@ #ifndef BYCORF_NET_RECEIVED_BUFFER_QUEUE_H_ #define BYCORF_NET_RECEIVED_BUFFER_QUEUE_H_ -#include -#include #include -#include -#include -#include -#include + +#include "bycorf/base/inline_ring_queue.h" namespace bycorf { @@ -34,169 +30,13 @@ struct ReceivedBuffer { }; // Owner-worker FIFO of receive descriptors, not the network bytes themselves. -// Resizing preserves buffer IDs and partial-read offsets; callers still own -// recycling the referenced backend buffers. Never retain an element reference -// across push_back/pop_front: either operation can relocate the descriptors. -class ReceivedBufferQueue { - public: - // User-provided so value-initializing the queue does not zero the unused - // inline slot before its first descriptor is constructed. - ReceivedBufferQueue() noexcept {} - ReceivedBufferQueue(const ReceivedBufferQueue&) = delete; - ReceivedBufferQueue& operator=(const ReceivedBufferQueue&) = delete; - ReceivedBufferQueue(ReceivedBufferQueue&& other) noexcept { - *this = std::move(other); - } - ReceivedBufferQueue& operator=(ReceivedBufferQueue&& 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. - data_.reset(); - if (other.size_ != 0) - std::construct_at(&inline_.value_, other.inline_.value_); - } else { - data_ = std::move(other.data_); - } - head_ = std::exchange(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_; } - ReceivedBuffer& operator[](std::uint32_t index) noexcept { - assert(index < size_); - return Storage()[(head_ + index) & (capacity_ - 1)].value_; - } - const ReceivedBuffer& operator[](std::uint32_t index) const noexcept { - assert(index < size_); - return Storage()[(head_ + index) & (capacity_ - 1)].value_; - } - ReceivedBuffer& front() noexcept { - assert(!empty()); - return Storage()[head_].value_; - } - const ReceivedBuffer& front() const noexcept { - assert(!empty()); - return Storage()[head_].value_; - } - - void push_back(ReceivedBuffer 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; - } - } - - // Like deque::clear, this discards descriptors without returning backend - // buffers. Production close paths must recycle each buffer before clearing. - void clear() noexcept { head_ = size_ = low_water_pops_ = 0; } - - private: - // An empty slot has no active value. Construct only live descriptors, both - // inline and on the heap; a union's empty constructor avoids - // value-initializing every unused slot during growth. Trivial descriptors - // need no destruction when popped, cleared, overwritten, or released with the - // slot array. - union Slot { - ReceivedBuffer value_; - Slot() noexcept {} - }; - static_assert(std::is_trivially_copyable_v); - static_assert(std::is_trivially_destructible_v); - - Slot* Storage() noexcept { - return capacity_ == kInlineCapacity ? &inline_ : data_.get(); - } - const Slot* Storage() const noexcept { - return capacity_ == kInlineCapacity ? &inline_ : data_.get(); - } - - 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 descriptor relocation out of inlined push/pop callers. - [[gnu::noinline]] void Grow() { - // Allocation failure leaves the source descriptors and ownership intact. - if (capacity_ == kMaxCapacity) throw std::bad_alloc(); - const auto capacity = - capacity_ == kInlineCapacity ? kMinHeapCapacity : capacity_ * 2; - auto data = std::make_unique(capacity); - CopyTo(data.get()); - data_ = std::move(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) { - CopyTo(&inline_); - data_.reset(); - } else { - std::unique_ptr data(new (std::nothrow) Slot[capacity]); - if (!data) return; - CopyTo(data.get()); - data_ = std::move(data); - } - head_ = 0; - capacity_ = capacity; - } - - void CopyTo(Slot* destination) const noexcept { - const auto* source = Storage(); - 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 = 1; - static constexpr std::uint32_t kMinHeapCapacity = 8; - static constexpr std::uint32_t kMaxCapacity = 1u << 30; - // The inline descriptor and frequently written counters occupy 28 bytes. - // Keep the heap pointer after them so the usual depth-one path need not - // access it. In the audited x86 layout, this queue starts at Connection+96: - // these 28 bytes fit one cache line if the Connection is 64-byte aligned. - // Recheck allocator addresses and throughput/tails after layout changes. - Slot inline_; - std::uint32_t head_ = 0; - std::uint32_t size_ = 0; - std::uint32_t capacity_ = kInlineCapacity; - std::uint32_t low_water_pops_ = 0; - std::unique_ptr data_; -}; +// 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 28 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 pointer follows them. +// Recheck allocator addresses and throughput/tails after layout changes. +using ReceivedBufferQueue = InlineRingQueue; } // namespace bycorf #endif // BYCORF_NET_RECEIVED_BUFFER_QUEUE_H_ diff --git a/tests/received_buffer_queue_check.cpp b/tests/received_buffer_queue_check.cpp index 9ec1df4..4504cb9 100644 --- a/tests/received_buffer_queue_check.cpp +++ b/tests/received_buffer_queue_check.cpp @@ -92,9 +92,130 @@ void CheckInlineAndHeapMoves() { 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); + } + } + 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(); bycorf::ReceivedBufferQueue queue; Check(queue.empty() && queue.capacity() == 1); @@ -157,6 +278,7 @@ int main() { queue.pop_front(); } Check(queue.capacity() == 1); - std::cout << "PASS FIFO, wrap, growth, delayed shrink, partial offsets, " - "move, clear\n"; + std::cout + << "PASS FIFO, wrap, growth, delayed shrink, partial offsets, " + "move, clear, generic types and inline capacities 1/2/4/8/16/32\n"; } From 8d6c983d2ca203ce8cb94f1656410cb76d148309 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 16:23:13 +0800 Subject: [PATCH 08/10] Share inline ring storage with heap ownership in a union --- include/bycorf/base/inline_ring_queue.h | 63 +++++++++++++++++----- include/bycorf/net/received_buffer_queue.h | 5 +- 2 files changed, 53 insertions(+), 15 deletions(-) diff --git a/include/bycorf/base/inline_ring_queue.h b/include/bycorf/base/inline_ring_queue.h index 24f4602..2bef9de 100644 --- a/include/bycorf/base/inline_ring_queue.h +++ b/include/bycorf/base/inline_ring_queue.h @@ -46,6 +46,9 @@ class InlineRingQueue { // 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) std::destroy_at(&storage_.data_); + } InlineRingQueue(const InlineRingQueue&) = delete; InlineRingQueue& operator=(const InlineRingQueue&) = delete; InlineRingQueue(InlineRingQueue&& other) noexcept { @@ -56,12 +59,20 @@ class InlineRingQueue { 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. - data_.reset(); - other.CopyTo(inline_); + if (capacity_ > kInlineCapacity) { + std::destroy_at(&storage_.data_); + std::construct_at(&storage_.inline_); + } + other.CopyTo(storage_.inline_.slots_); head_ = 0; } else { - data_ = std::move(other.data_); + if (capacity_ == kInlineCapacity) + std::construct_at(&storage_.data_, std::move(other.storage_.data_)); + else + storage_.data_ = std::move(other.storage_.data_); head_ = other.head_; + std::destroy_at(&other.storage_.data_); + std::construct_at(&other.storage_.inline_); } other.head_ = 0; size_ = std::exchange(other.size_, 0); @@ -133,11 +144,26 @@ class InlineRingQueue { }; 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_; + std::unique_ptr data_; + StorageUnion() noexcept : inline_() {} + // capacity_ selects the active member; the queue destroys heap ownership. + ~StorageUnion() {} + }; + Slot* Storage() noexcept { - return capacity_ == kInlineCapacity ? inline_ : data_.get(); + return capacity_ == kInlineCapacity ? storage_.inline_.slots_ + : storage_.data_.get(); } const Slot* Storage() const noexcept { - return capacity_ == kInlineCapacity ? inline_ : data_.get(); + return capacity_ == kInlineCapacity ? storage_.inline_.slots_ + : storage_.data_.get(); } std::uint32_t LowWaterLimit() const noexcept { @@ -154,7 +180,11 @@ class InlineRingQueue { capacity_ == kInlineCapacity ? kMinHeapCapacity : capacity_ * 2; auto data = std::make_unique(capacity); CopyTo(data.get()); - data_ = std::move(data); + // Allocation and copying finish before replacing the active union member. + if (capacity_ == kInlineCapacity) + std::construct_at(&storage_.data_, std::move(data)); + else + storage_.data_ = std::move(data); head_ = 0; capacity_ = capacity; low_water_pops_ = 0; @@ -165,20 +195,27 @@ class InlineRingQueue { const auto capacity = capacity_ == kMinHeapCapacity ? kInlineCapacity : capacity_ / 2; if (capacity == kInlineCapacity) { - CopyTo(inline_); - data_.reset(); + // Keep the source array alive locally before its owner is overwritten by + // inline slots. CopyFrom uses the old head/capacity until copying ends. + auto data = std::move(storage_.data_); + std::destroy_at(&storage_.data_); + std::construct_at(&storage_.inline_); + CopyFrom(data.get(), storage_.inline_.slots_); } else { std::unique_ptr data(new (std::nothrow) Slot[capacity]); if (!data) return; CopyTo(data.get()); - data_ = std::move(data); + storage_.data_ = std::move(data); } head_ = 0; capacity_ = capacity; } void CopyTo(Slot* destination) const noexcept { - const auto* source = Storage(); + 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_); @@ -190,13 +227,13 @@ class InlineRingQueue { static constexpr std::uint32_t kMinHeapCapacity = std::max(8u, InlineCapacity * 2); static constexpr std::uint32_t kMaxCapacity = 1u << 30; - // Keep inline values and mutable counters ahead of the cold heap pointer. - Slot inline_[InlineCapacity]; + // Inline values and heap ownership are mutually exclusive. capacity_ is the + // discriminator; every transition explicitly starts/ends member lifetimes. + StorageUnion storage_; std::uint32_t head_ = 0; std::uint32_t size_ = 0; std::uint32_t capacity_ = kInlineCapacity; std::uint32_t low_water_pops_ = 0; - std::unique_ptr data_; }; } // namespace bycorf diff --git a/include/bycorf/net/received_buffer_queue.h b/include/bycorf/net/received_buffer_queue.h index f84f325..2d70739 100644 --- a/include/bycorf/net/received_buffer_queue.h +++ b/include/bycorf/net/received_buffer_queue.h @@ -32,9 +32,10 @@ struct ReceivedBuffer { // 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 28 bytes. In the +// 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 pointer follows them. +// 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. using ReceivedBufferQueue = InlineRingQueue; From 3cc3586ab421f567ab1cae97a5c53841d5c8a271 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 16:26:14 +0800 Subject: [PATCH 09/10] Manage ring heap storage directly with an owning raw pointer --- include/bycorf/base/inline_ring_queue.h | 54 ++++++++++++------------- tests/received_buffer_queue_check.cpp | 19 +++++++++ 2 files changed, 45 insertions(+), 28 deletions(-) diff --git a/include/bycorf/base/inline_ring_queue.h b/include/bycorf/base/inline_ring_queue.h index 2bef9de..e776203 100644 --- a/include/bycorf/base/inline_ring_queue.h +++ b/include/bycorf/base/inline_ring_queue.h @@ -47,7 +47,7 @@ class InlineRingQueue { // inline slots before their first values are constructed. InlineRingQueue() noexcept {} ~InlineRingQueue() { - if (capacity_ > kInlineCapacity) std::destroy_at(&storage_.data_); + if (capacity_ > kInlineCapacity) delete[] storage_.data_; } InlineRingQueue(const InlineRingQueue&) = delete; InlineRingQueue& operator=(const InlineRingQueue&) = delete; @@ -60,18 +60,16 @@ class InlineRingQueue { // Inline elements belong to the queue object and must be copied before // the moved-from queue can be reused or destroyed. if (capacity_ > kInlineCapacity) { - std::destroy_at(&storage_.data_); + delete[] storage_.data_; std::construct_at(&storage_.inline_); } other.CopyTo(storage_.inline_.slots_); head_ = 0; } else { - if (capacity_ == kInlineCapacity) - std::construct_at(&storage_.data_, std::move(other.storage_.data_)); - else - storage_.data_ = std::move(other.storage_.data_); + 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::destroy_at(&other.storage_.data_); std::construct_at(&other.storage_.inline_); } other.head_ = 0; @@ -151,19 +149,17 @@ class InlineRingQueue { }; union StorageUnion { InlineStorage inline_; - std::unique_ptr data_; + Slot* data_; StorageUnion() noexcept : inline_() {} - // capacity_ selects the active member; the queue destroys heap ownership. - ~StorageUnion() {} }; Slot* Storage() noexcept { return capacity_ == kInlineCapacity ? storage_.inline_.slots_ - : storage_.data_.get(); + : storage_.data_; } const Slot* Storage() const noexcept { return capacity_ == kInlineCapacity ? storage_.inline_.slots_ - : storage_.data_.get(); + : storage_.data_; } std::uint32_t LowWaterLimit() const noexcept { @@ -178,13 +174,12 @@ class InlineRingQueue { if (capacity_ == kMaxCapacity) throw std::bad_alloc(); const auto capacity = capacity_ == kInlineCapacity ? kMinHeapCapacity : capacity_ * 2; - auto data = std::make_unique(capacity); - CopyTo(data.get()); - // Allocation and copying finish before replacing the active union member. - if (capacity_ == kInlineCapacity) - std::construct_at(&storage_.data_, std::move(data)); - else - storage_.data_ = std::move(data); + 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; @@ -195,17 +190,18 @@ class InlineRingQueue { const auto capacity = capacity_ == kMinHeapCapacity ? kInlineCapacity : capacity_ / 2; if (capacity == kInlineCapacity) { - // Keep the source array alive locally before its owner is overwritten by - // inline slots. CopyFrom uses the old head/capacity until copying ends. - auto data = std::move(storage_.data_); - std::destroy_at(&storage_.data_); + // 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.get(), storage_.inline_.slots_); + CopyFrom(data, storage_.inline_.slots_); + delete[] data; } else { - std::unique_ptr data(new (std::nothrow) Slot[capacity]); + auto* data = new (std::nothrow) Slot[capacity]; if (!data) return; - CopyTo(data.get()); - storage_.data_ = std::move(data); + CopyTo(data); + delete[] storage_.data_; + storage_.data_ = data; } head_ = 0; capacity_ = capacity; @@ -228,7 +224,9 @@ class InlineRingQueue { 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; every transition explicitly starts/ends member lifetimes. + // 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; diff --git a/tests/received_buffer_queue_check.cpp b/tests/received_buffer_queue_check.cpp index 4504cb9..9e2277e 100644 --- a/tests/received_buffer_queue_check.cpp +++ b/tests/received_buffer_queue_check.cpp @@ -153,6 +153,25 @@ void CheckGenericQueue() { 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); From 7e1eebbff8e5859ebad04bb8ad2227b5cd2d330e Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 21:10:10 +0800 Subject: [PATCH 10/10] Keep receive descriptor definitions with Connection --- CMakeLists.txt | 2 +- include/bycorf/net/connection.h | 18 ++++++++- include/bycorf/net/received_buffer_queue.h | 43 ---------------------- tests/received_buffer_queue_check.cpp | 16 ++++---- 4 files changed, 26 insertions(+), 53 deletions(-) delete mode 100644 include/bycorf/net/received_buffer_queue.h diff --git a/CMakeLists.txt b/CMakeLists.txt index 52e3f41..40e0d5a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -408,7 +408,7 @@ endif() if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS) add_executable(bycorf_received_buffer_queue_check tests/received_buffer_queue_check.cpp) - target_include_directories(bycorf_received_buffer_queue_check PRIVATE include) + 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) diff --git a/include/bycorf/net/connection.h b/include/bycorf/net/connection.h index f8bf7d2..10fd73e 100644 --- a/include/bycorf/net/connection.h +++ b/include/bycorf/net/connection.h @@ -26,7 +26,7 @@ #include "absl/cleanup/cleanup.h" #include "absl/status/statusor.h" -#include "bycorf/net/received_buffer_queue.h" +#include "bycorf/base/inline_ring_queue.h" namespace bycorf { @@ -59,6 +59,12 @@ struct RegisteredFile { bool is_fixed_ = false; }; +struct ReceivedBuffer { + std::uint16_t buffer_id_ = 0; + std::uint32_t size_ = 0; + std::uint32_t offset_ = 0; +}; + struct Connection { using PeerDisconnectCallback = void (*)(void*) noexcept; @@ -103,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; - ReceivedBufferQueue 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/include/bycorf/net/received_buffer_queue.h b/include/bycorf/net/received_buffer_queue.h deleted file mode 100644 index 2d70739..0000000 --- a/include/bycorf/net/received_buffer_queue.h +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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_NET_RECEIVED_BUFFER_QUEUE_H_ -#define BYCORF_NET_RECEIVED_BUFFER_QUEUE_H_ - -#include - -#include "bycorf/base/inline_ring_queue.h" - -namespace bycorf { - -struct ReceivedBuffer { - std::uint16_t buffer_id_ = 0; - std::uint32_t size_ = 0; - std::uint32_t offset_ = 0; -}; - -// 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. -using ReceivedBufferQueue = InlineRingQueue; - -} // namespace bycorf -#endif // BYCORF_NET_RECEIVED_BUFFER_QUEUE_H_ diff --git a/tests/received_buffer_queue_check.cpp b/tests/received_buffer_queue_check.cpp index 9e2277e..d292551 100644 --- a/tests/received_buffer_queue_check.cpp +++ b/tests/received_buffer_queue_check.cpp @@ -21,9 +21,11 @@ #include #include -#include "bycorf/net/received_buffer_queue.h" +#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) { @@ -36,7 +38,7 @@ 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 bycorf::ReceivedBufferQueue& queue, +void Compare(const ReceiveQueue& queue, const std::deque& reference) { Check(queue.size() == reference.size()); for (std::uint32_t i = 0; i < queue.size(); ++i) @@ -45,12 +47,12 @@ void Compare(const bycorf::ReceivedBufferQueue& queue, void CheckInlineAndHeapMoves() { for (bool source_heap : {false, true}) { for (bool destination_heap : {false, true}) { - bycorf::ReceivedBufferQueue destination; + ReceiveQueue destination; for (unsigned i = 0; i < (destination_heap ? 4u : 1u); ++i) destination.push_back({.buffer_id_ = 99}); std::deque expected; { - bycorf::ReceivedBufferQueue source; + ReceiveQueue source; for (unsigned i = 0; i < (source_heap ? 4u : 1u); ++i) { bycorf::ReceivedBuffer value{ .buffer_id_ = static_cast(i), @@ -66,7 +68,7 @@ void CheckInlineAndHeapMoves() { Compare(destination, expected); } Compare(destination, expected); - bycorf::ReceivedBufferQueue constructed(std::move(destination)); + ReceiveQueue constructed(std::move(destination)); Check(destination.empty() && destination.capacity() == 1); destination.push_back({.buffer_id_ = 88}); Compare(constructed, expected); @@ -75,7 +77,7 @@ void CheckInlineAndHeapMoves() { Compare(constructed, expected); } } - bycorf::ReceivedBufferQueue small; + ReceiveQueue small; for (unsigned cycle = 0; cycle < 100; ++cycle) { small.push_back({.buffer_id_ = 1}); Check(small.capacity() == 1); @@ -236,7 +238,7 @@ int main() { CheckGenericTypes<16>(); CheckGenericTypes<32>(); CheckInlineAndHeapMoves(); - bycorf::ReceivedBufferQueue queue; + ReceiveQueue queue; Check(queue.empty() && queue.capacity() == 1); std::deque reference; std::mt19937 random(20261019);