Commit 4a5cb104 authored by Jakob Otto's avatar Jakob Otto

Refactor udp

parent 0f2feb29
...@@ -98,17 +98,20 @@ public: ...@@ -98,17 +98,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", this->packet_queue_.size()));
// Try to write leftover data. auto get_messages = [&] {
write_some(); if (packet_queue_.empty()) {
// Get new data from parent. for (auto msg = manager.next_message(); msg != nullptr;
for (auto msg = manager.next_message(); msg != nullptr; msg = manager.next_message()) {
msg = manager.next_message()) { this->next_layer_.write_message(*this, std::move(msg));
this->next_layer_.write_message(*this, std::move(msg)); }
} }
// Write prepared data. return !packet_queue_.empty();
return write_some(); };
while (write_some() && get_messages())
; // nop
return !packet_queue_.empty();
} }
// TODO: remove this function. `resolve` should add workers when needed. // TODO: remove this function. `resolve` should add workers when needed.
...@@ -159,7 +162,6 @@ private: ...@@ -159,7 +162,6 @@ private:
} }
bool write_some() { bool 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();
...@@ -192,13 +194,11 @@ private: ...@@ -192,13 +194,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 false;
return true;
} }
} }
return false; return true;
} }
std::deque<packet> packet_queue_; std::deque<packet> packet_queue_;
......
...@@ -30,7 +30,4 @@ extern const size_t max_payload_buffers; ...@@ -30,7 +30,4 @@ extern const size_t max_payload_buffers;
/// Maximum number of cached buffers for sending headers. /// Maximum number of cached buffers for sending headers.
extern const size_t max_header_buffers; extern const size_t max_header_buffers;
/// Maximum number of packets to be held in write_queue.
extern const size_t max_write_queue_size;
} // namespace caf::defaults::middleman } // namespace caf::defaults::middleman
...@@ -104,12 +104,13 @@ public: ...@@ -104,12 +104,13 @@ 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)
<< CAF_ARG2("queue-size", write_queue_.size()));
auto get_messages = [&] { auto get_messages = [&] {
if (write_queue_.empty()) { if (write_queue_.empty()) {
for (auto msg = parent.next_message(); msg != nullptr; for (auto msg = manager.next_message(); msg != nullptr;
msg = parent.next_message()) { msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg)); this->next_layer_.write_message(*this, std::move(msg));
} }
} }
...@@ -167,7 +168,6 @@ private: ...@@ -167,7 +168,6 @@ private:
} }
bool write_some() { bool 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 = this->write_queue_.front(); auto& front = this->write_queue_.front();
......
...@@ -100,8 +100,6 @@ public: ...@@ -100,8 +100,6 @@ public:
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
manager_ = &parent; manager_ = &parent;
auto& cfg = system().config(); auto& cfg = system().config();
max_write_queue_size_ = get_or(cfg, "middleman.max-write-queue-size",
defaults::middleman::max_write_queue_size);
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers", auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
defaults::middleman::max_header_buffers); defaults::middleman::max_header_buffers);
header_bufs_.reserve(max_header_bufs); header_bufs_.reserve(max_header_bufs);
...@@ -226,8 +224,6 @@ protected: ...@@ -226,8 +224,6 @@ protected:
buffer_type read_buf_; buffer_type read_buf_;
endpoint_manager* manager_; endpoint_manager* manager_;
size_t max_write_queue_size_;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -24,6 +24,4 @@ const size_t max_payload_buffers = 100; ...@@ -24,6 +24,4 @@ const size_t max_payload_buffers = 100;
const size_t max_header_buffers = 10; const size_t max_header_buffers = 10;
const size_t max_write_queue_size = 2000;
} // namespace caf::defaults::middleman } // namespace caf::defaults::middleman
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