Commit f20cd436 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/neverlord/metric-instances'

parents a58cc0b2 052e3b60
...@@ -15,6 +15,9 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -15,6 +15,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
the closing parenthesis. the closing parenthesis.
- The JSON reader now automatically widens integers to doubles as necessary. - The JSON reader now automatically widens integers to doubles as necessary.
- Module options (e.g. for the `middleman`) now show up in `--long-help` output. - Module options (e.g. for the `middleman`) now show up in `--long-help` output.
- Fix undefined behavior in the Qt group chat example (#1336).
- The `..._instance` convenience functions on the registry metric now properly
support `double` metrics and histograms.
### Changed ### Changed
......
...@@ -121,7 +121,7 @@ private: ...@@ -121,7 +121,7 @@ private:
} // namespace caf } // namespace caf
namespace caf::detail { namespace caf::detail {
template <class F> template <class F, bool IsSingleShot>
struct default_action_impl : detail::atomic_ref_counted, action::impl { struct default_action_impl : detail::atomic_ref_counted, action::impl {
std::atomic<action::state> state_; std::atomic<action::state> state_;
F f_; F f_;
...@@ -144,11 +144,12 @@ struct default_action_impl : detail::atomic_ref_counted, action::impl { ...@@ -144,11 +144,12 @@ struct default_action_impl : detail::atomic_ref_counted, action::impl {
} }
void run() override { void run() override {
// Note: we do *not* set the state to disposed after running the function
// object. This allows the action to re-register itself when needed, e.g.,
// to implement time-based loops.
if (state_.load() == action::state::scheduled) { if (state_.load() == action::state::scheduled) {
f_(); f_();
if constexpr (IsSingleShot)
state_ = action::state::disposed;
// else: allow the action to re-register itself when needed by *not*
// setting the state to disposed, e.g., to implement time loops.
} }
} }
...@@ -177,7 +178,15 @@ namespace caf { ...@@ -177,7 +178,15 @@ namespace caf {
/// @param f The body for the action. /// @param f The body for the action.
template <class F> template <class F>
action make_action(F f) { action make_action(F f) {
using impl_t = detail::default_action_impl<F>; using impl_t = detail::default_action_impl<F, false>;
return action{make_counted<impl_t>(std::move(f))};
}
/// Convenience function for creating an @ref action from a function object.
/// @param f The body for the action.
template <class F>
action make_single_shot_action(F f) {
using impl_t = detail::default_action_impl<F, true>;
return action{make_counted<impl_t>(std::move(f))}; return action{make_counted<impl_t>(std::move(f))};
} }
......
...@@ -47,7 +47,7 @@ public: ...@@ -47,7 +47,7 @@ public:
g(static_cast<const error&>(std::get<error>(cp->value))); g(static_cast<const error&>(std::get<error>(cp->value)));
} }
}; };
auto cb_action = make_action(std::move(cb)); auto cb_action = make_single_shot_action(std::move(cb));
auto event = typename cell_type::event{ctx_, cb_action}; auto event = typename cell_type::event{ctx_, cb_action};
bool fire_immediately = false; bool fire_immediately = false;
{ // Critical section. { // Critical section.
...@@ -115,6 +115,16 @@ public: ...@@ -115,6 +115,16 @@ public:
return {ctx, cell_}; return {ctx, cell_};
} }
/// Queries whether the result of the asynchronous computation is still
/// pending, i.e., neither `set_value` nor `set_error` has been called on the
/// @ref promise.
/// @pre `valid()`
bool pending() const {
CAF_ASSERT(valid());
std::unique_lock guard{cell_->mtx};
return std::holds_alternative<none_t>(cell_->value);
}
private: private:
using cell_ptr = std::shared_ptr<detail::async_cell<T>>; using cell_ptr = std::shared_ptr<detail::async_cell<T>>;
......
...@@ -19,16 +19,25 @@ template <class T> ...@@ -19,16 +19,25 @@ template <class T>
class promise { class promise {
public: public:
promise(promise&&) noexcept = default; promise(promise&&) noexcept = default;
promise(const promise&) noexcept = default;
promise& operator=(promise&&) noexcept = default; promise& operator=(promise&&) noexcept = default;
promise& operator=(const promise&) noexcept = default;
promise(const promise& other) noexcept : promise(other.cell_) {
// nop
}
promise& operator=(const promise& other) noexcept {
promise copy{other};
cell_.swap(copy.cell_);
return *this;
}
promise() : cell_(std::make_shared<cell_type>()) { promise() : cell_(std::make_shared<cell_type>()) {
// nop // nop
} }
~promise() { ~promise() {
if (cell_) { if (valid()) {
auto& cnt = cell_->promises; auto& cnt = cell_->promises;
if (cnt == 1 || cnt.fetch_sub(1, std::memory_order_acq_rel) == 1) { if (cnt == 1 || cnt.fetch_sub(1, std::memory_order_acq_rel) == 1) {
typename cell_type::event_list events; typename cell_type::event_list events;
...@@ -59,7 +68,7 @@ public: ...@@ -59,7 +68,7 @@ public:
/// @pre `valid()` /// @pre `valid()`
void set_value(T value) { void set_value(T value) {
if (cell_) { if (valid()) {
do_set(value); do_set(value);
cell_ = nullptr; cell_ = nullptr;
} }
...@@ -67,7 +76,7 @@ public: ...@@ -67,7 +76,7 @@ public:
/// @pre `valid()` /// @pre `valid()`
void set_error(error reason) { void set_error(error reason) {
if (cell_) { if (valid()) {
do_set(reason); do_set(reason);
cell_ = nullptr; cell_ = nullptr;
} }
...@@ -82,7 +91,7 @@ private: ...@@ -82,7 +91,7 @@ private:
using cell_type = detail::async_cell<T>; using cell_type = detail::async_cell<T>;
using cell_ptr = std::shared_ptr<cell_type>; using cell_ptr = std::shared_ptr<cell_type>;
explicit promise(cell_type cell) : cell_(std::move(cell)) { explicit promise(cell_ptr cell) noexcept : cell_(std::move(cell)) {
CAF_ASSERT(cell_ != nullptr); CAF_ASSERT(cell_ != nullptr);
cell_->promises.fetch_add(1, std::memory_order_relaxed); cell_->promises.fetch_add(1, std::memory_order_relaxed);
} }
......
...@@ -36,11 +36,12 @@ public: ...@@ -36,11 +36,12 @@ public:
~actor_widget() { ~actor_widget() {
if (companion_) if (companion_)
self()->cleanup(error{}, &dummy_); self()->cleanup(error{}, &execution_unit_);
} }
void init(actor_system& system) { void init(actor_system& system) {
alive_ = true; alive_ = true;
execution_unit_.system_ptr(&system);
companion_ = actor_cast<strong_actor_ptr>(system.spawn<actor_companion>()); companion_ = actor_cast<strong_actor_ptr>(system.spawn<actor_companion>());
self()->on_enqueue([=](mailbox_element_ptr ptr) { self()->on_enqueue([=](mailbox_element_ptr ptr) {
qApp->postEvent(this, new event_type(std::move(ptr))); qApp->postEvent(this, new event_type(std::move(ptr)));
...@@ -66,7 +67,7 @@ public: ...@@ -66,7 +67,7 @@ public:
if (event->type() == static_cast<QEvent::Type>(EventId)) { if (event->type() == static_cast<QEvent::Type>(EventId)) {
auto ptr = dynamic_cast<event_type*>(event); auto ptr = dynamic_cast<event_type*>(event);
if (ptr && alive_) { if (ptr && alive_) {
switch (self()->activate(&dummy_, *(ptr->mptr))) { switch (self()->activate(&execution_unit_, *(ptr->mptr))) {
default: default:
break; break;
}; };
...@@ -89,7 +90,7 @@ public: ...@@ -89,7 +90,7 @@ public:
} }
private: private:
scoped_execution_unit dummy_; scoped_execution_unit execution_unit_;
strong_actor_ptr companion_; strong_actor_ptr companion_;
bool alive_; bool alive_;
}; };
......
...@@ -136,8 +136,9 @@ public: ...@@ -136,8 +136,9 @@ public:
gauge_instance(std::string_view prefix, std::string_view name, gauge_instance(std::string_view prefix, std::string_view name,
span_t<label_view> labels, std::string_view helptext, span_t<label_view> labels, std::string_view helptext,
std::string_view unit = "1", bool is_sum = false) { std::string_view unit = "1", bool is_sum = false) {
auto fptr = gauge_family<ValueType>(prefix, name, labels, helptext, unit, auto label_names = get_label_names(labels);
is_sum); auto fptr = gauge_family<ValueType>(prefix, name, label_names, helptext,
unit, is_sum);
return fptr->get_or_add(labels); return fptr->get_or_add(labels);
} }
...@@ -228,27 +229,6 @@ public: ...@@ -228,27 +229,6 @@ public:
is_sum); is_sum);
} }
/// @copydoc counter_family
template <class ValueType = int64_t>
metric_family_impl<counter<ValueType>>*
counter_family(std::string_view prefix, std::string_view name,
span_t<label_view> labels, std::string_view helptext,
std::string_view unit = "1", bool is_sum = false) {
using counter_type = counter<ValueType>;
using family_type = metric_family_impl<counter_type>;
std::unique_lock<std::mutex> guard{families_mx_};
if (auto ptr = fetch(prefix, name)) {
assert_properties(ptr, counter_type::runtime_type, labels, unit, is_sum);
return static_cast<family_type*>(ptr);
}
auto ptr = std::make_unique<family_type>(
std::string{prefix}, std::string{name}, to_sorted_vec(labels),
std::string{helptext}, std::string{unit}, is_sum);
auto result = ptr.get();
families_.emplace_back(std::move(ptr));
return result;
}
/// Returns a counter. Creates all objects lazily if necessary, but fails /// Returns a counter. Creates all objects lazily if necessary, but fails
/// if the full name already belongs to a different family. /// if the full name already belongs to a different family.
/// @param prefix The prefix (namespace) this family belongs to. Usually the /// @param prefix The prefix (namespace) this family belongs to. Usually the
...@@ -269,8 +249,9 @@ public: ...@@ -269,8 +249,9 @@ public:
counter_instance(std::string_view prefix, std::string_view name, counter_instance(std::string_view prefix, std::string_view name,
span_t<label_view> labels, std::string_view helptext, span_t<label_view> labels, std::string_view helptext,
std::string_view unit = "1", bool is_sum = false) { std::string_view unit = "1", bool is_sum = false) {
auto fptr = counter_family<ValueType>(prefix, name, labels, helptext, unit, auto label_names = get_label_names(labels);
is_sum); auto fptr = counter_family<ValueType>(prefix, name, label_names, helptext,
unit, is_sum);
return fptr->get_or_add(labels); return fptr->get_or_add(labels);
} }
...@@ -419,8 +400,10 @@ public: ...@@ -419,8 +400,10 @@ public:
span_t<label_view> labels, span_t<ValueType> upper_bounds, span_t<label_view> labels, span_t<ValueType> upper_bounds,
std::string_view helptext, std::string_view unit = "1", std::string_view helptext, std::string_view unit = "1",
bool is_sum = false) { bool is_sum = false) {
auto fptr = histogram_family<ValueType>(prefix, name, labels, upper_bounds, auto label_names = get_label_names(labels);
helptext, unit, is_sum); auto fptr = histogram_family<ValueType>(prefix, name, label_names,
upper_bounds, helptext, unit,
is_sum);
return fptr->get_or_add(labels); return fptr->get_or_add(labels);
} }
...@@ -432,8 +415,8 @@ public: ...@@ -432,8 +415,8 @@ public:
span_t<ValueType> upper_bounds, std::string_view helptext, span_t<ValueType> upper_bounds, std::string_view helptext,
std::string_view unit = "1", bool is_sum = false) { std::string_view unit = "1", bool is_sum = false) {
span_t<label_view> lbls{labels.begin(), labels.size()}; span_t<label_view> lbls{labels.begin(), labels.size()};
return histogram_instance(prefix, name, lbls, upper_bounds, helptext, unit, return histogram_instance<ValueType>(prefix, name, lbls, upper_bounds,
is_sum); helptext, unit, is_sum);
} }
/// Returns a histogram metric singleton, i.e., the single instance of a /// Returns a histogram metric singleton, i.e., the single instance of a
...@@ -489,6 +472,8 @@ private: ...@@ -489,6 +472,8 @@ private:
metric_family* fetch(const std::string_view& prefix, metric_family* fetch(const std::string_view& prefix,
const std::string_view& name); const std::string_view& name);
static std::vector<std::string_view> get_label_names(span_t<label_view> xs);
static std::vector<std::string> to_sorted_vec(span_t<std::string_view> xs); static std::vector<std::string> to_sorted_vec(span_t<std::string_view> xs);
static std::vector<std::string> to_sorted_vec(span_t<label_view> xs); static std::vector<std::string> to_sorted_vec(span_t<label_view> xs);
......
...@@ -410,6 +410,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0) ...@@ -410,6 +410,7 @@ CAF_BEGIN_TYPE_ID_BLOCK(core_module, 0)
CAF_ADD_TYPE_ID(core_module, (caf::timestamp)) CAF_ADD_TYPE_ID(core_module, (caf::timestamp))
CAF_ADD_TYPE_ID(core_module, (caf::unit_t)) CAF_ADD_TYPE_ID(core_module, (caf::unit_t))
CAF_ADD_TYPE_ID(core_module, (caf::uri)) CAF_ADD_TYPE_ID(core_module, (caf::uri))
CAF_ADD_TYPE_ID(core_module, (caf::uuid))
CAF_ADD_TYPE_ID(core_module, (caf::weak_actor_ptr)) CAF_ADD_TYPE_ID(core_module, (caf::weak_actor_ptr))
CAF_ADD_TYPE_ID(core_module, (std::vector<caf::actor>) ) CAF_ADD_TYPE_ID(core_module, (std::vector<caf::actor>) )
CAF_ADD_TYPE_ID(core_module, (std::vector<caf::actor_addr>) ) CAF_ADD_TYPE_ID(core_module, (std::vector<caf::actor_addr>) )
......
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "caf/timestamp.hpp" #include "caf/timestamp.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
#include "caf/uuid.hpp"
namespace caf::core { namespace caf::core {
......
...@@ -63,6 +63,15 @@ metric_family* metric_registry::fetch(const std::string_view& prefix, ...@@ -63,6 +63,15 @@ metric_family* metric_registry::fetch(const std::string_view& prefix,
return nullptr; return nullptr;
} }
std::vector<std::string_view>
metric_registry::get_label_names(span_t<label_view> xs) {
std::vector<std::string_view> result;
result.reserve(xs.size());
for (auto& x : xs)
result.push_back(x.name());
return result;
}
std::vector<std::string> std::vector<std::string>
metric_registry::to_sorted_vec(span<const std::string_view> xs) { metric_registry::to_sorted_vec(span<const std::string_view> xs) {
std::vector<std::string> result; std::vector<std::string> result;
......
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#include "core-test.hpp" #include "core-test.hpp"
#include "caf/flow/scoped_coordinator.hpp"
#include "caf/scheduled_actor/flow.hpp" #include "caf/scheduled_actor/flow.hpp"
using namespace caf; using namespace caf;
...@@ -71,4 +72,55 @@ SCENARIO("actors can observe futures") { ...@@ -71,4 +72,55 @@ SCENARIO("actors can observe futures") {
} }
} }
SCENARIO("never setting a value or an error breaks the promises") {
GIVEN("multiple promises that point to the same cell") {
WHEN("the last promise goes out of scope") {
THEN("the future reports a broken promise") {
using promise_t = async::promise<int32_t>;
using future_t = async::future<int32_t>;
future_t fut;
{
auto uut = promise_t{};
fut = uut.get_future();
CHECK(fut.pending());
{
// copy ctor
promise_t cpy{uut};
CHECK(fut.pending());
// move ctor
promise_t mv{std::move(cpy)};
CHECK(fut.pending());
{
// copy assign
promise_t cpy2;
cpy2 = mv;
CHECK(fut.pending());
// move assign
promise_t mv2;
mv2 = std::move(mv);
CHECK(fut.pending());
}
CHECK(fut.pending());
}
CHECK(fut.pending());
}
CHECK(!fut.pending());
auto ctx = flow::scoped_coordinator::make();
size_t observed_events = 0;
fut.bind_to(ctx.get()).then(
[&observed_events](int32_t) {
++observed_events;
FAIL("unexpected value");
},
[&observed_events](const error& err) {
++observed_events;
CHECK_EQ(err, make_error(sec::broken_promise));
});
ctx->run();
CHECK_EQ(observed_events, 1u);
}
}
}
}
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
...@@ -12,7 +12,7 @@ namespace caf::io::basp { ...@@ -12,7 +12,7 @@ namespace caf::io::basp {
/// @{ /// @{
/// The current BASP version. Note: BASP is not backwards compatible. /// The current BASP version. Note: BASP is not backwards compatible.
constexpr uint64_t version = 4; constexpr uint64_t version = 5;
/// @} /// @}
......
...@@ -66,15 +66,11 @@ middleman::~middleman() { ...@@ -66,15 +66,11 @@ middleman::~middleman() {
void middleman::start() { void middleman::start() {
if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) { if (!get_or(config(), "caf.middleman.manual-multiplexing", false)) {
mpx_thread_ = std::thread{[this] { mpx_thread_ = sys_.launch_thread("caf.net.mpx", [this] {
CAF_SET_LOGGER_SYS(&sys_);
detail::set_thread_name("caf.net.mpx");
sys_.thread_started();
mpx_->set_thread_id(); mpx_->set_thread_id();
launch_background_tasks(sys_); launch_background_tasks(sys_);
mpx_->run(); mpx_->run();
sys_.thread_terminates(); });
}};
} else { } else {
mpx_->set_thread_id(); mpx_->set_thread_id();
} }
......
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