Commit 89b58841 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #60

Refactor write events
parents 0bdff82e be6f5f7c
...@@ -103,17 +103,20 @@ public: ...@@ -103,17 +103,20 @@ public:
} }
bool handle_write_event(endpoint_manager& manager) override { bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id) CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size())); << CAF_ARG2("queue-size", packet_queue_.size()));
// Try to write leftover data. auto fetch_next_message = [&] {
write_some(); if (auto msg = manager.next_message()) {
// Get new data from parent. this->next_layer_.write_message(*this, std::move(msg));
for (auto msg = manager.next_message(); msg != nullptr; return true;
msg = manager.next_message()) { }
this->next_layer_.write_message(*this, std::move(msg)); return false;
} };
// Write prepared data. do {
return write_some(); if (auto err = write_some())
return err == sec::unavailable_or_would_block;
} while (fetch_next_message());
return !packet_queue_.empty();
} }
// TODO: remove this function. `resolve` should add workers when needed. // TODO: remove this function. `resolve` should add workers when needed.
...@@ -163,8 +166,7 @@ private: ...@@ -163,8 +166,7 @@ private:
this->read_buf_.resize(max_datagram_size); this->read_buf_.resize(max_datagram_size);
} }
bool write_some() { error write_some() {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id));
// Helper function to sort empty buffers back into the right caches. // Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() { auto recycle = [&]() {
auto& front = packet_queue_.front(); auto& front = packet_queue_.front();
...@@ -197,13 +199,11 @@ private: ...@@ -197,13 +199,11 @@ private:
if (err != sec::unavailable_or_would_block) { if (err != sec::unavailable_or_would_block) {
CAF_LOG_ERROR("write failed" << CAF_ARG(err)); CAF_LOG_ERROR("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err); this->next_layer_.handle_error(err);
return false;
} }
CAF_LOG_DEBUG("write returned `unavailable_or_would_block`"); return err;
return true;
} }
} }
return false; return none;
} }
std::deque<packet> packet_queue_; std::deque<packet> packet_queue_;
......
...@@ -107,18 +107,63 @@ public: ...@@ -107,18 +107,63 @@ public:
return true; return true;
} }
bool handle_write_event(endpoint_manager& parent) override { bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id)); CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id)
// Try to write leftover data. << CAF_ARG2("queue-size", write_queue_.size()));
write_some(); auto drain_write_queue = []() -> error_code<sec> {
// Get new data from parent. // Helper function to sort empty buffers back into the right caches.
// TODO: dont read all messages at once - get one by one. auto recycle = [&]() {
for (auto msg = parent.next_message(); msg != nullptr; auto& front = this->write_queue_.front();
msg = parent.next_message()) { auto& is_header = front.first;
this->next_layer_.write_message(*this, std::move(msg)); auto& buf = front.second;
} written_ = 0;
// Write prepared data. buf.clear();
return write_some(); if (is_header) {
if (this->header_bufs_.size() < this->header_bufs_.capacity())
this->header_bufs_.emplace_back(std::move(buf));
} else if (this->payload_bufs_.size() < this->payload_bufs_.capacity()) {
this->payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front();
};
// Write buffers from the write_queue_ for as long as possible.
while (!write_queue_.empty()) {
auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_;
auto len = buf.size() - written_;
auto write_ret = write(this->handle(), make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
written_ += *num_bytes;
if (written_ >= buf.size()) {
recycle();
written_ = 0;
}
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
}
return err;
}
}
return none;
};
auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
return true;
}
return false;
};
do {
if (auto err = drain_write_queue())
return err == sec::unavailable_or_would_block;
} while (fetch_next_message());
CAF_ASSERT(write_queue_.empty());
return false;
} }
void write_packet(id_type, span<buffer_type*> buffers) override { void write_packet(id_type, span<buffer_type*> buffers) override {
...@@ -167,50 +212,6 @@ private: ...@@ -167,50 +212,6 @@ private:
} }
} }
bool write_some() {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id));
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = this->write_queue_.front();
auto& is_header = front.first;
auto& buf = front.second;
written_ = 0;
buf.clear();
if (is_header) {
if (this->header_bufs_.size() < this->header_bufs_.capacity())
this->header_bufs_.emplace_back(std::move(buf));
} else if (this->payload_bufs_.size() < this->payload_bufs_.capacity()) {
this->payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front();
};
// Write buffers from the write_queue_ for as long as possible.
while (!write_queue_.empty()) {
auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_;
auto len = buf.size() - written_;
auto write_ret = write(this->handle(), make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
written_ += *num_bytes;
if (written_ >= buf.size()) {
recycle();
written_ = 0;
}
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
return true;
}
}
return false;
}
write_queue_type write_queue_; write_queue_type write_queue_;
size_t written_; size_t written_;
size_t read_threshold_; size_t read_threshold_;
......
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