Commit b7aedb1d authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/caf-net-prometheus'

parents 4b9e5e2a 02bb479c
...@@ -421,14 +421,17 @@ strong_actor_ptr middleman::remote_lookup(std::string name, ...@@ -421,14 +421,17 @@ strong_actor_ptr middleman::remote_lookup(std::string name,
void middleman::start() { void middleman::start() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// Launch background tasks. // Launch background tasks unless caf-net is also available. In that case, the
if (auto prom = get_if<config_value::dictionary>( // net::middleman takes care of these.
&system().config(), "caf.middleman.prometheus-http")) { if (!system().has_network_manager()) {
auto ptr = std::make_unique<prometheus_scraping>(system()); if (auto prom = get_if<config_value::dictionary>(
if (auto port = ptr->start(*prom)) { &system().config(), "caf.middleman.prometheus-http")) {
CAF_ASSERT(*port != 0); auto ptr = std::make_unique<prometheus_scraping>(system());
prometheus_scraping_port_ = *port; if (auto port = ptr->start(*prom)) {
background_tasks_.emplace_back(std::move(ptr)); CAF_ASSERT(*port != 0);
prometheus_scraping_port_ = *port;
background_tasks_.emplace_back(std::move(ptr));
}
} }
} }
// Launch backend. // Launch backend.
......
...@@ -51,6 +51,7 @@ caf_add_component( ...@@ -51,6 +51,7 @@ caf_add_component(
src/net/network_socket.cpp src/net/network_socket.cpp
src/net/pipe_socket.cpp src/net/pipe_socket.cpp
src/net/pollset_updater.cpp src/net/pollset_updater.cpp
src/net/prometheus/server.cpp
src/net/socket.cpp src/net/socket.cpp
src/net/socket_event_layer.cpp src/net/socket_event_layer.cpp
src/net/socket_manager.cpp src/net/socket_manager.cpp
...@@ -94,6 +95,7 @@ caf_add_component( ...@@ -94,6 +95,7 @@ caf_add_component(
net.operation net.operation
net.pipe_socket net.pipe_socket
net.producer_adapter net.producer_adapter
net.prometheus.server
net.socket net.socket
net.socket_guard net.socket_guard
net.ssl.transport net.ssl.transport
......
...@@ -21,10 +21,6 @@ public: ...@@ -21,10 +21,6 @@ public:
using factory_type = Factory; using factory_type = Factory;
using read_result = typename socket_event_layer::read_result;
using write_result = typename socket_event_layer::write_result;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> template <class... Ts>
...@@ -54,40 +50,34 @@ public: ...@@ -54,40 +50,34 @@ public:
return none; return none;
} }
read_result handle_read_event() override { void handle_read_event() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (auto x = accept(fd_)) { if (auto x = accept(fd_)) {
socket_manager_ptr child = factory_.make(owner_->mpx_ptr(), *x); socket_manager_ptr child = factory_.make(owner_->mpx_ptr(), *x);
if (!child) { if (!child) {
CAF_LOG_ERROR("factory failed to create a new child"); CAF_LOG_ERROR("factory failed to create a new child");
return read_result::abort; owner_->deregister();
return;
} }
if (auto err = child->init(cfg_)) { if (auto err = child->init(cfg_)) {
CAF_LOG_ERROR("failed to initialize new child:" << err); CAF_LOG_ERROR("failed to initialize new child:" << err);
return read_result::abort; owner_->deregister();
return;
} }
if (limit_ == 0) { if (limit_ != 0 && ++accepted_ == limit_) {
return read_result::again; // TODO: ask owner to close socket.
} else { owner_->deregister();
return ++accepted_ < limit_ ? read_result::again : read_result::stop; return;
} }
} else { } else {
CAF_LOG_ERROR("accept failed:" << x.error()); CAF_LOG_ERROR("accept failed:" << x.error());
return read_result::stop; owner_->deregister();
} }
} }
read_result handle_buffered_data() override { void handle_write_event() override {
return read_result::again;
}
read_result handle_continue_reading() override {
return read_result::again;
}
write_result handle_write_event() override {
CAF_LOG_ERROR("connection_acceptor received write event"); CAF_LOG_ERROR("connection_acceptor received write event");
return write_result::stop; owner_->deregister_writing();
} }
void abort(const error& reason) override { void abort(const error& reason) override {
......
...@@ -5,27 +5,28 @@ ...@@ -5,27 +5,28 @@
#pragma once #pragma once
#include "caf/async/consumer.hpp" #include "caf/async/consumer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net { namespace caf::net {
/// Connects a socket manager to an asynchronous consumer resource. Whenever new /// Connects a socket manager to an asynchronous consumer resource. Whenever new
/// data becomes ready, the adapter registers the socket manager for writing. /// data becomes ready, the adapter registers the socket manager for writing.
template <class Buffer> template <class Buffer>
class consumer_adapter final : public ref_counted, public async::consumer { class consumer_adapter final : public detail::atomic_ref_counted,
public async::consumer {
public: public:
using buf_ptr = intrusive_ptr<Buffer>; using buf_ptr = intrusive_ptr<Buffer>;
using ptr_type = intrusive_ptr<consumer_adapter>;
void on_producer_ready() override { void on_producer_ready() override {
// nop // nop
} }
void on_producer_wakeup() override { void on_producer_wakeup() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { // mgr_->schedule(do_wakeup_);
adapter->on_wakeup();
});
} }
void ref_consumer() const noexcept override { void ref_consumer() const noexcept override {
...@@ -44,22 +45,20 @@ public: ...@@ -44,22 +45,20 @@ public:
void cancel() { void cancel() {
buf_->cancel(); buf_->cancel();
buf_ = nullptr; buf_ = nullptr;
mgr_ = nullptr;
do_wakeup_.dispose();
do_wakeup_ = nullptr;
} }
bool has_data() const noexcept { bool has_data() const noexcept {
return buf_->has_data(); return buf_->has_data();
} }
/// Tries to open the resource for reading. static ptr_type make(buf_ptr buf, socket_manager_ptr mgr, action do_wakeup) {
/// @returns a connected adapter that reads from the resource on success, if (buf) {
/// `nullptr` otherwise. auto adapter = ptr_type{new consumer_adapter(buf, mgr,
template <class Resource> std::move(do_wakeup)), //
static intrusive_ptr<consumer_adapter> false};
try_open(socket_manager* owner, Resource src) {
CAF_ASSERT(owner != nullptr);
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<consumer_adapter>;
auto adapter = ptr_type{new consumer_adapter(owner, buf), false};
buf->set_consumer(adapter); buf->set_consumer(adapter);
return adapter; return adapter;
} else { } else {
...@@ -67,32 +66,18 @@ public: ...@@ -67,32 +66,18 @@ public:
} }
} }
friend void intrusive_ptr_add_ref(const consumer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const consumer_adapter* ptr) noexcept {
ptr->deref();
}
private: private:
consumer_adapter(socket_manager* owner, buf_ptr buf) consumer_adapter(buf_ptr buf, socket_manager_ptr mgr, action do_wakeup)
: mgr_(owner), buf_(std::move(buf)) { : buf_(std::move(buf)),
mgr_(std::move(mgr)),
do_wakeup_(std::move(do_wakeup)) {
// nop // nop
} }
auto strong_this() {
return intrusive_ptr{this};
}
void on_wakeup() {
if (buf_ && buf_->has_consumer_event()) {
mgr_->mpx().register_writing(mgr_);
}
}
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_; intrusive_ptr<Buffer> buf_;
socket_manager_ptr mgr_;
action do_wakeup_;
action do_cancel_;
}; };
template <class T> template <class T>
......
...@@ -5,6 +5,7 @@ ...@@ -5,6 +5,7 @@
#pragma once #pragma once
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
namespace caf::net { namespace caf::net {
...@@ -17,12 +18,18 @@ public: ...@@ -17,12 +18,18 @@ public:
/// Queries whether the output device can accept more data straight away. /// Queries whether the output device can accept more data straight away.
[[nodiscard]] virtual bool can_send_more() const noexcept = 0; [[nodiscard]] virtual bool can_send_more() const noexcept = 0;
/// Halt receiving of additional bytes or messages.
virtual void suspend_reading() = 0;
/// Queries whether the lower layer is currently configured to halt receiving /// Queries whether the lower layer is currently configured to halt receiving
/// of additional bytes or messages. /// of additional bytes or messages.
[[nodiscard]] virtual bool stopped_reading() const noexcept = 0; [[nodiscard]] virtual bool is_reading() const noexcept = 0;
/// Shuts down any connection or session gracefully. Any pending data gets
/// flushed before closing the socket.
virtual void close() = 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);
}; };
} // namespace caf::net } // namespace caf::net
...@@ -24,11 +24,6 @@ public: ...@@ -24,11 +24,6 @@ public:
/// function to decide whether it has to wait for write events on the socket. /// function to decide whether it has to wait for write events on the socket.
[[nodiscard]] virtual bool done_sending() = 0; [[nodiscard]] virtual bool done_sending() = 0;
/// Called by the lower layer after some event triggered re-registering the
/// socket manager for read operations after it has been stopped previously
/// by the read policy. May restart consumption of bytes or messages.
virtual void continue_reading() = 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.
virtual void abort(const error& reason) = 0; virtual void abort(const error& reason) = 0;
}; };
......
// 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
namespace caf::net::http {
/// Stores context information for a request. For HTTP/2, this is the stream ID.
class context {};
} // namespace caf::net::http
...@@ -10,7 +10,6 @@ ...@@ -10,7 +10,6 @@
namespace caf::net::http { namespace caf::net::http {
class context;
class header; class header;
class lower_layer; class lower_layer;
class server; class server;
......
...@@ -18,28 +18,33 @@ class CAF_NET_EXPORT lower_layer : public generic_lower_layer { ...@@ -18,28 +18,33 @@ class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public: public:
virtual ~lower_layer(); virtual ~lower_layer();
/// Start or re-start reading data from the client.
virtual void request_messages() = 0;
/// Stops reading messages until calling `request_messages`.
virtual void suspend_reading() = 0;
/// Sends the next header to the client. /// Sends the next header to the client.
virtual bool virtual bool send_header(status code, const header_fields_map& fields) = 0;
send_header(context ctx, status code, const header_fields_map& fields)
= 0;
/// Sends the payload after the header. /// Sends the payload after the header.
virtual bool send_payload(context, const_byte_span bytes) = 0; virtual bool send_payload(const_byte_span bytes) = 0;
/// Sends a chunk of data if the full payload is unknown when starting to /// Sends a chunk of data if the full payload is unknown when starting to
/// send. /// send.
virtual bool send_chunk(context, const_byte_span bytes) = 0; virtual bool send_chunk(const_byte_span bytes) = 0;
/// Sends the last chunk, completing a chunked payload. /// Sends the last chunk, completing a chunked payload.
virtual bool send_end_of_chunks() = 0; virtual bool send_end_of_chunks() = 0;
/// Terminates an HTTP context.
virtual void fin(context) = 0;
/// Convenience function for sending header and payload. Automatically sets /// Convenience function for sending header and payload. Automatically sets
/// the header fields `Content-Type` and `Content-Length`. /// the header fields `Content-Type` and `Content-Length`.
bool send_response(context ctx, status codek, std::string_view content_type, bool send_response(status code, std::string_view content_type,
const_byte_span content); const_byte_span content);
/// @copydoc send_response
bool send_response(status code, std::string_view content_type,
std::string_view content);
}; };
} // namespace caf::net::http } // namespace caf::net::http
...@@ -10,7 +10,6 @@ ...@@ -10,7 +10,6 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/connection_acceptor.hpp" #include "caf/net/connection_acceptor.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/http/context.hpp"
#include "caf/net/http/header.hpp" #include "caf/net/http/header.hpp"
#include "caf/net/http/header_fields_map.hpp" #include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/lower_layer.hpp" #include "caf/net/http/lower_layer.hpp"
...@@ -39,8 +38,6 @@ public: ...@@ -39,8 +38,6 @@ public:
using status_code_type = status; using status_code_type = status;
using context_type = context;
using header_type = header; using header_type = header;
enum class mode { enum class mode {
...@@ -80,20 +77,21 @@ public: ...@@ -80,20 +77,21 @@ public:
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
void suspend_reading() override; bool is_reading() const noexcept override;
bool stopped_reading() const noexcept override; void close() override;
bool send_header(context, status code, void request_messages() override;
const header_fields_map& fields) override;
bool send_payload(context, const_byte_span bytes) override; void suspend_reading() override;
bool send_chunk(context, const_byte_span bytes) override; bool send_header(status code, const header_fields_map& fields) override;
bool send_end_of_chunks() override; bool send_payload(const_byte_span bytes) override;
void fin(context) override; bool send_chunk(const_byte_span bytes) override;
bool send_end_of_chunks() override;
// -- stream_oriented::upper_layer implementation ---------------------------- // -- stream_oriented::upper_layer implementation ----------------------------
...@@ -106,8 +104,6 @@ public: ...@@ -106,8 +104,6 @@ public:
bool done_sending() override; bool done_sending() override;
void continue_reading() override;
ptrdiff_t consume(byte_span input, byte_span) override; ptrdiff_t consume(byte_span input, byte_span) override;
private: private:
......
...@@ -28,18 +28,12 @@ public: ...@@ -28,18 +28,12 @@ public:
= 0; = 0;
/// Consumes an HTTP message. /// Consumes an HTTP message.
/// @param ctx Identifies this request. The response message must pass this
/// back to the lower layer. Allows clients to send multiple
/// requests in "parallel" (i.e., send multiple requests before
/// receiving the response on the first one).
/// @param hdr The header fields for the received message. /// @param hdr The header fields for the received message.
/// @param payload The payload of the received message. /// @param payload The payload of the received message.
/// @returns The number of consumed bytes or a negative value to signal an /// @returns The number of consumed bytes or a negative value to signal an
/// error. /// error.
/// @note Discarded data is lost permanently. /// @note Discarded data is lost permanently.
virtual ptrdiff_t virtual ptrdiff_t consume(const header& hdr, const_byte_span payload) = 0;
consume(context ctx, const header& hdr, const_byte_span payload)
= 0;
}; };
} // namespace caf::net::http } // namespace caf::net::http
...@@ -81,8 +81,6 @@ public: ...@@ -81,8 +81,6 @@ public:
ptrdiff_t consume(byte_span buffer, byte_span delta) override; ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void continue_reading() override;
bool prepare_send() override; bool prepare_send() override;
bool done_sending() override; bool done_sending() override;
...@@ -95,7 +93,7 @@ public: ...@@ -95,7 +93,7 @@ public:
void suspend_reading() override; void suspend_reading() override;
bool stopped_reading() const noexcept override; bool is_reading() const noexcept override;
void begin_message() override; void begin_message() override;
...@@ -103,9 +101,7 @@ public: ...@@ -103,9 +101,7 @@ public:
bool end_message() override; bool end_message() override;
void send_close_message() override; void close() override;
void send_close_message(const error& reason) override;
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
......
...@@ -116,9 +116,9 @@ public: ...@@ -116,9 +116,9 @@ public:
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper); auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) { if (!again) {
if (helper.err) { if (helper.err) {
down_->send_close_message(helper.err); down_->close(helper.err);
} else { } else {
down_->send_close_message(); down_->close();
} }
in_ = nullptr; in_ = nullptr;
} else if (helper.aborted) { } else if (helper.aborted) {
......
...@@ -47,6 +47,9 @@ public: ...@@ -47,6 +47,9 @@ public:
/// Pulls messages from the transport until calling `suspend_reading`. /// Pulls messages from the transport until calling `suspend_reading`.
virtual void request_messages() = 0; virtual void request_messages() = 0;
/// Stops reading messages until calling `request_messages`.
virtual void suspend_reading() = 0;
/// Prepares the layer for an outgoing message, e.g., by allocating an /// Prepares the layer for an outgoing message, e.g., by allocating an
/// output buffer as necessary. /// output buffer as necessary.
virtual void begin_message() = 0; virtual void begin_message() = 0;
...@@ -62,17 +65,6 @@ public: ...@@ -62,17 +65,6 @@ public:
/// @note When returning `false`, clients must also call /// @note When returning `false`, clients must also call
/// `down.set_read_error(...)` with an appropriate error code. /// `down.set_read_error(...)` with an appropriate error code.
virtual bool end_message() = 0; virtual bool end_message() = 0;
/// Inform the remote endpoint that no more messages will arrive.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message() = 0;
/// Inform the remote endpoint that no more messages will arrive because of
/// an error.
/// @note Not every protocol has a dedicated close message. Some
/// implementation may simply do nothing.
virtual void send_close_message(const error& reason) = 0;
}; };
} // namespace caf::net::message_oriented } // namespace caf::net::message_oriented
...@@ -73,10 +73,10 @@ public: ...@@ -73,10 +73,10 @@ public:
size_t num_socket_managers() const noexcept; size_t num_socket_managers() const noexcept;
/// Returns the index of `mgr` in the pollset or `-1`. /// Returns the index of `mgr` in the pollset or `-1`.
ptrdiff_t index_of(const socket_manager_ptr& mgr); ptrdiff_t index_of(const socket_manager_ptr& mgr) const noexcept;
/// Returns the index of `fd` in the pollset or `-1`. /// Returns the index of `fd` in the pollset or `-1`.
ptrdiff_t index_of(socket fd); ptrdiff_t index_of(socket fd) const noexcept;
/// Returns the owning @ref middleman instance. /// Returns the owning @ref middleman instance.
middleman& owner(); middleman& owner();
...@@ -89,25 +89,9 @@ public: ...@@ -89,25 +89,9 @@ public:
// -- thread-safe signaling -------------------------------------------------- // -- thread-safe signaling --------------------------------------------------
/// Registers `mgr` for read events. /// Schedules a call to `mgr->handle_error(sec::disposed)`.
/// @thread-safe
void register_reading(const socket_manager_ptr& mgr);
/// Registers `mgr` for write events.
/// @thread-safe
void register_writing(const socket_manager_ptr& mgr);
/// Triggers a continue reading event for `mgr`.
/// @thread-safe
void continue_reading(const socket_manager_ptr& mgr);
/// Triggers a continue writing event for `mgr`.
/// @thread-safe /// @thread-safe
void continue_writing(const socket_manager_ptr& mgr); void dispose(const socket_manager_ptr& mgr);
/// Schedules a call to `mgr->handle_error(sec::discarded)`.
/// @thread-safe
void discard(const socket_manager_ptr& mgr);
/// Stops further reading by `mgr`. /// Stops further reading by `mgr`.
/// @thread-safe /// @thread-safe
...@@ -136,6 +120,29 @@ public: ...@@ -136,6 +120,29 @@ public:
/// @thread-safe /// @thread-safe
void shutdown(); void shutdown();
// -- callbacks for socket managers ------------------------------------------
/// Registers `mgr` for read events.
void register_reading(socket_manager* mgr);
/// Registers `mgr` for write events.
void register_writing(socket_manager* mgr);
/// Deregisters `mgr` from read events.
void deregister_reading(socket_manager* mgr);
/// Deregisters `mgr` from write events.
void deregister_writing(socket_manager* mgr);
/// Deregisters @p mgr from read and write events.
void deregister(socket_manager* mgr);
/// Queries whether `mgr` is currently registered for reading.
bool is_reading(const socket_manager* mgr) const noexcept;
/// Queries whether `mgr` is currently registered for writing.
bool is_writing(const socket_manager* mgr) const noexcept;
// -- control flow ----------------------------------------------------------- // -- control flow -----------------------------------------------------------
/// Polls I/O activity once and runs all socket event handlers that become /// Polls I/O activity once and runs all socket event handlers that become
...@@ -160,15 +167,12 @@ protected: ...@@ -160,15 +167,12 @@ protected:
/// Handles an I/O event on given manager. /// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, short events, short revents); void handle(const socket_manager_ptr& mgr, short events, short revents);
/// Transfers socket ownership from one manager to another.
void do_handover(const socket_manager_ptr& mgr);
/// Returns a change entry for the socket at given index. Lazily creates a new /// Returns a change entry for the socket at given index. Lazily creates a new
/// entry before returning if necessary. /// entry before returning if necessary.
poll_update& update_for(ptrdiff_t index); poll_update& update_for(ptrdiff_t index);
/// Returns a change entry for the socket of the manager. /// Returns a change entry for the socket of the manager.
poll_update& update_for(const socket_manager_ptr& mgr); poll_update& update_for(socket_manager* mgr);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event /// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater. /// later via the pollset updater.
...@@ -182,13 +186,7 @@ protected: ...@@ -182,13 +186,7 @@ protected:
} }
/// Queries the currently active event bitmask for `mgr`. /// Queries the currently active event bitmask for `mgr`.
short active_mask_of(const socket_manager_ptr& mgr); short active_mask_of(const socket_manager* mgr) const noexcept;
/// Queries whether `mgr` is currently registered for reading.
bool is_reading(const socket_manager_ptr& mgr);
/// Queries whether `mgr` is currently registered for writing.
bool is_writing(const socket_manager_ptr& mgr);
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
...@@ -226,13 +224,7 @@ private: ...@@ -226,13 +224,7 @@ private:
void do_register_reading(const socket_manager_ptr& mgr); void do_register_reading(const socket_manager_ptr& mgr);
void do_register_writing(const socket_manager_ptr& mgr); void do_dispose(const socket_manager_ptr& mgr);
void do_continue_reading(const socket_manager_ptr& mgr);
void do_continue_writing(const socket_manager_ptr& mgr);
void do_discard(const socket_manager_ptr& mgr);
void do_shutdown_reading(const socket_manager_ptr& mgr); void do_shutdown_reading(const socket_manager_ptr& mgr);
......
...@@ -25,12 +25,8 @@ public: ...@@ -25,12 +25,8 @@ public:
using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>; using msg_buf = std::array<std::byte, sizeof(intptr_t) + 1>;
enum class code : uint8_t { enum class code : uint8_t {
register_reading,
continue_reading,
register_writing,
continue_writing,
init_manager, init_manager,
discard_manager, dispose_manager,
shutdown_reading, shutdown_reading,
shutdown_writing, shutdown_writing,
run_action, run_action,
...@@ -49,18 +45,15 @@ public: ...@@ -49,18 +45,15 @@ public:
error init(socket_manager* owner, const settings& cfg) override; error init(socket_manager* owner, const settings& cfg) override;
read_result handle_read_event() override; void handle_read_event() override;
read_result handle_buffered_data() override; void handle_write_event() override;
read_result handle_continue_reading() override;
write_result handle_write_event() override;
void abort(const error& reason) override; void abort(const error& reason) override;
private: private:
pipe_socket fd_; pipe_socket fd_;
socket_manager* owner_ = nullptr;
multiplexer* mpx_ = nullptr; multiplexer* mpx_ = nullptr;
msg_buf buf_; msg_buf buf_;
size_t buf_size_ = 0; size_t buf_size_ = 0;
......
...@@ -8,36 +8,35 @@ ...@@ -8,36 +8,35 @@
#include <new> #include <new>
#include "caf/async/producer.hpp" #include "caf/async/producer.hpp"
#include "caf/detail/atomic_ref_counted.hpp"
#include "caf/flow/observer.hpp" #include "caf/flow/observer.hpp"
#include "caf/flow/subscription.hpp" #include "caf/flow/subscription.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/ref_counted.hpp"
namespace caf::net { namespace caf::net {
/// Connects a socket manager to an asynchronous producer resource. /// Connects a socket manager to an asynchronous producer resource.
template <class Buffer> template <class Buffer>
class producer_adapter final : public ref_counted, public async::producer { class producer_adapter final : public detail::atomic_ref_counted,
public async::producer {
public: public:
using atomic_count = std::atomic<size_t>;
using buf_ptr = intrusive_ptr<Buffer>; using buf_ptr = intrusive_ptr<Buffer>;
using value_type = typename Buffer::value_type; using value_type = typename Buffer::value_type;
using ptr_type = intrusive_ptr<producer_adapter>;
void on_consumer_ready() override { void on_consumer_ready() override {
// nop // nop
} }
void on_consumer_cancel() override { void on_consumer_cancel() override {
mgr_->mpx().schedule_fn([adapter = strong_this()] { // mgr_->schedule(do_cancel_);
adapter->on_cancel();
});
} }
void on_consumer_demand(size_t) override { void on_consumer_demand(size_t) override {
mgr_->continue_reading(); mgr_->schedule(do_resume_);
} }
void ref_producer() const noexcept override { void ref_producer() const noexcept override {
...@@ -48,16 +47,13 @@ public: ...@@ -48,16 +47,13 @@ public:
this->deref(); this->deref();
} }
/// Tries to open the resource for writing. static ptr_type make(buf_ptr buf, socket_manager_ptr mgr, action do_resume,
/// @returns a connected adapter that writes to the resource on success, action do_cancel) {
/// `nullptr` otherwise. if (buf) {
template <class Resource> auto adapter
static intrusive_ptr<producer_adapter> = ptr_type{new producer_adapter(buf, mgr, std::move(do_resume),
try_open(socket_manager* owner, Resource src) { std::move(do_cancel)),
CAF_ASSERT(owner != nullptr); false};
if (auto buf = src.try_open()) {
using ptr_type = intrusive_ptr<producer_adapter>;
auto adapter = ptr_type{new producer_adapter(owner, buf), false};
buf->set_producer(adapter); buf->set_producer(adapter);
return adapter; return adapter;
} else { } else {
...@@ -88,47 +84,40 @@ public: ...@@ -88,47 +84,40 @@ public:
void close() { void close() {
if (buf_) { if (buf_) {
buf_->close(); buf_->close();
buf_ = nullptr; reset();
mgr_ = nullptr;
} }
} }
void abort(error reason) { void abort(error reason) {
if (buf_) { if (buf_) {
buf_->abort(std::move(reason)); buf_->abort(std::move(reason));
buf_ = nullptr; reset();
mgr_ = nullptr;
} }
} }
friend void intrusive_ptr_add_ref(const producer_adapter* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const producer_adapter* ptr) noexcept {
ptr->deref();
}
private: private:
producer_adapter(socket_manager* owner, buf_ptr buf) producer_adapter(buf_ptr buf, socket_manager_ptr mgr, action do_resume,
: mgr_(owner), buf_(std::move(buf)) { action do_cancel)
: buf_(std::move(buf)),
mgr_(std::move(mgr)),
do_resume_(std::move(do_resume)),
do_cancel_(std::move(do_cancel)) {
// nop // nop
} }
void on_cancel() { void reset() {
if (buf_) { buf_ = nullptr;
mgr_->mpx().shutdown_reading(mgr_); mgr_ = nullptr;
buf_ = nullptr; do_resume_.dispose();
mgr_ = nullptr; do_resume_ = nullptr;
} do_cancel_.dispose();
} do_cancel_ = nullptr;
auto strong_this() {
return intrusive_ptr{this};
} }
intrusive_ptr<socket_manager> mgr_;
intrusive_ptr<Buffer> buf_; intrusive_ptr<Buffer> buf_;
intrusive_ptr<socket_manager> mgr_;
action do_resume_;
action do_cancel_;
}; };
template <class T> template <class T>
......
// 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/actor_system.hpp"
#include "caf/fwd.hpp"
#include "caf/net/http/server.hpp"
#include "caf/net/prometheus/server.hpp"
#include "caf/net/stream_transport.hpp"
namespace caf::detail {
template <class Transport>
class prometheus_acceptor_factory {
public:
using state_ptr = net::prometheus::server::scrape_state_ptr;
explicit prometheus_acceptor_factory(state_ptr ptr) : ptr_(std::move(ptr)) {
// nop
}
error init(net::socket_manager*, const settings&) {
return none;
}
template <class Socket>
net::socket_manager_ptr make(net::multiplexer* mpx, Socket fd) {
auto prom_serv = net::prometheus::server::make(ptr_);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto transport = Transport::make(fd, std::move(http_serv));
return net::socket_manager::make(mpx, fd, std::move(transport));
}
void abort(const error&) {
// nop
}
private:
state_ptr ptr_;
};
} // namespace caf::detail
namespace caf::net::prometheus {
/// Listens for incoming WebSocket connection on @p fd.
/// @param sys The host system.
/// @param fd An accept socket in listening mode. For a TCP socket, this socket
/// must already listen to a port.
template <class Transport = stream_transport, class Socket>
disposable serve(actor_system& sys, Socket fd) {
using factory_t = detail::prometheus_acceptor_factory<Transport>;
using impl_t = connection_acceptor<Socket, factory_t>;
auto mpx = &sys.network_manager().mpx();
auto registry = &sys.metrics();
auto state = prometheus::server::scrape_state::make(registry);
auto factory = factory_t{std::move(state)};
auto impl = impl_t::make(fd, 0, std::move(factory));
auto mgr = socket_manager::make(mpx, fd, std::move(impl));
mpx->init(mgr);
return mgr->make_disposer();
}
} // namespace caf::net::prometheus
// 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/fwd.hpp"
#include "caf/net/http/fwd.hpp"
#include "caf/net/http/upper_layer.hpp"
#include "caf/telemetry/collector/prometheus.hpp"
#include "caf/telemetry/importer/process.hpp"
#include <chrono>
#include <memory>
#include <string_view>
namespace caf::net::prometheus {
/// Makes metrics available to clients via the Prometheus exposition format.
class server : public http::upper_layer {
public:
// -- member types -----------------------------------------------------------
/// State for scraping Metrics data. Shared between all server instances.
class scrape_state {
public:
using clock_type = std::chrono::steady_clock;
using time_point = clock_type::time_point;
using duration = clock_type::duration;
std::string_view scrape();
explicit scrape_state(telemetry::metric_registry* ptr)
: registry(ptr), last_scrape(duration{0}), proc_importer(*ptr) {
// nop
}
static std::shared_ptr<scrape_state> make(telemetry::metric_registry* ptr) {
return std::make_shared<scrape_state>(ptr);
}
telemetry::metric_registry* registry;
std::chrono::steady_clock::time_point last_scrape;
telemetry::importer::process proc_importer;
telemetry::collector::prometheus collector;
};
using scrape_state_ptr = std::shared_ptr<scrape_state>;
// -- factories --------------------------------------------------------------
static std::unique_ptr<server> make(scrape_state_ptr state) {
return std::unique_ptr<server>{new server(std::move(state))};
}
// -- implementation of http::upper_layer ------------------------------------
bool prepare_send() override;
bool done_sending() override;
void abort(const error& reason) override;
error init(socket_manager* owner, http::lower_layer* down,
const settings& config) override;
ptrdiff_t consume(const http::header& hdr, const_byte_span payload) override;
private:
explicit server(scrape_state_ptr state) : state_(std::move(state)) {
// nop
}
scrape_state_ptr state_;
http::lower_layer* down_ = nullptr;
};
} // namespace caf::net::prometheus
...@@ -16,58 +16,14 @@ class CAF_NET_EXPORT socket_event_layer { ...@@ -16,58 +16,14 @@ class CAF_NET_EXPORT socket_event_layer {
public: public:
virtual ~socket_event_layer(); virtual ~socket_event_layer();
/// Encodes how to proceed after a read operation.
enum class read_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to write to the socket instead of reading
/// from the socket.
want_write,
/// Indicates that the manager no longer reads from the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Encodes how to proceed after a write operation.
enum class write_result {
/// Indicates that a manager wants to read again later.
again,
/// Indicates that a manager wants to stop reading until explicitly resumed.
stop,
/// Indicates that a manager wants to read from the socket instead of
/// writing to the socket.
want_read,
/// Indicates that the manager no longer writes to the socket.
close,
/// Indicates that the manager encountered a fatal error and stops both
/// reading and writing.
abort,
/// Indicates that a manager is done with the socket and hands ownership to
/// another manager.
handover,
};
/// Initializes the layer. /// Initializes the layer.
virtual error init(socket_manager* owner, const settings& cfg) = 0; virtual error init(socket_manager* owner, const settings& cfg) = 0;
/// Handles a read event on the managed socket. /// Handles a read event on the managed socket.
virtual read_result handle_read_event() = 0; virtual void handle_read_event() = 0;
/// Handles internally buffered data.
virtual read_result handle_buffered_data() = 0;
/// Handles a request to continue reading on the socket.
virtual read_result handle_continue_reading() = 0;
/// Handles a write event on the managed socket. /// Handles a write event on the managed socket.
virtual write_result handle_write_event() = 0; virtual void handle_write_event() = 0;
/// Called after returning `handover` from a read or write handler. /// Called after returning `handover` from a read or write handler.
virtual bool do_handover(std::unique_ptr<socket_event_layer>& next); virtual bool do_handover(std::unique_ptr<socket_event_layer>& next);
......
...@@ -4,6 +4,7 @@ ...@@ -4,6 +4,7 @@
#pragma once #pragma once
#include "caf/action.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/callback.hpp" #include "caf/callback.hpp"
...@@ -28,10 +29,6 @@ class CAF_NET_EXPORT socket_manager : public ref_counted { ...@@ -28,10 +29,6 @@ class CAF_NET_EXPORT socket_manager : public ref_counted {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using read_result = socket_event_layer::read_result;
using write_result = socket_event_layer::write_result;
using event_handler_ptr = std::unique_ptr<socket_event_layer>; using event_handler_ptr = std::unique_ptr<socket_event_layer>;
/// Stores manager-related flags in a single block. /// Stores manager-related flags in a single block.
...@@ -77,6 +74,10 @@ public: ...@@ -77,6 +74,10 @@ public:
return make_actor_shell<Handle>(std::move(f)); return make_actor_shell<Handle>(std::move(f));
} }
/// Returns a thread-safe disposer for stopping the socket manager from an
/// outside context.
disposable make_disposer();
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns the handle for the managed socket. /// Returns the handle for the managed socket.
...@@ -122,31 +123,54 @@ public: ...@@ -122,31 +123,54 @@ public:
return flags_.write_closed; return flags_.write_closed;
} }
/// Queries whether the manager is registered for reading.
bool is_reading() const noexcept;
/// Queries whether the manager is registered for writing.
bool is_writing() const noexcept;
// -- event loop management -------------------------------------------------- // -- event loop management --------------------------------------------------
/// Registers the manager for read operations on the @ref multiplexer. /// Registers the manager for read operations.
/// @thread-safe
void register_reading(); void register_reading();
/// Registers the manager for write operations on the @ref multiplexer. /// Registers the manager for write operations.
/// @thread-safe
void register_writing(); void register_writing();
/// Schedules a call to `handle_continue_reading` on the @ref multiplexer. /// Deregisters the manager from read operations.
/// This mechanism allows users to signal changes in the environment to the void deregister_reading();
/// manager that allow it to make progress, e.g., new demand in asynchronous
/// buffer that allow the manager to push available data downstream. The event /// Deregisters the manager from write operations.
/// is a no-op if the manager is already registered for reading. void deregister_writing();
/// @thread-safe
void continue_reading(); /// Deregisters the manager from both read and write operations.
void deregister();
/// Schedules a call to `handle_continue_reading` on the @ref multiplexer.
/// This mechanism allows users to signal changes in the environment to the /// Deregisters the manager from read operations and blocks any future
/// manager that allow it to make progress, e.g., new data for writing in an /// attempts to re-register it.
/// asynchronous buffer. The event is a no-op if the manager is already void shutdown_read();
/// registered for writing.
/// @thread-safe /// Deregisters the manager from write operations and blocks any future
void continue_writing(); /// 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.
void schedule_handover();
/// Schedules @p what to run later.
void schedule(action what);
/// Schedules @p what to run later.
template <class F>
void schedule_fn(F&& what) {
schedule(make_action(std::forward<F>(what)));
}
// -- callbacks for the multiplexer ------------------------------------------ // -- callbacks for the multiplexer ------------------------------------------
...@@ -160,30 +184,20 @@ public: ...@@ -160,30 +184,20 @@ public:
error init(const settings& cfg); error init(const settings& cfg);
/// Called whenever the socket received new data. /// Called whenever the socket received new data.
read_result handle_read_event(); void handle_read_event();
/// Called after handovers to allow the manager to process any data that is
/// already buffered at the transport policy and thus would not trigger a read
/// event on the socket.
read_result handle_buffered_data();
/// Restarts a socket manager that suspended reads. Calling this member
/// function on active managers is a no-op. This function also should read any
/// data buffered outside of the socket.
read_result handle_continue_reading();
/// Called whenever the socket is allowed to send data. /// Called whenever the socket is allowed to send data.
write_result handle_write_event(); void handle_write_event();
/// Called when the remote side becomes unreachable due to an error. /// Called when the remote side becomes unreachable due to an error.
/// @param code The error code as reported by the operating system. /// @param code The error code as reported by the operating system.
void handle_error(sec code); void handle_error(sec code);
/// Performs a handover to another transport after `handle_read_event` or
/// `handle_read_event` returned `handover`.
[[nodiscard]] bool do_handover();
private: private:
// -- utility functions ------------------------------------------------------
socket_manager_ptr strong_this();
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Stores the socket file descriptor. The socket manager automatically closes /// Stores the socket file descriptor. The socket manager automatically closes
......
...@@ -45,6 +45,10 @@ class CAF_NET_EXPORT lower_layer : public generic_lower_layer { ...@@ -45,6 +45,10 @@ class CAF_NET_EXPORT lower_layer : public generic_lower_layer {
public: public:
virtual ~lower_layer(); virtual ~lower_layer();
/// Queries whether the transport is currently configured to read from its
/// socket.
virtual bool is_reading() const noexcept = 0;
/// Configures threshold for the next receive operations. Policies remain /// Configures threshold for the next receive operations. Policies remain
/// active until calling this function again. /// active until calling this function again.
/// @warning Calling this function in `consume` invalidates both byte spans. /// @warning Calling this function in `consume` invalidates both byte spans.
......
...@@ -85,6 +85,9 @@ public: ...@@ -85,6 +85,9 @@ public:
/// Stores whether we left a write handler due to want_read. /// Stores whether we left a write handler due to want_read.
bool wanted_write_from_read_event : 1; bool wanted_write_from_read_event : 1;
/// Stores whether we shutdown write on the socket after flushing.
bool shutting_down : 1;
}; };
// -- constants -------------------------------------------------------------- // -- constants --------------------------------------------------------------
...@@ -118,9 +121,9 @@ public: ...@@ -118,9 +121,9 @@ public:
bool end_output() override; bool end_output() override;
void suspend_reading() override; bool is_reading() const noexcept override;
bool stopped_reading() const noexcept override; void close() override;
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
...@@ -152,17 +155,23 @@ public: ...@@ -152,17 +155,23 @@ public:
error init(socket_manager* owner, const settings& config) override; error init(socket_manager* owner, const settings& config) override;
read_result handle_read_event() override; void handle_read_event() override;
read_result handle_buffered_data() override;
write_result handle_write_event() override; void handle_write_event() override;
read_result handle_continue_reading() override;
void abort(const error& reason) override; void abort(const error& reason) override;
protected: protected:
// -- utility functions ------------------------------------------------------
/// Consumes as much data from the buffer as possible.
void handle_buffered_data();
/// Calls abort on the upper layer and deregisters the transport from events.
void fail(const error& reason);
// -- member variables -------------------------------------------------------
/// The socket file descriptor. /// The socket file descriptor.
stream_socket fd_; stream_socket fd_;
......
...@@ -24,8 +24,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket { ...@@ -24,8 +24,8 @@ struct CAF_NET_EXPORT tcp_accept_socket : network_socket {
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses. /// @param node The endpoint to listen on and the filter for incoming addresses.
/// Passing the address `0.0.0.0` will accept incoming connection from any host. /// Passing the address `0.0.0.0` will accept incoming connection
/// Passing port 0 lets the OS choose the port. /// from any host. Passing port 0 lets the OS choose the port.
/// @relates tcp_accept_socket /// @relates tcp_accept_socket
expected<tcp_accept_socket> expected<tcp_accept_socket>
CAF_NET_EXPORT make_tcp_accept_socket(ip_endpoint node, CAF_NET_EXPORT make_tcp_accept_socket(ip_endpoint node,
...@@ -33,14 +33,24 @@ expected<tcp_accept_socket> ...@@ -33,14 +33,24 @@ expected<tcp_accept_socket>
/// Creates a new TCP socket to accept connections on a given port. /// Creates a new TCP socket to accept connections on a given port.
/// @param node The endpoint to listen on and the filter for incoming addresses. /// @param node The endpoint to listen on and the filter for incoming addresses.
/// Passing the address `0.0.0.0` will accept incoming connection from any host. /// Passing the address `0.0.0.0` will accept incoming connection
/// Passing port 0 lets the OS choose the port. /// from any host. Passing port 0 lets the OS choose the port.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket. /// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates tcp_accept_socket /// @relates tcp_accept_socket
expected<tcp_accept_socket> expected<tcp_accept_socket>
CAF_NET_EXPORT make_tcp_accept_socket(const uri::authority_type& node, CAF_NET_EXPORT make_tcp_accept_socket(const uri::authority_type& node,
bool reuse_addr = false); bool reuse_addr = false);
/// Creates a new TCP socket to accept connections on a given port.
/// @param port The port for listening to incoming connection. Passing 0 lets
/// the OS choose a port.
/// @param addr The filter for incoming addresses. Passing the address `0.0.0.0`
/// will accept incoming connection from any host.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates tcp_accept_socket
expected<tcp_accept_socket> CAF_NET_EXPORT make_tcp_accept_socket(
uint16_t port, std::string addr = "0.0.0.0", bool reuse_addr = false);
/// Accepts a connection on `x`. /// Accepts a connection on `x`.
/// @param x Listening endpoint. /// @param x Listening endpoint.
/// @returns The socket that handles the accepted connection on success, an /// @returns The socket that handles the accepted connection on success, an
......
...@@ -74,8 +74,6 @@ public: ...@@ -74,8 +74,6 @@ public:
ptrdiff_t consume(byte_span buffer, byte_span delta) override; ptrdiff_t consume(byte_span buffer, byte_span delta) override;
void continue_reading() override;
bool prepare_send() override; bool prepare_send() override;
bool done_sending() override; bool done_sending() override;
......
...@@ -79,6 +79,10 @@ public: ...@@ -79,6 +79,10 @@ public:
} }
}; };
bool running() const noexcept {
return in_ || out_;
}
// -- implementation of web_socket::lower_layer ------------------------------ // -- implementation of web_socket::lower_layer ------------------------------
error init(net::socket_manager* mgr, web_socket::lower_layer* down, error init(net::socket_manager* mgr, web_socket::lower_layer* down,
...@@ -86,38 +90,48 @@ public: ...@@ -86,38 +90,48 @@ public:
down_ = down; down_ = down;
auto [err, pull, push] = conn_->on_request(cfg); auto [err, pull, push] = conn_->on_request(cfg);
if (!err) { if (!err) {
in_ = consumer_type::try_open(mgr, pull); auto do_wakeup = make_action([this] {
out_ = producer_type::try_open(mgr, push); prepare_send();
CAF_ASSERT(in_ != nullptr); if (!running())
CAF_ASSERT(out_ != nullptr); down_->close();
});
auto do_resume = make_action([this] { down_->request_messages(); });
auto do_cancel = make_action([this] {
out_ = nullptr;
if (!running())
down_->close();
});
in_ = consumer_type::make(pull.try_open(), mgr, std::move(do_wakeup));
out_ = producer_type::make(push.try_open(), mgr, std::move(do_resume),
std::move(do_cancel));
conn_ = nullptr; conn_ = nullptr;
return none; if (running())
return none;
else
return make_error(sec::runtime_error,
"cannot init flow bridge: no buffers");
} else { } else {
conn_ = nullptr; conn_ = nullptr;
return err; return err;
} }
} }
void continue_reading() override {
// nop
}
bool prepare_send() override { bool prepare_send() override {
write_helper helper{this}; write_helper helper{this};
while (down_->can_send_more() && in_) { while (down_->can_send_more() && in_) {
auto [again, consumed] = in_->pull(async::delay_errors, 1, helper); auto [again, consumed] = in_->pull(async::delay_errors, 1, helper);
if (!again) { if (!again) {
if (helper.err) { if (helper.err) {
down_->send_close_message(helper.err); down_->close(helper.err);
} else { } else {
down_->send_close_message(); down_->close();
} }
in_ = nullptr; in_ = nullptr;
return false; return true;
} else if (helper.aborted) { } else if (helper.aborted) {
in_->cancel(); in_->cancel();
in_ = nullptr; in_ = nullptr;
return false; return true;
} else if (consumed == 0) { } else if (consumed == 0) {
return true; return true;
} }
......
...@@ -77,11 +77,15 @@ public: ...@@ -77,11 +77,15 @@ public:
// -- web_socket::lower_layer implementation --------------------------------- // -- web_socket::lower_layer implementation ---------------------------------
using web_socket::lower_layer::close;
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
void suspend_reading() override; void suspend_reading() override;
bool stopped_reading() const noexcept override; bool is_reading() const noexcept override;
void close(status code, std::string_view desc) override;
void request_messages() override; void request_messages() override;
...@@ -97,17 +101,8 @@ public: ...@@ -97,17 +101,8 @@ public:
bool end_text_message() override; bool end_text_message() override;
void send_close_message() override;
void send_close_message(status code, std::string_view desc) override;
// -- interface for the lower layer ------------------------------------------ // -- interface for the lower layer ------------------------------------------
template <class LowerLayerPtr>
static void continue_reading(LowerLayerPtr down) {
down->configure_read(receive_policy::up_to(2048));
}
ptrdiff_t consume(byte_span input, byte_span); ptrdiff_t consume(byte_span input, byte_span);
private: private:
...@@ -117,10 +112,6 @@ private: ...@@ -117,10 +112,6 @@ private:
void ship_pong(byte_span payload); void ship_pong(byte_span payload);
void ship_close(uint16_t code, std::string_view msg);
void ship_close();
template <class T> template <class T>
void ship_frame(std::vector<T>& buf); void ship_frame(std::vector<T>& buf);
......
...@@ -22,6 +22,9 @@ public: ...@@ -22,6 +22,9 @@ public:
/// Pulls messages from the transport until calling `suspend_reading`. /// Pulls messages from the transport until calling `suspend_reading`.
virtual void request_messages() = 0; virtual void request_messages() = 0;
/// Stops reading messages until calling `request_messages`.
virtual void suspend_reading() = 0;
/// Begins transmission of a binary message. /// Begins transmission of a binary message.
virtual void begin_binary_message() = 0; virtual void begin_binary_message() = 0;
...@@ -42,14 +45,12 @@ public: ...@@ -42,14 +45,12 @@ public:
/// Seals the current text message buffer and ships a new WebSocket frame. /// Seals the current text message buffer and ships a new WebSocket frame.
virtual bool end_text_message() = 0; virtual bool end_text_message() = 0;
/// Sends the close message with @ref status `normal_close`. void close() override;
virtual void send_close_message() = 0;
/// Sends the close message with an custom @ref status. void close(const error& reason) override;
virtual void send_close_message(status code, std::string_view desc) = 0;
/// Sends the close message with @ref status `unexpected_condition`. /// Sends a close message with custom @ref status @p code and @p msg text.
void send_close_message(const error& reason); virtual void close(status code, std::string_view msg) = 0;
}; };
} // namespace caf::net::web_socket } // namespace caf::net::web_socket
...@@ -81,8 +81,6 @@ public: ...@@ -81,8 +81,6 @@ public:
ptrdiff_t consume(byte_span input, byte_span) override; ptrdiff_t consume(byte_span input, byte_span) override;
void continue_reading() override;
bool prepare_send() override; bool prepare_send() override;
bool done_sending() override; bool done_sending() override;
......
...@@ -10,4 +10,8 @@ generic_lower_layer::~generic_lower_layer() { ...@@ -10,4 +10,8 @@ generic_lower_layer::~generic_lower_layer() {
// nop // nop
} }
void generic_lower_layer::close(const error&) {
close();
}
} // namespace caf::net } // namespace caf::net
...@@ -4,7 +4,6 @@ ...@@ -4,7 +4,6 @@
#include "caf/net/http/lower_layer.hpp" #include "caf/net/http/lower_layer.hpp"
#include "caf/net/http/context.hpp"
#include "caf/net/http/header_fields_map.hpp" #include "caf/net/http/header_fields_map.hpp"
#include "caf/net/http/status.hpp" #include "caf/net/http/status.hpp"
...@@ -19,14 +18,18 @@ lower_layer::~lower_layer() { ...@@ -19,14 +18,18 @@ lower_layer::~lower_layer() {
// nop // nop
} }
bool lower_layer::send_response(context ctx, status code, bool lower_layer::send_response(status code, std::string_view content_type,
std::string_view content_type,
const_byte_span content) { const_byte_span content) {
auto content_size = std::to_string(content.size()); auto content_size = std::to_string(content.size());
header_fields_map fields; header_fields_map fields;
fields.emplace("Content-Type"sv, content_type); fields.emplace("Content-Type"sv, content_type);
fields.emplace("Content-Length"sv, content_size); fields.emplace("Content-Length"sv, content_size);
return send_header(ctx, code, fields) && send_payload(ctx, content); return send_header(code, fields) && send_payload(content);
}
bool lower_layer::send_response(status code, std::string_view content_type,
std::string_view content) {
return send_response(code, content_type, as_bytes(make_span(content)));
} }
} // namespace caf::net::http } // namespace caf::net::http
...@@ -14,23 +14,31 @@ bool server::can_send_more() const noexcept { ...@@ -14,23 +14,31 @@ bool server::can_send_more() const noexcept {
return down_->can_send_more(); return down_->can_send_more();
} }
void server::suspend_reading() { bool server::is_reading() const noexcept {
return down_->suspend_reading(); return down_->is_reading();
}
void server::close() {
down_->close();
} }
bool server::stopped_reading() const noexcept { void server::request_messages() {
return down_->stopped_reading(); if (!down_->is_reading())
down_->configure_read(receive_policy::up_to(max_request_size_));
} }
bool server::send_header(context, status code, void server::suspend_reading() {
const header_fields_map& fields) { down_->configure_read(receive_policy::stop());
}
bool server::send_header(status code, const header_fields_map& fields) {
down_->begin_output(); down_->begin_output();
v1::write_header(code, fields, down_->output_buffer()); v1::write_header(code, fields, down_->output_buffer());
down_->end_output(); down_->end_output();
return true; return true;
} }
bool server::send_payload(context, const_byte_span bytes) { bool server::send_payload(const_byte_span bytes) {
down_->begin_output(); down_->begin_output();
auto& buf = down_->output_buffer(); auto& buf = down_->output_buffer();
buf.insert(buf.end(), bytes.begin(), bytes.end()); buf.insert(buf.end(), bytes.begin(), bytes.end());
...@@ -38,7 +46,7 @@ bool server::send_payload(context, const_byte_span bytes) { ...@@ -38,7 +46,7 @@ bool server::send_payload(context, const_byte_span bytes) {
return true; return true;
} }
bool server::send_chunk(context, const_byte_span bytes) { bool server::send_chunk(const_byte_span bytes) {
down_->begin_output(); down_->begin_output();
auto& buf = down_->output_buffer(); auto& buf = down_->output_buffer();
auto size = bytes.size(); auto size = bytes.size();
...@@ -60,20 +68,15 @@ bool server::send_end_of_chunks() { ...@@ -60,20 +68,15 @@ bool server::send_end_of_chunks() {
return down_->end_output(); return down_->end_output();
} }
void server::fin(context) {
// nop
}
// -- stream_oriented::upper_layer implementation ------------------------------ // -- stream_oriented::upper_layer implementation ------------------------------
error server::init(socket_manager* owner, stream_oriented::lower_layer* down, error server::init(socket_manager* owner, stream_oriented::lower_layer* down,
const settings& cfg) { const settings& cfg) {
down_ = down; down_ = down;
if (auto err = up_->init(owner, this, cfg))
return err;
if (auto max_size = get_as<uint32_t>(cfg, "http.max-request-size")) if (auto max_size = get_as<uint32_t>(cfg, "http.max-request-size"))
max_request_size_ = *max_size; max_request_size_ = *max_size;
down_->configure_read(receive_policy::up_to(max_request_size_)); if (auto err = up_->init(owner, this, cfg))
return err;
return none; return none;
} }
...@@ -89,10 +92,6 @@ bool server::done_sending() { ...@@ -89,10 +92,6 @@ bool server::done_sending() {
return up_->done_sending(); return up_->done_sending();
} }
void server::continue_reading() {
down_->configure_read(receive_policy::up_to(max_request_size_));
}
ptrdiff_t server::consume(byte_span input, byte_span) { ptrdiff_t server::consume(byte_span input, byte_span) {
using namespace literals; using namespace literals;
CAF_LOG_TRACE(CAF_ARG2("bytes", input.size())); CAF_LOG_TRACE(CAF_ARG2("bytes", input.size()));
...@@ -174,7 +173,7 @@ void server::write_response(status code, std::string_view content) { ...@@ -174,7 +173,7 @@ void server::write_response(status code, std::string_view content) {
} }
bool server::invoke_upper_layer(const_byte_span payload) { bool server::invoke_upper_layer(const_byte_span payload) {
return up_->consume(context{}, hdr_, payload) >= 0; return up_->consume(hdr_, payload) >= 0;
} }
bool server::handle_header(std::string_view http) { bool server::handle_header(std::string_view http) {
......
...@@ -36,7 +36,7 @@ void length_prefix_framing::abort(const error& reason) { ...@@ -36,7 +36,7 @@ void length_prefix_framing::abort(const error& reason) {
} }
ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) { ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
CAF_LOG_TRACE("got" << input.size() << "bytes"); CAF_LOG_TRACE("got" << input.size() << "bytes\n");
if (input.size() < sizeof(uint32_t)) { if (input.size() < sizeof(uint32_t)) {
CAF_LOG_ERROR("received too few bytes from underlying transport"); CAF_LOG_ERROR("received too few bytes from underlying transport");
up_->abort(make_error(sec::logic_error, up_->abort(make_error(sec::logic_error,
...@@ -65,7 +65,7 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) { ...@@ -65,7 +65,7 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
if (msg_size == msg.size() && msg_size + hdr_size == input.size()) { if (msg_size == msg.size() && msg_size + hdr_size == input.size()) {
CAF_LOG_DEBUG("got message of size" << msg_size); CAF_LOG_DEBUG("got message of size" << msg_size);
if (up_->consume(msg) >= 0) { if (up_->consume(msg) >= 0) {
if (!down_->stopped_reading()) if (down_->is_reading())
down_->configure_read(receive_policy::exactly(hdr_size)); down_->configure_read(receive_policy::exactly(hdr_size));
return static_cast<ptrdiff_t>(input.size()); return static_cast<ptrdiff_t>(input.size());
} else { } else {
...@@ -79,10 +79,6 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) { ...@@ -79,10 +79,6 @@ ptrdiff_t length_prefix_framing::consume(byte_span input, byte_span) {
} }
} }
void length_prefix_framing::continue_reading() {
up_->continue_reading();
}
bool length_prefix_framing::prepare_send() { bool length_prefix_framing::prepare_send() {
return up_->prepare_send(); return up_->prepare_send();
} }
...@@ -98,15 +94,15 @@ bool length_prefix_framing::can_send_more() const noexcept { ...@@ -98,15 +94,15 @@ bool length_prefix_framing::can_send_more() const noexcept {
} }
void length_prefix_framing::suspend_reading() { void length_prefix_framing::suspend_reading() {
down_->suspend_reading(); down_->configure_read(receive_policy::stop());
} }
bool length_prefix_framing::stopped_reading() const noexcept { bool length_prefix_framing::is_reading() const noexcept {
return down_->stopped_reading(); return down_->is_reading();
} }
void length_prefix_framing::request_messages() { void length_prefix_framing::request_messages() {
if (down_->stopped_reading()) if (!down_->is_reading())
down_->configure_read(receive_policy::exactly(hdr_size)); down_->configure_read(receive_policy::exactly(hdr_size));
} }
...@@ -139,12 +135,8 @@ bool length_prefix_framing::end_message() { ...@@ -139,12 +135,8 @@ bool length_prefix_framing::end_message() {
} }
} }
void length_prefix_framing::send_close_message() { void length_prefix_framing::close() {
// nop: this layer has no close message down_->close();
}
void length_prefix_framing::send_close_message(const error&) {
// nop: this layer has no close message
} }
// -- utility functions ------------------------------------------------------ // -- utility functions ------------------------------------------------------
......
...@@ -7,6 +7,9 @@ ...@@ -7,6 +7,9 @@
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp" #include "caf/detail/set_thread_name.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/net/prometheus/serve.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -14,6 +17,40 @@ ...@@ -14,6 +17,40 @@
namespace caf::net { namespace caf::net {
namespace {
struct prom_config {
uint16_t port;
std::string address = "0.0.0.0";
bool reuse_address = false;
};
template <class Inspector>
bool inspect(Inspector& f, prom_config& x) {
return f.object(x).fields(
f.field("port", x.port), f.field("address", x.address).fallback("0.0.0.0"),
f.field("reuse-address", x.reuse_address).fallback(false));
}
void launch_prom_server(actor_system& sys, const prom_config& cfg) {
if (auto fd = make_tcp_accept_socket(cfg.port, cfg.address,
cfg.reuse_address)) {
CAF_LOG_INFO("start Prometheus server at port" << local_port(*fd));
prometheus::serve(sys, std::move(*fd));
} else {
CAF_LOG_WARNING("failed to start Prometheus server: " << fd.error());
}
}
void launch_background_tasks(actor_system& sys) {
auto& cfg = sys.config();
if (auto pcfg = get_as<prom_config>(cfg, "caf.middleman.prometheus-http")) {
launch_prom_server(sys, *pcfg);
}
}
} // namespace
void middleman::init_global_meta_objects() { void middleman::init_global_meta_objects() {
// nop // nop
} }
...@@ -33,6 +70,7 @@ void middleman::start() { ...@@ -33,6 +70,7 @@ void middleman::start() {
detail::set_thread_name("caf.net.mpx"); detail::set_thread_name("caf.net.mpx");
sys_.thread_started(); sys_.thread_started();
mpx_.set_thread_id(); mpx_.set_thread_id();
launch_background_tasks(sys_);
mpx_.run(); mpx_.run();
sys_.thread_terminates(); sys_.thread_terminates();
}}; }};
...@@ -89,6 +127,9 @@ void middleman::add_module_options(actor_system_config& cfg) { ...@@ -89,6 +127,9 @@ void middleman::add_module_options(actor_system_config& cfg) {
"max. time between messages before declaring a node dead " "max. time between messages before declaring a node dead "
"(disabled if 0, ignored if heartbeats are disabled)") "(disabled if 0, ignored if heartbeats are disabled)")
.add<std::string>("network-backend", "legacy option"); .add<std::string>("network-backend", "legacy option");
config_option_adder{cfg.custom_options(), "caf.middleman.prometheus-http"}
.add<uint16_t>("port", "listening port for incoming scrapes")
.add<std::string>("address", "bind address for the HTTP server socket");
} }
} // namespace caf::net } // namespace caf::net
...@@ -131,17 +131,18 @@ size_t multiplexer::num_socket_managers() const noexcept { ...@@ -131,17 +131,18 @@ size_t multiplexer::num_socket_managers() const noexcept {
return managers_.size(); return managers_.size();
} }
ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) { ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) const noexcept {
auto first = managers_.begin(); auto first = managers_.begin();
auto last = managers_.end(); auto last = managers_.end();
auto i = std::find(first, last, mgr); auto i = std::find(first, last, mgr);
return i == last ? -1 : std::distance(first, i); return i == last ? -1 : std::distance(first, i);
} }
ptrdiff_t multiplexer::index_of(socket fd) { ptrdiff_t multiplexer::index_of(socket fd) const noexcept {
auto first = pollset_.begin(); auto first = pollset_.begin();
auto last = pollset_.end(); auto last = pollset_.end();
auto i = std::find_if(first, last, [fd](pollfd& x) { return x.fd == fd.id; }); auto i = std::find_if(first, last,
[fd](const pollfd& x) { return x.fd == fd.id; });
return i == last ? -1 : std::distance(first, i); return i == last ? -1 : std::distance(first, i);
} }
...@@ -166,46 +167,12 @@ operation multiplexer::mask_of(const socket_manager_ptr& mgr) { ...@@ -166,46 +167,12 @@ operation multiplexer::mask_of(const socket_manager_ptr& mgr) {
// -- thread-safe signaling ---------------------------------------------------- // -- thread-safe signaling ----------------------------------------------------
void multiplexer::register_reading(const socket_manager_ptr& mgr) { void multiplexer::dispose(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
do_register_reading(mgr); do_dispose(mgr);
} else { } else {
write_to_pipe(pollset_updater::code::register_reading, mgr.get()); write_to_pipe(pollset_updater::code::dispose_manager, mgr.get());
}
}
void multiplexer::register_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
CAF_ASSERT(mgr != nullptr);
if (std::this_thread::get_id() == tid_) {
do_register_writing(mgr);
} else {
write_to_pipe(pollset_updater::code::register_writing, mgr.get());
}
}
void multiplexer::continue_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
write_to_pipe(pollset_updater::code::continue_reading, mgr.get());
}
void multiplexer::continue_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
CAF_ASSERT(mgr != nullptr);
if (std::this_thread::get_id() == tid_) {
do_continue_writing(mgr);
} else {
write_to_pipe(pollset_updater::code::continue_writing, mgr.get());
}
}
void multiplexer::discard(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (std::this_thread::get_id() == tid_) {
do_discard(mgr);
} else {
write_to_pipe(pollset_updater::code::discard_manager, mgr.get());
} }
} }
...@@ -251,6 +218,41 @@ void multiplexer::shutdown() { ...@@ -251,6 +218,41 @@ void multiplexer::shutdown() {
static_cast<socket_manager*>(nullptr)); static_cast<socket_manager*>(nullptr));
} }
// -- callbacks for socket managers --------------------------------------------
void multiplexer::register_reading(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events |= input_mask;
}
void multiplexer::register_writing(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events |= output_mask;
}
void multiplexer::deregister_reading(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events &= ~input_mask;
}
void multiplexer::deregister_writing(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events &= ~output_mask;
}
void multiplexer::deregister(socket_manager* mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
update_for(mgr).events = 0;
}
bool multiplexer::is_reading(const socket_manager* mgr) const noexcept {
return (active_mask_of(mgr) & input_mask) != 0;
}
bool multiplexer::is_writing(const socket_manager* mgr) const noexcept {
return (active_mask_of(mgr) & output_mask) != 0;
}
// -- control flow ------------------------------------------------------------- // -- control flow -------------------------------------------------------------
bool multiplexer::poll_once(bool blocking) { bool multiplexer::poll_once(bool blocking) {
...@@ -375,44 +377,16 @@ void multiplexer::handle(const socket_manager_ptr& mgr, ...@@ -375,44 +377,16 @@ void multiplexer::handle(const socket_manager_ptr& mgr,
CAF_ASSERT(mgr != nullptr); CAF_ASSERT(mgr != nullptr);
bool checkerror = true; bool checkerror = true;
// Note: we double-check whether the manager is actually reading because a // Note: we double-check whether the manager is actually reading because a
// previous action from the pipe may have called shutdown_reading. // previous action from the pipe may have disabled reading.
if ((events & revents & input_mask) != 0) { if ((revents & input_mask) != 0 && is_reading(mgr.get())) {
checkerror = false; checkerror = false;
switch (mgr->handle_read_event()) { mgr->handle_read_event();
default: // socket_manager::read_result::again
// Nothing to do, bitmask may remain unchanged.
break;
case socket_manager::read_result::abort:
case socket_manager::read_result::stop:
update_for(mgr).events &= ~input_mask;
break;
case socket_manager::read_result::want_write:
update_for(mgr).events = output_mask;
break;
case socket_manager::read_result::handover: {
do_handover(mgr);
return;
}
}
} }
// Similar reasoning than before: double-check whether this event should still // Similar reasoning than before: double-check whether this event should still
// get dispatched. // get dispatched.
if ((events & revents & output_mask) != 0) { if ((revents & output_mask) != 0 && is_writing(mgr.get())) {
checkerror = false; checkerror = false;
switch (mgr->handle_write_event()) { mgr->handle_write_event();
default: // socket_manager::write_result::again
break;
case socket_manager::write_result::abort:
case socket_manager::write_result::stop:
update_for(mgr).events &= ~output_mask;
break;
case socket_manager::write_result::want_read:
update_for(mgr).events = input_mask;
break;
case socket_manager::write_result::handover:
do_handover(mgr);
return;
}
} }
if (checkerror && ((revents & error_mask) != 0)) { if (checkerror && ((revents & error_mask) != 0)) {
if (revents & POLLNVAL) if (revents & POLLNVAL)
...@@ -421,38 +395,7 @@ void multiplexer::handle(const socket_manager_ptr& mgr, ...@@ -421,38 +395,7 @@ void multiplexer::handle(const socket_manager_ptr& mgr,
mgr->handle_error(sec::socket_disconnected); mgr->handle_error(sec::socket_disconnected);
else else
mgr->handle_error(sec::socket_operation_failed); mgr->handle_error(sec::socket_operation_failed);
update_for(mgr).events = 0; update_for(mgr.get()).events = 0;
}
}
void multiplexer::do_handover(const socket_manager_ptr& mgr) {
// Make sure to override the manager pointer in the update. Updates are
// associated to sockets, so the new manager is likely to modify this update
// again. Hence, it *must not* point to the old manager.
auto& update = update_for(mgr);
update.events = 0;
if (mgr->do_handover()) {
// If the new manager registered itself for reading, make sure it processes
// whatever data is available in buffers outside of the socket that may not
// trigger read events.
if ((update.events & input_mask)) {
switch (mgr->handle_buffered_data()) {
default: // socket_manager::read_result::again
// Nothing to do, bitmask may remain unchanged.
break;
case socket_manager::read_result::abort:
case socket_manager::read_result::stop:
update.events &= ~input_mask;
break;
case socket_manager::read_result::want_write:
update.events = output_mask;
break;
case socket_manager::read_result::handover: {
// Down the rabbit hole we go!
do_handover(mgr);
}
}
}
} }
} }
...@@ -467,14 +410,13 @@ multiplexer::poll_update& multiplexer::update_for(ptrdiff_t index) { ...@@ -467,14 +410,13 @@ multiplexer::poll_update& multiplexer::update_for(ptrdiff_t index) {
} }
} }
multiplexer::poll_update& multiplexer::poll_update& multiplexer::update_for(socket_manager* mgr) {
multiplexer::update_for(const socket_manager_ptr& mgr) {
auto fd = mgr->handle(); auto fd = mgr->handle();
if (auto i = updates_.find(fd); i != updates_.end()) { if (auto i = updates_.find(fd); i != updates_.end()) {
return i->second; return i->second;
} else if (auto index = index_of(fd); index != -1) { } else if (auto index = index_of(fd); index != -1) {
updates_.container().emplace_back(fd, updates_.container().emplace_back(fd, poll_update{pollset_[index].events,
poll_update{pollset_[index].events, mgr}); socket_manager_ptr{mgr}});
return updates_.container().back().second; return updates_.container().back().second;
} else { } else {
updates_.container().emplace_back(fd, poll_update{0, mgr}); updates_.container().emplace_back(fd, poll_update{0, mgr});
...@@ -501,7 +443,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, T* ptr) { ...@@ -501,7 +443,7 @@ void multiplexer::write_to_pipe(uint8_t opcode, T* ptr) {
intrusive_ptr_release(ptr); intrusive_ptr_release(ptr);
} }
short multiplexer::active_mask_of(const socket_manager_ptr& mgr) { short multiplexer::active_mask_of(const socket_manager* mgr) const noexcept {
auto fd = mgr->handle(); auto fd = mgr->handle();
if (auto i = updates_.find(fd); i != updates_.end()) { if (auto i = updates_.find(fd); i != updates_.end()) {
return i->second.events; return i->second.events;
...@@ -512,14 +454,6 @@ short multiplexer::active_mask_of(const socket_manager_ptr& mgr) { ...@@ -512,14 +454,6 @@ short multiplexer::active_mask_of(const socket_manager_ptr& mgr) {
} }
} }
bool multiplexer::is_reading(const socket_manager_ptr& mgr) {
return (active_mask_of(mgr) & input_mask) != 0;
}
bool multiplexer::is_writing(const socket_manager_ptr& mgr) {
return (active_mask_of(mgr) & output_mask) != 0;
}
// -- internal callbacks the pollset updater ----------------------------------- // -- internal callbacks the pollset updater -----------------------------------
void multiplexer::do_shutdown() { void multiplexer::do_shutdown() {
...@@ -537,61 +471,17 @@ void multiplexer::do_shutdown() { ...@@ -537,61 +471,17 @@ void multiplexer::do_shutdown() {
apply_updates(); apply_updates();
} }
void multiplexer::do_register_reading(const socket_manager_ptr& mgr) { void multiplexer::do_dispose(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
// When shutting down, no new reads are allowed.
if (shutting_down_)
mgr->close_read();
else if (!mgr->read_closed())
update_for(mgr).events |= input_mask;
}
void multiplexer::do_register_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
// When shutting down, we do allow managers to write whatever is currently
// pending but we make sure that all read channels are closed.
if (shutting_down_)
mgr->close_read();
if (!mgr->write_closed())
update_for(mgr).events |= output_mask;
}
void multiplexer::do_continue_reading(const socket_manager_ptr& mgr) {
if (!is_reading(mgr)) {
switch (mgr->handle_continue_reading()) {
default: // socket_manager::read_result::(stop | abort)
update_for(mgr).events &= ~input_mask;
break;
case socket_manager::read_result::again:
update_for(mgr).events |= input_mask;
break;
case socket_manager::read_result::want_write:
update_for(mgr).events = output_mask;
break;
case socket_manager::read_result::handover: {
do_handover(mgr);
}
}
}
}
void multiplexer::do_continue_writing(const socket_manager_ptr& mgr) {
if (!is_writing(mgr)) {
update_for(mgr).events |= output_mask;
}
}
void multiplexer::do_discard(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
mgr->handle_error(sec::disposed); mgr->handle_error(sec::disposed);
update_for(mgr).events = 0; update_for(mgr.get()).events = 0;
} }
void multiplexer::do_shutdown_reading(const socket_manager_ptr& mgr) { void multiplexer::do_shutdown_reading(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (!shutting_down_ && !mgr->read_closed()) { if (!shutting_down_ && !mgr->read_closed()) {
mgr->close_read(); mgr->close_read();
update_for(mgr).events &= ~input_mask; update_for(mgr.get()).events &= ~input_mask;
} }
} }
...@@ -599,7 +489,7 @@ void multiplexer::do_shutdown_writing(const socket_manager_ptr& mgr) { ...@@ -599,7 +489,7 @@ void multiplexer::do_shutdown_writing(const socket_manager_ptr& mgr) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id)); CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle().id));
if (!shutting_down_ && !mgr->write_closed()) { if (!shutting_down_ && !mgr->write_closed()) {
mgr->close_write(); mgr->close_write();
update_for(mgr).events &= ~output_mask; update_for(mgr.get()).events &= ~output_mask;
} }
} }
...@@ -615,10 +505,8 @@ void multiplexer::do_init(const socket_manager_ptr& mgr) { ...@@ -615,10 +505,8 @@ void multiplexer::do_init(const socket_manager_ptr& mgr) {
CAF_LOG_DEBUG("mgr->init failed: " << err); CAF_LOG_DEBUG("mgr->init failed: " << err);
// The socket manager should not register itself for any events if // The socket manager should not register itself for any events if
// initialization fails. Purge any state just in case. // initialization fails. Purge any state just in case.
update_for(mgr).events = 0; update_for(mgr.get()).events = 0;
} }
// Else: no update since the manager is supposed to call continue_reading
// and continue_writing as necessary.
} }
} }
......
...@@ -31,11 +31,12 @@ std::unique_ptr<pollset_updater> pollset_updater::make(pipe_socket fd) { ...@@ -31,11 +31,12 @@ std::unique_ptr<pollset_updater> pollset_updater::make(pipe_socket fd) {
error pollset_updater::init(socket_manager* owner, const settings&) { error pollset_updater::init(socket_manager* owner, const settings&) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
owner_ = owner;
mpx_ = owner->mpx_ptr(); mpx_ = owner->mpx_ptr();
return nonblocking(fd_, true); return nonblocking(fd_, true);
} }
pollset_updater::read_result pollset_updater::handle_read_event() { void pollset_updater::handle_read_event() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
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};
...@@ -56,23 +57,11 @@ pollset_updater::read_result pollset_updater::handle_read_event() { ...@@ -56,23 +57,11 @@ pollset_updater::read_result pollset_updater::handle_read_event() {
intptr_t ptr; intptr_t ptr;
memcpy(&ptr, buf_.data() + 1, sizeof(intptr_t)); memcpy(&ptr, buf_.data() + 1, sizeof(intptr_t));
switch (static_cast<code>(opcode)) { switch (static_cast<code>(opcode)) {
case code::register_reading:
mpx_->do_register_reading(as_mgr(ptr));
break;
case code::continue_reading:
mpx_->do_continue_reading(as_mgr(ptr));
break;
case code::register_writing:
mpx_->do_register_writing(as_mgr(ptr));
break;
case code::continue_writing:
mpx_->do_continue_writing(as_mgr(ptr));
break;
case code::init_manager: case code::init_manager:
mpx_->do_init(as_mgr(ptr)); mpx_->do_init(as_mgr(ptr));
break; break;
case code::discard_manager: case code::dispose_manager:
mpx_->do_discard(as_mgr(ptr)); mpx_->do_dispose(as_mgr(ptr));
break; break;
case code::shutdown_reading: case code::shutdown_reading:
mpx_->do_shutdown_reading(as_mgr(ptr)); mpx_->do_shutdown_reading(as_mgr(ptr));
...@@ -94,25 +83,20 @@ pollset_updater::read_result pollset_updater::handle_read_event() { ...@@ -94,25 +83,20 @@ pollset_updater::read_result pollset_updater::handle_read_event() {
} }
} else if (num_bytes == 0) { } else if (num_bytes == 0) {
CAF_LOG_DEBUG("pipe closed, assume shutdown"); CAF_LOG_DEBUG("pipe closed, assume shutdown");
return read_result::stop; owner_->shutdown();
return;
} else if (last_socket_error_is_temporary()) { } else if (last_socket_error_is_temporary()) {
return read_result::again; return;
} else { } else {
return read_result::stop; CAF_LOG_DEBUG("pollset updater failed to read from the pipe");
owner_->shutdown();
return;
} }
} }
} }
pollset_updater::read_result pollset_updater::handle_buffered_data() { void pollset_updater::handle_write_event() {
return read_result::again; owner_->deregister_writing();
}
pollset_updater::read_result pollset_updater::handle_continue_reading() {
return read_result::again;
}
pollset_updater::write_result pollset_updater::handle_write_event() {
return write_result::stop;
} }
void pollset_updater::abort(const error&) { void pollset_updater::abort(const error&) {
......
// 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.
#include "caf/net/prometheus/server.hpp"
#include "caf/net/http/header.hpp"
#include "caf/net/http/lower_layer.hpp"
using namespace std::literals;
namespace caf::net::prometheus {
// -- server::scrape_state -----------------------------------------------------
std::string_view server::scrape_state::scrape() {
// Scrape system metrics at most once per second. TODO: make configurable.
if (auto now = std::chrono::steady_clock::now(); last_scrape + 1s <= now) {
last_scrape = now;
proc_importer.update();
}
return collector.collect_from(*registry);
}
// -- implementation of http::upper_layer --------------------------------------
bool server::prepare_send() {
return true;
}
bool server::done_sending() {
return true;
}
void server::abort(const error&) {
// nop
}
error server::init(socket_manager*, http::lower_layer* down, const settings&) {
down_ = down;
down_->request_messages();
return caf::none;
}
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();
} else if (hdr.method() != http::method::get) {
down_->send_response(http::status::method_not_allowed, "text/plain",
"Method not allowed.");
down_->close();
} else if (!hdr.query().empty() || !hdr.fragment().empty()) {
down_->send_response(http::status::bad_request, "text/plain",
"No fragment or query allowed.");
down_->close();
} else {
auto str = state_->scrape();
down_->send_response(http::status::ok, "text/plain;version=0.0.4", str);
down_->close();
}
return static_cast<ptrdiff_t>(payload.size());
}
} // namespace caf::net::prometheus
...@@ -33,6 +33,49 @@ socket_manager_ptr socket_manager::make(multiplexer* mpx, socket handle, ...@@ -33,6 +33,49 @@ socket_manager_ptr socket_manager::make(multiplexer* mpx, socket handle,
return make_counted<socket_manager>(mpx, handle, std::move(handler)); return make_counted<socket_manager>(mpx, handle, std::move(handler));
} }
namespace {
class disposer : public detail::atomic_ref_counted, public disposable_impl {
public:
disposer(multiplexer* mpx, socket_manager_ptr mgr)
: mpx_(mpx), mgr_(std::move(mgr)) {
// nop
}
void dispose() {
std::unique_lock guard{mtx_};
if (mpx_) {
mpx_->dispose(mgr_);
mpx_ = nullptr;
mgr_ = nullptr;
}
}
bool disposed() const noexcept {
std::unique_lock guard{mtx_};
return mpx_ == nullptr;
}
void ref_disposable() const noexcept {
ref();
}
void deref_disposable() const noexcept {
deref();
}
private:
mutable std::mutex mtx_;
multiplexer* mpx_;
socket_manager_ptr mgr_;
};
} // namespace
disposable socket_manager::make_disposer() {
return disposable{make_counted<disposer>(mpx_, this)};
}
// -- properties --------------------------------------------------------------- // -- properties ---------------------------------------------------------------
actor_system& socket_manager::system() noexcept { actor_system& socket_manager::system() noexcept {
...@@ -40,22 +83,73 @@ actor_system& socket_manager::system() noexcept { ...@@ -40,22 +83,73 @@ actor_system& socket_manager::system() noexcept {
return mpx_->system(); return mpx_->system();
} }
bool socket_manager::is_reading() const noexcept {
return mpx_->is_reading(this);
}
bool socket_manager::is_writing() const noexcept {
return mpx_->is_writing(this);
}
// -- event loop management ---------------------------------------------------- // -- event loop management ----------------------------------------------------
void socket_manager::register_reading() { void socket_manager::register_reading() {
mpx_->register_reading(this); if (!flags_.read_closed)
mpx_->register_reading(this);
} }
void socket_manager::continue_reading() { void socket_manager::register_writing() {
mpx_->continue_reading(this); if (!flags_.write_closed)
mpx_->register_writing(this);
} }
void socket_manager::register_writing() { void socket_manager::deregister_reading() {
mpx_->register_writing(this); mpx_->deregister_reading(this);
}
void socket_manager::deregister_writing() {
mpx_->deregister_writing(this);
} }
void socket_manager::continue_writing() { void socket_manager::deregister() {
mpx_->continue_writing(this); 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;
deregister();
}
// -- callbacks for the handler ------------------------------------------------
void socket_manager::schedule_handover() {
deregister();
mpx_->schedule_fn([ptr = strong_this()] { //
event_handler_ptr next;
if (ptr->handler_->do_handover(next)) {
ptr->handler_.swap(next);
}
});
}
void socket_manager::schedule(action what) {
// Wrap the action to make sure the socket manager is still alive when running
// the action later.
mpx_->schedule_fn([ptr = strong_this(), f = std::move(what)]() mutable {
CAF_IGNORE_UNUSED(ptr);
f.run();
});
} }
// -- callbacks for the multiplexer -------------------------------------------- // -- callbacks for the multiplexer --------------------------------------------
...@@ -70,16 +164,6 @@ void socket_manager::close_write() noexcept { ...@@ -70,16 +164,6 @@ void socket_manager::close_write() noexcept {
flags_.write_closed = true; flags_.write_closed = true;
} }
bool socket_manager::do_handover() {
event_handler_ptr next;
if (handler_->do_handover(next)) {
handler_.swap(next);
return true;
} else {
return false;
}
}
error socket_manager::init(const settings& cfg) { error socket_manager::init(const settings& cfg) {
CAF_LOG_TRACE(CAF_ARG(cfg)); CAF_LOG_TRACE(CAF_ARG(cfg));
if (auto err = nonblocking(fd_, true)) { if (auto err = nonblocking(fd_, true)) {
...@@ -89,32 +173,12 @@ error socket_manager::init(const settings& cfg) { ...@@ -89,32 +173,12 @@ error socket_manager::init(const settings& cfg) {
return handler_->init(this, cfg); return handler_->init(this, cfg);
} }
socket_manager::read_result socket_manager::handle_read_event() { void socket_manager::handle_read_event() {
auto result = handler_->handle_read_event(); handler_->handle_read_event();
switch (result) {
default:
break;
case read_result::close:
flags_.read_closed = true;
break;
case read_result::abort:
flags_.read_closed = true;
flags_.write_closed = true;
break;
}
return result;
}
socket_manager::read_result socket_manager::handle_buffered_data() {
return handler_->handle_buffered_data();
}
socket_manager::read_result socket_manager::handle_continue_reading() {
return handler_->handle_continue_reading();
} }
socket_manager::write_result socket_manager::handle_write_event() { void socket_manager::handle_write_event() {
return handler_->handle_write_event(); handler_->handle_write_event();
} }
void socket_manager::handle_error(sec code) { void socket_manager::handle_error(sec code) {
...@@ -124,4 +188,10 @@ void socket_manager::handle_error(sec code) { ...@@ -124,4 +188,10 @@ void socket_manager::handle_error(sec code) {
} }
} }
// -- utility functions --------------------------------------------------------
socket_manager_ptr socket_manager::strong_this() {
return socket_manager_ptr{this};
}
} // namespace caf::net } // namespace caf::net
...@@ -26,10 +26,6 @@ public: ...@@ -26,10 +26,6 @@ public:
using super = socket_event_layer; using super = socket_event_layer;
using read_result = typename super::read_result;
using write_result = typename super::write_result;
using upper_layer_ptr = transport::upper_layer_ptr; using upper_layer_ptr = transport::upper_layer_ptr;
handshake_worker(connection conn, bool is_server, upper_layer_ptr up) handshake_worker(connection conn, bool is_server, upper_layer_ptr up)
...@@ -46,55 +42,56 @@ public: ...@@ -46,55 +42,56 @@ public:
return caf::none; return caf::none;
} }
read_result handle_read_event() override { void handle_read_event() override {
if (auto res = advance_handshake(); res > 0) { if (auto res = advance_handshake(); res > 0) {
return read_result::handover; owner_->deregister();
owner_->schedule_handover();
} else if (res == 0) { } else if (res == 0) {
up_->abort(make_error(sec::connection_closed)); up_->abort(make_error(sec::connection_closed));
return read_result::stop; owner_->deregister();
} else { } else {
auto err = policy_.last_error(policy_.conn.fd(), res); auto err = policy_.last_error(policy_.conn.fd(), res);
switch (err) { switch (err) {
case stream_transport_error::want_read: case stream_transport_error::want_read:
case stream_transport_error::temporary: case stream_transport_error::temporary:
return read_result::again; break;
case stream_transport_error::want_write: case stream_transport_error::want_write:
return read_result::want_write; owner_->deregister_reading();
default: owner_->register_writing();
break;
default: {
auto err = make_error(sec::cannot_connect_to_node, auto err = make_error(sec::cannot_connect_to_node,
policy_.conn.last_error_string(res)); policy_.conn.last_error_string(res));
up_->abort(std::move(err)); up_->abort(std::move(err));
return read_result::stop; owner_->deregister();
}
} }
} }
} }
read_result handle_buffered_data() override { void handle_write_event() override {
return read_result::again;
}
read_result handle_continue_reading() override {
return read_result::again;
}
write_result handle_write_event() override {
if (auto res = advance_handshake(); res > 0) { if (auto res = advance_handshake(); res > 0) {
return write_result::handover; owner_->deregister();
owner_->schedule_handover();
return;
} else if (res == 0) { } else if (res == 0) {
up_->abort(make_error(sec::connection_closed)); up_->abort(make_error(sec::connection_closed));
return write_result::stop; owner_->deregister();
} else { } else {
switch (policy_.last_error(policy_.conn.fd(), res)) { switch (policy_.last_error(policy_.conn.fd(), res)) {
case stream_transport_error::want_write: case stream_transport_error::want_write:
case stream_transport_error::temporary: case stream_transport_error::temporary:
return write_result::again; break;
case stream_transport_error::want_read: case stream_transport_error::want_read:
return write_result::want_read; owner_->deregister_writing();
default: owner_->register_reading();
break;
default: {
auto err = make_error(sec::cannot_connect_to_node, auto err = make_error(sec::cannot_connect_to_node,
policy_.conn.last_error_string(res)); policy_.conn.last_error_string(res));
up_->abort(std::move(err)); up_->abort(std::move(err));
return write_result::stop; owner_->deregister();
}
} }
} }
} }
......
...@@ -49,14 +49,14 @@ size_t stream_transport::default_policy::buffered() const noexcept { ...@@ -49,14 +49,14 @@ size_t stream_transport::default_policy::buffered() const noexcept {
// -- constructors, destructors, and assignment operators ---------------------- // -- constructors, destructors, and assignment operators ----------------------
stream_transport::stream_transport(stream_socket fd, upper_layer_ptr up) stream_transport::stream_transport(stream_socket fd, upper_layer_ptr up)
: fd_(fd), up_(std::move(up)) { : fd_(fd), up_(std::move(up)), policy_(&default_policy_) {
policy_ = &default_policy_; memset(&flags_, 0, sizeof(flags_t));
} }
stream_transport::stream_transport(stream_socket fd, upper_layer_ptr up, stream_transport::stream_transport(stream_socket fd, upper_layer_ptr up,
policy* policy) policy* policy)
: fd_(fd), up_(std::move(up)), policy_(policy) { : fd_(fd), up_(std::move(up)), policy_(policy) {
// nop memset(&flags_, 0, sizeof(flags_t));
} }
// -- factories ---------------------------------------------------------------- // -- factories ----------------------------------------------------------------
...@@ -73,11 +73,25 @@ bool stream_transport::can_send_more() const noexcept { ...@@ -73,11 +73,25 @@ bool stream_transport::can_send_more() const noexcept {
} }
void stream_transport::configure_read(receive_policy rd) { void stream_transport::configure_read(receive_policy rd) {
if (rd.max_size > 0 && max_read_size_ == 0) { auto restarting = rd.max_size > 0 && max_read_size_ == 0;
parent_->register_reading();
}
min_read_size_ = rd.min_size; min_read_size_ = rd.min_size;
max_read_size_ = rd.max_size; max_read_size_ = rd.max_size;
if (restarting && !parent_->is_reading()) {
if (buffered_ >= min_read_size_) {
// 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.
parent_->schedule_fn([this] {
parent_->register_reading();
handle_buffered_data();
});
} else {
// Simply ask for more data.
parent_->register_reading();
}
} else if (max_read_size_ == 0) {
parent_->deregister_reading();
}
} }
void stream_transport::begin_output() { void stream_transport::begin_output() {
...@@ -93,12 +107,16 @@ bool stream_transport::end_output() { ...@@ -93,12 +107,16 @@ bool stream_transport::end_output() {
return true; return true;
} }
void stream_transport::suspend_reading() { bool stream_transport::is_reading() const noexcept {
configure_read(receive_policy::stop()); return max_read_size_ > 0;
} }
bool stream_transport::stopped_reading() const noexcept { void stream_transport::close() {
return max_read_size_ == 0; parent_->shutdown_read();
if (write_buf_.empty())
parent_->shutdown_write();
else
flags_.shutting_down = true;
} }
// -- implementation of transport ---------------------------------------------- // -- implementation of transport ----------------------------------------------
...@@ -124,32 +142,28 @@ error stream_transport::init(socket_manager* owner, const settings& config) { ...@@ -124,32 +142,28 @@ error stream_transport::init(socket_manager* owner, const settings& config) {
return up_->init(owner, this, config); return up_->init(owner, this, config);
} }
stream_transport::read_result stream_transport::handle_read_event() { void stream_transport::handle_read_event() {
CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id)); CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id));
// Convenience lambda for failing the application.
auto fail = [this](error reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
up_->abort(reason);
return read_result::abort;
};
// Resume a write operation if the transport waited for the socket to be // Resume a write operation if the transport waited for the socket to be
// readable from the last call to handle_write_event. // readable from the last call to handle_write_event.
if (flags_.wanted_read_from_write_event) { if (flags_.wanted_read_from_write_event) {
flags_.wanted_read_from_write_event = false; flags_.wanted_read_from_write_event = false;
switch (handle_write_event()) { // The subsequent call to handle_write_event expects a writing manager.
case write_result::want_read: parent_->register_writing();
CAF_ASSERT(flags_.wanted_read_from_write_event); handle_write_event();
return read_result::again; if (!parent_->is_reading()) {
case write_result::handover: // The call to handle_write_event deregisters the manager from reading in
return read_result::handover; // case of an error. So we need to double-check that flag here.
case write_result::again: return;
parent_->register_writing(); }
break; // Check if we have actually some reading to do.
default: if (max_read_size_ == 0) {
break; if (!flags_.wanted_read_from_write_event)
parent_->deregister_reading();
return;
} }
} }
// Make sure the buffer is large enough. // Make sure our read buffer is large enough.
if (read_buf_.size() < max_read_size_) if (read_buf_.size() < max_read_size_)
read_buf_.resize(max_read_size_); read_buf_.resize(max_read_size_);
// Fill up our buffer. // Fill up our buffer.
...@@ -160,19 +174,23 @@ stream_transport::read_result stream_transport::handle_read_event() { ...@@ -160,19 +174,23 @@ stream_transport::read_result stream_transport::handle_read_event() {
switch (policy_->last_error(fd_, rd)) { switch (policy_->last_error(fd_, rd)) {
case stream_transport_error::temporary: case stream_transport_error::temporary:
case stream_transport_error::want_read: case stream_transport_error::want_read:
return read_result::again; // Try again later.
return;
case stream_transport_error::want_write: case stream_transport_error::want_write:
// Wait for writable socket and then call handle_read_event again.
flags_.wanted_write_from_read_event = true; flags_.wanted_write_from_read_event = true;
return read_result::want_write; parent_->register_writing();
parent_->deregister_reading();
return;
default: default:
return fail(sec::socket_operation_failed); return fail(make_error(sec::socket_operation_failed));
} }
} else if (rd == 0) { } else if (rd == 0) {
return fail(sec::socket_disconnected); return fail(make_error(sec::socket_disconnected));
} }
// Make sure we actually have all data currently available to us and the // Make sure we actually have all data currently available to us and the
// policy is not holding on to some bytes. This may happen when using // policy is not holding on to some bytes. This may happen when using
// OpenSSL or any other transport policy operating on blocks. // OpenSSL or any other transport policy that operates on blocks.
buffered_ += static_cast<size_t>(rd); buffered_ += static_cast<size_t>(rd);
if (auto policy_buffered = policy_->buffered(); policy_buffered > 0) { if (auto policy_buffered = policy_->buffered(); policy_buffered > 0) {
if (auto n = buffered_ + policy_buffered; n > read_buf_.size()) if (auto n = buffered_ + policy_buffered; n > read_buf_.size())
...@@ -181,23 +199,16 @@ stream_transport::read_result stream_transport::handle_read_event() { ...@@ -181,23 +199,16 @@ stream_transport::read_result stream_transport::handle_read_event() {
policy_buffered)); policy_buffered));
if (rd2 != static_cast<ptrdiff_t>(policy_buffered)) { if (rd2 != static_cast<ptrdiff_t>(policy_buffered)) {
CAF_LOG_ERROR("failed to read buffered data from the policy"); CAF_LOG_ERROR("failed to read buffered data from the policy");
return fail(sec::socket_operation_failed); return fail(make_error(sec::socket_operation_failed));
} }
buffered_ += static_cast<size_t>(rd2); buffered_ += static_cast<size_t>(rd2);
} }
// Try to make progress on the application and return control to the // Read buffered data and then allow other sockets to run.
// multiplexer to allow other sockets to run. handle_buffered_data();
return handle_buffered_data();
} }
stream_transport::read_result stream_transport::handle_buffered_data() { void stream_transport::handle_buffered_data() {
CAF_LOG_TRACE(CAF_ARG(buffered_)); CAF_LOG_TRACE(CAF_ARG(buffered_));
// Convenience lambda for failing the application.
auto fail = [this](error reason) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(reason));
up_->abort(reason);
return read_result::abort;
};
// Loop until we have drained the buffer as much as we can. // Loop until we have drained the buffer as much as we can.
CAF_ASSERT(min_read_size_ <= max_read_size_); CAF_ASSERT(min_read_size_ <= max_read_size_);
while (max_read_size_ > 0 && buffered_ >= min_read_size_) { while (max_read_size_ > 0 && buffered_ >= min_read_size_) {
...@@ -210,7 +221,8 @@ stream_transport::read_result stream_transport::handle_buffered_data() { ...@@ -210,7 +221,8 @@ stream_transport::read_result stream_transport::handle_buffered_data() {
// Negative values indicate that the application encountered an // Negative values indicate that the application encountered an
// unrecoverable error. // unrecoverable error.
up_->abort(make_error(caf::sec::runtime_error, "consumed < 0")); up_->abort(make_error(caf::sec::runtime_error, "consumed < 0"));
return read_result::abort; parent_->deregister();
return;
} else if (consumed == 0) { } else if (consumed == 0) {
// Returning 0 means that the application wants more data. Note: // Returning 0 means that the application wants more data. Note:
// max_read_size_ may have changed if the application realized it // max_read_size_ may have changed if the application realized it
...@@ -219,18 +231,19 @@ stream_transport::read_result stream_transport::handle_buffered_data() { ...@@ -219,18 +231,19 @@ stream_transport::read_result stream_transport::handle_buffered_data() {
delta_offset_ = static_cast<ptrdiff_t>(n); delta_offset_ = static_cast<ptrdiff_t>(n);
if (n == max_read_size_ || max_read_size_ < old_max_read_size) { if (n == max_read_size_ || max_read_size_ < old_max_read_size) {
CAF_LOG_ERROR("application failed to make progress"); CAF_LOG_ERROR("application failed to make progress");
return fail(sec::runtime_error); return fail(make_error(sec::runtime_error));
} else if (n == buffered_) { } else if (n == buffered_) {
// Either the application has increased max_read_size_ or we // Either the application has increased max_read_size_ or we
// did not reach max_read_size_ the first time. In both cases, we // did not reach max_read_size_ the first time. In both cases, we
// cannot proceed without receiving more data. // cannot proceed without receiving more data.
return read_result::again; return;
} }
} else if (static_cast<size_t>(consumed) > n) { } else if (static_cast<size_t>(consumed) > n) {
// Must not happen. An application cannot handle more data then we pass // Must not happen. An application cannot handle more data then we pass
// to it. // to it.
up_->abort(make_error(sec::logic_error, "consumed > buffer.size")); up_->abort(make_error(sec::logic_error, "consumed > buffer.size"));
return read_result::abort; parent_->deregister();
return;
} else { } else {
// Shove the unread bytes to the beginning of the buffer and continue // Shove the unread bytes to the beginning of the buffer and continue
// to the next loop iteration. // to the next loop iteration.
...@@ -245,75 +258,74 @@ stream_transport::read_result stream_transport::handle_buffered_data() { ...@@ -245,75 +258,74 @@ stream_transport::read_result stream_transport::handle_buffered_data() {
} }
} }
} }
return max_read_size_ > 0 ? read_result::again : read_result::stop; if (max_read_size_ == 0)
parent_->deregister_reading();
}
void stream_transport::fail(const error& reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
up_->abort(reason);
parent_->deregister();
} }
stream_transport::write_result stream_transport::handle_write_event() { void stream_transport::handle_write_event() {
CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id)); CAF_LOG_TRACE(CAF_ARG2("socket", fd_.id));
auto fail = [this](sec code) {
CAF_LOG_DEBUG("read failed" << CAF_ARG(code));
up_->abort(make_error(code));
return write_result::abort;
};
// Resume a read operation if the transport waited for the socket to be // Resume a read operation if the transport waited for the socket to be
// writable from the last call to handle_read_event. // writable from the last call to handle_read_event.
if (flags_.wanted_write_from_read_event) { if (flags_.wanted_write_from_read_event) {
flags_.wanted_write_from_read_event = false; flags_.wanted_write_from_read_event = false;
switch (handle_read_event()) { // The subsequent call to handle_read_event expects a writing manager.
case read_result::want_write: parent_->register_writing();
CAF_ASSERT(flags_.wanted_write_from_read_event); handle_read_event();
return write_result::again; if (!parent_->is_writing()) {
case read_result::handover: // The call to handle_read_event deregisters the manager from writing in
return write_result::handover; // case of an error. So we need to double-check that flag here.
case read_result::again: return;
parent_->register_reading();
break;
default:
break;
} }
// Fall though and see if we also have something to write.
} }
// Allow the upper layer to add extra data to the write buffer. // Allow the upper layer to add extra data to the write buffer.
if (!up_->prepare_send()) { if (!up_->prepare_send()) {
up_->abort(make_error(caf::sec::runtime_error, "prepare_send failed")); CAF_LOG_DEBUG("prepare_send failed");
return write_result::abort; parent_->deregister();
return;
}
if (write_buf_.empty()) {
if (up_->done_sending()) {
if (!flags_.shutting_down)
parent_->deregister_writing();
else
parent_->shutdown_write();
}
return;
} }
if (write_buf_.empty())
return !up_->done_sending() ? write_result::again : write_result::stop;
auto write_res = policy_->write(fd_, write_buf_); auto write_res = policy_->write(fd_, write_buf_);
if (write_res > 0) { if (write_res > 0) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + write_res); write_buf_.erase(write_buf_.begin(), write_buf_.begin() + write_res);
return !write_buf_.empty() || !up_->done_sending() ? write_result::again if (write_buf_.empty() && up_->done_sending()) {
: write_result::stop; if (!flags_.shutting_down)
parent_->deregister_writing();
else
parent_->shutdown_write();
}
return;
} else if (write_res < 0) { } else if (write_res < 0) {
// Try again later on temporary errors such as EWOULDBLOCK and // Try again later on temporary errors such as EWOULDBLOCK and
// stop writing to the socket on hard errors. // stop writing to the socket on hard errors.
switch (policy_->last_error(fd_, write_res)) { switch (policy_->last_error(fd_, write_res)) {
case stream_transport_error::temporary: case stream_transport_error::temporary:
case stream_transport_error::want_write: case stream_transport_error::want_write:
return write_result::again; return;
case stream_transport_error::want_read: case stream_transport_error::want_read:
flags_.wanted_read_from_write_event = true; flags_.wanted_read_from_write_event = true;
return write_result::want_read; parent_->register_reading();
parent_->deregister_writing();
return;
default: default:
return fail(sec::socket_operation_failed); return fail(make_error(sec::socket_operation_failed));
} }
} else { } else {
// write() returns 0 if the connection was closed. // write() returns 0 if the connection was closed.
return fail(sec::socket_disconnected); return fail(make_error(sec::socket_disconnected));
}
}
stream_transport::read_result stream_transport::handle_continue_reading() {
if (max_read_size_ == 0) {
up_->continue_reading();
if (max_read_size_ != 0) {
return handle_buffered_data();
} else {
return read_result::stop;
}
} else {
return handle_buffered_data();
} }
} }
......
...@@ -121,6 +121,14 @@ make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) { ...@@ -121,6 +121,14 @@ make_tcp_accept_socket(const uri::authority_type& node, bool reuse_addr) {
to_string(node)); to_string(node));
} }
expected<tcp_accept_socket>
make_tcp_accept_socket(uint16_t port, std::string addr, bool reuse_addr) {
uri::authority_type auth;
auth.port = port;
auth.host = std::move(addr);
return make_tcp_accept_socket(auth, reuse_addr);
}
expected<tcp_stream_socket> accept(tcp_accept_socket x) { expected<tcp_stream_socket> accept(tcp_accept_socket x) {
auto sock = ::accept(x.id, nullptr, nullptr); auto sock = ::accept(x.id, nullptr, nullptr);
if (sock == net::invalid_socket_id) { if (sock == net::invalid_socket_id) {
......
...@@ -95,11 +95,6 @@ ptrdiff_t client::consume(byte_span buffer, byte_span delta) { ...@@ -95,11 +95,6 @@ ptrdiff_t client::consume(byte_span buffer, byte_span delta) {
} }
} }
void client::continue_reading() {
if (handshake_completed())
upper_layer().continue_reading();
}
bool client::prepare_send() { bool client::prepare_send() {
return handshake_completed() ? upper_layer().prepare_send() : true; return handshake_completed() ? upper_layer().prepare_send() : true;
} }
......
...@@ -23,15 +23,34 @@ bool framing::can_send_more() const noexcept { ...@@ -23,15 +23,34 @@ bool framing::can_send_more() const noexcept {
} }
void framing::suspend_reading() { void framing::suspend_reading() {
down_->suspend_reading(); down_->configure_read(receive_policy::stop());
} }
bool framing::stopped_reading() const noexcept { bool framing::is_reading() const noexcept {
return down_->stopped_reading(); return down_->is_reading();
}
void framing::close(status code, std::string_view msg) {
auto code_val = static_cast<uint16_t>(code);
uint32_t mask_key = 0;
byte_buffer payload;
payload.reserve(msg.size() + 2);
payload.push_back(static_cast<std::byte>((code_val & 0xFF00) >> 8));
payload.push_back(static_cast<std::byte>(code_val & 0x00FF));
for (auto c : msg)
payload.push_back(static_cast<std::byte>(c));
if (mask_outgoing_frames) {
mask_key = static_cast<uint32_t>(rng_());
detail::rfc6455::mask_data(mask_key, payload);
}
down_->begin_output();
detail::rfc6455::assemble_frame(detail::rfc6455::connection_close, mask_key,
payload, down_->output_buffer());
down_->end_output();
} }
void framing::request_messages() { void framing::request_messages() {
if (down_->stopped_reading()) if (!down_->is_reading())
down_->configure_read(receive_policy::up_to(2048)); down_->configure_read(receive_policy::up_to(2048));
} }
...@@ -61,14 +80,6 @@ bool framing::end_text_message() { ...@@ -61,14 +80,6 @@ bool framing::end_text_message() {
return true; return true;
} }
void framing::send_close_message() {
ship_close();
}
void framing::send_close_message(status code, std::string_view desc) {
ship_close(static_cast<uint16_t>(code), desc);
}
// -- interface for the lower layer -------------------------------------------- // -- interface for the lower layer --------------------------------------------
ptrdiff_t framing::consume(byte_span input, byte_span) { ptrdiff_t framing::consume(byte_span input, byte_span) {
...@@ -99,9 +110,6 @@ ptrdiff_t framing::consume(byte_span input, byte_span) { ...@@ -99,9 +110,6 @@ ptrdiff_t framing::consume(byte_span input, byte_span) {
// Wait for more data if necessary. // Wait for more data if necessary.
size_t frame_size = hdr_bytes + hdr.payload_len; size_t frame_size = hdr_bytes + hdr.payload_len;
if (buffer.size() < frame_size) { if (buffer.size() < frame_size) {
// Ask for more data unless the upper layer called suspend_reading.
if (!down_->stopped_reading())
down_->configure_read(receive_policy::up_to(2048));
down_->configure_read(receive_policy::exactly(frame_size)); down_->configure_read(receive_policy::exactly(frame_size));
return consumed; return consumed;
} }
...@@ -170,7 +178,7 @@ ptrdiff_t framing::consume(byte_span input, byte_span) { ...@@ -170,7 +178,7 @@ ptrdiff_t framing::consume(byte_span input, byte_span) {
buffer = buffer.subspan(frame_size); buffer = buffer.subspan(frame_size);
if (buffer.empty()) { if (buffer.empty()) {
// Ask for more data unless the upper layer called suspend_reading. // Ask for more data unless the upper layer called suspend_reading.
if (!down_->stopped_reading()) if (down_->is_reading())
down_->configure_read(receive_policy::up_to(2048)); down_->configure_read(receive_policy::up_to(2048));
return consumed + static_cast<ptrdiff_t>(frame_size); return consumed + static_cast<ptrdiff_t>(frame_size);
} }
...@@ -215,40 +223,6 @@ void framing::ship_pong(byte_span payload) { ...@@ -215,40 +223,6 @@ void framing::ship_pong(byte_span payload) {
down_->end_output(); down_->end_output();
} }
void framing::ship_close(uint16_t code, std::string_view msg) {
uint32_t mask_key = 0;
byte_buffer payload;
payload.reserve(msg.size() + 2);
payload.push_back(static_cast<std::byte>((code & 0xFF00) >> 8));
payload.push_back(static_cast<std::byte>(code & 0x00FF));
for (auto c : msg)
payload.push_back(static_cast<std::byte>(c));
if (mask_outgoing_frames) {
mask_key = static_cast<uint32_t>(rng_());
detail::rfc6455::mask_data(mask_key, payload);
}
down_->begin_output();
detail::rfc6455::assemble_frame(detail::rfc6455::connection_close, mask_key,
payload, down_->output_buffer());
down_->end_output();
}
void framing::ship_close() {
uint32_t mask_key = 0;
std::byte payload[] = {// Error code 1000: normal close.
std::byte{0x03}, std::byte{0xE8},
// "EOF" string as goodbye message.
std::byte{'E'}, std::byte{'O'}, std::byte{'F'}};
if (mask_outgoing_frames) {
mask_key = static_cast<uint32_t>(rng_());
detail::rfc6455::mask_data(mask_key, payload);
}
down_->begin_output();
detail::rfc6455::assemble_frame(detail::rfc6455::connection_close, mask_key,
payload, down_->output_buffer());
down_->end_output();
}
template <class T> template <class T>
void framing::ship_frame(std::vector<T>& buf) { void framing::ship_frame(std::vector<T>& buf) {
uint32_t mask_key = 0; uint32_t mask_key = 0;
......
...@@ -13,8 +13,13 @@ lower_layer::~lower_layer() { ...@@ -13,8 +13,13 @@ lower_layer::~lower_layer() {
// nop // nop
} }
void lower_layer::send_close_message(const error& reason) { void lower_layer::close() {
send_close_message(status::unexpected_condition, to_string(reason)); using namespace std::literals;
close(status::normal_close, "EOF"sv);
}
void lower_layer::close(const error& reason) {
close(status::unexpected_condition, to_string(reason));
} }
} // namespace caf::net::web_socket } // namespace caf::net::web_socket
...@@ -74,11 +74,6 @@ ptrdiff_t server::consume(byte_span input, byte_span delta) { ...@@ -74,11 +74,6 @@ ptrdiff_t server::consume(byte_span input, byte_span delta) {
} }
} }
void server::continue_reading() {
auto rp = receive_policy::up_to(handshake::max_http_size);
lower_layer().configure_read(rp);
}
bool server::prepare_send() { bool server::prepare_send() {
return handshake_complete_ ? upper_layer().prepare_send() : true; return handshake_complete_ ? upper_layer().prepare_send() : true;
} }
......
...@@ -19,8 +19,12 @@ bool mock_stream_transport::can_send_more() const noexcept { ...@@ -19,8 +19,12 @@ bool mock_stream_transport::can_send_more() const noexcept {
return true; return true;
} }
void mock_stream_transport::suspend_reading() { bool mock_stream_transport::is_reading() const noexcept {
configure_read(net::receive_policy::stop()); return max_read_size > 0;
}
void mock_stream_transport::close() {
// nop
} }
void mock_stream_transport::configure_read(net::receive_policy policy) { void mock_stream_transport::configure_read(net::receive_policy policy) {
...@@ -40,10 +44,6 @@ bool mock_stream_transport::end_output() { ...@@ -40,10 +44,6 @@ bool mock_stream_transport::end_output() {
return true; return true;
} }
bool mock_stream_transport::stopped_reading() const noexcept {
return max_read_size == 0;
}
ptrdiff_t mock_stream_transport::handle_input() { ptrdiff_t mock_stream_transport::handle_input() {
ptrdiff_t result = 0; ptrdiff_t result = 0;
while (max_read_size > 0) { while (max_read_size > 0) {
......
...@@ -32,7 +32,9 @@ public: ...@@ -32,7 +32,9 @@ public:
bool can_send_more() const noexcept override; bool can_send_more() const noexcept override;
void suspend_reading() override; bool is_reading() const noexcept override;
void close() override;
void configure_read(caf::net::receive_policy policy) override; void configure_read(caf::net::receive_policy policy) override;
...@@ -42,8 +44,6 @@ public: ...@@ -42,8 +44,6 @@ public:
bool end_output() override; bool end_output() override;
bool stopped_reading() const noexcept override;
// -- initialization --------------------------------------------------------- // -- initialization ---------------------------------------------------------
caf::error init(const caf::settings& cfg) { caf::error init(const caf::settings& cfg) {
......
...@@ -65,10 +65,6 @@ public: ...@@ -65,10 +65,6 @@ public:
return !self->terminated(); return !self->terminated();
} }
void continue_reading() override {
CAF_FAIL("continue_reading called");
}
bool done_sending() override { bool done_sending() override {
return self->try_block_mailbox(); return self->try_block_mailbox();
} }
......
...@@ -81,8 +81,9 @@ public: ...@@ -81,8 +81,9 @@ public:
net::stream_oriented::lower_layer* down_ptr, net::stream_oriented::lower_layer* down_ptr,
const settings&) override { const settings&) override {
down = down_ptr; down = down_ptr;
if (auto ptr = adapter_type::try_open(mgr, std::move(input))) { if (auto buf = input.try_open()) {
adapter = std::move(ptr); auto do_wakeup = make_action([this] { prepare_send(); });
adapter = adapter_type::make(std::move(buf), mgr, std::move(do_wakeup));
return none; return none;
} else { } else {
FAIL("unable to open the resource"); FAIL("unable to open the resource");
...@@ -137,10 +138,6 @@ public: ...@@ -137,10 +138,6 @@ public:
return done || !adapter->has_data(); return done || !adapter->has_data();
} }
void continue_reading() override {
CAF_FAIL("continue_reading called");
}
void abort(const error& reason) override { void abort(const error& reason) override {
MESSAGE("app::abort: " << reason); MESSAGE("app::abort: " << reason);
} }
......
...@@ -51,6 +51,7 @@ public: ...@@ -51,6 +51,7 @@ public:
error init(net::socket_manager*, net::http::lower_layer* down_ptr, error init(net::socket_manager*, net::http::lower_layer* down_ptr,
const settings&) override { const settings&) override {
down = down_ptr; down = down_ptr;
down->request_messages();
return none; return none;
} }
...@@ -66,20 +67,15 @@ public: ...@@ -66,20 +67,15 @@ public:
return true; return true;
} }
ptrdiff_t consume(net::http::context ctx, ptrdiff_t consume(const net::http::header& request_hdr,
const net::http::header& request_hdr,
const_byte_span body) override { const_byte_span body) override {
hdr = request_hdr; hdr = request_hdr;
auto content = "Hello world!"sv; auto content = "Hello world!"sv;
down->send_response(ctx, net::http::status::ok, "text/plain", down->send_response(net::http::status::ok, "text/plain",
as_bytes(make_span(content))); as_bytes(make_span(content)));
payload.assign(body.begin(), body.end()); payload.assign(body.begin(), body.end());
return static_cast<ptrdiff_t>(body.size()); return static_cast<ptrdiff_t>(body.size());
} }
void continue_reading() override {
// nop
}
}; };
} // namespace } // namespace
......
...@@ -45,10 +45,11 @@ public: ...@@ -45,10 +45,11 @@ public:
// nop // nop
} }
caf::error init(net::socket_manager*, caf::error init(net::socket_manager* mgr_ptr,
net::message_oriented::lower_layer* down_ptr, net::message_oriented::lower_layer* down_ptr,
const settings&) override { const settings&) override {
// Start reading immediately. // Start reading immediately.
mgr = mgr_ptr;
down = down_ptr; down = down_ptr;
down->request_messages(); down->request_messages();
return none; return none;
...@@ -66,23 +67,23 @@ public: ...@@ -66,23 +67,23 @@ public:
// nop // nop
} }
void continue_reading() override { void continue_reading() {
down->request_messages(); mgr->schedule_fn([this] { down->request_messages(); });
} }
ptrdiff_t consume(byte_span buf) override { ptrdiff_t consume(byte_span buf) override {
printf("app_t::consume %d\n", __LINE__);
auto printable = [](std::byte x) { auto printable = [](std::byte x) {
return ::isprint(static_cast<uint8_t>(x)); return ::isprint(static_cast<uint8_t>(x));
}; };
if (CHECK(std::all_of(buf.begin(), buf.end(), printable))) { if (CHECK(std::all_of(buf.begin(), buf.end(), printable))) {
printf("app_t::consume %d\n", __LINE__);
auto str_buf = reinterpret_cast<char*>(buf.data()); auto str_buf = reinterpret_cast<char*>(buf.data());
inputs->emplace_back(std::string{str_buf, buf.size()}); inputs->emplace_back(std::string{str_buf, buf.size()});
printf("app_t::consume %d added %s\n", __LINE__, inputs->back().c_str()); MESSAGE("app: consumed " << inputs->back());
if constexpr (EnableSuspend) if constexpr (EnableSuspend)
if (inputs->back() == "pause") if (inputs->back() == "pause") {
MESSAGE("app: suspend reading");
down->suspend_reading(); down->suspend_reading();
}
std::string response = "ok "; std::string response = "ok ";
response += std::to_string(inputs->size()); response += std::to_string(inputs->size());
auto response_bytes = as_bytes(make_span(response)); auto response_bytes = as_bytes(make_span(response));
...@@ -92,11 +93,12 @@ public: ...@@ -92,11 +93,12 @@ public:
CHECK(down->end_message()); CHECK(down->end_message());
return static_cast<ptrdiff_t>(buf.size()); return static_cast<ptrdiff_t>(buf.size());
} else { } else {
printf("app_t::consume %d\n", __LINE__);
return -1; return -1;
} }
} }
net::socket_manager* mgr = nullptr;
net::message_oriented::lower_layer* down = nullptr; net::message_oriented::lower_layer* down = nullptr;
shared_string_list inputs; shared_string_list inputs;
...@@ -157,7 +159,7 @@ SCENARIO("length-prefix framing reads data with 32-bit size headers") { ...@@ -157,7 +159,7 @@ SCENARIO("length-prefix framing reads data with 32-bit size headers") {
} }
} }
SCENARIO("calling suspend_reading removes message apps temporarily") { SCENARIO("calling suspend_reading temporarily halts receiving of messages") {
using namespace std::literals; using namespace std::literals;
GIVEN("a length_prefix_framing with an app that consumes strings") { GIVEN("a length_prefix_framing with an app that consumes strings") {
auto [fd1, fd2] = unbox(net::make_stream_socket_pair()); auto [fd1, fd2] = unbox(net::make_stream_socket_pair());
...@@ -185,6 +187,7 @@ SCENARIO("calling suspend_reading removes message apps temporarily") { ...@@ -185,6 +187,7 @@ SCENARIO("calling suspend_reading removes message apps temporarily") {
CAF_FAIL("nonblocking returned an error: " << err); CAF_FAIL("nonblocking returned an error: " << err);
auto buf = std::make_shared<string_list>(); auto buf = std::make_shared<string_list>();
auto app = app_t<true>::make(buf); auto app = app_t<true>::make(buf);
auto app_ptr = app.get();
auto framing = net::length_prefix_framing::make(std::move(app)); auto framing = net::length_prefix_framing::make(std::move(app));
auto transport = net::stream_transport::make(fd2, std::move(framing)); auto transport = net::stream_transport::make(fd2, std::move(framing));
auto mgr = net::socket_manager::make(&mpx, fd2, std::move(transport)); auto mgr = net::socket_manager::make(&mpx, fd2, std::move(transport));
...@@ -201,8 +204,8 @@ SCENARIO("calling suspend_reading removes message apps temporarily") { ...@@ -201,8 +204,8 @@ SCENARIO("calling suspend_reading removes message apps temporarily") {
CHECK_EQ(buf->at(1), "second"); CHECK_EQ(buf->at(1), "second");
CHECK_EQ(buf->at(2), "pause"); CHECK_EQ(buf->at(2), "pause");
} }
THEN("users can resume it via continue_reading ") { THEN("users can resume it manually") {
mgr->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); CHECK_EQ(mpx.mask_of(mgr), net::operation::read);
......
...@@ -63,53 +63,33 @@ public: ...@@ -63,53 +63,33 @@ public:
// -- implementation of socket_event_layer ----------------------------------- // -- implementation of socket_event_layer -----------------------------------
error init(net::socket_manager*, const settings&) override { error init(net::socket_manager* mgr, const settings&) override {
mgr_ = mgr;
return none; return none;
} }
read_result handle_read_event() override { void handle_read_event() override {
// if (trigger_handover) {
// MESSAGE(name << " triggered a handover");
// return read_result::handover;
// }
if (read_capacity() < 1024) if (read_capacity() < 1024)
rd_buf_.resize(rd_buf_.size() + 2048); rd_buf_.resize(rd_buf_.size() + 2048);
auto num_bytes = read(fd_, auto res = read(fd_, make_span(read_position_begin(), read_capacity()));
make_span(read_position_begin(), read_capacity())); if (res > 0) {
if (num_bytes > 0) { CAF_ASSERT(res > 0);
CAF_ASSERT(num_bytes > 0); rd_buf_pos_ += res;
rd_buf_pos_ += num_bytes; } else if (res == 0 || !net::last_socket_error_is_temporary()) {
return read_result::again; mgr_->deregister();
} else if (num_bytes < 0 && net::last_socket_error_is_temporary()) {
return read_result::again;
} else {
return read_result::stop;
} }
} }
read_result handle_buffered_data() override { void handle_write_event() override {
return read_result::again; if (wr_buf_.size() == 0) {
} mgr_->deregister_writing();
} else if (auto res = write(fd_, wr_buf_); res > 0) {
read_result handle_continue_reading() override { wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + res);
return read_result::again; if (wr_buf_.size() == 0)
} mgr_->deregister_writing();
} else if (res == 0 || !net::last_socket_error_is_temporary()) {
write_result handle_write_event() override { mgr_->deregister();
// if (trigger_handover) {
// MESSAGE(name << " triggered a handover");
// return write_result::handover;
// }
if (wr_buf_.size() == 0)
return write_result::stop;
auto num_bytes = write(fd_, wr_buf_);
if (num_bytes > 0) {
wr_buf_.erase(wr_buf_.begin(), wr_buf_.begin() + num_bytes);
return wr_buf_.size() > 0 ? write_result::again : write_result::stop;
} }
return num_bytes < 0 && net::last_socket_error_is_temporary()
? write_result::again
: write_result::stop;
} }
void abort(const error& reason) override { void abort(const error& reason) override {
...@@ -130,7 +110,6 @@ private: ...@@ -130,7 +110,6 @@ private:
size_t read_capacity() const { size_t read_capacity() const {
return rd_buf_.size() - rd_buf_pos_; return rd_buf_.size() - rd_buf_pos_;
} }
net::stream_socket fd_; net::stream_socket fd_;
shared_count count_; shared_count count_;
...@@ -140,6 +119,8 @@ private: ...@@ -140,6 +119,8 @@ private:
byte_buffer wr_buf_; byte_buffer wr_buf_;
byte_buffer rd_buf_; byte_buffer rd_buf_;
net::socket_manager* mgr_ = nullptr;
}; };
struct fixture { struct fixture {
...@@ -168,7 +149,9 @@ struct fixture { ...@@ -168,7 +149,9 @@ struct fixture {
make_manager(net::stream_socket fd, std::string name) { make_manager(net::stream_socket fd, std::string name) {
auto mock = mock_event_layer::make(fd, std::move(name), manager_count); auto mock = mock_event_layer::make(fd, std::move(name), manager_count);
auto mock_ptr = mock.get(); auto mock_ptr = mock.get();
return {mock_ptr, net::socket_manager::make(&mpx, fd, std::move(mock))}; auto mgr = net::socket_manager::make(&mpx, fd, std::move(mock));
std::ignore = mgr->init(settings{});
return {mock_ptr, std::move(mgr)};
} }
void init() { void init() {
......
...@@ -75,18 +75,19 @@ public: ...@@ -75,18 +75,19 @@ public:
const settings&) override { const settings&) override {
down = down_ptr; down = down_ptr;
down->request_messages(); down->request_messages();
if (auto ptr = adapter_type::try_open(mgr, std::move(output_))) { if (auto buf = output_.try_open()) {
adapter_ = std::move(ptr); // 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(); });
adapter_ = adapter_type::make(std::move(buf), mgr, std::move(do_resume),
std::move(do_cancel));
return none; return none;
} else { } else {
FAIL("unable to open the resource"); FAIL("unable to open the resource");
} }
} }
void continue_reading() override {
down->request_messages();
}
bool prepare_send() override { bool prepare_send() override {
return true; return true;
} }
......
// 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
// prometheuss://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE net.prometheus.server
#include "caf/net/prometheus/server.hpp"
#include "net-test.hpp"
#include "caf/net/http/server.hpp"
#include "caf/telemetry/metric_registry.hpp"
using namespace caf;
using namespace caf::net;
using namespace std::literals;
constexpr std::string_view request_str
= "GET /metrics HTTP/1.1\r\n"
"Host: localhost:8090\r\n"
"User-Agent: Prometheus/2.18.1\r\n"
"Accept: text/plain;version=0.0.4\r\n"
"Accept-Encoding: gzip\r\n"
"X-Prometheus-Scrape-Timeout-Seconds: 5.000000\r\n\r\n";
SCENARIO("the Prometheus server responds to requests with scrape results") {
GIVEN("a valid Prometheus GET request") {
WHEN("sending it to an prometheus server") {
THEN("the Prometheus server responds with metrics text data") {
telemetry::metric_registry registry;
auto fb = registry.counter_singleton("foo", "bar", "test metric");
auto bf = registry.counter_singleton("bar", "foo", "test metric");
fb->inc(3);
bf->inc(7);
auto prom_state = prometheus::server::scrape_state::make(&registry);
auto prom_serv = prometheus::server::make(prom_state);
auto http_serv = net::http::server::make(std::move(prom_serv));
auto serv = mock_stream_transport::make(std::move(http_serv));
CHECK_EQ(serv->init(settings{}), error{});
serv->push(request_str);
CHECK_EQ(serv->handle_input(),
static_cast<ptrdiff_t>(request_str.size()));
auto out = serv->output_as_str();
CHECK(out.find("foo_bar 3"sv) != std::string_view::npos);
CHECK(out.find("bar_foo 7"sv) != std::string_view::npos);
}
}
}
}
...@@ -80,10 +80,6 @@ public: ...@@ -80,10 +80,6 @@ public:
return true; return true;
} }
void continue_reading() override {
// nop
}
ptrdiff_t consume(byte_span data, byte_span) override { ptrdiff_t consume(byte_span data, byte_span) override {
MESSAGE("dummy app received " << data.size() << " bytes"); MESSAGE("dummy app received " << data.size() << " bytes");
// Store the received bytes. // Store the received bytes.
...@@ -193,6 +189,7 @@ SCENARIO("ssl::transport::make_client performs the client handshake") { ...@@ -193,6 +189,7 @@ SCENARIO("ssl::transport::make_client performs the client handshake") {
THEN("CAF transparently calls SSL_connect") { THEN("CAF transparently calls SSL_connect") {
net::multiplexer mpx{nullptr}; net::multiplexer mpx{nullptr};
mpx.set_thread_id(); mpx.set_thread_id();
std::ignore = mpx.init();
auto ctx = unbox(ssl::context::make_client(ssl::tls::any)); auto ctx = unbox(ssl::context::make_client(ssl::tls::any));
auto conn = unbox(ctx.new_connection(client_fd)); auto conn = unbox(ctx.new_connection(client_fd));
auto done = std::make_shared<bool>(false); auto done = std::make_shared<bool>(false);
...@@ -230,6 +227,7 @@ SCENARIO("ssl::transport::make_server performs the server handshake") { ...@@ -230,6 +227,7 @@ SCENARIO("ssl::transport::make_server performs the server handshake") {
THEN("CAF transparently calls SSL_accept") { THEN("CAF transparently calls SSL_accept") {
net::multiplexer mpx{nullptr}; net::multiplexer mpx{nullptr};
mpx.set_thread_id(); mpx.set_thread_id();
std::ignore = mpx.init();
auto ctx = unbox(ssl::context::make_server(ssl::tls::any)); auto ctx = unbox(ssl::context::make_server(ssl::tls::any));
REQUIRE(ctx.use_certificate_from_file(cert_1_pem_path, // REQUIRE(ctx.use_certificate_from_file(cert_1_pem_path, //
ssl::format::pem)); ssl::format::pem));
......
...@@ -93,10 +93,6 @@ public: ...@@ -93,10 +93,6 @@ public:
return static_cast<ptrdiff_t>(recv_buf_->size()); return static_cast<ptrdiff_t>(recv_buf_->size());
} }
void continue_reading() override {
FAIL("continue_reading called");
}
bool prepare_send() override { bool prepare_send() override {
MESSAGE("prepare_send called"); MESSAGE("prepare_send called");
auto& buf = down->output_buffer(); auto& buf = down->output_buffer();
......
...@@ -65,10 +65,6 @@ public: ...@@ -65,10 +65,6 @@ public:
return !self->terminated(); return !self->terminated();
} }
void continue_reading() override {
CAF_FAIL("continue_reading called");
}
bool done_sending() override { bool done_sending() override {
return self->try_block_mailbox(); return self->try_block_mailbox();
} }
......
...@@ -40,10 +40,6 @@ public: ...@@ -40,10 +40,6 @@ public:
return true; return true;
} }
void continue_reading() override {
// nop
}
void abort(const error& reason) override { void abort(const error& reason) override {
CAF_FAIL("app::abort called: " << reason); CAF_FAIL("app::abort called: " << reason);
} }
......
...@@ -47,10 +47,6 @@ public: ...@@ -47,10 +47,6 @@ public:
return true; return true;
} }
void continue_reading() override {
// nop
}
void abort(const error& reason) override { void abort(const error& reason) override {
CAF_FAIL("app::abort called: " << reason); CAF_FAIL("app::abort called: " << reason);
} }
......
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