Unverified Commit 4b373edb authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #60

Refactor write events
parents cf2eb1e4 d89f0c67
......@@ -103,17 +103,20 @@ public:
}
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()));
// Try to write leftover data.
write_some();
// Get new data from parent.
for (auto msg = manager.next_message(); msg != nullptr;
msg = manager.next_message()) {
auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
return true;
}
// Write prepared data.
return write_some();
return false;
};
do {
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.
......@@ -163,8 +166,7 @@ private:
this->read_buf_.resize(max_datagram_size);
}
bool write_some() {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id));
error write_some() {
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = packet_queue_.front();
......@@ -197,13 +199,11 @@ private:
if (err != sec::unavailable_or_would_block) {
CAF_LOG_ERROR("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
CAF_LOG_DEBUG("write returned `unavailable_or_would_block`");
return true;
return err;
}
}
return false;
return none;
}
std::deque<packet> packet_queue_;
......
......@@ -107,18 +107,63 @@ public:
return true;
}
bool handle_write_event(endpoint_manager& parent) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id));
// Try to write leftover data.
write_some();
// Get new data from parent.
// TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle_.id)
<< CAF_ARG2("queue-size", write_queue_.size()));
auto drain_write_queue = []() -> error_code<sec> {
// 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 err;
}
}
return none;
};
auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
return true;
}
// Write prepared data.
return write_some();
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 {
......@@ -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_;
size_t written_;
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