Commit 0f2feb29 authored by Jakob Otto's avatar Jakob Otto

Reimplement write_event handling

parent 1bbb1f97
...@@ -30,4 +30,7 @@ extern const size_t max_payload_buffers; ...@@ -30,4 +30,7 @@ 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
...@@ -106,16 +106,18 @@ public: ...@@ -106,16 +106,18 @@ public:
bool handle_write_event(endpoint_manager& parent) override { bool handle_write_event(endpoint_manager& parent) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id)); CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id));
// Try to write leftover data. auto get_messages = [&] {
write_some(); if (write_queue_.empty()) {
// Get new data from parent. for (auto msg = parent.next_message(); msg != nullptr;
// TODO: dont read all messages at once - get one by one. msg = parent.next_message()) {
for (auto msg = parent.next_message(); msg != nullptr; this->next_layer_.write_message(*this, std::move(msg));
msg = parent.next_message()) { }
this->next_layer_.write_message(*this, std::move(msg)); }
} return !write_queue_.empty();
// Write prepared data. };
return write_some(); while (write_some() && get_messages())
; // nop
return !write_queue_.empty();
} }
void write_packet(id_type, span<buffer_type*> buffers) override { void write_packet(id_type, span<buffer_type*> buffers) override {
...@@ -200,12 +202,11 @@ private: ...@@ -200,12 +202,11 @@ private:
if (err != sec::unavailable_or_would_block) { if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err)); CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
this->next_layer_.handle_error(err); this->next_layer_.handle_error(err);
return false;
} }
return true; return false;
} }
} }
return false; return true;
} }
write_queue_type write_queue_; write_queue_type write_queue_;
......
...@@ -100,6 +100,8 @@ public: ...@@ -100,6 +100,8 @@ 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);
...@@ -224,6 +226,8 @@ protected: ...@@ -224,6 +226,8 @@ 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,4 +24,6 @@ const size_t max_payload_buffers = 100; ...@@ -24,4 +24,6 @@ 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