From 09b5d50722e24b39386176479f8cf6342c979a7e Mon Sep 17 00:00:00 2001 From: Alexander van der Grinten Date: Tue, 12 May 2026 22:06:16 +0200 Subject: [PATCH 1/2] queue: Move allocation out of mutex --- include/async/queue.hpp | 82 ++++++++++++++++++++++++++++++----------- 1 file changed, 61 insertions(+), 21 deletions(-) diff --git a/include/async/queue.hpp b/include/async/queue.hpp index fd4efb2..eec420c 100644 --- a/include/async/queue.hpp +++ b/include/async/queue.hpp @@ -11,9 +11,28 @@ namespace async { template struct queue { queue(Allocator allocator = {}) - : buffer_{allocator} {} + : allocator_{std::move(allocator)} {} + + ~queue() { + assert(sinks_.empty()); + while(!buffer_.empty()) + frg::destruct(allocator_, buffer_.pop_front()); + } + + queue(const queue &) = delete; + queue &operator=(const queue &) = delete; private: + // Note: element is heap allocated. It must be allocated *outside* the mutex. + struct element { + template + element(Ts&&... args) + : item(std::forward(args)...) {} + + T item; + frg::default_list_hook hook; + }; + struct sink { friend struct queue; @@ -24,7 +43,7 @@ struct queue { virtual void complete() = 0; protected: - frg::optional value; + element *el{nullptr}; private: frg::default_list_hook hook_; @@ -33,7 +52,7 @@ struct queue { bool try_cancel(sink *sp) { frg::unique_lock lock{mutex_}; - if(!sp->value) { + if(!sp->el) { auto it = sinks_.iterator_to(sp); sinks_.erase(it); return true; @@ -48,6 +67,8 @@ struct queue { template void emplace(Ts&&... arg) { + auto *el = frg::construct(allocator_, std::forward(arg)...); + sink *complete_sp = nullptr; { frg::unique_lock lock{mutex_}; @@ -55,10 +76,10 @@ struct queue { if(!sinks_.empty()) { assert(buffer_.empty()); auto sp = sinks_.pop_front(); - sp->value.emplace(std::forward(arg)...); + sp->el = el; complete_sp = sp; }else{ - buffer_.emplace_back(std::forward(arg)...); + buffer_.push_back(el); } } @@ -76,28 +97,27 @@ struct queue { : q_{q}, ct_{std::move(ct)}, r_{std::move(r)} { } void start() { - bool fast_path = false; + element *el = nullptr; { frg::unique_lock lock{q_->mutex_}; if(!q_->buffer_.empty()) { assert(q_->sinks_.empty()); - value = std::move(q_->buffer_.front()); - q_->buffer_.pop_front(); - fast_path = true; + el = q_->buffer_.pop_front(); }else{ q_->sinks_.push_back(this); } } - if(fast_path) - return execution::set_value(r_, std::move(value)); + if(el) { + frg::optional result{std::move(el->item)}; + frg::destruct(q_->allocator_, el); + return execution::set_value(r_, std::move(result)); + } cr_.listen(ct_); } private: - using sink::value; - struct try_cancel_fn { bool operator()(auto *cr) { auto self = frg::container_of(cr, &get_operation::cr_); @@ -107,7 +127,12 @@ struct queue { struct resume_fn { void operator()(auto *cr) { auto self = frg::container_of(cr, &get_operation::cr_); - execution::set_value(self->r_, std::move(self->value)); + frg::optional result; + if (self->el) { + result = std::move(self->el->item); + frg::destruct(self->q_->allocator_, self->el); + } + execution::set_value(self->r_, std::move(result)); } }; @@ -142,23 +167,38 @@ struct queue { } bool empty() { + frg::unique_lock lock{mutex_}; + return buffer_.empty(); } frg::optional maybe_get() { - frg::unique_lock lock{mutex_}; + element *el; + { + frg::unique_lock lock{mutex_}; - if(buffer_.empty()) - return {}; - auto object = std::move(buffer_.front()); - buffer_.pop_front(); - return object; + if(buffer_.empty()) + return frg::null_opt; + el = buffer_.pop_front(); + } + auto value = std::move(el->item); + frg::destruct(allocator_, el); + return value; } private: platform::mutex mutex_; - frg::list buffer_; + [[no_unique_address]] Allocator allocator_; + + frg::intrusive_list< + element, + frg::locate_member< + element, + frg::default_list_hook, + &element::hook + > + > buffer_; frg::intrusive_list< sink, From 53c28de404daf4978136891edada9153d66c1390 Mon Sep 17 00:00:00 2001 From: Alexander van der Grinten Date: Tue, 12 May 2026 22:11:48 +0200 Subject: [PATCH 2/2] mutex: Add test() function --- include/async/mutex.hpp | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/include/async/mutex.hpp b/include/async/mutex.hpp index 33c8149..abc84ad 100644 --- a/include/async/mutex.hpp +++ b/include/async/mutex.hpp @@ -109,6 +109,11 @@ namespace detail { // ------------------------------------------------------------------------------ + // Returns true if try_lock() would have succeeded. This is useful for TTAS. + bool test() { + return st_.load(std::memory_order_relaxed) == state::none; + } + bool try_lock() { auto st = state::none; return st_.compare_exchange_strong(