Commit 0cabe186 authored by Dominik Charousset's avatar Dominik Charousset

Collect metrics for in- and outbound buffers

parent 21cb3b2b
...@@ -339,6 +339,8 @@ public: ...@@ -339,6 +339,8 @@ public:
elements.erase(i); elements.erase(i);
return result; return result;
} }
std::pair<size_t, size_t> buffer_sizes();
/// @endcond /// @endcond
// -- overridden observers of abstract_actor --------------------------------- // -- overridden observers of abstract_actor ---------------------------------
...@@ -396,6 +398,7 @@ protected: ...@@ -396,6 +398,7 @@ protected:
datagram_servant_map& get_map(datagram_handle) { datagram_servant_map& get_map(datagram_handle) {
return datagram_servants_; return datagram_servants_;
} }
/// @endcond /// @endcond
/// Returns a `scribe` or `doorman` identified by `hdl`. /// Returns a `scribe` or `doorman` identified by `hdl`.
......
...@@ -58,6 +58,8 @@ public: ...@@ -58,6 +58,8 @@ public:
void graceful_shutdown() override; void graceful_shutdown() override;
std::pair<size_t, size_t> buffer_sizes() const noexcept override;
protected: protected:
template <class Policy> template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) { void handle_event_impl(io::network::operation op, Policy& policy) {
......
...@@ -93,6 +93,8 @@ public: ...@@ -93,6 +93,8 @@ public:
void graceful_shutdown() override; void graceful_shutdown() override;
std::pair<size_t, size_t> buffer_sizes() const noexcept override;
void add_endpoint(datagram_handle hdl, const ip_endpoint& ep, void add_endpoint(datagram_handle hdl, const ip_endpoint& ep,
const manager_ptr mgr); const manager_ptr mgr);
......
...@@ -24,6 +24,8 @@ ...@@ -24,6 +24,8 @@
#include "caf/io/network/operation.hpp" #include "caf/io/network/operation.hpp"
#include "caf/io/receive_policy.hpp" #include "caf/io/receive_policy.hpp"
#include <utility>
namespace caf::io::network { namespace caf::io::network {
/// A socket I/O event handler. /// A socket I/O event handler.
...@@ -64,6 +66,11 @@ public: ...@@ -64,6 +66,11 @@ public:
/// this event handler from the I/O loop. /// this event handler from the I/O loop.
virtual void graceful_shutdown() = 0; virtual void graceful_shutdown() = 0;
/// Returns the current size of any input (first) and output (second) buffer
/// managed by this handler. Always returns `(0, 0)` for event handlers
/// without buffers.
virtual std::pair<size_t, size_t> buffer_sizes() const noexcept = 0;
/// Returns the native socket handle for this handler. /// Returns the native socket handle for this handler.
native_socket fd() const { native_socket fd() const {
return fd_; return fd_;
......
...@@ -163,9 +163,18 @@ public: ...@@ -163,9 +163,18 @@ public:
} }
protected: protected:
void update_buffer_metrics(std::pair<size_t, size_t> prev,
std::pair<size_t, size_t> now);
/// Identifies the thread this multiplexer /// Identifies the thread this multiplexer
/// is running in. Must be set by the subclass. /// is running in. Must be set by the subclass.
std::thread::id tid_; std::thread::id tid_;
/// Caches middleman::metric_singletons.inbound_buffers_size.
telemetry::int_gauge* inbound_buffers_size_;
/// Caches middleman::metric_singletons.outbound_buffers_size.
telemetry::int_gauge* outbound_buffers_size_;
}; };
using multiplexer_ptr = std::unique_ptr<multiplexer>; using multiplexer_ptr = std::unique_ptr<multiplexer>;
......
...@@ -38,6 +38,8 @@ public: ...@@ -38,6 +38,8 @@ public:
void handle_event(operation op) override; void handle_event(operation op) override;
std::pair<size_t, size_t> buffer_sizes() const noexcept override;
void init(native_socket sock_fd); void init(native_socket sock_fd);
resumable* try_read_next(); resumable* try_read_next();
......
...@@ -40,6 +40,8 @@ public: ...@@ -40,6 +40,8 @@ public:
byte_buffer& rd_buf() override; byte_buffer& rd_buf() override;
std::pair<size_t, size_t> buffer_sizes() const noexcept override;
void graceful_shutdown() override; void graceful_shutdown() override;
void flush() override; void flush() override;
......
...@@ -80,6 +80,8 @@ public: ...@@ -80,6 +80,8 @@ public:
void graceful_shutdown() override; void graceful_shutdown() override;
std::pair<size_t, size_t> buffer_sizes() const noexcept override;
/// Forces this stream to subscribe to write events if no data is in the /// Forces this stream to subscribe to write events if no data is in the
/// write buffer. /// write buffer.
void force_empty_write(const manager_ptr& mgr); void force_empty_write(const manager_ptr& mgr);
......
...@@ -54,6 +54,11 @@ public: ...@@ -54,6 +54,11 @@ public:
/// Returns the current input buffer. /// Returns the current input buffer.
virtual byte_buffer& rd_buf() = 0; virtual byte_buffer& rd_buf() = 0;
/// Returns the current size of any input (first) and output (second) buffer
/// managed by this handler. Always returns `(0, 0)` for event handlers
/// without buffers.
virtual std::pair<size_t, size_t> buffer_sizes() const noexcept = 0;
/// Flushes the output buffer, i.e., sends the /// Flushes the output buffer, i.e., sends the
/// content of the buffer via the network. /// content of the buffer via the network.
virtual void flush() = 0; virtual void flush() = 0;
......
...@@ -377,6 +377,16 @@ abstract_broker::abstract_broker(actor_config& cfg) : scheduled_actor(cfg) { ...@@ -377,6 +377,16 @@ abstract_broker::abstract_broker(actor_config& cfg) : scheduled_actor(cfg) {
// nop // nop
} }
std::pair<size_t, size_t> abstract_broker::buffer_sizes() {
auto result = std::make_pair(size_t{0}, size_t{0});
for (auto& kvp : scribes_) {
auto [in_bufsize, out_bufsize] = kvp.second->buffer_sizes();
result.first += in_bufsize;
result.second += out_bufsize;
}
return result;
}
void abstract_broker::launch_servant(doorman_ptr& ptr) { void abstract_broker::launch_servant(doorman_ptr& ptr) {
// A doorman needs to be launched in addition to being initialized. This // A doorman needs to be launched in addition to being initialized. This
// allows CAF to assign doorman to uninitialized brokers. // allows CAF to assign doorman to uninitialized brokers.
......
...@@ -56,4 +56,8 @@ void acceptor::graceful_shutdown() { ...@@ -56,4 +56,8 @@ void acceptor::graceful_shutdown() {
shutdown_both(fd_); shutdown_both(fd_);
} }
std::pair<size_t, size_t> acceptor::buffer_sizes() const noexcept {
return {0, 0};
}
} // namespace caf::io::network } // namespace caf::io::network
...@@ -152,6 +152,17 @@ void datagram_handler::graceful_shutdown() { ...@@ -152,6 +152,17 @@ void datagram_handler::graceful_shutdown() {
// registered for reading or writing. // registered for reading or writing.
} }
std::pair<size_t, size_t> datagram_handler::buffer_sizes() const noexcept {
// Note: we could consider separate bookkeeping for avoiding O(n) here.
auto pending_bytes_total = size_t{0};
for (auto& wr_buf : wr_offline_buf_)
pending_bytes_total += wr_buf.second.size();
return {
0, // This handler does not cache data, only reads full datagrams.
pending_bytes_total,
};
}
void datagram_handler::prepare_next_read() { void datagram_handler::prepare_next_read() {
CAF_LOG_TRACE(CAF_ARG(wr_buf_.second.size()) CAF_LOG_TRACE(CAF_ARG(wr_buf_.second.size())
<< CAF_ARG(wr_offline_buf_.size())); << CAF_ARG(wr_offline_buf_.size()));
......
...@@ -154,6 +154,7 @@ default_multiplexer::default_multiplexer(actor_system* sys) ...@@ -154,6 +154,7 @@ default_multiplexer::default_multiplexer(actor_system* sys)
pipe_reader_(*this), pipe_reader_(*this),
servant_ids_(0), servant_ids_(0),
max_throughput_(0) { max_throughput_(0) {
CAF_ASSERT(sys != nullptr);
init(); init();
epollfd_ = epoll_create1(EPOLL_CLOEXEC); epollfd_ = epoll_create1(EPOLL_CLOEXEC);
if (epollfd_ == -1) { if (epollfd_ == -1) {
...@@ -237,10 +238,14 @@ void default_multiplexer::handle(const default_multiplexer::event& e) { ...@@ -237,10 +238,14 @@ void default_multiplexer::handle(const default_multiplexer::event& e) {
<< " from epoll"); << " from epoll");
op = EPOLL_CTL_DEL; op = EPOLL_CTL_DEL;
--shadow_; --shadow_;
if (e.ptr != nullptr)
update_buffer_metrics(e.ptr->buffer_sizes(), {0, 0});
} else if (old == 0) { } else if (old == 0) {
CAF_LOG_DEBUG("attempt to add socket " << CAF_ARG(e.fd) << " to epoll"); CAF_LOG_DEBUG("attempt to add socket " << CAF_ARG(e.fd) << " to epoll");
op = EPOLL_CTL_ADD; op = EPOLL_CTL_ADD;
++shadow_; ++shadow_;
if (e.ptr != nullptr)
update_buffer_metrics({0, 0}, e.ptr->buffer_sizes());
} else { } else {
CAF_LOG_DEBUG("modify epoll event mask for socket " CAF_LOG_DEBUG("modify epoll event mask for socket "
<< CAF_ARG(e.fd) << ": " << CAF_ARG(old) << " -> " << CAF_ARG(e.fd) << ": " << CAF_ARG(old) << " -> "
...@@ -253,6 +258,8 @@ void default_multiplexer::handle(const default_multiplexer::event& e) { ...@@ -253,6 +258,8 @@ void default_multiplexer::handle(const default_multiplexer::event& e) {
case EEXIST: case EEXIST:
CAF_LOG_ERROR("file descriptor registered twice"); CAF_LOG_ERROR("file descriptor registered twice");
--shadow_; --shadow_;
if (e.ptr != nullptr)
update_buffer_metrics(e.ptr->buffer_sizes(), {0, 0});
break; break;
// op was EPOLL_CTL_MOD or EPOLL_CTL_DEL, // op was EPOLL_CTL_MOD or EPOLL_CTL_DEL,
// and fd is not registered with this epoll instance. // and fd is not registered with this epoll instance.
...@@ -261,6 +268,8 @@ void default_multiplexer::handle(const default_multiplexer::event& e) { ...@@ -261,6 +268,8 @@ void default_multiplexer::handle(const default_multiplexer::event& e) {
"because it isn't registered"); "because it isn't registered");
if (e.mask == 0) { if (e.mask == 0) {
++shadow_; ++shadow_;
if (e.ptr != nullptr)
update_buffer_metrics({0, 0}, e.ptr->buffer_sizes());
} }
break; break;
default: default:
...@@ -423,10 +432,14 @@ void default_multiplexer::handle(const default_multiplexer::event& e) { ...@@ -423,10 +432,14 @@ void default_multiplexer::handle(const default_multiplexer::event& e) {
if (e.mask != 0) { if (e.mask != 0) {
pollset_.push_back(new_element); pollset_.push_back(new_element);
shadow_.push_back(e.ptr); shadow_.push_back(e.ptr);
if (e.ptr != nullptr)
update_buffer_metrics({0, 0}, e.ptr->buffer_sizes());
} }
} else if (i->fd == e.fd) { // modify } else if (i->fd == e.fd) { // modify
if (e.mask == 0) { if (e.mask == 0) {
// delete item // delete item
if (*j != nullptr)
update_buffer_metrics((*j)->buffer_sizes(), {0, 0});
pollset_.erase(i); pollset_.erase(i);
shadow_.erase(j); shadow_.erase(j);
} else { } else {
...@@ -446,6 +459,8 @@ void default_multiplexer::handle(const default_multiplexer::event& e) { ...@@ -446,6 +459,8 @@ void default_multiplexer::handle(const default_multiplexer::event& e) {
} else { // insert at iterator pos } else { // insert at iterator pos
pollset_.insert(i, new_element); pollset_.insert(i, new_element);
shadow_.insert(j, e.ptr); shadow_.insert(j, e.ptr);
if (e.ptr != nullptr)
update_buffer_metrics({0, 0}, e.ptr->buffer_sizes());
} }
} }
...@@ -559,6 +574,7 @@ void default_multiplexer::handle_socket_event(native_socket fd, int mask, ...@@ -559,6 +574,7 @@ void default_multiplexer::handle_socket_event(native_socket fd, int mask,
event_handler* ptr) { event_handler* ptr) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(mask)); CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(mask));
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
auto prev_buffer_sizes = ptr->buffer_sizes();
bool checkerror = true; bool checkerror = true;
if ((mask & input_mask) != 0) { if ((mask & input_mask) != 0) {
checkerror = false; checkerror = false;
...@@ -579,6 +595,7 @@ void default_multiplexer::handle_socket_event(native_socket fd, int mask, ...@@ -579,6 +595,7 @@ void default_multiplexer::handle_socket_event(native_socket fd, int mask,
del(operation::read, fd, ptr); del(operation::read, fd, ptr);
del(operation::write, fd, ptr); del(operation::write, fd, ptr);
} }
update_buffer_metrics(prev_buffer_sizes, ptr->buffer_sizes());
} }
void default_multiplexer::init() { void default_multiplexer::init() {
...@@ -616,6 +633,7 @@ bool default_multiplexer::poll_once(bool block) { ...@@ -616,6 +633,7 @@ bool default_multiplexer::poll_once(bool block) {
void default_multiplexer::resume(intrusive_ptr<resumable> ptr) { void default_multiplexer::resume(intrusive_ptr<resumable> ptr) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto impl = [&ptr, this](auto&& after_resume) {
switch (ptr->resume(this, max_throughput_)) { switch (ptr->resume(this, max_throughput_)) {
case resumable::resume_later: case resumable::resume_later:
// Delay resumable until next cycle. // Delay resumable until next cycle.
...@@ -625,7 +643,17 @@ void default_multiplexer::resume(intrusive_ptr<resumable> ptr) { ...@@ -625,7 +643,17 @@ void default_multiplexer::resume(intrusive_ptr<resumable> ptr) {
// Don't touch reference count of shutdown helpers. // Don't touch reference count of shutdown helpers.
ptr.release(); ptr.release();
break; break;
default:; // Done. Release reference to resumable. default:
after_resume();
}
};
if (ptr->subtype() == resumable::io_actor) {
auto dptr = static_cast<abstract_broker*>(ptr.get());
impl([this, dptr, prev_buffer_sizes{dptr->buffer_sizes()}] {
update_buffer_metrics(prev_buffer_sizes, dptr->buffer_sizes());
});
} else {
impl([] {});
} }
} }
......
...@@ -23,7 +23,15 @@ namespace caf::io::network { ...@@ -23,7 +23,15 @@ namespace caf::io::network {
multiplexer::multiplexer(actor_system* sys) multiplexer::multiplexer(actor_system* sys)
: execution_unit(sys), tid_(std::this_thread::get_id()) { : execution_unit(sys), tid_(std::this_thread::get_id()) {
// nop auto& reg = sys->metrics();
inbound_buffers_size_
= reg.gauge_singleton("caf.middleman", "inbound-buffers-size",
"The size of all inbound buffers combined.", "bytes",
true);
outbound_buffers_size_
= reg.gauge_singleton("caf.middleman", "outbound-buffers-size",
"The size of all outbound buffers combined.", "bytes",
true);
} }
multiplexer_ptr multiplexer::make(actor_system& sys) { multiplexer_ptr multiplexer::make(actor_system& sys) {
...@@ -51,4 +59,15 @@ void multiplexer::runnable::intrusive_ptr_release_impl() { ...@@ -51,4 +59,15 @@ void multiplexer::runnable::intrusive_ptr_release_impl() {
intrusive_ptr_release(this); intrusive_ptr_release(this);
} }
void multiplexer::update_buffer_metrics(std::pair<size_t, size_t> prev,
std::pair<size_t, size_t> now) {
auto delta = [](size_t x, size_t y) {
return static_cast<int64_t>(y) - static_cast<int64_t>(x);
};
if (auto din = delta(prev.first, now.first); din != 0)
inbound_buffers_size_->inc(din);
if (auto dout = delta(prev.second, now.second); dout != 0)
outbound_buffers_size_->inc(dout);
}
} // namespace caf::io::network } // namespace caf::io::network
...@@ -49,8 +49,8 @@ resumable* pipe_reader::try_read_next() { ...@@ -49,8 +49,8 @@ resumable* pipe_reader::try_read_next() {
std::intptr_t ptrval; std::intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles // on windows, we actually have sockets, otherwise we have file handles
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
auto res auto res = recv(fd(), reinterpret_cast<socket_recv_ptr>(&ptrval),
= recv(fd(), reinterpret_cast<socket_recv_ptr>(&ptrval), sizeof(ptrval), 0); sizeof(ptrval), 0);
#else #else
auto res = read(fd(), &ptrval, sizeof(ptrval)); auto res = read(fd(), &ptrval, sizeof(ptrval));
#endif #endif
...@@ -69,6 +69,10 @@ void pipe_reader::handle_event(operation op) { ...@@ -69,6 +69,10 @@ void pipe_reader::handle_event(operation op) {
// else: ignore errors // else: ignore errors
} }
std::pair<size_t, size_t> pipe_reader::buffer_sizes() const noexcept {
return {0, 0};
}
void pipe_reader::init(native_socket sock_fd) { void pipe_reader::init(native_socket sock_fd) {
fd_ = sock_fd; fd_ = sock_fd;
} }
......
...@@ -53,6 +53,10 @@ byte_buffer& scribe_impl::rd_buf() { ...@@ -53,6 +53,10 @@ byte_buffer& scribe_impl::rd_buf() {
return stream_.rd_buf(); return stream_.rd_buf();
} }
std::pair<size_t, size_t> scribe_impl::buffer_sizes() const noexcept {
return stream_.buffer_sizes();
}
void scribe_impl::graceful_shutdown() { void scribe_impl::graceful_shutdown() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
stream_.graceful_shutdown(); stream_.graceful_shutdown();
......
...@@ -100,6 +100,10 @@ void stream::graceful_shutdown() { ...@@ -100,6 +100,10 @@ void stream::graceful_shutdown() {
// Otherwise, send_fin() gets called after draining the send buffer. // Otherwise, send_fin() gets called after draining the send buffer.
} }
std::pair<size_t, size_t> stream::buffer_sizes() const noexcept {
return {rd_buf_.size(), wr_buf_.size() + wr_offline_buf_.size()};
}
void stream::force_empty_write(const manager_ptr& mgr) { void stream::force_empty_write(const manager_ptr& mgr) {
if (!state_.writing) { if (!state_.writing) {
backend().add(operation::write, fd(), this); backend().add(operation::write, fd(), this);
......
...@@ -100,6 +100,10 @@ scribe_ptr test_multiplexer::new_scribe(connection_handle hdl) { ...@@ -100,6 +100,10 @@ scribe_ptr test_multiplexer::new_scribe(connection_handle hdl) {
byte_buffer& rd_buf() override { byte_buffer& rd_buf() override {
return mpx_->input_buffer(hdl()); return mpx_->input_buffer(hdl());
} }
std::pair<size_t, size_t> buffer_sizes() const noexcept override {
return {mpx_->input_buffer(hdl()).size(),
mpx_->output_buffer(hdl()).size()};
}
void graceful_shutdown() override { void graceful_shutdown() override {
mpx_->stopped_reading(hdl()) = true; mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false); detach(mpx_, false);
......
...@@ -157,6 +157,10 @@ public: ...@@ -157,6 +157,10 @@ public:
return stream_.rd_buf(); return stream_.rd_buf();
} }
std::pair<size_t, size_t> buffer_sizes() const noexcept override {
return stream_.buffer_sizes();
}
void graceful_shutdown() override { void graceful_shutdown() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
stream_.graceful_shutdown(); stream_.graceful_shutdown();
......
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