summaryrefslogtreecommitdiff
path: root/boost/fiber/unbuffered_channel.hpp
diff options
context:
space:
mode:
Diffstat (limited to 'boost/fiber/unbuffered_channel.hpp')
-rw-r--r--boost/fiber/unbuffered_channel.hpp199
1 files changed, 93 insertions, 106 deletions
diff --git a/boost/fiber/unbuffered_channel.hpp b/boost/fiber/unbuffered_channel.hpp
index 582d9ae5a7..38a2d6111e 100644
--- a/boost/fiber/unbuffered_channel.hpp
+++ b/boost/fiber/unbuffered_channel.hpp
@@ -54,13 +54,14 @@ private:
};
// shared cacheline
- alignas(cache_alignment) std::atomic< slot * > slot_{ nullptr };
+ alignas(cache_alignment) std::atomic< slot * > slot_{ nullptr };
// shared cacheline
- alignas(cache_alignment) std::atomic_bool closed_{ false };
- mutable detail::spinlock splk_{};
- wait_queue_type waiting_producers_{};
- wait_queue_type waiting_consumers_{};
- char pad_[cacheline_length];
+ alignas(cache_alignment) std::atomic_bool closed_{ false };
+ alignas(cache_alignment) mutable detail::spinlock splk_producers_{};
+ wait_queue_type waiting_producers_{};
+ alignas( cache_alignment) mutable detail::spinlock splk_consumers_{};
+ wait_queue_type waiting_consumers_{};
+ char pad_[cacheline_length];
bool is_empty_() {
return nullptr == slot_.load( std::memory_order_acquire);
@@ -68,7 +69,7 @@ private:
bool try_push_( slot * own_slot) {
for (;;) {
- slot * s{ slot_.load( std::memory_order_acquire) };
+ slot * s = slot_.load( std::memory_order_acquire);
if ( nullptr == s) {
if ( ! slot_.compare_exchange_strong( s, own_slot, std::memory_order_acq_rel) ) {
continue;
@@ -81,9 +82,9 @@ private:
}
slot * try_pop_() {
- slot * nil_slot{ nullptr };
+ slot * nil_slot = nullptr;
for (;;) {
- slot * s{ slot_.load( std::memory_order_acquire) };
+ slot * s = slot_.load( std::memory_order_acquire);
if ( nullptr != s) {
if ( ! slot_.compare_exchange_strong( s, nil_slot, std::memory_order_acq_rel) ) {
continue;}
@@ -97,12 +98,12 @@ public:
~unbuffered_channel() {
close();
- slot * s{ nullptr };
+ slot * s = nullptr;
if ( nullptr != ( s = try_pop_() ) ) {
BOOST_ASSERT( nullptr != s);
BOOST_ASSERT( nullptr != s->ctx);
// value will be destructed in the context of the waiting fiber
- context::active()->set_ready( s->ctx);
+ context::active()->schedule( s->ctx);
}
}
@@ -114,90 +115,89 @@ public:
}
void close() noexcept {
- context * ctx{ context::active() };
- detail::spinlock_lock lk{ splk_ };
- closed_.store( true, std::memory_order_release);
+ context * active_ctx = context::active();
// notify all waiting producers
+ closed_.store( true, std::memory_order_release);
+ detail::spinlock_lock lk1{ splk_producers_ };
while ( ! waiting_producers_.empty() ) {
- context * producer_ctx{ & waiting_producers_.front() };
+ context * producer_ctx = & waiting_producers_.front();
waiting_producers_.pop_front();
- ctx->set_ready( producer_ctx);
+ active_ctx->schedule( producer_ctx);
}
// notify all waiting consumers
+ detail::spinlock_lock lk2{ splk_consumers_ };
while ( ! waiting_consumers_.empty() ) {
- context * consumer_ctx{ & waiting_consumers_.front() };
+ context * consumer_ctx = & waiting_consumers_.front();
waiting_consumers_.pop_front();
- ctx->set_ready( consumer_ctx);
+ active_ctx->schedule( consumer_ctx);
}
}
channel_op_status push( value_type const& value) {
- context * ctx{ context::active() };
- slot s{ value, ctx };
+ context * active_ctx = context::active();
+ slot s{ value, active_ctx };
for (;;) {
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( try_push_( & s) ) {
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_consumers_ };
// notify one waiting consumer
if ( ! waiting_consumers_.empty() ) {
- context * consumer_ctx{ & waiting_consumers_.front() };
+ context * consumer_ctx = & waiting_consumers_.front();
waiting_consumers_.pop_front();
- ctx->set_ready( consumer_ctx);
+ active_ctx->schedule( consumer_ctx);
}
// suspend till value has been consumed
- ctx->suspend( lk);
+ active_ctx->suspend( lk);
// resumed, value has been consumed
return channel_op_status::success;
} else {
- BOOST_ASSERT( ! ctx->wait_is_linked() );
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_producers_ };
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( is_empty_() ) {
continue;
}
- ctx->wait_link( waiting_producers_);
+ active_ctx->wait_link( waiting_producers_);
// suspend this producer
- ctx->suspend( lk);
+ active_ctx->suspend( lk);
// resumed, slot mabye free
}
}
}
channel_op_status push( value_type && value) {
- context * ctx{ context::active() };
- slot s{ std::move( value), ctx };
+ context * active_ctx = context::active();
+ slot s{ std::move( value), active_ctx };
for (;;) {
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( try_push_( & s) ) {
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_consumers_ };
// notify one waiting consumer
if ( ! waiting_consumers_.empty() ) {
- context * consumer_ctx{ & waiting_consumers_.front() };
+ context * consumer_ctx = & waiting_consumers_.front();
waiting_consumers_.pop_front();
- ctx->set_ready( consumer_ctx);
+ active_ctx->schedule( consumer_ctx);
}
// suspend till value has been consumed
- ctx->suspend( lk);
+ active_ctx->suspend( lk);
// resumed, value has been consumed
return channel_op_status::success;
} else {
- BOOST_ASSERT( ! ctx->wait_is_linked() );
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_producers_ };
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( is_empty_() ) {
continue;
}
- ctx->wait_link( waiting_producers_);
+ active_ctx->wait_link( waiting_producers_);
// suspend this producer
- ctx->suspend( lk);
+ active_ctx->suspend( lk);
// resumed, slot mabye free
}
}
@@ -220,51 +220,46 @@ public:
template< typename Clock, typename Duration >
channel_op_status push_wait_until( value_type const& value,
std::chrono::time_point< Clock, Duration > const& timeout_time_) {
- std::chrono::steady_clock::time_point timeout_time( detail::convert( timeout_time_) );
- context * ctx{ context::active() };
- slot s{ value, ctx };
+ context * active_ctx = context::active();
+ slot s{ value, active_ctx };
+ std::chrono::steady_clock::time_point timeout_time = detail::convert( timeout_time_);
for (;;) {
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( try_push_( & s) ) {
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_consumers_ };
// notify one waiting consumer
if ( ! waiting_consumers_.empty() ) {
- context * consumer_ctx{ & waiting_consumers_.front() };
+ context * consumer_ctx = & waiting_consumers_.front();
waiting_consumers_.pop_front();
- ctx->set_ready( consumer_ctx);
+ active_ctx->schedule( consumer_ctx);
}
// suspend this producer
- if ( ! ctx->wait_until( timeout_time, lk) ) {
+ if ( ! active_ctx->wait_until( timeout_time, lk) ) {
// clear slot
- slot * nil_slot{ nullptr }, * own_slot{ & s };
+ slot * nil_slot = nullptr, * own_slot = & s;
slot_.compare_exchange_strong( own_slot, nil_slot, std::memory_order_acq_rel);
- // relock local lk
- lk.lock();
- // remove from waiting-queue
- ctx->wait_unlink();
// resumed, value has not been consumed
return channel_op_status::timeout;
}
// resumed, value has been consumed
return channel_op_status::success;
} else {
- BOOST_ASSERT( ! ctx->wait_is_linked() );
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_producers_ };
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( is_empty_() ) {
continue;
}
- ctx->wait_link( waiting_producers_);
+ active_ctx->wait_link( waiting_producers_);
// suspend this producer
- if ( ! ctx->wait_until( timeout_time, lk) ) {
+ if ( ! active_ctx->wait_until( timeout_time, lk) ) {
// relock local lk
lk.lock();
// remove from waiting-queue
- ctx->wait_unlink();
+ waiting_producers_.remove( * active_ctx);
return channel_op_status::timeout;
}
// resumed, slot maybe free
@@ -275,51 +270,46 @@ public:
template< typename Clock, typename Duration >
channel_op_status push_wait_until( value_type && value,
std::chrono::time_point< Clock, Duration > const& timeout_time_) {
- std::chrono::steady_clock::time_point timeout_time( detail::convert( timeout_time_) );
- context * ctx{ context::active() };
- slot s{ std::move( value), ctx };
+ context * active_ctx = context::active();
+ slot s{ std::move( value), active_ctx };
+ std::chrono::steady_clock::time_point timeout_time = detail::convert( timeout_time_);
for (;;) {
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( try_push_( & s) ) {
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_consumers_ };
// notify one waiting consumer
if ( ! waiting_consumers_.empty() ) {
- context * consumer_ctx{ & waiting_consumers_.front() };
+ context * consumer_ctx = & waiting_consumers_.front();
waiting_consumers_.pop_front();
- ctx->set_ready( consumer_ctx);
+ active_ctx->schedule( consumer_ctx);
}
// suspend this producer
- if ( ! ctx->wait_until( timeout_time, lk) ) {
+ if ( ! active_ctx->wait_until( timeout_time, lk) ) {
// clear slot
- slot * nil_slot{ nullptr }, * own_slot{ & s };
+ slot * nil_slot = nullptr, * own_slot = & s;
slot_.compare_exchange_strong( own_slot, nil_slot, std::memory_order_acq_rel);
- // relock local lk
- lk.lock();
- // remove from waiting-queue
- ctx->wait_unlink();
// resumed, value has not been consumed
return channel_op_status::timeout;
}
// resumed, value has been consumed
return channel_op_status::success;
} else {
- BOOST_ASSERT( ! ctx->wait_is_linked() );
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_producers_ };
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( is_empty_() ) {
continue;
}
- ctx->wait_link( waiting_producers_);
+ active_ctx->wait_link( waiting_producers_);
// suspend this producer
- if ( ! ctx->wait_until( timeout_time, lk) ) {
+ if ( ! active_ctx->wait_until( timeout_time, lk) ) {
// relock local lk
lk.lock();
// remove from waiting-queue
- ctx->wait_unlink();
+ waiting_producers_.remove( * active_ctx);
return channel_op_status::timeout;
}
// resumed, slot maybe free
@@ -328,65 +318,63 @@ public:
}
channel_op_status pop( value_type & value) {
- context * ctx{ context::active() };
- slot * s{ nullptr };
+ context * active_ctx = context::active();
+ slot * s = nullptr;
for (;;) {
if ( nullptr != ( s = try_pop_() ) ) {
{
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_producers_ };
// notify one waiting producer
if ( ! waiting_producers_.empty() ) {
- context * producer_ctx{ & waiting_producers_.front() };
+ context * producer_ctx = & waiting_producers_.front();
waiting_producers_.pop_front();
lk.unlock();
- ctx->set_ready( producer_ctx);
+ active_ctx->schedule( producer_ctx);
}
}
// consume value
value = std::move( s->value);
// resume suspended producer
- ctx->set_ready( s->ctx);
+ active_ctx->schedule( s->ctx);
return channel_op_status::success;
} else {
- BOOST_ASSERT( ! ctx->wait_is_linked() );
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_consumers_ };
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( ! is_empty_() ) {
continue;
}
- ctx->wait_link( waiting_consumers_);
+ active_ctx->wait_link( waiting_consumers_);
// suspend this consumer
- ctx->suspend( lk);
+ active_ctx->suspend( lk);
// resumed, slot mabye set
}
}
}
value_type value_pop() {
- context * ctx{ context::active() };
- slot * s{ nullptr };
+ context * active_ctx = context::active();
+ slot * s = nullptr;
for (;;) {
if ( nullptr != ( s = try_pop_() ) ) {
{
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_producers_ };
// notify one waiting producer
if ( ! waiting_producers_.empty() ) {
- context * producer_ctx{ & waiting_producers_.front() };
+ context * producer_ctx = & waiting_producers_.front();
waiting_producers_.pop_front();
lk.unlock();
- ctx->set_ready( producer_ctx);
+ active_ctx->schedule( producer_ctx);
}
}
// consume value
value_type value{ std::move( s->value) };
// resume suspended producer
- ctx->set_ready( s->ctx);
+ active_ctx->schedule( s->ctx);
return std::move( value);
} else {
- BOOST_ASSERT( ! ctx->wait_is_linked() );
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_consumers_ };
if ( is_closed() ) {
throw fiber_error{
std::make_error_code( std::errc::operation_not_permitted),
@@ -395,9 +383,9 @@ public:
if ( ! is_empty_() ) {
continue;
}
- ctx->wait_link( waiting_consumers_);
+ active_ctx->wait_link( waiting_consumers_);
// suspend this consumer
- ctx->suspend( lk);
+ active_ctx->suspend( lk);
// resumed, slot mabye set
}
}
@@ -413,42 +401,41 @@ public:
template< typename Clock, typename Duration >
channel_op_status pop_wait_until( value_type & value,
std::chrono::time_point< Clock, Duration > const& timeout_time_) {
- std::chrono::steady_clock::time_point timeout_time( detail::convert( timeout_time_) );
- context * ctx{ context::active() };
- slot * s{ nullptr };
+ context * active_ctx = context::active();
+ slot * s = nullptr;
+ std::chrono::steady_clock::time_point timeout_time = detail::convert( timeout_time_);
for (;;) {
if ( nullptr != ( s = try_pop_() ) ) {
{
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_producers_ };
// notify one waiting producer
if ( ! waiting_producers_.empty() ) {
- context * producer_ctx{ & waiting_producers_.front() };
+ context * producer_ctx = & waiting_producers_.front();
waiting_producers_.pop_front();
lk.unlock();
- ctx->set_ready( producer_ctx);
+ active_ctx->schedule( producer_ctx);
}
}
// consume value
value = std::move( s->value);
// resume suspended producer
- ctx->set_ready( s->ctx);
+ active_ctx->schedule( s->ctx);
return channel_op_status::success;
} else {
- BOOST_ASSERT( ! ctx->wait_is_linked() );
- detail::spinlock_lock lk{ splk_ };
+ detail::spinlock_lock lk{ splk_consumers_ };
if ( is_closed() ) {
return channel_op_status::closed;
}
if ( ! is_empty_() ) {
continue;
}
- ctx->wait_link( waiting_consumers_);
+ active_ctx->wait_link( waiting_consumers_);
// suspend this consumer
- if ( ! ctx->wait_until( timeout_time, lk) ) {
+ if ( ! active_ctx->wait_until( timeout_time, lk) ) {
// relock local lk
lk.lock();
// remove from waiting-queue
- ctx->wait_unlink();
+ waiting_consumers_.remove( * active_ctx);
return channel_op_status::timeout;
}
}
@@ -459,8 +446,8 @@ public:
private:
typedef typename std::aligned_storage< sizeof( value_type), alignof( value_type) >::type storage_type;
- unbuffered_channel * chan_{ nullptr };
- storage_type storage_;
+ unbuffered_channel * chan_{ nullptr };
+ storage_type storage_;
void increment_() {
BOOST_ASSERT( nullptr != chan_);
generated by cgit v1.2.3 (git 2.25.1) at 2024-04-18 05:36:08 +0000