Commit 79dc3be5 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #59

Implement Read retries
parents e735d89a e3aaaf58
...@@ -80,20 +80,25 @@ public: ...@@ -80,20 +80,25 @@ public:
bool handle_read_event(endpoint_manager&) override { bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id)); CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) {
auto ret = read(this->handle_, this->read_buf_); auto ret = read(this->handle_, this->read_buf_);
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) { if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto num_bytes = res->first; auto& [num_bytes, ep] = *res;
CAF_LOG_DEBUG("received " << num_bytes << " bytes"); CAF_LOG_DEBUG("received " << num_bytes << " bytes");
auto ep = res->second;
this->read_buf_.resize(num_bytes); this->read_buf_.resize(num_bytes);
this->next_layer_.handle_data(*this, this->read_buf_, std::move(ep)); this->next_layer_.handle_data(*this, this->read_buf_, std::move(ep));
prepare_next_read(); prepare_next_read();
} else { } else {
auto err = get<sec>(ret); auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err)); if (err == sec::unavailable_or_would_block) {
break;
} else {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err); this->next_layer_.handle_error(err);
return false; return false;
} }
}
}
return true; return true;
} }
......
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp" #include "caf/net/receive_policy.hpp"
...@@ -74,10 +73,11 @@ public: ...@@ -74,10 +73,11 @@ public:
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
bool handle_read_event(endpoint_manager&) override { bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id));
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) {
auto buf = this->read_buf_.data() + this->collected_; auto buf = this->read_buf_.data() + this->collected_;
size_t len = this->read_threshold_ - this->collected_; size_t len = this->read_threshold_ - this->collected_;
CAF_LOG_TRACE(CAF_ARG2("handle", this->handle().id) CAF_LOG_DEBUG(CAF_ARG2("missing", len));
<< CAF_ARG2("missing", len));
auto ret = read(this->handle_, make_span(buf, len)); auto ret = read(this->handle_, make_span(buf, len));
// Update state. // Update state.
if (auto num_bytes = get_if<size_t>(&ret)) { if (auto num_bytes = get_if<size_t>(&ret)) {
...@@ -95,12 +95,15 @@ public: ...@@ -95,12 +95,15 @@ public:
} }
} else { } else {
auto err = get<sec>(ret); auto err = get<sec>(ret);
if (err != sec::unavailable_or_would_block) { if (err == sec::unavailable_or_would_block) {
break;
} else {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err)); CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err); this->next_layer_.handle_error(err);
return false; return false;
} }
} }
}
return true; return true;
} }
...@@ -214,8 +217,6 @@ private: ...@@ -214,8 +217,6 @@ private:
size_t collected_; size_t collected_;
size_t max_; size_t max_;
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
}; };
} // namespace caf::net } // namespace caf::net
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include "caf/defaults.hpp"
#include "caf/detail/overload.hpp" #include "caf/detail/overload.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
...@@ -100,6 +101,9 @@ public: ...@@ -100,6 +101,9 @@ public:
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
manager_ = &parent; manager_ = &parent;
auto& cfg = system().config(); auto& cfg = system().config();
max_consecutive_reads_ = get_or(this->system().config(),
"middleman.max-consecutive-reads",
defaults::middleman::max_consecutive_reads);
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 +228,8 @@ protected: ...@@ -224,6 +228,8 @@ protected:
buffer_type read_buf_; buffer_type read_buf_;
endpoint_manager* manager_; endpoint_manager* manager_;
size_t max_consecutive_reads_;
}; };
} // namespace caf::net } // namespace caf::net
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