Commit 617955aa authored by Dominik Charousset's avatar Dominik Charousset

Fix shutdown of binary and WebSocket flow bridges

parent 8fa1b449
...@@ -140,6 +140,11 @@ public: ...@@ -140,6 +140,11 @@ public:
impl_->cancel(); impl_->cancel();
} }
consumer_adapter& operator=(std::nullptr_t) {
impl_ = nullptr;
return *this;
}
template <class Policy> template <class Policy>
read_result pull(Policy policy, T& result) { read_result pull(Policy policy, T& result) {
if (impl_) if (impl_)
......
...@@ -107,6 +107,11 @@ public: ...@@ -107,6 +107,11 @@ public:
// nop // nop
} }
producer_adapter& operator=(std::nullptr_t) {
impl_ = nullptr;
return *this;
}
~producer_adapter() { ~producer_adapter() {
if (impl_) if (impl_)
impl_->close(); impl_->close();
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp"
#include "caf/fwd.hpp"
#include "caf/net/flow_connector.hpp"
#include "caf/sec.hpp"
#include <utility>
namespace caf::detail {
/// Translates between a message-oriented transport and data flows.
template <class UpperLayer, class LowerLayer, class Trait>
class flow_bridge_base : public UpperLayer {
public:
using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = async::consumer_adapter<output_type>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using connector_pointer = net::flow_connector_ptr<Trait>;
flow_bridge_base(async::execution_context_ptr loop, connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
virtual bool write(const output_type& item) = 0;
bool running() const noexcept {
return in_ || out_;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
}
// -- implementation of the lower_layer --------------------------------------
error start(LowerLayer* down, const settings& cfg) override {
CAF_ASSERT(down != nullptr);
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
auto do_wakeup = make_action([this] {
if (running())
prepare_send();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running()) {
down_->shutdown();
}
});
in_ = consumer_type::make(pull.try_open(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (running())
return none;
else
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
} else {
conn_ = nullptr;
return err;
}
}
void prepare_send() override {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
abort(trait_.last_error());
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
in_ = nullptr;
abort(error{});
down_->shutdown();
return;
case async::read_result::abort:
in_ = nullptr;
abort(in_.abort_reason());
down_->shutdown(in_.abort_reason());
return;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_.has_consumer_event();
}
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (!reason || reason == sec::connection_closed
|| reason == sec::socket_disconnected || reason == sec::disposed)
out_.close();
else
out_.abort(reason);
out_ = nullptr;
}
if (in_) {
in_.cancel();
in_ = nullptr;
}
self_ref_ = nullptr;
}
protected:
LowerLayer* down_;
/// The output of the application. Serialized to the socket.
consumer_type in_;
/// The input to the application. Deserialized from the socket.
producer_type out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Our event loop.
async::execution_context_ptr loop_;
/// 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::detail
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
#include "caf/async/consumer_adapter.hpp" #include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp" #include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp" #include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_bridge_base.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/binary/lower_layer.hpp" #include "caf/net/binary/lower_layer.hpp"
#include "caf/net/binary/upper_layer.hpp" #include "caf/net/binary/upper_layer.hpp"
...@@ -17,153 +18,48 @@ ...@@ -17,153 +18,48 @@
namespace caf::net::binary { namespace caf::net::binary {
/// Convenience alias for referring to the base type of @ref flow_bridge.
template <class Trait>
using flow_bridge_base_t
= detail::flow_bridge_base<upper_layer, lower_layer, Trait>;
/// Translates between a message-oriented transport and data flows. /// Translates between a message-oriented transport and data flows.
template <class Trait> template <class Trait>
class flow_bridge : public upper_layer { class flow_bridge : public flow_bridge_base_t<Trait> {
public: public:
using super = flow_bridge_base_t<Trait>;
using input_type = typename Trait::input_type; using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type; using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = async::consumer_adapter<output_type>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using connector_pointer = flow_connector_ptr<Trait>; using connector_pointer = flow_connector_ptr<Trait>;
explicit flow_bridge(async::execution_context_ptr loop, using super::super;
connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop, static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) { connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn)); return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
} }
bool write(const output_type& item) { bool write(const output_type& item) override {
down_->begin_message(); super::down_->begin_message();
auto& bytes = down_->message_buffer(); auto& bytes = super::down_->message_buffer();
return trait_.convert(item, bytes) && down_->end_message(); return super::trait_.convert(item, bytes) && super::down_->end_message();
}
bool running() const noexcept {
return in_ || out_;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
} }
// -- implementation of binary::lower_layer ---------------------------------- // -- implementation of binary::lower_layer ----------------------------------
error start(binary::lower_layer* down, const settings& cfg) override {
CAF_ASSERT(down != nullptr);
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
auto do_wakeup = make_action([this] {
prepare_send();
if (!running())
down_->shutdown();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running()) {
down_->shutdown();
}
});
in_ = consumer_type::make(pull.try_open(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (in_ && out_) {
return none;
} else {
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
}
} else {
conn_ = nullptr;
return err;
}
}
void prepare_send() override {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
down_->shutdown();
break;
case async::read_result::abort:
down_->shutdown(in_.abort_reason());
break;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_.has_consumer_event();
}
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::connection_closed || reason == sec::socket_disconnected
|| reason == sec::disposed)
out_.close();
else
out_.abort(reason);
}
in_.cancel();
self_ref_ = nullptr;
}
ptrdiff_t consume(byte_span buf) override { ptrdiff_t consume(byte_span buf) override {
if (!out_) if (!super::out_)
return -1; return -1;
input_type val; input_type val;
if (!trait_.convert(buf, val)) if (!super::trait_.convert(buf, val))
return -1; return -1;
if (out_.push(std::move(val)) == 0) if (super::out_.push(std::move(val)) == 0)
down_->suspend_reading(); super::down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size()); return static_cast<ptrdiff_t>(buf.size());
} }
private:
lower_layer* down_;
/// The output of the application. Serialized to the socket.
consumer_type in_;
/// The input to the application. Deserialized from the socket.
producer_type out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Our event loop.
async::execution_context_ptr loop_;
/// 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::binary } // namespace caf::net::binary
...@@ -4,10 +4,6 @@ ...@@ -4,10 +4,6 @@
#pragma once #pragma once
#include <memory>
#include <mutex>
#include <thread>
#include "caf/action.hpp" #include "caf/action.hpp"
#include "caf/async/execution_context.hpp" #include "caf/async/execution_context.hpp"
#include "caf/detail/atomic_ref_counted.hpp" #include "caf/detail/atomic_ref_counted.hpp"
...@@ -19,6 +15,11 @@ ...@@ -19,6 +15,11 @@
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/unordered_flat_map.hpp" #include "caf/unordered_flat_map.hpp"
#include <deque>
#include <memory>
#include <mutex>
#include <thread>
extern "C" { extern "C" {
struct pollfd; struct pollfd;
...@@ -150,6 +151,9 @@ public: ...@@ -150,6 +151,9 @@ public:
/// Applies all pending updates. /// Applies all pending updates.
void apply_updates(); void apply_updates();
/// Runs all pending actions.
void run_actions();
/// Sets the thread ID to `std::this_thread::id()`. /// Sets the thread ID to `std::this_thread::id()`.
void set_thread_id(); void set_thread_id();
...@@ -213,6 +217,9 @@ protected: ...@@ -213,6 +217,9 @@ protected:
/// Signals whether shutdown has been requested. /// Signals whether shutdown has been requested.
bool shutting_down_ = false; bool shutting_down_ = false;
/// Pending actions via `schedule`.
std::deque<action> pending_actions_;
/// Keeps track of watched disposables. /// Keeps track of watched disposables.
std::vector<disposable> watched_; std::vector<disposable> watched_;
......
...@@ -9,6 +9,7 @@ ...@@ -9,6 +9,7 @@
#include "caf/detail/accept_handler.hpp" #include "caf/detail/accept_handler.hpp"
#include "caf/detail/connection_factory.hpp" #include "caf/detail/connection_factory.hpp"
#include "caf/net/flow_connector.hpp" #include "caf/net/flow_connector.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/ssl/acceptor.hpp" #include "caf/net/ssl/acceptor.hpp"
#include "caf/net/ssl/transport.hpp" #include "caf/net/ssl/transport.hpp"
#include "caf/net/web_socket/default_trait.hpp" #include "caf/net/web_socket/default_trait.hpp"
......
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
#include "caf/async/consumer_adapter.hpp" #include "caf/async/consumer_adapter.hpp"
#include "caf/async/producer_adapter.hpp" #include "caf/async/producer_adapter.hpp"
#include "caf/async/spsc_buffer.hpp" #include "caf/async/spsc_buffer.hpp"
#include "caf/detail/flow_bridge_base.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/flow_connector.hpp" #include "caf/net/flow_connector.hpp"
#include "caf/net/web_socket/lower_layer.hpp" #include "caf/net/web_socket/lower_layer.hpp"
...@@ -18,170 +19,67 @@ ...@@ -18,170 +19,67 @@
namespace caf::net::web_socket { namespace caf::net::web_socket {
/// Convenience alias for referring to the base type of @ref flow_bridge.
template <class Trait>
using flow_bridge_base_t
= detail::flow_bridge_base<upper_layer, lower_layer, Trait>;
/// Translates between a message-oriented transport and data flows. /// Translates between a message-oriented transport and data flows.
template <class Trait> template <class Trait>
class flow_bridge : public web_socket::upper_layer { class flow_bridge : public flow_bridge_base_t<Trait> {
public: public:
using super = flow_bridge_base_t<Trait>;
using input_type = typename Trait::input_type; using input_type = typename Trait::input_type;
using output_type = typename Trait::output_type; using output_type = typename Trait::output_type;
/// Type for the consumer adapter. We consume the output of the application.
using consumer_type = async::consumer_adapter<output_type>;
/// Type for the producer adapter. We produce the input of the application.
using producer_type = async::producer_adapter<input_type>;
using request_type = request<Trait>;
using connector_pointer = flow_connector_ptr<Trait>; using connector_pointer = flow_connector_ptr<Trait>;
explicit flow_bridge(async::execution_context_ptr loop, using super::super;
connector_pointer conn)
: loop_(std::move(loop)), conn_(std::move(conn)) {
// nop
}
static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop, static std::unique_ptr<flow_bridge> make(async::execution_context_ptr loop,
connector_pointer conn) { connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(loop), std::move(conn)); return std::make_unique<flow_bridge>(std::move(loop), std::move(conn));
} }
bool write(const output_type& item) { bool write(const output_type& item) override {
if (trait_.converts_to_binary(item)) { if (super::trait_.converts_to_binary(item)) {
down_->begin_binary_message(); super::down_->begin_binary_message();
auto& bytes = down_->binary_message_buffer(); auto& bytes = super::down_->binary_message_buffer();
return trait_.convert(item, bytes) && down_->end_binary_message(); return super::trait_.convert(item, bytes)
&& super::down_->end_binary_message();
} else { } else {
down_->begin_text_message(); super::down_->begin_text_message();
auto& text = down_->text_message_buffer(); auto& text = super::down_->text_message_buffer();
return trait_.convert(item, text) && down_->end_text_message(); return super::trait_.convert(item, text)
&& super::down_->end_text_message();
} }
} }
bool running() const noexcept {
return in_ || out_;
}
void self_ref(disposable ref) {
self_ref_ = std::move(ref);
}
// -- implementation of web_socket::lower_layer ------------------------------ // -- implementation of web_socket::lower_layer ------------------------------
error start(web_socket::lower_layer* down, const settings& cfg) override {
down_ = down;
auto [err, pull, push] = conn_->on_request(cfg);
if (!err) {
auto do_wakeup = make_action([this] {
prepare_send();
if (!running())
down_->shutdown();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
if (!running())
down_->shutdown();
});
in_ = consumer_type::make(pull.try_open(), loop_, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), loop_, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr;
if (running())
return none;
else
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
} else {
conn_ = nullptr;
return err;
}
}
void prepare_send() override {
input_type tmp;
while (down_->can_send_more()) {
switch (in_.pull(async::delay_errors, tmp)) {
case async::read_result::ok:
if (!write(tmp)) {
down_->shutdown(trait_.last_error());
return;
}
break;
case async::read_result::stop:
down_->shutdown();
break;
case async::read_result::abort:
down_->shutdown(in_.abort_reason());
break;
default: // try later
return;
}
}
}
bool done_sending() override {
return !in_.has_consumer_event();
}
void abort(const error& reason) override {
CAF_LOG_TRACE(CAF_ARG(reason));
if (out_) {
if (reason == sec::connection_closed || reason == sec::socket_disconnected
|| reason == sec::disposed) {
out_.close();
} else {
out_.abort(reason);
}
}
in_.cancel();
self_ref_ = nullptr;
}
ptrdiff_t consume_binary(byte_span buf) override { ptrdiff_t consume_binary(byte_span buf) override {
if (!out_) if (!super::out_)
return -1; return -1;
input_type val; input_type val;
if (!trait_.convert(buf, val)) if (!super::trait_.convert(buf, val))
return -1; return -1;
if (out_.push(std::move(val)) == 0) if (super::out_.push(std::move(val)) == 0)
down_->suspend_reading(); super::down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size()); return static_cast<ptrdiff_t>(buf.size());
} }
ptrdiff_t consume_text(std::string_view buf) override { ptrdiff_t consume_text(std::string_view buf) override {
if (!out_) if (!super::out_)
return -1; return -1;
input_type val; input_type val;
if (!trait_.convert(buf, val)) if (!super::trait_.convert(buf, val))
return -1; return -1;
if (out_.push(std::move(val)) == 0) if (super::out_.push(std::move(val)) == 0)
down_->suspend_reading(); super::down_->suspend_reading();
return static_cast<ptrdiff_t>(buf.size()); return static_cast<ptrdiff_t>(buf.size());
} }
private:
web_socket::lower_layer* down_;
/// The output of the application. Serialized to the socket.
consumer_type in_;
/// The input to the application. Deserialized from the socket.
producer_type out_;
/// Converts between raw bytes and native C++ objects.
Trait trait_;
/// Runs callbacks in the I/O event loop.
async::execution_context_ptr loop_;
/// 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 } // namespace caf::net::web_socket
...@@ -171,8 +171,12 @@ void multiplexer::deref_execution_context() const noexcept { ...@@ -171,8 +171,12 @@ void multiplexer::deref_execution_context() const noexcept {
void multiplexer::schedule(action what) { void multiplexer::schedule(action what) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (std::this_thread::get_id() == tid_) {
pending_actions_.push_back(what);
} else {
auto ptr = std::move(what).as_intrusive_ptr().release(); auto ptr = std::move(what).as_intrusive_ptr().release();
write_to_pipe(pollset_updater::code::run_action, ptr); write_to_pipe(pollset_updater::code::run_action, ptr);
}
} }
void multiplexer::watch(disposable what) { void multiplexer::watch(disposable what) {
...@@ -309,6 +313,7 @@ void multiplexer::poll() { ...@@ -309,6 +313,7 @@ void multiplexer::poll() {
void multiplexer::apply_updates() { void multiplexer::apply_updates() {
CAF_LOG_DEBUG("apply" << updates_.size() << "updates"); CAF_LOG_DEBUG("apply" << updates_.size() << "updates");
for (;;) {
if (!updates_.empty()) { if (!updates_.empty()) {
for (auto& [fd, update] : updates_) { for (auto& [fd, update] : updates_) {
if (auto index = index_of(fd); index == -1) { if (auto index = index_of(fd); index == -1) {
...@@ -327,6 +332,14 @@ void multiplexer::apply_updates() { ...@@ -327,6 +332,14 @@ void multiplexer::apply_updates() {
} }
updates_.clear(); updates_.clear();
} }
while (!pending_actions_.empty()) {
auto next = std::move(pending_actions_.front());
pending_actions_.pop_front();
next.run();
}
if (updates_.empty())
return;
}
} }
void multiplexer::set_thread_id() { void multiplexer::set_thread_id() {
......
...@@ -45,9 +45,9 @@ void pollset_updater::handle_read_event() { ...@@ -45,9 +45,9 @@ void pollset_updater::handle_read_event() {
auto as_mgr = [](intptr_t ptr) { auto as_mgr = [](intptr_t ptr) {
return intrusive_ptr{reinterpret_cast<socket_manager*>(ptr), false}; return intrusive_ptr{reinterpret_cast<socket_manager*>(ptr), false};
}; };
auto run_action = [](intptr_t ptr) { auto add_action = [this](intptr_t ptr) {
auto f = action{intrusive_ptr{reinterpret_cast<action::impl*>(ptr), false}}; auto f = action{intrusive_ptr{reinterpret_cast<action::impl*>(ptr), false}};
f.run(); mpx_->pending_actions_.push_back(std::move(f));
}; };
for (;;) { for (;;) {
CAF_ASSERT((buf_.size() - buf_size_) > 0); CAF_ASSERT((buf_.size() - buf_size_) > 0);
...@@ -65,7 +65,7 @@ void pollset_updater::handle_read_event() { ...@@ -65,7 +65,7 @@ void pollset_updater::handle_read_event() {
mpx_->do_start(as_mgr(ptr)); mpx_->do_start(as_mgr(ptr));
break; break;
case code::run_action: case code::run_action:
run_action(ptr); add_action(ptr);
break; break;
case code::shutdown: case code::shutdown:
CAF_ASSERT(ptr == 0); CAF_ASSERT(ptr == 0);
......
...@@ -215,7 +215,6 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") { ...@@ -215,7 +215,6 @@ SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
app_ptr->continue_reading(); app_ptr->continue_reading();
mpx->apply_updates(); mpx->apply_updates();
mpx->poll_once(true); mpx->poll_once(true);
CHECK_EQ(mpx->mask_of(mgr), net::operation::read);
while (mpx->num_socket_managers() > 1u) while (mpx->num_socket_managers() > 1u)
mpx->poll_once(true); mpx->poll_once(true);
if (CHECK_EQ(buf->size(), 5u)) { if (CHECK_EQ(buf->size(), 5u)) {
......
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