Unverified Commit 47d0921e authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #740

Allow policies to signal internal buffering
parents 28e83bbe 0c76c395
......@@ -99,17 +99,23 @@ protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(op));
auto mcr = max_consecutive_reads_;
switch (op) {
case io::network::operation::read: {
// Loop until an error occurs or we have nothing more to read
// or until we have handled `mcr` reads.
size_t rb;
for (size_t i = 0; i < mcr; ++i) {
size_t rb = 0;
auto threshold = [&] {
CAF_ASSERT(read_threshold_ >= collected_);
return read_threshold_ - collected_;
};
size_t reads = 0;
while (reads < max_consecutive_reads_
|| policy.must_read_more(fd(), threshold())) {
auto res = policy.read_some(rb, fd(), rd_buf_.data() + collected_,
rd_buf_.size() - collected_);
if (!handle_read_result(res, rb))
return;
++reads;
}
break;
}
......
......@@ -47,6 +47,12 @@ struct tcp {
/// as long as
static bool try_accept(io::network::native_socket& result,
io::network::native_socket fd);
/// Always returns `false`. Native TCP I/O event handlers only rely on the
/// socket buffer.
static constexpr bool must_read_more(io::network::native_socket, size_t) {
return false;
}
};
} // namespace policy
......
......@@ -41,6 +41,12 @@ struct udp {
static bool write_datagram(size_t& result, io::network::native_socket fd,
void* buf, size_t buf_len,
const io::network::ip_endpoint& ep);
/// Always returns `false`. Native UDP I/O event handlers only rely on the
/// socket buffer.
static constexpr bool must_read_more(io::network::native_socket, size_t) {
return false;
}
};
} // namespace policy
......
......@@ -58,6 +58,9 @@ public:
size_t len);
bool try_connect(native_socket fd);
bool try_accept(native_socket fd);
bool must_read_more(native_socket, size_t threshold);
const char* openssl_passphrase();
private:
......
......@@ -89,6 +89,10 @@ struct ssl_policy {
return session_->try_accept(result);
}
bool must_read_more(native_socket fd, size_t threshold) {
return session_->must_read_more(fd, threshold);
}
private:
session_ptr session_;
};
......
......@@ -197,6 +197,10 @@ bool session::try_accept(native_socket fd) {
return handle_ssl_result(ret);
}
bool session::must_read_more(native_socket, size_t threshold) {
return static_cast<size_t>(SSL_pending(ssl_)) >= threshold;
}
const char* session::openssl_passphrase() {
return openssl_passphrase_.c_str();
}
......
......@@ -46,6 +46,11 @@ public:
add_message_type<std::vector<int>>("std::vector<int>");
actor_system_config::parse(test::engine::argc(),
test::engine::argv());
// Setting the "max consecutive reads" to 1 is highly likely to cause
// OpenSSL to buffer data internally and report "pending" data after a read
// operation. This will trigger `must_read_more` in the SSL read policy
// with high probability.
set("middleman.max-consecutive-reads", 1);
}
};
......
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