Commit 35737b2f authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #1276

Closes #1060.
parents 41abdfbc 9ed75df6
......@@ -15,6 +15,11 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Blocking actors now release their private thread before decrementing the
running-actors count to resolve a race condition during system shutdown that
could result in the system hanging (#1266).
- When using the OpenSSL module, CAF could run into a state where the SSL layer
wants to read data while CAF is trying to send data. In this case, CAF did not
properly back off, causing high CPU load due to spinning and in some scenarios
never recovering. This issue has been resolved by properly handling
`SSL_ERROR_WANT_READ` on the transport (#1060).
### Removed
......
......@@ -13,7 +13,9 @@ enum class rw_state {
/// Reports that the socket is closed or faulty.
failure,
/// Reports that an empty buffer is in use and no operation was performed.
indeterminate
indeterminate,
/// Reports that the transport wants to read data before it can write again.
want_read
};
} // namespace caf::io::network
......@@ -136,6 +136,7 @@ private:
size_t written_;
byte_buffer wr_buf_;
byte_buffer wr_offline_buf_;
bool wr_op_backoff_;
};
} // namespace caf::io::network
......@@ -21,7 +21,8 @@ stream::stream(default_multiplexer& backend_ref, native_socket sockfd)
defaults::middleman::max_consecutive_reads)),
read_threshold_(1),
collected_(0),
written_(0) {
written_(0),
wr_op_backoff_(false) {
configure_read(receive_policy::at_most(1024));
}
......@@ -143,6 +144,14 @@ bool stream::handle_read_result(rw_state read_result, size_t rb) {
case rw_state::indeterminate:
return false;
case rw_state::success:
// Recover previous pending write if it is the first successful read after
// want_read was reported.
if (wr_op_backoff_) {
backend().add(operation::write, fd(), this);
wr_op_backoff_ = false;
}
[[fallthrough]];
case rw_state::want_read:
if (rb == 0)
return false;
collected_ += rb;
......@@ -168,6 +177,15 @@ void stream::handle_write_result(rw_state write_result, size_t wb) {
case rw_state::indeterminate:
prepare_next_write();
break;
case rw_state::want_read:
// If the write operation returns want_read, we need to suspend writing to
// the socket until the next successful read. Otherwise, we may cause
// spinning and high CPU usage.
backend().del(operation::write, fd(), this);
wr_op_backoff_ = true;
if (wb == 0)
break;
[[fallthrough]];
case rw_state::success:
written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size());
......
......@@ -94,10 +94,7 @@ rw_state session::do_some(int (*f)(SSL*, void*, int), size_t& result, void* buf,
return rw_state::failure;
case SSL_ERROR_WANT_READ:
CAF_LOG_DEBUG("SSL_ERROR_WANT_READ reported");
// Report success to poll on this socket.
if (len == 0 && strcmp(debug_name, "write_some") == 0)
return rw_state::indeterminate;
return rw_state::success;
return rw_state::want_read;
case SSL_ERROR_WANT_WRITE:
CAF_LOG_DEBUG("SSL_ERROR_WANT_WRITE reported");
// Report success to poll on this socket.
......
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