Commit 77041c4b authored by Dominik Charousset's avatar Dominik Charousset

Fix signaling on the producer adapter

The current fix is to remove any demand bookkeeping. However, this may
cause frequent signaling now. Eventually, we need (working) a mechanism
that reduces wakeup signaling to a minimum.
parent 6e6e01a4
...@@ -37,9 +37,7 @@ public: ...@@ -37,9 +37,7 @@ public:
} }
void on_consumer_demand(size_t new_demand) override { void on_consumer_demand(size_t new_demand) override {
auto prev = demand_.fetch_add(new_demand); mgr_->continue_reading();
if (prev == 0)
mgr_->continue_reading();
} }
void ref_producer() const noexcept override { void ref_producer() const noexcept override {
...@@ -67,19 +65,11 @@ public: ...@@ -67,19 +65,11 @@ public:
} }
} }
/// Returns the current consumer demand.
size_t demand() const noexcept {
return demand_;
}
/// Makes `item` available to the consumer. /// Makes `item` available to the consumer.
/// @returns the remaining demand. /// @returns the remaining demand.
/// @pre `demand() > 0`
size_t push(const value_type& item) { size_t push(const value_type& item) {
if (buf_) { if (buf_) {
CAF_ASSERT(demand_ > 0); return buf_->push(item);
buf_->push(item);
return --demand_;
} else { } else {
return 0; return 0;
} }
...@@ -87,12 +77,9 @@ public: ...@@ -87,12 +77,9 @@ public:
/// Makes `items` available to the consumer. /// Makes `items` available to the consumer.
/// @returns the remaining demand. /// @returns the remaining demand.
/// @pre `demand() >= items.size()`
size_t push(span<const value_type> items) { size_t push(span<const value_type> items) {
if (buf_) { if (buf_) {
CAF_ASSERT(demand_ >= items.size()); return buf_->push(items);
buf_->push(items);
return demand_ -= items.size();
} else { } else {
return 0; return 0;
} }
...@@ -124,7 +111,7 @@ public: ...@@ -124,7 +111,7 @@ public:
private: private:
producer_adapter(socket_manager* owner, buf_ptr buf) producer_adapter(socket_manager* owner, buf_ptr buf)
: demand_(0), mgr_(owner), buf_(std::move(buf)) { : mgr_(owner), buf_(std::move(buf)) {
// nop // nop
} }
...@@ -140,9 +127,6 @@ private: ...@@ -140,9 +127,6 @@ private:
return intrusive_ptr{this}; return intrusive_ptr{this};
} }
atomic_count demand_;
char pad[CAF_CACHE_LINE_SIZE - sizeof(atomic_count)];
intrusive_ptr<socket_manager> mgr_; intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_; intrusive_ptr<Buffer> buf_;
}; };
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment