Commit ea9def46 authored by Dominik Charousset's avatar Dominik Charousset

Allow policies to signal internal buffering

Policies that internally buffer data, such as the OpenSSL policy, can
cause broker to block despite having data to read and even lose data at
the end of a transmission.
parent da9956ac
......@@ -99,18 +99,37 @@ 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) {
auto res = policy.read_some(rb, fd(), rd_buf_.data() + collected_,
rd_buf_.size() - collected_);
if (!handle_read_result(res, rb))
// Stores how many bytes the last read_some() procued.
size_t rb = 0;
// Threshold configured by the actor's read policy.
auto threshold = [&] {
CAF_ASSERT(read_threshold_ >= collected_);
return read_threshold_ - collected_;
};
// Returns how much bytes the next read_some() wants to read.
auto read_size = [&] {
return rd_buf_.size() - collected_;
};
// Returns `true` if we could read something, `false` otherwise.
auto read_some = [&] {
return policy.read_some(rb, fd(), rd_buf_.data() + collected_,
read_size());
};
// Loop until reaching the mcr threshold or an error occurs.
size_t reads = 0;
for (; reads < max_consecutive_reads_; ++reads)
if (!handle_read_result(read_some(), rb))
return;
// Some policies (such as SSL) buffer data internally. In this case, we
// have to read more despite having reached max_consecutive_reads_.
// Otherwise, brokers can block despite having data to read and lose
// data at the end of a transmission.
if (reads == max_consecutive_reads_)
while (policy.must_read_more(fd(), threshold()))
if (!handle_read_result(read_some(), rb))
return;
}
break;
}
case io::network::operation::write: {
......
......@@ -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