Commit 3fd37d26 authored by Dominik Charousset's avatar Dominik Charousset

Fix demand signaling and WebSocket processing

parent 823cfbea
......@@ -62,7 +62,7 @@ public:
buf_.insert(buf_.end(), items.begin(), items.end());
if (buf_.size() == items.size() && consumer_)
consumer_->on_producer_wakeup();
if (capacity_ >= buf_.size())
if (capacity_ > buf_.size())
return capacity_ - buf_.size();
else
return 0;
......@@ -235,13 +235,8 @@ public:
consumer_buf_.assign(make_move_iterator(buf_.begin()),
make_move_iterator(buf_.begin() + n));
buf_.erase(buf_.begin(), buf_.begin() + n);
if (overflow == 0) {
signal_demand(n);
} else if (n <= overflow) {
overflow -= n;
} else {
if (n > overflow) {
signal_demand(n - overflow);
overflow = 0;
}
guard.unlock();
auto items = span<const T>{consumer_buf_.data(), n};
......@@ -251,6 +246,7 @@ public:
consumed += n;
consumer_buf_.clear();
guard.lock();
overflow = buf_.size() <= capacity_ ? 0u : buf_.size() - capacity_;
}
if (!buf_.empty() || !closed_) {
return {true, consumed};
......@@ -268,9 +264,13 @@ private:
void ready() {
producer_->on_consumer_ready();
consumer_->on_producer_ready();
if (!buf_.empty())
if (!buf_.empty()) {
consumer_->on_producer_wakeup();
signal_demand(capacity_);
if (capacity_ > buf_.size())
signal_demand(capacity_ - buf_.size());
} else {
signal_demand(capacity_);
}
}
void signal_demand(uint32_t new_demand) {
......
......@@ -47,7 +47,7 @@ public:
}
void on_error(const error& what) {
CAF_ASSERT(sub->subscribed());
CAF_ASSERT(sub->in_.valid());
sub->in_.dispose();
sub->in_ = nullptr;
sub->err_ = what;
......@@ -215,7 +215,9 @@ private:
--demand_;
out_.on_next(item);
}
if (!in_ && buf_.empty()) {
if (in_) {
pull();
} else if (buf_.empty()) {
if (err_)
out_.on_error(err_);
else
......
......@@ -117,18 +117,23 @@ public:
void fwd_on_next(input_key key, const T& item) {
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(item));
if (auto ptr = get(key)) {
ptr->buf.push_back(item);
if (ptr->buf.size() == 1 && !flags_.running) {
flags_.running = true;
do_run();
if (!flags_.running && demand_ > 0) {
CAF_ASSERT(out_.valid());
--demand_;
out_.on_next(item);
ptr->sub.request(1);
} else {
ptr->buf.push_back(item);
}
}
}
void fwd_on_next(input_key key, const observable<T>& item) {
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(item));
if (auto ptr = get(key))
if (auto ptr = get(key)) {
subscribe_to(item);
ptr->sub.request(1);
}
}
// -- implementation of subscription_impl ------------------------------------
......@@ -150,7 +155,8 @@ public:
void request(size_t n) override {
CAF_ASSERT(out_.valid());
demand_ += n;
run_later();
if (demand_ == n)
run_later();
}
size_t buffered() const noexcept {
......@@ -181,9 +187,10 @@ private:
if (has_items_at(start))
return inputs_.begin() + start;
while (pos_ != start) {
if (has_items_at(pos_))
return inputs_.begin() + pos_;
auto p = pos_;
pos_ = (pos_ + 1) % inputs_.size();
if (has_items_at(p))
return inputs_.begin() + p;
}
return inputs_.end();
}
......@@ -195,14 +202,14 @@ private:
auto& buf = i->second->buf;
auto tmp = std::move(buf.front());
buf.pop_front();
if (auto& sub = i->second->sub)
if (auto& sub = i->second->sub) {
sub.request(1);
else if (buf.empty())
} else if (buf.empty()) {
inputs_.erase(i);
}
out_.on_next(tmp);
} else {
flags_.running = false;
return;
break;
}
}
if (out_ && inputs_.empty())
......
......@@ -104,12 +104,18 @@ public:
CAF_LOG_ERROR("connection_acceptor aborts due to an error:" << reason);
factory_->abort(reason);
on_conn_close_.dispose();
self_ref_ = nullptr;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
}
private:
void connection_closed() {
if (open_connections_-- == max_connections_)
if (open_connections_ == max_connections_)
owner_->register_reading();
--open_connections_;
}
Socket fd_;
......@@ -125,6 +131,11 @@ private:
action on_conn_close_;
settings cfg_;
/// Type-erased handle to the @ref socket_manager. This reference is important
/// to keep the acceptor alive while the manager is not registered for writing
/// or reading.
disposable self_ref_;
};
} // namespace caf::net
......@@ -33,9 +33,12 @@ public:
net::socket_manager_ptr make(net::multiplexer* mpx, socket_type fd) override {
auto app = net::web_socket::flow_bridge<Trait>::make(mpx, connector_);
auto app_ptr = app.get();
auto ws = net::web_socket::server::make(std::move(app));
auto transport = Transport::make(fd, std::move(ws));
return net::socket_manager::make(mpx, std::move(transport));
auto mgr = net::socket_manager::make(mpx, std::move(transport));
app_ptr->self_ref(mgr->as_disposable());
return mgr;
}
private:
......@@ -92,7 +95,9 @@ disposable accept(actor_system& sys, Socket fd, acceptor_resource_t<Ts...> out,
auto conn = std::make_shared<conn_t>(std::move(on_request), std::move(buf));
auto factory = std::make_unique<factory_t>(std::move(conn));
auto impl = impl_t::make(fd, std::move(factory), max_connections);
auto impl_ptr = impl.get();
auto ptr = socket_manager::make(&mpx, std::move(impl));
impl_ptr->self_ref(ptr->as_disposable());
mpx.start(ptr);
return disposable{std::move(ptr)};
} else {
......
......@@ -63,6 +63,10 @@ public:
return in_ || out_;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
}
// -- implementation of web_socket::lower_layer ------------------------------
error start(web_socket::lower_layer* down, const settings& cfg) override {
......@@ -124,12 +128,14 @@ public:
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::connection_closed || reason == sec::socket_disconnected
|| reason == sec::disposed)
|| reason == sec::disposed) {
out_.close();
else
} else {
out_.abort(reason);
}
}
in_.cancel();
self_ref_ = nullptr;
}
ptrdiff_t consume_binary(byte_span buf) override {
......@@ -171,6 +177,11 @@ private:
/// Initializes the bridge. Disposed (set to null) after initializing.
connector_pointer conn_;
/// Type-erased handle to the @ref socket_manager. This reference is important
/// to keep the bridge alive while the manager is not registered for writing
/// or reading.
disposable self_ref_;
};
} // namespace caf::net::web_socket
......@@ -24,8 +24,17 @@ socket_manager::socket_manager(multiplexer* mpx, event_handler_ptr handler)
}
socket_manager::~socket_manager() {
if (fd_)
// Note: may not call cleanup since it calls the multiplexer via deregister().
handler_.reset();
if (fd_) {
close(fd_);
fd_ = invalid_socket;
}
if (!cleanup_listeners_.empty()) {
for (auto& f : cleanup_listeners_)
mpx_->schedule(std::move(f));
cleanup_listeners_.clear();
}
}
// -- factories ----------------------------------------------------------------
......@@ -113,11 +122,11 @@ error socket_manager::start(const settings& cfg) {
CAF_LOG_TRACE(CAF_ARG(cfg));
if (auto err = nonblocking(fd_, true)) {
CAF_LOG_ERROR("failed to set nonblocking flag in socket:" << err);
close(fd_);
cleanup();
return err;
} else if (err = handler_->start(this, cfg)) {
} else if (err = handler_->start(this, cfg); err) {
CAF_LOG_DEBUG("failed to initialize handler:" << err);
close(fd_);
cleanup();
return err;
} else {
return none;
......
......@@ -77,7 +77,8 @@ void stream_transport::configure_read(receive_policy rd) {
min_read_size_ = rd.min_size;
max_read_size_ = rd.max_size;
if (restarting && !parent_->is_reading()) {
if (buffered_ > 0 && buffered_ >= min_read_size_) {
if (buffered_ > 0 && buffered_ >= min_read_size_
&& delta_offset_ < buffered_) {
// We can already make progress with the data we have. Hence, we need
// schedule a call to read from our buffer before we can wait for
// additional data from the peer.
......@@ -221,7 +222,8 @@ void stream_transport::handle_buffered_data() {
CAF_LOG_TRACE(CAF_ARG(buffered_));
// Loop until we have drained the buffer as much as we can.
CAF_ASSERT(min_read_size_ <= max_read_size_);
while (max_read_size_ > 0 && buffered_ >= min_read_size_) {
while (parent_->is_reading() && max_read_size_ > 0
&& buffered_ >= min_read_size_) {
auto n = std::min(buffered_, size_t{max_read_size_});
auto bytes = make_span(read_buf_.data(), n);
auto delta = bytes.subspan(delta_offset_);
......@@ -238,6 +240,16 @@ void stream_transport::handle_buffered_data() {
up_->abort(make_error(sec::logic_error, "consumed > buffer.size"));
parent_->deregister();
return;
} else if (consumed == 0) {
// See whether the next iteration would change what we pass to the
// application (max_read_size_ may have changed). Otherwise, we'll try
// again later.
delta_offset_ = static_cast<ptrdiff_t>(n);
if (n == std::min(buffered_, size_t{max_read_size_})) {
return;
} else {
// "Fall through".
}
} else {
// Shove the unread bytes to the beginning of the buffer and continue
// to the next loop iteration.
......@@ -252,8 +264,6 @@ void stream_transport::handle_buffered_data() {
}
}
}
if (max_read_size_ == 0)
parent_->deregister_reading();
}
void stream_transport::fail(const error& reason) {
......
......@@ -87,108 +87,94 @@ bool framing::end_text_message() {
// -- interface for the lower layer --------------------------------------------
ptrdiff_t framing::consume(byte_span input, byte_span) {
auto buffer = input;
ptrdiff_t consumed = 0;
// Parse all frames in the current input.
for (;;) {
// Parse header.
detail::rfc6455::header hdr;
auto hdr_bytes = detail::rfc6455::decode_header(buffer, hdr);
if (hdr_bytes < 0) {
CAF_LOG_DEBUG("decoded malformed data: hdr_bytes < 0");
ptrdiff_t framing::consume(byte_span buffer, byte_span) {
// Make sure we're overriding any 'exactly' setting.
down_->configure_read(receive_policy::up_to(2048));
// Parse header.
detail::rfc6455::header hdr;
auto hdr_bytes = detail::rfc6455::decode_header(buffer, hdr);
if (hdr_bytes < 0) {
CAF_LOG_DEBUG("decoded malformed data: hdr_bytes < 0");
up_->abort(make_error(sec::protocol_error,
"negative header size on WebSocket connection"));
return -1;
}
if (hdr_bytes == 0) {
// Wait for more input.
return 0;
}
// Make sure the entire frame (including header) fits into max_frame_size.
if (hdr.payload_len >= (max_frame_size - static_cast<size_t>(hdr_bytes))) {
CAF_LOG_DEBUG("WebSocket frame too large");
up_->abort(make_error(sec::protocol_error, "WebSocket frame too large"));
return -1;
}
// Wait for more data if necessary.
size_t frame_size = hdr_bytes + hdr.payload_len;
if (buffer.size() < frame_size) {
down_->configure_read(receive_policy::exactly(frame_size));
return 0;
}
// Decode frame.
auto payload_len = static_cast<size_t>(hdr.payload_len);
auto payload = buffer.subspan(hdr_bytes, payload_len);
if (hdr.mask_key != 0) {
detail::rfc6455::mask_data(hdr.mask_key, payload);
}
if (hdr.fin) {
if (opcode_ == nil_code) {
// Call upper layer.
if (hdr.opcode == detail::rfc6455::connection_close) {
up_->abort(make_error(sec::connection_closed));
return -1;
} else if (!handle(hdr.opcode, payload)) {
return -1;
}
} else if (hdr.opcode != detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("expected a WebSocket continuation_frame");
up_->abort(make_error(sec::protocol_error,
"negative header size on WebSocket connection"));
"expected a WebSocket continuation_frame"));
return -1;
}
if (hdr_bytes == 0) {
// Wait for more input.
down_->configure_read(receive_policy::up_to(2048));
return consumed;
}
// Make sure the entire frame (including header) fits into max_frame_size.
if (hdr.payload_len >= (max_frame_size - static_cast<size_t>(hdr_bytes))) {
CAF_LOG_DEBUG("WebSocket frame too large");
up_->abort(make_error(sec::protocol_error, "WebSocket frame too large"));
} else if (payload_buf_.size() + payload_len > max_frame_size) {
CAF_LOG_DEBUG("fragmented WebSocket payload exceeds maximum size");
up_->abort(make_error(sec::protocol_error, "fragmented WebSocket payload "
"exceeds maximum size"));
return -1;
}
// Wait for more data if necessary.
size_t frame_size = hdr_bytes + hdr.payload_len;
if (buffer.size() < frame_size) {
down_->configure_read(receive_policy::exactly(frame_size));
return consumed;
}
// Decode frame.
auto payload_len = static_cast<size_t>(hdr.payload_len);
auto payload = buffer.subspan(hdr_bytes, payload_len);
if (hdr.mask_key != 0) {
detail::rfc6455::mask_data(hdr.mask_key, payload);
}
if (hdr.fin) {
if (opcode_ == nil_code) {
// Call upper layer.
if (hdr.opcode == detail::rfc6455::connection_close) {
up_->abort(make_error(sec::connection_closed));
return -1;
} else if (!handle(hdr.opcode, payload)) {
return -1;
}
} else if (hdr.opcode != detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("expected a WebSocket continuation_frame");
up_->abort(make_error(sec::protocol_error,
"expected a WebSocket continuation_frame"));
return -1;
} else if (payload_buf_.size() + payload_len > max_frame_size) {
CAF_LOG_DEBUG("fragmented WebSocket payload exceeds maximum size");
up_->abort(make_error(sec::protocol_error,
"fragmented WebSocket payload "
"exceeds maximum size"));
} else {
if (hdr.opcode == detail::rfc6455::connection_close) {
up_->abort(make_error(sec::connection_closed));
return -1;
} else {
if (hdr.opcode == detail::rfc6455::connection_close) {
up_->abort(make_error(sec::connection_closed));
// End of fragmented input.
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
if (!handle(hdr.opcode, payload_buf_)) {
return -1;
} else {
// End of fragmented input.
payload_buf_.insert(payload_buf_.end(), payload.begin(),
payload.end());
if (!handle(hdr.opcode, payload_buf_))
return -1;
opcode_ = nil_code;
payload_buf_.clear();
}
opcode_ = nil_code;
payload_buf_.clear();
}
} else {
if (opcode_ == nil_code) {
if (hdr.opcode == detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("received WebSocket continuation "
"frame without prior opcode");
up_->abort(make_error(sec::protocol_error,
"received WebSocket continuation "
"frame without prior opcode"));
return -1;
}
opcode_ = hdr.opcode;
} else if (payload_buf_.size() + payload_len > max_frame_size) {
// Reject assembled payloads that exceed max_frame_size.
CAF_LOG_DEBUG("fragmented WebSocket payload exceeds maximum size");
}
} else {
if (opcode_ == nil_code) {
if (hdr.opcode == detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("received WebSocket continuation "
"frame without prior opcode");
up_->abort(make_error(sec::protocol_error,
"fragmented WebSocket payload "
"exceeds maximum size"));
"received WebSocket continuation "
"frame without prior opcode"));
return -1;
}
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
}
// Advance to next frame in the input.
buffer = buffer.subspan(frame_size);
if (buffer.empty()) {
// Ask for more data unless the upper layer called suspend_reading.
if (down_->is_reading())
down_->configure_read(receive_policy::up_to(2048));
return consumed + static_cast<ptrdiff_t>(frame_size);
opcode_ = hdr.opcode;
} else if (payload_buf_.size() + payload_len > max_frame_size) {
// Reject assembled payloads that exceed max_frame_size.
CAF_LOG_DEBUG("fragmented WebSocket payload exceeds maximum size");
up_->abort(make_error(sec::protocol_error, "fragmented WebSocket payload "
"exceeds maximum size"));
return -1;
}
consumed += static_cast<ptrdiff_t>(frame_size);
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
}
return static_cast<ptrdiff_t>(frame_size);
}
// -- implementation details ---------------------------------------------------
......@@ -211,7 +197,7 @@ bool framing::handle(uint8_t opcode, byte_span payload) {
// nop
return true;
default:
// nop
// error
return false;
}
}
......
......@@ -58,17 +58,8 @@ ptrdiff_t server::consume(byte_span input, byte_span delta) {
}
} else if (!handle_header(hdr)) {
return -1;
} else if (remainder.empty()) {
CAF_ASSERT(hdr.size() == input.size());
return hdr.size();
} else {
CAF_LOG_DEBUG(CAF_ARG2("remainder.size", remainder.size()));
if (auto sub_result = framing_.consume(remainder, remainder);
sub_result >= 0) {
return hdr.size() + sub_result;
} else {
return sub_result;
}
return hdr.size();
}
}
}
......@@ -126,6 +117,7 @@ bool server::handle_header(std::string_view http) {
put(fields, std::string{key}, std::string{val});
}
// Try to initialize the upper layer.
lower_layer().configure_read(receive_policy::stop());
if (auto err = upper_layer().start(&framing_, cfg_)) {
auto descr = to_string(err);
CAF_LOG_DEBUG("upper layer rejected a WebSocket connection:" << descr);
......
......@@ -50,28 +50,39 @@ bool mock_stream_transport::end_output() {
ptrdiff_t mock_stream_transport::handle_input() {
ptrdiff_t result = 0;
while (max_read_size > 0) {
CAF_ASSERT(max_read_size > static_cast<size_t>(read_size_));
size_t len = max_read_size - static_cast<size_t>(read_size_);
CAF_LOG_DEBUG(CAF_ARG2("available capacity:", len));
auto num_bytes = std::min(input.size(), len);
if (num_bytes == 0)
// Loop until we have drained the buffer as much as we can.
while (max_read_size > 0 && input.size() >= min_read_size) {
auto n = std::min(input.size(), size_t{max_read_size});
auto bytes = make_span(input.data(), n);
auto delta = bytes.subspan(delta_offset);
auto consumed = up->consume(bytes, delta);
if (consumed < 0) {
// Negative values indicate that the application encountered an
// unrecoverable error.
up->abort(make_error(caf::sec::runtime_error, "consumed < 0"));
return result;
auto delta_offset = static_cast<ptrdiff_t>(read_buf_.size());
read_buf_.insert(read_buf_.end(), input.begin(), input.begin() + num_bytes);
input.erase(input.begin(), input.begin() + num_bytes);
read_size_ += static_cast<ptrdiff_t>(num_bytes);
if (static_cast<size_t>(read_size_) < min_read_size)
} else if (static_cast<size_t>(consumed) > n) {
// Must not happen. An application cannot handle more data then we pass
// to it.
up->abort(make_error(sec::logic_error, "consumed > buffer.size"));
return result;
auto delta = make_span(read_buf_.data() + delta_offset,
read_size_ - delta_offset);
auto consumed = up->consume(make_span(read_buf_), delta);
if (consumed > 0) {
result += static_cast<ptrdiff_t>(consumed);
read_buf_.erase(read_buf_.begin(), read_buf_.begin() + consumed);
read_size_ -= consumed;
} else if (consumed < 0) {
return -1;
} else if (consumed == 0) {
// See whether the next iteration would change what we pass to the
// application (max_read_size_ may have changed). Otherwise, we'll try
// again later.
delta_offset = static_cast<ptrdiff_t>(n);
if (n == std::min(input.size(), size_t{max_read_size})) {
return result;
} else {
// "Fall through".
}
} else {
// Shove the unread bytes to the beginning of the buffer and continue
// to the next loop iteration.
result += consumed;
auto del = static_cast<size_t>(consumed);
delta_offset = static_cast<ptrdiff_t>(n - del);
input.erase(input.begin(), input.begin() + del);
}
}
return result;
......
......@@ -60,7 +60,7 @@ public:
// -- buffer management ------------------------------------------------------
void push(caf::span<const std::byte> bytes) {
input.insert(input.begin(), bytes.begin(), bytes.end());
input.insert(input.end(), bytes.begin(), bytes.end());
}
void push(std::string_view str) {
......@@ -91,11 +91,11 @@ public:
uint32_t max_read_size = 0;
size_t delta_offset = 0;
private:
caf::byte_buffer read_buf_;
ptrdiff_t read_size_ = 0;
caf::error abort_reason_;
};
......
......@@ -33,7 +33,9 @@ public:
return std::make_unique<app_t>();
}
error start(net::web_socket::lower_layer*, const settings& args) override {
error start(net::web_socket::lower_layer* down,
const settings& args) override {
down->request_messages();
cfg = args;
return none;
}
......
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