Commit c2e3d0e4 authored by Dominik Charousset's avatar Dominik Charousset

Fix shutdown behavior of socket managers

parent 599aafbb
......@@ -40,17 +40,17 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
return EXIT_FAILURE;
}
// Create buffers to signal events from the WebSocket server to the worker.
auto [worker_res, server_res] = net::http::make_request_resource();
auto [worker_pull, server_push] = net::http::make_request_resource();
// Spin up the HTTP server.
net::http::serve(sys, fd, std::move(server_res));
net::http::serve(sys, fd, std::move(server_push));
// Spin up a worker to handle the HTTP requests.
auto worker = sys.spawn([wres = worker_res](event_based_actor* self) {
auto worker = sys.spawn([wres = worker_pull](event_based_actor* self) {
// For each incoming request ...
wres
.observe_on(self) //
.for_each([](const net::http::request& req) {
// ... we simply return the current time as string.
// Note: we may not respond more than once to a single request.
// Note: we cannot respond more than once to a request.
auto str = caf::deep_to_string(make_timestamp());
req.respond(net::http::status::ok, "text/plain", str);
});
......
......@@ -50,8 +50,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
.for_each([self](const event_t& event) {
// ... that simply pushes data back to the sender.
auto [pull, push] = event.data();
self->make_observable()
.from_resource(pull)
pull.observe_on(self)
.do_on_next([](const ws::frame& x) {
if (x.is_binary()) {
std::cout << "*** received a binary WebSocket frame of size "
......
......@@ -60,8 +60,7 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
sys.spawn([conn, hello](caf::event_based_actor* self) {
auto [pull, push] = conn.data();
// Print everything from the server to stdout.
self->make_observable()
.from_resource(pull)
pull.observe_on(self)
.do_on_error([](const caf::error& what) {
std::cerr << "*** error while reading from the WebSocket: "
<< to_string(what) << '\n';
......
......@@ -128,7 +128,8 @@ int caf_main(caf::actor_system& sys, const config& cfg) {
auto quote = quotes.empty() ? "Try /epictetus, /seneca or /plato."
: f(quotes);
self->make_observable().just(ws::frame{quote}).subscribe(push);
// Note: we simply drop `pull` here, which will close the buffer.
// We ignore whatever the client may send to us.
pull.observe_on(self).subscribe(std::ignore);
});
});
// Callback for incoming WebSocket requests.
......
......@@ -470,6 +470,11 @@ disposable observable<T>::subscribe(async::producer_resource<T> resource) {
}
}
template <class T>
disposable observable<T>::subscribe(ignore_t) {
return subscribe(observer<T>::ignore());
}
template <class T>
template <class OnNext>
disposable observable<T>::for_each(OnNext on_next) {
......
......@@ -15,6 +15,7 @@
#include "caf/intrusive_ptr.hpp"
#include <cstddef>
#include <tuple>
#include <utility>
#include <vector>
......@@ -32,6 +33,9 @@ public:
/// The pointer-to-implementation type.
using pimpl_type = intrusive_ptr<op::base<T>>;
/// Type for drop-all subscribers.
using ignore_t = decltype(std::ignore);
// -- constructors, destructors, and assignment operators --------------------
explicit observable(pimpl_type pimpl) noexcept : pimpl_(std::move(pimpl)) {
......@@ -57,6 +61,9 @@ public:
/// Creates a new observer that pushes all observed items to the resource.
disposable subscribe(async::producer_resource<T> resource);
/// Subscribes a new observer to the items emitted by this observable.
disposable subscribe(ignore_t);
/// Calls `on_next` for each item emitted by this observable.
template <class OnNext>
disposable for_each(OnNext on_next);
......
......@@ -24,12 +24,12 @@ public:
/// Shuts down any connection or session gracefully. Any pending data gets
/// flushed before closing the socket.
virtual void close() = 0;
virtual void shutdown() = 0;
/// Shuts down any connection or session du to an error. Any pending data gets
/// flushed before closing the socket. Protocols with a dedicated closing
/// handshake such as WebSocket may send the close reason to the peer.
virtual void close(const error& reason);
virtual void shutdown(const error& reason);
};
} // namespace caf::net
......@@ -15,16 +15,16 @@ class CAF_NET_EXPORT generic_upper_layer {
public:
virtual ~generic_upper_layer();
/// Prepares any pending data for sending.
/// @returns `false` in case of an error to cause the lower layer to stop,
/// `true` otherwise.
[[nodiscard]] virtual bool prepare_send() = 0;
/// Gives the upper layer an opportunity to add additional data to the output
/// buffer.
virtual void prepare_send() = 0;
/// Queries whether all pending data has been sent. The lower calls this
/// function to decide whether it has to wait for write events on the socket.
[[nodiscard]] virtual bool done_sending() = 0;
/// Called by the lower layer for cleaning up any state in case of an error.
/// Called by the lower layer for cleaning up any state in case of an error or
/// when disposed.
virtual void abort(const error& reason) = 0;
};
......
......@@ -57,7 +57,7 @@ public:
// nop
}
bool prepare_send() override;
void prepare_send() override;
bool done_sending() override;
......@@ -144,7 +144,6 @@ disposable serve(actor_system& sys, Socket fd,
} else {
return disposable{};
}
// TODO: serve() should return a disposable to stop the HTTP server.
}
} // namespace caf::net::http
......@@ -81,7 +81,7 @@ public:
bool is_reading() const noexcept override;
void close() override;
void shutdown() override;
void request_messages() override;
......@@ -106,7 +106,7 @@ public:
void abort(const error& reason) override;
bool prepare_send() override;
void prepare_send() override;
bool done_sending() override;
......
......@@ -81,7 +81,7 @@ public:
ptrdiff_t consume(byte_span buffer, byte_span delta) override;
bool prepare_send() override;
void prepare_send() override;
bool done_sending() override;
......@@ -101,7 +101,7 @@ public:
bool end_message() override;
void close() override;
void shutdown() override;
// -- utility functions ------------------------------------------------------
......
......@@ -110,26 +110,23 @@ public:
}
};
bool prepare_send() override {
void prepare_send() override {
write_helper helper{this};
while (down_->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down_->close(helper.err);
down_->shutdown(helper.err);
} else {
down_->close();
down_->shutdown();
}
in_ = nullptr;
} else if (helper.aborted) {
in_->cancel();
in_ = nullptr;
return false;
} else if (consumed == 0) {
return true;
down_->shutdown(trait_.last_error());
}
}
return true;
}
bool done_sending() override {
......
......@@ -107,18 +107,6 @@ public:
// -- thread-safe signaling --------------------------------------------------
/// Schedules a call to `mgr->handle_error(sec::disposed)`.
/// @thread-safe
void dispose(socket_manager_ptr mgr);
/// Stops further reading by `mgr`.
/// @thread-safe
void shutdown_reading(socket_manager_ptr mgr);
/// Stops further writing by `mgr`.
/// @thread-safe
void shutdown_writing(socket_manager_ptr mgr);
/// Registers `mgr` for initialization in the multiplexer's thread.
/// @thread-safe
void init(socket_manager_ptr mgr);
......@@ -239,12 +227,6 @@ private:
void do_register_reading(const socket_manager_ptr& mgr);
void do_dispose(const socket_manager_ptr& mgr);
void do_shutdown_reading(const socket_manager_ptr& mgr);
void do_shutdown_writing(const socket_manager_ptr& mgr);
void do_init(const socket_manager_ptr& mgr);
};
......
......@@ -26,7 +26,6 @@ public:
enum class code : uint8_t {
init_manager,
dispose_manager,
shutdown_reading,
shutdown_writing,
run_action,
......
......@@ -55,7 +55,7 @@ public:
// -- implementation of http::upper_layer ------------------------------------
bool prepare_send() override;
void prepare_send() override;
bool done_sending() override;
......
......@@ -37,6 +37,14 @@ struct CAF_NET_EXPORT socket : detail::comparable<socket> {
return static_cast<signed_socket_id>(id)
- static_cast<signed_socket_id>(other.id);
}
constexpr explicit operator bool() const noexcept {
return id != invalid_socket_id;
}
constexpr bool operator!() const noexcept {
return id == invalid_socket_id;
}
};
/// @relates socket
......
......@@ -28,8 +28,12 @@ public:
/// Called after returning `handover` from a read or write handler.
virtual bool do_handover(std::unique_ptr<socket_event_layer>& next);
/// Called on hard errors on the managed socket.
/// Called on socket errors or when the manager gets disposed.
virtual void abort(const error& reason) = 0;
/// Queries whether the object can be safely discarded after calling
/// @ref abort on it, e.g., that pending data has been written.
virtual bool finalized() const noexcept;
};
} // namespace caf::net
......@@ -33,12 +33,6 @@ public:
using event_handler_ptr = std::unique_ptr<socket_event_layer>;
/// Stores manager-related flags in a single block.
struct flags_t {
bool read_closed : 1;
bool write_closed : 1;
};
// -- constructors, destructors, and assignment operators --------------------
/// @pre `handle != invalid_socket`
......@@ -111,16 +105,6 @@ public:
return mpx_;
}
/// Returns whether the manager closed read operations on the socket.
[[nodiscard]] bool read_closed() const noexcept {
return flags_.read_closed;
}
/// Returns whether the manager closed write operations on the socket.
[[nodiscard]] bool write_closed() const noexcept {
return flags_.write_closed;
}
/// Queries whether the manager is registered for reading.
bool is_reading() const noexcept;
......@@ -144,18 +128,6 @@ public:
/// Deregisters the manager from both read and write operations.
void deregister();
/// Deregisters the manager from read operations and blocks any future
/// attempts to re-register it.
void shutdown_read();
/// Deregisters the manager from write operations and blocks any future
/// attempts to re-register it.
void shutdown_write();
/// Deregisters the manager from both read and write operations and blocks any
/// future attempts to re-register it.
void shutdown();
// -- callbacks for the handler ----------------------------------------------
/// Schedules a call to `do_handover` on the handler.
......@@ -170,13 +142,10 @@ public:
schedule(make_action(std::forward<F>(what)));
}
// -- callbacks for the multiplexer ------------------------------------------
/// Closes the read channel of the socket.
void close_read() noexcept;
/// Shuts down this socket manager.
void shutdown();
/// Closes the write channel of the socket.
void close_write() noexcept;
// -- callbacks for the multiplexer ------------------------------------------
/// Initializes the manager and its all of its sub-components.
error init(const settings& cfg);
......@@ -187,8 +156,10 @@ public:
/// Called whenever the socket is allowed to send data.
void handle_write_event();
/// Called when the remote side becomes unreachable due to an error.
/// @param code The error code as reported by the operating system.
/// Called when the remote side becomes unreachable due to an error or after
/// calling @ref dispose.
/// @param code The error code as reported by the operating system or
/// @ref sec::disposed.
void handle_error(sec code);
// -- implementation of disposable_impl --------------------------------------
......@@ -204,6 +175,8 @@ public:
private:
// -- utility functions ------------------------------------------------------
void cleanup();
socket_manager_ptr strong_this();
// -- member variables -------------------------------------------------------
......@@ -219,12 +192,14 @@ private:
/// down (and the multiplexer is part of the actor system).
multiplexer* mpx_;
/// Stores flags for the socket file descriptor.
flags_t flags_;
/// Stores the event handler that operators on the socket file descriptor.
event_handler_ptr handler_;
/// Stores whether `shutdown` has been called.
bool shutting_down_ = false;
/// Stores whether the manager has been either explicitly disposed or shut
/// down by demand of the application.
std::atomic<bool> disposed_;
};
......
......@@ -86,7 +86,7 @@ public:
/// Stores whether we left a write handler due to want_read.
bool wanted_write_from_read_event : 1;
/// Stores whether we shutdown write on the socket after flushing.
/// Stores whether the application has asked to shut down.
bool shutting_down : 1;
};
......@@ -123,7 +123,7 @@ public:
bool is_reading() const noexcept override;
void close() override;
void shutdown() override;
// -- properties -------------------------------------------------------------
......@@ -161,6 +161,8 @@ public:
void abort(const error& reason) override;
bool finalized() const noexcept override;
protected:
// -- utility functions ------------------------------------------------------
......
......@@ -78,7 +78,7 @@ public:
ptrdiff_t consume(byte_span buffer, byte_span delta) override;
bool prepare_send() override;
void prepare_send() override;
bool done_sending() override;
......
......@@ -32,6 +32,8 @@ public:
bool convert(const_byte_span bytes, input_type& x);
bool convert(std::string_view text, input_type& x);
error last_error();
};
} // namespace caf::net::web_socket
......@@ -40,6 +40,13 @@ public:
// nop
}
~flow_bridge() {
if (in_)
in_->cancel();
if (out_)
out_->close();
}
static std::unique_ptr<flow_bridge> make(connector_pointer conn) {
return std::make_unique<flow_bridge>(std::move(conn));
}
......@@ -94,13 +101,13 @@ public:
auto do_wakeup = make_action([this] {
prepare_send();
if (!running())
down_->close();
down_->shutdown();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
out_ = nullptr;
if (!running())
down_->close();
down_->shutdown();
});
in_ = consumer_type::make(pull.try_open(), mgr, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), mgr, std::move(do_resume),
......@@ -117,27 +124,25 @@ public:
}
}
bool prepare_send() override {
void prepare_send() override {
write_helper helper{this};
while (down_->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) {
if (helper.err) {
down_->close(helper.err);
down_->shutdown(helper.err);
} else {
down_->close();
down_->shutdown();
}
in_ = nullptr;
return true;
} else if (helper.aborted) {
in_->cancel();
in_ = nullptr;
return true;
down_->shutdown(trait_.last_error());
} else if (consumed == 0) {
return true;
return;
}
}
return true;
}
bool done_sending() override {
......
......@@ -77,7 +77,7 @@ public:
// -- web_socket::lower_layer implementation ---------------------------------
using web_socket::lower_layer::close;
using web_socket::lower_layer::shutdown;
bool can_send_more() const noexcept override;
......@@ -85,7 +85,7 @@ public:
bool is_reading() const noexcept override;
void close(status code, std::string_view desc) override;
void shutdown(status code, std::string_view desc) override;
void request_messages() override;
......
......@@ -45,12 +45,12 @@ public:
/// Seals the current text message buffer and ships a new WebSocket frame.
virtual bool end_text_message() = 0;
void close() override;
void shutdown() override;
void close(const error& reason) override;
void shutdown(const error& reason) override;
/// Sends a close message with custom @ref status @p code and @p msg text.
virtual void close(status code, std::string_view msg) = 0;
/// Sends a shutdown message with custom @ref status @p code and @p msg text.
virtual void shutdown(status code, std::string_view msg) = 0;
};
} // namespace caf::net::web_socket
......@@ -81,7 +81,7 @@ public:
ptrdiff_t consume(byte_span input, byte_span) override;
bool prepare_send() override;
void prepare_send() override;
bool done_sending() override;
......
......@@ -10,8 +10,8 @@ generic_lower_layer::~generic_lower_layer() {
// nop
}
void generic_lower_layer::close(const error&) {
close();
void generic_lower_layer::shutdown(const error&) {
shutdown();
}
} // namespace caf::net
......@@ -35,8 +35,8 @@ bool http_request_producer::push(const net::http::request& item) {
// -- http_flow_adapter --------------------------------------------------------
bool http_flow_adapter::prepare_send() {
return true;
void http_flow_adapter::prepare_send() {
// nop
}
bool http_flow_adapter::done_sending() {
......@@ -59,6 +59,11 @@ error http_flow_adapter::init(net::socket_manager* owner,
ptrdiff_t http_flow_adapter::consume(const net::http::header& hdr,
const_byte_span payload) {
using namespace net::http;
if (!pending_.empty()) {
CAF_LOG_WARNING("received multiple requests from the same HTTP client: "
"not implemented yet (drop request)");
return static_cast<ptrdiff_t>(payload.size());
}
auto prom = async::promise<response>();
auto fut = prom.get_future();
auto buf = std::vector<std::byte>{payload.begin(), payload.end()};
......@@ -71,14 +76,13 @@ ptrdiff_t http_flow_adapter::consume(const net::http::header& hdr,
down_->add_header_field(key, val);
std::ignore = down_->end_header();
down_->send_payload(res.body());
// TODO: we should close the connection unless indicated otherwise
// (keepalive flag?). Also, we should clean up pending_.
down_->shutdown();
},
[this](const error& err) {
auto description = to_string(err);
down_->send_response(status::internal_server_error, "text/plain",
description);
// TODO: see above.
down_->shutdown();
});
pending_.emplace_back(std::move(hdl));
return static_cast<ptrdiff_t>(payload.size());
......
......@@ -18,8 +18,8 @@ bool server::is_reading() const noexcept {
return down_->is_reading();
}
void server::close() {
down_->close();
void server::shutdown() {
down_->shutdown();
}
void server::request_messages() {
......@@ -90,8 +90,8 @@ void server::abort(const error& reason) {
up_->abort(reason);
}
bool server::prepare_send() {
return up_->prepare_send();
void server::prepare_send() {
up_->prepare_send();
}
bool server::done_sending() {
......
......@@ -79,8 +79,8 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
}
}
bool length_prefix_framing::prepare_send() {
return up_->prepare_send();
void length_prefix_framing::prepare_send() {
up_->prepare_send();
}
bool length_prefix_framing::done_sending() {
......@@ -135,8 +135,8 @@ bool length_prefix_framing::end_message() {
}
}
void length_prefix_framing::close() {
down_->close();
void length_prefix_framing::shutdown() {
down_->shutdown();
}
// -- utility functions ------------------------------------------------------
......
......@@ -57,13 +57,8 @@ const short output_mask = POLLOUT;
// | (is_writing(mask) ? output_mask : 0));
// }
operation to_operation(const socket_manager_ptr& mgr,
std::optional<short> mask) {
operation to_operation(const socket_manager_ptr&, std::optional<short> mask) {
operation res = operation::none;
if (mgr->read_closed())
res = block_reads(res);
if (mgr->write_closed())
res = block_writes(res);
if (mask) {
if ((*mask & input_mask) != 0)
res = add_read_flag(res);
......@@ -187,33 +182,6 @@ void multiplexer::watch(disposable what) {
// -- thread-safe signaling ----------------------------------------------------
void multiplexer::dispose(socket_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
do_dispose(mgr);
} else {
write_to_pipe(pollset_updater::code::dispose_manager, mgr.release());
}
}
void multiplexer::shutdown_reading(socket_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
do_shutdown_reading(mgr);
} else {
write_to_pipe(pollset_updater::code::shutdown_reading, mgr.release());
}
}
void multiplexer::shutdown_writing(socket_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
do_shutdown_writing(mgr);
} else {
write_to_pipe(pollset_updater::code::shutdown_writing, mgr.release());
}
}
void multiplexer::init(socket_manager_ptr mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
......@@ -288,7 +256,6 @@ bool multiplexer::poll_once(bool blocking) {
CAF_LOG_DEBUG("poll() on" << pollset_.size() << "sockets reported"
<< presult << "event(s)");
// Scan pollset for events.
CAF_LOG_DEBUG("scan pollset for socket events");
if (auto revents = pollset_[0].revents; revents != 0) {
// Index 0 is always the pollset updater. This is the only handler that
// is allowed to modify pollset_ and managers_. Since this may very well
......@@ -297,6 +264,7 @@ bool multiplexer::poll_once(bool blocking) {
handle(mgr, pollset_[0].events, revents);
--presult;
}
apply_updates();
for (size_t i = 1; i < pollset_.size() && presult > 0; ++i) {
if (auto revents = pollset_[i].revents; revents != 0) {
handle(managers_[i], pollset_[i].events, revents);
......@@ -369,6 +337,8 @@ void multiplexer::set_thread_id() {
void multiplexer::run() {
CAF_LOG_TRACE("");
CAF_LOG_DEBUG("run multiplexer" << CAF_ARG(input_mask) << CAF_ARG(error_mask)
<< CAF_ARG(output_mask));
// On systems like Linux, we cannot disable sigpipe on the socket alone. We
// need to block the signal at thread level since some APIs (such as OpenSSL)
// are unsafe to call otherwise.
......@@ -393,6 +363,8 @@ void multiplexer::handle(const socket_manager_ptr& mgr,
<< CAF_ARG(events) << CAF_ARG(revents));
CAF_ASSERT(mgr != nullptr);
bool checkerror = true;
CAF_LOG_DEBUG("handle event on socket" << mgr->handle().id << CAF_ARG(events)
<< CAF_ARG(revents));
// Note: we double-check whether the manager is actually reading because a
// previous action from the pipe may have disabled reading.
if ((revents & input_mask) != 0 && is_reading(mgr.get())) {
......@@ -478,36 +450,11 @@ void multiplexer::do_shutdown() {
shutting_down_ = true;
apply_updates();
// Skip the first manager (the pollset updater).
for (size_t i = 1; i < managers_.size(); ++i) {
auto& mgr = managers_[i];
mgr->close_read();
update_for(static_cast<ptrdiff_t>(i)).events &= ~input_mask;
}
for (size_t i = 1; i < managers_.size(); ++i)
managers_[i]->dispose();
apply_updates();
}
void multiplexer::do_dispose(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
mgr->handle_error(sec::disposed);
update_for(mgr.get()).events = 0;
}
void multiplexer::do_shutdown_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (!shutting_down_ && !mgr->read_closed()) {
mgr->close_read();
update_for(mgr.get()).events &= ~input_mask;
}
}
void multiplexer::do_shutdown_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (!shutting_down_ && !mgr->write_closed()) {
mgr->close_write();
update_for(mgr.get()).events &= ~output_mask;
}
}
void multiplexer::do_init(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (!shutting_down_) {
......
......@@ -60,15 +60,6 @@ void pollset_updater::handle_read_event() {
case code::init_manager:
mpx_->do_init(as_mgr(ptr));
break;
case code::dispose_manager:
mpx_->do_dispose(as_mgr(ptr));
break;
case code::shutdown_reading:
mpx_->do_shutdown_reading(as_mgr(ptr));
break;
case code::shutdown_writing:
mpx_->do_shutdown_writing(as_mgr(ptr));
break;
case code::run_action:
run_action(ptr);
break;
......@@ -83,13 +74,13 @@ void pollset_updater::handle_read_event() {
}
} else if (num_bytes == 0) {
CAF_LOG_DEBUG("pipe closed, assume shutdown");
owner_->shutdown();
owner_->deregister();
return;
} else if (last_socket_error_is_temporary()) {
return;
} else {
CAF_LOG_DEBUG("pollset updater failed to read from the pipe");
owner_->shutdown();
CAF_LOG_ERROR("pollset updater failed to read from the pipe");
owner_->deregister();
return;
}
}
......
......@@ -24,8 +24,8 @@ std::string_view server::scrape_state::scrape() {
// -- implementation of http::upper_layer --------------------------------------
bool server::prepare_send() {
return true;
void server::prepare_send() {
// nop
}
bool server::done_sending() {
......@@ -45,19 +45,19 @@ error server::init(socket_manager*, http::lower_layer* down, const settings&) {
ptrdiff_t server::consume(const http::header& hdr, const_byte_span payload) {
if (hdr.path() != "/metrics") {
down_->send_response(http::status::not_found, "text/plain", "Not found.");
down_->close();
down_->shutdown();
} else if (hdr.method() != http::method::get) {
down_->send_response(http::status::method_not_allowed, "text/plain",
"Method not allowed.");
down_->close();
down_->shutdown();
} else if (!hdr.query().empty() || !hdr.fragment().empty()) {
down_->send_response(http::status::bad_request, "text/plain",
"No fragment or query allowed.");
down_->close();
down_->shutdown();
} else {
auto str = state_->scrape();
down_->send_response(http::status::ok, "text/plain;version=0.0.4", str);
down_->close();
down_->shutdown();
}
return static_cast<ptrdiff_t>(payload.size());
}
......
......@@ -14,4 +14,8 @@ bool socket_event_layer::do_handover(std::unique_ptr<socket_event_layer>&) {
return false;
}
bool socket_event_layer::finalized() const noexcept {
return true;
}
} // namespace caf::net
......@@ -19,10 +19,10 @@ socket_manager::socket_manager(multiplexer* mpx, socket fd,
CAF_ASSERT(fd_ != invalid_socket);
CAF_ASSERT(mpx_ != nullptr);
CAF_ASSERT(handler_ != nullptr);
memset(&flags_, 0, sizeof(flags_t));
}
socket_manager::~socket_manager() {
if (fd_)
close(fd_);
}
......@@ -53,12 +53,10 @@ bool socket_manager::is_writing() const noexcept {
// -- event loop management ----------------------------------------------------
void socket_manager::register_reading() {
if (!flags_.read_closed)
mpx_->register_reading(this);
}
void socket_manager::register_writing() {
if (!flags_.write_closed)
mpx_->register_writing(this);
}
......@@ -74,23 +72,6 @@ void socket_manager::deregister() {
mpx_->deregister(this);
}
void socket_manager::shutdown_read() {
deregister_reading();
flags_.read_closed = true;
}
void socket_manager::shutdown_write() {
deregister_writing();
flags_.write_closed = true;
}
void socket_manager::shutdown() {
flags_.read_closed = true;
flags_.write_closed = true;
disposed_ = true;
deregister();
}
// -- callbacks for the handler ------------------------------------------------
void socket_manager::schedule_handover() {
......@@ -112,21 +93,20 @@ void socket_manager::schedule(action what) {
});
}
// -- callbacks for the multiplexer --------------------------------------------
void socket_manager::close_read() noexcept {
// TODO: extend transport API for closing read operations.
flags_.read_closed = true;
if (flags_.write_closed)
disposed_ = true;
void socket_manager::shutdown() {
if (!shutting_down_) {
shutting_down_ = true;
dispose();
} else {
// This usually only happens after disposing the manager if the handler
// still had data to send.
mpx_->schedule_fn([ptr = strong_this()] { //
ptr->cleanup();
});
}
}
void socket_manager::close_write() noexcept {
// TODO: extend transport API for closing write operations.
flags_.write_closed = true;
if (flags_.read_closed)
disposed_ = true;
}
// -- callbacks for the multiplexer --------------------------------------------
error socket_manager::init(const settings& cfg) {
CAF_LOG_TRACE(CAF_ARG(cfg));
......@@ -138,26 +118,46 @@ error socket_manager::init(const settings& cfg) {
}
void socket_manager::handle_read_event() {
if (handler_)
handler_->handle_read_event();
else
deregister();
}
void socket_manager::handle_write_event() {
if (handler_)
handler_->handle_write_event();
else
deregister();
}
void socket_manager::handle_error(sec code) {
if (!disposed_)
disposed_ = true;
if (handler_) {
if (!shutting_down_) {
handler_->abort(make_error(code));
handler_ = nullptr;
disposed_ = true;
shutting_down_ = true;
}
if (code == sec::disposed && !handler_->finalized()) {
// When disposing the manger, the transport is still allowed to send any
// pending data and it will call shutdown() later to trigger cleanup().
deregister_reading();
} else {
cleanup();
}
}
}
// -- implementation of disposable_impl ----------------------------------------
void socket_manager::dispose() {
if (!disposed())
mpx_->dispose(this);
bool expected = false;
if (disposed_.compare_exchange_strong(expected, true)) {
mpx_->schedule_fn([ptr = strong_this()] { //
ptr->handle_error(sec::disposed);
});
}
}
bool socket_manager::disposed() const noexcept {
......@@ -174,6 +174,15 @@ void socket_manager::deref_disposable() const noexcept {
// -- utility functions --------------------------------------------------------
void socket_manager::cleanup() {
deregister();
handler_.reset();
if (fd_) {
close(fd_);
fd_ = invalid_socket;
}
}
socket_manager_ptr socket_manager::strong_this() {
return socket_manager_ptr{this};
}
......
......@@ -111,12 +111,14 @@ bool stream_transport::is_reading() const noexcept {
return max_read_size_ > 0;
}
void stream_transport::close() {
parent_->shutdown_read();
if (write_buf_.empty())
parent_->shutdown_write();
else
void stream_transport::shutdown() {
if (write_buf_.empty()) {
parent_->shutdown();
} else {
configure_read(receive_policy::stop());
parent_->deregister_reading();
flags_.shutting_down = true;
}
}
// -- implementation of transport ----------------------------------------------
......@@ -283,29 +285,25 @@ void stream_transport::handle_write_event() {
return;
}
}
// Allow the upper layer to add extra data to the write buffer.
if (!up_->prepare_send()) {
CAF_LOG_DEBUG("prepare_send failed");
parent_->deregister();
return;
}
// When shutting down, we flush our buffer and then shut down the manager.
if (flags_.shutting_down) {
if (write_buf_.empty()) {
if (up_->done_sending()) {
if (!flags_.shutting_down)
parent_->deregister_writing();
else
parent_->shutdown_write();
}
parent_->shutdown();
return;
}
} else if (can_send_more()) {
// Allow the upper layer to add extra data to the write buffer.
up_->prepare_send();
}
auto write_res = policy_->write(fd_, write_buf_);
if (write_res > 0) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + write_res);
if (write_buf_.empty() && up_->done_sending()) {
if (!flags_.shutting_down)
if (!flags_.shutting_down) {
parent_->deregister_writing();
else
parent_->shutdown_write();
} else {
parent_->shutdown();
}
}
return;
} else if (write_res < 0) {
......@@ -331,6 +329,11 @@ void stream_transport::handle_write_event() {
void stream_transport::abort(const error& reason) {
up_->abort(reason);
flags_.shutting_down = true;
}
bool stream_transport::finalized() const noexcept {
return write_buf_.empty();
}
// -- free functions -----------------------------------------------------------
......
......@@ -75,6 +75,7 @@ expected<tcp_accept_socket> new_tcp_acceptor_impl(uint16_t port,
CAF_NET_SYSCALL("bind", res, !=, 0,
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<socket_size_type>(sizeof(sa))));
CAF_LOG_DEBUG("bound socket" << fd << "to listen on port" << port);
return sguard.release();
}
......@@ -130,6 +131,7 @@ make_tcp_accept_socket(uint16_t port, std::string addr, bool reuse_addr) {
}
expected<tcp_stream_socket> accept(tcp_accept_socket x) {
CAF_LOG_TRACE(CAF_ARG(x));
auto sock = ::accept(x.id, nullptr, nullptr);
if (sock == net::invalid_socket_id) {
auto err = net::last_socket_error();
......@@ -139,6 +141,7 @@ expected<tcp_stream_socket> accept(tcp_accept_socket x) {
}
return caf::make_error(sec::socket_operation_failed, "tcp accept failed");
}
CAF_LOG_DEBUG("accepted TCP socket" << sock << "on accept socket" << x.id);
return tcp_stream_socket{sock};
}
......
......@@ -96,8 +96,9 @@ ptrdiff_t client::consume(byte_span buffer, byte_span delta) {
}
}
bool client::prepare_send() {
return handshake_completed() ? upper_layer().prepare_send() : true;
void client::prepare_send() {
if (handshake_completed())
upper_layer().prepare_send();
}
bool client::done_sending() {
......
......@@ -4,7 +4,10 @@
#include "caf/net/web_socket/default_trait.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/net/web_socket/frame.hpp"
#include "caf/sec.hpp"
namespace caf::net::web_socket {
......@@ -34,4 +37,9 @@ bool default_trait::convert(std::string_view text, input_type& x) {
return true;
}
error default_trait::last_error() {
CAF_LOG_ERROR("default_trait::last_error called");
return {sec::logic_error};
}
} // namespace caf::net::web_socket
......@@ -30,7 +30,7 @@ bool framing::is_reading() const noexcept {
return down_->is_reading();
}
void framing::close(status code, std::string_view msg) {
void framing::shutdown(status code, std::string_view msg) {
auto code_val = static_cast<uint16_t>(code);
uint32_t mask_key = 0;
byte_buffer payload;
......@@ -47,6 +47,7 @@ void framing::close(status code, std::string_view msg) {
detail::rfc6455::assemble_frame(detail::rfc6455::connection_close, mask_key,
payload, down_->output_buffer());
down_->end_output();
down_->shutdown();
}
void framing::request_messages() {
......
......@@ -13,13 +13,13 @@ lower_layer::~lower_layer() {
// nop
}
void lower_layer::close() {
void lower_layer::shutdown() {
using namespace std::literals;
close(status::normal_close, "EOF"sv);
shutdown(status::normal_close, "EOF"sv);
}
void lower_layer::close(const error& reason) {
close(status::unexpected_condition, to_string(reason));
void lower_layer::shutdown(const error& reason) {
shutdown(status::unexpected_condition, to_string(reason));
}
} // namespace caf::net::web_socket
......@@ -75,8 +75,9 @@ ptrdiff_t server::consume(byte_span input, byte_span delta) {
}
}
bool server::prepare_send() {
return handshake_complete_ ? upper_layer().prepare_send() : true;
void server::prepare_send() {
if (handshake_complete_)
upper_layer().prepare_send();
}
bool server::done_sending() {
......
......@@ -23,7 +23,7 @@ bool mock_stream_transport::is_reading() const noexcept {
return max_read_size > 0;
}
void mock_stream_transport::close() {
void mock_stream_transport::shutdown() {
// nop
}
......
......@@ -34,7 +34,7 @@ public:
bool is_reading() const noexcept override;
void close() override;
void shutdown() override;
void configure_read(caf::net::receive_policy policy) override;
......
......@@ -59,18 +59,17 @@ public:
return none;
}
bool prepare_send() override {
while (self->consume_message())
void prepare_send() override {
while (!self->terminated() && self->consume_message())
; // repeat
return !self->terminated();
}
bool done_sending() override {
return self->try_block_mailbox();
}
void abort(const error& reason) override {
CAF_FAIL("app::abort called: " << reason);
void abort(const error&) override {
// nop
}
ptrdiff_t consume(byte_span buf, byte_span) override {
......
......@@ -115,9 +115,8 @@ public:
}
};
bool prepare_send() override {
if (done || !adapter)
return true;
void prepare_send() override {
if (!done && adapter) {
auto helper = send_helper{this};
while (down->can_send_more()) {
auto [again, consumed] = adapter->pull(async::delay_errors, 1, helper);
......@@ -131,7 +130,7 @@ public:
}
}
MESSAGE(written_bytes.size() << " bytes written");
return true;
}
}
bool done_sending() override {
......
......@@ -59,8 +59,8 @@ public:
CAF_FAIL("app::abort called: " << reason);
}
bool prepare_send() override {
return true;
void prepare_send() override {
// nop
}
bool done_sending() override {
......
......@@ -55,8 +55,8 @@ public:
return none;
}
bool prepare_send() override {
return true;
void prepare_send() override {
// nop
}
bool done_sending() override {
......
......@@ -93,11 +93,13 @@ public:
}
void abort(const error& reason) override {
FAIL("abort called: " << reason);
abort_reason = reason;
}
std::string name;
error abort_reason;
private:
std::byte* read_position_begin() {
return rd_buf_.data() + rd_buf_pos_;
......@@ -232,8 +234,8 @@ SCENARIO("a multiplexer terminates its thread after shutting down") {
mpx->shutdown();
THEN("the thread terminates and all socket managers get shut down") {
mpx_thread.join();
CHECK(alice_mgr->read_closed());
CHECK(bob_mgr->read_closed());
CHECK(alice_mgr->disposed());
CHECK(bob_mgr->disposed());
}
}
}
......
......@@ -79,7 +79,7 @@ public:
// Note: the execution unit is the owner of this object. As as long as the
// execution unit lives, accessing `this` is safe.
auto do_resume = make_action([this] { down->request_messages(); });
auto do_cancel = make_action([this] { down->close(); });
auto do_cancel = make_action([this] { down->shutdown(); });
adapter_ = adapter_type::make(std::move(buf), mgr, std::move(do_resume),
std::move(do_cancel));
return none;
......@@ -88,8 +88,8 @@ public:
}
}
bool prepare_send() override {
return true;
void prepare_send() override {
// nop
}
bool done_sending() override {
......
......@@ -72,8 +72,8 @@ public:
return none;
}
bool prepare_send() override {
return true;
void prepare_send() override {
// nop
}
bool done_sending() override {
......
......@@ -93,12 +93,11 @@ public:
return static_cast<ptrdiff_t>(recv_buf_->size());
}
bool prepare_send() override {
void prepare_send() override {
MESSAGE("prepare_send called");
auto& buf = down->output_buffer();
auto data = as_bytes(make_span(hello_manager));
buf.insert(buf.end(), data.begin(), data.end());
return true;
}
bool done_sending() override {
......
......@@ -59,18 +59,17 @@ public:
return none;
}
bool prepare_send() override {
while (self->consume_message())
void prepare_send() override {
while (!self->terminated() && self->consume_message())
; // repeat
return !self->terminated();
}
bool done_sending() override {
return self->try_block_mailbox();
}
void abort(const error& reason) override {
CAF_FAIL("app::abort called: " << reason);
void abort(const error&) override {
// nop
}
ptrdiff_t consume(byte_span buf, byte_span) override {
......
......@@ -32,8 +32,8 @@ public:
return none;
}
bool prepare_send() override {
return true;
void prepare_send() override {
// nop
}
bool done_sending() override {
......
......@@ -39,8 +39,8 @@ public:
return none;
}
bool prepare_send() override {
return true;
void prepare_send() override {
// nop
}
bool done_sending() override {
......
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